[client] introduce client-side event aggregation

[dmitri-netbird]

Describe your changes

This is the same as #6394, but using a branch in the main repo.

Issue ticket number and link

Stack

Checklist

• Is it a bug fix
• Is a typo/documentation fix
• Is a feature enhancement
• It is a refactor
• Created tests that fail without the change (if possible)
• This change does not modify the public API, gRPC protocols, functionality behavior, CLI / service flags, or introduce a new feature — OR I have discussed it with the NetBird team beforehand (link the issue / Slack thread in the description). See CONTRIBUTING.md.

By submitting this pull request, you confirm that you have read and agree to the terms of the Contributor License Agreement.

Documentation

Select exactly one:

• I added/updated documentation for this change
• Documentation is not needed for this change (explain why)

Docs PR URL (required if "docs added" is checked)

Paste the PR link from https://github.com/netbirdio/docs here:

netbirdio/docs#800

Summary by CodeRabbit

• New Features
  • Netflow/traffic events now include aggregation window timestamps (window_start, window_end) and counters (num_of_starts, num_of_ends, num_of_drops).
• Bug Fixes
  • Netflow event delivery is more reliable: unacknowledged events are tracked and automatically retried until acknowledged.
• API Changes
  • Management API schema/JSON responses now expose the new aggregation fields.
• Tests
  • Added unit and integration tests covering event aggregation, ack handling, and retry behavior.

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

This PR adds aggregating netflow storage, retry-and-ack delivery in the manager, new aggregation fields across protobuf and HTTP schemas, and a mutex unlock adjustment in the GRPC client send path.

Changes

Netflow aggregation and reliable delivery

Layer / File(s)	Summary
Core type and interface changes for aggregation client/internal/netflow/types/types.go	Event gains WindowStart/WindowEnd, EventFields gains NumOfStarts/NumOfEnds/NumOfDrops, Event.Clone() is added, and FlowLogger now uses FlowEventAggregator and AggregatingStore.
Aggregating memory store implementation and tests client/internal/netflow/store/memory.go, client/internal/netflow/store/event_aggregation_test.go	AggregatingMemory tracks aggregation windows, groups events by aggregation key, resets windows, and is covered by aggregation and reset tests.
Logger wiring to aggregating store client/internal/netflow/logger/logger.go	Logger.Store switches to types.AggregatingStore, New() constructs the aggregating memory store, and ResetAggregationWindow() delegates to the store.
Manager retry-and-ack delivery client/internal/netflow/manager.go, client/internal/netflow/manager_integration_test.go	Manager buffers unacked events, sends aggregated events, retries with backoff, extends proto conversion, and is covered by send/ack/retry integration tests.
Proto and OpenAPI schema updates flow/proto/flow.proto, shared/management/http/api/openapi.yml, shared/management/http/api/types.gen.go	FlowEvent, FlowFields, and NetworkTrafficEvent add window and aggregation-counter fields.

Flow GRPC client locking

Layer / File(s)	Summary
Send lock scope adjustment flow/client/client.go	GRPCClient.Send now defers unlocking its mutex until function exit after capturing the stream reference.

Estimated code review effort: 4 (Complex) | ~60 minutes

Poem

I hop through windows, round and bright,
Count the starts and drops just right.
Acks come back, retries take flight,
Tiny flows hum through the night,
🐇✨ All snug in aggregating light.

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (1 warning, 1 inconclusive)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 0.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Description check	❓ Inconclusive	The template is only partially filled: the change description is vague, and the issue ticket and stack sections are left blank.	Add a concrete change summary, the issue ticket/link, the branch stack, and clarify the checklist/documentation entries.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title is concise and accurately summarizes the main change: adding client-side event aggregation.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch dmitri-event-aggregation

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[coderabbitai]

