Mountain/IPC/DevLog/
EmitOTLPSpan.rs

1//! Fire-and-forget OTLP span exporter. Sends a single
2//! `resourceSpans` payload over plain HTTP to the collector at
3//! `Pipe` (default `127.0.0.1:4318`, configurable via
4//! `.env.Land.PostHog`). Stops trying after the first failure
5//! (`OTLP_AVAILABLE` flips to `false`) so a missing collector
6//! doesn't tax every IPC call. Release builds are compiled out
7//! via `cfg!(debug_assertions)`. Honors the `Capture` master
8//! telemetry kill switch and the per-pipe `Emit` toggle.
9
10use std::{
11	collections::hash_map::DefaultHasher,
12	hash::{Hash, Hasher},
13	sync::{
14		OnceLock,
15		atomic::{AtomicBool, Ordering},
16	},
17};
18
19use crate::{Binary::Build::PostHogPlugin::Constants, IPC::DevLog::NowNano};
20
21static OTLP_AVAILABLE:AtomicBool = AtomicBool::new(true);
22
23static OTLP_TRACE_ID:OnceLock<String> = OnceLock::new();
24
25fn GetTraceId() -> &'static str {
26	OTLP_TRACE_ID.get_or_init(|| {
27		let mut H = DefaultHasher::new();
28
29		std::process::id().hash(&mut H);
30
31		NowNano::Fn().hash(&mut H);
32
33		format!("{:032x}", H.finish() as u128)
34	})
35}
36
37fn RandU64() -> u64 {
38	let mut H = DefaultHasher::new();
39
40	std::thread::current().id().hash(&mut H);
41
42	NowNano::Fn().hash(&mut H);
43
44	H.finish()
45}
46
47pub fn Fn(Name:&str, StartNano:u64, EndNano:u64, Attributes:&[(&str, &str)]) {
48	if !cfg!(debug_assertions) {
49		return;
50	}
51
52	if matches!(Constants::TELEMETRY_CAPTURE, "false" | "0" | "off") {
53		return;
54	}
55
56	if matches!(Constants::OTLP_ENABLED, "false" | "0" | "off") {
57		return;
58	}
59
60	if !OTLP_AVAILABLE.load(Ordering::Relaxed) {
61		return;
62	}
63
64	let SpanId = format!("{:016x}", RandU64());
65
66	let TraceId = GetTraceId().to_string();
67
68	let SpanName = Name.to_string();
69
70	let AttributesJson:Vec<String> = Attributes
71		.iter()
72		.map(|(K, V)| {
73			format!(
74				r#"{{"key":"{}","value":{{"stringValue":"{}"}}}}"#,
75				K,
76				V.replace('\\', "\\\\").replace('"', "\\\"")
77			)
78		})
79		.collect();
80
81	let IsError = SpanName.contains("error");
82
83	let StatusCode = if IsError { 2 } else { 1 };
84
85	let Payload = format!(
86		concat!(
87			r#"{{"resourceSpans":[{{"resource":{{"attributes":["#,
88			r#"{{"key":"service.name","value":{{"stringValue":"land-editor-mountain"}}}},"#,
89			r#"{{"key":"service.version","value":{{"stringValue":"0.0.1"}}}}"#,
90			r#"]}},"scopeSpans":[{{"scope":{{"name":"mountain.ipc","version":"1.0.0"}},"#,
91			r#""spans":[{{"traceId":"{}","spanId":"{}","name":"{}","kind":1,"#,
92			r#""startTimeUnixNano":"{}","endTimeUnixNano":"{}","#,
93			r#""attributes":[{}],"status":{{"code":{}}}}}]}}]}}]}}"#,
94		),
95		TraceId,
96		SpanId,
97		SpanName,
98		StartNano,
99		EndNano,
100		AttributesJson.join(","),
101		StatusCode,
102	);
103
104	// Resolve `Pipe` (e.g. `http://127.0.0.1:4318`) → host:port + path.
105	// Strip scheme, split on `/` for the path component if any, default to
106	// `/v1/traces`.
107	let (HostAddress, PathSegment) = ParseEndpoint(Constants::OTLP_ENDPOINT);
108
109	std::thread::spawn(move || {
110		use std::{
111			io::{Read as IoRead, Write as IoWrite},
112			net::TcpStream,
113			time::Duration,
114		};
115
116		let Ok(SocketAddress) = HostAddress.parse() else {
117			OTLP_AVAILABLE.store(false, Ordering::Relaxed);
118
119			return;
120		};
121
122		let Ok(mut Stream) = TcpStream::connect_timeout(&SocketAddress, Duration::from_millis(200)) else {
123			OTLP_AVAILABLE.store(false, Ordering::Relaxed);
124
125			return;
126		};
127
128		let _ = Stream.set_write_timeout(Some(Duration::from_millis(200)));
129
130		let _ = Stream.set_read_timeout(Some(Duration::from_millis(200)));
131
132		let HttpReq = format!(
133			"POST {} HTTP/1.1\r\nHost: {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: \
134			 close\r\n\r\n",
135			PathSegment,
136			HostAddress,
137			Payload.len()
138		);
139
140		if Stream.write_all(HttpReq.as_bytes()).is_err() {
141			return;
142		}
143
144		if Stream.write_all(Payload.as_bytes()).is_err() {
145			return;
146		}
147
148		let mut Buf = [0u8; 32];
149
150		let _ = Stream.read(&mut Buf);
151
152		if !(Buf.starts_with(b"HTTP/1.1 2") || Buf.starts_with(b"HTTP/1.0 2")) {
153			OTLP_AVAILABLE.store(false, Ordering::Relaxed);
154		}
155	});
156}
157
158/// Split `http://host:port/path` into `(host:port, /path)`. Defaults the
159/// path to `/v1/traces` when the endpoint has none. Returns owned `String`s
160/// so the spawned thread does not borrow the build-time `&'static str`.
161fn ParseEndpoint(Endpoint:&str) -> (String, String) {
162	let WithoutScheme = Endpoint
163		.strip_prefix("http://")
164		.or_else(|| Endpoint.strip_prefix("https://"))
165		.unwrap_or(Endpoint);
166
167	let (HostPort, Path) = match WithoutScheme.split_once('/') {
168		Some((HP, Rest)) => (HP.to_string(), format!("/{}", Rest.trim_start_matches('/'))),
169
170		None => (WithoutScheme.to_string(), "/v1/traces".to_string()),
171	};
172
173	let PathFinal = if Path == "/" { "/v1/traces".to_string() } else { Path };
174
175	(HostPort, PathFinal)
176}
Mountain/IPC/DevLog/EmitOTLPSpan.rs

Mountain/IPC/DevLog/
EmitOTLPSpan.rs
