SK-246 // transform trace on export via a SKEL file

sk-api/schema/v1/simkube.yml

@@ -45,6 +45,8 @@ paths:
                       type: array
                       items:
                         $ref: 'https://raw.githubusercontent.com/kubernetes/kubernetes/master/api/openapi-spec/v3/api__v1_openapi.json#/components/schemas/io.k8s.apimachinery.pkg.apis.meta.v1.LabelSelector'  # yamllint disable-line rule:line-length
+                transform:
+                  type: string
       responses:
         '200':
           description: OK

sk-api/src/v1/export_filters.rs

@@ -10,7 +10,7 @@
 
 use super::*;
 
-#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)]
 pub struct ExportFilters {
     #[serde(rename = "excluded_namespaces")]
     pub excluded_namespaces: Vec<String>,

sk-api/src/v1/export_request.rs

@@ -10,7 +10,7 @@
 
 use super::*;
 
-#[derive(Clone, Default, Debug, PartialEq, Serialize, Deserialize)]
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)]
 pub struct ExportRequest {
     #[serde(rename = "start_ts")]
     pub start_ts: i64,
@@ -20,6 +20,8 @@ pub struct ExportRequest {
     pub export_path: String,
     #[serde(rename = "filters")]
     pub filters: Box<ExportFilters>,
+    #[serde(rename = "transform", skip_serializing_if = "Option::is_none")]
+    pub transform: Option<String>,
 }
 
 impl ExportRequest {
@@ -29,6 +31,7 @@ impl ExportRequest {
             end_ts,
             export_path,
             filters: Box::new(filters),
+            transform: None,
         }
     }
 }

sk-cli/src/export.rs

@@ -1,3 +1,5 @@
+use std::fs;
+
 use anyhow::bail;
 use sk_api::v1::{
     ExportFilters,
@@ -43,6 +45,9 @@ pub struct Args {
     )]
     pub excluded_namespaces: Vec<String>,
 
+    #[arg(long, long_help = "location of .skel transformation file")]
+    pub transform: Option<String>,
+
     #[arg(
         long,
         long_help = "sk-tracer server address",
