Add back metric integration testing to the otelgocql/test package

instrumentation/github.com/gocql/gocql/otelgocql/test/go.mod

@@ -8,7 +8,9 @@ require (
 	go.opentelemetry.io/contrib v1.12.0
 	go.opentelemetry.io/contrib/instrumentation/github.com/gocql/gocql/otelgocql v0.37.0
 	go.opentelemetry.io/otel v1.11.2
+	go.opentelemetry.io/otel/metric v0.34.0
 	go.opentelemetry.io/otel/sdk v1.11.2
+	go.opentelemetry.io/otel/sdk/metric v0.34.0
 	go.opentelemetry.io/otel/trace v1.11.2
 )
 
@@ -19,7 +21,6 @@ require (
 	github.com/golang/snappy v0.0.3 // indirect
 	github.com/hailocab/go-hostpool v0.0.0-20160125115350-e80d13ce29ed // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	go.opentelemetry.io/otel/metric v0.34.0 // indirect
 	golang.org/x/sys v0.0.0-20220919091848-fb04ddd9f9c8 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

instrumentation/github.com/gocql/gocql/otelgocql/test/go.sum

@@ -38,6 +38,8 @@ go.opentelemetry.io/otel/metric v0.34.0 h1:MCPoQxcg/26EuuJwpYN1mZTeCYAUGx8ABxfW0
 go.opentelemetry.io/otel/metric v0.34.0/go.mod h1:ZFuI4yQGNCupurTXCwkeD/zHBt+C2bR7bw5JqUm/AP8=
 go.opentelemetry.io/otel/sdk v1.11.2 h1:GF4JoaEx7iihdMFu30sOyRx52HDHOkl9xQ8SMqNXUiU=
 go.opentelemetry.io/otel/sdk v1.11.2/go.mod h1:wZ1WxImwpq+lVRo4vsmSOxdd+xwoUJ6rqyLc3SyX9aU=
+go.opentelemetry.io/otel/sdk/metric v0.34.0 h1:7ElxfQpXCFZlRTvVRTkcUvK8Gt5DC8QzmzsLsO2gdzo=
+go.opentelemetry.io/otel/sdk/metric v0.34.0/go.mod h1:l4r16BIqiqPy5rd14kkxllPy/fOI4tWo1jkpD9Z3ffQ=
 go.opentelemetry.io/otel/trace v1.11.2 h1:Xf7hWSF2Glv0DE3MH7fBHvtpSBsjcBUe5MYAmZM/+y0=
 go.opentelemetry.io/otel/trace v1.11.2/go.mod h1:4N+yC7QEz7TTsG9BSRLNAa63eg5E06ObSbKPmxQ/pKA=
 golang.org/x/sys v0.0.0-20220919091848-fb04ddd9f9c8 h1:h+EGohizhe9XlX18rfpa8k8RAc5XyaeamM+0VHRd4lc=

instrumentation/github.com/gocql/gocql/otelgocql/test/gocql_test.go

@@ -30,17 +30,16 @@ import (
 	"go.opentelemetry.io/contrib/instrumentation/github.com/gocql/gocql/otelgocql/internal"
 	"go.opentelemetry.io/contrib/internal/util"
 	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric/unit"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	semconv "go.opentelemetry.io/otel/semconv/v1.12.0"
 	"go.opentelemetry.io/otel/trace"
 )
 
-// TODO(#2761): Add metric integration tests for the instrumentation. These
-// tests depend on
-// https://github.com/open-telemetry/opentelemetry-go/issues/3031 being
-// resolved.
-
 const (
 	keyspace  string = "gotest"
 	tableName string = "test_table"
@@ -59,13 +58,16 @@ func TestQuery(t *testing.T) {
 	cluster := getCluster()
 	sr := tracetest.NewSpanRecorder()
 	tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	reader := metric.NewManualReader()
+	meterProvider := metric.NewMeterProvider(metric.WithReader(reader))
 
 	ctx, parentSpan := tracerProvider.Tracer(internal.InstrumentationName).Start(context.Background(), "gocql-test")
 
 	session, err := otelgocql.NewSessionWithTracing(
 		ctx,
 		cluster,
 		otelgocql.WithTracerProvider(tracerProvider),
+		otelgocql.WithMeterProvider(meterProvider),
 		otelgocql.WithConnectInstrumentation(false),
 	)
 	require.NoError(t, err)
@@ -101,20 +103,32 @@ func TestQuery(t *testing.T) {
 		}
 		assertConnectionLevelAttributes(t, span)
 	}
+
+	rm, err := reader.Collect(context.Background())
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	sm := rm.ScopeMetrics[0]
+	assertScope(t, sm)
+	assertQueriesMetric(t, 1, insertStmt, requireMetric(t, "db.cassandra.queries", sm.Metrics))
+	assertRowsMetric(t, 1, requireMetric(t, "db.cassandra.rows", sm.Metrics))
+	assertLatencyMetric(t, 1, requireMetric(t, "db.cassandra.latency", sm.Metrics))
 }
 
 func TestBatch(t *testing.T) {
 	defer afterEach(t)
 	cluster := getCluster()
 	sr := tracetest.NewSpanRecorder()
 	tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	reader := metric.NewManualReader()
+	meterProvider := metric.NewMeterProvider(metric.WithReader(reader))
 
 	ctx, parentSpan := tracerProvider.Tracer(internal.InstrumentationName).Start(context.Background(), "gocql-test")
 
 	session, err := otelgocql.NewSessionWithTracing(
 		ctx,
 		cluster,
 		otelgocql.WithTracerProvider(tracerProvider),
+		otelgocql.WithMeterProvider(meterProvider),
 		otelgocql.WithConnectInstrumentation(false),
 	)
 	require.NoError(t, err)
@@ -144,20 +158,31 @@ func TestBatch(t *testing.T) {
 		assert.Contains(t, span.Attributes(), semconv.DBOperationKey.String("db.cassandra.batch.query"))
 		assertConnectionLevelAttributes(t, span)
 	}
+
+	rm, err := reader.Collect(context.Background())
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	sm := rm.ScopeMetrics[0]
+	assertScope(t, sm)
+	assertBatchQueriesMetric(t, 1, requireMetric(t, "db.cassandra.batch.queries", sm.Metrics))
+	assertLatencyMetric(t, 1, requireMetric(t, "db.cassandra.latency", sm.Metrics))
 }
 
 func TestConnection(t *testing.T) {
 	defer afterEach(t)
 	cluster := getCluster()
 	sr := tracetest.NewSpanRecorder()
 	tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sr))
