Make a crate of grpc_stream

[rtso]

Description

Clean up transaction_stream.rs' (renamed from grpc_stream.rs`) so we can make a crate of it, which will be used in the SDK.

TransactionStream

transaction_stream.rs defines a struct TransactionStream that has these methods:

• init_stream establishes a connection with the grpc stream
• get_next_transaction_batch fetches the next transaction response from grpc
• is_end_of_stream signals if you've reached the end of the stream
• reconnect_to_grpc will attempt to reconnect with grpc if there's an error with the stream
• reconnect_to_grpc_with_retries is a wrapper around reconnect_to_grpc that adds retry logic

How processors use TransactionStream

In worker.rs, fetch_transactions_from_transaction_stream will initialize and manage the TransactionStream, filtering transactions, sending the transactions to channel, handle reconnection if necessary.

Testing

cargo run --release -- -c config.yaml

[banool]

Apologies if I left comments for things that are just copy pasted. If this isn't the right place to address them, please open issues / a tracker item for those rather than fixing them here.

[banool]

Does it matter if they clash, don't they still mean the same thing?

[rtso]

Prometheus will try to register the metric with the same name and will error. I can't import the metrics from processor crate because of cyclic dep.

[CapCap]

I suggest renaming them and changing them- ideally each crate is responsible for emitting its own metrics; if anything, we can allow for a "metrics_prefix" and then allow dynamically creating them (instead of a single global lazy static) per instance of the streamer.

We can then create nice and specific metrics- maybe an "items received", and then a "filtered" either book or separate series, etc

Ie there should not be any overlap between crates and the SDK because the SDK will theoretically never have the same context on what's happening inside a given step as the step does (especially for complicated things like this)

[CapCap]

Ie we should move away from thinking "this is the processor GRPC client" and think of it as "this is the way anyone in any application, even python, should use our GRPC APIs, and will put it on crates.io and it'll go viral and the world will love us"- I find that generalized approach helps create nice, hermetic, batteries-included crates, vs crates that rely on specific behavior from other implementing crates for our own purposes (within reason ofc)

[CapCap]

Also duplicate metrics- ie the GRPC emitting channel size and also processor with the instrumented channel- it's fine :-)

[banool]

Got it, didn't realize you'd be using this crate alongside the old code and the registration itself would clash.

These clients support setting a prefix. If we don't do that today, I agree we should. Example from API Gateway: https://github.com/aptos-labs/api-gateway/blob/fa335677d1c80de9dc103396d563d4e47006a661/src/metrics/labels.rs#L37.

[banool]

Can we do this instead:

pub enum ProcessorStep {
    /// Received transactions from GRPC. Sending transactions to channel.
    ReceivedTxnsFromGrpc,
    /// Processor finished processing one batch of transaction
    ProcessedBatch,
    /// Processor finished processing multiple batches of transactions
    ProcessedMultipleBatches,
}

[banool]

Also what do we mean by "Sending transactions to channel.". Did that happen as part of this step or is that what happens next?

[rtso]

Sending transactions to channel is what happens next. This is copy/pasted from the processors crate. We'll redo all these metrics as part of SDK, so this is just throwaway for backwards compatibility for our dashboards.

[CapCap]

Can we kill this enum and giant, single overloaded logging method as a pattern? It's quite awful and confusing hahaha

[rtso]

giant, single overloaded logging method as a pattern

We can kill it in this crate. For more context, "giant, single overloaded logging method as a pattern" got introduced in the indexer grpc stack to reduce duplicated code and make sure we are logging all the same metrics in every relevant place. I'm open to other suggestions on how to do this :)

[banool]

I actually don't mind this enum ¯_(ツ)_/¯. I like it for the reason Renee mentions, it's easier to do things consistently. If anything I'd go further and wrap this in a struct that also contains all the relevant labels, and then some methods on that struct bump the relevant counts (so the counters become private to this file). You can see the API Gateway code for some inspiration on that: https://github.com/aptos-labs/api-gateway/blob/fa335677d1c80de9dc103396d563d4e47006a661/src/metrics/labels.rs#L134-L135. In this case we only bump the relevant metrics when the struct is dropped, but you could also have explicit functions for it (which makes more sense in this case).

[banool]

Should this 3k value be configurable? (with a default).