Implement MeterProvider's Meter method

sdk/metric/meter.go

@@ -19,21 +19,87 @@ package metric // import "go.opentelemetry.io/otel/sdk/metric"
 
 import (
 	"context"
+	"sync"
 
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/metric/instrument"
 	"go.opentelemetry.io/otel/metric/instrument/asyncfloat64"
 	"go.opentelemetry.io/otel/metric/instrument/asyncint64"
 	"go.opentelemetry.io/otel/metric/instrument/syncfloat64"
 	"go.opentelemetry.io/otel/metric/instrument/syncint64"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
 )
 
+// meterRegistry keeps a record of initialized meters for instrumentation
+// libraries. A meter is unique to an instrumentation library and if multiple
+// requests for that meter are made a meterRegistry ensure the same instance
+// is used.
+//
+// The zero meterRegistry is empty and ready for use.
+//
+// A meterRegistry must not be copied after first use.
+//
+// All methods of a meterRegistry are safe to call concurrently.
+type meterRegistry struct {
+	sync.Mutex
+
+	meters map[instrumentation.Library]*meter
+}
+
+// Get returns a registered meter matching the instrumentation library if it
+// exists in the meterRegistry. Otherwise, a new meter configured for the
+// instrumentation library is registered and then returned.
+//
+// The returned found bool is true if the returned meter already existed in
+// the registry and false if a new meter was created.
+//
+// Get is safe to call concurrently.
+func (r *meterRegistry) Get(l instrumentation.Library) (m *meter, found bool) {
+	r.Lock()
+	defer r.Unlock()
+
+	if r.meters == nil {
+		m = &meter{Library: l}
+		r.meters = map[instrumentation.Library]*meter{l: m}
+		return m, false
+	}
+
+	m, ok := r.meters[l]
+	if ok {
+		return m, true
+	}
+
+	m = &meter{Library: l}
+	r.meters[l] = m
+	return m, false
+}
+
+// Range calls f sequentially for each meter present in the meterRegistry. If
+// f returns false, the iteration is stopped.
+//
+// Range is safe to call concurrently.
+func (r *meterRegistry) Range(f func(*meter) bool) {
+	r.Lock()
+	defer r.Unlock()
+
+	for _, m := range r.meters {
+		if !f(m) {
+			return
+		}
+	}
+}
+
 // meter handles the creation and coordination of all metric instruments. A
 // meter represents a single instrumentation scope; all metric telemetry
 // produced by an instrumentation scope will use metric instruments from a
 // single meter.
 type meter struct {
-	// TODO (#2821, #2815, 2814): implement.
+	instrumentation.Library
+
+	// provider is the MeterProvider that created this meter.
+	provider *MeterProvider
+
+	// TODO (#2815, 2814): implement.
 }
 
 // Compile-time check meter implements metric.Meter.

sdk/metric/meter_test.go

@@ -0,0 +1,72 @@
+// Copyright The OpenTelemetry Authors
+//
+// 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.
+
+//go:build go1.17
+// +build go1.17
+
+package metric
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+)
+
+func TestMeterRegistryGetDoesNotPanicForZeroValue(t *testing.T) {
+	r := meterRegistry{}
+	assert.NotPanics(t, func() { _, _ = r.Get(instrumentation.Library{}) })
+}
+
+func TestMeterRegistry(t *testing.T) {
+	il0 := instrumentation.Library{Name: "zero"}
+	il1 := instrumentation.Library{Name: "one"}
+
+	r := meterRegistry{}
+	m0, ok := r.Get(il0)
+	t.Run("ZeroValueGet", func(t *testing.T) {
+		assert.Falsef(t, ok, "meter was in registry: %v", il0)
+	})
+
+	m01, ok := r.Get(il0)
+	t.Run("GetSameMeter", func(t *testing.T) {
+		assert.Truef(t, ok, "meter was not in registry: %v", il0)
+		assert.Samef(t, m0, m01, "returned different meters: %v", il0)
+	})
+
+	m1, ok := r.Get(il1)
+	t.Run("GetDifferentMeter", func(t *testing.T) {
+		assert.Falsef(t, ok, "meter was in registry: %v", il1)
+		assert.NotSamef(t, m0, m1, "returned same meters: %v", il1)
+	})
+
+	t.Run("RangeOrdered", func(t *testing.T) {
+		var got []*meter
+		r.Range(func(m *meter) bool {
+			got = append(got, m)
+			return true
+		})
+		assert.Equal(t, []*meter{m0, m1}, got)
+	})
+
+	t.Run("RangeStopIteration", func(t *testing.T) {
+		var i int
+		r.Range(func(m *meter) bool {
+			i++
+			return false
+		})
+		assert.Equal(t, 1, i, "iteration not stopped after first flase return")
+	})
+}

sdk/metric/provider.go

@@ -21,6 +21,7 @@ import (
 	"context"
 
 	"go.opentelemetry.io/otel/metric"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
 	"go.opentelemetry.io/otel/sdk/resource"
 )
 
@@ -31,6 +32,8 @@ import (
 type MeterProvider struct {
 	res *resource.Resource
 
+	meters meterRegistry
+
 	forceFlush, shutdown func(context.Context) error
 }
 
@@ -69,10 +72,16 @@ func NewMeterProvider(options ...Option) *MeterProvider {
 //
 // This method is safe to call concurrently.
 func (mp *MeterProvider) Meter(name string, options ...metric.MeterOption) metric.Meter {
-	// TODO (#2821): ensure this is concurrent safe.
-	// TODO: test this is concurrent safe.
-	// TODO (#2821): register and track the created Meter.
-	return &meter{}
+	c := metric.NewMeterConfig(options...)
+	m, existing := mp.meters.Get(instrumentation.Library{
+		Name:      name,
+		Version:   c.InstrumentationVersion(),
+		SchemaURL: c.SchemaURL(),
+	})
+	if !existing {
+		m.provider = mp
+	}
+	return m
 }
 
 // ForceFlush flushes all pending telemetry.

sdk/metric/provider_test.go

@@ -24,6 +24,17 @@ import (
 	"github.com/stretchr/testify/assert"
 )
 
+func TestMeterConcurrentSafe(t *testing.T) {
+	const name = "TestMeterConcurrentSafe meter"
+	mp := NewMeterProvider()
+
+	go func() {
+		_ = mp.Meter(name)
+	}()
+
+	_ = mp.Meter(name)
+}
+
 func TestForceFlushConcurrentSafe(t *testing.T) {
 	mp := NewMeterProvider()
 
@@ -44,6 +55,11 @@ func TestShutdownConcurrentSafe(t *testing.T) {
 	_ = mp.Shutdown(context.Background())
 }
 
+func TestMeterDoesNotPanicForEmptyMeterProvider(t *testing.T) {
+	mp := MeterProvider{}
+	assert.NotPanics(t, func() { _ = mp.Meter("") })
+}
+
 func TestForceFlushDoesNotPanicForEmptyMeterProvider(t *testing.T) {
 	mp := MeterProvider{}
 	assert.NotPanics(t, func() { _ = mp.ForceFlush(context.Background()) })