Skip to content

Commit

Permalink
Merge branch 'main' into bugfix5705
Browse files Browse the repository at this point in the history
  • Loading branch information
zhihali authored Sep 7, 2024
2 parents 0f832b9 + b37e8a9 commit d250d03
Show file tree
Hide file tree
Showing 9 changed files with 232 additions and 62 deletions.
3 changes: 3 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,13 +11,16 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
### Added

- Support `OTEL_EXPORTER_OTLP_LOGS_INSECURE` and `OTEL_EXPORTER_OTLP_INSECURE` environments in `go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc`. (#5739)
- The `WithResource` option for `NewMeterProvider` now merges the provided resources with the ones from environment variables. (#5773)
- The `WithResource` option for `NewLoggerProvider` now merges the provided resources with the ones from environment variables. (#5773)

### Fixed

- Fix memory leak in the global `MeterProvider` when identical instruments are repeatedly created. (#5754)
- Fix panic instruments creation when setting meter provider. (#5758)
- Stop percent encoding header environment variables in `open-telemetry/opentelemetry-go/exporters/otlp/otlplog/otlploggrpc` and `open-telemetry/opentelemetry-go/exporters/otlp/otlplog/otlploghttp` (#5792)
- Remove invalid environment variable header keys in `open-telemetry/opentelemetry-go/exporters/otlp/otlplog/otlploggrpc` and `open-telemetry/opentelemetry-go/exporters/otlp/otlplog/otlploghttp` (#5792)
- Fix an issue where `SetMeterProvider` in `go.opentelemetry.io/otel` might miss the delegation for instruments and registries. (#5780)

### Removed

Expand Down
133 changes: 80 additions & 53 deletions internal/global/meter.go
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,6 @@ import (
"container/list"
"reflect"
"sync"
"sync/atomic"

"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/metric/embedded"
Expand Down Expand Up @@ -97,7 +96,7 @@ type meter struct {

registry list.List

delegate atomic.Value // metric.Meter
delegate metric.Meter
}

type delegatedInstrument interface {
Expand All @@ -123,12 +122,12 @@ type instID struct {
//
// It is guaranteed by the caller that this happens only once.
func (m *meter) setDelegate(provider metric.MeterProvider) {
meter := provider.Meter(m.name, m.opts...)
m.delegate.Store(meter)

m.mtx.Lock()
defer m.mtx.Unlock()

meter := provider.Meter(m.name, m.opts...)
m.delegate = meter

for _, inst := range m.instruments {
inst.setDelegate(meter)
}
Expand All @@ -141,16 +140,18 @@ func (m *meter) setDelegate(provider metric.MeterProvider) {
m.registry.Remove(e)
}

clear(m.instruments)
m.instruments = nil
m.registry.Init()
}

func (m *meter) Int64Counter(name string, options ...metric.Int64CounterOption) (metric.Int64Counter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64Counter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64Counter(name, options...)
}

i := &siCounter{name: name, opts: options}
cfg := metric.NewInt64CounterConfig(options...)
id := instID{
Expand All @@ -164,11 +165,13 @@ func (m *meter) Int64Counter(name string, options ...metric.Int64CounterOption)
}

func (m *meter) Int64UpDownCounter(name string, options ...metric.Int64UpDownCounterOption) (metric.Int64UpDownCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64UpDownCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64UpDownCounter(name, options...)
}

i := &siUpDownCounter{name: name, opts: options}
cfg := metric.NewInt64UpDownCounterConfig(options...)
id := instID{
Expand All @@ -182,11 +185,13 @@ func (m *meter) Int64UpDownCounter(name string, options ...metric.Int64UpDownCou
}

func (m *meter) Int64Histogram(name string, options ...metric.Int64HistogramOption) (metric.Int64Histogram, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64Histogram(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64Histogram(name, options...)
}

i := &siHistogram{name: name, opts: options}
cfg := metric.NewInt64HistogramConfig(options...)
id := instID{
Expand All @@ -200,11 +205,13 @@ func (m *meter) Int64Histogram(name string, options ...metric.Int64HistogramOpti
}

func (m *meter) Int64Gauge(name string, options ...metric.Int64GaugeOption) (metric.Int64Gauge, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64Gauge(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64Gauge(name, options...)
}

i := &siGauge{name: name, opts: options}
cfg := metric.NewInt64GaugeConfig(options...)
id := instID{
Expand All @@ -218,11 +225,13 @@ func (m *meter) Int64Gauge(name string, options ...metric.Int64GaugeOption) (met
}

func (m *meter) Int64ObservableCounter(name string, options ...metric.Int64ObservableCounterOption) (metric.Int64ObservableCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64ObservableCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64ObservableCounter(name, options...)
}

i := &aiCounter{name: name, opts: options}
cfg := metric.NewInt64ObservableCounterConfig(options...)
id := instID{
Expand All @@ -236,11 +245,13 @@ func (m *meter) Int64ObservableCounter(name string, options ...metric.Int64Obser
}

func (m *meter) Int64ObservableUpDownCounter(name string, options ...metric.Int64ObservableUpDownCounterOption) (metric.Int64ObservableUpDownCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64ObservableUpDownCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64ObservableUpDownCounter(name, options...)
}

i := &aiUpDownCounter{name: name, opts: options}
cfg := metric.NewInt64ObservableUpDownCounterConfig(options...)
id := instID{
Expand All @@ -254,11 +265,13 @@ func (m *meter) Int64ObservableUpDownCounter(name string, options ...metric.Int6
}

func (m *meter) Int64ObservableGauge(name string, options ...metric.Int64ObservableGaugeOption) (metric.Int64ObservableGauge, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Int64ObservableGauge(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Int64ObservableGauge(name, options...)
}

i := &aiGauge{name: name, opts: options}
cfg := metric.NewInt64ObservableGaugeConfig(options...)
id := instID{
Expand All @@ -272,11 +285,13 @@ func (m *meter) Int64ObservableGauge(name string, options ...metric.Int64Observa
}

func (m *meter) Float64Counter(name string, options ...metric.Float64CounterOption) (metric.Float64Counter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64Counter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64Counter(name, options...)
}

i := &sfCounter{name: name, opts: options}
cfg := metric.NewFloat64CounterConfig(options...)
id := instID{
Expand All @@ -290,11 +305,13 @@ func (m *meter) Float64Counter(name string, options ...metric.Float64CounterOpti
}

func (m *meter) Float64UpDownCounter(name string, options ...metric.Float64UpDownCounterOption) (metric.Float64UpDownCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64UpDownCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64UpDownCounter(name, options...)
}

i := &sfUpDownCounter{name: name, opts: options}
cfg := metric.NewFloat64UpDownCounterConfig(options...)
id := instID{
Expand All @@ -308,11 +325,13 @@ func (m *meter) Float64UpDownCounter(name string, options ...metric.Float64UpDow
}

func (m *meter) Float64Histogram(name string, options ...metric.Float64HistogramOption) (metric.Float64Histogram, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64Histogram(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64Histogram(name, options...)
}

i := &sfHistogram{name: name, opts: options}
cfg := metric.NewFloat64HistogramConfig(options...)
id := instID{
Expand All @@ -326,11 +345,13 @@ func (m *meter) Float64Histogram(name string, options ...metric.Float64Histogram
}

func (m *meter) Float64Gauge(name string, options ...metric.Float64GaugeOption) (metric.Float64Gauge, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64Gauge(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64Gauge(name, options...)
}

i := &sfGauge{name: name, opts: options}
cfg := metric.NewFloat64GaugeConfig(options...)
id := instID{
Expand All @@ -344,11 +365,13 @@ func (m *meter) Float64Gauge(name string, options ...metric.Float64GaugeOption)
}

func (m *meter) Float64ObservableCounter(name string, options ...metric.Float64ObservableCounterOption) (metric.Float64ObservableCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64ObservableCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64ObservableCounter(name, options...)
}

i := &afCounter{name: name, opts: options}
cfg := metric.NewFloat64ObservableCounterConfig(options...)
id := instID{
Expand All @@ -362,11 +385,13 @@ func (m *meter) Float64ObservableCounter(name string, options ...metric.Float64O
}

func (m *meter) Float64ObservableUpDownCounter(name string, options ...metric.Float64ObservableUpDownCounterOption) (metric.Float64ObservableUpDownCounter, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64ObservableUpDownCounter(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64ObservableUpDownCounter(name, options...)
}

i := &afUpDownCounter{name: name, opts: options}
cfg := metric.NewFloat64ObservableUpDownCounterConfig(options...)
id := instID{
Expand All @@ -380,11 +405,13 @@ func (m *meter) Float64ObservableUpDownCounter(name string, options ...metric.Fl
}

func (m *meter) Float64ObservableGauge(name string, options ...metric.Float64ObservableGaugeOption) (metric.Float64ObservableGauge, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
return del.Float64ObservableGauge(name, options...)
}
m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
return m.delegate.Float64ObservableGauge(name, options...)
}

i := &afGauge{name: name, opts: options}
cfg := metric.NewFloat64ObservableGaugeConfig(options...)
id := instID{
Expand All @@ -399,14 +426,14 @@ func (m *meter) Float64ObservableGauge(name string, options ...metric.Float64Obs

// RegisterCallback captures the function that will be called during Collect.
func (m *meter) RegisterCallback(f metric.Callback, insts ...metric.Observable) (metric.Registration, error) {
if del, ok := m.delegate.Load().(metric.Meter); ok {
insts = unwrapInstruments(insts)
return del.RegisterCallback(f, insts...)
}

m.mtx.Lock()
defer m.mtx.Unlock()

if m.delegate != nil {
insts = unwrapInstruments(insts)
return m.delegate.RegisterCallback(f, insts...)
}

reg := &registration{instruments: insts, function: f}
e := m.registry.PushBack(reg)
reg.unreg = func() error {
Expand Down
15 changes: 11 additions & 4 deletions internal/global/meter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -83,6 +83,12 @@ func TestMeterConcurrentSafe(t *testing.T) {
mtr.setDelegate(noop.NewMeterProvider())
close(finish)
<-done

// No instruments should be left after the meter is replaced.
assert.Empty(t, mtr.instruments)

// No callbacks should be left after the meter is replaced.
assert.Zero(t, mtr.registry.Len())
}

func TestUnregisterConcurrentSafe(t *testing.T) {
Expand Down Expand Up @@ -162,8 +168,9 @@ func testSetupAllInstrumentTypes(t *testing.T, m metric.Meter) (metric.Float64Co
// This is to emulate a read from an exporter.
func testCollect(t *testing.T, m metric.Meter) {
if tMeter, ok := m.(*meter); ok {
m, ok = tMeter.delegate.Load().(metric.Meter)
if !ok {
// This changes the input m to the delegate.
m = tMeter.delegate
if m == nil {
t.Error("meter was not delegated")
return
}
Expand Down Expand Up @@ -261,7 +268,7 @@ func TestMeterDelegatesCalls(t *testing.T) {

// Calls to Meter methods after setDelegate() should be executed by the delegate
require.IsType(t, &meter{}, m)
tMeter := m.(*meter).delegate.Load().(*testMeter)
tMeter := m.(*meter).delegate.(*testMeter)
require.NotNil(t, tMeter)
assert.Equal(t, 1, tMeter.afCount)
assert.Equal(t, 1, tMeter.afUDCount)
Expand Down Expand Up @@ -309,7 +316,7 @@ func TestMeterDefersDelegations(t *testing.T) {

// Calls to Meter() before setDelegate() should be the delegated type
require.IsType(t, &meter{}, m)
tMeter := m.(*meter).delegate.Load().(*testMeter)
tMeter := m.(*meter).delegate.(*testMeter)
require.NotNil(t, tMeter)
assert.Equal(t, 1, tMeter.afCount)
assert.Equal(t, 1, tMeter.afUDCount)
Expand Down
1 change: 1 addition & 0 deletions sdk/log/go.mod
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ go 1.22
require (
github.com/go-logr/logr v1.4.2
github.com/go-logr/stdr v1.2.2
github.com/google/go-cmp v0.6.0
github.com/stretchr/testify v1.9.0
go.opentelemetry.io/otel v1.29.0
go.opentelemetry.io/otel/log v0.5.0
Expand Down
7 changes: 6 additions & 1 deletion sdk/log/provider.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"sync"
"sync/atomic"

"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/internal/global"
"go.opentelemetry.io/otel/log"
"go.opentelemetry.io/otel/log/embedded"
Expand Down Expand Up @@ -196,7 +197,11 @@ func (fn loggerProviderOptionFunc) apply(c providerConfig) providerConfig {
// go.opentelemetry.io/otel/sdk/resource package will be used.
func WithResource(res *resource.Resource) LoggerProviderOption {
return loggerProviderOptionFunc(func(cfg providerConfig) providerConfig {
cfg.resource = res
var err error
cfg.resource, err = resource.Merge(resource.Environment(), res)
if err != nil {
otel.Handle(err)
}
return cfg
})
}
Expand Down
Loading

0 comments on commit d250d03

Please sign in to comment.