feat(scheduler): Report lack of dataflow engines in pipeline statuses

[agrski]

Why

Issues

N/A

Motivation

Currently, the pipeline status returned from the scheduler to the operator does not report if there are no dataflow engines available on which to schedule that pipeline. The pipeline is left in state PipelineCreate with no further details; users must check the scheduler's logs in order to determine that this is the cause.

This PR ameliorates this situation by propagating this information back to the pipeline status in the scheduler-internal data store, which in turn is used to inform subscribers such as the Core v2 operator.

This PR may, at first glance, look like it changes more than required; there is good reason for this. The chainer server (dataflow engine server) is both a consumer of pipeline events on the internal event bus and also a publisher to this particular stream of events. The simple solution of just updating the pipeline state when no dataflow engines are registered causes a potentially infinite loop in which the server component receives an event, sees there are no engines, publishes an event, then listens to this same event it has just published. This loop can run very quickly, constantly spawning new coroutines, in fact to the point that with a single pipeline it can OOM the scheduler (running with the default 1GB memory) in ~40 seconds.

To break this cycle, we need a mechanism for de-duplicating or otherwise ignoring events. The solution provided here is to add an event source, such that the producer of a message can ignore anything produced by itself. This is relatively simple and also quite generic in its approach--there is no need to cache particular event IDs or anything like that, or to split things up into many separate topics.

What

Summary of changes

• Set pipeline status message when no dataflow engines are available.
• Add source for pipeline events for publishers to ignore their own output.

Testing

Prior to my changes, the pipeline status is:

k -n seldon get pipelines.mlops.seldon.io dummy-pipeline -o json \
  | jq '.status.conditions[] | select(.type == "PipelineReady")'

{
  "lastTransitionTime": "2023-08-10T14:01:18Z",
  "reason": "PipelineCreate",
  "status": "False",
  "type": "PipelineReady"
}

By building a local scheduler image and pushing this into a kind cluster:

make -C scheduler docker-build-scheduler
make -C scheduler kind-image-install-scheduler
kubectl rollout restart -n seldon sts seldon-scheduler

then deleting and re-creating a dummy pipeline, the status now looks like:

kubectl -n seldon get pipelines.mlops.seldon.io dummy-pipeline -o json \
  | jq '.status.conditions[] | select(.type == "PipelineReady")'

{
  "lastTransitionTime": "2023-08-09T11:11:09Z",
  "message": "no dataflow engines available to handle pipeline",
  "reason": "PipelineCreate",
  "status": "False",
  "type": "PipelineReady"
}

[jesse-c]

LGTM

We were able to reproduce the issue yesterday, apply the fix, and confirm it!