dataflow_rs/observe/observer.rs
1//! Observers: what a run tells a host as it goes, inline from the
2//! recorder, so an attached observer adds no await.
3
4use super::{RunStatus, WorkflowRecord};
5use crate::base::{ProgramId, WorkflowIx};
6use crate::function::CallInfo;
7use std::fmt;
8use std::ops::{BitOr, BitOrAssign};
9use std::time::Duration;
10
11/// What a host is told about a run as it goes: counters, histograms and
12/// spans, kept by the host.
13///
14/// An engine has one ([`EngineConfig::observer`]) and a run may add one
15/// beside it ([`RunOptions::observer`]). Each is called only with the events
16/// its [`interest`](Self::interest) names, and the clock is read only when
17/// one of them asks for [`Interest::TIMINGS`].
18///
19/// [`on`](Self::on) is called inline, synchronously, from the run: it must
20/// not block, re-enter the engine or panic. Push to a channel or bump an
21/// atomic and return.
22///
23/// ```
24/// # tokio::runtime::Builder::new_current_thread().build().unwrap().block_on(async {
25/// use dataflow_rs::observe::{Event, Observer};
26/// use dataflow_rs::run::EngineConfig;
27/// use dataflow_rs::{Engine, Message};
28/// use serde_json::json;
29/// use std::sync::Arc;
30/// use std::sync::atomic::{AtomicU64, Ordering};
31///
32/// #[derive(Default)]
33/// struct Calls(AtomicU64);
34///
35/// impl Observer for Calls {
36/// fn on(&self, event: &Event<'_>) {
37/// if let Event::TaskFinished { .. } = event {
38/// self.0.fetch_add(1, Ordering::Relaxed);
39/// }
40/// }
41/// }
42///
43/// let calls = Arc::new(Calls::default());
44/// let engine = Engine::builder()
45/// .config(EngineConfig::default().observer(calls.clone()))
46/// .workflow_json(&json!({"id": "w", "tasks": [
47/// {"id": "t", "function": {"name": "map", "input": {
48/// "mappings": [{"path": "data.a", "logic": 1}]}}}
49/// ]}))
50/// .build()
51/// .unwrap();
52/// engine.run(&mut Message::from_json(&json!({}))).await;
53/// assert_eq!(calls.0.load(Ordering::Relaxed), 1);
54/// # });
55/// ```
56///
57/// [`EngineConfig::observer`]: crate::run::EngineConfig::observer()
58/// [`RunOptions::observer`]: crate::run::RunOptions::observer()
59pub trait Observer: Send + Sync + 'static {
60 /// The events this observer is called with. Read once per run, when the
61 /// run starts. `WORKFLOWS | TASKS` by default, which reads no clock.
62 fn interest(&self) -> Interest {
63 Interest::WORKFLOWS | Interest::TASKS
64 }
65
66 /// One event of a run, inline and synchronously. It must not block or
67 /// re-enter the engine; what it keeps, it copies.
68 fn on(&self, event: &Event<'_>);
69}
70
71/// What an [`Observer`] asks for: a set of flags. `|` and `|=` unite two
72/// sets, [`union`](Self::union) does in a `const`, and
73/// [`contains`](Self::contains) tests one.
74///
75/// ```
76/// use dataflow_rs::observe::Interest;
77///
78/// const TIMED_TASKS: Interest = Interest::TASKS.union(Interest::TIMINGS);
79/// assert_eq!(TIMED_TASKS, Interest::TASKS | Interest::TIMINGS);
80/// ```
81///
82/// Every observer receives [`RunStarted`](Event::RunStarted) and
83/// [`RunFinished`](Event::RunFinished).
84#[derive(Clone, Copy, PartialEq, Eq, Hash)]
85pub struct Interest(u8);
86
87impl Interest {
88 /// No flag: the run's events only.
89 pub const NONE: Self = Self(0);
90 /// [`WorkflowStarted`](Event::WorkflowStarted) and
91 /// [`WorkflowFinished`](Event::WorkflowFinished).
92 pub const WORKFLOWS: Self = Self(1);
93 /// [`TaskFinished`](Event::TaskFinished).
94 pub const TASKS: Self = Self(1 << 1);
95 /// Each `elapsed` this observer is given is `Some`: the clock is read.
96 pub const TIMINGS: Self = Self(1 << 2);
97 /// Every flag.
98 pub const ALL: Self = Self::WORKFLOWS.union(Self::TASKS).union(Self::TIMINGS);
99
100 /// The flags' names, in `Debug`'s order.
101 const NAMES: [(Self, &'static str); 3] = [
102 (Self::WORKFLOWS, "WORKFLOWS"),
103 (Self::TASKS, "TASKS"),
104 (Self::TIMINGS, "TIMINGS"),
105 ];
106
107 /// The flags of either set: `|`, usable in a `const`.
108 #[must_use]
109 pub const fn union(self, other: Self) -> Self {
110 Self(self.0 | other.0)
111 }
112
113 /// Whether every flag of `other` is in this set: always for
114 /// [`NONE`](Self::NONE).
115 #[must_use]
116 pub const fn contains(self, other: Self) -> bool {
117 self.0 & other.0 == other.0
118 }
119
120 /// Whether the set holds no flag.
121 #[must_use]
122 pub const fn is_empty(self) -> bool {
123 self.0 == 0
124 }
125}
126
127impl BitOr for Interest {
128 type Output = Self;
129
130 /// The flags of either set: [`Interest::union`].
131 fn bitor(self, other: Self) -> Self {
132 self.union(other)
133 }
134}
135
136impl BitOrAssign for Interest {
137 /// Adds `other`'s flags.
138 fn bitor_assign(&mut self, other: Self) {
139 *self = self.union(other);
140 }
141}
142
143impl fmt::Debug for Interest {
144 /// The flags set, by name: `Interest(WORKFLOWS | TASKS)`, or
145 /// `Interest(NONE)`.
146 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
147 f.write_str("Interest(")?;
148 if self.is_empty() {
149 f.write_str("NONE")?;
150 }
151 let set = Self::NAMES.iter().filter(|(flag, _)| self.contains(*flag));
152 for (i, (_, name)) in set.enumerate() {
153 if i > 0 {
154 f.write_str(" | ")?;
155 }
156 f.write_str(name)?;
157 }
158 f.write_str(")")
159 }
160}
161
162/// One event of a run, borrowed for the call to [`Observer::on`]: an
163/// observer copies out what it keeps.
164///
165/// An `elapsed` is `Some` only for an observer whose interest holds
166/// [`Interest::TIMINGS`]. Each variant is `#[non_exhaustive]`: match its
167/// fields with `..`, so a fact added later breaks no host.
168#[derive(Debug)]
169#[non_exhaustive]
170pub enum Event<'a> {
171 /// The run starts, after its stamps, before any workflow.
172 #[non_exhaustive]
173 RunStarted {
174 /// The program the run runs.
175 program: ProgramId,
176 /// The message's id.
177 message_id: &'a str,
178 /// The channel the run was started on.
179 channel: Option<&'a str>,
180 /// How many workflows the run considers: its channel's Active ones,
181 /// before any rollout or condition is judged.
182 considered: usize,
183 },
184 /// A workflow was selected: its condition admitted it, or it has none.
185 /// Once per workflow, a loop's included. A workflow that ends `Ran`
186 /// with no judgment of its condition reached (a loop that enters no
187 /// sweep, or a run stopped before the condition) gets it as it ends,
188 /// before its `WorkflowFinished`.
189 #[non_exhaustive]
190 WorkflowStarted {
191 /// The workflow's position in the program.
192 workflow: WorkflowIx,
193 /// The workflow's id.
194 id: &'a str,
195 },
196 /// A call was recorded: once per call, a fan-out's once per element it
197 /// folds, and once with no element for a fan-out with nothing to call or
198 /// cut short. A task its condition refused is not.
199 #[non_exhaustive]
200 TaskFinished {
201 /// The call: its workflow, its task and the function's registered
202 /// name, its loop counter and its element.
203 info: &'a CallInfo<'a>,
204 /// The status recorded; `None` for a `Skip`.
205 status: Option<u16>,
206 /// How long the call ran, the recording not included.
207 elapsed: Option<Duration>,
208 },
209 /// A workflow that was selected ended. Its record says how, and how many
210 /// sweeps it ran.
211 #[non_exhaustive]
212 WorkflowFinished {
213 /// What the workflow did.
214 record: &'a WorkflowRecord<'a>,
215 /// How long since its [`WorkflowStarted`](Event::WorkflowStarted).
216 elapsed: Option<Duration>,
217 },
218 /// The run ended, whatever its status.
219 #[non_exhaustive]
220 RunFinished {
221 /// The message's id, as its `RunStarted` named it: what ties the end
222 /// to the start when one observer sees runs overlap.
223 message_id: &'a str,
224 /// How the run ended.
225 status: &'a RunStatus,
226 /// How many error records the message holds.
227 errors: usize,
228 /// How long the run took.
229 elapsed: Option<Duration>,
230 },
231}