ref: Move rust-arroyo from snuba to arroyo #323

untitaker · 2024-01-17T12:37:59Z

Move ./rust_snuba/rust_arroyo/ from snuba into arroyo, preserving history.

Steps used to preserve history:

git remote add snuba git@github.com:getsentry/snuba
git checkout -b ref/add-rust-arroyo
git reset --hard snuba/master
git filter-branch --subdirectory-filter rust_snuba/rust_arroyo/ -- --all
mkdir rust-arroyo
mv Cargo.toml examples/ src/ rust-arroyo/
git add ...
git commit -am 'move arroyo to subdirectory'
git merge --allow-unrelated-histories main

maybe we should not squash-merge this one? unsure.

this PR doesn't contain any other code changes. followup PRs include:

ref(ci): Add Rust CI #324

@dbanda

Experimental project that will allow us to develop and test Rust consumers alongside the existing Python ones. This is entirely based off @dbanda's branch `dbanda/rust-consumer` except with all of the code except `rust_arroyo` removed. The `rust_arroyo` crate is an exact copy https://github.com/getsentry/rust_arroyo, which will allow us to use it here without publishing a separate library yet.

Arroyo is a bit finnicky about the types of configuration you are supposed to use with it, let's use the `new_consumer_config` way

`Position` struct is no longer needed. It was a hack used for the legacy Snuba consumer, and has been removed in Python Arroyo: #165. We are removing it here to bring the two libraries in line.

…3885) Same as we did in Python Arroyo #134. The processing strategy interface needs to support messages that represent a batch or a filter message - i.e. not necessarily a 1:1 relationship between messages from the Kafka broker and a message passed to the processing strategy. This means a message can be associated with a number of partitions and offsets, not just a single one.

Follw up to getsentry/snuba#3883. commit_positions and stage_positions should be called commit_offsets/ stage_offset like in Python's Arroyo library.

* merge in produce example * rename hash password to reverse string. more accurately describes strategy * add using futures in the queue * handle future * use tokio runtime

Tests are currently broken - this fixes them and runs in CI so we don't break them again

I think this test was flaky because it used the same topic as the test that preceded it. Seems to do a lot better with a different name.

Currently the consumer does not do any batching, add a reduce strategy to help with this.

Now it passes batches to the fake ClickHouse writer step instead of one message at a time. Depends on getsentry/snuba#4061

* unlock mutex to avoid deadlock * use take to make linter happy

Now ClickHouse is available in Rust CI so we can test writing to ClickHouse functionality. We are also running Rust CI runs on all branches now. This is needed since Rust consumer calls Python so a change in Python can potentially break the Rust consumer.

feat(rust-consumer): Add basic clickhouse strategy Same idea as getsentry/snuba#4437 with a few differences: - Does not switch the arroyo interface to async - Does not write to ClickHouse by default (only if --no-skip-write is passed) - Batches inserted rows into fewer writes with multiple rows

There's no point having in the clickhouse_client living in the arroyo package since it's only by the ClickHouse strategy in Snuba which is not part of Arroyo.

Break the task scheduling code out of the Clickhouse writer strategy so it can be used for parallel execution of any functions, such as custom message processors. The ClickHouse strategy now uses RunTaskInThreads under the hood.

Also refactor rust CI setup to work with Snuba installed

Fixes failing CI since Arroyo dependencies are not currently pinned.

* feat(rust): Multiprocessing with Python message processors Spawn multiple subprocesses in Rust, and run the Python message processor in each subprocess By default there is only one process. * add tests and fix a ton of bugs * fix typing * fix rust linting * just use a real processor lol

…4655)" This reverts commit bb1e799dcb8a6b03ece9ef49051737105bf5c8ba. Co-authored-by: lynnagara <1779792+lynnagara@users.noreply.github.com>

