feat: common flusher for reporting with mtu handler

[satishrudderstack]

Description

• In this pr, along with mtu flushing support, we are also adding a common flusher module.

Motivation for common flusher

• We have multiple outbox tables on Rudder Server which needs to be flushed to our central Reporting Service. Currently we have multiple flushers with duplicate code.

Flusher

• CronRunner: Invokes Flusher's Flush() function periodically and if there's lag increase, it calls Flush() aggressively
• Flusher: Core flush logic. It uses db and and a aggregator. This is like common platform for flushing reports from different outbox tables
• Aggregator: We can have in-app or in-db aggregators for each table. This takes start, end times as input and returns the aggregated reports as jsons

Flush steps

1. Get time range (start, end) to flush. start will be from db. end will be min(now() - 30s, end-of-current-hour). We don't want to flush reports where inserts are happening. We always try to flush full window durations which is configured via flushWindow unless window is crossing multiple hour boundaries
2. Get aggregated reports. Aggregation can be in-app or in-db
3. Send the aggregated reports to Reporting Service using go routines
4. Delete the reports in the range

Features

• Flush/Aggregation window as configuration. We can increase this window to decrease load on Reporting Service
• Aggregation can be done in-app or in db
• Ability to flush aggressively if lag increases. This is done by not sleeping and increasing the concurrency of go-routines used to make calls to Reporting Service
• Batching of payloads to Reporting Service

Linear Ticket

https://linear.app/rudderstack/issue/OBS-504/reporting-client-for-tracked-users completes OBS-504

Security

• [ ] The code changed/added as part of this pull request won't create any security issues with how the software is being used.

[coderabbitai[bot]]

[!IMPORTANT]

Review skipped

Auto reviews are disabled on this repository.

Please check the settings in the CodeRabbit UI or the .coderabbit.yaml file in this repository. To trigger a single review, invoke the @coderabbitai review command.

You can disable this status message by setting the reviews.review_status to false in the CodeRabbit configuration file.

---

Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media?

Share

