replace uber atomic with sync/atomic

.golangci.yml

@@ -23,7 +23,14 @@ linters:
     - unused
     - usestdlibvars
     - whitespace
+    - depguard
   settings:
+    depguard:
+      rules:
+        denied-deps:
+          deny:
+            - pkg: go.uber.org/atomic
+              desc: use `sync/atomic` instead
     dupl:
       threshold: 120
     forbidigo:

fd/file.d.go

@@ -9,6 +9,7 @@ import (
 	"net/http/pprof"
 	"runtime"
 	"runtime/debug"
+	"sync/atomic"
 
 	"github.com/bitly/go-simplejson"
 	"github.com/ozontech/file.d/buildinfo"
@@ -20,7 +21,6 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/collectors"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
-	"go.uber.org/atomic"
 )
 
 type FileD struct {

go.mod

@@ -47,7 +47,6 @@ require (
 	github.com/twmb/franz-go/plugin/kzap v1.1.2
 	github.com/twmb/tlscfg v1.2.1
 	github.com/valyala/fasthttp v1.48.0
-	go.uber.org/atomic v1.11.0
 	go.uber.org/automaxprocs v1.5.3
 	go.uber.org/zap v1.27.0
 	golang.org/x/oauth2 v0.36.0
@@ -136,6 +135,7 @@ require (
 	go.opentelemetry.io/otel v1.34.0 // indirect
 	go.opentelemetry.io/otel/metric v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.34.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect

pipeline/action_watcher.go

@@ -6,7 +6,7 @@ import (
 	"sync"
 	"time"
 
-	"go.uber.org/atomic"
+	"sync/atomic"
 )
 
 // actionWatcher is the struct for debugging actions.
@@ -38,7 +38,7 @@ func newActionWatcher(procID int) *actionWatcher {
 	return &actionWatcher{
 		procID:     procID,
 		samples:    make(map[int][]*sample),
-		samplesLen: atomic.NewInt64(0),
+		samplesLen: &atomic.Int64{},
 		samplesMu:  sync.Mutex{},
 	}
 }
@@ -70,7 +70,7 @@ func (aw *actionWatcher) addSample(actionIdx int) *sample {
 	}
 	aw.samplesMu.Lock()
 	aw.samples[actionIdx] = append(aw.samples[actionIdx], s)
-	aw.samplesLen.Inc()
+	aw.samplesLen.Add(1)
 	aw.samplesMu.Unlock()
 
 	return s
@@ -94,7 +94,7 @@ func (aw *actionWatcher) deleteSample(actionIdx int, sample *sample) {
 	samples[deleteInd] = samples[len(samples)-1]
 	samples[len(samples)-1] = nil
 	aw.samples[actionIdx] = samples[:len(samples)-1]
-	aw.samplesLen.Dec()
+	aw.samplesLen.Add(-1)
 }
 
 var timeoutEvent = []byte("<timeout event>")

pipeline/antispam/antispammer.go

@@ -3,12 +3,12 @@ package antispam
 import (
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/ozontech/file.d/cfg/matchrule"
 	"github.com/ozontech/file.d/logger"
 	"github.com/ozontech/file.d/metric"
-	"go.uber.org/atomic"
 	"go.uber.org/zap"
 )
 
@@ -172,7 +172,7 @@ func (a *Antispammer) IsSpam(id string, name string, isNewSource bool, event []b
 	x := src.counter.Load()
 	diff := timeEventSeconds - src.timestamp.Swap(timeEventSeconds)
 	if diff < a.maintenanceInterval.Nanoseconds() {
-		x = src.counter.Inc()
+		x = src.counter.Add(1)
 	}
 	if x == int32(threshold) {
 		src.counter.Swap(int32(a.unbanIterations * threshold))

pipeline/backoff_test.go

@@ -2,13 +2,13 @@ package pipeline
 
 import (
 	"errors"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ozontech/file.d/metric"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/atomic"
 )
 
 func TestBackoff(t *testing.T) {
@@ -19,15 +19,15 @@ func TestBackoff(t *testing.T) {
 	eventCountBefore := eventCount.Load()
 
 	errorFn := func(err error, events []*Event) {
-		errorCount.Inc()
+		errorCount.Add(1)
 	}
 
 	batcherBackoff := NewRetriableBatcher(
 		&BatcherOptions{
 			MetricCtl: metric.NewCtl("", prometheus.NewRegistry(), time.Minute, 0),
 		},
 		func(workerData *WorkerData, batch *Batch) error {
-			eventCount.Inc()
+			eventCount.Add(1)
 			return nil
 		},
 		BackoffOpts{
@@ -49,7 +49,7 @@ func TestBackoffWithError(t *testing.T) {
 	errorCount := &atomic.Int32{}
 	prevValue := errorCount.Load()
 	errorFn := func(err error, events []*Event) {
-		errorCount.Inc()
+		errorCount.Add(1)
 	}
 
 	batcherBackoff := NewRetriableBatcher(
@@ -79,7 +79,7 @@ func TestBackoffWithErrorWithDeadQueue(t *testing.T) {
 	errorCount := &atomic.Int32{}
 	prevValue := errorCount.Load()
 	errorFn := func(err error, events []*Event) {
-		errorCount.Inc()
+		errorCount.Add(1)
 	}
 
 	batcherBackoff := NewRetriableBatcher(

pipeline/batch_test.go

@@ -3,14 +3,14 @@ package pipeline
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/ozontech/file.d/logger"
 	"github.com/ozontech/file.d/metric"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
-	"go.uber.org/atomic"
 )
 
 type batcherTail struct {
@@ -41,7 +41,7 @@ func TestBatcher(t *testing.T) {
 			*workerData = batchCount
 		}
 		counter := (*workerData).(*atomic.Int32)
-		counter.Inc()
+		counter.Add(1)
 	}
 
 	seqIDs := make(map[SourceID]uint64)
@@ -55,7 +55,7 @@ func TestBatcher(t *testing.T) {
 		}
 		seqIDs[event.SourceID] = event.SeqID
 
-		commitsCount.Inc()
+		commitsCount.Add(1)
 		wg.Done()
 	}}
 
@@ -112,7 +112,7 @@ func TestBatcherMaxSize(t *testing.T) {
 			*workerData = batchCount
 		}
 		counter := (*workerData).(*atomic.Int32)
-		counter.Inc()
+		counter.Add(1)
 	}
 
 	seqIDs := make(map[SourceID]uint64)
@@ -126,7 +126,7 @@ func TestBatcherMaxSize(t *testing.T) {
 		}
 		seqIDs[event.SourceID] = event.SeqID
 
-		commitsCount.Inc()
+		commitsCount.Add(1)
 		wg.Done()
 	}}
 

pipeline/event.go

@@ -8,9 +8,10 @@ import (
 	"sync"
 	"time"
 
+	"sync/atomic"
+
 	"github.com/ozontech/file.d/logger"
 	insaneJSON "github.com/ozontech/insane-json"
-	"go.uber.org/atomic"
 )
 
 type Event struct {
@@ -226,19 +227,23 @@ func newEventPool(capacity, avgEventSize int) *eventPool {
 		avgEventSize:     avgEventSize,
 		capacity:         capacity,
 		getMu:            &sync.Mutex{},
-		backCounter:      *atomic.NewInt64(int64(capacity)),
 		runHeartbeatOnce: &sync.Once{},
-		stopped:          atomic.NewBool(false),
-		slowWaiters:      atomic.NewInt64(0),
+		stopped:          &atomic.Bool{},
+		slowWaiters:      &atomic.Int64{},
 		wakeupInterval:   time.Second * 5,
 	}
 
+	eventPool.backCounter.Store(int64(capacity))
 	eventPool.getCond = sync.NewCond(eventPool.getMu)
 
-	for range capacity {
-		eventPool.free1 = append(eventPool.free1, *atomic.NewBool(true))
-		eventPool.free2 = append(eventPool.free2, *atomic.NewBool(true))
-		eventPool.events = append(eventPool.events, newEvent())
+	eventPool.free1 = make([]atomic.Bool, capacity)
+	eventPool.free2 = make([]atomic.Bool, capacity)
+	eventPool.events = make([]*Event, capacity)
+
+	for i := range capacity {
+		eventPool.free1[i].Store(true)
+		eventPool.free2[i].Store(true)
+		eventPool.events[i] = newEvent()
 	}
 
 	return eventPool
@@ -247,7 +252,7 @@ func newEventPool(capacity, avgEventSize int) *eventPool {
 const maxTries = 3
 
 func (p *eventPool) get(size int) *Event {
-	x := (p.getCounter.Inc() - 1) % int64(p.capacity)
+	x := (p.getCounter.Add(1) - 1) % int64(p.capacity)
 	var tries int
 	for {
 		if x < p.backCounter.Load() {
@@ -273,26 +278,26 @@ func (p *eventPool) get(size int) *Event {
 			})
 
 			// slowest path
-			p.slowWaiters.Inc()
+			p.slowWaiters.Add(1)
 			p.getMu.Lock()
 			p.getCond.Wait()
 			p.getMu.Unlock()
-			p.slowWaiters.Dec()
+			p.slowWaiters.Add(-1)
 			tries = 0
 		}
 	}
 	event := p.events[x]
 	p.events[x] = nil
 	p.free2[x].Store(false)
-	p.inUseEvents.Inc()
+	p.inUseEvents.Add(1)
 	event.stage = eventStageInput
 	event.Size = size
 	return event
 }
 
 func (p *eventPool) back(event *Event) {
 	event.stage = eventStagePool
-	x := (p.backCounter.Inc() - 1) % int64(p.capacity)
+	x := (p.backCounter.Add(1) - 1) % int64(p.capacity)
 	var tries int
 	for {
 		// fast path
@@ -318,7 +323,7 @@ func (p *eventPool) back(event *Event) {
 	p.resetEvent(event)
 	p.events[x] = event
 	p.free1[x].Store(true)
-	p.inUseEvents.Dec()
+	p.inUseEvents.Add(-1)
 	p.getCond.Broadcast()
 }
 
@@ -438,7 +443,7 @@ func (p *lowMemoryEventPool) get(size int) *Event {
 	getPool := p.pools[index]
 
 again:
-	inUse := int(p.inUseEvents.Inc())
+	inUse := int(p.inUseEvents.Add(1))
 	// Fast path: we're not over the capacity.
 	if inUse <= p.capacity {
 		e := getPool.Get().(*Event)
@@ -447,21 +452,21 @@ again:
 	}
 
 	// Slow path: wait until we fit in the capacity.
-	p.inUseEvents.Dec()
+	p.inUseEvents.Add(-1)
 
 	// Run heartbeat to periodically wake up goroutines that are waiting.
 	p.runHeartbeatOnce.Do(func() {
 		go p.wakeupWaiters()
 	})
 
 	// Wait until we fit in the capacity.
-	p.slowWaiters.Inc()
+	p.slowWaiters.Add(1)
 	p.getCond.L.Lock()
 	if !p.eventsAvailable() {
 		p.getCond.Wait()
 	}
 	p.getCond.L.Unlock()
-	p.slowWaiters.Dec()
+	p.slowWaiters.Add(-1)
 	goto again
 }
 
@@ -471,7 +476,7 @@ func (p *lowMemoryEventPool) back(event *Event) {
 
 	event.reset()
 	backPool.Put(event)
-	p.inUseEvents.Dec()
+	p.inUseEvents.Add(-1)
 	p.getCond.Broadcast()
 }