DM-39485: Use redis for work dispatch

[mfisherlevine]

No description provided.

[TallJimbo]

Lots of comments, some minor, some about the future, but really not that many given the PR size.

I did not look too closely at:

• logic where I didn't have enough context/background to understand it (e.g. decoding redis objects, non-redis channels)
• scripts (that's all machine-generated, right, for gitops, except scripts/meta?)
• places where you already had "TODO" comments that suggested that you wanted to rewrite that code substantially already.

[TallJimbo]

Is this comment supposed to be here? Looks like a pipeline yaml file to me, not a Redis config (same in other similar files).

[TallJimbo]

Are these doubled #s significant?

[TallJimbo]

Args here are now out of sync with docs.

[TallJimbo]

Given how things have gone since we first added this, we might want to evaluate whether we can drop the stuff that uses it here instead. I may be wrong, but it seemed like being able to change the pipeline without restarting a pod was looking less important.

[TallJimbo]

I generally very much recommend using pydantic for structs that get converted to/from JSON, but this expansion-by-butler at least that makes that less of an obvious win.

[TallJimbo]

Suggested change

	raise RuntimeError(f"Unexpected error connecting to redis: {e}")
	raise RuntimeError(f"Unexpected error connecting to redis: {e}") from e

[TallJimbo]

Or maybe just let this raise...I'm not sure your RuntimeError is adding anything useful here.

[TallJimbo]

I'd seen processingId in a few places before but didn't see what it meant until I got here. I think it's going to need some kind of abstraction to work with the AOS pipelines, from what I've seen of them.

This is just one example of what I think is the biggest weakness of this system in its current state, so I'm gonna use this thread as a place to start that conversation:: different kinds of data IDs (like visits and exposures) are all over the place, and in some case they're instrument-specific. Instead, I think the redis, control layer, and worker code should instead recognize three kinds of processing:

• per-detector (these layers do need to know about detectors because that's the worker-affinity dimension);
• gather-of-detectors (per exposure, per visit, per intra/extra-focal pair-of-visits, maybe per-group someday);
• cumulative nightly.

I think we can introspect a PipelineGraph to split it up tasks into a sequence of those (or die if that can't be done). The ambitious version of this goal gets rid of all mentions of expRecord from anything other than user-display messages (e.g. logs, plots) and things that actually need exposure metadata (I'm thinking it'd be replaced by some ABC), and it'd get rid of all mention of "steps". But that ambitious version may not be doable until older Rapid Analysis channel types are migrated, and it may never be worth doing at all. The argument for doing is that it prepares us for the next time somebody comes up with a new problem for Rapid Analysis to solve (i.e. new pipeline for it to run), by separating the orchestration logic further from the assumptions about the kinds of pipelines it runs.

[TallJimbo]

Is the redis content persistent, then, so if the redis DB goes down it remembers this sort of thing when it comes back up?

[TallJimbo]

We usually put __all__ before all of the imports (except __future__ imports).

[TallJimbo]

Some documentation of the Redis schema would be really useful: e.g. keys, types, and what they mean. Absent that, I can't really claim I understood how various states for workers are represented and hence how different methods interact.