- [X](https://twitter.com/intent/tweet?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A&url=https%3A//coderabbit.ai) - [Mastodon](https://mastodon.social/share?text=I%20just%20used%20%40coderabbitai%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20the%20proprietary%20code.%20Check%20it%20out%3A%20https%3A%2F%2Fcoderabbit.ai) - [Reddit](https://www.reddit.com/submit?title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&text=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code.%20Check%20it%20out%3A%20https%3A//coderabbit.ai) - [LinkedIn](https://www.linkedin.com/sharing/share-offsite/?url=https%3A%2F%2Fcoderabbit.ai&mini=true&title=Great%20tool%20for%20code%20review%20-%20CodeRabbit&summary=I%20just%20used%20CodeRabbit%20for%20my%20code%20review%2C%20and%20it%27s%20fantastic%21%20It%27s%20free%20for%20OSS%20and%20offers%20a%20free%20trial%20for%20proprietary%20code)

Tips

### Chat There are 3 ways to chat with [CodeRabbit](https://coderabbit.ai): - Review comments: Directly reply to a review comment made by CodeRabbit. Example: - `I pushed a fix in commit .` - `Generate unit testing code for this file.` - `Open a follow-up GitHub issue for this discussion.` - Files and specific lines of code (under the "Files changed" tab): Tag `@coderabbitai` in a new review comment at the desired location with your query. Examples: - `@coderabbitai generate unit testing code for this file.` - `@coderabbitai modularize this function.` - PR comments: Tag `@coderabbitai` in a new PR comment to ask questions about the PR branch. For the best results, please provide a very specific query, as very limited context is provided in this mode. Examples: - `@coderabbitai generate interesting stats about this repository and render them as a table.` - `@coderabbitai show all the console.log statements in this repository.` - `@coderabbitai read src/utils.ts and generate unit testing code.` - `@coderabbitai read the files in the src/scheduler package and generate a class diagram using mermaid and a README in the markdown format.` - `@coderabbitai help me debug CodeRabbit configuration file.` Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. ### CodeRabbit Commands (invoked as PR comments) - `@coderabbitai pause` to pause the reviews on a PR. - `@coderabbitai resume` to resume the paused reviews. - `@coderabbitai review` to trigger an incremental review. This is useful when automatic reviews are disabled for the repository. - `@coderabbitai full review` to do a full review from scratch and review all the files again. - `@coderabbitai summary` to regenerate the summary of the PR. - `@coderabbitai resolve` resolve all the CodeRabbit review comments. - `@coderabbitai configuration` to show the current CodeRabbit configuration for the repository. - `@coderabbitai help` to get help. Additionally, you can add `@coderabbitai ignore` anywhere in the PR description to prevent this PR from being reviewed. ### CodeRabbit Configration File (`.coderabbit.yaml`) - You can programmatically configure CodeRabbit by adding a `.coderabbit.yaml` file to the root of your repository. - Please see the [configuration documentation](https://docs.coderabbit.ai/guides/configure-coderabbit) for more information. - If your editor has YAML language server enabled, you can add the path at the top of this file to enable auto-completion and validation: `# yaml-language-server: $schema=https://coderabbit.ai/integrations/schema.v2.json` ### Documentation and Community - Visit our [Documentation](https://coderabbit.ai/docs) for detailed information on how to use CodeRabbit. - Join our [Discord Community](https://discord.com/invite/GsXnASn26c) to get help, request features, and share feedback. - Follow us on [X/Twitter](https://twitter.com/coderabbitai) for updates and announcements.

[codecov[bot]]

Codecov Report

Attention: Patch coverage is 69.84127% with 114 lines in your changes missing coverage. Please review.

Project coverage is 74.35%. Comparing base (d263cc5) to head (e3425f7).

Files	Patch %	Lines
enterprise/reporting/flusher/flusher.go	68.25%	33 Missing and 27 partials :warning:
...eporting/flusher/aggregator/tracked_users_inapp.go	68.75%	10 Missing and 10 partials :warning:
...se/reporting/flusher/aggregator/aggregator_mock.go	0.00%	15 Missing :warning:
enterprise/reporting/flusher/cron_runner.go	88.33%	5 Missing and 2 partials :warning:
enterprise/reporting/flusher/factory.go	72.00%	4 Missing and 3 partials :warning:
enterprise/reporting/mediator.go	76.92%	2 Missing and 1 partial :warning:
enterprise/reporting/flusher/aggregator/nop.go	0.00%	2 Missing :warning:

Additional details and impacted files

```diff @@ Coverage Diff @@ ## master #4823 +/- ## ========================================== - Coverage 74.37% 74.35% -0.03% ========================================== Files 420 427 +7 Lines 49232 49610 +378 ========================================== + Hits 36617 36886 +269 - Misses 10214 10284 +70 - Partials 2401 2440 +39 ```

:umbrella: View full report in Codecov by Sentry.
:loudspeaker: Have feedback on the report? Share it here.

[snarkychef]

How do we ensure atomicity of steps 3 & 4 in the above diagram? viz. send aggregated reports to reporting service and delete sent reports?

As per my understanding, we used to ensure idempotency using timestamp along with grouping labels as report id. Just wondering how it's affected with the introduction of Flush/Aggregation window.

[satishrudderstack]

How do we ensure atomicity of steps 3 & 4 in the above diagram? viz. send aggregated reports to reporting service and delete sent reports?

As per my understanding, we used to ensure idempotency using timestamp along with grouping labels as report id. Just wondering how it's affected with the introduction of Flush/Aggregation window.

On Reporting Service, we are planning to have hash id generated from the payload for each report and have it as a id column on Timescale table. If we receive the same report on Reporting Service, we ignore the report and respond with 200. We have similar logic for metrics table

From Rudder Server side, we are calculating the flush window first and we are deleting all the reports in the flush window if all the aggregated reports were all sent successfully. In the aggregated report, we have min(reported_at) for the group as timestamp which will be sent to reporting.

Now, for the case where some reports were sent and there is an error for some reports while sending, we error out and don't delete the reports. We will retry the entire window later and reporting will ignore the reports which it consumed already.

FYI, we don't consider most recent data where inserts are happening. We have config recentExclusionWindowInSeconds

Nice point @snarkychef . I have added some code so that our flush window doesn't change on retries