Add entry-point plugin system to pegasus-monitord (#2194) - #2222
Open
mjstealey wants to merge 14 commits into
Open
Add entry-point plugin system to pegasus-monitord (#2194)#2222mjstealey wants to merge 14 commits into
mjstealey wants to merge 14 commits into
Conversation
Adds a MonitordEventPlugin base class plus a manager that discovers plugins via the `pegasus.monitord.plugins` entry-point group and runs each enabled plugin (pegasus.monitord.plugins.<name>.enabled=true) on its own bounded-queue daemon thread. Plugins are driven through a new PluginHostEventSink attached to monitord's existing multiplex fan-out via an injected `plugins://` endpoint, so a slow or crashing plugin can neither block the parse loop nor disable database population. - monitoring/plugin.py: MonitordEventPlugin, _PluginWorker (queue+thread), MonitordPluginManager (entry-point discovery + explicit-enable gate) - monitoring/event_output.py: PluginHostEventSink + plugins:// scheme - cli/pegasus-monitord.py: inject the plugin host endpoint; thread the full props through as monitord_props (multiplex strips the plugin keys) - test/test_monitord_plugin.py: 12 unit + integration tests Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
PluginHostEventSink enqueued the live event-payload dict by reference to per-plugin worker threads that read it asynchronously, while monitord's main thread keeps reusing/mutating that same dict after dispatch (the rc.meta loop in workflow.py overwrites key/value and re-sends one dict; wf.plan adds db_url post-dispatch). Safe before the plugin system because the only consumer (DBEventSink.send) copies synchronously; the new queue-backed worker reads the dict milliseconds later, observing torn payloads -- and a corrupted dict when two plugins share one object. _PluginWorker.submit now snapshots the payload with dict(kw) at enqueue time: taken synchronously on the main thread before any later mutation, and per-worker so plugins cannot corrupt each other's view. Shallow copy suffices because every payload value monitord produces is an immutable scalar. Adds a deterministic regression test (GatedRecordingPlugin pins the worker inside the first handle_event so the mutation provably precedes the worker's read) and a writeup of the finding, the fix, and gotchas. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Plugins are purely event-driven: _PluginWorker blocks on queue.get() with no timeout, so a plugin gets no CPU during quiet periods -- exactly when wall-clock work like polling an external system matters most (idle or stalled jobs produce no stampede events). Add an opt-in tick() hook on MonitordEventPlugin, driven by the EXISTING worker thread: when pegasus.monitord.plugins.<name>.tick_interval > 0 the worker waits with get(timeout=<remainder of interval>) and calls tick() on queue.Empty, plus a starvation guard that ticks between events under continuous flow. tick() runs on the same thread as handle_event (no locking needed between them), is exception-isolated identically, and can never fire after the shutdown sentinel -- the FIFO drain-and-join close semantics are unchanged. With the property unset the worker runs the byte-identical blocking loop as before. Co-Authored-By: Claude Fable 5 <[email protected]>
…tes (#2194) Reference guide gains an "Event Plugins" section under Monitoring > pegasus-monitord: writing a plugin (start/handle_event/tick/stop with the threading contract), entry-point registration under pegasus.monitord.plugins, the property namespace (enabled, queue_size, join_timeout, tick_interval), delivery semantics (per-plugin payload snapshots, in-order per plugin, bounded best-effort queue, isolated failures), the static.bp non-epoch ts payload caveat, and a pointer to the wfmonitor production example. monitord-plugin-payload-race.md updated with post-fix field findings from the FABRIC deployments (2026-06-11/12): the snapshot fix validated live; a triage warning that small-integer "torn" timestamps are usually the planner's 1970-era static.bp roster stamps, NOT this race (they survive the fix and sit in every 5.1.2 stampede DB); the tick() addition (2a6a23f) analyzed -- same worker thread, serialized with handle_event, unreachable after the shutdown sentinel, so the race analysis is unchanged; verification section updated to the current suite (15 passed, 3 skipped) with the working dev-env command. Co-Authored-By: Claude Fable 5 <[email protected]>
This developer scratch/analysis doc (monitord-plugin-payload-race.md) is kept locally only and was not intended for the shared tree. Removed from the branch; contributors keep it on disk via .git/info/exclude. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Adds IdentityRecordingPlugin and test_each_plugin_gets_an_isolated_payload_copy, which enables two plugins and asserts each worker receives a distinct payload copy (not the producer's dict, nor each other's). dispatch() hands the same producer dict to every worker, so the per-worker dict(kw) snapshot in _PluginWorker.submit is the only thing isolating plugins; this guards the #2194 cross-thread payload race for the >=2 plugin case. Fails deterministically when the dict(kw) snapshot is reverted. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
…stop() Builds on the #2194 cross-thread payload race fix with three robustness improvements, all confined to the opt-in plugin host path (no effect when pegasus.monitord.plugins.* is unset): - Deep payload snapshot: _PluginWorker.submit now uses copy.deepcopy(kw) instead of a shallow dict(kw), so nested mutable payload values (e.g. job_inst.composite invocation lists / multipart records) are isolated per plugin, not just top-level keys. - Bounded stop(): plugin.stop() runs in a daemon helper thread joined with join_timeout; a wedged stop() is logged and abandoned rather than hanging monitord shutdown. Wired into both stop_all() and startup cleanup. - Startup watchdog: plugin.start() runs in a daemon helper thread bounded by a new pegasus.monitord.plugins.<name>.start_timeout (default: that plugin's join_timeout). A start() that exceeds the bound is skipped and monitord continues; if it later returns, a bounded stop() cleanup runs via a cross-thread handshake. Python cannot forcibly kill the thread, so timed-out startup is documented as best-effort abandonment. Adds regressions for nested-payload isolation, blocking stop(), and startup timeout + late cleanup. Updates the reference-guide plugin contract. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
…ating Six review follow-ups to the monitord event plugin host (#2194): - Gate host activation on a truthy pegasus.monitord.plugins.<name>.enabled instead of any plugin property, so leftover disabled config no longer flips monitord's sink topology to a multiplex. New enabled_plugin_names() helper shared by the gate and the manager. - Warn at startup for names enabled in properties but not registered under the entry-point group (previously only "started 0 plugin(s)" at INFO). - Log a final per-plugin dropped-events total (and filtered total) when a worker closes, covering the abandoned-on-timeout path. - Per-plugin event filtering: an event_filter prefix tuple on the plugin (or the .events property, which replaces it; '*' = all, () = tick-only), checked before the per-plugin deepcopy so uninteresting events cost nothing on the parse loop. - Configurable overflow policy: .overflow_policy = drop-newest (default, current behavior) | drop-oldest, which evicts the oldest queued event so live monitors keep the freshest state; sentinel-safe and bounded under the single-producer invariant. _worker_config now returns a namedtuple. - Expose monitord's replay/recovery restart flag to plugins via start(props, restart=...), signature-gated so old-style start(self, props=None) overrides are called exactly as before; the flag rides the existing restart kwarg fan-out into PluginHostEventSink. Docs: gate wording, event filtering, overflow policy, and a replay/recovery idempotency section for plugin authors in monitoring.rst. Tests: 30 new (56 total), including deepcopy-skip proof via a __deepcopy__ probe, drop-oldest eviction order, sentinel protection, old/new/kwargs start signatures, and the disabled-config plain-sink regression. Co-Authored-By: Claude Fable 5 <[email protected]>
Author
|
Adding for reference (not part of the commit) |
This file contains hidden or 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
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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.
Summary
Review request for: @vahi or @mayani
Adds an opt-in, entry-point-based plugin system to
pegasus-monitordso third-party Python packages can receive every stampede workflow event live, in-process — without forking monitord or scraping the stampede database after the fact.Closes #2194.
How it works
MonitordEventPlugin(newPegasus.monitoring.pluginmodule) and register under thepegasus.monitord.pluginsentry-point group.pegasus.monitord.plugins.<name>.enabled = true. With no plugin enabled, monitord's sink topology is completely unchanged.plugins://endpoint into thepegasus.catalog.workflow.*.urlnamespace so the existing multiplex fan-out drives the newPluginHostEventSinkalongside the database sink (reserved-name probing avoids clobbering user endpoints).send()only enqueues — it never blocks and never raises — so a slow or crashing plugin can neither stall monitord's parse loop nor disable the database writer.Robustness
start()/stop()so a misbehaving plugin cannot hang monitord startup/shutdown.tick(), and a restart flag — all viapegasus.monitord.plugins.<name>.*properties.pegasus.monitord.plugins.*from per-endpoint sink properties so plugin config never leaks into other sinks.Docs and tests
doc/sphinx/reference-guide/monitoring.rst(authoring, registration, properties, delivery semantics).test/test_monitord_plugin.pycovering the plugin manager, host sink, filtering, overflow, lifecycle bounds, and payload isolation.🤖 Generated with Claude Code