Skip to content
Sign up

flagon-io/g1t

public
0
836 linesCodeBlameRaw
1//! Observability over REST and MCP: what a project's running software did,
2//! as requests, logs, traces, errors and metrics, and how it is collected
3//! (settings, ingest tokens and exports).
4//!
5//! The telemetry service keeps each project's store and checks nothing:
6//! this resolves the project through the projects service and decides
7//! first, as the site does. Reading takes a role of the viewer's own on
8//! the project's repository (logs and request addresses can hold private
9//! data, so a public repository is not enough); resolving and ignoring
10//! errors takes Triage; settings, ingest tokens, exports and usage take
11//! Admin (`manage_integrations`). Anyone who may not read is told there is
12//! no such project.
13
14use g1t_contracts::access::{self, Capability, RepoRef};
15use g1t_contracts::{FailureCode, Outcome, Viewer};
16use serde_json::{Map, Value, json};
17use worker::{Headers, Method, Request, RequestInit, Result};
18
19use crate::operations::Services;
20
21/// One operation on a project's observability.
22#[derive(Clone, Copy, Debug, PartialEq, Eq)]
23pub enum TelemetryOp {
24 GetOverview,
25 ListRequests,
26 GetRequest,
27 ListLogs,
28 ListTraces,
29 GetTrace,
30 ListErrors,
31 UpdateError,
32 ListMetrics,
33 GetMetric,
34 GetSettings,
35 UpdateSettings,
36 ListTokens,
37 CreateToken,
38 DeleteToken,
39 ListExports,
40 CreateExport,
41 UpdateExport,
42 DeleteExport,
43 GetUsage,
44}
45
46/// Log levels, as the telemetry service groups them.
47const SEVERITIES: [&str; 6] = ["trace", "debug", "info", "warn", "error", "fatal"];
48/// An error group's states.
49const ERROR_STATES: [&str; 3] = ["open", "resolved", "ignored"];
50/// What a metric's values in a step may be reduced to.
51const AGGREGATES: [&str; 7] = ["avg", "sum", "min", "max", "p50", "p95", "p99"];
52/// The signals an export may send.
53const SIGNALS: [&str; 3] = ["traces", "logs", "metrics"];
54/// Retention a project can choose, in days (`RETENTION_CHOICES` in @g1t/contracts).
55const RETENTION_DAYS: [u64; 5] = [1, 3, 7, 14, 30];
56
57/// Fields keyed by what the software sent (OpenTelemetry attributes and
58/// resources), passed through as they are rather than put in `snake_case`.
59pub const AS_GIVEN: &[&str] = &["attributes", "resource"];
60
61/// What an operation needs of the viewer on the project's repository.
62#[derive(Clone, Copy, Debug, PartialEq, Eq)]
63enum Need {
64 /// A role of their own: a member, a collaborator, or the workspace's token.
65 Insider,
66 Can(Capability),
67}
68
69impl TelemetryOp {
70 /// Every one: `Op::ALL` lists each as `Op::Telemetry(…)`, which a test
71 /// checks against this.
72 #[cfg(test)]
73 pub const ALL: [TelemetryOp; 20] = [
74 TelemetryOp::GetOverview,
75 TelemetryOp::ListRequests,
76 TelemetryOp::GetRequest,
77 TelemetryOp::ListLogs,
78 TelemetryOp::ListTraces,
79 TelemetryOp::GetTrace,
80 TelemetryOp::ListErrors,
81 TelemetryOp::UpdateError,
82 TelemetryOp::ListMetrics,
83 TelemetryOp::GetMetric,
84 TelemetryOp::GetSettings,
85 TelemetryOp::UpdateSettings,
86 TelemetryOp::ListTokens,
87 TelemetryOp::CreateToken,
88 TelemetryOp::DeleteToken,
89 TelemetryOp::ListExports,
90 TelemetryOp::CreateExport,
91 TelemetryOp::UpdateExport,
92 TelemetryOp::DeleteExport,
93 TelemetryOp::GetUsage,
94 ];
95
96 pub fn name(self) -> &'static str {
97 match self {
98 TelemetryOp::GetOverview => "get_observability_overview",
99 TelemetryOp::ListRequests => "list_requests",
100 TelemetryOp::GetRequest => "get_request",
101 TelemetryOp::ListLogs => "list_logs",
102 TelemetryOp::ListTraces => "list_traces",
103 TelemetryOp::GetTrace => "get_trace",
104 TelemetryOp::ListErrors => "list_errors",
105 TelemetryOp::UpdateError => "update_error",
106 TelemetryOp::ListMetrics => "list_metrics",
107 TelemetryOp::GetMetric => "get_metric",
108 TelemetryOp::GetSettings => "get_observability_settings",
109 TelemetryOp::UpdateSettings => "update_observability_settings",
110 TelemetryOp::ListTokens => "list_ingest_tokens",
111 TelemetryOp::CreateToken => "create_ingest_token",
112 TelemetryOp::DeleteToken => "delete_ingest_token",
113 TelemetryOp::ListExports => "list_telemetry_exports",
114 TelemetryOp::CreateExport => "create_telemetry_export",
115 TelemetryOp::UpdateExport => "update_telemetry_export",
116 TelemetryOp::DeleteExport => "delete_telemetry_export",
117 TelemetryOp::GetUsage => "get_observability_usage",
118 }
119 }
120
121 /// For the API reference: "List logs".
122 pub fn title(self) -> &'static str {
123 match self {
124 TelemetryOp::GetOverview => "Get a project's observability overview",
125 TelemetryOp::ListRequests => "List requests",
126 TelemetryOp::GetRequest => "Get a request",
127 TelemetryOp::ListLogs => "List logs",
128 TelemetryOp::ListTraces => "List traces",
129 TelemetryOp::GetTrace => "Get a trace",
130 TelemetryOp::ListErrors => "List errors",
131 TelemetryOp::UpdateError => "Update an error",
132 TelemetryOp::ListMetrics => "List metrics",
133 TelemetryOp::GetMetric => "Get a metric",
134 TelemetryOp::GetSettings => "Get observability settings",
135 TelemetryOp::UpdateSettings => "Update observability settings",
136 TelemetryOp::ListTokens => "List ingest tokens",
137 TelemetryOp::CreateToken => "Create an ingest token",
138 TelemetryOp::DeleteToken => "Delete an ingest token",
139 TelemetryOp::ListExports => "List telemetry exports",
140 TelemetryOp::CreateExport => "Create a telemetry export",
141 TelemetryOp::UpdateExport => "Update a telemetry export",
142 TelemetryOp::DeleteExport => "Delete a telemetry export",
143 TelemetryOp::GetUsage => "Get observability usage",
144 }
145 }
146
147 pub fn description(self) -> &'static str {
148 match self {
149 TelemetryOp::GetOverview => "A project's numbers over a period (the last 24 hours unless you say): totals (requests, errors, error_rate, and latency and CPU percentiles p50, p75, p95 and p99 in milliseconds) and the same for the period before it, to compare; points, one per step (step_s seconds), with requests, errors, counts by status class and percentiles; routes, by requests; deployments, each one's numbers newest first, so a regression points at the deployment it began with; services and regions. from and to are RFC 3339 times or Unix milliseconds. Narrow it by environment, service or deployment_id. Needs a role on the project's repository.",
150 TelemetryOp::ListRequests => "The requests (and other invocations) a project's software answered, newest first: each with its method, url, route (ids taken out), status, outcome, duration_us, cpu_ms, region, trace_id, log_count, error, and where it came from (service, environment, deployment_id, commit_sha, source). Apps on g1t.page send them with nothing to set up; software elsewhere, as OpenTelemetry server spans. Filter by words in q, status (2xx, 3xx, 4xx, 5xx or a status, comma-separated), errors_only (a 5xx or an outcome other than ok), route, min_duration_ms, environment, service and deployment_id, between from and to. A page holds limit; pass next back as cursor for the next. Needs a role on the project's repository.",
151 TelemetryOp::GetRequest => "One request by id, with every log line it wrote, oldest first. Needs a role on the project's repository.",
152 TelemetryOp::ListLogs => "Search a project's logs, newest first. q finds lines holding all of its words, or a phrase in quotes. level narrows to levels (trace, debug, info, warn, error, fatal, comma-separated); trace_id and request_id to one trace's or one request's lines; environment, service and deployment_id to where they ran; from and to to a period. Each line has its time_us, severity, body, attributes, trace_id, span_id and request_id. A page holds limit; pass next back as cursor for the next. Needs a role on the project's repository.",
153 TelemetryOp::ListTraces => "A project's traces, newest first: each its root span's name, start_us, duration_us, span_count, error_count, the services it crossed and its HTTP status. Filter by words in q, errors_only, min_duration_ms, environment and service, between from and to. A page holds limit; pass next back as cursor for the next. Needs a role on the project's repository.",
154 TelemetryOp::GetTrace => "One trace by its id (32 hex characters): every span, with its parent, kind, timing, status, attributes, resource, events and links, and the log lines written under it. Needs a role on the project's repository.",
155 TelemetryOp::ListErrors => "A project's errors, grouped: the same exception (its name, its message with numbers and ids taken out, and the top frame of its stack) is one group, with its count, first and last time seen, the deployment it was first seen on (what likely brought it in), its last request and trace, its state (open, resolved or ignored) and the issue made for it. Most recently seen first, at most 200. Filter by state, environment and service, between from and to. Needs a role on the project's repository.",
156 TelemetryOp::UpdateError => "Resolve, ignore or reopen an error group (state), or link it to the issue made for it (issue_number; null unlinks it). A resolved group that happens again opens again. Needs the Triage role or higher.",
157 TelemetryOp::ListMetrics => "The metrics a project's software has sent: each with its name, kind (gauge, sum, histogram or summary), unit, description, the services that send it and when it last did. Needs a role on the project's repository.",
158 TelemetryOp::GetMetric => "One metric over time, one value per step: gauges averaged in each step (or min or max), sums added (a cumulative sum becomes its increase), histograms their mean or a percentile (p50, p95, p99) from their buckets. group_by splits it into a series per value of an attribute (or service). The last 24 hours unless from and to say. Needs a role on the project's repository.",
159 TelemetryOp::GetSettings => "How long a project's requests, logs and spans are kept (retention_days; charts are kept longer, as per-minute counts) and whether its apps on g1t.page send telemetry (deployments). Needs the Admin role.",
160 TelemetryOp::UpdateSettings => "Change how long a project's telemetry is kept (retention_days: 1, 3, 7, 14 or 30) and whether its apps on g1t.page send it (deployments; off keeps the tokens and history). Only what you give changes. Needs the Admin role.",
161 TelemetryOp::ListTokens => "A project's ingest tokens, which OpenTelemetry SDKs and collectors send with: each its name, hint (its first characters), who made it, when, and when it was last used. managed marks the one g1t made for its own deployments, which people cannot delete. Secrets are never returned. Needs the Admin role.",
162 TelemetryOp::CreateToken => "Make an ingest token for a project. Returns token and secret: the secret is shown this once, so keep it. Send it as `Authorization: Bearer <secret>` (or `x-g1t-token`) with OTLP over HTTP to the project's endpoint. Needs the Admin role.",
163 TelemetryOp::DeleteToken => "Delete an ingest token: what sends with it is refused from then on. The token g1t made for its own deployments cannot be deleted. Needs the Admin role.",
164 TelemetryOp::ListExports => "Where a project's telemetry is also sent as it arrives: each export's name, OTLP/HTTP endpoint, signals (traces, logs, metrics), header_names (values are never returned), whether it is enabled, and its last error and send. Needs the Admin role.",
165 TelemetryOp::CreateExport => "Send a project's telemetry on to any OTLP/HTTP endpoint as it arrives: a collector, or a vendor that takes OTLP. signals are traces, logs (requests go as log records) and metrics, all three unless you say; headers are sent with each request, such as an API key, and are written, never read back. Needs the Admin role.",
166 TelemetryOp::UpdateExport => "Change an export: its name, endpoint, signals and whether it is enabled. Only what you give changes. Headers given are set; a header given as \"\" is removed; those not given are kept. Needs the Admin role.",
167 TelemetryOp::DeleteExport => "Stop sending a project's telemetry to an export. Needs the Admin role.",
168 TelemetryOp::GetUsage => "What a project has sent and kept this month: ingested_bytes (what ingest is metered by), requests, logs, spans, metric_points and stored_bytes. Needs the Admin role.",
169 }
170 }
171
172 /// Whether it changes anything.
173 pub fn writes(self) -> bool {
174 matches!(
175 self,
176 TelemetryOp::UpdateError
177 | TelemetryOp::UpdateSettings
178 | TelemetryOp::CreateToken
179 | TelemetryOp::DeleteToken
180 | TelemetryOp::CreateExport
181 | TelemetryOp::UpdateExport
182 | TelemetryOp::DeleteExport
183 )
184 }
185
186 fn need(self) -> Need {
187 match self {
188 TelemetryOp::UpdateError => Need::Can(Capability::Triage),
189 TelemetryOp::GetSettings
190 | TelemetryOp::UpdateSettings
191 | TelemetryOp::ListTokens
192 | TelemetryOp::CreateToken
193 | TelemetryOp::DeleteToken
194 | TelemetryOp::ListExports
195 | TelemetryOp::CreateExport
196 | TelemetryOp::UpdateExport
197 | TelemetryOp::DeleteExport
198 | TelemetryOp::GetUsage => Need::Can(Capability::ManageIntegrations),
199 _ => Need::Insider,
200 }
201 }
202
203 pub fn input(self) -> Value {
204 let text = |what: &str| json!({ "type": "string", "description": what });
205 let integer = |what: &str| json!({ "type": "integer", "description": what });
206 let mut properties = Map::new();
207 properties.insert("workspace".into(), text("The workspace's slug, e.g. \"flagon-io\"."));
208 properties.insert("project".into(), text("The project's slug, as in g1t.sh/{workspace}/{project}."));
209 let mut required = vec!["workspace", "project"];
210 let mut add = |name: &str, schema: Value| {
211 properties.insert(name.to_owned(), schema);
212 };
213 let period = |add: &mut dyn FnMut(&str, Value)| {
214 add("from", text("The start of the period: an RFC 3339 time or Unix milliseconds. 24 hours before `to` unless you say."));
215 add("to", text("The end of the period: an RFC 3339 time or Unix milliseconds. Now unless you say."));
216 };
217 let place = |add: &mut dyn FnMut(&str, Value)| {
218 add("environment", text("Only what ran in this environment, such as production or preview; several comma-separated."));
219 add("service", text("Only what this service sent (OpenTelemetry's service.name; an app on g1t.page is its project's name); several comma-separated."));
220 };
221 let page = |add: &mut dyn FnMut(&str, Value)| {
222 add("cursor", text("Where to carry on from: the last page's `next`."));
223 add("limit", integer("How many a page holds."));
224 };
225 let words = "Words to find, all of them, or a phrase in quotes.";
226 match self {
227 TelemetryOp::GetOverview => {
228 period(&mut add);
229 place(&mut add);
230 add("deployment_id", text("Only this deployment's (dpl_…)."));
231 add("step", integer("Seconds per point of the charts; chosen from the period unless you say."));
232 }
233 TelemetryOp::ListRequests => {
234 period(&mut add);
235 add("q", text(words));
236 add("status", text("Only these statuses: 2xx, 3xx, 4xx, 5xx or a status such as 404, comma-separated."));
237 add("errors_only", json!({ "type": "boolean", "description": "Only those that failed: a 5xx, or an outcome other than ok." }));
238 add("route", text("Only this route, as the list shows it, such as /users/:id."));
239 add("min_duration_ms", integer("Only those that took at least this many milliseconds."));
240 place(&mut add);
241 add("deployment_id", text("Only this deployment's (dpl_…)."));
242 page(&mut add);
243 }
244 TelemetryOp::GetRequest => {
245 add("id", text("The request's id."));
246 required.push("id");
247 }
248 TelemetryOp::ListLogs => {
249 period(&mut add);
250 add("q", text(words));
251 add("level", json!({ "type": "string", "description": "Only these levels: trace, debug, info, warn, error or fatal, comma-separated." }));
252 add("trace_id", text("Only the lines written under this trace."));
253 add("request_id", text("Only the lines this request wrote."));
254 place(&mut add);
255 add("deployment_id", text("Only this deployment's (dpl_…)."));
256 page(&mut add);
257 }
258 TelemetryOp::ListTraces => {
259 period(&mut add);
260 add("q", text(words));
261 add("errors_only", json!({ "type": "boolean", "description": "Only traces with a span that failed." }));
262 add("min_duration_ms", integer("Only traces that took at least this many milliseconds."));
263 place(&mut add);
264 page(&mut add);
265 }
266 TelemetryOp::GetTrace => {
267 add("trace_id", text("The trace's id, 32 hex characters."));
268 required.push("trace_id");
269 }
270 TelemetryOp::ListErrors => {
271 add("state", json!({ "type": "string", "enum": ERROR_STATES, "description": "Only groups in this state." }));
272 place(&mut add);
273 period(&mut add);
274 }
275 TelemetryOp::UpdateError => {
276 add("fingerprint", text("The error group's fingerprint, as list_errors gives it."));
277 add("state", json!({ "type": "string", "enum": ERROR_STATES, "description": "Its state now: resolved, ignored, or open again." }));
278 add("issue_number", json!({ "type": ["integer", "null"], "description": "The issue made for it; null unlinks it." }));
279 required.push("fingerprint");
280 }
281 TelemetryOp::ListMetrics
282 | TelemetryOp::GetSettings
283 | TelemetryOp::ListTokens
284 | TelemetryOp::ListExports
285 | TelemetryOp::GetUsage => {}
286 TelemetryOp::GetMetric => {
287 add("metric", text("The metric's name, such as http.server.request.duration."));
288 required.push("metric");
289 period(&mut add);
290 add("aggregate", json!({ "type": "string", "enum": AGGREGATES, "description": "How each step's values are reduced: avg for gauges and sum for sums unless you say; a percentile for histograms." }));
291 add("group_by", text("An attribute (or service) to split it into a series per value."));
292 add("step", integer("Seconds per point; chosen from the period unless you say."));
293 place(&mut add);
294 }
295 TelemetryOp::UpdateSettings => {
296 add("retention_days", json!({ "type": "integer", "enum": RETENTION_DAYS, "description": "Days requests, logs and spans are kept." }));
297 add("deployments", json!({ "type": "boolean", "description": "Whether the project's apps on g1t.page send telemetry." }));
298 }
299 TelemetryOp::CreateToken => {
300 add("name", text("What it is for, such as the machine or collector that sends with it."));
301 required.push("name");
302 }
303 TelemetryOp::DeleteToken => {
304 add("id", text("The token's id."));
305 required.push("id");
306 }
307 TelemetryOp::CreateExport | TelemetryOp::UpdateExport => {
308 if self == TelemetryOp::UpdateExport {
309 add("id", text("The export's id."));
310 required.push("id");
311 } else {
312 required.extend(["name", "endpoint"]);
313 }
314 add("name", text("Its name, for people."));
315 add("endpoint", text("The OTLP/HTTP base address, such as https://otlp.example.com; /v1/traces and the rest are added."));
316 add("signals", json!({ "type": "array", "items": { "type": "string", "enum": SIGNALS }, "description": "What it sends: traces, logs and metrics. All three unless you say." }));
317 add("headers", json!({ "type": "object", "additionalProperties": { "type": "string" }, "description": "Headers sent with each request, such as an API key. Written, never read back; \"\" removes one." }));
318 add("enabled", json!({ "type": "boolean", "description": "Whether it sends. True unless you say." }));
319 }
320 TelemetryOp::DeleteExport => {
321 add("id", text("The export's id."));
322 required.push("id");
323 }
324 }
325 json!({ "type": "object", "properties": properties, "required": required })
326 }
327}
328
329/// A non-empty string field, trimmed.
330fn text(input: &Value, key: &str) -> Option<String> {
331 input[key].as_str().map(str::trim).filter(|value| !value.is_empty()).map(str::to_owned)
332}
333
334/// A number given as a number or as text.
335fn number(input: &Value, key: &str) -> std::result::Result<Option<u64>, String> {
336 match &input[key] {
337 Value::Null => Ok(None),
338 Value::Number(number) => number.as_u64().map(Some).ok_or_else(|| format!("{key} must be a whole number.")),
339 Value::String(digits) if digits.trim().is_empty() => Ok(None),
340 Value::String(digits) => digits.trim().parse().map(Some).map_err(|_| format!("{key} must be a whole number.")),
341 _ => Err(format!("{key} must be a whole number.")),
342 }
343}
344
345/// A flag given as a boolean or as text.
346fn flag(input: &Value, key: &str) -> std::result::Result<Option<bool>, String> {
347 match &input[key] {
348 Value::Null => Ok(None),
349 Value::Bool(value) => Ok(Some(*value)),
350 Value::String(text) => match text.trim() {
351 "" => Ok(None),
352 "true" | "1" => Ok(Some(true)),
353 "false" | "0" => Ok(Some(false)),
354 _ => Err(format!("{key} is true or false.")),
355 },
356 _ => Err(format!("{key} is true or false.")),
357 }
358}
359
360/// A list given as an array or as comma-separated text.
361fn list(input: &Value, key: &str) -> Vec<String> {
362 match &input[key] {
363 Value::Array(items) => items.iter().filter_map(Value::as_str).map(str::trim).filter(|item| !item.is_empty()).map(str::to_owned).collect(),
364 Value::String(text) => text.split(',').map(str::trim).filter(|item| !item.is_empty()).map(str::to_owned).collect(),
365 _ => Vec::new(),
366 }
367}
368
369/// A time as microseconds since the epoch, from Unix milliseconds or an
370/// RFC 3339 time, in UTC (`Z`) or with an offset (`+02:00`).
371pub(crate) fn micros(given: &Value) -> Option<u64> {
372 let text = match given {
373 Value::Number(number) => return number.as_u64().map(|ms| ms * 1000),
374 Value::String(text) => text.trim(),
375 _ => return None,
376 };
377 if !text.is_empty() && text.bytes().all(|byte| byte.is_ascii_digit()) {
378 return text.parse::<u64>().ok().map(|ms| ms * 1000);
379 }
380 if let Some(ms) = g1t_contracts::time::parse_rfc3339(text) {
381 return Some(ms * 1000);
382 }
383 // An offset: the time there, less the offset, is UTC.
384 let at = text.len().checked_sub(6)?;
385 let (local, offset) = text.split_at(at);
386 let sign: i64 = match offset.as_bytes().first() {
387 Some(b'+') => 1,
388 Some(b'-') => -1,
389 _ => return None,
390 };
391 let (hours, minutes) = offset[1..].split_once(':')?;
392 let shift = sign * (hours.parse::<i64>().ok()? * 60 + minutes.parse::<i64>().ok()?) * 60_000;
393 let ms = g1t_contracts::time::parse_rfc3339(&format!("{local}Z"))? as i64 - shift;
394 u64::try_from(ms).ok().map(|ms| ms * 1000)
395}
396
397/// The period and the place every list narrows by, in the service's names.
398fn filter(input: &Value, out: &mut Map<String, Value>) -> std::result::Result<(), String> {
399 for (key, field) in [("from", "from_us"), ("to", "to_us")] {
400 let given = &input[key];
401 if given.is_null() || given.as_str().is_some_and(|text| text.trim().is_empty()) {
402 continue;
403 }
404 let at = micros(given).ok_or_else(|| format!("{key} must be an RFC 3339 time or Unix milliseconds."))?;
405 out.insert(field.into(), at.into());
406 }
407 for (key, field) in [("environment", "environments"), ("service", "services")] {
408 let values = list(input, key);
409 if !values.is_empty() {
410 out.insert(field.into(), json!(values));
411 }
412 }
413 Ok(())
414}
415
416/// The arguments the telemetry service takes for `op`, from the
417/// operation's input, without the project; `Err` says what is missing or
418/// wrong. `actor` is who makes a token.
419pub(crate) fn args(op: TelemetryOp, input: &Value, actor: Option<&str>) -> std::result::Result<(&'static str, Map<String, Value>), String> {
420 let mut out = Map::new();
421 let copy = |out: &mut Map<String, Value>, keys: &[(&str, &str)]| {
422 for (key, field) in keys {
423 if let Some(value) = text(input, key) {
424 out.insert((*field).to_owned(), value.into());
425 }
426 }
427 };
428 let page = |out: &mut Map<String, Value>| -> std::result::Result<(), String> {
429 copy(out, &[("cursor", "cursor")]);
430 if let Some(limit) = number(input, "limit")? {
431 out.insert("limit".into(), limit.into());
432 }
433 Ok(())
434 };
435 let required = |key: &str, what: &str| text(input, key).ok_or_else(|| format!("Give {what}."));
436 let method = match op {
437 TelemetryOp::GetOverview => {
438 filter(input, &mut out)?;
439 copy(&mut out, &[("deployment_id", "deployment_id")]);
440 if let Some(step) = number(input, "step")? {
441 out.insert("step_s".into(), step.into());
442 }
443 "overview"
444 }
445 TelemetryOp::ListRequests => {
446 filter(input, &mut out)?;
447 copy(&mut out, &[("q", "text"), ("route", "route"), ("deployment_id", "deployment_id")]);
448 let status = list(input, "status");
449 if !status.is_empty() {
450 out.insert("status".into(), json!(status));
451 }
452 if let Some(errors) = flag(input, "errors_only")? {
453 out.insert("errors_only".into(), errors.into());
454 }
455 if let Some(ms) = number(input, "min_duration_ms")? {
456 out.insert("min_duration_ms".into(), ms.into());
457 }
458 page(&mut out)?;
459 "requests"
460 }
461 TelemetryOp::GetRequest => {
462 out.insert("id".into(), required("id", "the request's id")?.into());
463 "request"
464 }
465 TelemetryOp::ListLogs => {
466 filter(input, &mut out)?;
467 copy(&mut out, &[("q", "text"), ("trace_id", "trace_id"), ("request_id", "request_id"), ("deployment_id", "deployment_id")]);
468 let levels: Vec<String> = list(input, "level").into_iter().map(|level| level.to_lowercase()).collect();
469 if let Some(unknown) = levels.iter().find(|level| !SEVERITIES.contains(&level.as_str())) {
470 return Err(format!("{unknown} is not a level: use trace, debug, info, warn, error or fatal."));
471 }
472 if !levels.is_empty() {
473 out.insert("severities".into(), json!(levels));
474 }
475 page(&mut out)?;
476 "logs"
477 }
478 TelemetryOp::ListTraces => {
479 filter(input, &mut out)?;
480 copy(&mut out, &[("q", "text")]);
481 if let Some(errors) = flag(input, "errors_only")? {
482 out.insert("errors_only".into(), errors.into());
483 }
484 if let Some(ms) = number(input, "min_duration_ms")? {
485 out.insert("min_duration_ms".into(), ms.into());
486 }
487 page(&mut out)?;
488 "traces"
489 }
490 TelemetryOp::GetTrace => {
491 out.insert("trace_id".into(), required("trace_id", "the trace's id")?.to_lowercase().into());
492 "trace"
493 }
494 TelemetryOp::ListErrors => {
495 filter(input, &mut out)?;
496 if let Some(state) = text(input, "state") {
497 if !ERROR_STATES.contains(&state.as_str()) {
498 return Err(format!("{state} is not a state: use open, resolved or ignored."));
499 }
500 out.insert("state".into(), state.into());
501 }
502 "errors"
503 }
504 TelemetryOp::UpdateError => {
505 out.insert("fingerprint".into(), required("fingerprint", "the error group's fingerprint")?.into());
506 if let Some(state) = text(input, "state") {
507 if !ERROR_STATES.contains(&state.as_str()) {
508 return Err(format!("{state} is not a state: use open, resolved or ignored."));
509 }
510 out.insert("state".into(), state.into());
511 }
512 match input.get("issue_number") {
513 None => {}
514 Some(Value::Null) => {
515 out.insert("issue_number".into(), Value::Null);
516 }
517 Some(_) => {
518 let issue = number(input, "issue_number")?.ok_or("issue_number must be a whole number, or null.")?;
519 out.insert("issue_number".into(), issue.into());
520 }
521 }
522 if !out.contains_key("state") && !out.contains_key("issue_number") {
523 return Err("Give a state, or an issue_number.".to_owned());
524 }
525 "set_error"
526 }
527 TelemetryOp::ListMetrics => "metrics",
528 TelemetryOp::GetMetric => {
529 out.insert("name".into(), required("metric", "the metric's name")?.into());
530 filter(input, &mut out)?;
531 if let Some(aggregate) = text(input, "aggregate") {
532 if !AGGREGATES.contains(&aggregate.as_str()) {
533 return Err(format!("{aggregate} is not an aggregate: use avg, sum, min, max, p50, p95 or p99."));
534 }
535 out.insert("aggregate".into(), aggregate.into());
536 }
537 copy(&mut out, &[("group_by", "group_by")]);
538 if let Some(step) = number(input, "step")? {
539 out.insert("step_s".into(), step.into());
540 }
541 "metric"
542 }
543 TelemetryOp::GetSettings => "settings",
544 TelemetryOp::UpdateSettings => {
545 if let Some(days) = number(input, "retention_days")? {
546 if !RETENTION_DAYS.contains(&days) {
547 return Err("retention_days is 1, 3, 7, 14 or 30.".to_owned());
548 }
549 out.insert("retention_days".into(), days.into());
550 }
551 if let Some(on) = flag(input, "deployments")? {
552 out.insert("deployments".into(), on.into());
553 }
554 "set_settings"
555 }
556 TelemetryOp::ListTokens => "tokens",
557 TelemetryOp::CreateToken => {
558 out.insert("name".into(), required("name", "the token a name")?.into());
559 out.insert("created_by".into(), actor.map_or(Value::Null, Value::from));
560 "create_token"
561 }
562 TelemetryOp::DeleteToken => {
563 out.insert("id".into(), required("id", "the token's id")?.into());
564 "revoke_token"
565 }
566 TelemetryOp::ListExports => "exports",
567 TelemetryOp::CreateExport | TelemetryOp::UpdateExport => {
568 if op == TelemetryOp::UpdateExport {
569 out.insert("id".into(), required("id", "the export's id")?.into());
570 } else {
571 required("name", "the export a name")?;
572 required("endpoint", "the endpoint to send to")?;
573 }
574 copy(&mut out, &[("name", "name"), ("endpoint", "endpoint")]);
575 if !input["signals"].is_null() {
576 let signals = list(input, "signals");
577 if let Some(unknown) = signals.iter().find(|signal| !SIGNALS.contains(&signal.as_str())) {
578 return Err(format!("{unknown} is not a signal: use traces, logs or metrics."));
579 }
580 out.insert("signals".into(), json!(signals));
581 }
582 match &input["headers"] {
583 Value::Null => {}
584 Value::Object(headers) => {
585 let mut sent = Map::new();
586 for (name, value) in headers {
587 let value = value.as_str().ok_or_else(|| format!("The header {name} must be text."))?;
588 sent.insert(name.trim().to_owned(), value.into());
589 }
590 out.insert("headers".into(), Value::Object(sent));
591 }
592 _ => return Err("headers must be an object of names and values.".to_owned()),
593 }
594 if let Some(enabled) = flag(input, "enabled")? {
595 out.insert("enabled".into(), enabled.into());
596 }
597 "put_export"
598 }
599 TelemetryOp::DeleteExport => {
600 out.insert("id".into(), required("id", "the export's id")?.into());
601 "delete_export"
602 }
603 TelemetryOp::GetUsage => "usage",
604 };
605 Ok((method, out))
606}
607
608/// Calls the telemetry service: its answer, or its refusal of a bad
609/// argument (`400`, in words) as invalid.
610async fn ask(services: &Services, method: &str, args: &Value) -> Result<Outcome<Value>> {
611 let headers = Headers::new();
612 headers.set("content-type", "application/json")?;
613 let mut init = RequestInit::new();
614 init.with_method(Method::Post)
615 .with_headers(headers)
616 .with_body(Some(serde_json::to_string(args)?.into()));
617 let request = Request::new_with_init(&format!("https://service/rpc/{method}"), &init)?;
618 let mut response = services.telemetry.fetch_request(request).await?;
619 let text = response.text().await.unwrap_or_default();
620 match response.status_code() {
621 200 => Ok(Outcome::Ok(g1t_kit::read_answer(method, &text)?)),
622 400 => Ok(Outcome::fail(FailureCode::Invalid, text.trim())),
623 status => Err(worker::Error::RustError(format!("telemetry {method} failed with status {status}: {text}"))),
624 }
625}
626
627/// The repository a project's permission comes from, as the deployments
628/// service reads it: a mirrored project has none, and its workspace's base
629/// permission decides.
630fn repo_of(project: &Value) -> (String, String, bool) {
631 let source = &project["source"];
632 let private = project["private"].as_bool().unwrap_or(true);
633 if source["kind"] == "hosted" {
634 (
635 source["repoId"].as_str().unwrap_or_default().to_owned(),
636 source["repo"]["namespace"].as_str().unwrap_or_default().to_owned(),
637 private,
638 )
639 } else {
640 (String::new(), project["workspace"].as_str().unwrap_or_default().to_owned(), private)
641 }
642}
643
644/// Whether the viewer may do what `need` says on the project; otherwise
645/// the refusal.
646fn allowed(viewer: &Viewer, project: &Value, need: Need) -> std::result::Result<(), Outcome<Value>> {
647 let (id, namespace, private) = repo_of(project);
648 let repo = RepoRef { id: &id, namespace: &namespace, private };
649 let user = viewer.as_ref();
650 if !user.is_some_and(|user| access::granted(user, repo).is_some()) {
651 return Err(Outcome::fail(FailureCode::NotFound, "There is no such project."));
652 }
653 if let Need::Can(capability) = need
654 && !access::can(user, repo, capability)
655 {
656 let name = format!("{}/{}", project["workspace"].as_str().unwrap_or_default(), project["slug"].as_str().unwrap_or_default());
657 return Err(Outcome::fail(FailureCode::Forbidden, access::needs(capability, &name)));
658 }
659 Ok(())
660}
661
662pub async fn run(op: TelemetryOp, services: &Services, viewer: &Viewer, input: &Value) -> Result<Outcome<Value>> {
663 let slug = |key: &str| text(input, key).map(|value| value.to_lowercase());
664 let (Some(workspace), Some(slug)) = (slug("workspace"), slug("project")) else {
665 return Ok(Outcome::fail(FailureCode::Invalid, "Give the workspace's slug and the project's."));
666 };
667 let Some(user) = viewer else {
668 let what = if op.writes() { "Changing" } else { "Reading" };
669 return Ok(Outcome::fail(FailureCode::Unauthenticated, format!("{what} a project's observability needs a g1t access token.")));
670 };
671 let (method, mut args) = match args(op, input, Some(&user.username)) {
672 Ok(found) => found,
673 Err(message) => return Ok(Outcome::fail(FailureCode::Invalid, message)),
674 };
675 let found: Outcome<Value> =
676 g1t_kit::call(&services.projects, "get", &json!({ "workspace": workspace, "slug": slug, "viewer": viewer })).await?;
677 let project = match found {
678 Outcome::Ok(project) => project,
679 Outcome::Fail(failure) => return Ok(Outcome::Fail(failure)),
680 };
681 if let Err(refused) = allowed(viewer, &project, op.need()) {
682 return Ok(refused);
683 }
684 let Some(project_id) = project["id"].as_str() else {
685 return Ok(Outcome::fail(FailureCode::NotFound, "There is no such project."));
686 };
687 args.insert("project_id".into(), project_id.into());
688
689 // A change to an export names only what changes: the rest is as it is.
690 if op == TelemetryOp::UpdateExport {
691 let listed = ask(services, "exports", &json!({ "project_id": project_id })).await?;
692 let Outcome::Ok(Value::Array(exports)) = listed else {
693 return Ok(listed);
694 };
695 let Some(current) = exports.iter().find(|export| export["id"] == args["id"]) else {
696 return Ok(Outcome::fail(FailureCode::NotFound, "The project has no export with that id."));
697 };
698 for key in ["name", "endpoint", "signals", "enabled"] {
699 if !args.contains_key(key) {
700 args.insert(key.into(), current[key].clone());
701 }
702 }
703 }
704 if op == TelemetryOp::CreateExport {
705 args.entry("signals").or_insert_with(|| json!(SIGNALS));
706 args.entry("enabled").or_insert(Value::Bool(true));
707 }
708
709 let answered = ask(services, method, &Value::Object(args)).await?;
710 // Nothing found answers null, and deleting nothing false.
711 let missing = match op {
712 TelemetryOp::GetRequest => Some("The project has no request with that id."),
713 TelemetryOp::GetTrace => Some("The project has no trace with that id."),
714 TelemetryOp::UpdateError => Some("The project has no error with that fingerprint."),
715 TelemetryOp::DeleteToken => Some("The project has no ingest token with that id that people can delete."),
716 TelemetryOp::DeleteExport => Some("The project has no export with that id."),
717 _ => None,
718 };
719 Ok(match (answered, missing) {
720 (Outcome::Ok(Value::Null | Value::Bool(false)), Some(message)) => Outcome::fail(FailureCode::NotFound, message),
721 (answered, _) => answered,
722 })
723}
724
725#[cfg(test)]
726mod tests {
727 use super::*;
728
729 #[test]
730 fn a_query_becomes_the_services_arguments() {
731 let (method, args) = super::args(
732 TelemetryOp::ListLogs,
733 &json!({ "workspace": "acme", "project": "web", "q": "timeout", "level": "Error,warn", "environment": "production", "from": "1790918179000", "limit": "50", "ignored": 1 }),
734 None,
735 )
736 .unwrap();
737 assert_eq!(method, "logs");
738 assert_eq!(args["text"], json!("timeout"));
739 assert_eq!(args["severities"], json!(["error", "warn"]));
740 assert_eq!(args["environments"], json!(["production"]));
741 assert_eq!(args["from_us"], json!(1_790_918_179_000_000u64));
742 assert_eq!(args["limit"], json!(50));
743 assert!(!args.contains_key("ignored") && !args.contains_key("workspace"));
744 assert!(super::args(TelemetryOp::ListLogs, &json!({ "level": "loud" }), None).is_err());
745 assert!(super::args(TelemetryOp::ListLogs, &json!({ "from": "yesterday" }), None).is_err());
746
747 let (method, args) = super::args(
748 TelemetryOp::ListRequests,
749 &json!({ "status": "5xx,404", "errors_only": "true", "min_duration_ms": 250, "service": "api" }),
750 None,
751 )
752 .unwrap();
753 assert_eq!(method, "requests");
754 assert_eq!((args["status"].clone(), args["errors_only"].clone(), args["min_duration_ms"].clone()), (json!(["5xx", "404"]), json!(true), json!(250)));
755 assert_eq!(args["services"], json!(["api"]));
756
757 let (method, args) = super::args(TelemetryOp::GetMetric, &json!({ "metric": "http.server.request.duration", "aggregate": "p95", "step": "60" }), None).unwrap();
758 assert_eq!((method, args["name"].clone(), args["step_s"].clone()), ("metric", json!("http.server.request.duration"), json!(60)));
759 assert!(super::args(TelemetryOp::GetMetric, &json!({ "metric": "x", "aggregate": "median" }), None).is_err());
760 }
761
762 #[test]
763 fn changes_say_only_what_they_change() {
764 let (method, args) = super::args(TelemetryOp::UpdateError, &json!({ "fingerprint": "f1", "issue_number": null }), None).unwrap();
765 assert_eq!(method, "set_error");
766 assert_eq!(args.get("issue_number"), Some(&Value::Null), "null unlinks the issue");
767 assert!(!args.contains_key("state"));
768 assert!(super::args(TelemetryOp::UpdateError, &json!({ "fingerprint": "f1" }), None).is_err());
769 assert!(super::args(TelemetryOp::UpdateError, &json!({ "fingerprint": "f1", "state": "fixed" }), None).is_err());
770 assert!(super::args(TelemetryOp::UpdateSettings, &json!({ "retention_days": 5 }), None).is_err());
771 let (method, args) = super::args(TelemetryOp::CreateToken, &json!({ "name": "laptop" }), Some("ada")).unwrap();
772 assert_eq!((method, args["created_by"].clone()), ("create_token", json!("ada")));
773 let (method, args) = super::args(TelemetryOp::UpdateExport, &json!({ "id": "e1", "headers": { "x-api-key": "" } }), None).unwrap();
774 assert_eq!(method, "put_export");
775 assert_eq!(args["headers"], json!({ "x-api-key": "" }));
776 assert!(!args.contains_key("signals"));
777 assert!(super::args(TelemetryOp::CreateExport, &json!({ "name": "vendor" }), None).is_err());
778 assert!(super::args(TelemetryOp::CreateExport, &json!({ "name": "v", "endpoint": "https://x", "signals": ["spans"] }), None).is_err());
779 }
780
781 #[test]
782 fn times_are_milliseconds_or_rfc_3339() {
783 assert_eq!(micros(&json!(1_790_918_179_000u64)), Some(1_790_918_179_000_000));
784 assert_eq!(micros(&json!("2026-10-02T05:16:19Z")), Some(1_790_918_179_000_000));
785 assert_eq!(micros(&json!("2026-10-02T07:16:19+02:00")), Some(1_790_918_179_000_000));
786 assert_eq!(micros(&json!("2026-10-02T00:16:19.500-05:00")), Some(1_790_918_179_500_000));
787 assert_eq!(micros(&json!("soon")), None);
788 }
789
790 #[test]
791 fn reading_takes_a_role_of_ones_own() {
792 use g1t_contracts::User;
793 let project = json!({
794 "id": "prj_1", "workspace": "acme", "slug": "web", "private": false,
795 "source": { "kind": "hosted", "repoId": "repo_1", "repo": { "namespace": "acme", "name": "web" } },
796 });
797 let person = |username: &str, membership: Value| -> Viewer {
798 let mut user = User { username: username.into(), ..User::default() };
799 if !membership.is_null() {
800 user.workspaces = vec![serde_json::from_value(membership).unwrap()];
801 }
802 Some(user)
803 };
804 let refused = |viewer: &Viewer, need: Need| match allowed(viewer, &project, need) {
805 Err(Outcome::Fail(failure)) => Some(failure.code),
806 _ => None,
807 };
808 // A public project's telemetry is not public.
809 assert_eq!(refused(&person("eve", Value::Null), Need::Insider), Some(FailureCode::NotFound));
810 assert_eq!(refused(&None, Need::Insider), Some(FailureCode::NotFound));
811 let reader = person("ada", json!({ "slug": "acme", "role": "member", "base_permission": "read" }));
812 assert_eq!(refused(&reader, Need::Insider), None);
813 assert_eq!(refused(&reader, TelemetryOp::UpdateError.need()), Some(FailureCode::Forbidden));
814 let writer = person("bo", json!({ "slug": "acme", "role": "member" }));
815 assert_eq!(refused(&writer, TelemetryOp::UpdateError.need()), None);
816 assert_eq!(refused(&writer, TelemetryOp::ListTokens.need()), Some(FailureCode::Forbidden));
817 let owner = person("ana", json!({ "slug": "acme", "role": "owner" }));
818 for op in TelemetryOp::ALL {
819 assert_eq!(refused(&owner, op.need()), None, "{}", op.name());
820 }
821 assert_eq!(TelemetryOp::GetSettings.need(), Need::Can(Capability::ManageIntegrations));
822 }
823
824 #[test]
825 fn each_operation_is_described_with_a_schema() {
826 for op in TelemetryOp::ALL {
827 assert!(crate::operations::Op::ALL.contains(&crate::operations::Op::Telemetry(op)), "{}", op.name());
828 assert!(!op.title().is_empty() && op.description().len() > 40, "{}", op.name());
829 let required = op.input()["required"].as_array().unwrap().clone();
830 assert!(required.contains(&json!("workspace")) && required.contains(&json!("project")), "{}", op.name());
831 for field in required {
832 assert!(op.input()["properties"].get(field.as_str().unwrap()).is_some(), "{}: {field}", op.name());
833 }
834 }
835 }
836}