Metrics: add attributes and switch to snake_case (#63)

* Metrics: add attributes and switch to camel_case

* Fix examples

Author: martin-majlis-s1

examples/client/main.go

@@ -21,6 +21,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/meter_config"
+
 	"go.opentelemetry.io/otel"
 
 	"github.com/scalyr/dataset-go/pkg/server_host_config"
@@ -67,7 +69,7 @@ func main() {
 		&http.Client{},
 		zap.Must(zap.NewDevelopment()),
 		&libraryConsumerUserAgentSuffix,
-		&meter,
+		meter_config.NewMeterConfig(&meter, "all", "example"),
 	)
 	if err != nil {
 		panic(err)

examples/readme/main.go

@@ -21,6 +21,8 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/meter_config"
+
 	"go.opentelemetry.io/otel"
 
 	"github.com/scalyr/dataset-go/pkg/server_host_config"
@@ -84,7 +86,7 @@ func main() {
 	libraryConsumerUserAgentSuffix := "OtelCollector-readme;1.2.3"
 	meter := otel.Meter("example.readme")
 	// build client
-	cl, err := client.NewClient(cfg, &http.Client{}, logger, &libraryConsumerUserAgentSuffix, &meter)
+	cl, err := client.NewClient(cfg, &http.Client{}, logger, &libraryConsumerUserAgentSuffix, meter_config.NewMeterConfig(&meter, "all", "example"))
 	if err != nil {
 		panic(err)
 	}

pkg/client/add_events_test.go

@@ -31,10 +31,10 @@ import (
 	"testing"
 	"time"
 
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/metric"
+	"github.com/scalyr/dataset-go/pkg/meter_config"
 
 	"github.com/scalyr/dataset-go/pkg/server_host_config"
+	"go.opentelemetry.io/otel"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -241,15 +241,15 @@ func TestAddEvents(t *testing.T) {
 	meter := otel.Meter("test")
 	tests := []struct {
 		name  string
-		meter *metric.Meter
+		meter *meter_config.MeterConfig
 	}{
 		{
 			name:  "no meter",
 			meter: nil,
 		},
 		{
 			name:  "with meter",
-			meter: &meter,
+			meter: meter_config.NewMeterConfig(&meter, "e", "n"),
 		},
 	}
 	for _, tt := range tests {

pkg/client/client.go

@@ -27,6 +27,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/meter_config"
+
 	"golang.org/x/exp/slices"
 
 	"github.com/scalyr/dataset-go/pkg/version"
@@ -43,7 +45,6 @@ import (
 
 	"github.com/cskr/pubsub"
 	"github.com/google/uuid"
-	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 )
 
@@ -104,7 +105,7 @@ func NewClient(
 	client *http.Client,
 	logger *zap.Logger,
 	userAgentSuffix *string,
-	meter *metric.Meter,
+	meterConfig *meter_config.MeterConfig,
 ) (*DataSetClient, error) {
 	logger.Info(
 		"Using config: ",
@@ -158,7 +159,7 @@ func NewClient(
 	}
 	logger.Info("Using User-Agent: ", zap.String("User-Agent", userAgent))
 
-	stats, err := statistics.NewStatistics(meter, logger)
+	stats, err := statistics.NewStatistics(meterConfig, logger)
 	if err != nil {
 		return nil, fmt.Errorf("it was not possible to create statistics: %w", err)
 	}

pkg/meter_config/meter_config.go

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2023 SentinelOne, Inc.
+ *
+ * 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 meter_config
+
+import "go.opentelemetry.io/otel/metric"
+
+type MeterConfig struct {
+	entity string
+	name   string
+	meter  *metric.Meter
+}
+
+func NewMeterConfig(meter *metric.Meter, entity string, name string) *MeterConfig {
+	return &MeterConfig{
+		entity: entity,
+		name:   name,
+		meter:  meter,
+	}
+}
+
+func (c *MeterConfig) Entity() string {
+	return c.entity
+}
+
+func (c *MeterConfig) Name() string {
+	return c.name
+}
+
+func (c *MeterConfig) Meter() *metric.Meter {
+	return c.meter
+}

pkg/meter_config/meter_config_test.go

@@ -0,0 +1,32 @@
+/*
+ * Copyright 2023 SentinelOne, Inc.
+ *
+ * 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 meter_config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/otel"
+)
+
+func TestNewMeterConfig(t *testing.T) {
+	meter := otel.Meter("AAA")
+	cfg := NewMeterConfig(&meter, "entity", "name")
+	assert.Equal(t, "entity", cfg.Entity())
+	assert.Equal(t, "name", cfg.Name())
+	assert.NotNil(t, cfg.Meter())
+}

pkg/statistics/statistics.go

@@ -22,6 +22,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/meter_config"
+	"go.opentelemetry.io/otel/attribute"
+
 	"go.uber.org/zap"
 
 	"github.com/scalyr/dataset-go/pkg/buffer_config"
@@ -44,8 +47,10 @@ type Statistics struct {
 	bytesAPISent     atomic.Uint64
 	bytesAPIAccepted atomic.Uint64
 
-	meter  *metric.Meter
-	logger *zap.Logger
+	config     *meter_config.MeterConfig
+	meter      *metric.Meter
+	logger     *zap.Logger
+	attributes []attribute.KeyValue
 
 	cBuffersEnqueued  metric.Int64UpDownCounter
 	cBuffersProcessed metric.Int64UpDownCounter
@@ -67,7 +72,7 @@ type Statistics struct {
 // NewStatistics creates structure to keep track of data processing.
 // If meter is not nil, then Open Telemetry is used for collecting metrics
 // as well.
-func NewStatistics(meter *metric.Meter, logger *zap.Logger) (*Statistics, error) {
+func NewStatistics(config *meter_config.MeterConfig, logger *zap.Logger) (*Statistics, error) {
 	logger.Info("Initialising statistics")
 	statistics := &Statistics{
 		buffersEnqueued:  atomic.Uint64{},
@@ -83,8 +88,10 @@ func NewStatistics(meter *metric.Meter, logger *zap.Logger) (*Statistics, error)
 		bytesAPIAccepted: atomic.Uint64{},
 		bytesAPISent:     atomic.Uint64{},
 
-		meter:  meter,
-		logger: logger,
+		config:     config,
+		meter:      nil,
+		logger:     logger,
+		attributes: []attribute.KeyValue{},
 	}
 
 	err := statistics.initMetrics()
@@ -97,54 +104,70 @@ func key(key string) string {
 }
 
 func (stats *Statistics) initMetrics() error {
+	// if there is no config, there is no need to initialise counters
+	if stats.config == nil {
+		stats.logger.Info("OTel metrics WILL NOT be collected (MeterConfig is nil)")
+		return nil
+	}
+
+	// update meter with config meter
+	stats.meter = stats.config.Meter()
 	meter := stats.meter
+
 	// if there is no meter, there is no need to initialise counters
 	if meter == nil {
-		stats.logger.Info("OTel metrics WILL NOT be collected")
+		stats.logger.Info("OTel metrics WILL NOT be collected (Meter is nil)")
 		return nil
 	}
 	stats.logger.Info("OTel metrics WILL be collected")
 
+	// set attributes so that we can track multiple instances
+	stats.attributes = []attribute.KeyValue{
+		{Key: "entity", Value: attribute.StringValue(stats.config.Entity())},
+		{Key: "name", Value: attribute.StringValue(stats.config.Name())},
+	}
+	metric.WithAttributes(stats.attributes...)
+
 	err := error(nil)
-	stats.cBuffersEnqueued, err = (*meter).Int64UpDownCounter(key("buffersEnqueued"))
+	stats.cBuffersEnqueued, err = (*meter).Int64UpDownCounter(key("buffers_enqueued"))
 	if err != nil {
 		return err
 	}
-	stats.cBuffersProcessed, err = (*meter).Int64UpDownCounter(key("buffersProcessed"))
+	stats.cBuffersProcessed, err = (*meter).Int64UpDownCounter(key("buffers_processed"))
 	if err != nil {
 		return err
 	}
-	stats.cBuffersDropped, err = (*meter).Int64UpDownCounter(key("buffersDropped"))
+	stats.cBuffersDropped, err = (*meter).Int64UpDownCounter(key("buffers_dropped"))
 	if err != nil {
 		return err
 	}
-	stats.cBuffersBroken, err = (*meter).Int64UpDownCounter(key("buffersBroken"))
+	stats.cBuffersBroken, err = (*meter).Int64UpDownCounter(key("buffers_broken"))
 	if err != nil {
 		return err
 	}
 
-	stats.cEventsEnqueued, err = (*meter).Int64UpDownCounter(key("eventsEnqueued"))
+	stats.cEventsEnqueued, err = (*meter).Int64UpDownCounter(key("events_enqueued"))
 	if err != nil {
 		return err
 	}
-	stats.cEventsProcessed, err = (*meter).Int64UpDownCounter(key("eventsProcessed"))
+	stats.cEventsProcessed, err = (*meter).Int64UpDownCounter(key("events_processed"))
 	if err != nil {
 		return err
 	}
-	stats.cEventsDropped, err = (*meter).Int64UpDownCounter(key("eventsDropped"))
+	stats.cEventsDropped, err = (*meter).Int64UpDownCounter(key("events_dropped"))
 	if err != nil {
 		return err
 	}
-	stats.cEventsBroken, err = (*meter).Int64UpDownCounter(key("eventsBroken"))
+	stats.cEventsBroken, err = (*meter).Int64UpDownCounter(key("events_broken"))
 	if err != nil {
 		return err
 	}
 
-	stats.cBytesAPISent, err = (*meter).Int64UpDownCounter(key("bytesAPISent"))
+	stats.cBytesAPISent, err = (*meter).Int64UpDownCounter(key("bytes_api_sent"))
 	if err != nil {
 		return err
 	}
-	stats.cBytesAPIAccepted, err = (*meter).Int64UpDownCounter(key("bytesAPIAccepted"))
+	stats.cBytesAPIAccepted, err = (*meter).Int64UpDownCounter(key("bytes_api_accepted"))
 	if err != nil {
 		return err
 	}
@@ -158,7 +181,7 @@ func (stats *Statistics) initMetrics() error {
 		zap.Float64s("buckets", payloadBuckets),
 	)
 	stats.hPayloadSize, err = (*meter).Int64Histogram(key(
-		"payloadSize"),
+		"payload_size"),
 		metric.WithExplicitBucketBoundaries(payloadBuckets...),
 		metric.WithUnit("b"),
 	)
@@ -175,7 +198,7 @@ func (stats *Statistics) initMetrics() error {
 		zap.Float64s("buckets", responseBuckets),
 	)
 	stats.hResponseTime, err = (*meter).Int64Histogram(key(
-		"responseTime"),
+		"response_time"),
 		metric.WithExplicitBucketBoundaries(responseBuckets...),
 		metric.WithUnit("ms"),
 	)
@@ -278,19 +301,31 @@ func (stats *Statistics) BytesAPIAcceptedAdd(i uint64) {
 
 func (stats *Statistics) PayloadSizeRecord(payloadSizeInBytes int64) {
 	if stats.hPayloadSize != nil {
-		stats.hPayloadSize.Record(context.Background(), payloadSizeInBytes)
+		stats.hPayloadSize.Record(
+			context.Background(),
+			payloadSizeInBytes,
+			metric.WithAttributes(stats.attributes...),
+		)
 	}
 }
 
 func (stats *Statistics) ResponseTimeRecord(duration time.Duration) {
 	if stats.hResponseTime != nil {
-		stats.hResponseTime.Record(context.Background(), duration.Milliseconds())
+		stats.hResponseTime.Record(
+			context.Background(),
+			duration.Milliseconds(),
+			metric.WithAttributes(stats.attributes...),
+		)
 	}
 }
 
 func (stats *Statistics) add(counter metric.Int64UpDownCounter, i uint64) {
 	if counter != nil {
-		counter.Add(context.Background(), int64(i))
+		counter.Add(
+			context.Background(),
+			int64(i),
+			metric.WithAttributes(stats.attributes...),
+		)
 	}
 }
 

pkg/statistics/statistics_test.go

@@ -21,28 +21,32 @@ import (
 	"testing"
 	"time"
 
+	"github.com/scalyr/dataset-go/pkg/meter_config"
+
 	"go.uber.org/zap"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/metric"
-
-	"github.com/stretchr/testify/assert"
 )
 
 var (
 	meter = otel.Meter("test")
 	tests = []struct {
 		name  string
-		meter *metric.Meter
+		meter *meter_config.MeterConfig
 	}{
 		{
-			name:  "no meter",
+			name:  "no config",
 			meter: nil,
 		},
+		{
+			name:  "no meter",
+			meter: meter_config.NewMeterConfig(nil, "e", "n"),
+		},
 		{
 			name:  "with meter",
-			meter: &meter,
+			meter: meter_config.NewMeterConfig(&meter, "e", "n"),
 		},
 	}
 )