1use crate::{
2 llm::actions::{
3 client_trait::{Client, ClientActionResult},
4 protocol_trait::Protocol,
5 ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
6 },
7 protocol::{log_template::LogTemplate, ConnectContext, EventType},
8 state::AppState,
9};
10use anyhow::{bail, Context, Result};
11use serde_json::{json, Value};
12use std::sync::LazyLock;
13fn field(name: &str, hint: &str, description: &str, required: bool) -> Parameter {
14 Parameter {
15 name: name.into(),
16 type_hint: hint.into(),
17 description: description.into(),
18 required,
19 }
20}
21fn event(id: &str, description: &str, parameters: Vec<Parameter>) -> EventType {
22 EventType::new(id, description, json!({"type":"wait_for_more"}))
23 .with_parameters(parameters)
24 .with_actions(OtlpClientProtocol.get_sync_actions())
25}
26pub static CONNECTED: LazyLock<EventType> = LazyLock::new(|| {
27 event(
28 "otlp_connected",
29 "OTLP transport connected",
30 vec![
31 field(
32 "remote_addr",
33 "string",
34 "Remote socket address of the OpenTelemetry receiver",
35 true,
36 ),
37 field(
38 "transport",
39 "string",
40 "Transport used for this export: grpc or http",
41 true,
42 ),
43 field(
44 "tls_verified",
45 "boolean",
46 "TLS certificate and server name verified",
47 true,
48 ),
49 field("server_name", "string", "Authenticated TLS name", false),
50 ],
51 )
52});
53pub static RESULT: LazyLock<EventType> = LazyLock::new(|| {
54 event(
55 "otlp_export_result",
56 "Export acknowledgement or gRPC failure; partial success is never retryable",
57 vec![
58 field("signal", "string", "traces, metrics or logs", true),
59 field("transport", "string", "Transport used for this export: grpc or http", true),
60 field("service_name", "string", "Exported service", true),
61 field("items", "number", "Exported item count", true),
62 field("result", "string", "accepted, partial_success or rejected", true),
63 field("rejected", "number", "Rejected item count on partial success", true),
64 field("message", "string", "Bounded status diagnostic", true),
65 field(
66 "retryable", "boolean",
67 "Whether an explicit later retry is appropriate; NetGet never retries automatically",
68 true,
69 ),
70 field("retry_after_secs", "number", "Receiver retry delay, if supplied", false),
71 field("http_status", "number", "HTTP status for HTTP transport", false),
72 field("grpc_code", "number", "Receiver or library gRPC status code", false),
73 ],
74 )
75});
76pub static ERROR: LazyLock<EventType> = LazyLock::new(|| {
77 event(
78 "otlp_export_error",
79 "Local transport, response decoding or export deadline failure",
80 vec![
81 field("action_type", "string", "Failed export action", true),
82 field("error", "string", "Bounded local error", true),
83 ],
84 )
85});
86#[derive(Default)]
87pub struct OtlpClientProtocol;
88impl OtlpClientProtocol {
89 pub fn new() -> Self {
90 Self
91 }
92}
93impl Protocol for OtlpClientProtocol {
94 fn protocol_name(&self) -> &'static str {
95 "OTLP"
96 }
97 fn stack_name(&self) -> &'static str {
98 "ETH>IP>TCP>HTTP>OTLP"
99 }
100 fn description(&self) -> &'static str {
101 "Bounded typed telemetry exporter over OTLP/gRPC or OTLP/HTTP protobuf"
102 }
103 fn keywords(&self) -> Vec<&'static str> {
104 vec![
105 "otlp",
106 "otel",
107 "opentelemetry",
108 "export telemetry",
109 "collector",
110 ]
111 }
112 fn group_name(&self) -> &'static str {
113 "Application"
114 }
115 fn example_prompt(&self) -> &'static str {
116 "Export a gauge data point to a trusted OpenTelemetry Collector"
117 }
118 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
119 self.get_sync_actions()
120 }
121 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
122 let common = || {
123 vec![
124 field("service_name", "string", "Service name, 1..256 bytes", true),
125 field(
126 "resource_attributes", "object",
127 "At most 32 flat scalar attributes; keys 256 bytes, strings 1024 bytes; use service_name for service.name",
128 false,
129 ),
130 field("scope_name", "string", "Instrumentation scope, at most 256 bytes; default netget", false),
131 ]
132 };
133 let action = |name: &str,
134 description: &str,
135 parameters: Vec<Parameter>,
136 example: Value,
137 log_message: &str| {
138 ActionDefinition {
139 name: name.into(),
140 description: description.into(),
141 parameters,
142 example,
143 log_template: Some(LogTemplate::new().with_info(log_message)),
144 }
145 };
146 let mut traces = common();
147 traces.push(field("spans","array","1..128 spans: name; nonzero trace_id (32 hex), span_id (16 hex), optional parent_span_id (16 hex); positive start_time_unix_nano/end_time_unix_nano u64 with end>=start; optional kind internal/server/client/producer/consumer, status unset/ok/error, status_message and flat attributes",true));
148 let mut logs = common();
149 logs.push(field("logs","array","1..128 records: body (1..4096 bytes), positive time_unix_nano u64; optional severity_number 0..24 (default 9), severity_text, paired trace_id/span_id and flat attributes",true));
150 let mut gauge = common();
151 gauge.extend([
152 field("name", "string", "Metric name, 1..256 bytes", true),
153 field("description", "string", "Optional gauge metric description, at most 256 bytes", false),
154 field("unit", "string", "Optional gauge metric unit, at most 64 bytes", false),
155 field(
156 "data_points", "array",
157 "1..128 gauge points: positive time_unix_nano u64, value (i64 or finite float), optional flat attributes",
158 true,
159 ),
160 ]);
161 vec![
162 action(
163 "export_otlp_traces",
164 "Export typed spans; no binary payload, span events or links",
165 traces,
166 json!({"type":"export_otlp_traces","service_name":"checkout","spans":[{"name":"charge card","trace_id":"0102030405060708090a0b0c0d0e0f10","span_id":"0102030405060708","start_time_unix_nano":1720000000000000000u64,"end_time_unix_nano":1720000000001000000u64,"kind":"client","status":"ok"}]}),
167 "Export OTLP traces for {service_name}",
168 ),
169 action(
170 "export_otlp_logs",
171 "Export UTF-8 text log records",
172 logs,
173 json!({"type":"export_otlp_logs","service_name":"checkout","logs":[{"body":"payment accepted","time_unix_nano":1720000000000000000u64,"severity_number":9}]}),
174 "Export OTLP logs for {service_name}",
175 ),
176 action(
177 "export_otlp_gauge",
178 "Export one gauge metric with typed numeric data points",
179 gauge,
180 json!({"type":"export_otlp_gauge","service_name":"checkout","name":"queue.depth","unit":"1","data_points":[{"value":3,"time_unix_nano":1720000000000000000u64}]}),
181 "Export OTLP gauge {name} for {service_name}",
182 ),
183 action(
184 "disconnect",
185 "Cancel exports and handlers, close the owned transport",
186 vec![],
187 json!({"type":"disconnect"}),
188 "Disconnect OTLP exporter",
189 ),
190 action(
191 "wait_for_more",
192 "Wait for another event or injected export",
193 vec![],
194 json!({"type":"wait_for_more"}),
195 "Wait for the next OTLP exporter event",
196 ),
197 ]
198 }
199 fn get_event_types(&self) -> Vec<EventType> {
200 vec![CONNECTED.clone(), RESULT.clone(), ERROR.clone()]
201 }
202 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
203 let p =
204 |name: &str, hint: &str, description: &str, example: Value, default: Option<Value>| {
205 ParameterDefinition {
206 name: name.into(),
207 type_hint: hint.into(),
208 description: description.into(),
209 required: false,
210 example,
211 default,
212 }
213 };
214 vec![
215 p(
216 "transport",
217 "string",
218 "grpc (default) or http; HTTP uses protobuf over HTTP/1.1",
219 json!("grpc"),
220 Some(json!(super::DEFAULT_TRANSPORT)),
221 ),
222 p(
223 "tls",
224 "boolean",
225 "Use authenticated TLS (default true); false explicitly selects cleartext",
226 json!(true),
227 Some(json!(super::DEFAULT_TLS)),
228 ),
229 p(
230 "gzip",
231 "boolean",
232 "Compress export requests (default false); responses support none/gzip",
233 json!(true),
234 Some(json!(super::DEFAULT_GZIP)),
235 ),
236 p(
237 "ca_cert_path",
238 "string",
239 "Operator PEM trust file, at most 1 MiB, added to public roots",
240 json!("collector-ca.pem"),
241 None,
242 ),
243 p(
244 "server_name",
245 "string",
246 "TLS certificate name override; default receiver hostname",
247 json!("collector.example"),
248 None,
249 ),
250 p(
251 "connect_timeout_secs",
252 "number",
253 "Whole DNS/TCP/TLS/transport connection deadline, 1..60 seconds",
254 json!(10),
255 Some(json!(super::CONNECT_TIMEOUT.as_secs())),
256 ),
257 p(
258 "export_timeout_secs",
259 "number",
260 "Whole export and response deadline including queue wait, 1..300 seconds",
261 json!(30),
262 Some(json!(super::EXPORT_TIMEOUT.as_secs())),
263 ),
264 p(
265 "idle_timeout_secs",
266 "number",
267 "Inactivity without an export or handler, 1..3600 seconds",
268 json!(120),
269 Some(json!(super::IDLE_TIMEOUT.as_secs())),
270 ),
271 ]
272 }
273 fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
274 use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
275 ProtocolMetadataV2::builder().state(DevelopmentState::Experimental)
276 .implementation("Generated opentelemetry-proto0.29/tonic0.12 unary gRPC exporter and owned hyper HTTP/1.1 protobuf connection; authenticated rustls TLS, optional custom CA and gzip")
277 .llm_control("Typed spans, UTF-8 logs and one gauge metric per export; semantic receiver verdicts, partial success and retry hints; injected actions and shared static/script/manual/model handlers")
278 .e2e_testing("tests/client/otlp requires official OpenTelemetry Collector0.162.0 for both transports and all three signals; direct NetGet pairing and transport/response/cancellation bounds")
279 .notes("128 items/export, 32 scalar attributes/object, 1 MiB encoded export, 4 MiB response before/after inflation, 16 total exports/handlers, four follow-up levels. Socket guard closes internally spawned tonic transport tasks. No automatic retry/reconnect, persistent telemetry, raw protobuf/JSON actions, histogram/sum/exemplar/profile export, span events/links, mTLS or proxy. Expanded subset remains Experimental")
280 .build()
281 }
282 fn get_startup_examples(&self) -> StartupExamples {
283 let base = || json!({"type":"open_client","protocol":"otlp","remote_addr":"collector.example:4317","startup_params":{"transport":"grpc","tls":true}});
284 let mut llm = base();
285 llm["instruction"]=json!("Export a gauge named queue.depth from checkout, with the current Unix timestamp in nanoseconds, then disconnect after its result. Never retry a partial success.");
286 let mut script = base();
287 script["event_handlers"] = json!([{"event_pattern":"*","handler":{"type":"script","language":"python","code":"import json,sys,time\nt=json.load(sys.stdin)['event_type_id']\na=[]\nif t=='otlp_connected': a=[{'type':'export_otlp_gauge','service_name':'checkout','name':'queue.depth','data_points':[{'value':3,'time_unix_nano':time.time_ns()}]}]\nelif t in ('otlp_export_result','otlp_export_error'): a=[{'type':'disconnect'}]\nprint(json.dumps({'actions':a}))"}}]);
288 let mut fixed = base();
289 fixed["event_handlers"] = json!([{"event_pattern":"otlp_connected","handler":{"type":"static","actions":[self.get_sync_actions()[2].example.clone()]}},{"event_pattern":"otlp_export_result","handler":{"type":"static","actions":[{"type":"disconnect"}]}},{"event_pattern":"otlp_export_error","handler":{"type":"static","actions":[{"type":"disconnect"}]}}]);
290 StartupExamples::new(llm, script, fixed)
291 }
292}
293impl Client for OtlpClientProtocol {
294 fn connect(
295 &self,
296 ctx: ConnectContext,
297 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
298 {
299 Box::pin(super::OtlpClient::connect(ctx))
300 }
301 fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
302 match action["type"].as_str().context("missing type")? {
303 "export_otlp_traces" | "export_otlp_logs" | "export_otlp_gauge" => {
304 super::wire::build(&action)?;
305 Ok(ClientActionResult::Custom {
306 name: "otlp_export".into(),
307 data: action,
308 })
309 }
310 "disconnect" => Ok(ClientActionResult::Disconnect),
311 "wait_for_more" => Ok(ClientActionResult::WaitForMore),
312 name => bail!("unknown OTLP client action: {name}"),
313 }
314 }
315}