From 8c716f00705a74e133d6716bbc9b41c0118a53b6 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Tue, 23 Jun 2026 16:23:46 +0000 Subject: [PATCH] [fda] Add commands for viewing and managing tags and tagged pipelines Signed-off-by: Karakatiza666 --- crates/fda/src/cli.rs | 31 ++++ crates/fda/src/main.rs | 120 ++++++++++++- crates/fda/src/tags.rs | 239 +++++++++++++++++++++++++ docs.feldera.com/docs/interface/cli.md | 9 + 4 files changed, 398 insertions(+), 1 deletion(-) create mode 100644 crates/fda/src/tags.rs diff --git a/crates/fda/src/cli.rs b/crates/fda/src/cli.rs index 9509739912f..e0da94e7f73 100644 --- a/crates/fda/src/cli.rs +++ b/crates/fda/src/cli.rs @@ -320,6 +320,12 @@ pub enum PipelineAction { /// The compilation profile to use. #[arg(default_value = "optimized")] profile: CompilationProfile, + /// A tag to assign to the pipeline. + /// + /// Repeat the flag to assign several tags, e.g. `--tag prod --tag team-billing`. + /// Tags are deduplicated and stored in sorted order. + #[arg(long = "tag", value_hint = ValueHint::Other)] + tags: Vec, /// Read the program code from stdin. /// /// EXAMPLES: @@ -515,6 +521,31 @@ pub enum PipelineAction { /// The new value for the configuration. value: String, }, + /// Retrieve the tags of a pipeline. + /// + /// Prints the tags as a comma-separated list, in sorted order. + Tags { + /// The name of the pipeline. + #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))] + name: String, + }, + /// Replace the tags of a pipeline. + /// + /// Takes the new tags as a single comma-separated list, replacing whatever the + /// pipeline carried before; pass an empty list to clear all tags. To append + /// instead, include the current tags, e.g. + /// `fda set-tags my-pipeline $(fda tags my-pipeline),d,e`. + /// + /// A tag containing spaces must be quoted, e.g. `"team billing",prod`. Each tag + /// may be named alone; its color is filled in from the same tag used elsewhere. + SetTags { + /// The name of the pipeline. + #[arg(value_hint = ValueHint::Other, add = ArgValueCompleter::new(pipeline_names))] + name: String, + /// The new tags, as a comma-separated list. Omit to clear all tags. + #[arg(default_value = "")] + tags: String, + }, /// Recompile a pipeline with the Feldera runtime version included in the /// currently installed Feldera platform. UpdateRuntime { diff --git a/crates/fda/src/main.rs b/crates/fda/src/main.rs index ef8a13a11c4..5f71c69e25d 100644 --- a/crates/fda/src/main.rs +++ b/crates/fda/src/main.rs @@ -33,6 +33,7 @@ mod bench; mod cli; mod debug; mod shell; +mod tags; pub(crate) const UPGRADE_NOTICE: &str = "Try upgrading to the latest CLI version to resolve this issue. Also make sure the pipeline is recompiled with the latest version of feldera. Report it on github.com/feldera/feldera if the issue persists."; @@ -677,6 +678,98 @@ async fn wait_for_checkpoint(client: &Client, name: String, seq_number: u64, wai } } +/// Fetch a pipeline's current tags. +async fn get_pipeline_tags(client: &Client, name: &str) -> Vec { + client + .get_pipeline() + .pipeline_name(name) + .send() + .await + .map_err(handle_errors_fatal( + client.baseurl().clone(), + "Failed to get pipeline tags", + 1, + )) + .unwrap() + .into_inner() + .tags +} + +/// Collect the tags used across every pipeline. +/// +/// This is the pool of "known tags": when a tag name is colored on one pipeline, +/// setting it on another borrows that color so the name stays one consistent +/// color everywhere. +async fn get_known_tags(client: &Client) -> Vec { + client + .list_pipelines() + .send() + .await + .map_err(handle_errors_fatal( + client.baseurl().clone(), + "Failed to list pipeline tags", + 1, + )) + .unwrap() + .into_inner() + .into_iter() + .flat_map(|pipeline| pipeline.tags) + .collect() +} + +/// Replace a pipeline's tags and return the stored result. +async fn set_pipeline_tags(client: &Client, name: &str, tags: Vec) -> Vec { + client + .patch_pipeline() + .pipeline_name(name) + .body(PatchPipeline { + description: None, + tags: Some(tags), + name: None, + program_code: None, + udf_rust: None, + udf_toml: None, + program_config: None, + runtime_config: None, + }) + .send() + .await + .map_err(handle_errors_fatal( + client.baseurl().clone(), + "Failed to update pipeline tags", + 1, + )) + .unwrap() + .into_inner() + .tags +} + +/// Print a pipeline's tags as a comma-separated list of display names, or as a +/// JSON array of the raw stored strings. +/// +/// Text output shows display names: the color suffix that encodes a tag's color +/// is stripped, since it is noise to a human reader and the same comma-separated +/// list feeds straight back into `set-tags`. JSON output keeps the raw strings, +/// so a tag's color round-trips for machine consumers. +fn print_pipeline_tags(format: OutputFormat, tags: &[String]) { + match format { + OutputFormat::Text => { + let names: Vec<&str> = tags.iter().map(|tag| tags::tag_display_name(tag)).collect(); + println!("{}", names.join(",")); + } + OutputFormat::Json => { + println!( + "{}", + serde_json::to_string_pretty(tags).expect("Failed to serialize tags") + ); + } + _ => { + eprintln!("Unsupported output format: {}", format); + std::process::exit(1); + } + } +} + async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client) { match action { PipelineAction::Create { @@ -685,6 +778,7 @@ async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client) runtime_version, use_platform_compiler, profile, + tags, udf_rs, udf_toml, stdin, @@ -694,11 +788,19 @@ async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client) read_file(udf_rs).await, read_file(udf_toml).await, ) { + // Each `--tag` borrows its color from the tags already used across + // pipelines, keeping a name one consistent color. The lookup is + // skipped when no tags were given. + let tags = if tags.is_empty() { + Vec::new() + } else { + tags::set_tags(tags, &get_known_tags(&client).await) + }; let response = client .post_pipeline() .body(PostPutPipeline { description: None, - tags: Vec::new(), + tags, name: name.to_string(), program_code: program_code.unwrap_or_default(), udf_rust, @@ -1438,6 +1540,22 @@ async fn pipeline(format: OutputFormat, action: PipelineAction, client: Client) } } } + PipelineAction::Tags { name } => { + let current = get_pipeline_tags(&client, &name).await; + print_pipeline_tags(format, ¤t); + } + PipelineAction::SetTags { name, tags } => { + let requested = tags::split_tag_list(&tags); + // A tag may be named alone; its color is borrowed from the same tag + // used across pipelines. + let known = get_known_tags(&client).await; + let updated = + set_pipeline_tags(&client, &name, tags::set_tags(requested, &known)).await; + if matches!(format, OutputFormat::Text) { + println!("Tags updated successfully."); + } + print_pipeline_tags(format, &updated); + } PipelineAction::UpdateRuntime { name } => { let response = client .post_update_runtime() diff --git a/crates/fda/src/tags.rs b/crates/fda/src/tags.rs new file mode 100644 index 00000000000..2c2088b902b --- /dev/null +++ b/crates/fda/src/tags.rs @@ -0,0 +1,239 @@ +//! Pipeline tag conventions shared across Feldera clients. +//! +//! The backend stores a tag as an opaque string, validates it against a fixed +//! character set, and knows nothing about color. A tag's color is encoded into +//! the string itself as a suffix that character set permits: the visible name, +//! then a `|` separator, then a six-digit `rrggbb` hex color (no leading `#`, +//! which the backend disallows) -- e.g. `"prod|ef4444"`. The suffix is optional, +//! so a bare `"prod"` is an uncolored tag; `"prod|ef4444"` and `"prod|22c55e"` +//! are the same tag in two colors. +//! +//! The CLI treats a tag's *display name* -- its name with any color suffix +//! removed -- as the tag's identity, and enforces two invariants whenever a +//! pipeline's tags are written: +//! +//! * **Variant exclusivity.** A pipeline carries at most one tag per display +//! name. When the input holds several color variants of the same name, the +//! last one wins. +//! * **Lexicographic order.** Tags are stored sorted, so the stored order is +//! stable regardless of the order they were supplied in. +//! +//! Both operate on the full string, so a tag's color is preserved; only the +//! display name decides which tags collide. A name alone is enough to refer to a +//! tag: its color is filled in from the same tag elsewhere, so no command +//! requires the user to type the raw color suffix. + +use std::collections::HashMap; + +/// Split a comma-separated tag list into individual tags. +/// +/// Empty segments are dropped, so an empty string yields no tags -- the way +/// `set-tags` clears a pipeline -- and a stray trailing comma is harmless. Only +/// the comma separates tags, so spaces are kept and a quoted tag such as +/// `"team billing"` survives intact. +pub fn split_tag_list(input: &str) -> Vec { + input + .split(',') + .filter(|tag| !tag.is_empty()) + .map(|tag| tag.to_string()) + .collect() +} + +/// A tag's visible text: its name, with any trailing `|rrggbb` color suffix +/// removed. +/// +/// The suffix is recognized only when the text after the last `|` is exactly six +/// hex digits, so a name that itself contains `|` (without a trailing color) is +/// returned whole. +pub fn tag_display_name(tag: &str) -> &str { + if let Some(separator) = tag.rfind('|') + && is_hex_color(&tag[separator + 1..]) + { + return &tag[..separator]; + } + tag +} + +/// Whether `text` is a six-digit `rrggbb` hex color. +fn is_hex_color(text: &str) -> bool { + text.len() == 6 && text.bytes().all(|b| b.is_ascii_hexdigit()) +} + +/// Collapse color variants of the same tag and sort the result. +/// +/// Tags that share a display name are deduplicated, keeping the last occurrence, +/// and the survivors are returned in lexicographic order. The color suffix that +/// encodes a tag's color is preserved. +pub fn normalize_tags(tags: I) -> Vec +where + I: IntoIterator, + S: AsRef, +{ + let mut by_display_name: HashMap = HashMap::new(); + for tag in tags { + let tag = tag.as_ref(); + by_display_name.insert(tag_display_name(tag).to_string(), tag.to_string()); + } + let mut result: Vec = by_display_name.into_values().collect(); + result.sort(); + result +} + +/// Index a pool of tags by display name, mapping each name to one colored +/// variant. +/// +/// When the pool holds several color variants of the same name -- which can +/// happen across pipelines if some client diverged -- the lexicographically +/// smallest one wins, so the choice is deterministic. +fn variant_index(pool: &[String]) -> HashMap { + let mut sorted = pool.to_vec(); + sorted.sort(); + let mut index: HashMap = HashMap::new(); + for tag in sorted { + index + .entry(tag_display_name(&tag).to_string()) + .or_insert(tag); + } + index +} + +/// Choose the colored variant to store for `name`. +/// +/// The CLI cannot express a color, so it borrows one rather than inventing it: +/// any variant of this name known across all pipelines (`known`) wins, and only +/// a name seen nowhere is stored uncolored. This keeps a given name one +/// consistent color everywhere. +fn resolve_variant(name: &str, known: &HashMap) -> String { + known + .get(tag_display_name(name)) + .cloned() + .unwrap_or_else(|| name.to_string()) +} + +/// Resolve `requested` against the colors in use across all pipelines and return +/// the normalized result. +/// +/// Each requested name borrows a color via [`resolve_variant`]: a variant known +/// across all pipelines (`known`), else uncolored. +pub fn set_tags(requested: I, known: &[String]) -> Vec +where + I: IntoIterator, + S: AsRef, +{ + let known = variant_index(known); + normalize_tags( + requested + .into_iter() + .map(|tag| resolve_variant(tag.as_ref(), &known)), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn display_name_strips_color_suffix() { + assert_eq!(tag_display_name("prod"), "prod"); + assert_eq!(tag_display_name("prod|ef4444"), "prod"); + assert_eq!(tag_display_name("team billing|22c55e"), "team billing"); + // Uppercase hex digits are a valid color. + assert_eq!(tag_display_name("prod|ABCDEF"), "prod"); + // A name containing '|' but no valid trailing color is returned whole. + assert_eq!(tag_display_name("a|b"), "a|b"); + assert_eq!(tag_display_name("env|staging"), "env|staging"); + // Wrong-length or malformed suffixes are not colors. + assert_eq!(tag_display_name("prod|fff"), "prod|fff"); + assert_eq!(tag_display_name("prod|gggggg"), "prod|gggggg"); + // Only the final '|' segment is the color; an inner '|' stays in the name. + assert_eq!(tag_display_name("a|b|ef4444"), "a|b"); + assert_eq!(tag_display_name(""), ""); + } + + #[test] + fn sorts_lexicographically() { + assert_eq!( + normalize_tags(["prod", "dev", "staging"]), + vec!["dev", "prod", "staging"] + ); + } + + #[test] + fn is_idempotent() { + let once = normalize_tags(["beta", "alpha", "alpha|ef4444"]); + assert_eq!(normalize_tags(&once), once); + } + + #[test] + fn empty_input() { + assert_eq!(normalize_tags(Vec::::new()), Vec::::new()); + } + + #[test] + fn collapses_color_variants_keeping_last() { + assert_eq!(normalize_tags(["prod", "prod|ef4444"]), vec!["prod|ef4444"]); + assert_eq!(normalize_tags(["prod|ef4444", "prod"]), vec!["prod"]); + } + + #[test] + fn preserves_color_of_survivor() { + assert_eq!( + normalize_tags(["dev", "prod|ef4444", "qa"]), + vec!["dev", "prod|ef4444", "qa"] + ); + } + + #[test] + fn distinct_names_with_colors_all_kept() { + assert_eq!( + normalize_tags(["b|ef4444", "a|22c55e", "c"]), + vec!["a|22c55e", "b|ef4444", "c"] + ); + } + + #[test] + fn split_drops_empty_segments_and_keeps_spaces() { + assert_eq!(split_tag_list(""), Vec::::new()); + assert_eq!(split_tag_list("a,b,c"), vec!["a", "b", "c"]); + // A trailing or doubled comma yields no empty tag. + assert_eq!(split_tag_list("a,,b,"), vec!["a", "b"]); + // A quoted tag arrives as one argument with its space intact. + assert_eq!( + split_tag_list("team billing,prod"), + vec!["team billing", "prod"] + ); + } + + #[test] + fn set_borrows_color_from_known_pool() { + // "prod" is requested by name and colored in the known pool; setting it + // adopts that color, while "dev" is unknown and stored uncolored. + assert_eq!( + set_tags(["prod", "dev"], &["prod|ef4444".to_string()]), + vec!["dev", "prod|ef4444"] + ); + } + + #[test] + fn set_color_conflict_is_deterministic() { + // The known pool colors "prod" two ways; the lexicographically smallest + // variant ("...|22c55e" < "...|ef4444") wins, so the choice is stable. + assert_eq!( + set_tags( + ["prod"], + &["prod|ef4444".to_string(), "prod|22c55e".to_string()] + ), + vec!["prod|22c55e"] + ); + } + + #[test] + fn set_adds_uncolored_name_when_absent() { + assert_eq!(set_tags(["prod"], &[]), vec!["prod"]); + } + + #[test] + fn set_with_no_tags_clears() { + assert_eq!(set_tags(Vec::::new(), &[]), Vec::::new()); + } +} diff --git a/docs.feldera.com/docs/interface/cli.md b/docs.feldera.com/docs/interface/cli.md index 46709d04a1e..53bb5eec13a 100644 --- a/docs.feldera.com/docs/interface/cli.md +++ b/docs.feldera.com/docs/interface/cli.md @@ -261,6 +261,15 @@ Enable storage for `p1`: fda set-config p1 storage true ``` +View and set the tags of `p1`: + +```bash +fda tags p1 # prints the tags as a comma-separated list +fda set-tags p1 prod,team-billing +fda set-tags p1 `fda tags p1`,qa # append to the existing tags +fda set-tags p1 # clear all tags +``` + Add Rust UDF code to `p1`: ```bash