Consumer supplied filters when producing, along with a producer filter #36

huntc · 2023-10-05T09:47:43Z

When producing from the edge, the consumer may also supply filters. The changes here implement that by providing a watch channel for the upstream gRPC producer. Please start with the sample temperature_production.rs file changes; the addition of conveying a criteria watch channel is consistent with when the edge is consuming, although in this instance it cannot be optional. We do not know if the remote consumer will be supplying criteria or not, hence the channel being mandatory.

A producer filter is also catered for.

Filters are applied prior to any event transformation and intentionally result in seq_nr gaps when events go unmatched. The consuming side has previously been enhanced to expect these gaps.

TODO:

Implement the consumer_filter matches function
Tests for the consumer_filter matches function

Questions/thoughts/issues:

~~We can add a filter parameter to the grpc_producer.handler(consumer_filters_receiver, transformer) method if we want, or just have the transformer double-up as a producer-side filter.~~
~~It appears that a consumer can only send a filter upon starting the protocol, and not subsequently. Is this correct?~~
~~We don't presently support tags in akka-edge-rs, and so we can't filter against them. Should we support tags? (I recall an early conversation that we wouldn't, but I may have misheard that).~~
I've just noticed that the JVM code holds the notion of merging and diffing filters. I'll need more of an explanation on that. Right now, we simply merge filter commands with the current filter state, and remember this filter state across connections.
~~I haven't been able to find tests that test out the filter matching.~~

Fixes #25

When producing from the edge, the consumer may also supply filters. The changes here implement that by providing a watch channel for the upstream gRPC producer.

patriknw · 2023-10-05T13:53:15Z

examples/iot-service/backend/src/temperature_production.rs

@@ -28,6 +28,7 @@ pub async fn task(
    // Establish a sink of envelopes that will be forwarded
    // on to a consumer via gRPC event producer.

+    let (consumer_filters, consumer_filters_receiver) = watch::channel(vec![]);
    let (grpc_producer, grpc_producer_receiver) = mpsc::channel(10);

    let grpc_producer =


unrelated, isn't it confusing to reuse the same name grpc_producer for a new thing, or is that a rust thing?

I’d say that it is a more general personal style thing. Rust linting via clippy has no problem with it. Clippy is king. Perhaps I could use a different name here though.

Commit 8730696 deals with this.

examples/iot-service/backend/src/temperature_production.rs

patriknw · 2023-10-06T06:05:17Z

It appears that a consumer can only send a filter upon starting the protocol, and not subsequently. Is this correct?

That is correct. We have not implemented dynamic consumer filter on jvm side for this case. We might do that later, but currently it can only be set at start.

patriknw · 2023-10-06T06:14:51Z

We don't presently support tags in akka-edge-rs, and so we can't filter against them. Should we support tags? (I recall an early conversation that we wouldn't, but I may have misheard that).

We have found that tags is a pretty good way to filter events (consumer defined filters, evaluated on producer side). Without tags those filters can only be on entity ids (exact or regex). I think we have been and still are uncertain about how useful consumer defined filters are for the case when edge is producer. I guess the consumer would be interested in all events from the edge in most cases, and the edge itself decides what to publish (producer defined filter). I would say that tag filters can be added when we have more important things in place. Also depends on effort, if it's a small thing to add tagging support we can just do it.

Also note that topic filters is actually a specific type of tag filters https://doc.akka.io/docs/akka-projection/snapshot/grpc.html#topics
Topic is not the same concept as a streambed topic.

patriknw · 2023-10-06T06:36:37Z

We can add a filter parameter to the grpc_producer.handler(consumer_filters_receiver, transformer) method if we want, or just have the transformer double-up as a producer-side filter.

What's your opinion on this @johanandren ?
For producer push I can agree that transformation and producerFilter looks very similar and we will handle them in same way (for -rs we don't have to send the FilteredEvent since gaps will be filled on other side)

johanandren · 2023-10-06T07:14:50Z

I'll just copy-paste what I said in Slack:

I think it would be good if the API aligns with the Akka side and has that (producer filter) as well, to figure out that something called transformer can also be used for filtering is also a form of complexity to push to the developer IMO. The producer filter is optional so shouldn't be needed to define (or think about) unless the dev wants to.

huntc · 2023-10-06T07:59:12Z

I'll just copy-paste what I said in Slack:

I think it would be good if the API aligns with the Akka side and has that (producer filter) as well, to figure out that something called transformer can also be used for filtering is also a form of complexity to push to the developer IMO. The producer filter is optional so shouldn't be needed to define (or think about) unless the dev wants to.

Do you feel the same now looking at the code?

johanandren · 2023-10-06T09:26:46Z

Yes, but I wonder if we could make all of the optional handler parameters a single setup-kinda thing, so that for the simplest case it would be grpc_flow.handler(None) or something like a factory with an empty default grpc_flow.handler(options_empty) and then you could build from that for the case where you want filtering grpc_flow.handler(options_empty.with_producer_filter(...).with_event_transform(transform)

That would somewhat align with the EventProducerSource we use for setup in the Akka JVM side of things.

The general design idea is easy discoverability of opt-ins, but with no mental overhead needing to know all possible details when not used.

huntc · 2023-10-06T09:31:23Z

Yes, but I wonder if we could make all of the optional handler parameters a single setup-kinda thing, so that for the simplest case it would be grpc_flow.handler(None) or something like a factory with an empty default grpc_flow.handler(options_empty) and then you could build from that for the case where you want filtering grpc_flow.handler(options_empty.with_producer_filter(...).with_event_transform(transform)

That would somewhat align with the EventProducerSource we use for setup in the Akka JVM side of things.

The general design idea is easy discoverability of opt-ins, but with no mental overhead needing to know all possible details when not used.

I'll have a think on that a bit more... you can't have default params or overloaded functions in Rust, but you can have constrructors that return Self and then have a with_ style thing.

Note quite working yet.

johanandren · 2023-10-06T09:39:47Z

If it's not too far from what Rust devs would expect ofc. my suggestions was mostly coming from being used to Scala/Java. It could be possible to do nicer in some very different way in rust, but the aim of not having to know all options up front, but easily being able to discover them, I think is more general.

By conveying a PersistenceId instead of an EntityId for envelopes, we can reduce the places where we need to declare the event type.

huntc · 2023-10-07T04:47:05Z

It appears that a consumer can only send a filter upon starting the protocol, and not subsequently. Is this correct?

That is correct. We have not implemented dynamic consumer filter on jvm side for this case. We might do that later, but currently it can only be set at start.

Great. FYI, the code here will permit the filter to be updated in the future (it was easy and natural to cater for this).

huntc · 2023-10-07T04:49:22Z

We don't presently support tags in akka-edge-rs, and so we can't filter against them. Should we support tags? (I recall an early conversation that we wouldn't, but I may have misheard that).

We have found that tags is a pretty good way to filter events (consumer defined filters, evaluated on producer side). Without tags those filters can only be on entity ids (exact or regex). I think we have been and still are uncertain about how useful consumer defined filters are for the case when edge is producer. I guess the consumer would be interested in all events from the edge in most cases, and the edge itself decides what to publish (producer defined filter). I would say that tag filters can be added when we have more important things in place. Also depends on effort, if it's a small thing to add tagging support we can just do it.

We now have a WithTags trait for envelopes. At present, it'll always return an empty vector. I'll create a separate issue to circle back on considering the utility of tags at the edge. UPDATE: Issue: #38

Also note that topic filters is actually a specific type of tag filters https://doc.akka.io/docs/akka-projection/snapshot/grpc.html#topics Topic is not the same concept as a streambed topic.

Understood.

huntc · 2023-10-07T06:01:29Z

Yes, but I wonder if we could make all of the optional handler parameters a single setup-kinda thing, ...

I've now included an explicit producer_filter parameter via commit 715c7b4.

I've separated out the declarations so that you can see how the grpc_flow handler is constructed:

    let grpc_flow = GrpcEventFlow::new(entity_type, grpc_producer);
    let producer_filter = |_: &CommitLogEventEnvelope<temperature::Event>| true;
    let consumer_filter = Filter::default();
    let event_handler = grpc_flow.handler(
        producer_filter,
        consumer_filters_receiver,
        consumer_filter,
        transformer,
    );

We have a separate issue to see if we can improve the DX on some of these calls. My present focus is to standup the full API.

Was previously running the stream only once

We should always endeavour to avoid running out of heap

patriknw · 2023-10-09T13:30:30Z

akka-projection-rs/src/consumer_filter.rs

-    IncludeEntityIds {
-        entity_id_offsets: Vec<EntityIdOffset>,
+    IncludePersistenceIds {
+        persistence_id_offsets: Vec<PersistenceIdIdOffset>,
    },


What happened with EntityId vs PersistenceId?
The filters should be defined in terms of entity id on the consumer side.

Then there can be a mapping between stream id and entity type. This mapping is on the producer side.

There's obviously something I'm not understanding here. I took inspiration from the protobuf to filter conversions on the JVM: https://github.com/akka/akka-projection/blob/main/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/ProtobufProtocolConversions.scala#L242-L285

And also more specifically here: https://github.com/akka/akka-projection/blob/6f13d3c378e4cbd6f2e7d214f25ce63dd902f85f/akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala#L77

I might be lost in the implementation details, but this is my understanding of how it should work.

edge is consumer

The user API to define a filter at the edge should be IncludeEntityIds, ExcludeEntityIds and corresponding Remove.
That is transformed pretty much 1:1 to the corresponding EntityId protobuf messages.

Those filters are evaluated on the jvm consumer side, so you don't have to care about mapping between streamId and entityType for this case.

edge is producer

The incoming protobuf filters are IncludeEntityIds and ExcludeEntityIds. When filtering the events from the journal you have to filter on the EntityType and EntityId. You know the EntityType on the edge producer side, because that was defined when setting up the producer push stream. That producer push stream should only emit events with the given EntityType to begin with.

Thanks for the offline discussions. Commit d2e1550 restores things back to filtering by entity ids.

patriknw · 2023-10-09T13:42:24Z

akka-projection-rs-grpc/src/lib.rs

+    EntityId, EntityType, Offset, PersistenceId, Tag, TimestampOffset, WithOffset,
+};
+use akka_projection_rs::consumer_filter::{ComparableRegex, FilterCriteria, PersistenceIdIdOffset};
+use mqtt::TopicFilter;


Given that we don't have tags (yet) I'm not sure we should add the mqtt dependency.

If there's the potential of supporting tags in the future i.e. if we're still unsure, I'd prefer to retain the code. I've now hidden the MQTT types from the developer though as per commit 0fc0f50.

patriknw · 2023-10-09T13:59:06Z

akka-projection-rs/src/consumer_filter.rs

    /// Remove a previously added `IncludeTopics`.
-    RemoveIncludeTopics { expressions: Vec<TopicMatcher> },
+    RemoveIncludeTopics { expressions: Vec<TopicFilter> },


I'm not sure we should tie this to the mqtt dependency. I picked mqtt for the semantics of the topic filter wildcards but we don't aim for speaking mqtt protocol in any way.

I can understand that you added the library instead of implementing the evaluation of the topic filter, but I think we should keep that as an implementation detail (not exposed to user).

Note that these filters are also used for the case where edge is consumer, and filter is evaluated by (jvm) producer.

The MQTT dependency is now isolated to one place and the TopicFilter type has been wrapped with our own TopicMatcher. We are therefore free to change its implementation if required.

Commit 0fc0f50 resolves this.

The MQTT TopicFilter is hidden to the outside so that we can freely change it, as we're not necessarily supporting all of MQTT's capabilities.

patriknw

LGTM

Consumer supplied filters when producing

10e97e5

When producing from the edge, the consumer may also supply filters. The changes here implement that by providing a watch channel for the upstream gRPC producer.

patriknw reviewed Oct 5, 2023

View reviewed changes

patriknw mentioned this pull request Oct 5, 2023

Higher level api for typical usage #37

Closed

Possibly better name for flowing gRPC events

8730696

huntc force-pushed the consumer-filter branch from d585367 to 8730696 Compare October 6, 2023 02:14

Beginnings of fleshing out the filtering further

3384e04

Note quite working yet.

huntc added 3 commits October 7, 2023 13:06

Replace WithEntityId and streamlined EventType usage

a07ea02

By conveying a PersistenceId instead of an EntityId for envelopes, we can reduce the places where we need to declare the event type.

Includes a test for the consumer filter

e8b7b35

Makes the filter update-able as per the JVM

6cba9bb

huntc added 2 commits October 7, 2023 16:52

Doc

8b1231e

Provides a producer filter

715c7b4

Formatting

e60ac7d

huntc marked this pull request as ready for review October 7, 2023 06:05

huntc self-assigned this Oct 7, 2023

huntc added the enhancement New feature or request label Oct 7, 2023

huntc changed the title ~~Consumer supplied filters when producing~~ Consumer supplied filters when producing, along with a producer filter Oct 7, 2023

huntc added 3 commits October 8, 2023 14:57

Correctly merge criteria into filters

9c6911f

Filter tests

4028a94

Use more refs - reduces the clones

7a51787

huntc added 5 commits October 8, 2023 16:43

Improved bench

820dae8

Was previously running the stream only once

Improved some names

ba3fae7

Restrict the max size of a filter's fields

cc3e556

We should always endeavour to avoid running out of heap

Improve the filter dx

928b5d8

More DX tidy-up

f1162a4

patriknw reviewed Oct 9, 2023

View reviewed changes

huntc added 2 commits October 10, 2023 10:21

MQTT protocol is now hidden

0fc0f50

The MQTT TopicFilter is hidden to the outside so that we can freely change it, as we're not necessarily supporting all of MQTT's capabilities.

Filter by entity ids for edge based consumers

d2e1550

patriknw approved these changes Oct 10, 2023

View reviewed changes

huntc merged commit f12f526 into main Oct 10, 2023
1 check passed

huntc deleted the consumer-filter branch October 10, 2023 21:56

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Consumer supplied filters when producing, along with a producer filter #36

Consumer supplied filters when producing, along with a producer filter #36

huntc commented Oct 5, 2023 •

edited

Loading

patriknw Oct 5, 2023

huntc Oct 5, 2023

huntc Oct 6, 2023

patriknw commented Oct 6, 2023

patriknw commented Oct 6, 2023

patriknw commented Oct 6, 2023

johanandren commented Oct 6, 2023

huntc commented Oct 6, 2023

johanandren commented Oct 6, 2023 •

edited

Loading

huntc commented Oct 6, 2023 •

edited

Loading

johanandren commented Oct 6, 2023

huntc commented Oct 7, 2023

huntc commented Oct 7, 2023 •

edited

Loading

huntc commented Oct 7, 2023 •

edited

Loading

patriknw Oct 9, 2023

huntc Oct 10, 2023 •

edited

Loading

patriknw Oct 10, 2023

huntc Oct 10, 2023

patriknw Oct 9, 2023

huntc Oct 9, 2023

patriknw Oct 9, 2023

huntc Oct 9, 2023

patriknw left a comment

Consumer supplied filters when producing, along with a producer filter #36

Consumer supplied filters when producing, along with a producer filter #36

Conversation

huntc commented Oct 5, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw commented Oct 6, 2023

patriknw commented Oct 6, 2023

patriknw commented Oct 6, 2023

johanandren commented Oct 6, 2023

huntc commented Oct 6, 2023

johanandren commented Oct 6, 2023 • edited Loading

huntc commented Oct 6, 2023 • edited Loading

johanandren commented Oct 6, 2023

huntc commented Oct 7, 2023

huntc commented Oct 7, 2023 • edited Loading

huntc commented Oct 7, 2023 • edited Loading

Choose a reason for hiding this comment

huntc Oct 10, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw left a comment

Choose a reason for hiding this comment

huntc commented Oct 5, 2023 •

edited

Loading

johanandren commented Oct 6, 2023 •

edited

Loading

huntc commented Oct 6, 2023 •

edited

Loading

huntc commented Oct 7, 2023 •

edited

Loading

huntc commented Oct 7, 2023 •

edited

Loading

huntc Oct 10, 2023 •

edited

Loading