platform_core/
telemetry.rs

1//
2// Copyright 2018-2026 Accenture Technology
3//
4// Licensed under the Apache License, Version 2.0 (the "License");
5// you may not use this file except in compliance with the License.
6// You may obtain a copy of the License at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15//
16
17//! Rust port of the Java `Telemetry` service
18//! (`org.platformlambda.core.services.Telemetry`) — the built-in
19//! **`distributed.tracing`** function, registered by the lifecycle's
20//! essential-services phase (the Java `EssentialServiceLoader`).
21//!
22//! When a traced function finishes, its route worker sends the
23//! performance-metrics dataset here: `{ trace: { id, span_id, parent_span_id,
24//! service, path, from, origin, start, exec_time, success, status,
25//! exception? }, annotations: {...} }`. By default the dataset is **logged**
26//! (the real-time telemetry stream); two reserved extension routes forward it
27//! elsewhere:
28//!
29//! - **`distributed.trace.forwarder`** — register a function here (e.g. an
30//!   OpenTelemetry OTLP exporter) and every dataset is forwarded to it;
31//! - **`transaction.journal.recorder`** — receives request/response journals
32//!   when journaling is enabled (payloads may contain PII/PHI/PCI — handle per
33//!   your organization's security policy).
34//!
35//! Reserved for system use — do not call this route from application code.
36
37use std::collections::HashMap;
38
39use async_trait::async_trait;
40
41use crate::envelope::EventEnvelope;
42use crate::function::{AppError, ComposableFunction};
43use crate::platform::Platform;
44
45pub const DISTRIBUTED_TRACING: &str = "distributed.tracing";
46pub const DISTRIBUTED_TRACE_FORWARDER: &str = "distributed.trace.forwarder";
47pub const TRANSACTION_JOURNAL_RECORDER: &str = "transaction.journal.recorder";
48
49/// Routes that must never appear as a traced service — the telemetry
50/// plumbing itself and the RPC reply listener (Java `ZERO_TRACING_FILTER`;
51/// exact names only, no prefix matching).
52pub const ZERO_TRACING_FILTER: [&str; 4] = [
53    DISTRIBUTED_TRACING,
54    DISTRIBUTED_TRACE_FORWARDER,
55    TRANSACTION_JOURNAL_RECORDER,
56    crate::inbox::TEMPORARY_INBOX,
57];
58
59/// The built-in telemetry sink. Holds a handle to its own platform so it can
60/// forward datasets to the optional extension routes.
61pub struct Telemetry {
62    platform: Platform,
63}
64
65impl Telemetry {
66    pub fn new(platform: &Platform) -> Self {
67        Telemetry {
68            platform: platform.clone(),
69        }
70    }
71}
72
73#[async_trait]
74impl ComposableFunction for Telemetry {
75    async fn handle_event(
76        &self,
77        _headers: HashMap<String, String>,
78        input: EventEnvelope,
79        _instance: usize,
80    ) -> Result<EventEnvelope, AppError> {
81        let Ok(payload) = input.body_as::<serde_json::Value>() else {
82            return Ok(EventEnvelope::new());
83        };
84        let Some(payload) = payload.as_object() else {
85            return Ok(EventEnvelope::new());
86        };
87        let mut metrics = match payload.get("trace").and_then(|t| t.as_object()) {
88            Some(m) if !m.is_empty() => m.clone(),
89            _ => return Ok(EventEnvelope::new()),
90        };
91        // filter the telemetry plumbing itself; trim any "@origin" suffix
92        let Some(service) = permitted_route(metrics.get("service")) else {
93            return Ok(EventEnvelope::new());
94        };
95        metrics.insert("service".to_string(), serde_json::Value::String(service));
96        if let Some(from) = metrics.get("from").and_then(|f| f.as_str()) {
97            let trimmed = trim_origin(from).to_string();
98            metrics.insert("from".to_string(), serde_json::Value::String(trimmed));
99        }
100        let annotations = payload
101            .get("annotations")
102            .and_then(|a| a.as_object())
103            .cloned()
104            .unwrap_or_default();
105        let mut dataset = serde_json::Map::new();
106        dataset.insert("trace".to_string(), serde_json::Value::Object(metrics));
107        if !annotations.is_empty() {
108            dataset.insert(
109                "annotations".to_string(),
110                serde_json::Value::Object(annotations),
111            );
112        }
113        let dataset = serde_json::Value::Object(dataset);
114        // the default sink: the real-time telemetry log stream
115        log::info!("{dataset}");
116        // optional forwarders (must be registered in the same application)
117        if self.platform.has_route(DISTRIBUTED_TRACE_FORWARDER) {
118            let event = EventEnvelope::new()
119                .set_to(DISTRIBUTED_TRACE_FORWARDER)
120                .set_body(&dataset)?;
121            let _ = self
122                .platform
123                .deliver(DISTRIBUTED_TRACE_FORWARDER, event)
124                .await;
125        }
126        if payload.contains_key("journal") && self.platform.has_route(TRANSACTION_JOURNAL_RECORDER)
127        {
128            let mut forward = dataset;
129            if let (Some(map), Some(journal)) = (forward.as_object_mut(), payload.get("journal")) {
130                map.insert("journal".to_string(), journal.clone());
131            }
132            let event = EventEnvelope::new()
133                .set_to(TRANSACTION_JOURNAL_RECORDER)
134                .set_body(&forward)?;
135            let _ = self
136                .platform
137                .deliver(TRANSACTION_JOURNAL_RECORDER, event)
138                .await;
139        }
140        Ok(EventEnvelope::new())
141    }
142}
143
144/// Extract the service route, dropping the telemetry plumbing routes and
145/// trimming any `@origin` suffix (Java `getPermittedRoute`).
146fn permitted_route(service: Option<&serde_json::Value>) -> Option<String> {
147    let route = service?.as_str()?;
148    let name = trim_origin(route);
149    if ZERO_TRACING_FILTER.contains(&name) {
150        None
151    } else {
152        Some(name.to_string())
153    }
154}
155
156fn trim_origin(route: &str) -> &str {
157    match route.find('@') {
158        Some(at) => &route[..at],
159        None => route,
160    }
161}
162
163#[cfg(test)]
164mod tests {
165    use super::*;
166
167    #[test]
168    fn plumbing_routes_are_filtered() {
169        let v = |s: &str| serde_json::Value::String(s.to_string());
170        assert_eq!(
171            permitted_route(Some(&v("v1.hello"))),
172            Some("v1.hello".into())
173        );
174        assert_eq!(
175            permitted_route(Some(&v("v1.hello@abc123"))),
176            Some("v1.hello".into())
177        );
178        assert_eq!(permitted_route(Some(&v("distributed.tracing"))), None);
179        assert_eq!(
180            permitted_route(Some(&v("distributed.trace.forwarder@x"))),
181            None
182        );
183        assert_eq!(permitted_route(None), None);
184    }
185}
platform_core/telemetry.rs

platform_core/
telemetry.rs
