Allow streaming events while running a pipeline

[schustmi]

Describe changes

Live event streaming on pipeline runs over SSE. Can be enabled by setting stream_broker_implementation_source.

Producer

Call zenml.streaming.publish({...}, kind=..., correlation_id=..., index=...) from inside any step or dynamic pipeline.

Server

Add StreamBroker interface and RedisStreamsBroker implementation. StreamBroadcaster runs one broker reader per stream, fans out to N SSE consumers, handles catch-up, gap signalling, idle cleanup, per-stream consumer cap.

Endpoint

GET /api/v1/runs/{id}/events/stream

Multi-value filters: ?kinds=, ?step_names=, ?correlation_ids=. Resume via Last-Event-ID header or ?since=.

Pre-requisites

Please ensure you have done the following:

• I have read the CONTRIBUTING.md document.
• I have added tests to cover my changes.
• I have based my new branch on develop and the open PR is targeting develop. If your branch wasn't based on develop read Contribution guide on rebasing branch to develop.
• IMPORTANT: I made sure that my changes are reflected properly in the following resources:
  • ZenML Docs
  • Dashboard: Needs to be communicated to the frontend team.
  • Templates: Might need adjustments (that are not reflected in the template tests) in case of non-breaking changes and deprecations.
  • Projects: Depending on the version dependencies, different projects might get affected.

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Other (add details above)

[github-actions]

✅ No broken links found!

[Json-Andriopoulos]

This is interesting I like it. We can reach such conditions, I think it would be good to have it implemented as part of this PR.

[schustmi]

The only reason I didn't implement this was that I'm concerned about overloading our server.

• there might be multiple consumers of the same stream on different replicas
• for each consumer, this would send a request for the run status every 30 seconds or whatever the configured heartbeat interval
• the current implementation scales pretty well with multiple consumers, as each replica only subscribes to each stream only once. but this would now add DB load relative to the consumer count

From my understanding (but I might be wrong), idling consumers waiting on an empty stream shouldn't consume any noticeable resources.

On the other hand, it usually won't really matter for the consumer if the connection isn't closed as soon as the run ends. The user will usually be on a website that displays the stream, and once they change the URL the browser will close the connection.

So essentially, we'd add these potentially many run status checks for little benefit, and the happy case should already publish the end frame just fine, so all of this would be just to cover some unlikely edge cases.

What do you think?

[Json-Andriopoulos]

Hmm I was thinking about a more infrequent HB (5, 10 minutes would make sense) to avoid the mid/long-term effect of dangling sessions but even that is probably a stretch. An alternative here is: We don't care as long as we can inspect it. For instance, there is a server performance issue how can I rule out SSE, queues etc?

[schustmi]

@Json-Andriopoulos What do you mean by not caring here? Not caring about idle connections? Or not caring about DB calls?

[Json-Andriopoulos]

Idle connections :)

[socket-security]

Review the following changes in direct dependencies. Learn more about Socket for GitHub.

Diff	Package	Supply Chain Security	Vulnerability	Quality	Maintenance	License
	redis@​7.4.0

View full report