@@ -61,7 +66,14 @@ pub struct Args {
 
 pub async fn cmd(args: &Args) -> EmptyResult {
     let filters = ExportFilters::new(args.excluded_namespaces.clone(), vec![]);
-    let req = ExportRequest::new(args.start_time, args.end_time, args.output_path.clone(), filters);
+    let transform = args.transform.as_ref().map(fs::read_to_string).transpose()?;
+    let req = ExportRequest {
+        start_ts: args.start_time,
+        end_ts: args.end_time,
+        export_path: args.output_path.clone(),
+        filters: Box::new(filters),
+        transform,
+    };
     let endpoint = format!("{}/export", args.tracer_address);
 
     println!("exporting trace data");

sk-cli/src/snapshot.rs

@@ -42,7 +42,12 @@ pub async fn cmd(args: &Args) -> EmptyResult {
     let filters = ExportFilters::new(args.excluded_namespaces.clone(), vec![]);
     let start_ts = UtcClock.now_ts();
     let end_ts = start_ts + 1;
-    let data = manager.get_store().lock().await.export(start_ts, end_ts, &filters).await?;
+    let data = manager
+        .get_store()
+        .lock()
+        .await
+        .export(start_ts, end_ts, &filters, None)
+        .await?;
 
     println!("Writing trace file: {}", args.output);
     let mut file = File::create(&args.output)?;

sk-cli/src/transform.rs

@@ -1,3 +1,4 @@
+use std::fs;
 use std::sync::mpsc;
 use std::time::Duration;
 
@@ -22,9 +23,9 @@ use sk_core::external_storage::{
     SkObjectStore,
 };
 use sk_core::prelude::*;
-use sk_skel::apply_skel_file;
 use sk_skel::metric_names::*;
 use sk_store::ExportedTrace;
+use sk_store::process::apply_skel;
 
 const SPINNER_REFRESH_RATE: u64 = 50;
 const SPINNER_DOTS: &[&str] = &["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
@@ -62,8 +63,8 @@ pub async fn cmd(args: &Args) -> EmptyResult {
 
     let clock = UtcClock::new();
     let start_time = clock.now();
-    let skel_filename = args.skel_file.clone();
-    let transform_task = tokio::spawn(async move { apply_skel_file(&trace, &skel_filename, tx).await });
+    let skel_contents = fs::read_to_string(args.skel_file.clone())?;
+    let transform_task = tokio::spawn(async move { apply_skel(&trace, &skel_contents, tx).await });
     loop {
         // don't really care about errors on progress bar updates right now
         let _ = progress_bar.refresh();

sk-cli/src/validation/rules/missing_resources.rs

@@ -9,9 +9,9 @@ use corev1::{
 };
 use json_patch_ext::prelude::*;
 use k8s_openapi::Resource;
+use sk_core::event::TraceEvent;
 use sk_core::k8s::GVK;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 
 use crate::validation::validator::{
     Diagnostic,

sk-cli/src/validation/rules/status_field_populated.rs

@@ -1,5 +1,5 @@
+use sk_core::event::TraceEvent;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 
 use crate::validation::validator::{
     Diagnostic,

sk-cli/src/validation/rules/tests/missing_resources_test.rs

@@ -1,6 +1,6 @@
 use assertables::*;
 use serde_json::json;
-use sk_store::TraceEvent;
+use sk_core::event::TraceEvent;
 
 use super::missing_resources::service_account_validator;
 use super::*;

sk-cli/src/validation/rules/tests/status_field_populated_test.rs

@@ -1,6 +1,6 @@
 use assertables::*;
 use serde_json::json;
-use sk_store::TraceEvent;
+use sk_core::event::TraceEvent;
 
 use super::*;
 

sk-cli/src/validation/tests/mod.rs

@@ -2,8 +2,8 @@ mod validation_store_test;
 
 use std::collections::BTreeMap;
 
+use sk_core::event::TraceEvent;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 use sk_testutils::*;
 
 use super::validator::{

sk-cli/src/validation/validator.rs

@@ -6,8 +6,8 @@ use serde::{
     Serialize,
     Serializer,
 };
+use sk_core::event::TraceEvent;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 
 #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd, Serialize)]
 pub enum ValidatorType {

sk-cli/src/xray/tests/app_test.rs

@@ -1,8 +1,6 @@
 use ratatui::widgets::ListState;
-use sk_store::{
-    ExportedTrace,
-    TraceEvent,
-};
+use sk_core::event::TraceEvent;
+use sk_store::ExportedTrace;
 
 use super::*;
 

sk-cli/src/xray/view/helpers.rs

@@ -2,8 +2,8 @@ use std::sync::LazyLock;
 
 use chrono::TimeDelta;
 use ratatui::prelude::*;
+use sk_core::event::TraceEvent;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 
 use crate::validation::{
     Annotations,

sk-core/src/event.rs

@@ -0,0 +1,29 @@
+use serde::{
+    Deserialize,
+    Serialize,
+};
+
+use crate::prelude::*;
+
+#[derive(Clone, Copy, Debug)]
+pub enum TraceAction {
+    ObjectApplied,
+    ObjectDeleted,
+}
+
+#[derive(Clone, Debug, Default, Deserialize, PartialEq, Serialize)]
+pub struct TraceEvent {
+    pub ts: i64,
+    pub applied_objs: Vec<DynamicObject>,
+    pub deleted_objs: Vec<DynamicObject>,
+}
+
+impl TraceEvent {
+    pub fn len(&self) -> usize {
+        self.applied_objs.len() + self.deleted_objs.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.applied_objs.is_empty() && self.deleted_objs.is_empty()
+    }
+}

sk-core/src/lib.rs

@@ -2,6 +2,7 @@
 pub mod config;
 pub mod constants;
 pub mod errors;
+pub mod event;
 pub mod events;
 pub mod external_storage;
 pub mod hooks;

sk-driver/src/tests/helpers.rs

@@ -1,6 +1,6 @@
 use std::sync::Arc;
 
-use sk_store::TraceEvent;
+use sk_core::event::TraceEvent;
 use tokio::sync::Mutex;
 
 use super::*;

sk-skel/Cargo.toml

@@ -18,7 +18,6 @@ pest_derive = { workspace = true }
 regex = { workspace = true }
 serde_json = { workspace = true }
 sk-core = { workspace = true }
-sk-store = { workspace = true }
 thiserror = { workspace = true }
 
 [dev-dependencies]

sk-skel/src/ast.rs

@@ -71,7 +71,7 @@ pub struct Command {
     pub(super) action: CommandAction,
 }
 
-pub(super) fn parse_command(cmd: Pair<Rule>, trace_start_ts: i64) -> anyhow::Result<Command> {
+pub fn parse_command(cmd: Pair<Rule>, trace_start_ts: i64) -> anyhow::Result<Command> {
     let rule = cmd.as_rule();
     let args = cmd.into_inner();
 

sk-skel/src/engine.rs

@@ -4,8 +4,8 @@ use json_patch_ext::prelude::*;
 use metrics::counter;
 use regex::Regex;
 use serde_json::Value;
+use sk_core::event::TraceEvent;
 use sk_core::prelude::*;
-use sk_store::TraceEvent;
 
 use crate::ast::{
     Command,
@@ -20,7 +20,7 @@ use crate::context::*;
 use crate::errors::SkelError;
 use crate::metric_names::*;
 
-pub(super) fn process_event(cmd: &Command, mut evt: TraceEvent) -> anyhow::Result<TraceEvent> {
+pub fn process_event(cmd: &Command, mut evt: TraceEvent) -> anyhow::Result<TraceEvent> {
     // Reinterpret the event as a JSON object so JSON pointers work for all fields
     // This seems sortof inefficient, we'll have to see how it works in practice.
     let mut applied_objs = serde_json::to_value(evt.applied_objs)?;

sk-skel/src/lib.rs

@@ -1,23 +1,10 @@
-mod ast;
+pub mod ast;
+pub mod engine;
+
 mod context;
-mod engine;
 mod errors;
 
-use std::fs;
-use std::sync::mpsc;
-
-use pest::Parser;
 use pest_derive::Parser;
-use sk_store::{
-    ExportedTrace,
-    TraceEvent,
-};
-
-use self::ast::{
-    Command,
-    parse_command,
-};
-use self::engine::process_event;
 
 pub mod metric_names {
     pub const EVENT_MATCHED_COUNTER: &str = "trace_events_matched";
@@ -28,58 +15,7 @@ pub mod metric_names {
 #[allow(dead_code)]
 #[derive(Parser)]
 #[grammar = "src/skel.pest"]
-struct SkelParser;
-
-pub fn process_trace(
-    trace: &ExportedTrace,
-    commands: &Vec<Command>,
-    update_channel: Option<mpsc::Sender<()>>,
-) -> anyhow::Result<Vec<TraceEvent>> {
-    let mut new_events = Vec::with_capacity(trace.events.len());
-    for (evt, _) in trace.iter() {
-        let mut new_event = evt.clone();
-        for cmd in commands {
-            new_event = process_event(cmd, new_event)?;
-        }
-
-        // Only add the event if it actually does anything
-        if !new_event.applied_objs.is_empty() || !new_event.deleted_objs.is_empty() {
-            new_events.push(new_event);
-        }
-        if let Some(ref c) = update_channel {
-            let _ = c.send(()); // if we can't send on the channel, nbd
-        }
-    }
-
-    Ok(new_events)
-}
-
-pub async fn apply_skel_file(
-    trace: &ExportedTrace,
-    skel_file: &str,
-    update_channel: mpsc::Sender<()>,
-) -> anyhow::Result<ExportedTrace> {
-    let skel_str = fs::read_to_string(skel_file)?;
-    let skel = SkelParser::parse(Rule::skel, &skel_str)?;
-
-    let parsed_commands = skel
-        .filter_map(|cmd| match cmd.as_rule() {
-            Rule::EOI => None,
-            _ => Some(parse_command(cmd, trace.start_ts().unwrap_or_default())),
-        })
-        .collect::<Result<Vec<_>, anyhow::Error>>()?;
-
-    let new_events = process_trace(trace, &parsed_commands, Some(update_channel))?;
-    let new_trace = ExportedTrace {
-        version: trace.version,
-        config: trace.config.clone(),
-        events: new_events,
-        index: trace.index.clone(),
-        pod_lifecycles: trace.pod_lifecycles.clone(),
-    };
-
-    Ok(new_trace)
-}
+pub struct SkelParser;
 
 #[cfg(test)]
 mod tests;

sk-skel/src/tests/ast_test.rs

@@ -1,6 +1,7 @@
 use std::collections::HashSet;
 
 use json_patch_ext::PointerBuf;
+use pest::Parser;
 use serde_json::{
     Value,
     json,