diff --git a/client/internal/netflow/manager.go b/client/internal/netflow/manager.go index ba07c8522..06e644f4f 100644 --- a/client/internal/netflow/manager.go +++ b/client/internal/netflow/manager.go @@ -223,12 +223,12 @@ func (m *Manager) startSender(ctx context.Context, flowConfigInterval time.Durat collectedEvents := m.logger.ResetAggregationWindow() events := collectedEvents.GetAggregatedEvents() for _, event := range events { + m.eventsWithoutAcks.StoreEvent(event) if err := m.send(event); err != nil { log.Errorf("failed to send flow event to server: %v", err) } else { log.Tracef("sent flow event: %s", event.ID) } - m.eventsWithoutAcks.StoreEvent(event) } } } @@ -271,6 +271,7 @@ func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Dura case <-ctx.Done(): return case <-timer.C: + resetBackoff := true for _, e := range m.eventsWithoutAcks.GetEvents() { if e.Timestamp.Add(time.Second).After(time.Now()) { // grace period on retries to avoid early retries @@ -278,12 +279,15 @@ func (m *Manager) startRetries(ctx context.Context, flowConfigInterval time.Dura continue } if err := m.send(e); err != nil { - timer = time.NewTimer(retryBackoff.NextBackOff()) //nolint:staticcheck,wastedassign + timer = time.NewTimer(retryBackoff.NextBackOff()) + resetBackoff = false break } } - retryBackoff.Reset() - timer = time.NewTimer(m.retryInterval) + if resetBackoff { // use regular retry interval in absense of network errors + retryBackoff.Reset() + timer = time.NewTimer(m.retryInterval) + } } } } diff --git a/client/internal/netflow/store/memory.go b/client/internal/netflow/store/memory.go index 0343b15c8..a34e4be63 100644 --- a/client/internal/netflow/store/memory.go +++ b/client/internal/netflow/store/memory.go @@ -70,7 +70,7 @@ func (am *AggregatingMemory) ResetAggregationWindow() types.FlowEventAggregator defer am.mux.Unlock() now := time.Now() - toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}} + toret := AggregatingMemory{WindowStart: am.WindowStart, WindowEnd: now, Memory: Memory{events: am.events}, rnd: v2.NewPCG(rand.Uint64(), rand.Uint64())} am.events = make(map[uuid.UUID]*types.Event) am.WindowStart = now diff --git a/flow/client/client.go b/flow/client/client.go index 180a4b441..3f31c2464 100644 --- a/flow/client/client.go +++ b/flow/client/client.go @@ -109,7 +109,7 @@ func (c *GRPCClient) Close() error { func (c *GRPCClient) Send(event *proto.FlowEvent) error { c.mu.Lock() stream := c.stream - c.mu.Unlock() + defer c.mu.Unlock() // stream.Send() is not safe to call concurrently from multiple goroutines if stream == nil { return errors.New("stream not initialized")