netget/client/loki/
actions.rs

1use crate::llm::actions::{
2    client_trait::{Client, ClientActionResult},
3    protocol_trait::Protocol,
4    ActionDefinition, ParameterDefinition, StartupExamples,
5};
6use crate::protocol::{
7    log_template::LogTemplate,
8    metadata::{DevelopmentState, ProtocolMetadataV2},
9    ConnectContext, EventType,
10};
11use crate::server::loki::{
12    actions::parameter,
13    codec::{self, PushBatch},
14};
15use crate::state::app_state::AppState;
16use anyhow::{bail, Context, Result};
17use serde_json::{json, Value};
18use std::sync::LazyLock;
19#[derive(Default)]
20pub struct LokiClientProtocol;
21impl LokiClientProtocol {
22    pub fn new() -> Self {
23        Self
24    }
25}
26pub fn write_action() -> ActionDefinition {
27    ActionDefinition{name:"push_loki_entries".into(),description:"Submit one validated typed Loki push, with explicit status/error/tenant/encoding completion and no automatic retries.204 means receiver acceptance.".into(),parameters:vec![parameter("batch","object","optional tenant_id,encoding json(default)/gzip_json/snappy_protobuf,streams[{labels:{name:value},entries:[{timestamp_ns:i64,line:string,structured_metadata:{name:value}(optional)}]}]",true)],example:json!({"type":"push_loki_entries","batch":{"tenant_id":"tenant-one","encoding":"snappy_protobuf","streams":[{"labels":{"app":"example"},"entries":[{"timestamp_ns":1700000000000000000i64,"line":"Hello 名","structured_metadata":{"trace_id":"123"}}]}]}}),log_template:Some(LogTemplate::new().with_info("Push Loki log entries"))}
28}
29fn disconnect() -> ActionDefinition {
30    ActionDefinition {
31        name: "disconnect".into(),
32        description: "Cancel logical session, parked handlers and pending HTTP exchange".into(),
33        parameters: vec![],
34        example: json!({"type":"disconnect"}),
35        log_template: Some(LogTemplate::new().with_info("Disconnect Loki client")),
36    }
37}
38pub static LOKI_CONNECTED_EVENT: LazyLock<EventType> = LazyLock::new(|| {
39    EventType::new(
40        "loki_connected",
41        "Logical HTTP origin ready; no TCP until a fully validated push",
42        write_action().example,
43    )
44    .with_parameters(vec![parameter(
45        "remote_addr",
46        "string",
47        "Cleartext HTTP origin of the Loki push receiver",
48        true,
49    )])
50    .with_actions(vec![write_action(), disconnect()])
51});
52pub static LOKI_RESPONSE_EVENT: LazyLock<EventType> = LazyLock::new(|| {
53    EventType::new("loki_push_response","Typed push outcome;204accepted,260blocked,other supported error statuses explicit. No retry/persistence/atomicity promise.",write_action().example).with_parameters(vec![parameter("tenant_id","string","Submitted tenant or fake",true),parameter("encoding","string","Carrier used for the submitted batch: json, gzip_json or snappy_protobuf",true),parameter("stream_count","number","Number of label streams in the submitted batch",true),parameter("entry_count","number","Number of log entries in the submitted batch",true),parameter("status","number","Receiver HTTP response status; 204 accepts the push, 260 blocks ingestion",true),parameter("error","object|null","message for rejection, including260",true),parameter("retry_after_seconds","number|null","Optional receiver retry delay in seconds; this client never retries automatically",true)]).with_actions(vec![write_action(),disconnect()])
54});
55impl Protocol for LokiClientProtocol {
56    fn protocol_name(&self) -> &'static str {
57        "Loki"
58    }
59    fn stack_name(&self) -> &'static str {
60        "ETH>IP>TCP>HTTP>Loki"
61    }
62    fn keywords(&self) -> Vec<&'static str> {
63        vec!["loki", "loki-push"]
64    }
65    fn group_name(&self) -> &'static str {
66        "Network Services"
67    }
68    fn description(&self) -> &'static str {
69        "Typed Loki JSON/gzip/protobuf-Snappy push emitter with tenant and rejection events"
70    }
71    fn example_prompt(&self) -> &'static str {
72        "Push typed logs to Loki at http://localhost:3100"
73    }
74    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
75        vec![]
76    }
77    fn get_async_actions(&self, _: &AppState) -> Vec<ActionDefinition> {
78        vec![write_action(), disconnect()]
79    }
80    fn get_event_types(&self) -> Vec<EventType> {
81        vec![LOKI_CONNECTED_EVENT.clone(), LOKI_RESPONSE_EVENT.clone()]
82    }
83    fn get_startup_parameters(&self) -> Vec<ParameterDefinition> {
84        vec![ParameterDefinition{name:"auth_token".into(),type_hint:"string".into(),description:"Optional proxy-style Bearer token<=1024printable ASCII; not built-in Loki authentication; excluded from handlers".into(),required:false,example:json!("collector-secret"),default:None}]
85    }
86    fn metadata(&self) -> ProtocolMetadataV2 {
87        ProtocolMetadataV2::builder().state(DevelopmentState::Experimental).well_known_port(3100).implementation("Native bounded public protobuf/Snappy block encoder and JSON/gzip; owned Hyper HTTP/1.1 exchanges").llm_control("Connected/typed response events, common shared memory and command injection independent of parked handlers/IO").e2e_testing("Official Loki3.7.8 readback and Alloy1.20.1 writer; native pair/errors/lifecycle/bounds").notes("Cleartext HTTP origin only; POST /loki/api/v1/push; optional tenant/Bearer;JSON/gzip JSON/Snappy protobuf.256KiB wire/decoded body,64streams/1024entries,16KiB line,32labels/64metadata,128byte ASCII names/2048values,150tenant/1024token;64KiB response/4096byte error,32KiB/64headers,10s whole exchange,one in-flight request,32queued events/actions,followup depth8.204only success;260blocked and supported4xx/5xx typed; numeric Retry-After1..3600only429/503. No TLS/proxy/basic/cloud auth, redirects, retries, queries, durable store, order/retention/ACL enforcement, OTLP,partial acceptance,fuzz orpcap. Logical local address0.0.0.0:0; fresh TCP per push.").build()
88    }
89    fn get_startup_examples(&self) -> StartupExamples {
90        StartupExamples::new(
91            json!({"type":"open_client","base_stack":"loki","remote_addr":"http://localhost:3100","instruction":"Push one log entry"}),
92            json!({"type":"open_client","base_stack":"loki","remote_addr":"http://localhost:3100","event_handlers":[{"event_pattern":"loki_connected","handler":{"type":"script","language":"python","code":"import json,sys\njson.load(sys.stdin)\nprint(json.dumps({'actions':[{'type':'disconnect'}]}))"}}]}),
93            json!({"type":"open_client","base_stack":"loki","remote_addr":"http://localhost:3100","event_handlers":[{"event_pattern":"loki_connected","handler":{"type":"static","actions":[write_action().example]}}]}),
94        )
95    }
96}
97impl Client for LokiClientProtocol {
98    fn connect(
99        &self,
100        ctx: ConnectContext,
101    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::net::SocketAddr>> + Send>>
102    {
103        Box::pin(super::LokiClient::connect(ctx))
104    }
105    fn execute_action(&self, action: Value) -> Result<ClientActionResult> {
106        match action["type"].as_str() {
107            Some("push_loki_entries") => {
108                let batch: PushBatch =
109                    serde_json::from_value(action.get("batch").context("batch required")?.clone())?;
110                codec::encode_batch(&batch)?;
111                Ok(ClientActionResult::Custom {
112                    name: "push_loki_entries".into(),
113                    data: serde_json::to_value(batch)?,
114                })
115            }
116            Some("disconnect") => Ok(ClientActionResult::Disconnect),
117            _ => bail!("unknown Loki emitter action"),
118        }
119    }
120}
netget/client/loki/actions.rs

netget/client/loki/
actions.rs
