Report metrics from system-log-monitor

pkg/util/metrics/fakes.go

@@ -0,0 +1,108 @@
+/*
+Copyright 2019 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package metrics
+
+import (
+	"errors"
+	"fmt"
+	"reflect"
+)
+
+// Int64MetricRepresentation represents a snapshot of an int64 metrics.
+// This is used for inspecting fake metrics.
+type Int64MetricRepresentation struct {
+	// Name is the metric name.
+	Name string
+	// Labels contains all metric labels in key-value pair format.
+	Labels map[string]string
+	// Value is the value of the metric.
+	Value int64
+}
+
+// Int64MetricInterface is used to create test double for Int64Metric.
+type Int64MetricInterface interface {
+	// Record records a measurement for the metric, with provided tags as metric labels.
+	Record(tags map[string]string, measurement int64) error
+}
+
+// FakeInt64Metric implements Int64MetricInterface.
+// FakeInt64Metric can be used as a test double for Int64MetricInterface, allowing
+// inspection of the metrics.
+type FakeInt64Metric struct {
+	name        string
+	aggregation Aggregation
+	allowedTags map[string]bool
+	metrics     []Int64MetricRepresentation
+}
+
+func NewFakeInt64Metric(name string, aggregation Aggregation, tagNames []string) *FakeInt64Metric {
+	if name == "" {
+		return nil
+	}
+
+	allowedTags := make(map[string]bool)
+	for _, tagName := range tagNames {
+		allowedTags[tagName] = true
+	}
+
+	fake := FakeInt64Metric{name, aggregation, allowedTags, []Int64MetricRepresentation{}}
+	return &fake
+}
+
+func (fake *FakeInt64Metric) Record(tags map[string]string, measurement int64) error {
+	labels := make(map[string]string)
+	for tagName, tagValue := range tags {
+		if _, ok := fake.allowedTags[tagName]; !ok {
+			return fmt.Errorf("tag %q is not allowed", tagName)
+		}
+		labels[tagName] = tagValue
+	}
+
+	metric := Int64MetricRepresentation{
+		Name:   fake.name,
+		Labels: labels,
+	}
+
+	// If there is a metric with equavalent labels, reuse it.
+	metricIndex := -1
+	for index, existingMetric := range fake.metrics {
+		if !reflect.DeepEqual(existingMetric.Labels, metric.Labels) {
+			continue
+		}
+		metricIndex = index
+		break
+	}
+	// If there is no metric with equalvalent labels, create a new one.
+	if metricIndex == -1 {
+		fake.metrics = append(fake.metrics, metric)
+		metricIndex = len(fake.metrics) - 1
+	}
+
+	switch fake.aggregation {
+	case LastValue:
+		fake.metrics[metricIndex].Value = measurement
+	case Sum:
+		fake.metrics[metricIndex].Value += measurement
+	default:
+		return errors.New("unsupported aggregation type")
+	}
+	return nil
+}
+
+// ListMetrics returns a snapshot of the current metrics.
+func (fake *FakeInt64Metric) ListMetrics() []Int64MetricRepresentation {
+	return fake.metrics
+}

pkg/util/metrics/fakes_test.go

@@ -0,0 +1,243 @@
+/*
+Copyright 2019 The Kubernetes Authors All rights reserved.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package metrics
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFakeInt64Metric(t *testing.T) {
+	type recordType struct {
+		tags        map[string]string
+		measurement int64
+	}
+	testCases := []struct {
+		name            string
+		metricName      string
+		aggregation     Aggregation
+		tagNames        []string
+		records         []recordType
+		expectedMetrics []Int64MetricRepresentation
+	}{
+		{
+			name:            "empty sum metric",
+			metricName:      "foo",
+			aggregation:     Sum,
+			tagNames:        []string{},
+			records:         []recordType{},
+			expectedMetrics: []Int64MetricRepresentation{},
+		},
+		{
+			name:        "sum metric with no tag",
+			metricName:  "foo",
+			aggregation: Sum,
+			tagNames:    []string{},
+			records: []recordType{
+				{
+					tags:        map[string]string{},
+					measurement: 1,
+				},
+				{
+					tags:        map[string]string{},
+					measurement: 2,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{},
+					Value:  3,
+				},
+			},
+		},
+		{
+			name:        "sum metric with one tag",
+			metricName:  "foo",
+			aggregation: Sum,
+			tagNames:    []string{"A"},
+			records: []recordType{
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 1,
+				},
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 2,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{"A": "1"},
+					Value:  3,
+				},
+			},
+		},
+		{
+			name:        "sum metric with different tags",
+			metricName:  "foo",
+			aggregation: Sum,
+			tagNames:    []string{"A", "B"},
+			records: []recordType{
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 1,
+				},
+				{
+					tags:        map[string]string{"B": "2"},
+					measurement: 2,
+				},
+				{
+					tags:        map[string]string{},
+					measurement: 4,
+				},
+				{
+					tags:        map[string]string{"B": "3"},
+					measurement: 8,
+				},
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 16,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{},
+					Value:  4,
+				},
+				{
+					Name:   "foo",
+					Labels: map[string]string{"A": "1"},
+					Value:  17,
+				},
+				{
+					Name:   "foo",
+					Labels: map[string]string{"B": "2"},
+					Value:  2,
+				},
+				{
+					Name:   "foo",
+					Labels: map[string]string{"B": "3"},
+					Value:  8,
+				},
+			},
+		},
+		{
+			name:            "empty gauge metric",
+			metricName:      "foo",
+			aggregation:     LastValue,
+			tagNames:        []string{},
+			records:         []recordType{},
+			expectedMetrics: []Int64MetricRepresentation{},
+		},
+		{
+			name:        "gauge metric with one measurement",
+			metricName:  "foo",
+			aggregation: LastValue,
+			tagNames:    []string{},
+			records: []recordType{
+				{
+					tags:        map[string]string{},
+					measurement: 2,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{},
+					Value:  2,
+				},
+			},
+		},
+		{
+			name:        "gauge metric with multiple measurements under same tag",
+			metricName:  "foo",
+			aggregation: LastValue,
+			tagNames:    []string{"A"},
+			records: []recordType{
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 2,
+				},
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 4,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{"A": "1"},
+					Value:  4,
+				},
+			},
+		},
+		{
+			name:        "gauge metric with multiple measurements under different tags",
+			metricName:  "foo",
+			aggregation: LastValue,
+			tagNames:    []string{"A", "B"},
+			records: []recordType{
+				{
+					tags:        map[string]string{"A": "1"},
+					measurement: 2,
+				},
+				{
+					tags:        map[string]string{"B": "2"},
+					measurement: 4,
+				},
+				{
+					tags:        map[string]string{"A": "1", "B": "2"},
+					measurement: 8,
+				},
+			},
+			expectedMetrics: []Int64MetricRepresentation{
+				{
+					Name:   "foo",
+					Labels: map[string]string{"A": "1"},
+					Value:  2,
+				},
+				{
+					Name:   "foo",
+					Labels: map[string]string{"B": "2"},
+					Value:  4,
+				},
+				{
+					Name:   "foo",
+					Labels: map[string]string{"A": "1", "B": "2"},
+					Value:  8,
+				},
+			},
+		},
+	}
+
+	for _, test := range testCases {
+		t.Run(test.name, func(t *testing.T) {
+			metric := NewFakeInt64Metric(test.metricName, test.aggregation, test.tagNames)
+
+			for _, record := range test.records {
+				metric.Record(record.tags, record.measurement)
+			}
+
+			gotMetrics := metric.ListMetrics()
+			assert.ElementsMatch(t, test.expectedMetrics, gotMetrics,
+				"expected metrics: %+v, got: %+v", test.expectedMetrics, gotMetrics)
+		})
+	}
+}

pkg/util/metrics/helpers.go

@@ -16,41 +16,174 @@ limitations under the License.
 package metrics
 
 import (
+	"context"
+	"fmt"
+	"sync"
+
 	"go.opencensus.io/stats"
 	"go.opencensus.io/stats/view"
 	"go.opencensus.io/tag"
 )
 
-// NewInt64Metric create a stats.Int64 metrics, returns nil when name is empty.
-func NewInt64Metric(name string, description string, unit string, aggregation *view.Aggregation, tagKeys []tag.Key) *stats.Int64Measure {
+var tagMap map[string]tag.Key
+var tagMapMutex sync.RWMutex
+
+func init() {
+	tagMapMutex.Lock()
+	tagMap = make(map[string]tag.Key)
+	tagMapMutex.Unlock()
+}
+
+// Int64Metric represents an int64 metric.
+type Int64Metric struct {
+	name    string
+	measure *stats.Int64Measure
+}
+
+// Aggregation defines how measurements should be aggregated into data points.
+type Aggregation string
+
+const (
+	// LastValue means last measurement overwrites previous measurements (gauge metric).
+	LastValue Aggregation = "LastValue"
+	// Sum means last measurement will be added onto previous measurements (counter metric).
+	Sum Aggregation = "Sum"
+)
+
+// NewInt64Metric create a Int64Metric metric, returns nil when name is empty.
+func NewInt64Metric(name string, description string, unit string, aggregation Aggregation, tagNames []string) (*Int64Metric, error) {
 	if name == "" {
-		return nil
+		return nil, nil
 	}
+
+	tagKeys, err := getTagKeysFromNames(tagNames)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create metric %q because of tag creation failure: %v", name, err)
+	}
+
+	var aggregationMethod *view.Aggregation
+	switch aggregation {
+	case LastValue:
+		aggregationMethod = view.LastValue()
+	case Sum:
+		aggregationMethod = view.Sum()
+	default:
+		return nil, fmt.Errorf("unknown aggregation option %q", aggregation)
+	}
+
 	measure := stats.Int64(name, description, unit)
 	newView := &view.View{
 		Name:        name,
 		Measure:     measure,
 		Description: description,
-		Aggregation: aggregation,
+		Aggregation: aggregationMethod,
 		TagKeys:     tagKeys,
 	}
 	view.Register(newView)
-	return measure
+
+	metric := Int64Metric{name, measure}
+	return &metric, nil
 }
 
-// NewFloat64Metric create a stats.Float64 metrics, returns nil when name is empty.
-func NewFloat64Metric(name string, description string, unit string, aggregation *view.Aggregation, tagKeys []tag.Key) *stats.Float64Measure {
-	if name == "" {
-		return nil
+// Record records a measurement for the metric, with provided tags as metric labels.
+func (metric *Int64Metric) Record(tags map[string]string, measurement int64) error {
+	var mutators []tag.Mutator
+
+	tagMapMutex.RLock()
+	defer tagMapMutex.RUnlock()
+
+	for tagName, tagValue := range tags {
+		tagKey, ok := tagMap[tagName]
+		if !ok {
+			return fmt.Errorf("referencing none existing tag %q in metric %q", tagName, metric.name)
+		}
+		mutators = append(mutators, tag.Upsert(tagKey, tagValue))
 	}
+
+	return stats.RecordWithTags(
+		context.Background(),
+		mutators,
+		metric.measure.M(measurement))
+}
+
+// Float64Metric represents an float64 metric.
+type Float64Metric struct {
+	name    string
+	measure *stats.Float64Measure
+}
+
+// NewFloat64Metric create a Float64Metric metrics, returns nil when name is empty.
+func NewFloat64Metric(name string, description string, unit string, aggregation Aggregation, tagNames []string) (*Float64Metric, error) {
+	if name == "" {
+		return nil, nil
+	}
+
+	tagKeys, err := getTagKeysFromNames(tagNames)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create metric %q because of tag creation failure: %v", name, err)
+	}
+
+	var aggregationMethod *view.Aggregation
+	switch aggregation {
+	case LastValue:
+		aggregationMethod = view.LastValue()
+	case Sum:
+		aggregationMethod = view.Sum()
+	default:
+		return nil, fmt.Errorf("unknown aggregation option %q", aggregation)
+	}
+
 	measure := stats.Float64(name, description, unit)
 	newView := &view.View{
 		Name:        name,
 		Measure:     measure,
 		Description: description,
-		Aggregation: aggregation,
+		Aggregation: aggregationMethod,
 		TagKeys:     tagKeys,
 	}
 	view.Register(newView)
-	return measure
+
+	metric := Float64Metric{name, measure}
+	return &metric, nil
+}
+
+// Record records a measurement for the metric, with provided tags as metric labels.
+func (metric *Float64Metric) Record(tags map[string]string, measurement float64) error {
+	var mutators []tag.Mutator
+
+	tagMapMutex.RLock()
+	defer tagMapMutex.RUnlock()
+
+	for tagName, tagValue := range tags {
+		tagKey, ok := tagMap[tagName]
+		if !ok {
+			return fmt.Errorf("referencing none existing tag %q in metric %q", tagName, metric.name)
+		}
+		mutators = append(mutators, tag.Upsert(tagKey, tagValue))
+	}
+
+	return stats.RecordWithTags(
+		context.Background(),
+		mutators,
+		metric.measure.M(measurement))
+}
+
+func getTagKeysFromNames(tagNames []string) ([]tag.Key, error) {
+	tagMapMutex.Lock()
+	defer tagMapMutex.Unlock()
+
+	var tagKeys []tag.Key
+	var err error
+	for _, tagName := range tagNames {
+		tagKey, ok := tagMap[tagName]
+		if !ok {
+			tagKey, err = tag.NewKey(tagName)
+			if err != nil {
+				return []tag.Key{}, fmt.Errorf("failed to create tag %q: %v", tagName, err)
+			}
+			tagMap[tagName] = tagKey
+		}
+		tagKeys = append(tagKeys, tagKey)
+	}
+	return tagKeys, nil
 }