…tempt (#4673) * Revert "Revert "feat(rust): Multiprocessing with Python message processors (#4655)"" This reverts commit 2a45cdd4a55d84c5b9bfbf684b85aa214cdb37f9. * run without process overhead if only one process * do not commit broken code to master, impossible challenge * attempt to fix keyerror in CI * fix tests on macos * fix python typing and rustfmt

Add statsd and testing metric backends to rust_snuba. Remove the metric implementation and dependency on cadence from Arroyo - users of Arroyo should provide the trait implementation. Pass the dogstatsd host and port from Python to Rust via config. Also start logging one metric ("arroyo.consumer.join.time") to verify it works.

* wip * fix up tests and remove useless seek * clippy and fmt

… 1 (#5284)

* ref(rust): Remove explicit buffering of gauges statsdproxy should be able to buffer gauges efficiently * fmt

…5294)" This reverts commit 0409b3f45fd65f89885d9be6bf048b16416fb70d. Co-authored-by: untitaker <837573+untitaker@users.noreply.github.com>

* add testcase * fix assertion

Especially in benchmarks, creating a fresh runtime for every run leads to thousands of threads being spawned which makes local profiling difficult

That function is interning the topic name, and thus aquiring a Mutex. If we keep a list of subscribed topics locally, we avoid that and can also guard against errors.

…og implementation (#5311) * ref(devserver): It's all Rust now Move all consumers to use Rust consumers, be it either using the Rust message processors or using the hybrid mode to execute the existing processing functions written in Python. * fixup * remove debug logging too * fix benchmark * also run full sentry testsuite when devserver changes * add no-skip-write * fix commitlog implementation * formatting and fix benchmarks * fmt * fix arroyo test

* fix: Remove any panics in threads This causes a lot of double-erroring, and at least within RunTaskInThreads we can carry forward errors normally. We still have to remove panics in the main thread, but this should already cut down a few duplicate errors we are seeing today. * fix tests

* Add Metrics impl based on `merni` I pretty much copy-pasted the core `merni` code, which is not that much. Also built shims to statsdproxy, and updated all usages of the previous `Metrics` trait and `get_metrics` to use the global macros. * Update rust_snuba/rust_arroyo/src/metrics/globals.rs Co-authored-by: Sebastian Zivota <loewenheim@users.noreply.github.com> --------- Co-authored-by: Sebastian Zivota <loewenheim@users.noreply.github.com>

…ion tag (#5346) * ref(metrics): Refactor how global tags work, and introduce min_partition tag Port the min_partition tag from snuba's python consumer. In order to do that, we need a way to mutate global tags. So we're now using a custom global tags implementation based on statsdproxy instead of cadence's global tags support. This might be more efficient than cadence's global tags because adding them to a metric happens after aggregation, but i haven't measured it. Additionally, I found that our strategy factory's "create" function doesn't receive partitions. I don't want to add that argument though because doing so has been really painful for optimizing rebalancing performance in the past. Instead I propose to split up the API into two parts: a callback to recreate strategies, and a callback to update the currently assigned partitions. depends on getsentry/statsdproxy#42 * bump statsdproxy * fix formatting and benchmarks * bump statsdproxy to crates.io version * add basic test * better test, copypasted from statsdproxy * make private

* add rust concurrency metric * share metrics prefix * move metric strategy_name to self

loewenheim · 2024-01-17T12:48:03Z

I would agree with not squashing this.

lynnagara and others added 30 commits February 22, 2023 13:17

move examples to right folder (#3772)

715d749

wip: Scaffolding for generic Rust consumer (#3866)

5fe2f74

feat(rust-consumer): Remove option to build raw config (#3870)

cb2a998

Arroyo is a bit finnicky about the types of configuration you are supposed to use with it, let's use the `new_consumer_config` way

feat: The Rust consumer calls Python message processors (#3871)

4b5b36b

feat(rust-arroyo): Get rid of Position (#3883)

63ffe67

`Position` struct is no longer needed. It was a hack used for the legacy Snuba consumer, and has been removed in Python Arroyo: #165. We are removing it here to bring the two libraries in line.

ref: Rename positions to offsets everywhere (#3888)

e57a192

Follw up to getsentry/snuba#3883. commit_positions and stage_positions should be called commit_offsets/ stage_offset like in Python's Arroyo library.

rust: add simple transform and produce strategy example (#3796)

b262ff6

* merge in produce example * rename hash password to reverse string. more accurately describes strategy * add using futures in the queue * handle future * use tokio runtime

ci: Run Rust linter (#3884)

ae4dadc

feat(rust-consumer): Commit offsets every second (#3919)

aacaabd

feat(rust-consumer): Handle shutdown signal (#4034)

7bce9e2

fix(rust-consumer): Stop spamming the log with messages (#4039)

3e01fb5

fix: Fix Rust tests and run in CI (#4042)

fda75fb

Tests are currently broken - this fixes them and runs in CI so we don't break them again

ci(rust): Fix rust-arroyo linting errors and run check in CI (#4064)

1f464a8

test: Fix flaky Rust consumer tests (#4067)

73a988c

I think this test was flaky because it used the same topic as the test that preceded it. Seems to do a lot better with a different name.

feat(rust-arroyo): Reduce strategy (#4061)

e25c825

Currently the consumer does not do any batching, add a reduce strategy to help with this.

feat(rust-consumer): Add batching step to the Rust consumer (#4066)

4edae8a

Now it passes batches to the fake ClickHouse writer step instead of one message at a time. Depends on getsentry/snuba#4061

rust: unlock mutex to avoid deadlock (#4393)

1076ba7

* unlock mutex to avoid deadlock * use take to make linter happy

ref(rust-consumer): Move clickhouse_client out of rust_arroyo (#4499)

c004468

There's no point having in the clickhouse_client living in the arroyo package since it's only by the ClickHouse strategy in Snuba which is not part of Arroyo.

feat(rust-consumer): Get runtime config from rust (#4551)

0e0aad1

Also refactor rust CI setup to work with Snuba installed

build: chrono 0.4.26 and remove use of deprecated function (#4652)

2e575a2

Fixes failing CI since Arroyo dependencies are not currently pinned.

Revert "feat(rust): Multiprocessing with Python message processors (#…

6bfd19f

…4655)" This reverts commit bb1e799dcb8a6b03ece9ef49051737105bf5c8ba. Co-authored-by: lynnagara <1779792+lynnagara@users.noreply.github.com>

feat(rust-arroyo): Implement join/terminate for RunTaskInThreads (#4676)

edddb17

untitaker and others added 19 commits December 28, 2023 19:01

fix(rust): pause/resume bugfixes (#5253)

4d7a124

* wip * fix up tests and remove useless seek * clippy and fmt

fix(rust): Remove print stmt (#5269)

74ec6cf

feat(rust): Add number of rows processed to the batch size instead of…

a5087b8

… 1 (#5284)

ref(rust): Do not do extra work when merging if not needed (#5294)

b86f377

ref(rust): Remove explicit buffering of gauges (#5293)

bf2b063

* ref(rust): Remove explicit buffering of gauges statsdproxy should be able to buffer gauges efficiently * fmt

Revert "ref(rust): Do not do extra work when merging if not needed (#…

a6dacae

…5294)" This reverts commit 0409b3f45fd65f89885d9be6bf048b16416fb70d. Co-authored-by: untitaker <837573+untitaker@users.noreply.github.com>

fix(rust): add testcase for empty batches (#5299)

a1ba62a

* add testcase * fix assertion

ref(rust): Log actual error if strategy panics (#5317)

5e132fe

Reuse Tokio Handle in DlqPolicy (#5329)

d67673c

Especially in benchmarks, creating a fresh runtime for every run leads to thousands of threads being spawned which makes local profiling difficult

Avoid calling Topic::new for every received Message (#5331)

d373819

That function is interning the topic name, and thus aquiring a Mutex. If we keep a list of subscribed topics locally, we avoid that and can also guard against errors.

deps(rust): Change rdkafka dep to upstream master (#5386)

f16e23d

ref(rust): Don't panic in RunTaskInThreads::poll (#5387)

dfafaa5

rust: add rust concurrency metric (#5341)

bac32f3

* add rust concurrency metric * share metrics prefix * move metric strategy_name to self

move rust-arroyo to subdirectory

9abd81f

Merge branch 'main' into ref/add-rust-arroyo

3690356

untitaker requested a review from a team as a code owner January 17, 2024 12:37

untitaker changed the title ~~ref/add rust arroyo~~ ref: Move rust-arroyo from snuba to arroyo Jan 17, 2024

loewenheim approved these changes Jan 17, 2024

View reviewed changes

lynnagara approved these changes Jan 17, 2024

View reviewed changes

untitaker merged commit 0a20c30 into main Jan 17, 2024
9 checks passed

untitaker deleted the ref/add-rust-arroyo branch January 17, 2024 19:50

untitaker mentioned this pull request Jan 17, 2024

Revert "ref: Move rust-arroyo from snuba to arroyo" #325

Merged

untitaker restored the ref/add-rust-arroyo branch January 17, 2024 19:51

untitaker mentioned this pull request Jan 17, 2024

ref: Add rust arroyo, second attempt #326

Merged

untitaker deleted the ref/add-rust-arroyo branch January 17, 2024 19:53

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ref: Move rust-arroyo from snuba to arroyo #323

ref: Move rust-arroyo from snuba to arroyo #323

untitaker commented Jan 17, 2024 •

edited

Loading

loewenheim commented Jan 17, 2024

ref: Move rust-arroyo from snuba to arroyo #323

ref: Move rust-arroyo from snuba to arroyo #323

Conversation

untitaker commented Jan 17, 2024 • edited Loading

loewenheim commented Jan 17, 2024

untitaker commented Jan 17, 2024 •

edited

Loading