+	reader := metric.NewManualReader()
+	meterProvider := metric.NewMeterProvider(metric.WithReader(reader))
 	connectObserver := &mockConnectObserver{0}
 	ctx := context.Background()
 
 	session, err := otelgocql.NewSessionWithTracing(
 		ctx,
 		cluster,
 		otelgocql.WithTracerProvider(tracerProvider),
+		otelgocql.WithMeterProvider(meterProvider),
 		otelgocql.WithConnectObserver(connectObserver),
 	)
 	require.NoError(t, err)
@@ -174,6 +199,13 @@ func TestConnection(t *testing.T) {
 		assert.Contains(t, span.Attributes(), semconv.DBOperationKey.String("db.cassandra.connect"))
 		assertConnectionLevelAttributes(t, span)
 	}
+
+	rm, err := reader.Collect(context.Background())
+	require.NoError(t, err)
+	require.Len(t, rm.ScopeMetrics, 1)
+	sm := rm.ScopeMetrics[0]
+	assertScope(t, sm)
+	assertConnectionsMetric(t, requireMetric(t, "db.cassandra.connections", sm.Metrics))
 }
 
 func TestHostOrIP(t *testing.T) {
@@ -218,6 +250,147 @@ func getCluster() *gocql.ClusterConfig {
 	return cluster
 }
 
+func assertScope(t *testing.T, sm metricdata.ScopeMetrics) {
+	assert.Equal(t, instrumentation.Scope{
+		Name:    "go.opentelemetry.io/contrib/instrumentation/github.com/gocql/gocql/otelgocql",
+		Version: otelgocql.SemVersion(),
+	}, sm.Scope)
+}
+
+func requireMetric(t *testing.T, name string, metrics []metricdata.Metrics) metricdata.Metrics {
+	m, ok := getMetric(name, metrics)
+	require.Truef(t, ok, "missing metric %q", name)
+	return m
+}
+
+func getMetric(name string, metrics []metricdata.Metrics) (metricdata.Metrics, bool) {
+	for _, m := range metrics {
+		if m.Name == name {
+			return m, true
+		}
+	}
+	return metricdata.Metrics{}, false
+}
+
+func assertQueriesMetric(t *testing.T, value int64, stmt string, m metricdata.Metrics) {
+	assert.Equal(t, "db.cassandra.queries", m.Name)
+	assert.Equal(t, "Number queries executed", m.Description)
+	require.IsType(t, m.Data, metricdata.Sum[int64]{})
+	data := m.Data.(metricdata.Sum[int64])
+	assert.Equal(t, metricdata.CumulativeTemporality, data.Temporality, "Temporality")
+	assert.True(t, data.IsMonotonic, "IsMonotonic")
+	require.Len(t, data.DataPoints, 1, "DataPoints")
+	dPt := data.DataPoints[0]
+	assert.Equal(t, value, dPt.Value, "Value")
+	assertAttrSet(t, []attribute.KeyValue{
+		internal.CassDBSystem(),
+		internal.CassPeerIP("127.0.0.1"),
+		internal.CassPeerPort(9042),
+		internal.CassVersion("3"),
+		internal.CassHostID("test-id"),
+		internal.CassHostState("UP"),
+		internal.CassKeyspace(keyspace),
+		internal.CassStatement(stmt),
+	}, dPt.Attributes)
+}
+
+func assertBatchQueriesMetric(t *testing.T, value int64, m metricdata.Metrics) {
+	assert.Equal(t, "db.cassandra.batch.queries", m.Name)
+	assert.Equal(t, "Number of batch queries executed", m.Description)
+	require.IsType(t, m.Data, metricdata.Sum[int64]{})
+	data := m.Data.(metricdata.Sum[int64])
+	assert.Equal(t, metricdata.CumulativeTemporality, data.Temporality, "Temporality")
+	assert.True(t, data.IsMonotonic, "IsMonotonic")
+	require.Len(t, data.DataPoints, 1, "DataPoints")
+	dPt := data.DataPoints[0]
+	assert.Equal(t, value, dPt.Value, "Value")
+	assertAttrSet(t, []attribute.KeyValue{
+		internal.CassDBSystem(),
+		internal.CassPeerIP("127.0.0.1"),
+		internal.CassPeerPort(9042),
+		internal.CassVersion("3"),
+		internal.CassHostID("test-id"),
+		internal.CassHostState("UP"),
+		internal.CassKeyspace(keyspace),
+	}, dPt.Attributes)
+}
+
+func assertConnectionsMetric(t *testing.T, m metricdata.Metrics) {
+	assert.Equal(t, "db.cassandra.connections", m.Name)
+	assert.Equal(t, "Number of connections created", m.Description)
+	require.IsType(t, m.Data, metricdata.Sum[int64]{})
+	data := m.Data.(metricdata.Sum[int64])
+	assert.Equal(t, metricdata.CumulativeTemporality, data.Temporality, "Temporality")
+	assert.True(t, data.IsMonotonic, "IsMonotonic")
+	for _, dPt := range data.DataPoints {
+		assertAttrSet(t, []attribute.KeyValue{
+			internal.CassDBSystem(),
+			internal.CassPeerIP("127.0.0.1"),
+			internal.CassPeerPort(9042),
+			internal.CassVersion("3"),
+			internal.CassHostID("test-id"),
+			internal.CassHostState("UP"),
+		}, dPt.Attributes)
+	}
+}
+
+func assertRowsMetric(t *testing.T, count uint64, m metricdata.Metrics) {
+	assert.Equal(t, "db.cassandra.rows", m.Name)
+	assert.Equal(t, "Number of rows returned from query", m.Description)
+	require.IsType(t, m.Data, metricdata.Histogram{})
+	data := m.Data.(metricdata.Histogram)
+	assert.Equal(t, metricdata.CumulativeTemporality, data.Temporality, "Temporality")
+	require.Len(t, data.DataPoints, 1, "DataPoints")
+	dPt := data.DataPoints[0]
+	assert.Equal(t, count, dPt.Count, "Count")
+	assertAttrSet(t, []attribute.KeyValue{
+		internal.CassDBSystem(),
+		internal.CassPeerIP("127.0.0.1"),
+		internal.CassPeerPort(9042),
+		internal.CassVersion("3"),
+		internal.CassHostID("test-id"),
+		internal.CassHostState("UP"),
+		internal.CassKeyspace(keyspace),
+	}, dPt.Attributes)
+}
+
+func assertLatencyMetric(t *testing.T, count uint64, m metricdata.Metrics) {
+	assert.Equal(t, "db.cassandra.latency", m.Name)
+	assert.Equal(t, "Sum of latency to host in milliseconds", m.Description)
+	assert.Equal(t, unit.Milliseconds, m.Unit)
+	require.IsType(t, m.Data, metricdata.Histogram{})
+	data := m.Data.(metricdata.Histogram)
+	assert.Equal(t, metricdata.CumulativeTemporality, data.Temporality, "Temporality")
+	require.Len(t, data.DataPoints, 1, "DataPoints")
+	dPt := data.DataPoints[0]
+	assert.Equal(t, count, dPt.Count, "Count")
+	assertAttrSet(t, []attribute.KeyValue{
+		internal.CassDBSystem(),
+		internal.CassPeerIP("127.0.0.1"),
+		internal.CassPeerPort(9042),
+		internal.CassVersion("3"),
+		internal.CassHostID("test-id"),
+		internal.CassHostState("UP"),
+		internal.CassKeyspace(keyspace),
+	}, dPt.Attributes)
+}
+
+func assertAttrSet(t *testing.T, want []attribute.KeyValue, got attribute.Set) {
+	for _, attr := range want {
+		actual, ok := got.Value(attr.Key)
+		if !assert.Truef(t, ok, "missing attribute %s", attr.Key) {
+			continue
+		}
+		switch attr.Key {
+		case internal.CassHostIDKey, internal.CassVersionKey:
+			// Host ID and Version will change between test runs.
+			assert.NotEmpty(t, actual)
+		default:
+			assert.Equal(t, attr.Value, actual)
+		}
+	}
+}
+
 // beforeAll creates the testing keyspace and table if they do not already exist.
 func beforeAll() error {
 	cluster := gocql.NewCluster("localhost")