greentic_runner_host/
telemetry.rs

1use greentic_types::telemetry::set_current_tenant_ctx;
2use greentic_types::{EnvId, TenantCtx, TenantId};
3
4/// Canonical `TenantCtx.attributes` keys for deploy-spec rollout identifiers
5/// (B11/C5) and the messaging endpoint discriminator (M1.4).
6///
7/// Previously exported by `greentic_types::telemetry::attr_keys`; the upstream
8/// module was removed in greentic-types 1.1.0-dev.27836473437 and the new
9/// `set_current_tenant_ctx` no longer projects these from the attributes map.
10/// The runner-host is the sole producer of these values, so the constants live
11/// here now. The string values are unchanged to keep existing telemetry
12/// dashboards and queries stable.
13pub(crate) mod attr_keys {
14    pub const CUSTOMER_ID: &str = "gt.customer_id";
15    pub const DEPLOYMENT_ID: &str = "gt.deployment_id";
16    pub const BUNDLE_ID: &str = "gt.bundle_id";
17    pub const REVISION_ID: &str = "gt.revision_id";
18    pub const PACK_ID: &str = "gt.pack_id";
19    pub const MESSAGING_ENDPOINT_ID: &str = "gt.messaging_endpoint_id";
20}
21use rand::{RngExt, rng};
22use std::str::FromStr;
23use tracing::Span;
24
25pub const PROVIDER_ID: &str = "greentic-runner";
26
27#[derive(Debug, Clone)]
28pub struct FlowSpanAttributes<'a> {
29    pub tenant: &'a str,
30    pub flow_id: &'a str,
31    pub node_id: Option<&'a str>,
32    pub tool: Option<&'a str>,
33    pub action: Option<&'a str>,
34}
35
36pub fn annotate_span(span: &Span, attrs: &FlowSpanAttributes<'_>) {
37    span.record("tenant", attrs.tenant);
38    span.record("flow_id", attrs.flow_id);
39    if let Some(node) = attrs.node_id {
40        span.record("node_id", node);
41    }
42    if let Some(tool) = attrs.tool {
43        span.record("tool", tool);
44    }
45    if let Some(action) = attrs.action {
46        span.record("action", action);
47    }
48}
49
50pub fn tenant_context(
51    env: &str,
52    tenant: &str,
53    flow_id: Option<&str>,
54    node_id: Option<&str>,
55    provider_id: Option<&str>,
56    session_id: Option<&str>,
57) -> TenantCtx {
58    let env_id = EnvId::from_str(env).expect("invalid env id");
59    let tenant_id = TenantId::from_str(tenant).expect("invalid tenant id");
60    let mut ctx = TenantCtx::new(env_id, tenant_id);
61    let provider = provider_id.unwrap_or(PROVIDER_ID);
62    ctx = ctx.with_provider(provider.to_string());
63    if let Some(flow) = flow_id {
64        ctx = ctx.with_flow(flow.to_string());
65    }
66    if let Some(node) = node_id {
67        ctx = ctx.with_node(node.to_string());
68    }
69    if let Some(session) = session_id {
70        ctx = ctx.with_session(session.to_string());
71    }
72    ctx
73}
74
75/// Build the per-invocation tenant context for a flow execution, stamping the
76/// resolved `pack_id` and any rollout identifiers onto `attributes`.
77///
78/// `pack_id` is always live (the engine knows it per invocation). The rollout
79/// IDs come from the engine's owning revision-keyed runtime and are empty until
80/// the Phase-D revision dispatcher constructs revision runtimes. Pure so the
81/// stamping can be unit-tested without the task-local slot.
82#[allow(clippy::too_many_arguments)]
83fn flow_tenant_ctx(
84    env: &str,
85    tenant: &str,
86    flow_id: &str,
87    node_id: Option<&str>,
88    provider_id: Option<&str>,
89    session_id: Option<&str>,
90    pack_id: &str,
91    rollout: &RolloutIds,
92) -> TenantCtx {
93    let mut ctx = tenant_context(env, tenant, Some(flow_id), node_id, provider_id, session_id);
94    if !pack_id.is_empty() {
95        ctx.attributes
96            .insert(attr_keys::PACK_ID.to_string(), pack_id.to_string());
97    }
98    stamp_rollout_ids(&mut ctx, rollout);
99    ctx
100}
101
102/// Project a [`TenantCtx`] into a [`greentic_telemetry::TelemetryCtx`] for span
103/// annotation export.
104///
105/// This replaces the removed `greentic_types::telemetry::tenant_ctx_to_telemetry`.
106/// The new `set_current_tenant_ctx` only copies the core identity fields into the
107/// task-local slot; it no longer projects rollout/pack/messaging attributes. The
108/// `annotate_span` export path still needs the full projection so that `gt.*`
109/// attributes reach exported OTLP spans.
110#[cfg(feature = "telemetry")]
111fn tenant_ctx_to_telemetry(ctx: &TenantCtx) -> greentic_telemetry::TelemetryCtx {
112    let mut t =
113        greentic_telemetry::TelemetryCtx::new(ctx.tenant_id.as_ref()).with_env(ctx.env.as_str());
114    if let Some(team) = ctx.team_id.as_ref().or(ctx.team.as_ref()) {
115        t = t.with_team(team.as_str());
116    }
117    if let Some(session) = ctx.session_id() {
118        t = t.with_session(session);
119    }
120    if let Some(flow) = ctx.flow_id() {
121        t = t.with_flow(flow);
122    }
123    if let Some(node) = ctx.node_id() {
124        t = t.with_node(node);
125    }
126    if let Some(provider) = ctx.provider_id() {
127        t = t.with_provider(provider);
128    }
129    if let Some(v) = ctx.attributes.get(attr_keys::CUSTOMER_ID) {
130        t = t.with_customer_id(v);
131    }
132    if let Some(v) = ctx.attributes.get(attr_keys::DEPLOYMENT_ID) {
133        t = t.with_deployment_id(v);
134    }
135    if let Some(v) = ctx.attributes.get(attr_keys::BUNDLE_ID) {
136        t = t.with_bundle_id(v);
137    }
138    if let Some(v) = ctx.attributes.get(attr_keys::REVISION_ID) {
139        t = t.with_revision_id(v);
140    }
141    if let Some(v) = ctx.attributes.get(attr_keys::PACK_ID) {
142        t = t.with_pack_id(v);
143    }
144    if let Some(v) = ctx.attributes.get(attr_keys::MESSAGING_ENDPOINT_ID) {
145        t = t.with_messaging_endpoint_id(v);
146    }
147    t
148}
149
150/// Install per-invocation telemetry for `span`: stamp the tenant context (live
151/// `pack_id` + rollout IDs) into the task-local slot, and export the `gt.*`
152/// attribution as real OpenTelemetry attributes on `span`.
153///
154/// The export step is load-bearing: greentic-telemetry's `ContextLayer` only
155/// stashes the task-local context in span extensions for capture layers — it
156/// does NOT record `gt.*` onto exported spans (`Span::record` is a no-op for
157/// callsite-unknown fields). The supported export primitive is
158/// `greentic_telemetry::annotate_span`, which sets the attributes directly on
159/// the owned span handle; without it the IDs never reach exported telemetry.
160#[allow(clippy::too_many_arguments)]
161pub fn set_flow_context(
162    span: &Span,
163    env: &str,
164    tenant: &str,
165    flow_id: &str,
166    node_id: Option<&str>,
167    provider_id: Option<&str>,
168    session_id: Option<&str>,
169    pack_id: &str,
170    rollout: &RolloutIds,
171) {
172    let ctx = flow_tenant_ctx(
173        env,
174        tenant,
175        flow_id,
176        node_id,
177        provider_id,
178        session_id,
179        pack_id,
180        rollout,
181    );
182    set_current_tenant_ctx(&ctx);
183    #[cfg(feature = "telemetry")]
184    greentic_telemetry::annotate_span(span, &tenant_ctx_to_telemetry(&ctx));
185    #[cfg(not(feature = "telemetry"))]
186    let _ = span;
187}
188
189/// Deploy-spec rollout identifiers stamped onto the per-invocation
190/// [`TenantCtx`] for telemetry attribution (B11). All optional — the producer
191/// (the revision dispatcher resolving a deployment/revision) is Phase D, so
192/// today these are `None` and [`stamp_rollout_ids`] is a no-op.
193#[derive(Debug, Clone, Default, PartialEq, Eq)]
194pub struct RolloutIds {
195    pub customer_id: Option<String>,
196    pub deployment_id: Option<String>,
197    pub bundle_id: Option<String>,
198    pub revision_id: Option<String>,
199}
200
201impl RolloutIds {
202    /// True when no identifier is set (the common case until Phase D wires the
203    /// dispatcher producer).
204    pub fn is_empty(&self) -> bool {
205        self.customer_id.is_none()
206            && self.deployment_id.is_none()
207            && self.bundle_id.is_none()
208            && self.revision_id.is_none()
209    }
210}
211
212/// Stamp the rollout IDs onto `ctx.attributes` under the canonical
213/// [`attr_keys`], so the local [`tenant_ctx_to_telemetry`] projection copies
214/// them into `TelemetryCtx` for span export.
215///
216/// Authoritative over these four keys: a present ID is written, an absent one
217/// is cleared. Stamping is therefore safe to re-run on a reused `TenantCtx`
218/// (e.g. a session that migrates between revisions) — an ID dropped on a
219/// re-stamp won't linger as a stale attribute from an earlier stamp.
220pub fn stamp_rollout_ids(ctx: &mut TenantCtx, ids: &RolloutIds) {
221    set_or_clear(ctx, attr_keys::CUSTOMER_ID, ids.customer_id.as_deref());
222    set_or_clear(ctx, attr_keys::DEPLOYMENT_ID, ids.deployment_id.as_deref());
223    set_or_clear(ctx, attr_keys::BUNDLE_ID, ids.bundle_id.as_deref());
224    set_or_clear(ctx, attr_keys::REVISION_ID, ids.revision_id.as_deref());
225}
226
227fn set_or_clear(ctx: &mut TenantCtx, key: &str, value: Option<&str>) {
228    match value {
229        Some(v) => {
230            ctx.attributes.insert(key.to_string(), v.to_string());
231        }
232        None => {
233            ctx.attributes.remove(key);
234        }
235    }
236}
237
238pub fn backoff_delay_ms(base: u64, attempt: u32) -> u64 {
239    let multiplier = 1_u64 << attempt.min(10);
240    let exp = base.saturating_mul(multiplier);
241    let mut rng = rng();
242    let jitter = rng.random_range(0..=exp.min(1000));
243    exp + jitter
244}
245
246#[cfg(test)]
247mod tests {
248    use super::*;
249
250    fn ctx() -> TenantCtx {
251        tenant_context("prod-eu", "acme", None, None, None, None)
252    }
253
254    #[test]
255    fn stamp_sets_present_ids_under_canonical_keys() {
256        let mut c = ctx();
257        let ids = RolloutIds {
258            customer_id: Some("cust-acme".into()),
259            deployment_id: Some("01JTKS".into()),
260            bundle_id: Some("customer.support".into()),
261            revision_id: Some("01JTKR".into()),
262        };
263        stamp_rollout_ids(&mut c, &ids);
264        assert_eq!(
265            c.attributes.get(attr_keys::CUSTOMER_ID).map(String::as_str),
266            Some("cust-acme")
267        );
268        assert_eq!(
269            c.attributes
270                .get(attr_keys::DEPLOYMENT_ID)
271                .map(String::as_str),
272            Some("01JTKS")
273        );
274        assert_eq!(
275            c.attributes.get(attr_keys::BUNDLE_ID).map(String::as_str),
276            Some("customer.support")
277        );
278        assert_eq!(
279            c.attributes.get(attr_keys::REVISION_ID).map(String::as_str),
280            Some("01JTKR")
281        );
282    }
283
284    #[test]
285    fn stamp_empty_is_noop() {
286        let mut c = ctx();
287        let before = c.attributes.len();
288        stamp_rollout_ids(&mut c, &RolloutIds::default());
289        assert_eq!(c.attributes.len(), before);
290        assert!(RolloutIds::default().is_empty());
291    }
292
293    #[test]
294    fn stamp_only_sets_present_subset() {
295        let mut c = ctx();
296        stamp_rollout_ids(
297            &mut c,
298            &RolloutIds {
299                deployment_id: Some("01JTKS".into()),
300                ..Default::default()
301            },
302        );
303        assert!(c.attributes.contains_key(attr_keys::DEPLOYMENT_ID));
304        assert!(!c.attributes.contains_key(attr_keys::CUSTOMER_ID));
305    }
306
307    #[test]
308    fn flow_ctx_stamps_pack_id_live() {
309        let ctx = flow_tenant_ctx(
310            "prod-eu",
311            "acme",
312            "support",
313            None,
314            None,
315            None,
316            "customer.support@1.2.0",
317            &RolloutIds::default(),
318        );
319        assert_eq!(
320            ctx.attributes.get(attr_keys::PACK_ID).map(String::as_str),
321            Some("customer.support@1.2.0")
322        );
323        // No revision runtime today → rollout IDs absent.
324        assert!(!ctx.attributes.contains_key(attr_keys::REVISION_ID));
325    }
326
327    #[test]
328    fn flow_ctx_stamps_pack_id_and_rollout_ids() {
329        let ctx = flow_tenant_ctx(
330            "prod-eu",
331            "acme",
332            "support",
333            None,
334            None,
335            None,
336            "customer.support@1.2.0",
337            &RolloutIds {
338                customer_id: Some("cust-acme".into()),
339                deployment_id: Some("01JTKS".into()),
340                bundle_id: Some("customer.support".into()),
341                revision_id: Some("01JTKR".into()),
342            },
343        );
344        assert_eq!(
345            ctx.attributes.get(attr_keys::PACK_ID).map(String::as_str),
346            Some("customer.support@1.2.0")
347        );
348        assert_eq!(
349            ctx.attributes
350                .get(attr_keys::REVISION_ID)
351                .map(String::as_str),
352            Some("01JTKR")
353        );
354        assert_eq!(
355            ctx.attributes
356                .get(attr_keys::DEPLOYMENT_ID)
357                .map(String::as_str),
358            Some("01JTKS")
359        );
360    }
361
362    #[test]
363    fn flow_ctx_skips_empty_pack_id() {
364        let ctx = flow_tenant_ctx(
365            "prod-eu",
366            "acme",
367            "support",
368            None,
369            None,
370            None,
371            "",
372            &RolloutIds::default(),
373        );
374        assert!(!ctx.attributes.contains_key(attr_keys::PACK_ID));
375    }
376
377    #[test]
378    fn stamp_clears_stale_ids_on_restamp() {
379        let mut c = ctx();
380        stamp_rollout_ids(
381            &mut c,
382            &RolloutIds {
383                customer_id: Some("cust-acme".into()),
384                deployment_id: Some("01JTKS".into()),
385                bundle_id: Some("customer.support".into()),
386                revision_id: Some("01JTKR".into()),
387            },
388        );
389        // Re-stamp with only a new revision (e.g. a session migrating revisions):
390        // the other three IDs must be cleared, not left stale from the first stamp.
391        stamp_rollout_ids(
392            &mut c,
393            &RolloutIds {
394                revision_id: Some("01JTKZ".into()),
395                ..Default::default()
396            },
397        );
398        assert_eq!(
399            c.attributes.get(attr_keys::REVISION_ID).map(String::as_str),
400            Some("01JTKZ")
401        );
402        assert!(!c.attributes.contains_key(attr_keys::CUSTOMER_ID));
403        assert!(!c.attributes.contains_key(attr_keys::DEPLOYMENT_ID));
404        assert!(!c.attributes.contains_key(attr_keys::BUNDLE_ID));
405    }
406}
407
408/// End-to-end export regression (C5.4): proves the stamped `gt.*` attribution
409/// actually reaches an **exported** OTLP span — not just the task-local slot.
410/// Guards against the failure mode where `set_current_tenant_ctx` is called but
411/// no `annotate_span` export happens, so production spans omit pack/rollout
412/// attribution while pure-context tests still pass.
413#[cfg(all(test, feature = "telemetry"))]
414mod export_tests {
415    use super::*;
416    use opentelemetry::trace::TracerProvider as _;
417    use opentelemetry_sdk::error::OTelSdkResult;
418    use opentelemetry_sdk::trace::{SdkTracerProvider, SpanData, SpanExporter};
419    use std::sync::{Arc, Mutex};
420    use tracing::subscriber;
421    use tracing_subscriber::prelude::*;
422
423    #[derive(Clone, Debug)]
424    struct TestExporter {
425        spans: Arc<Mutex<Vec<SpanData>>>,
426    }
427
428    impl SpanExporter for TestExporter {
429        fn export(
430            &self,
431            batch: Vec<SpanData>,
432        ) -> impl std::future::Future<Output = OTelSdkResult> + Send {
433            let spans = self.spans.clone();
434            async move {
435                spans
436                    .lock()
437                    .unwrap_or_else(|e| e.into_inner())
438                    .extend(batch);
439                Ok(())
440            }
441        }
442    }
443
444    fn attr_value(span: &SpanData, key: &str) -> Option<String> {
445        span.attributes
446            .iter()
447            .find(|kv| kv.key.as_str() == key)
448            .map(|kv| kv.value.to_string())
449    }
450
451    #[test]
452    fn set_flow_context_exports_pack_id_and_rollout_ids() {
453        let exported = Arc::new(Mutex::new(Vec::new()));
454        let provider = SdkTracerProvider::builder()
455            .with_simple_exporter(TestExporter {
456                spans: exported.clone(),
457            })
458            .build();
459        let tracer = provider.tracer("c5.4-flow-export");
460        let subscriber =
461            tracing_subscriber::registry().with(tracing_opentelemetry::layer().with_tracer(tracer));
462        let _guard = subscriber::set_default(subscriber);
463
464        // A span whose callsite declares NONE of the gt.* fields — exactly the
465        // `flow.execute` situation. `set_flow_context` must export them anyway.
466        let span = tracing::info_span!("flow.execute");
467        set_flow_context(
468            &span,
469            "prod-eu",
470            "acme",
471            "support",
472            None,
473            None,
474            Some("sess-1"),
475            "customer.support@1.2.0",
476            &RolloutIds {
477                customer_id: Some("cust-acme".into()),
478                deployment_id: Some("01JTKS".into()),
479                bundle_id: Some("customer.support".into()),
480                revision_id: Some("01JTKR".into()),
481            },
482        );
483        {
484            let _enter = span.enter();
485        }
486        drop(span);
487
488        let _ = provider.force_flush();
489        let _ = provider.shutdown();
490
491        let spans = exported.lock().unwrap_or_else(|e| e.into_inner());
492        // Assert on the named span, not just the last one — a future extra span
493        // from the subscriber stack would otherwise yield confusing failures.
494        let span = spans
495            .iter()
496            .find(|s| s.name == "flow.execute")
497            .expect("flow.execute span exported");
498        assert_eq!(
499            attr_value(span, attr_keys::PACK_ID).as_deref(),
500            Some("customer.support@1.2.0"),
501            "pack_id must reach the exported span"
502        );
503        assert_eq!(
504            attr_value(span, attr_keys::REVISION_ID).as_deref(),
505            Some("01JTKR")
506        );
507        assert_eq!(
508            attr_value(span, attr_keys::DEPLOYMENT_ID).as_deref(),
509            Some("01JTKS")
510        );
511        assert_eq!(
512            attr_value(span, attr_keys::BUNDLE_ID).as_deref(),
513            Some("customer.support")
514        );
515        assert_eq!(
516            attr_value(span, attr_keys::CUSTOMER_ID).as_deref(),
517            Some("cust-acme")
518        );
519    }
520}
greentic_runner_host/telemetry.rs

greentic_runner_host/
telemetry.rs
