experimental: sse client support #592
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Dependency ReviewThe following issues were found:
License IssuesCargo.lock
server/Cargo.toml
Denied Licenses: GPL-1.0, GPL-2.0, GPL-3.0, LGPL-2.1, LGPL-3.0, AGPL-3.0 OpenSSF ScorecardScorecard details
Scanned Files
|
thomasheartman
commented
Dec 5, 2024
kwasniew
reviewed
Dec 5, 2024
kwasniew
reviewed
Dec 5, 2024
thomasheartman
force-pushed
the
streaming-spike
branch
from
b3122da to
ae70a59
Compare
kwasniew
reviewed
Dec 6, 2024
thomasheartman
commented
Dec 9, 2024
thomasheartman
commented
Dec 10, 2024
| } | ||
| } | ||
| }) | ||
| .map_err(|e| warn!("Error in SSE stream: {:?}", e)); |
There was a problem hiding this comment.
handle 404s etc, maybe launch polling if can't connect
There was a problem hiding this comment.
This is a more complex topic. I suggest we move handling this one to a little later.
chriswk
approved these changes
Dec 10, 2024
nunogois
approved these changes
Dec 11, 2024
|
|
||
| #[cfg(feature = "streaming")] | ||
| #[get("/streaming")] | ||
| pub async fn stream_features( |
There was a problem hiding this comment.
Maybe this would be slightly more consistent with Unleash's counterpart?
Suggested change
| pub async fn stream_features( | |
| pub async fn connect( |
There was a problem hiding this comment.
Sure. But as the API endpoint, maybe it should be more explicit? Feels like stream_features is clearer to me. But then again, it's a separate controller in Unleash. I don't care much either way, but this feels more explicit in this context.
| impl From<SendError<Event>> for EdgeError { | ||
| // todo: create better enum representation. use this is placeholder | ||
| fn from(_value: SendError<Event>) -> Self { | ||
| EdgeError::TlsError |
There was a problem hiding this comment.
Might be weird seeing a TlsError here, but I'm guessing we're okay with this for now?
There was a problem hiding this comment.
Yes, I agree. Hence the comment. I just didn't want to start implementing a new error enum member when cleaning this up. We should handle this as part of testing, because I don't think it is the right error.
server/src/http/feature_refresher.rs
Outdated
| engine_cache: engines, | ||
| refresh_interval: features_refresh_interval, | ||
| persistence, | ||
| strict, | ||
| app_name: app_name.into(), | ||
| #[cfg(feature = "streaming")] | ||
| broadcaster: Broadcaster::new(features.clone()), |
There was a problem hiding this comment.
How about we move this above features_cache? That way we can clone here if needed, but leave features_cache: features intact as the owner of the features.
| } | ||
| _ = metrics_pusher::prometheus_remote_write(prom_registry_for_write, edge.prometheus_remote_write_url, edge.prometheus_push_interval, edge.prometheus_username, edge.prometheus_password, app_name) => { | ||
| tracing::info!("Prometheus push unexpectedly shut down"); | ||
| } |
There was a problem hiding this comment.
Wish we could simplify this and reduce the duplicated code but I don't have any better ideas atm.
There was a problem hiding this comment.
Yeah, I know. I hate this. But we can refactor it later. I don't think this'll be the final version (I hope not!).
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Overview
tl;dr: This is a very experimental implementation of streaming in Edge: it both listens to the experimental Unleash streaming endpoint and pushes updates to any listeners subscribing to Edge (in effect mirroring the Unleash endpoint). All related code is behind the "streaming" feature gate.
More detail:
I have added an event source client to the
features_refresherfile. If the streaming flag is on, we'll spawn a task that takes care of listening to Unleash for events instead of starting the poll loop for flags.There is a new endpoint at
api/client/streaming(mirroring the Unleash endpoint) that you can hit to start listening for updates.The updates are handled by a new
broadcastermodule (largely stolen from this Actix example).The broadcaster stores its client in a hash map that uses the flag query as the key and maps it to a vec of clients that use the same query.
Left to do
Regarding the implementation: I'm not very familiar with Actix, and I haven't done a whole lot of async / multithreaded rust before, so there's probably gonna be a whole lot of things that we can improve, from the architecture level to the specific data structures used.
Also, due to the very time-limited spike we did, we need to actually do some real error handling. There's a good few places where we either ignore errors or would just panic if we ever encountered them.But aside from that, there's a few other things to do too: