1use super::codec::{DEFAULT_LLM_FALLBACK, DEFAULT_REQUIRE_TENANT, MAX_BODY_BYTES};
2use crate::llm::actions::{
3 protocol_trait::{ActionResult, Protocol, Server},
4 ActionDefinition, Parameter, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::{
7 log_template::LogTemplate,
8 metadata::{DevelopmentState, ProtocolMetadataV2},
9 EventType, SpawnContext,
10};
11use crate::state::app_state::AppState;
12use anyhow::{bail, ensure, Context, Result};
13use serde::{Deserialize, Serialize};
14use serde_json::{json, Value};
15use std::sync::LazyLock;
16#[derive(Clone, Debug, Serialize, Deserialize)]
17#[serde(tag = "mode", rename_all = "snake_case")]
18pub enum Decision {
19 Accept,
20 Reject {
21 status: u16,
22 message: String,
23 retry_after_seconds: Option<u16>,
24 },
25}
26pub fn parameter(name: &str, type_hint: &str, description: &str, required: bool) -> Parameter {
27 Parameter {
28 name: name.into(),
29 type_hint: type_hint.into(),
30 description: description.into(),
31 required,
32 }
33}
34fn accept() -> ActionDefinition {
35 ActionDefinition{name:"accept_loki_entries".into(),description:"Accept all validated entries with204. Acceptance means handler observation, not durable storage; no private stream database.".into(),parameters:vec![],example:json!({"type":"accept_loki_entries"}),log_template:Some(LogTemplate::new().with_info("Accept Loki log entries"))}
36}
37fn reject() -> ActionDefinition {
38 ActionDefinition{name:"reject_loki_entries".into(),description:"Reject this batch with a bounded Loki plain-text HTTP error.260 explicitly means blocked ingestion. Retry advice never schedules retries.".into(),parameters:vec![parameter("status","number","HTTP refusal status: 260 for blocked ingestion, or 400, 401, 403, 404, 413, 415, 422, 429, 500 or 503",true),parameter("message","string","UTF-8 explanation<=4096bytes; tab/newline allowed",true),parameter("retry_after_seconds","number","Optional Retry-After delay, 1..3600 seconds; valid only with status 429 or 503",false)],example:json!({"type":"reject_loki_entries","status":429,"message":"Tenant ingestion quota","retry_after_seconds":3}),log_template:Some(LogTemplate::new().with_info("Reject Loki log entries with status {status}"))}
39}
40pub static LOKI_PUSH_EVENT: LazyLock<EventType> = LazyLock::new(|| {
41 EventType::new("loki_push","Validated typed stream labels, entries with signed nanosecond timestamps/log lines/string metadata, tenant identity and encoding. Raw compressed bodies and bearer credentials are excluded.",accept().example).with_parameters(vec![parameter("tenant_id","string","Single tenant or fake when header absent",true),parameter("tenant_provided","boolean","Explicit X-Scope-OrgID received",true),parameter("encoding","string","json/gzip_json/snappy_protobuf",true),parameter("streams","array","labels and typed entries(timestamp_ns,line,structured_metadata)",true),parameter("authenticated","boolean","Token verified or no token configured",true),parameter("auth_required","boolean","Bearer token configured",true),parameter("source_addr","string","Remote socket address of the HTTP peer that submitted this push",true)]).with_actions(vec![accept(),reject()])
42});
43#[derive(Default)]
44pub struct LokiProtocol;
45impl LokiProtocol {
46 pub fn new() -> Self {
47 Self
48 }
49}
50impl Protocol for LokiProtocol {
51 fn protocol_name(&self) -> &'static str {
52 "Loki"
53 }
54 fn stack_name(&self) -> &'static str {
55 "ETH>IP>TCP>HTTP>Loki"
56 }
57 fn keywords(&self) -> Vec<&'static str> {
58 vec!["loki", "loki-push"]
59 }
60 fn group_name(&self) -> &'static str {
61 "Network Services"
62 }
63 fn description(&self) -> &'static str {
64 "Bounded typed Loki JSON/gzip and protobuf/Snappy collector; unmatched logs collect without model calls"
65 }
66 fn example_prompt(&self) -> &'static str {
67 "Collect Loki log pushes on localhost port3100"
68 }
69 fn get_sync_actions(&self) -> Vec<ActionDefinition> {
70 vec![accept(), reject()]
71 }
72 fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
73 vec![]
74 }
75 fn get_event_types(&self) -> Vec<EventType> {
76 vec![LOKI_PUSH_EVENT.clone()]
77 }
78 fn metadata(&self) -> ProtocolMetadataV2 {
79 ProtocolMetadataV2::builder().answers_on_failure().state(DevelopmentState::Experimental).well_known_port(3100).max_inbound_bytes(MAX_BODY_BYTES).request_only("HTTP push request/response; no unsolicited messages")
80.implementation("Native public push schema and bounded Snappy block codec; JSON and existing flate2 gzip; owned Hyper HTTP/1.1 collector")
81.llm_control("Typed logs/labels/metadata and tenant facts through common handlers/shared memory; unmatched collection has no model calls unless llm_fallback=true")
82.e2e_testing("Pinned official Loki3.7.8 service readback and maintained Alloy1.20.1 protobuf/Snappy writer; native pair, malformed/bounds/auth/error/cancellation checks")
83.notes("Experimental POST /loki/api/v1/push only: JSON, gzip JSON, Snappy protobuf; optional single tenant header (fake if absent), require_tenant flag, optional proxy-style Bearer check (not Loki-built-in authentication).256KiB wire/decoded body;64streams/1024entries;16KiB UTF-8 lines;32labels/64metadata;ASCII identifier names<=128bytes,values<=2048bytes; tenant150/token1024bytes;16384protobuf fields/JSON depth8;256connections;32KiB/64headers;30s header/body,10s decision-write. Reserved __ stream labels and duplicate stream sets/keys rejected; strict push fields and signed i64 nanoseconds. Empty successful handler accepts; failed/multiple decisions503. No TLS/proxy/cloud auth, durable store, per-tenant ACL, querying, retention/order enforcement, partial acceptance, OTLP, retries, fuzz or pcap.").build()
84 }
85 fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
86 vec![ParameterDefinition{name:"auth_token".into(),type_hint:"string".into(),description:"Optional proxy-style Bearer token<=1024printable ASCII; never included in event data; absent means anonymous".into(),required:false,example:json!("collector-secret"),default:None},ParameterDefinition{name:"require_tenant".into(),type_hint:"boolean".into(),description:"Require one valid X-Scope-OrgID. Identity is not authorization or a tenant database.".into(),required:false,example:json!(true),default:Some(json!(DEFAULT_REQUIRE_TENANT))},ParameterDefinition{name:"llm_fallback".into(),type_hint:"boolean".into(),description:"Opt unmatched pushes into model calls; explicit handlers always run".into(),required:false,example:json!(true),default:Some(json!(DEFAULT_LLM_FALLBACK))}]
87 }
88 fn get_startup_examples(&self) -> StartupExamples {
89 StartupExamples::new(
90 json!({"type":"open_server","base_stack":"loki","port":3100,"startup_params":{"llm_fallback":true},"instruction":"Decide which log pushes to accept"}),
91 json!({"type":"open_server","base_stack":"loki","port":3100,"event_handlers":[{"event_pattern":"loki_push","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'accept_loki_entries'}]}))"}}]}),
92 json!({"type":"open_server","base_stack":"loki","port":3100,"event_handlers":[{"event_pattern":"loki_push","handler":{"type":"static","actions":[accept().example]}}]}),
93 )
94 }
95}
96impl Server for LokiProtocol {
97 fn spawn(
98 &self,
99 ctx: SpawnContext,
100 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
101 {
102 Box::pin(super::LokiServer::spawn(ctx))
103 }
104 fn execute_action(&self, action: Value) -> Result<ActionResult> {
105 let decision = match action["type"].as_str() {
106 Some("accept_loki_entries") => Decision::Accept,
107 Some("reject_loki_entries") => {
108 let status = u16::try_from(action["status"].as_u64().context("status required")?)?;
109 ensure!(
110 [260, 400, 401, 403, 404, 413, 415, 422, 429, 500, 503].contains(&status),
111 "unsupported rejection status"
112 );
113 let message = action["message"].as_str().context("message required")?;
114 ensure!(
115 !message.is_empty()
116 && message.len() <= 4096
117 && !message
118 .chars()
119 .any(|c| c.is_control() && !matches!(c, '\t' | '\n' | '\r')),
120 "error message byte/control limit"
121 );
122 let retry = action
123 .get("retry_after_seconds")
124 .map(|v| {
125 u16::try_from(v.as_u64().context("integer retry advice required")?)
126 .map_err(anyhow::Error::from)
127 })
128 .transpose()?;
129 if let Some(n) = retry {
130 ensure!(
131 (1..=3600).contains(&n) && [429, 503].contains(&status),
132 "Retry-After1..3600only429/503"
133 );
134 }
135 Decision::Reject {
136 status,
137 message: message.into(),
138 retry_after_seconds: retry,
139 }
140 }
141 _ => bail!("unknown Loki collector action"),
142 };
143 Ok(ActionResult::Custom {
144 name: "loki_push_decision".into(),
145 data: serde_json::to_value(decision)?,
146 })
147 }
148}