netget/server/spark/
actions.rs

1//! Apache Spark monitoring REST API (`/api/v1/...`) protocol actions and event types.
2//!
3//! The LLM roleplays a Spark application's control plane and invents the applications, jobs,
4//! stages and executors it reports. Every success response is a top-level **JSON array**, the
5//! exact shape Spark's monitoring REST API (and its History Server) emits, so `curl` / a real
6//! client parses it unchanged.
7//!
8//! No storage: the model supplies all state per request. One event type (`spark_request`),
9//! emitted for every LLM-handled request; the mechanical `GET /api/v1/version` banner is
10//! answered statically in `mod.rs`.
11
12use crate::llm::actions::protocol_trait::{ActionResult, Protocol, Server};
13use crate::llm::actions::{ActionDefinition, Parameter};
14use crate::protocol::log_template::LogTemplate;
15use crate::protocol::EventType;
16use crate::state::app_state::AppState;
17use anyhow::Result;
18use serde_json::{json, Value};
19use std::sync::LazyLock;
20
21/// Apache Spark monitoring-API protocol handler.
22pub struct SparkProtocol {}
23
24impl SparkProtocol {
25    pub fn new() -> Self {
26        Self {}
27    }
28}
29
30impl Default for SparkProtocol {
31    fn default() -> Self {
32        Self::new()
33    }
34}
35
36/// Fired for every Spark monitoring-API request that needs an invented answer (everything
37/// except the static `/api/v1/version` banner).
38pub static SPARK_REQUEST_EVENT: LazyLock<EventType> = LazyLock::new(|| {
39    EventType::new(
40        "spark_request",
41        "Apache Spark monitoring REST API request received",
42        json!({
43            "type": "send_spark_applications",
44            "applications": [{
45                "id": "app-20161116163331-0000",
46                "name": "Spark shell",
47                "attempts": [{
48                    "startTime": "2016-11-16T22:33:29.916GMT",
49                    "completed": false,
50                    "sparkUser": "jose",
51                    "appSparkVersion": "3.5.1"
52                }]
53            }]
54        }),
55    )
56    .with_parameters(vec![
57        Parameter {
58            name: "method".to_string(),
59            type_hint: "string".to_string(),
60            description: "HTTP method (GET)".to_string(),
61            required: true,
62        },
63        Parameter {
64            name: "path".to_string(),
65            type_hint: "string".to_string(),
66            description: "Request path, e.g. /api/v1/applications/app-1/jobs".to_string(),
67            required: true,
68        },
69        Parameter {
70            name: "operation".to_string(),
71            type_hint: "string".to_string(),
72            description: "Detected operation: applications (the list), application (one, \
73                          GET /api/v1/applications/{id}), jobs, stages, executors, rdd \
74                          (/storage/rdd), or application_resource (any other per-application \
75                          path: answer it or 404 it with send_spark_error)"
76                .to_string(),
77            required: true,
78        },
79        Parameter {
80            name: "app_id".to_string(),
81            type_hint: "string".to_string(),
82            description: "Application id from the path (for per-application operations)"
83                .to_string(),
84            required: false,
85        },
86    ])
87    .with_actions(spark_actions())
88    // Only fields `spark_request` actually carries. `{client_ip}` and `{client_port}` were
89    // in this template and in none of the event data, and a missing field renders as the
90    // empty string — so the INFO line read "Spark  GET /api/v1/applications".
91    .with_log_template(
92        LogTemplate::new()
93            .with_info("Spark {method} {path}")
94            .with_debug("Spark {method} {path} op={operation} app={app_id}")
95            .with_trace("Spark: {json_pretty(.)}"),
96    )
97});
98
99fn spark_actions() -> Vec<ActionDefinition> {
100    vec![
101        send_spark_applications_action(),
102        send_spark_jobs_action(),
103        send_spark_stages_action(),
104        send_spark_executors_action(),
105        send_spark_rdds_action(),
106        send_spark_error_action(),
107    ]
108}
109
110fn send_spark_rdds_action() -> ActionDefinition {
111    ActionDefinition {
112        name: "send_spark_rdds".to_string(),
113        description: "Answer GET /api/v1/applications/{id}/storage/rdd (operation rdd): the \
114                      application's cached RDDs. Supply the `rdds` array; the response is the \
115                      array itself, and an empty array means nothing is cached."
116            .to_string(),
117        parameters: vec![Parameter {
118            name: "rdds".to_string(),
119            type_hint: "array".to_string(),
120            description: "Array of RDD storage objects. Common fields: id (number), name, \
121                          numPartitions, numCachedPartitions, storageLevel (e.g. \"Memory \
122                          Deserialized 1x Replicated\"), memoryUsed, diskUsed (bytes)."
123                .to_string(),
124            required: true,
125        }],
126        example: json!({
127            "type": "send_spark_rdds",
128            "rdds": [{
129                "id": 3, "name": "ParallelCollectionRDD", "numPartitions": 4,
130                "numCachedPartitions": 4, "storageLevel": "Memory Deserialized 1x Replicated",
131                "memoryUsed": 2048, "diskUsed": 0
132            }]
133        }),
134        log_template: Some(LogTemplate::new().with_info("-> Spark {rdds_len} cached RDDs")),
135    }
136}
137
138fn send_spark_applications_action() -> ActionDefinition {
139    ActionDefinition {
140        name: "send_spark_applications".to_string(),
141        description: "Answer GET /api/v1/applications (list) or GET /api/v1/applications/{id} \
142                      (single). Supply the `applications` array; the response is the array \
143                      itself (Spark returns a bare JSON array). For the single-app endpoint \
144                      (operation application) return exactly one element; it is sent as the \
145                      bare object, as Spark sends it."
146            .to_string(),
147        parameters: vec![Parameter {
148            name: "applications".to_string(),
149            type_hint: "array".to_string(),
150            description: "Array of application objects. Each: id (e.g. \
151                          app-20161116163331-0000), name, and an `attempts` array whose \
152                          elements carry startTime, endTime, lastUpdated, duration, sparkUser, \
153                          completed (bool), appSparkVersion."
154                .to_string(),
155            required: true,
156        }],
157        example: json!({
158            "type": "send_spark_applications",
159            "applications": [{
160                "id": "app-20161116163331-0000", "name": "Spark shell",
161                "attempts": [{
162                    "startTime": "2016-11-16T22:33:29.916GMT",
163                    "endTime": "1969-12-31T23:59:59.999GMT",
164                    "duration": 0, "sparkUser": "jose", "completed": false,
165                    "appSparkVersion": "3.5.1"
166                }]
167            }]
168        }),
169        log_template: Some(
170            LogTemplate::new().with_info("-> Spark {applications_len} applications"),
171        ),
172    }
173}
174
175fn send_spark_jobs_action() -> ActionDefinition {
176    ActionDefinition {
177        name: "send_spark_jobs".to_string(),
178        description: "Answer GET /api/v1/applications/{id}/jobs. Supply the `jobs` array; the \
179                      response is the array itself."
180            .to_string(),
181        parameters: vec![Parameter {
182            name: "jobs".to_string(),
183            type_hint: "array".to_string(),
184            description: "Array of job objects. Common fields: jobId (number), name, \
185                          submissionTime, completionTime, stageIds (array of numbers), status \
186                          (RUNNING/SUCCEEDED/FAILED/UNKNOWN), numTasks, numActiveTasks, \
187                          numCompletedTasks, numSkippedTasks, numFailedTasks, numActiveStages, \
188                          numCompletedStages, numFailedStages."
189                .to_string(),
190            required: true,
191        }],
192        example: json!({
193            "type": "send_spark_jobs",
194            "jobs": [{
195                "jobId": 0, "name": "count at <console>:15", "status": "SUCCEEDED",
196                "stageIds": [0], "numTasks": 8, "numActiveTasks": 0, "numCompletedTasks": 8,
197                "numFailedTasks": 0, "numActiveStages": 0, "numCompletedStages": 1,
198                "numFailedStages": 0
199            }]
200        }),
201        log_template: Some(LogTemplate::new().with_info("-> Spark {jobs_len} jobs")),
202    }
203}
204
205fn send_spark_stages_action() -> ActionDefinition {
206    ActionDefinition {
207        name: "send_spark_stages".to_string(),
208        description: "Answer GET /api/v1/applications/{id}/stages. Supply the `stages` array; \
209                      the response is the array itself."
210            .to_string(),
211        parameters: vec![Parameter {
212            name: "stages".to_string(),
213            type_hint: "array".to_string(),
214            description: "Array of stage objects. Common fields: status \
215                          (ACTIVE/COMPLETE/PENDING/FAILED/SKIPPED), stageId (number), attemptId \
216                          (number), name, numTasks, numActiveTasks, numCompleteTasks, \
217                          numFailedTasks, inputBytes, outputBytes, shuffleReadBytes, \
218                          shuffleWriteBytes."
219                .to_string(),
220            required: true,
221        }],
222        example: json!({
223            "type": "send_spark_stages",
224            "stages": [{
225                "status": "COMPLETE", "stageId": 0, "attemptId": 0, "numTasks": 8,
226                "numActiveTasks": 0, "numCompleteTasks": 8, "numFailedTasks": 0,
227                "name": "count at <console>:15"
228            }]
229        }),
230        log_template: Some(LogTemplate::new().with_info("-> Spark {stages_len} stages")),
231    }
232}
233
234fn send_spark_executors_action() -> ActionDefinition {
235    ActionDefinition {
236        name: "send_spark_executors".to_string(),
237        description: "Answer GET /api/v1/applications/{id}/executors. Supply the `executors` \
238                      array; the response is the array itself. There is always at least the \
239                      `driver` executor."
240            .to_string(),
241        parameters: vec![Parameter {
242            name: "executors".to_string(),
243            type_hint: "array".to_string(),
244            description: "Array of executor objects. Common fields: id (\"driver\" or a number \
245                          as string), hostPort, isActive (bool), rddBlocks, memoryUsed, \
246                          diskUsed, totalCores, maxTasks, activeTasks, failedTasks, \
247                          completedTasks, totalTasks, totalDuration, maxMemory."
248                .to_string(),
249            required: true,
250        }],
251        example: json!({
252            "type": "send_spark_executors",
253            "executors": [{
254                "id": "driver", "hostPort": "10.0.0.1:57971", "isActive": true, "rddBlocks": 0,
255                "memoryUsed": 0, "diskUsed": 0, "totalCores": 8, "maxTasks": 8, "activeTasks": 0,
256                "failedTasks": 0, "completedTasks": 16, "totalTasks": 16, "maxMemory": 384093388
257            }]
258        }),
259        log_template: Some(LogTemplate::new().with_info("-> Spark {executors_len} executors")),
260    }
261}
262
263fn send_spark_error_action() -> ActionDefinition {
264    ActionDefinition {
265        name: "send_spark_error".to_string(),
266        description: "Return an error with an explicit HTTP status. Spark's REST API sends error \
267                      bodies as plain text (e.g. 404 \"unknown app: app-1\"); this action does \
268                      the same. status must be 100-599."
269            .to_string(),
270        parameters: vec![
271            Parameter {
272                name: "status".to_string(),
273                type_hint: "number".to_string(),
274                description: "HTTP status code (400, 404, 500, ...)".to_string(),
275                required: true,
276            },
277            Parameter {
278                name: "message".to_string(),
279                type_hint: "string".to_string(),
280                description: "Plain-text error message".to_string(),
281                required: true,
282            },
283        ],
284        example: json!({
285            "type": "send_spark_error",
286            "status": 404,
287            "message": "unknown app: app-does-not-exist"
288        }),
289        log_template: Some(LogTemplate::new().with_info("-> Spark error {status}")),
290    }
291}
292
293pub fn get_spark_event_types() -> Vec<EventType> {
294    vec![SPARK_REQUEST_EVENT.clone()]
295}
296
297fn array_body(action: &Value, key: &str) -> Result<Value> {
298    action
299        .get(key)
300        .filter(|v| v.is_array())
301        .cloned()
302        .ok_or_else(|| anyhow::anyhow!("Missing '{key}' array"))
303}
304
305fn parse_status(action: &Value) -> Result<u16> {
306    let raw = action
307        .get("status")
308        .and_then(|v| v.as_u64())
309        .ok_or_else(|| anyhow::anyhow!("Missing or invalid 'status': expected a number"))?;
310    if !(100..=599).contains(&raw) {
311        return Err(anyhow::anyhow!(
312            "Invalid status {raw}: must be an HTTP status between 100 and 599"
313        ));
314    }
315    Ok(raw as u16)
316}
317
318impl Protocol for SparkProtocol {
319    fn get_startup_parameters(&self) -> Vec<crate::llm::actions::ParameterDefinition> {
320        use crate::llm::actions::ParameterDefinition;
321        vec![ParameterDefinition {
322            name: "spark_version".to_string(),
323            type_hint: "string".to_string(),
324            description: "Version string reported by the static GET /api/v1/version banner. \
325                          Default 3.5.1."
326                .to_string(),
327            required: false,
328            example: json!("3.5.1"),
329            default: None,
330        }]
331    }
332    fn get_async_actions(&self, _state: &AppState) -> Vec<ActionDefinition> {
333        vec![]
334    }
335    fn get_sync_actions(&self) -> Vec<ActionDefinition> {
336        spark_actions()
337    }
338    fn protocol_name(&self) -> &'static str {
339        "Spark"
340    }
341    fn get_event_types(&self) -> Vec<EventType> {
342        get_spark_event_types()
343    }
344    fn stack_name(&self) -> &'static str {
345        "ETH>IP>TCP>HTTP>SPARK"
346    }
347    fn keywords(&self) -> Vec<&'static str> {
348        vec!["spark", "apache spark", "spark rest api"]
349    }
350    fn metadata(&self) -> crate::protocol::metadata::ProtocolMetadataV2 {
351        use crate::protocol::metadata::{DevelopmentState, ProtocolMetadataV2};
352        ProtocolMetadataV2::builder()
353            .state(DevelopmentState::Experimental)
354            .well_known_port(4040)
355            .implementation("hyper v1 HTTP/1.1 server, manual Spark monitoring REST arrays")
356            .llm_control("Applications, jobs, stages, executors and cached RDDs invented by the LLM")
357            .e2e_testing(
358                "tests/server/spark/datadog_check_test.rs runs Datadog's Spark integration \
359                 (datadog_checks.spark, integrations-core tag spark-7.8.0) once in driver mode \
360                 against the server and asserts the metrics, tags, service checks and version it \
361                 submitted from the model's answers (applications, jobs, stages, executors, \
362                 storage/rdd, a streaming-statistics 404). e2e_test.rs checks the documented \
363                 shapes with reqwest. NetGet's own Spark client is tested against real Spark, \
364                 not against this server",
365            )
366            .notes(
367                "Virtual application, no storage. GET /api/v1/version is answered \
368                    statically; applications/jobs/stages/executors/storage-rdd are LLM-driven, \
369                    and GET /api/v1/applications/{id} is sent as the one object Spark sends. \
370                    Fail-closed: LLM failure returns 503/500 JSON error, never an empty-but-200 \
371                    array. Experimental: one independent consumer (Datadog's check), which reads \
372                    the list endpoints only; nothing has yet driven the per-job, per-stage, \
373                    environment or SQL endpoints, which reach the model as \
374                    application_resource.",
375            )
376            .max_inbound_bytes(crate::server::spark::MAX_REQUEST_BYTES)
377            .request_only("The Spark REST API is HTTP request/response; a server cannot send a peer anything unprompted")
378            .build()
379    }
380    fn description(&self) -> &'static str {
381        "Apache Spark monitoring REST / history API"
382    }
383    fn example_prompt(&self) -> &'static str {
384        "Start an Apache Spark REST API on port 4040 reporting one running application"
385    }
386    fn group_name(&self) -> &'static str {
387        "Application"
388    }
389
390    fn get_startup_examples(&self) -> crate::llm::actions::StartupExamples {
391        use crate::llm::actions::StartupExamples;
392
393        // Deterministic: report one running application for every monitoring
394        // API request, no LLM call.
395        let script = r#"import json, sys
396data = json.load(sys.stdin)
397if data["event_type_id"] == "spark_request":
398    actions = [{"type": "send_spark_applications",
399                "applications": [{"id": "app-netget-0000", "name": "netget-app",
400                    "attempts": [{"completed": False, "appSparkVersion": "3.5.1"}]}]}]
401else:
402    actions = []
403print(json.dumps({"actions": actions}))"#;
404
405        StartupExamples::new(
406            json!({
407                "type": "open_server",
408                "port": 4040,
409                "base_stack": "spark",
410                "instruction": "Apache Spark driver monitoring API with one running job"
411            }),
412            json!({
413                "type": "open_server",
414                "port": 4040,
415                "base_stack": "spark",
416                "event_handlers": [{
417                    "event_pattern": "spark_request",
418                    "handler": { "type": "script", "language": "python", "code": script }
419                }]
420            }),
421            json!({
422                "type": "open_server",
423                "port": 4040,
424                "base_stack": "spark",
425                "event_handlers": [{
426                    "event_pattern": "spark_request",
427                    "handler": {
428                        "type": "static",
429                        "actions": [{
430                            "type": "send_spark_applications",
431                            "applications": [{
432                                "id": "app-netget-0000", "name": "netget-app",
433                                "attempts": [{ "completed": false, "appSparkVersion": "3.5.1" }]
434                            }]
435                        }]
436                    }
437                }]
438            }),
439        )
440    }
441}
442
443impl Server for SparkProtocol {
444    fn spawn(
445        &self,
446        ctx: crate::protocol::SpawnContext,
447    ) -> std::pin::Pin<
448        Box<dyn std::future::Future<Output = anyhow::Result<std::net::SocketAddr>> + Send>,
449    > {
450        Box::pin(async move {
451            let spark_version = ctx
452                .startup_params
453                .as_ref()
454                .and_then(|p| p.get_optional_string("spark_version").ok().flatten())
455                .unwrap_or_else(|| "3.5.1".to_string());
456
457            crate::server::spark::SparkServer::spawn_with_llm_actions(
458                ctx.legacy_listen_addr(),
459                ctx.llm_client,
460                ctx.state,
461                ctx.status_tx,
462                ctx.server_id,
463                spark_version,
464            )
465            .await
466        })
467    }
468
469    fn execute_action(&self, action: Value) -> Result<ActionResult> {
470        let action_type = action
471            .get("type")
472            .and_then(|v| v.as_str())
473            .ok_or_else(|| anyhow::anyhow!("Missing action type"))?;
474
475        // Spark success bodies are bare JSON arrays; errors are plain text.
476        let json_array = |status: u16, body: Value| {
477            Ok(ActionResult::Custom {
478                name: "spark_response".to_string(),
479                data: json!({
480                    "status": status,
481                    "body": serde_json::to_string(&body).unwrap_or_else(|_| "[]".to_string()),
482                    "content_type": "application/json",
483                }),
484            })
485        };
486
487        match action_type {
488            "send_spark_applications" => json_array(200, array_body(&action, "applications")?),
489            "send_spark_jobs" => json_array(200, array_body(&action, "jobs")?),
490            "send_spark_stages" => json_array(200, array_body(&action, "stages")?),
491            "send_spark_executors" => json_array(200, array_body(&action, "executors")?),
492            "send_spark_rdds" => json_array(200, array_body(&action, "rdds")?),
493            "send_spark_error" => {
494                let status = parse_status(&action)?;
495                let message = action
496                    .get("message")
497                    .and_then(|v| v.as_str())
498                    .ok_or_else(|| anyhow::anyhow!("Missing 'message'"))?;
499                Ok(ActionResult::Custom {
500                    name: "spark_response".to_string(),
501                    data: json!({
502                        "status": status,
503                        "body": message,
504                        "content_type": "text/plain",
505                    }),
506                })
507            }
508            _ => Err(anyhow::anyhow!("Unknown action type: {}", action_type)),
509        }
510    }
511}
netget/server/spark/actions.rs

netget/server/spark/
actions.rs
