From 321e4e6bfda082c78fe8407bdcde06d66f197f95 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:37:42 +0530 Subject: [PATCH 01/12] fix: install script (#1820) * fix: install script make installer platform aware, shell profile safe * coderabbit comments fix * coderabbit comments fix * add separator newline when missing --- scripts/download.sh | 228 ++++++++++++++++++++++++++++++++------------ 1 file changed, 165 insertions(+), 63 deletions(-) diff --git a/scripts/download.sh b/scripts/download.sh index d22b7403b..86117fc5a 100755 --- a/scripts/download.sh +++ b/scripts/download.sh @@ -1,89 +1,191 @@ -#!/bin/zsh +#!/bin/sh -# supported CPU architectures and operating systems -SUPPORTED_ARCH=("x86_64" "arm64") -SUPPORTED_OS=("linux" "darwin") DOWNLOAD_BASE_URL="https://github.com/parseablehq/parseable/releases/download" -ARM_APPLE_PREFIX="Parseable_OSS_aarch64-apple-darwin" -ARM_LINUX_PREFIX="Parseable_OSS_aarch64-unknown-linux-gnu" -INTEL_LINUX_PREFIX="Parseable_OSS_x86_64-unknown-linux-gnu" -PARSEABLE_PREFIX=${ARM_APPLE_PREFIX} +INSTALL_DIR="${HOME}/.parseable" +BIN_DIR="${INSTALL_DIR}/bin" +BIN_NAME="${BIN_DIR}/parseable" -# Get the system's CPU architecture and operating system +# Get the system's CPU architecture and operating system. CPU_ARCH=$(uname -m) OS=$(uname -s | tr '[:upper:]' '[:lower:]') -printf "\n=========================\n" -printf "Detected CPU architecture: %s\n" "$CPU_ARCH" -printf "Detected operating system: %s\n" "$OS" +printf '\n=========================\n' +printf 'Detected CPU architecture: %s\n' "$CPU_ARCH" +printf 'Detected operating system: %s\n' "$OS" -SHELL_NAME=$(basename $SHELL) -RC_FILE=".${SHELL_NAME}rc" -RC_FILE_PATH="${HOME}/${RC_FILE}" -INSTALL_DIR="${HOME}/.parseable" -BIN_DIR="${INSTALL_DIR}/bin" -BIN_NAME="${BIN_DIR}/parseable" +# Select one downloader and use it for both the release lookup and the binary. +if command -v curl >/dev/null 2>&1; then + DOWNLOADER="curl" +elif command -v wget >/dev/null 2>&1; then + DOWNLOADER="wget" +else + printf 'Error: Neither curl nor wget was found. Please install either curl or wget.\n' >&2 + exit 1 +fi + +# Match the platform names used by the release assets. Linux commonly reports +# ARM64 as "aarch64", while macOS reports it as "arm64". +case "${OS}_${CPU_ARCH}" in + darwin_arm64) + PARSEABLE_PREFIX="Parseable_OSS_aarch64-apple-darwin" + ;; + linux_arm64|linux_aarch64) + PARSEABLE_PREFIX="Parseable_OSS_aarch64-unknown-linux-gnu" + ;; + linux_x86_64|linux_amd64) + PARSEABLE_PREFIX="Parseable_OSS_x86_64-unknown-linux-gnu" + ;; + *) + printf 'Error: unsupported platform %s/%s.\n' "$OS" "$CPU_ARCH" >&2 + exit 1 + ;; +esac + +# Parseable's Linux release binaries require glibc. +if [ "$OS" = "linux" ]; then + MUSL_FOUND="false" + if command -v ldd >/dev/null 2>&1 && ldd --version 2>&1 | grep -qi musl; then + MUSL_FOUND="true" + else + for MUSL_LOADER in /lib/ld-musl-*.so.1; do + if [ -e "$MUSL_LOADER" ]; then + MUSL_FOUND="true" + break + fi + done + fi + if [ "$MUSL_FOUND" = "true" ]; then + printf 'Error: Alpine Linux and other musl-based systems are not supported by the available release binaries.\n' >&2 + exit 1 + fi +fi -# Check if the CPU architecture is supported -if ! echo "${SUPPORTED_ARCH[@]}" | grep -q "\\b${CPU_ARCH}\\b"; then - echo "Error: Unsupported CPU architecture (${CPU_ARCH})." +# Get the latest release tag from the GitHub API. +fetch_release() { + case "$DOWNLOADER" in + curl) + curl -fsSL "https://api.github.com/repos/parseablehq/parseable/releases/latest" + ;; + wget) + wget -q -O - "https://api.github.com/repos/parseablehq/parseable/releases/latest" + ;; + esac +} +if ! release=$(fetch_release); then + printf 'Error: Could not fetch the latest release information.\n' >&2 exit 1 fi -# Check if the OS is supported -if ! echo "${SUPPORTED_OS[@]}" | grep -q "\\b${OS}\\b"; then - echo "Error: Unsupported operating system (${OS})." +release_tag=$(printf '%s\n' "$release" | sed -n 's/.*"tag_name"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' | head -n 1) +if [ -z "$release_tag" ]; then + printf 'Error: Could not determine the latest release version.\n' >&2 exit 1 fi -# Get the latest release information using GitHub API -release=$(curl -s "https://api.github.com/repos/parseablehq/parseable/releases/latest") -# find the release tag -release_tag=$(echo "$release" | grep -o "\"tag_name\":\s*\"[^\"]*\"" | cut -d '"' -f 4) -if [[ -z "$release_tag" ]]; then - echo "Error: Could not determine the latest release version." +printf 'Latest Parseable version: %s\n' "$release_tag" + +download_url="${DOWNLOAD_BASE_URL}/${release_tag}/${PARSEABLE_PREFIX}" + +if [ -d "$INSTALL_DIR" ]; then + printf "A previous version of Parseable already exists. Run 'parseable --version' to check the version,\n" + printf 'or remove %s before installing a new version.\n' "$INSTALL_DIR" exit 1 fi +mkdir -p "$BIN_DIR" || { + printf 'Error: Could not create installation directory %s.\n' "$BIN_DIR" >&2 + rmdir "$BIN_DIR" "$INSTALL_DIR" 2>/dev/null || : + exit 1 +} -printf "Latest Parseable version: $release_tag\n" - -# Determine the appropriate binary prefix based on OS and CPU architecture -declare -A PREFIX_MAP=( - ["darwin_arm64"]=$ARM_APPLE_PREFIX - ["linux_arm64"]=$ARM_LINUX_PREFIX - ["linux_x86_64"]=$INTEL_LINUX_PREFIX -) -key="${OS}_${CPU_ARCH}" -PARSEABLE_PREFIX=${PREFIX_MAP[$key]:-""} || { - echo "Error: unsupported platform $OS/$CPU_ARCH"; exit 1; +TEMP_BIN="${BIN_NAME}.tmp" +# Remove files and empty directories created by an incomplete installation. +cleanup_install() { + rm -f "$TEMP_BIN" "$BIN_NAME" + rmdir "$BIN_DIR" "$INSTALL_DIR" 2>/dev/null || : } +trap 'cleanup_install' 0 +trap 'cleanup_install; exit 1' 1 2 15 -download_url=${DOWNLOAD_BASE_URL}/${release_tag}/${PARSEABLE_PREFIX} +printf 'Downloading Parseable version %s, for OS: %s, CPU architecture: %s\n' "$release_tag" "$OS" "$CPU_ARCH" +printf 'Download URL: %s\n\n' "$download_url" -if [[ -d ${INSTALL_DIR} ]]; then - printf "A Previous version of parseable already exists. Run 'parseable --version' to check the version." - printf "or consider removing that before new installation\n" +# Download the selected release asset to the temporary installation path. +download_binary() { + case "$DOWNLOADER" in + curl) + curl -fL -o "$TEMP_BIN" "$download_url" + ;; + wget) + wget -q -O "$TEMP_BIN" "$download_url" + ;; + esac +} +if ! download_binary; then + printf 'Error: download failed.\n' >&2 exit 1 -else - mkdir -p ${BIN_DIR} fi -# Download the binary using curl or wget -printf "Downloading Parseable version $release_tag, for OS: $OS, CPU architecture: $CPU_ARCH\n" -printf "Download URL: $download_url\n\n" - -if command -v curl &>/dev/null; then - curl -fL -o "${BIN_NAME}" "$download_url" || { echo "Error: download failed"; exit 1; } -elif command -v wget &>/dev/null; then - wget -q -O "${BIN_NAME}" "$download_url" || { echo "Error: download failed"; exit 1; } -else - echo "Error: Neither curl nor wget found. Please install either curl or wget." +chmod +x "$TEMP_BIN" || { + printf 'Error: Could not make the downloaded binary executable.\n' >&2 exit 1 -fi +} +mv "$TEMP_BIN" "$BIN_NAME" || { + printf 'Error: Could not finish installing %s.\n' "$BIN_NAME" >&2 + exit 1 +} +trap - 0 1 2 15 -printf "Parseable Server was successfully installed at: ${BIN_NAME}\n" +printf 'Parseable Server was successfully installed at: %s\n' "$BIN_NAME" -chmod +x "${BIN_NAME}" +# Use the user's login shell to select its startup file. Do not source that file +# here: this installer may be running under a different shell via `... | sh`. +SHELL_NAME=${SHELL##*/} +case "$SHELL_NAME" in + zsh) + RC_FILE_PATH="${ZDOTDIR:-$HOME}/.zshrc" + ;; + bash) + RC_FILE_PATH="${HOME}/.bash_profile" + for BASH_PROFILE in .bash_profile .bash_login .profile; do + if [ -f "${HOME}/${BASH_PROFILE}" ]; then + RC_FILE_PATH="${HOME}/${BASH_PROFILE}" + break + fi + done + ;; + sh|dash|ksh) + RC_FILE_PATH="${HOME}/.profile" + ;; + *) + RC_FILE_PATH="" + ;; +esac -PATH_STR="export PATH=${BIN_DIR}"':$PATH' -echo ${PATH_STR} >> ${RC_FILE_PATH} -source ${RC_FILE_PATH} +PATH_STR="export PATH=\"${BIN_DIR}:\$PATH\"" +# Add the PATH export to a shell profile unless it is already present. +append_path() { + PROFILE_PATH=$1 + if grep -Fqx "$PATH_STR" "$PROFILE_PATH" 2>/dev/null; then + printf 'Parseable is already configured in %s.\n' "$PROFILE_PATH" + elif { + if [ -s "$PROFILE_PATH" ] && ! tail -c 1 "$PROFILE_PATH" | grep -q '^$'; then + printf '\n' + fi + printf '%s\n' "$PATH_STR" + } >> "$PROFILE_PATH"; then + printf 'Added Parseable to PATH in %s.\n' "$PROFILE_PATH" + else + printf 'Warning: Could not update %s.\n' "$PROFILE_PATH" >&2 + fi +} + +if [ "$SHELL_NAME" = "bash" ]; then + append_path "${HOME}/.bashrc" + if [ "$RC_FILE_PATH" != "${HOME}/.bashrc" ]; then + append_path "$RC_FILE_PATH" + fi +elif [ -n "$RC_FILE_PATH" ]; then + append_path "$RC_FILE_PATH" +else + printf 'No supported shell profile was detected.\n' +fi +printf 'To use Parseable in the current shell, run:\n %s\n' "$PATH_STR" From e3e4c8c5aab8f2df7887ef015942edfeb59c2f5d Mon Sep 17 00:00:00 2001 From: parmesant Date: Mon, 12 Oct 2026 22:47:48 +0530 Subject: [PATCH 02/12] Misc updates (#1807) * Misc updates - added query param `?format=compact` to get a compact schema (defaults to JSON) - if severity text/number is missing then the value is derived using the other field if present * remove nullable from compact schema --- src/handlers/http/logstream.rs | 322 ++++++++++++++++++++++++++++++++- src/otel/logs.rs | 128 +++++++++++-- 2 files changed, 428 insertions(+), 22 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index 46f7df355..fdfe0f84a 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -37,13 +37,15 @@ use crate::utils::json::flatten::{ }; use crate::{LOCK_EXPECT, stats, validator}; -use actix_web::http::StatusCode; -use actix_web::web::{Json, Path}; -use actix_web::{HttpRequest, Responder, web}; +use actix_web::http::{StatusCode, header::ContentType}; +use actix_web::web::{Json, Path, Query}; +use actix_web::{HttpRequest, HttpResponse, Responder, web}; use arrow_json::reader::infer_json_schema_from_iterator; +use arrow_schema::{DataType, Field, Schema, UnionFields}; use bytes::Bytes; use chrono::Utc; use itertools::Itertools; +use serde::Deserialize; use serde_json::{Value, json}; use std::fs; use std::sync::Arc; @@ -171,14 +173,126 @@ pub async fn detect_schema(Json(json): Json) -> Result HttpResponse { + match format { + SchemaFormat::Json => HttpResponse::Ok().json(schema), + SchemaFormat::Compact => HttpResponse::Ok() + .insert_header(ContentType::plaintext()) + .body(compact_schema(schema)), + } +} + +fn compact_schema(schema: &Schema) -> String { + let mut output = String::from("name\ttype\n"); + for field in schema.fields() { + append_escaped_compact_field(&mut output, field.name()); + output.push('\t'); + append_escaped_compact_field( + &mut output, + &compact_data_type(field.data_type()).to_string(), + ); + output.push('\n'); + } + output +} + +/// Clone nested fields without metadata before using Arrow's type display, preserving type structure. +fn compact_data_type(data_type: &DataType) -> DataType { + match data_type { + DataType::List(field) => DataType::List(compact_field(field)), + DataType::ListView(field) => DataType::ListView(compact_field(field)), + DataType::FixedSizeList(field, size) => { + DataType::FixedSizeList(compact_field(field), *size) + } + DataType::LargeList(field) => DataType::LargeList(compact_field(field)), + DataType::LargeListView(field) => DataType::LargeListView(compact_field(field)), + DataType::Struct(fields) => DataType::Struct( + fields + .iter() + .map(|field| compact_field(field)) + .collect::>() + .into(), + ), + DataType::Union(fields, mode) => DataType::Union( + UnionFields::try_new( + fields.iter().map(|(type_id, _)| type_id), + fields.iter().map(|(_, field)| compact_field(field)), + ) + .expect("existing union fields are valid"), + *mode, + ), + DataType::Dictionary(key, value) => DataType::Dictionary( + Box::new(compact_data_type(key)), + Box::new(compact_data_type(value)), + ), + DataType::Map(field, sorted) => DataType::Map(compact_field(field), *sorted), + DataType::RunEndEncoded(run_ends, values) => { + DataType::RunEndEncoded(compact_field(run_ends), compact_field(values)) + } + _ => data_type.clone(), + } +} + +fn compact_field(field: &Field) -> Arc { + Arc::new(Field::new( + field.name(), + compact_data_type(field.data_type()), + field.is_nullable(), + )) +} + +fn append_escaped_compact_field(output: &mut String, value: &str) { + if !value.contains(['\\', '\t', '\r', '\n']) { + output.push_str(value); + return; + } + + for character in value.chars() { + match character { + '\\' => output.push_str("\\\\"), + '\t' => output.push_str("\\t"), + '\r' => output.push_str("\\r"), + '\n' => output.push_str("\\n"), + _ => output.push(character), + } + } +} + +/// Fetch the full Arrow JSON schema, or request `?format=compact` for a smaller TSV response. +/// Compact output includes all column names, type displays, but omits +/// schema/field metadata, top-level nullability and dictionary IPC attributes; it is not an Arrow interchange format. pub async fn get_schema( req: HttpRequest, logstream: Path, + query: Query, ) -> Result { let stream_name = logstream.into_inner(); let tenant_id = get_tenant_id_from_request(&req); let schema = get_schema_internal(&stream_name, &tenant_id).await?; - Ok((web::Json(schema), StatusCode::OK)) + Ok(schema_response(&schema, query.format)) } /// Shared schema implementation for HTTP handlers and in-process callers. @@ -196,7 +310,10 @@ pub async fn get_schema_internal( return Err(StreamNotFound(stream_name.to_owned()).into()); } match update_schema_when_distributed(&vec![stream_name.to_owned()], tenant_id).await { - Ok(_) => Ok(stream.get_schema()), + Ok(_) => { + let schema = stream.get_schema(); + Ok(schema) + } Err(err) => Err(StreamError::Custom { msg: err.to_string(), status: StatusCode::EXPECTATION_FAILED, @@ -769,10 +886,203 @@ pub mod error { #[cfg(test)] mod tests { + use super::{SchemaQuery, compact_schema, schema_response}; use crate::{ event::format::LogSource, handlers::http::modal::utils::logstream_utils::PutStreamHeaders, }; - use actix_web::test::TestRequest; + use actix_web::{ + App, HttpResponse, + http::{StatusCode, header}, + test, + test::TestRequest, + web::{self, Query}, + }; + use arrow_schema::{DataType, Field, Schema, TimeUnit}; + use std::collections::HashMap; + use std::sync::Arc; + + async fn schema_test_handler( + query: Query, + schema: web::Data, + ) -> HttpResponse { + schema_response(schema.get_ref(), query.into_inner().format) + } + + fn test_schema() -> Schema { + Schema::new_with_metadata( + vec![ + Arc::new(Field::new( + "nested", + DataType::Struct( + vec![Arc::new( + Field::new("inner", DataType::Int64, false).with_metadata( + HashMap::from([( + String::from("inner-metadata"), + String::from("omitted"), + )]), + ), + )] + .into(), + ), + true, + )), + Arc::new(Field::new_list( + "list", + Arc::new(Field::new_list_field(DataType::Utf8, false)), + false, + )), + Arc::new(Field::new_dictionary( + "dictionary", + DataType::Int32, + DataType::Utf8, + true, + )), + Arc::new(Field::new( + "dictionary_struct", + DataType::Dictionary( + Box::new(DataType::Int32), + Box::new(DataType::Struct( + vec![Arc::new( + Field::new("dictionary_inner", DataType::Utf8, true).with_metadata( + HashMap::from([( + String::from("dictionary-inner-metadata"), + String::from("omitted"), + )]), + ), + )] + .into(), + )), + ), + false, + )), + Arc::new(Field::new( + "timestamp", + DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into())), + false, + )), + Arc::new(Field::new("awk\\ward\tname\r\n", DataType::Utf8, true)), + ], + HashMap::from([(String::from("schema-metadata"), String::from("omitted"))]), + ) + } + + #[actix_web::test] + async fn schema_format_query_selects_byte_compatible_json_or_compact_tsv() { + let schema = test_schema(); + let expected_json = serde_json::to_vec(&schema).unwrap(); + let app = test::init_service( + App::new() + .app_data(web::Data::new(schema)) + .route("/schema", web::get().to(schema_test_handler)), + ) + .await; + + let default_response = + test::call_service(&app, TestRequest::get().uri("/schema").to_request()).await; + assert_eq!(default_response.status(), StatusCode::OK); + assert_eq!( + default_response + .headers() + .get(header::CONTENT_TYPE) + .unwrap(), + "application/json" + ); + assert_eq!(test::read_body(default_response).await, expected_json); + + let json_response = test::call_service( + &app, + TestRequest::get().uri("/schema?format=json").to_request(), + ) + .await; + assert_eq!(json_response.status(), StatusCode::OK); + assert_eq!( + json_response.headers().get(header::CONTENT_TYPE).unwrap(), + "application/json" + ); + assert_eq!(test::read_body(json_response).await, expected_json); + + let compact_response = test::call_service( + &app, + TestRequest::get() + .uri("/schema?format=compact") + .to_request(), + ) + .await; + assert_eq!(compact_response.status(), StatusCode::OK); + assert_eq!( + compact_response + .headers() + .get(header::CONTENT_TYPE) + .unwrap(), + "text/plain; charset=utf-8" + ); + let compact = std::str::from_utf8(&test::read_body(compact_response).await) + .unwrap() + .to_owned(); + assert_eq!( + compact, + "name\ttype\n\ + nested\tStruct(\"inner\": non-null Int64)\n\ + list\tList(non-null Utf8)\n\ + dictionary\tDictionary(Int32, Utf8)\n\ + dictionary_struct\tDictionary(Int32, Struct(\"dictionary_inner\": Utf8))\n\ + timestamp\tTimestamp(ns, \"UTC\")\n\ + awk\\\\ward\\tname\\r\\n\tUtf8\n" + ); + assert!(!compact.contains("schema-metadata")); + assert!(!compact.contains("inner-metadata")); + assert!(!compact.contains("dictionary-inner-metadata")); + } + + #[actix_web::test] + async fn schema_format_query_rejects_invalid_format() { + let app = test::init_service( + App::new() + .app_data(web::Data::new(Schema::empty())) + .route("/schema", web::get().to(schema_test_handler)), + ) + .await; + + let response = test::call_service( + &app, + TestRequest::get().uri("/schema?format=xml").to_request(), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + } + + #[core::prelude::v1::test] + fn compact_schema_handles_empty_schemas_and_reduces_500_column_payloads() { + assert_eq!(compact_schema(&Schema::empty()), "name\ttype\n"); + + let schema = Schema::new( + (0..500) + .map(|index| { + Arc::new(Field::new( + format!("field_{index:03}"), + DataType::Utf8, + index % 2 == 0, + )) + }) + .collect::>(), + ); + let compact_len = compact_schema(&schema).len(); + let arrow_json_len = serde_json::to_vec(&schema).unwrap().len(); + println!("compact schema: {compact_len} bytes vs Arrow JSON: {arrow_json_len} bytes"); + assert_eq!(compact_schema(&schema).lines().count(), 501); + assert!(compact_len * 4 < arrow_json_len); + } + + #[core::prelude::v1::test] + fn compact_schema_uses_arrow_display_for_complex_types() { + let schema = test_schema(); + let compact = compact_schema(&schema); + assert!(compact.contains("Struct(\"inner\": non-null Int64)")); + assert!(compact.contains("List(non-null Utf8)")); + assert!(compact.contains("Dictionary(Int32, Utf8)")); + assert!(compact.contains("Dictionary(Int32, Struct(\"dictionary_inner\": Utf8))")); + assert!(compact.contains("Timestamp(ns, \"UTC\")")); + } // TODO: Fix this test with routes // #[actix_web::test] diff --git a/src/otel/logs.rs b/src/otel/logs.rs index f6d770fad..a7cbc0b42 100644 --- a/src/otel/logs.rs +++ b/src/otel/logs.rs @@ -47,26 +47,80 @@ pub const OTEL_LOG_KNOWN_FIELD_LIST: [&str; 17] = [ "event_name", "p_log_category", ]; -/// otel log event has severity number -/// there is a mapping of severity number to severity text provided in proto -/// this function fetches the severity text from the severity number -/// and adds it to the flattened json -fn flatten_severity(severity_number: i32) -> Map { +/// Derive whichever OTel severity field is absent from the other. +fn normalize_severity(severity_number: i32, severity_text: &str) -> (i32, String) { + let severity_text_is_unspecified = + severity_text.trim().is_empty() || severity_text.trim().eq_ignore_ascii_case("UNSPECIFIED"); + let severity_number = if severity_number == 0 { + severity_number_from_text(severity_text).unwrap_or(severity_number) + } else { + severity_number + }; + + let severity_text = if severity_text_is_unspecified { + SeverityNumber::try_from(severity_number) + .ok() + .map(|severity| { + let name = severity.as_str_name(); + name.strip_prefix("SEVERITY_NUMBER_") + .unwrap_or(name) + .to_string() + }) + .unwrap_or_else(|| "UNSPECIFIED".to_string()) + } else { + severity_text.to_string() + }; + + (severity_number, severity_text) +} + +/// Convert standard OTel severity names (and the common WARNING alias) to their +/// lowest severity number in the corresponding range. +fn severity_number_from_text(severity_text: &str) -> Option { + let severity_text = severity_text.trim().to_ascii_uppercase(); + let severity_text = severity_text + .strip_prefix("SEVERITY_NUMBER_") + .unwrap_or(&severity_text); + + match severity_text { + "TRACE" => Some(1), + "TRACE2" => Some(2), + "TRACE3" => Some(3), + "TRACE4" => Some(4), + "DEBUG" => Some(5), + "DEBUG2" => Some(6), + "DEBUG3" => Some(7), + "DEBUG4" => Some(8), + "INFO" => Some(9), + "INFO2" => Some(10), + "INFO3" => Some(11), + "INFO4" => Some(12), + "WARN" | "WARNING" => Some(13), + "WARN2" => Some(14), + "WARN3" => Some(15), + "WARN4" => Some(16), + "ERROR" => Some(17), + "ERROR2" => Some(18), + "ERROR3" => Some(19), + "ERROR4" => Some(20), + "FATAL" => Some(21), + "FATAL2" => Some(22), + "FATAL3" => Some(23), + "FATAL4" => Some(24), + _ => None, + } +} + +/// Add the normalized severity fields to the flattened JSON. +fn flatten_severity(severity_number: i32, severity_text: &str) -> Map { let mut severity_json: Map = Map::new(); severity_json.insert( "severity_number".to_string(), Value::Number(severity_number.into()), ); - let severity = SeverityNumber::try_from(severity_number).unwrap(); - let severity_text = severity.as_str_name().to_string(); severity_json.insert( "severity_text".to_string(), - Value::String( - severity_text - .strip_prefix("SEVERITY_NUMBER_") - .unwrap_or(&severity_text) - .to_string(), - ), + Value::String(severity_text.to_string()), ); severity_json } @@ -131,10 +185,12 @@ pub fn flatten_log_record(log_record: &LogRecord) -> Map { )), ); - log_record_json.extend(flatten_severity(log_record.severity_number)); + let (severity_number, severity_text) = + normalize_severity(log_record.severity_number, &log_record.severity_text); + log_record_json.extend(flatten_severity(severity_number, &severity_text)); - // Primary: derive category from severity_number - let mut log_category = category_from_severity(log_record.severity_number); + // Primary: derive category from the normalized severity_number + let mut log_category = category_from_severity(severity_number); if log_record.body.is_some() { let body = &log_record.body; @@ -304,3 +360,43 @@ pub fn flatten_otel_logs(message: &LogsData, tenant_id: &str) -> Vec { tenant_id, ) } + +#[cfg(test)] +mod tests { + use super::normalize_severity; + + #[test] + fn derives_severity_text_from_number_when_missing() { + assert_eq!(normalize_severity(17, ""), (17, "ERROR".to_string())); + assert_eq!( + normalize_severity(17, "UNSPECIFIED"), + (17, "ERROR".to_string()) + ); + } + + #[test] + fn derives_severity_number_from_text_when_unspecified() { + assert_eq!( + normalize_severity(0, "warning"), + (13, "warning".to_string()) + ); + assert_eq!( + normalize_severity(0, "unspecified"), + (0, "UNSPECIFIED".to_string()) + ); + assert_eq!( + normalize_severity(0, "SEVERITY_NUMBER_FATAL4"), + (24, "SEVERITY_NUMBER_FATAL4".to_string()) + ); + } + + #[test] + fn leaves_both_fields_unspecified_when_neither_is_present() { + assert_eq!(normalize_severity(0, ""), (0, "UNSPECIFIED".to_string())); + } + + #[test] + fn preserves_provided_severity_fields() { + assert_eq!(normalize_severity(18, "error"), (18, "error".to_string())); + } +} From c7dfbefee00bff39eea2344a8e47aa4429d8e68a Mon Sep 17 00:00:00 2001 From: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Date: Tue, 13 Oct 2026 13:10:27 +0530 Subject: [PATCH 03/12] remove pmeta creation (#1822) --- src/handlers/http/modal/query_server.rs | 3 - src/parseable/mod.rs | 76 +------------------------ 2 files changed, 2 insertions(+), 77 deletions(-) diff --git a/src/handlers/http/modal/query_server.rs b/src/handlers/http/modal/query_server.rs index dd2208f79..f4808dbe2 100644 --- a/src/handlers/http/modal/query_server.rs +++ b/src/handlers/http/modal/query_server.rs @@ -120,9 +120,6 @@ impl ParseableServer for QueryServer { .await; migration::run_migration(&PARSEABLE).await?; - // create internal stream at server start - // Multi-tenant mode is not allowed in OSS hence no need to have knowledge of tenant id - PARSEABLE.create_internal_stream_if_not_exists().await?; // load on init load_on_init().await?; // track all parquet files already in the data directory diff --git a/src/parseable/mod.rs b/src/parseable/mod.rs index 1db2bfe85..cd127adbf 100644 --- a/src/parseable/mod.rs +++ b/src/parseable/mod.rs @@ -21,14 +21,10 @@ use std::{ collections::{HashMap, HashSet}, num::NonZeroU32, path::PathBuf, - str::FromStr, sync::{Arc, LazyLock, RwLock}, }; -use actix_web::http::{ - StatusCode, - header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue}, -}; +use actix_web::http::{StatusCode, header::HeaderMap}; use arrow_schema::{Field, Schema}; use bytes::Bytes; use chrono::Utc; @@ -53,12 +49,10 @@ use crate::{ format::{LogSource, LogSourceEntry}, }, handlers::{ - DatasetTag, STREAM_TYPE_KEY, TelemetryType, + DatasetTag, TelemetryType, http::{ - cluster::{PMETA_STREAM_NAME, sync_streams_with_ingestors}, ingest::PostError, logstream::error::{CreateStreamError, StreamError}, - middleware::{CLUSTER_SECRET, CLUSTER_SECRET_HEADER}, modal::{ ingest_server::INGESTOR_META, utils::{logstream_utils::PutStreamHeaders, rbac_utils::get_metadata}, @@ -570,72 +564,6 @@ impl Parseable { Ok(true) } - pub async fn create_internal_stream_if_not_exists(&self) -> Result<(), StreamError> { - let log_source_entry = LogSourceEntry::new(LogSource::Pmeta, HashSet::new()); - let tenants = if let Some(tenants) = PARSEABLE.list_tenants() { - tenants.into_iter().map(Some).collect() - } else { - vec![None] - }; - for tenant_id in tenants { - let internal_stream_result = self - .create_stream_if_not_exists( - PMETA_STREAM_NAME, - StreamType::Internal, - None, - vec![log_source_entry.clone()], - TelemetryType::Logs, - &tenant_id, - vec![], - vec![], - ) - .await; - - // Check if either stream creation failed - if let Err(e) = &internal_stream_result { - tracing::error!("Failed to create pmeta stream: {:?}", e); - } - - // Check if both streams already existed - if matches!(internal_stream_result, Ok(true)) { - continue; - } - - let mut header_map = HeaderMap::new(); - header_map.insert( - HeaderName::from_str(STREAM_TYPE_KEY).unwrap(), - HeaderValue::from_str(&StreamType::Internal.to_string()).unwrap(), - ); - header_map.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - - if let Some((_, hash)) = CLUSTER_SECRET.get() { - header_map.insert( - HeaderName::from_static(CLUSTER_SECRET_HEADER), - HeaderValue::from_str(hash).unwrap(), - ); - header_map.insert( - HeaderName::from_static("intra-cluster-tenant"), - HeaderValue::from_str(tenant_id.as_deref().unwrap_or(DEFAULT_TENANT)).unwrap(), - ); - } - - // Sync only the streams that were created successfully - if matches!(internal_stream_result, Ok(false)) - && let Err(e) = sync_streams_with_ingestors( - header_map.clone(), - Bytes::new(), - PMETA_STREAM_NAME, - &tenant_id, - ) - .await - { - tracing::error!("Failed to sync pmeta stream with ingestors: {:?}", e); - } - } - - Ok(()) - } - // Check if the stream exists and create a new stream if doesn't exist #[allow(clippy::too_many_arguments)] pub async fn create_stream_if_not_exists( From c94bd14e3065f047b6f9538854f258ea1d5c67a9 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Date: Tue, 13 Oct 2026 18:21:39 +0530 Subject: [PATCH 04/12] add log, trace errors as tools (#1823) * add log, trace errors as tools * deesource error fix * get_dataset_schema tool to return compact TSV response --- src/handlers/http/logstream.rs | 4 +- src/tool_catalog.rs | 72 ++++++++++++- src/tool_executor.rs | 189 +++++++++++++++++++++++++++++++-- 3 files changed, 250 insertions(+), 15 deletions(-) diff --git a/src/handlers/http/logstream.rs b/src/handlers/http/logstream.rs index fdfe0f84a..82c58bd70 100644 --- a/src/handlers/http/logstream.rs +++ b/src/handlers/http/logstream.rs @@ -205,7 +205,9 @@ fn schema_response(schema: &Schema, format: SchemaFormat) -> HttpResponse { } } -fn compact_schema(schema: &Schema) -> String { +/// Render the token-efficient schema representation used by the compact HTTP +/// response and by agent-facing schema tools. +pub fn compact_schema(schema: &Schema) -> String { let mut output = String::from("name\ttype\n"); for field in schema.fields() { append_escaped_compact_field(&mut output, field.name()); diff --git a/src/tool_catalog.rs b/src/tool_catalog.rs index 196893fb5..70acf79f3 100644 --- a/src/tool_catalog.rs +++ b/src/tool_catalog.rs @@ -104,7 +104,9 @@ impl ToolSpec { | "explain_query" | "discover_datasets" | "get_traces" - | "get_trace" => ToolEffect::ExpensiveRead, + | "get_trace" + | "list_log_errors" + | "list_trace_errors" => ToolEffect::ExpensiveRead, "create_alert" | "enable_alert" | "disable_alert" @@ -354,7 +356,7 @@ pub fn oss_read_tool_specs() -> Vec { ToolSpec::new( "get_dataset_schema", "Get dataset schema", - "Get columns and data types for a dataset.", + "Get all dataset columns and data types as compact tab-separated text with a name/type header.", dataset_schema(), ), ToolSpec::new( @@ -402,6 +404,34 @@ pub fn oss_read_tool_specs() -> Vec { &["dataset", "pTimestamp"], ), ), + ToolSpec::new( + "list_log_errors", + "List log errors", + "List grouped error, fatal, and exception log records from one logs dataset in a bounded time range.", + object_schema( + json!({ + "dataset": string_property("One logs dataset."), + "startTime": string_property("RFC3339 range start."), + "endTime": string_property("RFC3339 range end."), + "limit": { "type": "integer", "minimum": 1, "maximum": 500, "default": 50 } + }), + &["dataset", "startTime", "endTime"], + ), + ), + ToolSpec::new( + "list_trace_errors", + "List trace errors", + "List failed and error spans from one traces dataset in a bounded time range.", + object_schema( + json!({ + "dataset": string_property("One traces dataset."), + "startTime": string_property("RFC3339 range start."), + "endTime": string_property("RFC3339 range end."), + "limit": { "type": "integer", "minimum": 1, "maximum": 500, "default": 50 } + }), + &["dataset", "startTime", "endTime"], + ), + ), ToolSpec::new( "query_sql", "Run SQL query", @@ -824,7 +854,7 @@ mod tests { let tools = oss_tool_specs(); let names = tools.iter().map(|tool| tool.name).collect::>(); - assert_eq!(tools.len(), 47); + assert_eq!(tools.len(), 49); assert_eq!(tools.len(), names.len()); } @@ -863,7 +893,7 @@ mod tests { let read_tools = oss_read_tool_specs(); let action_tools = oss_action_tool_specs(); - assert_eq!(read_tools.len(), 34); + assert_eq!(read_tools.len(), 36); assert_eq!(action_tools.len(), 13); assert!(read_tools.iter().all(|tool| matches!( tool.effect(), @@ -913,6 +943,40 @@ mod tests { ); } + #[actix_web::test] + async fn oss_catalog_exposes_list_log_errors() { + let tool = oss_read_tool_specs() + .into_iter() + .find(|tool| tool.name == "list_log_errors") + .expect("list_log_errors must be available in OSS"); + + assert_eq!(tool.effect(), ToolEffect::ExpensiveRead); + assert_eq!(tool.input_schema["properties"]["dataset"]["type"], "string"); + assert!( + tool.input_schema["required"] + .as_array() + .unwrap() + .contains(&serde_json::json!("dataset")) + ); + } + + #[actix_web::test] + async fn oss_catalog_exposes_list_trace_errors() { + let tool = oss_read_tool_specs() + .into_iter() + .find(|tool| tool.name == "list_trace_errors") + .expect("list_trace_errors must be available in OSS"); + + assert_eq!(tool.effect(), ToolEffect::ExpensiveRead); + assert_eq!(tool.input_schema["properties"]["dataset"]["type"], "string"); + assert!( + tool.input_schema["required"] + .as_array() + .unwrap() + .contains(&serde_json::json!("dataset")) + ); + } + #[actix_web::test] async fn tool_call_defaults_to_empty_arguments() { let call: ToolCallRequest = serde_json::from_value(serde_json::json!({ diff --git a/src/tool_executor.rs b/src/tool_executor.rs index a68bf88a5..c0eaeaf6d 100644 --- a/src/tool_executor.rs +++ b/src/tool_executor.rs @@ -272,12 +272,10 @@ async fn execute_dataset_metadata_tool( match name { "get_dataset_schema" => { authorize(context.session_key, Action::GetSchema, Some(dataset), None)?; - serde_json::to_value( - logstream::get_schema_internal(dataset, context.tenant_id) - .await - .map_err(|error| error.to_string())?, - ) - .map_err(|error| error.to_string()) + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + Ok(Value::String(logstream::compact_schema(&schema))) } "get_dataset_info" => { authorize( @@ -409,6 +407,152 @@ async fn get_log_context(arguments: &Value, context: OssToolContext<'_>) -> Resu response_value(response).await } +fn first_present_field(fields: &[String], candidates: &[&str]) -> Option { + candidates + .iter() + .find(|candidate| fields.iter().any(|field| field == **candidate)) + .map(|field| (*field).to_owned()) +} + +pub fn list_log_errors_query( + dataset: &str, + fields: &[String], + limit: u64, +) -> Result { + let service = first_present_field(fields, &["service.name", "service_name"]).map_or_else( + || "'unknown'".to_owned(), + |field| { + format!( + "COALESCE(CAST({} AS VARCHAR), 'unknown')", + quote_identifier(&field) + ) + }, + ); + let severity_text = first_present_field(fields, &["severity_text", "severity_description"]); + let exception_type = first_present_field( + fields, + &["event_exception.type", "exception.type", "error.type"], + ); + let error_type = exception_type + .as_ref() + .or(severity_text.as_ref()) + .map_or_else( + || "'error'".to_owned(), + |field| { + format!( + "COALESCE(CAST({} AS VARCHAR), 'error')", + quote_identifier(field) + ) + }, + ); + let message = first_present_field( + fields, + &[ + "event_exception.message", + "exception.message", + "error.message", + "body", + "message", + "log", + ], + ); + let sample_message = message.as_ref().map_or_else( + || "CAST(NULL AS VARCHAR)".to_owned(), + |field| format!("MAX(CAST({} AS VARCHAR))", quote_identifier(field)), + ); + let last_seen = first_present_field(fields, &["p_timestamp", "time_unix_nano"]).map_or_else( + || "CAST(NULL AS VARCHAR)".to_owned(), + |field| format!("MAX({})", quote_identifier(&field)), + ); + + let mut predicates = Vec::new(); + if fields.iter().any(|field| field == "severity_number") { + predicates.push("CAST(\"severity_number\" AS BIGINT) >= 17".to_owned()); + } + if let Some(field) = &severity_text { + predicates.push(format!( + "UPPER(CAST({} AS VARCHAR)) IN ('ERROR', 'FATAL', 'CRITICAL')", + quote_identifier(field) + )); + } + if let Some(field) = &exception_type { + predicates.push(format!( + "{} IS NOT NULL AND CAST({} AS VARCHAR) <> ''", + quote_identifier(field), + quote_identifier(field) + )); + } + if predicates.is_empty() + && let Some(field) = &message + { + let field = quote_identifier(field); + predicates.push(format!( + "(LOWER(CAST({field} AS VARCHAR)) LIKE '%error%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%exception%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%failed%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%fatal%' OR LOWER(CAST({field} AS VARCHAR)) LIKE '%panic%')" + )); + } + if predicates.is_empty() { + return Err( + "log dataset has no supported severity, exception, or message field".to_owned(), + ); + } + + Ok(format!( + "SELECT {service} AS \"serviceName\", {error_type} AS \"errorType\", {sample_message} AS \"sampleMessage\", COUNT(*) AS \"occurrences\", {last_seen} AS \"lastSeen\" FROM {} WHERE ({}) GROUP BY {service}, {error_type} ORDER BY \"occurrences\" DESC LIMIT {}", + quote_identifier(dataset), + predicates.join(" OR "), + limit.clamp(1, 500), + )) +} + +async fn list_log_errors(arguments: &Value, context: OssToolContext<'_>) -> Result { + let dataset = required_string(arguments, "dataset")?; + authorize(context.session_key, Action::Query, Some(dataset), None)?; + let info = serde_json::to_value( + get_prism_logstream_info(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?, + ) + .map_err(|error| error.to_string())?; + if !nested_string(&info, "telemetryType").is_some_and(|value| value == "logs") { + return Err(format!( + "tool requires a logs dataset, but `{dataset}` is not logs telemetry" + )); + } + let schema = logstream::get_schema_internal(dataset, context.tenant_id) + .await + .map_err(|error| error.to_string())?; + let fields = schema + .fields() + .iter() + .map(|field| field.name().to_owned()) + .collect::>(); + let limit = bounded_integer(arguments, "limit", 50, 1, 500); + let query_text = bounded_select(&list_log_errors_query(dataset, &fields, limit)?)?; + let response = query::query_internal( + query::Query { + query: query_text, + start_time: required_string(arguments, "startTime")?.to_owned(), + end_time: required_string(arguments, "endTime")?.to_owned(), + send_null: true, + fields: false, + streaming: false, + filter_tags: None, + }, + context.session_key, + context.tenant_id, + ) + .await + .map_err(|error| error.to_string())?; + let records = response_value(response).await?; + let total = records.as_array().map_or(0, Vec::len); + Ok(json!({ + "total": total, + "offset": 0, + "limit": limit, + "records": records, + })) +} + async fn execute_sql_tool( name: &str, arguments: &Value, @@ -563,7 +707,8 @@ async fn execute_trace_tool( let datasets = [required_string(arguments, "dataset")?.to_owned()]; let start_time = required_string(arguments, "startTime")?.to_owned(); let end_time = required_string(arguments, "endTime")?.to_owned(); - let limit = bounded_integer(arguments, "limit", 100, 1, 500) as usize; + let default_limit = if name == "list_trace_errors" { 50 } else { 100 }; + let limit = bounded_integer(arguments, "limit", default_limit, 1, 500) as usize; let mut results = Vec::new(); let mut errors = Vec::new(); for dataset in datasets { @@ -596,7 +741,8 @@ async fn execute_trace_tool( end_time: end_time.clone(), sort_by: None, conditions: None, - options: None, + options: (name == "list_trace_errors") + .then_some(crate::handlers::http::traces::TraceListOption::ErrorSpans), limit: Some(limit), offset: Some(0), }, @@ -818,13 +964,16 @@ async fn execute_oss_tool_inner( } "sample_events" => sample_events(arguments, context).await, "get_log_context" => get_log_context(arguments, context).await, + "list_log_errors" => list_log_errors(arguments, context).await, "query_sql" | "explain_query" => execute_sql_tool(name, arguments, context).await, "list_alerts" | "get_alert" | "list_alert_tags" | "enable_alert" | "disable_alert" | "evaluate_alert" | "create_alert" => execute_alert_tool(name, arguments, context).await, "list_alert_targets" | "get_alert_target" | "create_alert_target" => { execute_alert_target_tool(name, arguments, context).await } - "get_traces" | "get_trace" => execute_trace_tool(name, arguments, context).await, + "get_traces" | "get_trace" | "list_trace_errors" => { + execute_trace_tool(name, arguments, context).await + } "ping" | "get_cluster_status" | "get_cluster_metrics" @@ -894,7 +1043,27 @@ mod tests { use chrono::{DateTime, Duration, Utc}; use serde_json::{Value, json}; - use super::{call_tool_registry, normalized_log_context_bounds}; + use super::{call_tool_registry, list_log_errors_query, normalized_log_context_bounds}; + + #[actix_web::test] + async fn log_error_query_uses_otel_severity_fields() { + let query = list_log_errors_query( + "app-logs", + &[ + "p_timestamp".to_owned(), + "service.name".to_owned(), + "severity_number".to_owned(), + "severity_text".to_owned(), + "body".to_owned(), + ], + 5, + ) + .unwrap(); + + assert!(query.contains("CAST(\"severity_number\" AS BIGINT) >= 17")); + assert!(query.contains("'ERROR', 'FATAL', 'CRITICAL'")); + assert!(query.contains("LIMIT 5")); + } #[actix_web::test] async fn log_context_bounds_normalize_relative_times() { From 9baabc595e7c562f213de6cb5f1750cabed49520 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Date: Thu, 15 Oct 2026 15:48:11 +0530 Subject: [PATCH 05/12] set P_SEND_ANONYMOUS_USAGE_DATA=false for docker compose tests (#1825) --- docker-compose-distributed-test-with-kafka.yaml | 3 +++ docker-compose-distributed-test.yaml | 2 ++ docker-compose-gcs-distributed-test.yaml | 4 +++- docker-compose-test-with-kafka.yaml | 1 + docker-compose-test.yaml | 1 + 5 files changed, 10 insertions(+), 1 deletion(-) diff --git a/docker-compose-distributed-test-with-kafka.yaml b/docker-compose-distributed-test-with-kafka.yaml index 7ad424d9d..f173e22b0 100644 --- a/docker-compose-distributed-test-with-kafka.yaml +++ b/docker-compose-distributed-test-with-kafka.yaml @@ -72,6 +72,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=query - RUST_LOG=warn @@ -113,6 +114,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=ingest - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000 @@ -161,6 +163,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=ingest - P_INGESTOR_ENDPOINT=parseable-ingest-two:8000 diff --git a/docker-compose-distributed-test.yaml b/docker-compose-distributed-test.yaml index 25b246638..52a5352f3 100644 --- a/docker-compose-distributed-test.yaml +++ b/docker-compose-distributed-test.yaml @@ -43,6 +43,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=query - RUST_LOG=warn @@ -80,6 +81,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=ingest - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000 diff --git a/docker-compose-gcs-distributed-test.yaml b/docker-compose-gcs-distributed-test.yaml index 16c4c79bd..a90e53e2f 100644 --- a/docker-compose-gcs-distributed-test.yaml +++ b/docker-compose-gcs-distributed-test.yaml @@ -18,6 +18,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=query - RUST_LOG=warn @@ -33,7 +34,7 @@ services: restart_policy: condition: on-failure delay: 20s - max_attempts: + max_attempts: 3 volumes: - "${GCS_CREDENTIALS_PATH:-./credentials}:/parseable/svc/:ro,z" # ingest server one @@ -52,6 +53,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_MODE=ingest - P_INGESTOR_ENDPOINT=parseable-ingest-one:8000 diff --git a/docker-compose-test-with-kafka.yaml b/docker-compose-test-with-kafka.yaml index 6e630e202..ff8f09f2e 100644 --- a/docker-compose-test-with-kafka.yaml +++ b/docker-compose-test-with-kafka.yaml @@ -44,6 +44,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - P_KAFKA_CONSUMER_TOPICS=test-logs-stream - P_KAFKA_BOOTSTRAP_SERVERS=kafka-0:9092 diff --git a/docker-compose-test.yaml b/docker-compose-test.yaml index 635e31ceb..062cc04dc 100644 --- a/docker-compose-test.yaml +++ b/docker-compose-test.yaml @@ -42,6 +42,7 @@ services: - P_USERNAME=parseableadmin - P_PASSWORD=parseableadmin - P_CHECK_UPDATE=false + - P_SEND_ANONYMOUS_USAGE_DATA=false - P_PARQUET_COMPRESSION_ALGO=snappy - MAX_PERIODIC_LOCAL_SYNC_CYCLES=1 - RUST_LOG=warn From a5976d2e911cdd4391dbc6360b00470281f619f9 Mon Sep 17 00:00:00 2001 From: Nikhil Sinha <131262146+nikhilsinhaparseable@users.noreply.github.com> Date: Thu, 15 Oct 2026 16:42:43 +0530 Subject: [PATCH 06/12] fix: create/update sql filters (#1824) currently failing because UI sends `stream_name=sql` in create/update filters when called from sql editor the stream name validation was added in the PR https://github.com/parseablehq/parseable/pull/1727 for a security fix https://github.com/parseablehq/parseable/security/advisories/GHSA-q8xx-r33r-6mwv with this change, the security fix still works also, UI will be able to save the filter as-is --- src/handlers/http/users/filters.rs | 41 ++++++++++++++++++++++-------- 1 file changed, 31 insertions(+), 10 deletions(-) diff --git a/src/handlers/http/users/filters.rs b/src/handlers/http/users/filters.rs index c7a4d5681..85cf0d92e 100644 --- a/src/handlers/http/users/filters.rs +++ b/src/handlers/http/users/filters.rs @@ -22,7 +22,7 @@ use crate::{ parseable::PARSEABLE, rbac::{Users, map::SessionKey}, storage::{ObjectStorageError, StreamType}, - users::filters::{CURRENT_FILTER_VERSION, FILTERS, Filter}, + users::filters::{CURRENT_FILTER_VERSION, FILTERS, Filter, FilterType}, utils::{ actix::extract_session_key_from_req, get_hash, get_user_and_tenant_from_request, is_admin, is_admin_for_session, @@ -38,6 +38,8 @@ use actix_web::{ use serde_json::Error as SerdeError; use ulid::Ulid; +const SQL_FILTER_STREAM_NAME: &str = "sql"; + pub async fn list(req: HttpRequest) -> Result { let key = extract_session_key_from_req(&req).map_err(|e| FiltersError::Custom(e.to_string()))?; @@ -82,7 +84,7 @@ pub async fn post( req: HttpRequest, Json(mut filter): Json, ) -> Result { - validate_stream_name(&filter.stream_name)?; + validate_stream_name(&mut filter.stream_name, &filter.query.filter_type)?; let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; user_id = get_hash(&user_id); @@ -102,7 +104,7 @@ pub async fn update( filter_id: Path, Json(mut filter): Json, ) -> Result { - validate_stream_name(&filter.stream_name)?; + validate_stream_name(&mut filter.stream_name, &filter.query.filter_type)?; let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; user_id = get_hash(&user_id); @@ -153,7 +155,15 @@ pub async fn delete( Ok(HttpResponse::Ok().finish()) } -fn validate_stream_name(stream_name: &str) -> Result<(), FiltersError> { +fn validate_stream_name( + stream_name: &mut String, + filter_type: &FilterType, +) -> Result<(), FiltersError> { + if matches!(filter_type, FilterType::SQL) { + SQL_FILTER_STREAM_NAME.clone_into(stream_name); + return Ok(()); + } + validator::stream_name(stream_name, StreamType::UserDefined) .map_err(FiltersError::InvalidStreamName) } @@ -205,20 +215,31 @@ impl actix_web::ResponseError for FiltersError { #[cfg(test)] mod tests { - use super::validate_stream_name; + use super::{SQL_FILTER_STREAM_NAME, validate_stream_name}; + use crate::users::filters::FilterType; use crate::validator::error::StreamNameValidationError; #[test] fn accepts_flat_stream_names() { - assert!(validate_stream_name("team1").is_ok()); - assert!(validate_stream_name("orders-prod").is_ok()); - assert!(validate_stream_name("user_events_v2").is_ok()); + for stream_name in ["team1", "orders-prod", "user_events_v2"] { + let mut stream_name = stream_name.to_owned(); + assert!(validate_stream_name(&mut stream_name, &FilterType::Filter).is_ok()); + } + } + + #[test] + fn normalizes_sql_filter_stream_name() { + let mut stream_name = "../../unsafe-path".to_owned(); + + assert!(validate_stream_name(&mut stream_name, &FilterType::SQL).is_ok()); + assert_eq!(stream_name, SQL_FILTER_STREAM_NAME); } #[test] fn rejects_path_like_stream_names() { - let err = - validate_stream_name("team1/serviceA").expect_err("path-like names must be rejected"); + let mut stream_name = "team1/serviceA".to_owned(); + let err = validate_stream_name(&mut stream_name, &FilterType::Search) + .expect_err("path-like names must be rejected"); assert!(matches!( err, super::FiltersError::InvalidStreamName(StreamNameValidationError::NameSpecialChar { From 99bba954bcbee033fab96cbe283deefc7b67f581 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:10:42 -0500 Subject: [PATCH 07/12] Preserve fork release targets and OTEL regression coverage --- README.md | 8 +- scripts/download.ps1 | 18 +++- scripts/download.sh | 8 +- scripts/test-download.py | 181 +++++++++++++++++++++++++++++++++++++++ src/about.rs | 2 +- src/otel/logs.rs | 64 +++++++++++++- src/utils/update.rs | 10 +-- 7 files changed, 273 insertions(+), 18 deletions(-) create mode 100644 scripts/test-download.py diff --git a/README.md b/README.md index 8906ca895..a8cb676fb 100644 --- a/README.md +++ b/README.md @@ -37,17 +37,19 @@ Some of the key highlights include: ## Quickstart -Download and run the Parseable binary on your laptop: +Download the latest published binary from this fork (`VibeEngineeringAI/parseable`). +The installer requires a published release containing the asset for your platform; +it will fail if one is unavailable. ```bash -curl -fsSL https://logg.ing/install | bash +curl -fsSL https://raw.githubusercontent.com/VibeEngineeringAI/parseable/main/scripts/download.sh | sh ```
For Windows ```pwsh -powershell -c "irm https://logg.ing/install-windows | iex" +powershell -c "irm https://raw.githubusercontent.com/VibeEngineeringAI/parseable/main/scripts/download.ps1 | iex" ```
diff --git a/scripts/download.ps1 b/scripts/download.ps1 index 6ce086d16..37be258e0 100644 --- a/scripts/download.ps1 +++ b/scripts/download.ps1 @@ -107,7 +107,13 @@ function Install-Parseable { Write-Output "Fetching latest release..." # Get the latest release information using GitHub API - $release = Invoke-RestMethod -Uri "https://api.github.com/repos/parseablehq/parseable/releases/latest" + try { + $release = Invoke-RestMethod -Uri "https://api.github.com/repos/VibeEngineeringAI/parseable/releases/latest" -ErrorAction Stop + } catch { + Write-Error "Could not fetch the latest release information from VibeEngineeringAI/parseable. Check that this fork has a published release." + exit 1 + } + $release_tag = $release.tag_name # Write-Host ($release | Format-List | Out-String) # Write-Host formats the object while printing $installation_found_flag = 0 @@ -125,10 +131,16 @@ function Install-Parseable { Write-Output "Downloading Parseable version $release_tag, for OS: $OS, CPU architecture: $CPU_ARCH`n`n" # Download the binary using Invoke-WebRequest if ($installation_found_flag) { - Invoke-WebRequest -Uri $download_url -OutFile $BIN + try { + Invoke-WebRequest -Uri $download_url -OutFile $BIN -ErrorAction Stop + } catch { + Remove-Item -LiteralPath $BIN -Force -ErrorAction SilentlyContinue + Write-Error "Download failed: $_" + exit 1 + } } else { Write-Output "Unable to find correct installation. Exiting" - exit + exit 1 } diff --git a/scripts/download.sh b/scripts/download.sh index 86117fc5a..66254ce81 100755 --- a/scripts/download.sh +++ b/scripts/download.sh @@ -1,6 +1,6 @@ #!/bin/sh -DOWNLOAD_BASE_URL="https://github.com/parseablehq/parseable/releases/download" +DOWNLOAD_BASE_URL="https://github.com/VibeEngineeringAI/parseable/releases/download" INSTALL_DIR="${HOME}/.parseable" BIN_DIR="${INSTALL_DIR}/bin" BIN_NAME="${BIN_DIR}/parseable" @@ -64,15 +64,15 @@ fi fetch_release() { case "$DOWNLOADER" in curl) - curl -fsSL "https://api.github.com/repos/parseablehq/parseable/releases/latest" + curl -fsSL "https://api.github.com/repos/VibeEngineeringAI/parseable/releases/latest" ;; wget) - wget -q -O - "https://api.github.com/repos/parseablehq/parseable/releases/latest" + wget -q -O - "https://api.github.com/repos/VibeEngineeringAI/parseable/releases/latest" ;; esac } if ! release=$(fetch_release); then - printf 'Error: Could not fetch the latest release information.\n' >&2 + printf 'Error: Could not fetch the latest release information from VibeEngineeringAI/parseable. Check that this fork has a published release.\n' >&2 exit 1 fi release_tag=$(printf '%s\n' "$release" | sed -n 's/.*"tag_name"[[:space:]]*:[[:space:]]*"\([^"]*\)".*/\1/p' | head -n 1) diff --git a/scripts/test-download.py b/scripts/test-download.py new file mode 100644 index 000000000..88f214e82 --- /dev/null +++ b/scripts/test-download.py @@ -0,0 +1,181 @@ +#!/usr/bin/env python3 +"""Installer regression tests: only mocked HTTP and disposable shell profiles. + +Run with python3 scripts/test-download.py. No downloaded file is executed. +Requires a glibc runner: the installer also checks the real /lib for musl loaders. +""" +import os +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + + +INSTALLER = Path(__file__).with_name('download.sh').resolve() +REPO = 'VibeEngineeringAI/parseable' + + +class InstallerTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory(prefix='parseable-installer-') + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.home = self.root / 'home with spaces' + self.home.mkdir() + self.tools = self.root / 'tools' + self.tools.mkdir() + # A closed PATH ensures missing-downloader tests cannot contact the network. + for name in ('tr', 'grep', 'sed', 'head', 'mkdir', 'rmdir', 'rm', 'chmod', 'mv', 'tail'): + (self.tools / name).symlink_to(shutil.which(name)) + self.env = dict(os.environ, HOME=str(self.home), PATH=str(self.tools), + SHELL='/bin/bash', MOCK_OS='Linux', MOCK_ARCH='x86_64', + MOCK_LIBC='glibc', MOCK_FAILURE='', + MOCK_LOG=str(self.root / 'requests')) + self.mock('uname', '''case "$1" in +-m) printf '%s\\n' "$MOCK_ARCH" ;; +-s) printf '%s\\n' "$MOCK_OS" ;; +esac''') + self.mock('ldd', "printf '%s\\n' \"$MOCK_LIBC\"") + + def mock(self, name, body): + path = self.tools / name + path.write_text('#!/bin/sh\n' + body + '\n') + path.chmod(0o755) + + def downloader(self, name='curl'): + self.mock(name, '''output='' +url='' +while [ "$#" -gt 0 ]; do + case "$1" in + -o|-O) output=$2; shift ;; + https://*) url=$1 ;; + esac + shift +done +printf '%s\\n' "$url" >> "$MOCK_LOG" +case "$url" in + https://api.github.com/repos/VibeEngineeringAI/parseable/releases/latest) + [ "$MOCK_FAILURE" = metadata ] && exit 22 + if [ "$MOCK_FAILURE" = missing_tag ]; then + printf '{"message":"Not Found"}\\n' + else + printf '{"tag_name": "v1.2.3"}\\n' + fi + ;; + https://github.com/VibeEngineeringAI/parseable/releases/download/v1.2.3/*) + printf 'mock binary, never execute\\n' > "$output" + [ "$MOCK_FAILURE" = download ] && exit 22 + ;; + *) printf 'Unexpected network request: %s\\n' "$url" >&2; exit 99 ;; +esac +exit 0''') + + def run_installer(self, **env): + return subprocess.run(['/bin/sh', str(INSTALLER)], env=dict(self.env, **env), + capture_output=True, text=True) + + def requests(self): + path = self.root / 'requests' + return path.read_text().splitlines() if path.exists() else [] + + def test_platform_assets_and_downloaders(self): + platforms = [('Linux', 'x86_64', 'x86_64-unknown-linux-gnu'), + ('Linux', 'amd64', 'x86_64-unknown-linux-gnu'), + ('Linux', 'aarch64', 'aarch64-unknown-linux-gnu'), + ('Linux', 'arm64', 'aarch64-unknown-linux-gnu'), + ('Darwin', 'arm64', 'aarch64-apple-darwin')] + for downloader in ('curl', 'wget'): + self.downloader(downloader) + for system, arch, asset in platforms: + with self.subTest(downloader=downloader, system=system, arch=arch): + result = self.run_installer(MOCK_OS=system, MOCK_ARCH=arch) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.requests()[-2:], [ + f'https://api.github.com/repos/{REPO}/releases/latest', + f'https://github.com/{REPO}/releases/download/v1.2.3/Parseable_OSS_{asset}']) + binary = self.home / '.parseable/bin/parseable' + self.assertEqual(binary.read_text(), 'mock binary, never execute\n') + self.assertTrue(os.access(binary, os.X_OK)) + shutil.rmtree(self.home / '.parseable') + (self.tools / downloader).unlink() + + def test_no_downloader_fails_without_installation(self): + result = self.run_installer() + self.assertNotEqual(result.returncode, 0) + self.assertIn('Neither curl nor wget', result.stderr) + self.assertFalse((self.home / '.parseable').exists()) + + def test_unsupported_platforms_and_musl_fail_before_network(self): + self.downloader() + for system, arch, libc in [('Darwin', 'x86_64', 'glibc'), + ('Linux', 'riscv64', 'glibc'), + ('Windows', 'x86_64', 'glibc'), + ('Linux', 'x86_64', 'musl libc')]: + with self.subTest(system=system, arch=arch, libc=libc): + result = self.run_installer(MOCK_OS=system, MOCK_ARCH=arch, MOCK_LIBC=libc) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(self.requests(), []) + self.assertFalse((self.home / '.parseable').exists()) + + def test_release_failures_leave_home_unchanged(self): + self.downloader() + for failure in ('metadata', 'missing_tag'): + with self.subTest(failure=failure): + result = self.run_installer(MOCK_FAILURE=failure) + self.assertNotEqual(result.returncode, 0) + self.assertEqual(list(self.home.iterdir()), []) + self.assertNotIn('successfully installed', result.stdout) + + def test_failed_asset_download_cleans_up_and_can_be_retried(self): + self.downloader() + result = self.run_installer(MOCK_FAILURE='download') + self.assertNotEqual(result.returncode, 0) + self.assertEqual(list(self.home.iterdir()), []) + self.assertEqual(self.run_installer().returncode, 0) + + def test_existing_installation_is_preserved(self): + self.downloader() + binary = self.home / '.parseable/bin/parseable' + binary.parent.mkdir(parents=True) + binary.write_text('keep existing binary') + result = self.run_installer() + self.assertNotEqual(result.returncode, 0) + self.assertEqual(binary.read_text(), 'keep existing binary') + self.assertEqual(len(self.requests()), 1) + self.assertFalse((self.home / '.bashrc').exists()) + + def test_bash_profiles_are_not_sourced_and_path_is_not_duplicated(self): + self.downloader() + profile = self.home / '.profile' + profile.write_text('touch "${HOME}/profile-was-sourced"') + result = self.run_installer() + self.assertEqual(result.returncode, 0, result.stderr) + path_line = f'export PATH="{self.home}/.parseable/bin:$PATH"' + self.assertEqual(profile.read_text(), 'touch "${HOME}/profile-was-sourced"\n' + path_line + '\n') + self.assertEqual((self.home / '.bashrc').read_text(), path_line + '\n') + self.assertFalse((self.home / 'profile-was-sourced').exists()) + shutil.rmtree(self.home / '.parseable') + self.assertEqual(self.run_installer().returncode, 0) + self.assertEqual(profile.read_text().count(path_line), 1) + self.assertEqual((self.home / '.bashrc').read_text().count(path_line), 1) + + def test_zsh_respects_zdotdir(self): + self.downloader() + zdotdir = self.root / 'zsh profiles' + zdotdir.mkdir() + result = self.run_installer(SHELL='/bin/zsh', ZDOTDIR=str(zdotdir)) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertTrue((zdotdir / '.zshrc').exists()) + self.assertFalse((self.home / '.zshrc').exists()) + + def test_unknown_shell_prints_manual_path_without_profile_changes(self): + self.downloader() + result = self.run_installer(SHELL='/bin/fish') + self.assertEqual(result.returncode, 0, result.stderr) + self.assertIn('To use Parseable in the current shell', result.stdout) + self.assertEqual([path.name for path in self.home.iterdir()], ['.parseable']) + + +if __name__ == '__main__': + unittest.main(verbosity=2) diff --git a/src/about.rs b/src/about.rs index 0816f9c69..d2d8a3e29 100644 --- a/src/about.rs +++ b/src/about.rs @@ -129,7 +129,7 @@ fn print_latest_release(latest_release: LatestRelease) { let time_since_latest_release = chrono::Utc::now() - latest_release.date; let time_since_latest_release = humanize_time(time_since_latest_release); let fmt_latest_version = format!( - " ( v{} released {} ago. Download new release from https://github.com/parseablehq/parseable/releases/latest )", + " ( v{} released {} ago. Download new release from https://github.com/VibeEngineeringAI/parseable/releases/latest )", latest_release.version, time_since_latest_release ); eprint!("{}", fmt_latest_version.red()); diff --git a/src/otel/logs.rs b/src/otel/logs.rs index a7cbc0b42..eb53bb39a 100644 --- a/src/otel/logs.rs +++ b/src/otel/logs.rs @@ -363,7 +363,69 @@ pub fn flatten_otel_logs(message: &LogsData, tenant_id: &str) -> Vec { #[cfg(test)] mod tests { - use super::normalize_severity; + use super::{flatten_log_record, normalize_severity}; + use opentelemetry_proto::tonic::common::v1::{AnyValue, any_value}; + use opentelemetry_proto::tonic::logs::v1::LogRecord; + use serde_json::json; + + fn flatten_record( + number: i32, + text: &str, + body: &str, + ) -> serde_json::Map { + flatten_log_record(&LogRecord { + severity_number: number, + severity_text: text.to_string(), + body: Some(AnyValue { + value: Some(any_value::Value::StringValue(body.to_string())), + }), + ..Default::default() + }) + } + + #[test] + fn text_only_error_record_is_categorized_with_a_neutral_body() { + let record = flatten_record(0, "ERROR", "request completed"); + assert_eq!(record["severity_number"], json!(17)); + assert_eq!(record["severity_text"], json!("ERROR")); + assert_eq!(record["p_log_category"], json!("ERROR")); + } + + #[test] + fn invalid_positive_and_negative_severity_numbers_do_not_panic() { + for number in [25, i32::MAX, -1, i32::MIN] { + let record = flatten_record(number, "", "request completed"); + assert_eq!(record["severity_number"], json!(number)); + assert_eq!(record["severity_text"], json!("UNSPECIFIED")); + assert_eq!(record["p_log_category"], json!("UNSPECIFIED")); + + let record = flatten_record(number, "custom", "ERROR: request failed"); + assert_eq!(record["severity_text"], json!("custom")); + assert_eq!(record["p_log_category"], json!("ERROR")); + } + } + + #[test] + fn explicit_number_wins_over_contradictory_text_and_body_for_category() { + let record = flatten_record(9, "ERROR", "fatal request failure"); + assert_eq!(record["severity_number"], json!(9)); + assert_eq!(record["severity_text"], json!("ERROR")); + assert_eq!(record["p_log_category"], json!("INFO")); + + let record = flatten_record(0, "INFO", "fatal request failure"); + assert_eq!(record["severity_number"], json!(9)); + assert_eq!(record["p_log_category"], json!("INFO")); + } + + #[test] + fn unspecified_severity_retains_body_fallback() { + for text in ["", "UNSPECIFIED", " unspecified "] { + let record = flatten_record(0, text, "Error: request failed"); + assert_eq!(record["severity_number"], json!(0)); + assert_eq!(record["severity_text"], json!("UNSPECIFIED")); + assert_eq!(record["p_log_category"], json!("ERROR")); + } + } #[test] fn derives_severity_text_from_number_when_missing() { diff --git a/src/utils/update.rs b/src/utils/update.rs index 73c3f4c3d..f8f9fda8c 100644 --- a/src/utils/update.rs +++ b/src/utils/update.rs @@ -21,8 +21,6 @@ use std::time::Duration; use anyhow::anyhow; use chrono::{DateTime, Utc}; -use crate::{about, parseable::PARSEABLE}; - use super::uid; #[derive(Debug, Clone)] @@ -31,18 +29,18 @@ pub struct LatestRelease { pub date: DateTime, } -pub async fn get_latest(deployment_id: &uid::Uid) -> Result { - let send_analytics = PARSEABLE.options.send_analytics; +pub async fn get_latest(_deployment_id: &uid::Uid) -> Result { let agent = reqwest::ClientBuilder::new() - .user_agent(about::user_agent(deployment_id, send_analytics)) + .user_agent(concat!("Parseable/", env!("CARGO_PKG_VERSION"))) .timeout(Duration::from_secs(8)) .build() .expect("client can be built on this system"); let json: serde_json::Value = agent - .get("https://download.parseable.io/latest-version") + .get("https://api.github.com/repos/VibeEngineeringAI/parseable/releases/latest") .send() .await? + .error_for_status()? .json() .await?; let version = json["tag_name"] From ac1424124972ecf6eb6d96b7daa0f123f13ab6ea Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:25:24 -0500 Subject: [PATCH 08/12] Support gradual SQL filter conversion with restart-safe reconciliation --- docs/sql-filter-conversion.md | 73 ++ scripts/test-sql-filter-conversion.py | 160 +++++ src/handlers/http/modal/server.rs | 9 +- src/handlers/http/users/filters.rs | 160 ++++- src/metastore/metastore_traits.rs | 8 +- .../metastores/object_store_metastore.rs | 632 ++++++++++++++---- src/rbac/role.rs | 25 + src/users/filters.rs | 99 ++- 8 files changed, 1004 insertions(+), 162 deletions(-) create mode 100644 docs/sql-filter-conversion.md create mode 100644 scripts/test-sql-filter-conversion.py diff --git a/docs/sql-filter-conversion.md b/docs/sql-filter-conversion.md new file mode 100644 index 000000000..a476a675c --- /dev/null +++ b/docs/sql-filter-conversion.md @@ -0,0 +1,73 @@ +# Saved SQL filter compatibility and conversion + +Dataset-scoped SQL filters and upstream SQL filters using `stream_name: "sql"` +are both supported. Existing views are not automatically converted. New native +Logs views keep their dataset-scoped format until the owner chooses **Convert to +SQL view** in Saved views. Converted views still open in the original dataset; +`explorer_state` retains their dataset, query controls, time range and display +settings. Editing a converted view keeps it converted. + +The native explorer reads both formats. Classic filters without native explorer +state remain visible under their original dataset, but cannot be opened or +converted by the native explorer. The literal `sql` is a reserved dataset name, +so it cannot collide with a user-created dataset. + +## Backend operation + +``` +POST /api/v1/filters/{filter_id}/convert-to-sql +``` + +The endpoint takes no JSON payload and returns the complete saved filter. It +requires filter-update permission and an owner-or-admin check. Only SQL filters +can be converted. The server copies the authorized stored record, preserving +its ID, owner, query, time filter and extra metadata. Repeating the request is +safe, including after an interrupted cleanup. Admin updates also preserve the +original owner. + +Ordinary create and update requests accept a valid dataset name or, for SQL +filters only, the literal `sql`. Path-like names remain invalid. A stale client +cannot turn a converted SQL filter back into a dataset-scoped one by sending its +old stream name. + +## Reconciliation and recovery + +Persistence uses `users//filters//.json`. Conversion writes +the `sql` copy first, updates the server cache, then removes obsolete copies of +the same owner's filter ID. Cleanup failure returns an error, while the durable +new copy remains available for retry. Deletion removes aliases before the +current copy, so a successful deletion cannot leave an older copy to return on +restart. + +Startup reconciliation groups records by owner and ID within each tenant. When +both canonical SQL and legacy SQL copies exist, it deterministically keeps the +canonical copy and removes obsolete SQL copies. A legacy record with no +canonical copy is left in its existing format. Ambiguous legacy-only duplicates +are selected deterministically without deleting the alternatives. Cleanup +failures are logged and can be retried on the next startup or conversion; there +is no one-time completion flag that could hide unfinished work. + +A process-wide async lock serializes filter persistence and cache changes. +This follows the existing single-query-server ownership model: it is not a +distributed lock, and concurrent writes through multiple independent query +servers sharing the same metastore are not coordinated by this change. + +Before a rollout, back up stored filter metadata and verify a legacy view can +be opened, converted, edited, restarted and deleted without reappearing. A +rollback to an older native UI will hide converted views because that UI only +understands dataset-scoped records; keep this compatible UI when rolling back +other changes. Conversion does not change log data or delete datasets. + +## Regression checks + +Run the isolated HTTP/restart smoke test against a newly built server: + +```sh +PARSEABLE_BIN=/absolute/path/to/parseable python3 scripts/test-sql-filter-conversion.py +``` + +It starts its own local-store process and temporary data directories; it does not +contact an existing deployment. The fixture checks owner authorization, admin +updates, conversion retries, stale-client edits, startup cleanup of an injected +legacy duplicate, and delete followed by restart. Rust metastore tests also +exercise cleanup failures, tenant isolation, and ambiguous legacy records. diff --git a/scripts/test-sql-filter-conversion.py b/scripts/test-sql-filter-conversion.py new file mode 100644 index 000000000..c0b8e01ab --- /dev/null +++ b/scripts/test-sql-filter-conversion.py @@ -0,0 +1,160 @@ +#!/usr/bin/env python3 +"""Live saved-filter conversion/restart regression using an isolated local-store. + +Run: PARSEABLE_BIN=/absolute/path/to/parseable python3 scripts/test-sql-filter-conversion.py +No existing server, data directory, credentials, or Parseable environment is used. +""" +import base64 +import json +import os +from pathlib import Path +import socket +import subprocess +import tempfile +import time +import urllib.error +import urllib.request +import uuid + + +def free_port(): + with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + return sock.getsockname()[1] + + +def main(): + binary = Path(os.environ['PARSEABLE_BIN']).expanduser().resolve() + if not binary.is_file() or not os.access(binary, os.X_OK): + raise RuntimeError('PARSEABLE_BIN must point to an executable server') + with tempfile.TemporaryDirectory(prefix='parseable-filter-smoke-') as directory: + root = Path(directory) + ports = set() + while len(ports) < 4: + ports.add(free_port()) + http_port, grpc_port, flight_port, query_port = ports + url = f'http://127.0.0.1:{http_port}' + admin = ('filter-smoke-admin', uuid.uuid4().hex) + env = { + 'PATH': os.environ.get('PATH', '/usr/bin:/bin'), + 'P_ADDR': f'127.0.0.1:{http_port}', + 'P_GRPC_PORT': str(grpc_port), 'P_FLIGHT_PORT': str(flight_port), + 'P_QUERY_GRPC_PORT': str(query_port), + 'P_USERNAME': admin[0], 'P_PASSWORD': admin[1], + 'P_FS_DIR': str(root / 'data'), 'P_STAGING_DIR': str(root / 'staging'), + 'P_CHECK_UPDATE': 'false', 'P_SEND_ANONYMOUS_USAGE_DATA': 'false', + 'P_ACTIX_NUM_WORKERS': '2', + } + process = None + log = (root / 'server.log').open('wb') + + def request(method, path, body=None, auth=admin, expected=200): + headers = {'Accept': 'application/json'} + if auth is not None: + token = base64.b64encode(f'{auth[0]}:{auth[1]}'.encode()).decode() + headers['Authorization'] = f'Basic {token}' + data = None + if body is not None: + data = json.dumps(body).encode() + headers['Content-Type'] = 'application/json' + req = urllib.request.Request(url + path, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=15) as response: + status, raw = response.status, response.read() + except urllib.error.HTTPError as error: + status, raw = error.code, error.read() + if expected is None: + assert status >= 400, f'{method} {path}: expected denial, got {status}' + return None + assert status == expected, f'{method} {path}: expected {expected}, got {status}: {raw[:500]!r}' + try: + return json.loads(raw) if raw else None + except json.JSONDecodeError: + return raw.decode() + + def stop(): + nonlocal process + if process is not None: + process.terminate() + try: + process.wait(timeout=15) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + process = None + + def start(): + nonlocal process + process = subprocess.Popen([str(binary), 'local-store'], cwd=root, env=env, + stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT) + deadline = time.monotonic() + 45 + while time.monotonic() < deadline: + if process.poll() is not None: + raise RuntimeError('Disposable server exited during startup') + try: + request('GET', '/api/v1/liveness') + return + except (urllib.error.URLError, TimeoutError): + time.sleep(0.2) + raise RuntimeError('Disposable server did not become ready') + + try: + start() + dataset = 'filter_smoke_logs' + request('PUT', f'/api/v1/logstream/{dataset}', {}) + request('PUT', '/api/v1/role/filter-smoke-reader', { + 'roleType': 'user', + 'actions': [{'privilege': 'reader', 'resource': {'stream': dataset}}], + }) + owner = ('filter-owner', request('POST', '/api/v1/user/filter-owner', ['filter-smoke-reader'])) + other = ('filter-other', request('POST', '/api/v1/user/filter-other', ['filter-smoke-reader'])) + body = { + 'stream_name': dataset, 'filter_name': 'Incident view', + 'query': {'filter_type': 'sql', 'filter_query': f'SELECT * FROM {dataset}', 'filter_builder': None}, + 'time_filter': None, + 'explorer_state': 'preserve-this-native-state', 'labels': ['incident'], + } + legacy = request('POST', '/api/v1/filters', body, auth=owner) + assert legacy['stream_name'] == dataset + identifier = legacy['filter_id'] + path = f'/api/v1/filters/{identifier}' + request('POST', path + '/convert-to-sql', auth=other, expected=None) + request('POST', path + '/convert-to-sql', auth=None, expected=None) + # Admin editing must not move the record into the admin's directory. + edited = request('PUT', path, {**legacy, 'filter_name': 'Admin edit'}) + assert edited['user_id'] == legacy['user_id'] + converted = request('POST', path + '/convert-to-sql', auth=owner) + assert converted == {**edited, 'stream_name': 'sql'} + assert request('POST', path + '/convert-to-sql', auth=owner) == converted + # A stale client must not undo conversion on a normal edit. + updated = request('PUT', path, {**edited, 'filter_name': 'After conversion'}, auth=owner) + assert updated['stream_name'] == 'sql' + assert updated['user_id'] == legacy['user_id'] + assert updated['explorer_state'] == body['explorer_state'] + assert any(view['filter_id'] == identifier for view in request('GET', '/api/v1/filters', auth=owner)) + request('POST', '/api/v1/filters', {**body, 'stream_name': '../../bad'}, auth=owner, expected=None) + classic = request('POST', '/api/v1/filters', {**body, 'query': {**body['query'], 'filter_type': 'search'}}, auth=owner) + request('POST', f"/api/v1/filters/{classic['filter_id']}/convert-to-sql", auth=owner, expected=None) + stop() + files = list((root / 'data').rglob(f'{identifier}.json')) + assert len(files) == 1 and files[0].parent.name == 'sql' + # Emulate a canonical write that crashed before removing the old copy. + stale = files[0].parent.parent / dataset / files[0].name + stale.parent.mkdir(parents=True, exist_ok=True) + stale.write_text(json.dumps(legacy)) + start() + assert request('GET', path, auth=owner)['filter_name'] == 'After conversion' + assert not stale.exists(), 'Startup must reconcile the obsolete copy' + request('DELETE', path, auth=owner) + stop() + start() + assert all(view['filter_id'] != identifier for view in request('GET', '/api/v1/filters', auth=owner)) + assert not list((root / 'data').rglob(f'{identifier}.json')) + print('PASS: live legacy/create, owner authorization, admin ownership, conversion/retry, stale update, restart reconciliation, deletion/restart') + finally: + stop() + log.close() + + +if __name__ == '__main__': + main() diff --git a/src/handlers/http/modal/server.rs b/src/handlers/http/modal/server.rs index ae94eb526..c40be0234 100644 --- a/src/handlers/http/modal/server.rs +++ b/src/handlers/http/modal/server.rs @@ -500,6 +500,13 @@ impl Server { ) .route(web::get().to(filters::list).authorize(Action::ListFilter)), ) + .service( + web::resource("/{filter_id}/convert-to-sql").route( + web::post() + .to(filters::convert_to_sql) + .authorize(Action::UpdateFilter), + ), + ) .service( web::resource("/{filter_id}") .route(web::get().to(filters::get).authorize(Action::GetFilter)) @@ -511,7 +518,7 @@ impl Server { .route( web::put() .to(filters::update) - .authorize(Action::CreateFilter), + .authorize(Action::UpdateFilter), ), ) } diff --git a/src/handlers/http/users/filters.rs b/src/handlers/http/users/filters.rs index 85cf0d92e..d51afdf77 100644 --- a/src/handlers/http/users/filters.rs +++ b/src/handlers/http/users/filters.rs @@ -22,7 +22,10 @@ use crate::{ parseable::PARSEABLE, rbac::{Users, map::SessionKey}, storage::{ObjectStorageError, StreamType}, - users::filters::{CURRENT_FILTER_VERSION, FILTERS, Filter, FilterType}, + users::filters::{ + CURRENT_FILTER_VERSION, FILTER_MUTATION_LOCK, FILTERS, Filter, FilterType, + SQL_FILTER_STREAM_NAME, + }, utils::{ actix::extract_session_key_from_req, get_hash, get_user_and_tenant_from_request, is_admin, is_admin_for_session, @@ -38,8 +41,6 @@ use actix_web::{ use serde_json::Error as SerdeError; use ulid::Ulid; -const SQL_FILTER_STREAM_NAME: &str = "sql"; - pub async fn list(req: HttpRequest) -> Result { let key = extract_session_key_from_req(&req).map_err(|e| FiltersError::Custom(e.to_string()))?; @@ -84,7 +85,8 @@ pub async fn post( req: HttpRequest, Json(mut filter): Json, ) -> Result { - validate_stream_name(&mut filter.stream_name, &filter.query.filter_type)?; + validate_stream_name(&filter.stream_name, &filter.query.filter_type)?; + let _mutation = FILTER_MUTATION_LOCK.lock().await; let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; user_id = get_hash(&user_id); @@ -95,6 +97,10 @@ pub async fn post( filter.tenant_id.clone_from(&tenant_id); PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?; FILTERS.update(&filter, &tenant_id).await; + PARSEABLE + .metastore + .reconcile_filter(&filter, &tenant_id) + .await?; Ok((web::Json(filter), StatusCode::OK)) } @@ -104,29 +110,31 @@ pub async fn update( filter_id: Path, Json(mut filter): Json, ) -> Result { - validate_stream_name(&mut filter.stream_name, &filter.query.filter_type)?; + validate_stream_name(&filter.stream_name, &filter.query.filter_type)?; + let _mutation = FILTER_MUTATION_LOCK.lock().await; let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; user_id = get_hash(&user_id); let filter_id = filter_id.into_inner(); let is_admin = is_admin(&req).map_err(|e| FiltersError::Custom(e.to_string()))?; - if FILTERS + let existing = FILTERS .get_filter(&filter_id, &user_id, is_admin, &tenant_id) .await - .is_none() - { - return Err(FiltersError::Metadata( + .ok_or(FiltersError::Metadata( "Filter does not exist or user is not authorized", - )); - } + ))?; + preserve_filter_identity(&existing, &mut filter); filter.filter_id = Some(filter_id.clone()); - filter.user_id = Some(user_id.clone()); filter.version = Some(CURRENT_FILTER_VERSION.to_string()); filter.tenant_id.clone_from(&tenant_id); PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?; FILTERS.update(&filter, &tenant_id).await; + PARSEABLE + .metastore + .reconcile_filter(&filter, &tenant_id) + .await?; Ok((web::Json(filter), StatusCode::OK)) } @@ -135,6 +143,7 @@ pub async fn delete( req: HttpRequest, filter_id: Path, ) -> Result { + let _mutation = FILTER_MUTATION_LOCK.lock().await; let (mut user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; user_id = get_hash(&user_id); let filter_id = filter_id.into_inner(); @@ -150,20 +159,62 @@ pub async fn delete( .metastore .delete_filter(&filter, &tenant_id) .await?; - FILTERS.delete_filter(&filter_id, &tenant_id).await; + FILTERS.delete_filter(&filter, &tenant_id).await; Ok(HttpResponse::Ok().finish()) } -fn validate_stream_name( - stream_name: &mut String, - filter_type: &FilterType, -) -> Result<(), FiltersError> { - if matches!(filter_type, FilterType::SQL) { - SQL_FILTER_STREAM_NAME.clone_into(stream_name); - return Ok(()); +/// Conversion deliberately accepts no client filter payload: metadata and ownership +/// come from the authorized stored record. +pub async fn convert_to_sql( + req: HttpRequest, + filter_id: Path, +) -> Result { + let _mutation = FILTER_MUTATION_LOCK.lock().await; + let (user_id, tenant_id) = get_user_and_tenant_from_request(&req)?; + let is_admin = is_admin(&req).map_err(|e| FiltersError::Custom(e.to_string()))?; + let mut filter = FILTERS + .get_filter(&filter_id, &get_hash(&user_id), is_admin, &tenant_id) + .await + .ok_or(FiltersError::Metadata( + "Filter does not exist or user is not authorized", + ))?; + convert_filter_to_sql(&mut filter)?; + filter.tenant_id.clone_from(&tenant_id); + PARSEABLE.metastore.put_filter(&filter, &tenant_id).await?; + // Once the write succeeds, serve the new record even if alias cleanup fails. + // A retry or startup reconciliation can safely finish cleanup. + FILTERS.update(&filter, &tenant_id).await; + PARSEABLE + .metastore + .reconcile_filter(&filter, &tenant_id) + .await?; + Ok((web::Json(filter), StatusCode::OK)) +} + +fn convert_filter_to_sql(filter: &mut Filter) -> Result<(), FiltersError> { + if filter.query.filter_type != FilterType::SQL { + return Err(FiltersError::Metadata("Only SQL filters can be converted")); } + SQL_FILTER_STREAM_NAME.clone_into(&mut filter.stream_name); + Ok(()) +} +fn preserve_filter_identity(existing: &Filter, incoming: &mut Filter) { + incoming.user_id.clone_from(&existing.user_id); + if existing.query.filter_type == FilterType::SQL + && existing.stream_name == SQL_FILTER_STREAM_NAME + && incoming.query.filter_type == FilterType::SQL + { + // A client holding an older dataset-scoped copy must not undo conversion. + SQL_FILTER_STREAM_NAME.clone_into(&mut incoming.stream_name); + } +} + +fn validate_stream_name(stream_name: &str, filter_type: &FilterType) -> Result<(), FiltersError> { + if matches!(filter_type, FilterType::SQL) && stream_name == SQL_FILTER_STREAM_NAME { + return Ok(()); + } validator::stream_name(stream_name, StreamType::UserDefined) .map_err(FiltersError::InvalidStreamName) } @@ -219,26 +270,77 @@ mod tests { use crate::users::filters::FilterType; use crate::validator::error::StreamNameValidationError; + #[actix_web::test] + async fn conversion_requires_authenticated_request() { + let request = actix_web::test::TestRequest::default().to_http_request(); + assert!(matches!( + super::convert_to_sql(request, actix_web::web::Path::from("id".to_owned())).await, + Err(super::FiltersError::UserDoesNotExist(_)) + )); + } + + #[test] + fn administrator_update_preserves_owner_and_canonical_sql_scope() { + let existing: crate::users::filters::Filter = serde_json::from_value(serde_json::json!({ + "version":"v2", "user_id":"original-owner", "filter_id":"id", + "stream_name":"sql", "filter_name":"View", "tenant_id":null, + "query":{"filter_type":"sql","filter_query":"select * from orders","filter_builder":null}, + "time_filter":null, "explorer_state":{"tab":"sql"} + })).unwrap(); + let mut incoming = existing.clone(); + incoming.user_id = Some("administrator".into()); + incoming.stream_name = "orders".into(); + super::preserve_filter_identity(&existing, &mut incoming); + assert_eq!(incoming, existing); + } + + #[test] + fn conversion_preserves_complete_record_and_rejects_non_sql_filters() { + let legacy: crate::users::filters::Filter = serde_json::from_value(serde_json::json!({ + "version":"v2", "user_id":"original-owner", "filter_id":"id", + "stream_name":"orders", "filter_name":"View", "tenant_id":"tenant", + "query":{"filter_type":"sql","filter_query":"select * from orders","filter_builder":null}, + "time_filter":null, "explorer_state":{"tab":"sql"}, "future_field":[1,2] + })).unwrap(); + let mut converted = legacy.clone(); + super::convert_filter_to_sql(&mut converted).unwrap(); + assert_eq!(converted.stream_name, "sql"); + let once = converted.clone(); + super::convert_filter_to_sql(&mut converted).unwrap(); + assert_eq!(converted, once); + converted.stream_name = legacy.stream_name.clone(); + assert_eq!(converted, legacy); + for kind in [FilterType::Filter, FilterType::Search] { + converted.query.filter_type = kind; + let unchanged = converted.clone(); + assert!(super::convert_filter_to_sql(&mut converted).is_err()); + assert_eq!(converted, unchanged); + } + } + #[test] fn accepts_flat_stream_names() { for stream_name in ["team1", "orders-prod", "user_events_v2"] { - let mut stream_name = stream_name.to_owned(); - assert!(validate_stream_name(&mut stream_name, &FilterType::Filter).is_ok()); + let stream_name = stream_name.to_owned(); + assert!(validate_stream_name(&stream_name, &FilterType::Filter).is_ok()); } } #[test] - fn normalizes_sql_filter_stream_name() { - let mut stream_name = "../../unsafe-path".to_owned(); - - assert!(validate_stream_name(&mut stream_name, &FilterType::SQL).is_ok()); - assert_eq!(stream_name, SQL_FILTER_STREAM_NAME); + fn accepts_legacy_and_canonical_sql_filter_stream_names() { + for stream_name in ["orders-prod", SQL_FILTER_STREAM_NAME] { + assert!(validate_stream_name(stream_name, &FilterType::SQL).is_ok()); + } + for stream_name in ["../../unsafe-path", "team1/serviceA", "", "sql/../orders"] { + assert!(validate_stream_name(stream_name, &FilterType::SQL).is_err()); + } + assert!(validate_stream_name(SQL_FILTER_STREAM_NAME, &FilterType::Search).is_err()); } #[test] fn rejects_path_like_stream_names() { - let mut stream_name = "team1/serviceA".to_owned(); - let err = validate_stream_name(&mut stream_name, &FilterType::Search) + let stream_name = "team1/serviceA".to_owned(); + let err = validate_stream_name(&stream_name, &FilterType::Search) .expect_err("path-like names must be rejected"); assert!(matches!( err, diff --git a/src/metastore/metastore_traits.rs b/src/metastore/metastore_traits.rs index 267428d7a..4bd6c9415 100644 --- a/src/metastore/metastore_traits.rs +++ b/src/metastore/metastore_traits.rs @@ -192,9 +192,15 @@ pub trait Metastore: std::fmt::Debug + Send + Sync { obj: &dyn MetastoreObject, tenant_id: &Option, ) -> Result<(), MetastoreError>; + /// Remove obsolete copies after the destination has been durably written. + async fn reconcile_filter( + &self, + filter: &Filter, + tenant_id: &Option, + ) -> Result<(), MetastoreError>; async fn delete_filter( &self, - obj: &dyn MetastoreObject, + filter: &Filter, tenant_id: &Option, ) -> Result<(), MetastoreError>; diff --git a/src/metastore/metastores/object_store_metastore.rs b/src/metastore/metastores/object_store_metastore.rs index 867544601..3ebcece96 100644 --- a/src/metastore/metastores/object_store_metastore.rs +++ b/src/metastore/metastores/object_store_metastore.rs @@ -62,7 +62,7 @@ use crate::{ to_bytes, }, }, - users::filters::{Filter, migrate_v1_v2}, + users::filters::{Filter, FilterType, SQL_FILTER_STREAM_NAME, migrate_v1_v2}, }; /// Using PARSEABLE's storage as a metastore (default) @@ -71,6 +71,204 @@ pub struct ObjectStoreMetastore { pub storage: Arc, } +impl ObjectStoreMetastore { + async fn delete_filter_path( + &self, + path: &RelativePathBuf, + tenant_id: &Option, + ) -> Result<(), MetastoreError> { + match self.storage.delete_object(path, tenant_id).await { + Ok(()) => Ok(()), + Err(error) if is_missing_optional_dir(&error) => Ok(()), + Err(error) => Err(error.into()), + } + } + + /// Enumerate only this owner's ID, never a filter belonging to another user. + async fn filter_aliases( + &self, + filter: &Filter, + tenant_id: &Option, + ) -> Result, MetastoreError> { + let owner = filter + .user_id + .as_deref() + .expect("stored filters have an owner"); + let id = filter + .filter_id + .as_deref() + .expect("stored filters have an ID"); + let root = filter_path(owner, "", "", tenant_id); + let mut copies = Vec::new(); + for stream in self.storage.list_dirs_relative(&root, tenant_id).await? { + let path = root.join(&stream).join(format!("{id}.json")); + let bytes = match self.storage.get_object(&path, tenant_id).await { + Ok(bytes) => bytes, + Err(error) if is_missing_optional_dir(&error) => continue, + Err(error) => return Err(error.into()), + }; + let copy: Filter = serde_json::from_slice(&bytes)?; + if copy.filter_id == filter.filter_id && copy.user_id == filter.user_id { + copies.push((path, copy)); + } + } + copies.sort_by(|a, b| a.0.cmp(&b.0)); + Ok(copies) + } + + async fn load_tenant_filters( + &self, + tenant_id: &Option, + ) -> Result, MetastoreError> { + let users_dir = + RelativePathBuf::from_iter([tenant_id.as_deref().unwrap_or(""), USERS_ROOT_DIR]); + let mut grouped: BTreeMap<(String, String), Vec<(RelativePathBuf, Filter)>> = + BTreeMap::new(); + for user in self + .storage + .list_dirs_relative(&users_dir, tenant_id) + .await? + { + let root = users_dir.join(&user).join("filters"); + for stream in self.storage.list_dirs_relative(&root, tenant_id).await? { + let directory = root.join(&stream); + let objects = self + .storage + .get_objects( + Some(&directory), + Box::new(|name| name.ends_with(".json")), + tenant_id, + ) + .await?; + for bytes in objects { + let mut value: serde_json::Value = serde_json::from_slice(&bytes)?; + let is_v1 = value.get("version").and_then(|v| v.as_str()) == Some("v1"); + if is_v1 { + value = migrate_v1_v2(value, tenant_id); + } + let Ok(mut filter) = serde_json::from_value::(value) else { + continue; + }; + // The discovered tenant directory is authoritative, including + // default-tenant records whose payload predates tenant metadata. + filter.tenant_id = tenant_id.clone().filter(|tenant| !tenant.is_empty()); + let (Some(owner), Some(id)) = (&filter.user_id, &filter.filter_id) else { + continue; + }; + // Stored identity fields must remain single path components. + if !safe_filter_component(owner) + || !safe_filter_component(id) + || !safe_filter_component(&filter.stream_name) + || filter.stream_name != stream + || (!is_v1 && owner != &user) + { + warn!( + "Ignoring saved filter whose identity does not match its storage directory" + ); + continue; + } + let source = directory.join(format!("{id}.json")); + // get_objects returns bytes without filenames. Verify the inferred + // filename before any cleanup; never delete an unrelated object. + match self.storage.get_object(&source, tenant_id).await { + Ok(stored) if stored == bytes => {} + Ok(_) => continue, + Err(error) if is_missing_optional_dir(&error) => continue, + Err(error) => return Err(error.into()), + } + if is_v1 { + // Persist before removing the original so interrupted migration is safe. + self.put_filter(&filter, tenant_id).await?; + if source.as_str() != filter.get_object_path() { + self.delete_filter_path(&source, tenant_id).await?; + } + } + grouped + .entry((owner.clone(), id.clone())) + .or_default() + .push(( + if is_v1 { + RelativePathBuf::from(filter.get_object_path()) + } else { + source + }, + filter, + )); + } + } + } + let mut filters = Vec::new(); + for (_, mut copies) in grouped { + // A v1 migration may write a hashed-owner directory that is visited + // later in this same scan. It is still one physical object. + copies.sort_by(|a, b| a.0.cmp(&b.0)); + copies.dedup_by(|a, b| a.0 == b.0); + copies.sort_by(|a, b| { + let canonical = |filter: &Filter| { + filter.query.filter_type == FilterType::SQL + && filter.stream_name == SQL_FILTER_STREAM_NAME + }; + canonical(&b.1).cmp(&canonical(&a.1)).then(a.0.cmp(&b.0)) + }); + let (canonical_path, filter) = &copies[0]; + if filter.query.filter_type == FilterType::SQL + && filter.stream_name == SQL_FILTER_STREAM_NAME + { + for (path, alias) in copies.iter().skip(1) { + if path != canonical_path && alias.query.filter_type == FilterType::SQL { + // The canonical object is already durable. Retain it in memory + // even if cleanup is temporarily unavailable; retry on restart. + if let Err(error) = self.delete_filter_path(path, tenant_id).await { + warn!("Failed to clean saved SQL filter alias {path}: {error}"); + } + } else if alias.query.filter_type != FilterType::SQL { + warn!( + "Retaining non-SQL saved filter copy {path} with the same owner {:?} and ID {:?} as canonical SQL filter {canonical_path}", + filter.user_id, filter.filter_id + ); + } + } + } + if copies.len() > 1 + && !(filter.query.filter_type == FilterType::SQL + && filter.stream_name == SQL_FILTER_STREAM_NAME) + { + warn!( + "Multiple saved filter copies for owner {:?}, ID {:?}; using deterministic path {canonical_path}", + filter.user_id, filter.filter_id + ); + } + filters.push(filter.clone()); + } + Ok(filters) + } +} + +fn stored_filter_path(filter: &Filter, tenant_id: &Option) -> RelativePathBuf { + filter_path( + filter + .user_id + .as_deref() + .expect("stored filters have an owner"), + &filter.stream_name, + &format!( + "{}.json", + filter + .filter_id + .as_deref() + .expect("stored filters have an ID") + ), + tenant_id, + ) +} + +fn safe_filter_component(component: &str) -> bool { + !component.is_empty() + && component != "." + && component != ".." + && !component.contains(['/', '\\']) +} + pub fn is_missing_optional_dir(err: &ObjectStorageError) -> bool { match err { ObjectStorageError::NoSuchKey(_) => true, @@ -629,113 +827,23 @@ impl Metastore for ObjectStoreMetastore { .await?) } - // for get filters, take care of migration and removal of incorrect/old filters - // return deserialized filter + // Reconcile duplicate SQL paths before exposing filters to the cache. async fn get_filters(&self) -> Result>, MetastoreError> { - let mut this = HashMap::new(); - let base_paths = PARSEABLE.list_tenants().unwrap_or_else(|| vec!["".into()]); - - for mut tenant in base_paths { - let users_dir = RelativePathBuf::from_iter([&tenant, USERS_ROOT_DIR]); - let tenant_id = &Some(tenant.clone()); - let mut filters = Vec::new(); - for user in self - .storage - .list_dirs_relative(&users_dir, tenant_id) - .await? - { - let stream_dir = users_dir.join(&user).join("filters"); - - for stream in self - .storage - .list_dirs_relative(&stream_dir, tenant_id) - .await? - { - let filters_path = stream_dir.join(&stream); - - // read filter object - let filter_bytes = self - .storage - .get_objects( - Some(&filters_path), - Box::new(|file_name| file_name.ends_with(".json")), - tenant_id, - ) - .await?; - - for filter in filter_bytes { - // deserialize into Value - let mut filter_value = - serde_json::from_slice::(&filter)?; - - if let Some(meta) = filter_value.clone().as_object() { - let version = meta.get("version").and_then(|version| version.as_str()); - - if version == Some("v1") { - // delete older version of the filter - // get filter id to delete - let filterid = meta - .get("filter_id") - .and_then(|filter_id| filter_id.as_str()); - if let Some(filterid) = filterid { - self.storage - .delete_object( - &filters_path.join(format!("{filterid}.json")), - tenant_id, - ) - .await?; - } - // self.storage.delete_object(&filters_path, tenant_id).await?; - - filter_value = migrate_v1_v2(filter_value, tenant_id); - let user_id = filter_value - .as_object() - .unwrap() - .get("user_id") - .and_then(|user_id| user_id.as_str()); - let filter_id = filter_value - .as_object() - .unwrap() - .get("filter_id") - .and_then(|filter_id| filter_id.as_str()); - let stream_name = filter_value - .as_object() - .unwrap() - .get("stream_name") - .and_then(|stream_name| stream_name.as_str()); - - // if these values are present, create a new file - if let (Some(user_id), Some(stream_name), Some(filter_id)) = - (user_id, stream_name, filter_id) - { - let path = filter_path( - user_id, - stream_name, - &format!("{filter_id}.json"), - tenant_id, - ); - let filter_bytes = to_bytes(&filter_value); - self.storage - .put_object(&path, filter_bytes.clone(), tenant_id) - .await?; - } - } - - if let Ok(filter) = serde_json::from_value::(filter_value) { - filters.retain(|f: &Filter| f.filter_id != filter.filter_id); - filters.push(filter); - } - } - } - } - } - if tenant.is_empty() { - tenant.clone_from(&DEFAULT_TENANT.to_string()); - } - this.insert(tenant, filters); + let mut all = HashMap::new(); + let tenants = PARSEABLE.list_tenants().unwrap_or_else(|| vec!["".into()]); + for tenant in tenants { + let tenant_id = Some(tenant.clone()); + let filters = self.load_tenant_filters(&tenant_id).await?; + all.insert( + if tenant.is_empty() { + DEFAULT_TENANT.to_owned() + } else { + tenant + }, + filters, + ); } - - Ok(this) + Ok(all) } /// Save a filter @@ -753,18 +861,30 @@ impl Metastore for ObjectStoreMetastore { .await?) } - /// Delete a filter - async fn delete_filter( + async fn reconcile_filter( &self, - obj: &dyn MetastoreObject, + filter: &Filter, tenant_id: &Option, ) -> Result<(), MetastoreError> { - let path = obj.get_object_path(); + let canonical = stored_filter_path(filter, tenant_id); + for (path, _) in self.filter_aliases(filter, tenant_id).await? { + if path != canonical { + self.delete_filter_path(&path, tenant_id).await?; + } + } + Ok(()) + } - Ok(self - .storage - .delete_object(&RelativePathBuf::from(path), tenant_id) - .await?) + /// Delete aliases first. If any cleanup fails, the canonical copy remains + /// available for a retry instead of allowing an old alias to reappear. + async fn delete_filter( + &self, + filter: &Filter, + tenant_id: &Option, + ) -> Result<(), MetastoreError> { + self.reconcile_filter(filter, tenant_id).await?; + self.delete_filter_path(&stored_filter_path(filter, tenant_id), tenant_id) + .await } /// Fetch an `ObjectStoreFormat` file @@ -1408,6 +1528,290 @@ mod tests { } } + fn saved_sql(owner: &str, id: &str, stream: &str, tenant: &Option) -> Filter { + serde_json::from_value(serde_json::json!({ + "version": "v2", "user_id": owner, "filter_id": id, + "stream_name": stream, "filter_name": "Revenue", "tenant_id": tenant, + "query": {"filter_type": "sql", "filter_query": "select * from orders", "filter_builder": null}, + "time_filter": null, "explorer_state": {"columns": ["amount"]}, + "future_metadata": {"preserve": true} + })).unwrap() + } + + #[tokio::test] + async fn saved_filter_conversion_persists_and_delete_cannot_resurrect() { + for tenant in [None, Some("tenant-a".to_owned())] { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let legacy = saved_sql("owner", "same-id", "orders", &tenant); + store.put_filter(&legacy, &tenant).await.unwrap(); + // Sharing an ID with a different owner or tenant cannot remove their filter. + let other_owner = saved_sql("other-owner", "same-id", "orders", &tenant); + store.put_filter(&other_owner, &tenant).await.unwrap(); + let mut other_id = saved_sql("owner", "different-id", "orders", &tenant); + other_id.query.filter_type = FilterType::Search; + store.put_filter(&other_id, &tenant).await.unwrap(); + let other_tenant = saved_sql("owner", "same-id", "orders", &Some("tenant-b".into())); + store + .put_filter(&other_tenant, &other_tenant.tenant_id) + .await + .unwrap(); + let mut canonical = legacy.clone(); + canonical.stream_name = "sql".into(); + for _ in 0..2 { + store.put_filter(&canonical, &tenant).await.unwrap(); + store.reconcile_filter(&canonical, &tenant).await.unwrap(); + let bytes = storage + .get_object(&RelativePathBuf::from(canonical.get_object_path()), &tenant) + .await + .unwrap(); + assert_eq!(serde_json::from_slice::(&bytes).unwrap(), canonical); + } + assert!( + storage + .get_object(&RelativePathBuf::from(legacy.get_object_path()), &tenant) + .await + .is_err() + ); + let restarted = ObjectStoreMetastore { + storage: storage.clone(), + }; + let loaded = restarted.load_tenant_filters(&tenant).await.unwrap(); + assert_eq!(loaded.len(), 3); + assert!(loaded.contains(&canonical)); + // Simulate a stale legacy alias left by an interrupted earlier conversion. + restarted.put_filter(&legacy, &tenant).await.unwrap(); + restarted.delete_filter(&canonical, &tenant).await.unwrap(); + restarted.delete_filter(&canonical, &tenant).await.unwrap(); + let loaded = restarted.load_tenant_filters(&tenant).await.unwrap(); + assert_eq!(loaded.len(), 2); + assert!(loaded.contains(&other_owner)); + assert!(loaded.contains(&other_id)); + assert_eq!( + restarted + .load_tenant_filters(&other_tenant.tenant_id) + .await + .unwrap(), + vec![other_tenant] + ); + } + } + + #[tokio::test] + async fn saved_filter_startup_prefers_sql_and_keeps_untouched_legacy() { + for sql_first in [true, false] { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = None; + let old = saved_sql("owner", "converted", "orders", &tenant); + let mut canonical = old.clone(); + canonical.stream_name = "sql".into(); + canonical.filter_name = "Latest name".into(); + let untouched = saved_sql("owner", "untouched", "orders", &tenant); + for filter in if sql_first { + [&canonical, &old] + } else { + [&old, &canonical] + } { + store.put_filter(filter, &tenant).await.unwrap(); + } + store.put_filter(&untouched, &tenant).await.unwrap(); + let loaded = store.load_tenant_filters(&tenant).await.unwrap(); + assert_eq!(loaded.len(), 2); + assert!(loaded.contains(&canonical)); + assert!(loaded.contains(&untouched)); + assert!( + storage + .get_object(&RelativePathBuf::from(old.get_object_path()), &tenant) + .await + .is_err() + ); + assert_eq!(store.load_tenant_filters(&tenant).await.unwrap(), loaded); + } + } + + #[tokio::test] + async fn saved_filter_startup_uses_directory_tenant_for_null_payload() { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = Some("tenant-a".to_owned()); + let legacy = saved_sql("owner", "id", "orders", &tenant); + let mut payload = serde_json::to_value(&legacy).unwrap(); + payload["tenant_id"] = serde_json::Value::Null; + let legacy_path = RelativePathBuf::from(legacy.get_object_path()); + storage + .put_object(&legacy_path, to_bytes(&payload), &tenant) + .await + .unwrap(); + let loaded = store.load_tenant_filters(&tenant).await.unwrap(); + assert_eq!(loaded, vec![legacy.clone()]); + let mut canonical = loaded[0].clone(); + canonical.stream_name = "sql".into(); + store.put_filter(&canonical, &tenant).await.unwrap(); + store.reconcile_filter(&canonical, &tenant).await.unwrap(); + assert!(storage.get_object(&legacy_path, &tenant).await.is_err()); + assert_eq!( + store.load_tenant_filters(&tenant).await.unwrap(), + vec![canonical.clone()] + ); + store.delete_filter(&canonical, &tenant).await.unwrap(); + assert!(store.load_tenant_filters(&tenant).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn saved_filter_startup_retains_non_sql_copy_with_same_identity() { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = None; + let canonical = saved_sql("owner", "id", "sql", &tenant); + let mut non_sql = saved_sql("owner", "id", "orders", &tenant); + non_sql.query.filter_type = FilterType::Search; + store.put_filter(&canonical, &tenant).await.unwrap(); + store.put_filter(&non_sql, &tenant).await.unwrap(); + for _ in 0..2 { + assert_eq!( + store.load_tenant_filters(&tenant).await.unwrap(), + vec![canonical.clone()] + ); + let bytes = storage + .get_object(&RelativePathBuf::from(non_sql.get_object_path()), &tenant) + .await + .unwrap(); + assert_eq!(serde_json::from_slice::(&bytes).unwrap(), non_sql); + } + } + + #[tokio::test] + async fn saved_filter_failed_cleanup_keeps_canonical_for_retry() { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = None; + let legacy = saved_sql("owner", "id", "orders", &tenant); + let mut canonical = legacy.clone(); + canonical.stream_name = "sql".into(); + store.put_filter(&legacy, &tenant).await.unwrap(); + store.put_filter(&canonical, &tenant).await.unwrap(); + let legacy_path = RelativePathBuf::from(legacy.get_object_path()); + // An unreadable alias causes cleanup to fail after the canonical write. + storage + .put_object(&legacy_path, Bytes::from_static(b"invalid json"), &tenant) + .await + .unwrap(); + assert!(store.reconcile_filter(&canonical, &tenant).await.is_err()); + assert!(store.delete_filter(&canonical, &tenant).await.is_err()); + let canonical_path = RelativePathBuf::from(canonical.get_object_path()); + let bytes = storage.get_object(&canonical_path, &tenant).await.unwrap(); + assert_eq!(serde_json::from_slice::(&bytes).unwrap(), canonical); + // Once the transient alias failure is repaired, both retries are safe. + store.put_filter(&legacy, &tenant).await.unwrap(); + store.reconcile_filter(&canonical, &tenant).await.unwrap(); + store.reconcile_filter(&canonical, &tenant).await.unwrap(); + assert_eq!( + store.load_tenant_filters(&tenant).await.unwrap(), + vec![canonical.clone()] + ); + store.delete_filter(&canonical, &tenant).await.unwrap(); + assert!(store.load_tenant_filters(&tenant).await.unwrap().is_empty()); + } + + #[cfg(unix)] + #[tokio::test] + async fn saved_filter_delete_failure_reconciles_on_restart_before_delete_retry() { + use std::os::unix::fs::PermissionsExt; + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = None; + let legacy = saved_sql("owner", "id", "orders", &tenant); + let mut canonical = legacy.clone(); + canonical.stream_name = "sql".into(); + store.put_filter(&legacy, &tenant).await.unwrap(); + store.put_filter(&canonical, &tenant).await.unwrap(); + let directory = temp.path().join( + RelativePathBuf::from(legacy.get_object_path()) + .parent() + .unwrap() + .as_str(), + ); + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o555)).unwrap(); + // Root or unusual filesystems may not enforce write permissions. + if std::fs::write(directory.join("permission-probe"), b"probe").is_ok() { + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o755)).unwrap(); + return; + } + let cleanup = store.reconcile_filter(&canonical, &tenant).await; + let deletion = store.delete_filter(&canonical, &tenant).await; + let restart = store.load_tenant_filters(&tenant).await; + let canonical_exists = storage + .get_object(&RelativePathBuf::from(canonical.get_object_path()), &tenant) + .await + .is_ok(); + std::fs::set_permissions(&directory, std::fs::Permissions::from_mode(0o755)).unwrap(); + assert!(cleanup.is_err()); + assert!(deletion.is_err()); + assert!(canonical_exists); + assert_eq!(restart.unwrap(), vec![canonical.clone()]); + assert_eq!( + store.load_tenant_filters(&tenant).await.unwrap(), + vec![canonical.clone()] + ); + assert!( + storage + .get_object(&RelativePathBuf::from(legacy.get_object_path()), &tenant) + .await + .is_err() + ); + store.delete_filter(&canonical, &tenant).await.unwrap(); + assert!(store.load_tenant_filters(&tenant).await.unwrap().is_empty()); + } + + #[tokio::test] + async fn saved_filter_legacy_duplicates_are_deterministic_and_not_deleted() { + let temp = tempfile::tempdir().unwrap(); + let storage = Arc::new(LocalFS::new(temp.path().to_path_buf())); + let store = ObjectStoreMetastore { + storage: storage.clone(), + }; + let tenant = None; + let a = saved_sql("owner", "id", "a", &tenant); + let z = saved_sql("owner", "id", "z", &tenant); + store.put_filter(&z, &tenant).await.unwrap(); + store.put_filter(&a, &tenant).await.unwrap(); + assert_eq!( + store.load_tenant_filters(&tenant).await.unwrap(), + vec![a.clone()] + ); + assert!( + storage + .get_object(&RelativePathBuf::from(z.get_object_path()), &tenant) + .await + .is_ok() + ); + assert!( + storage + .get_object(&RelativePathBuf::from(a.get_object_path()), &tenant) + .await + .is_ok() + ); + } + #[tokio::test] async fn delete_alert_uses_request_tenant_path_and_is_idempotent() { let temp_dir = tempfile::tempdir().unwrap(); diff --git a/src/rbac/role.rs b/src/rbac/role.rs index d682b1108..4520e57f0 100644 --- a/src/rbac/role.rs +++ b/src/rbac/role.rs @@ -70,6 +70,7 @@ pub enum Action { ListFilter, GetFilter, CreateFilter, + UpdateFilter, DeleteFilter, Login, Metrics, @@ -139,6 +140,7 @@ impl RoleBuilder { | Action::GetFilter | Action::ListFilter | Action::CreateFilter + | Action::UpdateFilter | Action::DeleteFilter | Action::PutAlert | Action::GetAlert @@ -360,6 +362,7 @@ pub mod model { Action::QueryLLM, Action::ListLLM, Action::CreateFilter, + Action::UpdateFilter, Action::ListFilter, Action::GetFilter, Action::DeleteFilter, @@ -403,6 +406,7 @@ pub mod model { Action::GetFilter, Action::ListFilter, Action::CreateFilter, + Action::UpdateFilter, Action::DeleteFilter, Action::GetUserRoles, Action::PatchUser, @@ -426,6 +430,7 @@ pub mod model { Action::ListFilter, Action::GetFilter, Action::CreateFilter, + Action::UpdateFilter, Action::DeleteFilter, Action::ListDashboard, Action::GetDashboard, @@ -449,3 +454,23 @@ pub mod model { } } } + +#[cfg(test)] +mod saved_filter_permission_tests { + use super::{Action, Permission, RoleBuilder, model::DefaultPrivilege}; + + #[test] + fn saved_filter_update_is_granted_to_existing_filter_roles() { + for privilege in [ + DefaultPrivilege::Editor, + DefaultPrivilege::Writer { resource: None }, + DefaultPrivilege::Reader { resource: None }, + ] { + let permissions = RoleBuilder::from(&privilege).build(); + assert!(permissions.contains(&Permission::Unit(Action::CreateFilter))); + assert!(permissions.contains(&Permission::Unit(Action::UpdateFilter))); + } + let permissions = RoleBuilder::from(&DefaultPrivilege::Ingestor { resource: None }).build(); + assert!(!permissions.contains(&Permission::Unit(Action::UpdateFilter))); + } +} diff --git a/src/users/filters.rs b/src/users/filters.rs index 1817cbd20..66abae66a 100644 --- a/src/users/filters.rs +++ b/src/users/filters.rs @@ -21,7 +21,7 @@ use std::collections::HashMap; use once_cell::sync::Lazy; use serde::{Deserialize, Serialize}; use serde_json::Value; -use tokio::sync::RwLock; +use tokio::sync::{Mutex, RwLock}; use super::TimeFilter; use crate::{ @@ -33,6 +33,9 @@ use crate::{ }; pub static FILTERS: Lazy = Lazy::new(Filters::default); +/// Serializes storage relocation and cache mutation within this server. +pub static FILTER_MUTATION_LOCK: Mutex<()> = Mutex::const_new(()); +pub const SQL_FILTER_STREAM_NAME: &str = "sql"; pub const CURRENT_FILTER_VERSION: &str = "v2"; #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] pub struct Filter { @@ -117,12 +120,9 @@ pub struct Filters(RwLock>>); impl Filters { pub async fn load(&self) -> anyhow::Result<()> { - let all_filters = PARSEABLE.metastore.get_filters().await.unwrap_or_default(); - - let mut s = self.0.write().await; - for (tenant_id, filters) in all_filters { - s.entry(tenant_id).or_default().extend(filters); - } + let _mutation = FILTER_MUTATION_LOCK.lock().await; + let all_filters = PARSEABLE.metastore.get_filters().await?; + *self.0.write().await = all_filters; Ok(()) } @@ -131,15 +131,15 @@ impl Filters { let mut s = self.0.write().await; let tenant = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT).to_owned(); let filters = s.entry(tenant).or_default(); - filters.retain(|f| f.filter_id != filter.filter_id); + filters.retain(|f| f.filter_id != filter.filter_id || f.user_id != filter.user_id); filters.push(filter.clone()); } - pub async fn delete_filter(&self, filter_id: &str, tenant_id: &Option) { + pub async fn delete_filter(&self, filter: &Filter, tenant_id: &Option) { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); let mut s = self.0.write().await; if let Some(filters) = s.get_mut(tenant_id) { - filters.retain(|f| f.filter_id != Some(filter_id.to_string())); + filters.retain(|f| f.filter_id != filter.filter_id || f.user_id != filter.user_id); } } @@ -152,13 +152,17 @@ impl Filters { ) -> Option { let tenant_id = tenant_id.as_deref().unwrap_or(DEFAULT_TENANT); if let Some(filters) = self.0.read().await.get(tenant_id) { - filters - .iter() - .find(|f| { - f.filter_id == Some(filter_id.to_string()) - && (f.user_id == Some(user_id.to_string()) || is_admin) - }) - .cloned() + let mut matches = filters.iter().filter(|f| { + f.filter_id.as_deref() == Some(filter_id) + && (f.user_id.as_deref() == Some(user_id) || is_admin) + }); + let filter = matches.next()?; + // IDs are normally globally unique. Fail closed for an administrator + // if imported records make this owner-less URL ambiguous. + if matches.next().is_some() { + return None; + } + Some(filter.clone()) } else { None } @@ -222,3 +226,64 @@ pub fn migrate_v1_v2(mut filter_meta: Value, tenant_id: &Option) -> Valu filter_meta } + +#[cfg(test)] +mod tests { + use super::*; + + fn saved_filter(owner: &str, tenant: &Option) -> Filter { + serde_json::from_value(serde_json::json!({ + "version":"v2", "user_id":owner, "filter_id":"shared-id", + "stream_name":"orders", "filter_name":"View", "tenant_id":tenant, + "query":{"filter_type":"sql","filter_query":"select * from orders","filter_builder":null}, + "time_filter":null + })).unwrap() + } + + #[tokio::test] + async fn saved_filter_cache_enforces_owner_tenant_and_preserves_same_id_other_owner() { + let cache = Filters::default(); + let tenant = Some("tenant".into()); + let owner = saved_filter("owner", &tenant); + let other = saved_filter("other", &tenant); + cache.update(&owner, &tenant).await; + assert!( + cache + .get_filter("shared-id", "intruder", false, &tenant) + .await + .is_none() + ); + assert!( + cache + .get_filter("shared-id", "owner", false, &None) + .await + .is_none() + ); + assert_eq!( + cache + .get_filter("shared-id", "administrator", true, &tenant) + .await, + Some(owner.clone()) + ); + cache.update(&other, &tenant).await; + assert_eq!( + cache.get_filter("shared-id", "owner", false, &tenant).await, + Some(owner.clone()) + ); + assert_eq!( + cache.get_filter("shared-id", "other", false, &tenant).await, + Some(other.clone()) + ); + assert!( + cache + .get_filter("shared-id", "administrator", true, &tenant) + .await + .is_none() + ); + cache.delete_filter(&owner, &tenant).await; + assert_eq!( + cache.get_filter("shared-id", "other", false, &tenant).await, + Some(other) + ); + } +} From 639597db67112b79f3bde607de5f4a4af9d28376 Mon Sep 17 00:00:00 2001 From: VibeSWE <338779468+VibeSWE@users.noreply.github.com> Date: Fri, 16 Oct 2026 11:25:24 -0500 Subject: [PATCH 09/12] Keep both saved-view formats usable and add explicit SQL conversion --- frontend/README.md | 2 +- frontend/docs/api-contracts.md | 2 +- frontend/e2e/logs-sql-view-conversion.spec.ts | 218 ++++++++++++++++++ frontend/src/features/logs/SavedViews.test.ts | 46 +++- frontend/src/features/logs/SavedViews.tsx | 68 ++++-- frontend/src/lib/client.ts | 2 + frontend/src/lib/demoLogs.ts | 19 +- frontend/src/lib/logsContract.test.ts | 32 ++- frontend/src/lib/types.ts | 1 + 9 files changed, 365 insertions(+), 25 deletions(-) create mode 100644 frontend/e2e/logs-sql-view-conversion.spec.ts diff --git a/frontend/README.md b/frontend/README.md index 33fce92ff..677c5b101 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -74,7 +74,7 @@ This creates `build.zip` with a top-level `dist/` directory, including `dist/ind - Keystone and LLM settings: native capability pages with explicit limitations and deployment-dependent classic links. See [native screen scope](docs/prism-parity.md#native-missing-screen-scope-2026-10-12). - Component gallery at `/components`, plus isolated Storybook stories for UI, PromQL, charts, Traces, APM and ingestion setup. -The default log query reads at most 100 rows, with choices up to 10,000. Message search assumes a `message` field. The histogram separately aggregates every matching event through `/query`, using the same captured bounds and predicates as the rows. Its 48 buckets divide elapsed time evenly; they are not calendar-day buckets. AND/OR filters have two levels, at most 12 groups and 50 rules, and support `=`/`!=`; SQL NULL values are excluded by both operators. Saved log views use server `/filters` CRUD, while copy links encode versioned explorer state with fixed absolute bounds. Context uses the existing `/query/context` endpoint with a 101-event bound and requires `p_timestamp` plus a string `message`, `body` or `log`. Large/wide results virtualize rows and columns; sorting and export cover loaded rows only. Wrapped virtual rows show a 96px preview; event details retain the full value. Timezones change display and wall-clock editing while requests retain UTC instants; DST gaps are rejected and repeated times require an earlier/later choice. Demo SQL intentionally supports only a small SELECT subset and rejects unsupported syntax. Full SQL requires a server. SQL multi-dataset generation uses `UNION ALL BY NAME`, fills missing columns with NULL and requires compatible types for common columns, and keeps the 100-row default; demo does not support unions. Incoming query links open or focus a tab while preserving other drafts. Reopening an already-open saved query focuses that tab and preserves its edits. The saved SQL library uses persistent browser local storage, while draft workspaces use session storage independent to each browser tab and survive navigation/reload. Both are scoped to the server and signed-in user (with a separate demo scope); they are not server-synced, and result rows are not persisted. Query handoff parameters are consumed after opening the tab so reloading an edited draft does not reopen the original query. SQL chart previews use the executed query and captured time bounds. Adding a chart to an existing dashboard preserves that dashboard’s time range; a new dashboard starts with the executed range. +The default log query reads at most 100 rows, with choices up to 10,000. Message search assumes a `message` field. The histogram separately aggregates every matching event through `/query`, using the same captured bounds and predicates as the rows. Its 48 buckets divide elapsed time evenly; they are not calendar-day buckets. AND/OR filters have two levels, at most 12 groups and 50 rules, and support `=`/`!=`; SQL NULL values are excluded by both operators. Saved log views use server `/filters` CRUD and support dataset-scoped and canonical SQL views. New saves stay dataset scoped; owners can explicitly convert compatible views to SQL while retaining their query and dataset. Copy links encode versioned explorer state with fixed absolute bounds. Context uses the existing `/query/context` endpoint with a 101-event bound and requires `p_timestamp` plus a string `message`, `body` or `log`. Large/wide results virtualize rows and columns; sorting and export cover loaded rows only. Wrapped virtual rows show a 96px preview; event details retain the full value. Timezones change display and wall-clock editing while requests retain UTC instants; DST gaps are rejected and repeated times require an earlier/later choice. Demo SQL intentionally supports only a small SELECT subset and rejects unsupported syntax. Full SQL requires a server. SQL multi-dataset generation uses `UNION ALL BY NAME`, fills missing columns with NULL and requires compatible types for common columns, and keeps the 100-row default; demo does not support unions. Incoming query links open or focus a tab while preserving other drafts. Reopening an already-open saved query focuses that tab and preserves its edits. The saved SQL library uses persistent browser local storage, while draft workspaces use session storage independent to each browser tab and survive navigation/reload. Both are scoped to the server and signed-in user (with a separate demo scope); they are not server-synced, and result rows are not persisted. Query handoff parameters are consumed after opening the tab so reloading an edited draft does not reopen the original query. SQL chart previews use the executed query and captured time bounds. Adding a chart to an existing dashboard preserves that dashboard’s time range; a new dashboard starts with the executed range. Logs implementation keeps persisted settings in `logViewReducer.ts`, with explicit dataset reset, saved-view and classic-link transitions. `logFilters.ts` owns expression limits and additions; adding within a group and narrowing the whole expression remain distinct operations. `logQuery.ts` compiles canonical criteria once for the page's row and histogram queries. Changes to client-only filter IDs or empty groups do not refresh the captured time window. Histogram responses carry their bucket geometry to the renderer. Both table modes use `logTable.tsx` for sorting, labels and cells. `logsContract.ts` validates server payloads; `demoLogs.ts` implements the demo methods separately. Saved-view instants, server timestamps and timezone wall-clock inputs retain separate validators because their accepted formats differ. diff --git a/frontend/docs/api-contracts.md b/frontend/docs/api-contracts.md index 0ae0cf18d..4de086ade 100644 --- a/frontend/docs/api-contracts.md +++ b/frontend/docs/api-contracts.md @@ -16,7 +16,7 @@ The client is React independent and uses the Rust handlers in this repository. ` Rows and histogram issue separate `POST /api/v1/query` requests with one captured UTC start/end pair and identical search/group predicates. Rows default to `LIMIT 100`, with selectable limits 1,000, 5,000 and 10,000; histogram SQL uses `DATE_BIN`, `COUNT(*)`, grouping and ordering without a result-row limit. An explicit half-open timestamp predicate enforces the same window for both reads. Dataset information selects the timestamp field. The chart fills 48 fixed elapsed-time buckets and distinguishes matching totals from loaded rows, with independent loading/error/retry states. `/counts` manifest statistics are not used for exact totals. Refresh captures new bounds; changing filters preserves the current window. Two-level AND/OR expressions allow 12 groups and 50 rules, escape identifiers/literals, ignore empty groups, and support `=`/`!=` with SQL NULL semantics (NULL is excluded). -Saved log views use existing user/tenant-authorized `GET/POST /api/v1/filters` and `PUT/DELETE /api/v1/filters/{encoded id}` through `listLogViews`, `createLogView`, `updateLogView` and `deleteLogView`. Documents keep `stream_name`, `filter_name`, SQL `query {filter_type:"sql",filter_query,filter_builder:null}`, optional classic `time_filter {from,to}`, and flattened `explorer_state` containing validated version-1 JSON. State includes dataset, range, search, grouped filters, columns, timezone, result limit, wrapping and histogram visibility. Saved SQL uses the dataset timestamp partition rather than assuming `p_timestamp`. Updates preserve unrelated server metadata. Only verified owners can update/delete a view; authorized views owned by others are labeled read only and can still be applied or copied with Save new view. Administrator access does not bypass this UI ownership constraint because the backend update path reassigns ownership. Opening the dialog refetches the list so views created in another tab appear. Classic/incompatible views are labeled and cannot be applied or overwritten here. Failed CRUD remains visible, without browser-storage fallback. +Saved log views use existing user/tenant-authorized `GET/POST /api/v1/filters` and `PUT/DELETE /api/v1/filters/{encoded id}` through `listLogViews`, `createLogView`, `updateLogView` and `deleteLogView`. Documents keep `stream_name`, `filter_name`, SQL `query {filter_type:"sql",filter_query,filter_builder:null}`, optional classic `time_filter {from,to}`, and flattened `explorer_state` containing validated version-1 JSON. State includes dataset, range, search, grouped filters, columns, timezone, result limit, wrapping and histogram visibility. Saved SQL uses the dataset timestamp partition rather than assuming `p_timestamp`. New views retain dataset-scoped `stream_name` for gradual adoption. Compatible owned views offer **Convert to SQL view**, using a bodyless `POST /api/v1/filters/{encoded id}/convert-to-sql` through `convertLogViewToSql`. Conversion retains the ID, owner, query, explorer state and other metadata, and is idempotent. Canonical `stream_name:"sql"` views resolve their dataset from validated explorer state; malformed or unrelated canonical states do not appear in this dataset’s list. Updates preserve canonical SQL storage and unrelated server metadata. Only verified owners can update/delete/convert a view in this UI; authorized views owned by others are labeled read only and can still be applied or copied with Save new view. Backend authorization remains authoritative and also permits administrators to convert views. Opening the dialog refetches the list so views created in another tab appear. Classic/incompatible views are labeled and cannot be applied or overwritten here. Failed CRUD remains visible, without browser-storage fallback. Copy links put raw version-1 JSON in the URLSearchParams `logView` parameter with fixed absolute bounds. Validation bounds state at 64KiB, checks timestamp/timezone/limit/filter constraints, rejects duplicate identifiers/NUL characters and discards unrecognized keys. Clipboard feedback follows successful copying, with a selectable fallback for blocked clipboard access. The native `logView` state is consumed after applying it, preserving later edits on reload; classic `view` retains its saved-view-ID meaning. Classic Logs parameters remain supported; linked datasets remain subject to server authorization. diff --git a/frontend/e2e/logs-sql-view-conversion.spec.ts b/frontend/e2e/logs-sql-view-conversion.spec.ts new file mode 100644 index 000000000..9a05260d4 --- /dev/null +++ b/frontend/e2e/logs-sql-view-conversion.spec.ts @@ -0,0 +1,218 @@ +import { createHash } from 'node:crypto'; +import { expect, test, type Page } from '@playwright/test'; +import type { SavedLogView } from '../src/lib/types'; + +const owner = createHash('sha256').update('reader').digest('hex'); +const state = { + version: 1, + dataset: 'events', + range: { startTime: '2026-10-12T12:00:00Z', endTime: '2026-10-12T13:00:00Z' }, + search: 'saved search', + expression: { combinator: 'and', groups: [] }, + timezone: 'UTC', + limit: 100, + wrap: false, + showHistogram: true, +}; +const makeView = (filter_id: string, extra: Partial = {}): SavedLogView => ({ + filter_id, + filter_name: filter_id, + stream_name: 'events', + user_id: owner, + query: { filter_type: 'sql', filter_query: 'SELECT * FROM events', filter_builder: null }, + explorer_state: JSON.stringify(state), + custom: { pinned: true }, + ...extra, +}); +async function fixture( + page: Page, + views: SavedLogView[], + options: { conversionError?: boolean; waitForConversion?: Promise } = {}, +) { + const mutations: { method: string; path: string; body: unknown }[] = []; + await page + .context() + .addCookies([{ name: 'user_id', value: 'reader', url: 'http://127.0.0.1:5173' }]); + await page.route('**/api/v1/**', async (route) => { + const request = route.request(), + path = new URL(request.url()).pathname; + if (path.endsWith('/users/reader')) + return route.fulfill({ json: { id: 'reader', username: 'Reader' } }); + if (path.endsWith('/logstream')) + return route.fulfill({ json: [{ name: 'events' }, { name: 'other' }] }); + if (path.endsWith('/schema')) + return route.fulfill({ json: { fields: [{ name: 'p_timestamp' }, { name: 'message' }] } }); + if (path.endsWith('/info')) return route.fulfill({ json: { timePartition: 'p_timestamp' } }); + if (path.endsWith('/filters') && request.method() === 'GET') + return route.fulfill({ json: views }); + if (path.includes('/filters')) { + mutations.push({ + method: request.method(), + path, + body: request.postData() ? request.postDataJSON() : null, + }); + if (path.endsWith('/convert-to-sql')) { + await options.waitForConversion; + if (options.conversionError) + return route.fulfill({ status: 503, body: 'Conversion unavailable. Try again.' }); + const view = views.find((view) => view.filter_id === path.split('/').at(-2))!; + view.stream_name = 'sql'; + return route.fulfill({ json: view }); + } + if (request.method() === 'POST') { + const view = { ...request.postDataJSON(), filter_id: 'created', user_id: owner }; + views.push(view); + return route.fulfill({ json: view }); + } + if (request.method() === 'PUT') { + const id = path.split('/').at(-1), + index = views.findIndex((view) => view.filter_id === id); + views[index] = { ...request.postDataJSON(), filter_id: id, user_id: owner }; + return route.fulfill({ json: views[index] }); + } + } + if (path.endsWith('/query')) return route.fulfill({ json: [] }); + return route.fulfill({ json: {} }); + }); + await page.goto( + `/logs/explore/events?${new URLSearchParams({ logView: JSON.stringify(state) })}`, + ); + await page.getByRole('button', { name: 'Saved views', exact: true }).click(); + return { dialog: page.getByRole('dialog', { name: 'Saved log views' }), mutations }; +} + +test('new views remain dataset scoped until converted, reopen from SQL storage, and retain SQL on update', async ({ + page, +}) => { + const views: SavedLogView[] = []; + const { dialog, mutations } = await fixture(page, views); + await dialog.getByLabel('Log view name').fill('My saved view'); + await dialog.getByRole('button', { name: 'Save new view', exact: true }).click(); + await expect(dialog.getByText('View saved on the server.', { exact: true })).toBeVisible(); + expect(views[0].stream_name).toBe('events'); + const legacy = structuredClone(views[0]); + await dialog.getByLabel('Log view name').fill('Unsaved rename'); + await expect( + dialog.getByText( + 'Convert this saved view to SQL while keeping its existing query and dataset.', + ), + ).toBeVisible(); + await dialog.getByRole('button', { name: 'Convert to SQL view' }).click(); + await expect( + dialog.getByText('View converted to SQL. Its query and dataset were retained.'), + ).toBeVisible(); + expect(views[0]).toEqual({ ...legacy, stream_name: 'sql' }); + expect(mutations.at(-1)).toEqual({ + method: 'POST', + path: '/api/v1/filters/created/convert-to-sql', + body: null, + }); + await expect(dialog.getByLabel('Log view name')).toHaveValue('My saved view'); + await expect(dialog.getByRole('button', { name: 'Convert to SQL view' })).toBeDisabled(); + await page.reload(); + await page.getByRole('button', { name: 'Saved views', exact: true }).click(); + await dialog.getByLabel('Saved log view').selectOption('created'); + await dialog.getByRole('button', { name: 'Apply view', exact: true }).click(); + await expect(page.getByLabel('Search messages', { exact: true })).toHaveValue('saved search'); + await page.getByLabel('Search messages', { exact: true }).fill('updated search'); + await page.getByRole('button', { name: 'Saved views', exact: true }).click(); + await dialog.getByLabel('Saved log view').selectOption('created'); + await dialog.getByLabel('Log view name').fill('Updated SQL view'); + await dialog.getByRole('button', { name: 'Update selected' }).click(); + await expect(dialog.getByText('View saved on the server.', { exact: true })).toBeVisible(); + expect(views[0]).toMatchObject({ + stream_name: 'sql', + filter_id: legacy.filter_id, + user_id: owner, + filter_name: 'Updated SQL view', + }); + expect(JSON.parse(views[0].explorer_state!)).toMatchObject({ + dataset: 'events', + search: 'updated search', + }); +}); + +test('SQL views stay within their explorer dataset and read-only or incompatible views cannot convert', async ({ + page, +}) => { + const { dialog, mutations } = await fixture(page, [ + makeView('owned'), + makeView('shared', { user_id: 'someone-else' }), + makeView('canonical', { stream_name: 'sql' }), + makeView('other-canonical', { + stream_name: 'sql', + explorer_state: JSON.stringify({ ...state, dataset: 'other' }), + }), + makeView('malformed', { stream_name: 'sql', explorer_state: '{' }), + makeView('invalid', { stream_name: 'sql', explorer_state: '{}' }), + makeView('classic', { explorer_state: null }), + ]); + const select = dialog.getByLabel('Saved log view'); + await expect(select.locator('option')).toHaveCount(5); + await expect(select.locator('option[value="other-canonical"]')).toHaveCount(0); + await expect(select.locator('option[value="malformed"]')).toHaveCount(0); + await expect(select.locator('option[value="invalid"]')).toHaveCount(0); + await select.selectOption('shared'); + await expect(dialog.getByRole('button', { name: 'Convert to SQL view' })).toBeDisabled(); + await expect(dialog.getByRole('button', { name: 'Apply view' })).toBeEnabled(); + await select.selectOption('classic'); + await expect(dialog.getByRole('button', { name: 'Convert to SQL view' })).toBeDisabled(); + await expect(dialog.getByRole('button', { name: 'Apply view' })).toBeDisabled(); + await select.selectOption('canonical'); + await expect(dialog.getByRole('button', { name: 'Convert to SQL view' })).toBeDisabled(); + await expect(dialog.getByRole('button', { name: 'Apply view' })).toBeEnabled(); + await dialog.getByRole('button', { name: 'Close dialog' }).click(); + await page.getByLabel('Dataset', { exact: true }).selectOption('other'); + await page.getByRole('button', { name: 'Saved views', exact: true }).click(); + await expect(select.locator('option')).toHaveCount(2); + await expect(select.locator('option[value="other-canonical"]')).toHaveCount(1); + expect(mutations).toEqual([]); +}); + +test('conversion failure retains the legacy view and explains the server error', async ({ + page, +}) => { + const views = [makeView('owned')], + options = { conversionError: true }; + const { dialog } = await fixture(page, views, options); + await dialog.getByLabel('Saved log view').selectOption('owned'); + await dialog.getByRole('button', { name: 'Convert to SQL view' }).click(); + await expect(dialog.getByRole('alert')).toContainText('Conversion unavailable. Try again.'); + expect(views[0].stream_name).toBe('events'); + await expect(dialog.getByRole('button', { name: 'Convert to SQL view' })).toBeEnabled(); + options.conversionError = false; + await dialog.getByRole('button', { name: 'Convert to SQL view' }).click(); + await expect( + dialog.getByText('View converted to SQL. Its query and dataset were retained.'), + ).toBeVisible(); +}); + +test('late conversion responses do not change a reopened dialog selection or name', async ({ + page, +}) => { + let finish!: () => void; + const waitForConversion = new Promise((resolve) => { + finish = resolve; + }); + const { dialog, mutations } = await fixture( + page, + [makeView('converting'), makeView('other-selection')], + { waitForConversion }, + ); + await dialog.getByLabel('Saved log view').selectOption('converting'); + await dialog.getByRole('button', { name: 'Convert to SQL view' }).click(); + await expect.poll(() => mutations.length).toBe(1); + await dialog.getByRole('button', { name: 'Close dialog' }).click(); + await page.getByRole('button', { name: 'Saved views', exact: true }).click(); + await dialog.getByLabel('Saved log view').selectOption('other-selection'); + await dialog.getByLabel('Log view name').fill('Keep my draft'); + const response = page.waitForResponse('**/filters/converting/convert-to-sql'); + finish(); + await response; + await expect(dialog.getByLabel('Saved log view')).toHaveValue('other-selection'); + await expect(dialog.getByLabel('Log view name')).toHaveValue('Keep my draft'); + await expect( + dialog.getByText('View converted to SQL. Its query and dataset were retained.'), + ).toHaveCount(0); + await expect(dialog.getByRole('button', { name: 'Update selected' })).toBeEnabled(); +}); diff --git a/frontend/src/features/logs/SavedViews.test.ts b/frontend/src/features/logs/SavedViews.test.ts index 39863ee01..eeec3a63c 100644 --- a/frontend/src/features/logs/SavedViews.test.ts +++ b/frontend/src/features/logs/SavedViews.test.ts @@ -1,5 +1,10 @@ import { describe, expect, it } from 'vitest'; -import { ownsSavedLogView, savedLogViewBody, savedLogViewState } from './SavedViews'; +import { + ownsSavedLogView, + savedLogViewBody, + savedLogViewState, + savedLogViewsForDataset, +} from './SavedViews'; import { canShowContext, contextField } from './LogContext'; import type { LogViewState } from './logViewState'; @@ -44,6 +49,45 @@ describe('server saved views', () => { expect(next.time_filter).toBeNull(); expect(savedLogViewState(next).wrap).toBe(true); }); + it('opens canonical SQL views using their explorer dataset and preserves conversion on update', () => { + const legacy = savedLogViewBody(state, 'Legacy'); + expect(legacy.stream_name).toBe('app'); + const canonical = { ...legacy, stream_name: 'sql', filter_id: 'same-id', user_id: 'owner' }; + expect(savedLogViewState(canonical)).toEqual(state); + const updated = savedLogViewBody({ ...state, search: 'new search' }, 'New', canonical); + expect(updated).toMatchObject({ stream_name: 'sql', filter_id: 'same-id', user_id: 'owner' }); + expect(savedLogViewState(updated).search).toBe('new search'); + expect(savedLogViewBody(state, 'Copy').stream_name).toBe('app'); + }); + it('lists native views by their decoded dataset while retaining dataset-scoped classic entries', () => { + const legacy = savedLogViewBody(state, 'Legacy'); + const canonical = { ...legacy, stream_name: 'sql', filter_name: 'Canonical' }; + const { explorer_state: _state, ...classic } = { ...legacy, filter_name: 'Classic' }; + const available = savedLogViewsForDataset( + [ + legacy, + canonical, + classic, + { + ...canonical, + filter_name: 'Other dataset', + explorer_state: JSON.stringify({ ...state, dataset: 'other' }), + }, + { ...canonical, filter_name: 'Broken JSON', explorer_state: '{' }, + { ...canonical, filter_name: 'Invalid state', explorer_state: '{}' }, + { ...classic, stream_name: 'sql' }, + { ...classic, stream_name: 'other' }, + ], + 'app', + ); + expect(available.map(({ view }) => view.filter_name)).toEqual([ + 'Legacy', + 'Canonical', + 'Classic', + ]); + expect(available.map(({ decoded }) => decoded?.dataset)).toEqual(['app', 'app', undefined]); + expect(savedLogViewsForDataset([canonical], 'sql')).toEqual([]); + }); it('rejects classic formats and mismatched dataset metadata', () => { const { explorer_state: _state, ...classic } = savedLogViewBody(state, 'Classic'); expect(() => savedLogViewState(classic)).toThrow('classic'); diff --git a/frontend/src/features/logs/SavedViews.tsx b/frontend/src/features/logs/SavedViews.tsx index 88500a7a0..619ac4d74 100644 --- a/frontend/src/features/logs/SavedViews.tsx +++ b/frontend/src/features/logs/SavedViews.tsx @@ -9,13 +9,27 @@ import type { SavedLogView } from '../../lib/types'; import { decodeLogView, encodeLogView, type LogViewState } from './logViewState'; export function savedLogViewState(view: SavedLogView): LogViewState { - if (typeof view.explorer_state !== 'string') + if (typeof view.explorer_state !== 'string' || view.query.filter_type !== 'sql') throw new Error('This saved view uses a classic format and cannot be opened in this explorer.'); const state = decodeLogView(view.explorer_state); - if (state.dataset !== view.stream_name) + if (view.stream_name !== 'sql' && state.dataset !== view.stream_name) throw new Error('This saved view has inconsistent dataset metadata.'); return state; } +export function savedLogViewsForDataset(views: SavedLogView[], dataset: string) { + return views.flatMap<{ view: SavedLogView; decoded: LogViewState | undefined }>((view) => { + try { + const decoded = savedLogViewState(view); + return decoded.dataset === dataset ? [{ view, decoded }] : []; + } catch { + // Classic dataset-scoped views remain visible, but canonical views need a + // valid explorer state to establish which dataset they belong to. + return view.stream_name !== 'sql' && view.stream_name === dataset + ? [{ view, decoded: undefined }] + : []; + } + }); +} /** Ownership is a display constraint; backend authorization remains authoritative. */ export function ownsSavedLogView( view: SavedLogView | undefined, @@ -36,7 +50,7 @@ export function savedLogViewBody( return { ...previous, filter_name, - stream_name: state.dataset, + stream_name: previous?.stream_name === 'sql' ? 'sql' : state.dataset, query: { ...(previous?.query ?? {}), filter_type: 'sql', @@ -98,22 +112,13 @@ export function SavedViews({ ), ); const available = useMemo( - () => - (views.data ?? []) - .filter((view) => view.stream_name === state.dataset) - .map((view) => { - try { - return { view, decoded: savedLogViewState(view) }; - } catch { - return { view, decoded: undefined }; - } - }), + () => savedLogViewsForDataset(views.data ?? [], state.dataset), [views.data, state.dataset], ); const selectedView = available.find((item) => item.view.filter_id === selected); const view = selectedView?.view; const owned = ownsSavedLogView(view, ownerHash.data, mode === 'demo'); - async function mutate(action: 'create' | 'update' | 'delete') { + async function mutate(action: 'create' | 'update' | 'delete' | 'convert') { const requestGeneration = generation.current; setError(''); setNotice(''); @@ -130,9 +135,14 @@ export function SavedViews({ if (!view?.filter_id) throw new Error('Select a saved view first.'); if (!owned) throw new Error( - 'Only the owner can update or delete this saved view. Apply it or save a new copy instead.', + 'Only the owner can change this saved view. Apply it or save a new copy instead.', ); - if (action === 'update') { + if (action === 'convert') { + if (!selectedView?.decoded) throw new Error('This saved view cannot be converted here.'); + const converted = await client.convertLogViewToSql(view.filter_id); + if (requestGeneration !== generation.current) return; + setName(converted.filter_name); + } else if (action === 'update') { // Classic views remain readable in the list but are never overwritten. if (!selectedView?.decoded) throw new Error('This saved view cannot be overwritten here.'); @@ -148,7 +158,13 @@ export function SavedViews({ } } views.reload(); - setNotice(action === 'delete' ? 'View deleted.' : 'View saved on the server.'); + setNotice( + action === 'delete' + ? 'View deleted.' + : action === 'convert' + ? 'View converted to SQL. Its query and dataset were retained.' + : 'View saved on the server.', + ); } catch (cause) { if (requestGeneration === generation.current) setError(cause instanceof Error ? cause.message : String(cause)); @@ -165,7 +181,13 @@ export function SavedViews({ { + if (!nextOpen) { + generation.current += 1; + setBusy(false); + } + setOpen(nextOpen); + }} title="Saved log views" description="Views are saved on your Parseable server for this dataset." > @@ -217,6 +239,9 @@ export function SavedViews({ apply it or save a new copy.

)} + {view && owned && compatible && view.stream_name !== 'sql' && ( +

Convert this saved view to SQL while keeping its existing query and dataset.

+ )}