Actionable comments posted: 4

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@client/internal/netflow/manager.go`:
- Around line 223-232: The event retry queue update in the flow sending path is
happening after the first send attempt, which allows a fast ACK to delete the
event before it is tracked. Update the logic in the event loop around m.send,
m.eventsWithoutAcks.StoreEvent, and the related retry path at the other
mentioned block so each event is stored in eventsWithoutAcks before any send
attempt. Keep the existing send/logging behavior in send and the ACK cleanup in
receiveACKs/DeleteEvents unchanged, but ensure the queue insertion happens first
for every event.
- Around line 257-287: The retry path in manager.go is creating a backoff timer
in the send loop, but that value is immediately replaced by the fixed retry
interval, so the exponential backoff never takes effect. Update the retry logic
in the event loop around m.send, retryBackoff.NextBackOff, and m.retryInterval
so failures actually wait using the backoff delay, and only reset to the normal
retry interval after a successful pass; also avoid creating a timer that gets
overwritten without being used.
- Around line 118-130: `startSender()` and `startRetries()` can both reach
`m.send()` at the same time, so writes to the flow stream are still concurrent
even though `GRPCClient.Send()` only protects access to `c.stream`. Update the
send path in the netflow manager to serialize the actual `stream.Send()` call,
either by adding a dedicated mutex around `m.send()`/`GRPCClient.Send()` or by
funneling all outbound flow messages through a single writer goroutine; use the
`startSender`, `startRetries`, and `m.send` symbols to place the fix.

In `@client/internal/netflow/store/memory.go`:
- Around line 73-78: The reset snapshot returned by ResetAggregationWindow() is
missing rnd, which can cause a nil dereference later when GetAggregatedEvents()
handles unknown protocols. Update the snapshot creation in
ResetAggregationWindow() so the returned AggregatingMemory carries over or
initializes rnd before aggregation continues, and make sure the same fix is
applied in the other reset path referenced by the comment.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: f351fb17-db1c-475a-81c5-52f293b3d3f8

📥 Commits

Reviewing files that changed from the base of the PR and between 2ab99ee and b684f85.

⛔ Files ignored due to path filters (2)

• flow/proto/flow.pb.go is excluded by !**/*.pb.go
• flow/proto/flow_grpc.pb.go is excluded by !**/*.pb.go

📒 Files selected for processing (9)

• client/internal/netflow/logger/logger.go
• client/internal/netflow/manager.go
• client/internal/netflow/manager_integration_test.go
• client/internal/netflow/store/event_aggregation_test.go
• client/internal/netflow/store/memory.go
• client/internal/netflow/types/types.go
• flow/proto/flow.proto
• shared/management/http/api/openapi.yml
• shared/management/http/api/types.gen.go

[github-actions]

Release artifacts

Built for PR head c160fdd in workflow run #16509.

Artifact	Link
All release artifacts	Download
Linux packages	Download
Windows packages	Download
macOS packages	Download
UI artifacts	Download
UI macOS artifacts	Download

GHCR images (amd64)

• ghcr.io/netbirdio/management:pr-6627
• ghcr.io/netbirdio/netbird-server:pr-6627
• ghcr.io/netbirdio/netbird:pr-6627
• ghcr.io/netbirdio/relay:pr-6627
• ghcr.io/netbirdio/reverse-proxy:pr-6627
• ghcr.io/netbirdio/signal:pr-6627
• ghcr.io/netbirdio/upload:pr-6627

This comment is updated by the Release workflow. Artifact links expire according to the workflow retention policy.

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@client/internal/netflow/manager.go`:
- Line 287: Fix the codespell failure by correcting the typo in the comment
inside the netflow retry logic, where resetBackoff is used; update the word
“absense” to the proper spelling in that conditional comment so the surrounding
manager.go code passes spell checking.
- Around line 281-289: Guard the retry timer setup in `Manager` so
`time.NewTimer` is never given the `backoff.Stop` value from
`retryBackoff.NextBackOff()`. In the retry path inside `m.send(e)`, check the
returned duration before creating the timer; if the backoff is exhausted, fall
back to `m.retryInterval` instead of using the stopped backoff value. Keep the
change localized to the retry loop logic and the `retryBackoff`/`resetBackoff`
handling in `Manager`.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro

Run ID: 1e3d7165-57f6-4526-9c82-cfcb834d6ced

📥 Commits

Reviewing files that changed from the base of the PR and between b684f85 and fb22695.

📒 Files selected for processing (3)

• client/internal/netflow/manager.go
• client/internal/netflow/store/memory.go
• flow/client/client.go

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud