replacing go.uber.org/atomic by sync/atomic (#7220)

replacing go.uber.org/atomic by sync/atomic
wrt issue #7160

.chloggen/use-sync/atomic.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: all
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "replacing uber-go/atomic by sync/atomic"
+
+# One or more tracking issues or pull requests related to the change
+issues: [7160]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

.golangci.yml

@@ -103,9 +103,8 @@ linters-settings:
     list-type: denylist
     include-go-root: true
     packages-with-error-message:
-      # See https://github.com/open-telemetry/opentelemetry-collector/issues/5200 for rationale
-      - sync/atomic: "Use go.uber.org/atomic instead of sync/atomic"
       - github.com/pkg/errors: "Use 'errors' or 'fmt' instead of github.com/pkg/errors"
+      - go.uber.org/atomic: "Use 'sync/atomic' instead of go.uber.org/atomic"
     # Add a different guard rule so that we can ignore tests.
     additional-guards:
       - list-type: denylist

confmap/resolver_test.go

@@ -19,11 +19,11 @@ import (
 	"errors"
 	"path/filepath"
 	"sync"
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 )
 
 type mockProvider struct {
@@ -270,7 +270,7 @@ func TestBackwardsCompatibilityForFilePath(t *testing.T) {
 }
 
 func TestResolver(t *testing.T) {
-	numCalls := atomic.NewInt32(0)
+	numCalls := atomic.Int32{}
 	resolver, err := NewResolver(ResolverSettings{
 		URIs: []string{"mock:"},
 		Providers: makeMapProvidersMap(&mockProvider{retM: map[string]any{}, closeFunc: func(ctx context.Context) error {

exporter/exporterhelper/internal/bounded_memory_queue.go

@@ -18,8 +18,7 @@ package internal // import "go.opentelemetry.io/collector/exporter/exporterhelpe
 
 import (
 	"sync"
-
-	"go.uber.org/atomic"
+	"sync/atomic"
 )
 
 // boundedMemoryQueue implements a producer-consumer exchange similar to a ring buffer queue,
@@ -40,8 +39,8 @@ type boundedMemoryQueue struct {
 func NewBoundedMemoryQueue(capacity int) ProducerConsumerQueue {
 	return &boundedMemoryQueue{
 		items:    make(chan Request, capacity),
-		stopped:  atomic.NewBool(false),
-		size:     atomic.NewUint32(0),
+		stopped:  &atomic.Bool{},
+		size:     &atomic.Uint32{},
 		capacity: uint32(capacity),
 	}
 }
@@ -57,7 +56,7 @@ func (q *boundedMemoryQueue) StartConsumers(numWorkers int, callback func(item R
 			startWG.Done()
 			defer q.stopWG.Done()
 			for item := range q.items {
-				q.size.Sub(1)
+				q.size.Add(^uint32(0))
 				callback(item)
 			}
 		}()
@@ -84,7 +83,7 @@ func (q *boundedMemoryQueue) Produce(item Request) bool {
 		return true
 	default:
 		// should not happen, as overflows should have been captured earlier
-		q.size.Sub(1)
+		q.size.Add(^uint32(0))
 		return false
 	}
 }

exporter/exporterhelper/internal/bounded_memory_queue_test.go

@@ -19,12 +19,12 @@ package internal
 import (
 	"reflect"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 )
 
 type stringRequest struct {
@@ -159,7 +159,7 @@ func newConsumerState(t *testing.T) *consumerState {
 	return &consumerState{
 		t:            t,
 		consumed:     make(map[string]bool),
-		consumedOnce: atomic.NewBool(false),
+		consumedOnce: &atomic.Bool{},
 	}
 }
 

exporter/exporterhelper/internal/persistent_queue_test.go

@@ -17,12 +17,12 @@ package internal
 import (
 	"context"
 	"fmt"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
@@ -176,10 +176,10 @@ func TestPersistentQueue_ConsumersProducers(t *testing.T) {
 			defer tq.Stop()
 			t.Cleanup(func() { assert.NoError(t, ext.Shutdown(context.Background())) })
 
-			numMessagesConsumed := atomic.NewInt32(0)
+			numMessagesConsumed := &atomic.Int32{}
 			tq.StartConsumers(c.numConsumers, func(item Request) {
 				if item != nil {
-					numMessagesConsumed.Inc()
+					numMessagesConsumed.Add(int32(1))
 				}
 			})
 

exporter/exporterhelper/internal/persistent_storage.go

@@ -19,8 +19,8 @@ import (
 	"errors"
 	"strconv"
 	"sync"
+	"sync/atomic"
 
-	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/extension/experimental/storage"
@@ -101,7 +101,7 @@ func newPersistentContiguousStorage(ctx context.Context, queueName string, capac
 		putChan:     make(chan struct{}, capacity),
 		reqChan:     make(chan Request),
 		stopChan:    make(chan struct{}),
-		itemsCount:  atomic.NewUint64(0),
+		itemsCount:  &atomic.Uint64{},
 	}
 
 	initPersistentContiguousStorage(ctx, pcs)

exporter/exporterhelper/queued_retry_test.go

@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"strconv"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -28,7 +29,6 @@ import (
 	"go.opencensus.io/metric/metricdata"
 	"go.opencensus.io/metric/metricproducer"
 	"go.opencensus.io/tag"
-	"go.uber.org/atomic"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/component/componenttest"
@@ -583,7 +583,8 @@ func TestQueuedRetryPersistenceEnabledStorageError(t *testing.T) {
 }
 
 func TestQueuedRetryPersistentEnabled_shutdown_dataIsRequeued(t *testing.T) {
-	produceCounter := atomic.NewUint32(0)
+
+	produceCounter := &atomic.Uint32{}
 
 	qCfg := NewDefaultQueueSettings()
 	qCfg.NumConsumers = 1
@@ -662,7 +663,7 @@ type mockRequest struct {
 }
 
 func (m *mockRequest) Export(ctx context.Context) error {
-	m.requestCount.Inc()
+	m.requestCount.Add(int64(1))
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	err := m.consumeError
@@ -703,7 +704,7 @@ func newMockRequest(ctx context.Context, cnt int, consumeError error) *mockReque
 		baseRequest:  baseRequest{ctx: ctx},
 		cnt:          cnt,
 		consumeError: consumeError,
-		requestCount: atomic.NewInt64(0),
+		requestCount: &atomic.Int64{},
 	}
 }
 
@@ -718,8 +719,8 @@ func newObservabilityConsumerSender(nextSender requestSender) *observabilityCons
 	return &observabilityConsumerSender{
 		waitGroup:         new(sync.WaitGroup),
 		nextSender:        nextSender,
-		droppedItemsCount: atomic.NewInt64(0),
-		sentItemsCount:    atomic.NewInt64(0),
+		droppedItemsCount: &atomic.Int64{},
+		sentItemsCount:    &atomic.Int64{},
 	}
 }
 

exporter/otlpexporter/go.mod

@@ -10,7 +10,6 @@ require (
 	go.opentelemetry.io/collector/consumer v0.72.0
 	go.opentelemetry.io/collector/exporter v0.0.0-20230303211526-ec5d71fec2da
 	go.opentelemetry.io/collector/pdata v1.0.0-rc6
-	go.uber.org/atomic v1.10.0
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f
 	google.golang.org/grpc v1.53.0
 	google.golang.org/protobuf v1.28.1
@@ -41,6 +40,7 @@ require (
 	go.opentelemetry.io/otel v1.13.0 // indirect
 	go.opentelemetry.io/otel/metric v0.36.0 // indirect
 	go.opentelemetry.io/otel/trace v1.13.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	go.uber.org/zap v1.24.0 // indirect
 	golang.org/x/net v0.7.0 // indirect

exporter/otlpexporter/otlp_test.go

@@ -20,12 +20,12 @@ import (
 	"path/filepath"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 	"google.golang.org/genproto/googleapis/rpc/errdetails"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
@@ -77,7 +77,7 @@ type mockTracesReceiver struct {
 }
 
 func (r *mockTracesReceiver) Export(ctx context.Context, req ptraceotlp.ExportRequest) (ptraceotlp.ExportResponse, error) {
-	r.requestCount.Inc()
+	r.requestCount.Add(int32(1))
 	td := req.Traces()
 	r.totalItems.Add(int32(td.SpanCount()))
 	r.mux.Lock()
@@ -112,8 +112,8 @@ func otlpTracesReceiverOnGRPCServer(ln net.Listener, useTLS bool) (*mockTracesRe
 	rcv := &mockTracesReceiver{
 		mockReceiver: mockReceiver{
 			srv:          grpc.NewServer(sopts...),
-			requestCount: atomic.NewInt32(0),
-			totalItems:   atomic.NewInt32(0),
+			requestCount: &atomic.Int32{},
+			totalItems:   &atomic.Int32{},
 		},
 	}
 
@@ -133,7 +133,7 @@ type mockLogsReceiver struct {
 }
 
 func (r *mockLogsReceiver) Export(ctx context.Context, req plogotlp.ExportRequest) (plogotlp.ExportResponse, error) {
-	r.requestCount.Inc()
+	r.requestCount.Add(int32(1))
 	ld := req.Logs()
 	r.totalItems.Add(int32(ld.LogRecordCount()))
 	r.mux.Lock()
@@ -153,8 +153,8 @@ func otlpLogsReceiverOnGRPCServer(ln net.Listener) *mockLogsReceiver {
 	rcv := &mockLogsReceiver{
 		mockReceiver: mockReceiver{
 			srv:          grpc.NewServer(),
-			requestCount: atomic.NewInt32(0),
-			totalItems:   atomic.NewInt32(0),
+			requestCount: &atomic.Int32{},
+			totalItems:   &atomic.Int32{},
 		},
 	}
 
@@ -175,7 +175,7 @@ type mockMetricsReceiver struct {
 
 func (r *mockMetricsReceiver) Export(ctx context.Context, req pmetricotlp.ExportRequest) (pmetricotlp.ExportResponse, error) {
 	md := req.Metrics()
-	r.requestCount.Inc()
+	r.requestCount.Add(int32(1))
 	r.totalItems.Add(int32(md.DataPointCount()))
 	r.mux.Lock()
 	defer r.mux.Unlock()
@@ -194,8 +194,8 @@ func otlpMetricsReceiverOnGRPCServer(ln net.Listener) *mockMetricsReceiver {
 	rcv := &mockMetricsReceiver{
 		mockReceiver: mockReceiver{
 			srv:          grpc.NewServer(),
-			requestCount: atomic.NewInt32(0),
-			totalItems:   atomic.NewInt32(0),
+			requestCount: &atomic.Int32{},
+			totalItems:   &atomic.Int32{},
 		},
 	}
 

featuregate/gate.go

@@ -14,7 +14,7 @@
 
 package featuregate // import "go.opentelemetry.io/collector/featuregate"
 
-import "go.uber.org/atomic"
+import "sync/atomic"
 
 // Gate is an immutable object that is owned by the Registry and represents an individual feature that
 // may be enabled or disabled based on the lifecycle state of the feature and CLI flags specified by the user.

featuregate/gate_test.go

@@ -15,17 +15,19 @@
 package featuregate
 
 import (
+	"sync/atomic"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/atomic"
 )
 
 func TestGate(t *testing.T) {
+	enabled := &atomic.Bool{}
+	enabled.Store(true)
 	g := &Gate{
 		id:             "test",
 		description:    "test gate",
-		enabled:        atomic.NewBool(true),
+		enabled:        enabled,
 		stage:          StageAlpha,
 		referenceURL:   "http://example.com",
 		removalVersion: "v0.64.0",

featuregate/go.mod

@@ -2,10 +2,7 @@ module go.opentelemetry.io/collector/featuregate
 
 go 1.19
 
-require (
-	github.com/stretchr/testify v1.8.2
-	go.uber.org/atomic v1.10.0
-)
+require github.com/stretchr/testify v1.8.2
 
 require (
 	github.com/davecgh/go-spew v1.1.1 // indirect

featuregate/go.sum

@@ -10,8 +10,6 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
 github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
-go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ=
-go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

featuregate/registry.go

@@ -18,8 +18,7 @@ import (
 	"fmt"
 	"sort"
 	"sync"
-
-	"go.uber.org/atomic"
+	"sync/atomic"
 )
 
 var globalRegistry = NewRegistry()
@@ -91,9 +90,11 @@ func (r *Registry) Register(id string, stage Stage, opts ...RegisterOption) (*Ga
 	}
 	switch g.stage {
 	case StageAlpha:
-		g.enabled = atomic.NewBool(false)
+		g.enabled = &atomic.Bool{}
 	case StageBeta, StageStable:
-		g.enabled = atomic.NewBool(true)
+		enabled := &atomic.Bool{}
+		enabled.Store(true)
+		g.enabled = enabled
 	default:
 		return nil, fmt.Errorf("unknown stage value %q for gate %q", stage, id)
 	}

go.mod

@@ -34,7 +34,6 @@ require (
 	go.opentelemetry.io/otel/sdk v1.13.0
 	go.opentelemetry.io/otel/sdk/metric v0.36.0
 	go.opentelemetry.io/otel/trace v1.13.0
-	go.uber.org/atomic v1.10.0
 	go.uber.org/multierr v1.9.0
 	go.uber.org/zap v1.24.0
 	golang.org/x/net v0.7.0
@@ -77,6 +76,7 @@ require (
 	github.com/tklauser/numcpus v0.6.0 // indirect
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.opentelemetry.io/contrib/zpages v0.39.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
 	google.golang.org/genproto v0.0.0-20230110181048-76db0878b65f // indirect
 	google.golang.org/protobuf v1.28.1 // indirect

otelcol/collector.go

@@ -22,9 +22,9 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"sync/atomic"
 	"syscall"
 
-	"go.uber.org/atomic"
 	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
@@ -119,9 +119,11 @@ func NewCollector(set CollectorSettings) (*Collector, error) {
 		return nil, errors.New("invalid nil config provider")
 	}
 
+	state := &atomic.Int32{}
+	state.Store(int32(StateStarting))
 	return &Collector{
 		set:          set,
-		state:        atomic.NewInt32(int32(StateStarting)),
+		state:        state,
 		shutdownChan: make(chan struct{}),
 		// Per signal.Notify documentation, a size of the channel equaled with
 		// the number of signals getting notified on is recommended.

processor/memorylimiterprocessor/go.mod

@@ -9,7 +9,6 @@ require (
 	go.opentelemetry.io/collector/confmap v0.72.0
 	go.opentelemetry.io/collector/consumer v0.72.0
 	go.opentelemetry.io/collector/pdata v1.0.0-rc6
-	go.uber.org/atomic v1.10.0
 	go.uber.org/zap v1.24.0
 )
 
@@ -39,6 +38,7 @@ require (
 	go.opentelemetry.io/otel v1.13.0 // indirect
 	go.opentelemetry.io/otel/metric v0.36.0 // indirect
 	go.opentelemetry.io/otel/trace v1.13.0 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.9.0 // indirect
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect

processor/memorylimiterprocessor/memorylimiter.go

@@ -20,9 +20,9 @@ import (
 	"fmt"
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"time"
 
-	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
@@ -129,7 +129,7 @@ func newMemoryLimiter(set processor.CreateSettings, cfg *Config) (*memoryLimiter
 		ticker:         time.NewTicker(cfg.CheckInterval),
 		readMemStatsFn: runtime.ReadMemStats,
 		logger:         logger,
-		forceDrop:      atomic.NewBool(false),
+		forceDrop:      &atomic.Bool{},
 		obsrep:         obsrep,
 	}
 

processor/memorylimiterprocessor/memorylimiter_test.go

@@ -17,12 +17,12 @@ package memorylimiterprocessor
 import (
 	"context"
 	"runtime"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.uber.org/atomic"
 	"go.uber.org/zap"
 
 	"go.opentelemetry.io/collector/component"
@@ -112,7 +112,7 @@ func TestMetricsMemoryPressureResponse(t *testing.T) {
 		usageChecker: memUsageChecker{
 			memAllocLimit: 1024,
 		},
-		forceDrop: atomic.NewBool(false),
+		forceDrop: &atomic.Bool{},
 		readMemStatsFn: func(ms *runtime.MemStats) {
 			ms.Alloc = currentMemAlloc
 		},
@@ -181,7 +181,7 @@ func TestTraceMemoryPressureResponse(t *testing.T) {
 		usageChecker: memUsageChecker{
 			memAllocLimit: 1024,
 		},
-		forceDrop: atomic.NewBool(false),
+		forceDrop: &atomic.Bool{},
 		readMemStatsFn: func(ms *runtime.MemStats) {
 			ms.Alloc = currentMemAlloc
 		},
@@ -250,7 +250,7 @@ func TestLogMemoryPressureResponse(t *testing.T) {
 		usageChecker: memUsageChecker{
 			memAllocLimit: 1024,
 		},
-		forceDrop: atomic.NewBool(false),
+		forceDrop: &atomic.Bool{},
 		readMemStatsFn: func(ms *runtime.MemStats) {
 			ms.Alloc = currentMemAlloc
 		},