diff --git a/projects/caliper/cli/commands.py b/projects/caliper/cli/commands.py index 1d155aebb..8c6fcf285 100644 --- a/projects/caliper/cli/commands.py +++ b/projects/caliper/cli/commands.py @@ -12,6 +12,7 @@ from projects.caliper.engine.ai_eval import run_ai_eval_export from projects.caliper.engine.file_export.artifacts_export_run import run_artifacts_export from projects.caliper.engine.file_export.artifacts_import_run import run_artifacts_import +from projects.caliper.engine.file_export.censoring import apply_censoring_to_artifacts from projects.caliper.engine.file_export.mlflow_config import load_mlflow_config_yaml from projects.caliper.engine.kpi.generate import run_kpi_generate from projects.caliper.engine.kpi.import_export import ( @@ -1239,15 +1240,17 @@ def artifacts_export( } try: - run_artifacts_export( + exit_code = run_artifacts_export( from_path=from_path, backend=list(backend) if backend else ["mlflow"], - mlflow_tracking_uri=mlflow_tracking_uri, - mlflow_experiment=mlflow_experiment, - mlflow_run_id=mlflow_run_id, - mlflow_run_name=mlflow_run_name, - mlflow_insecure_tls=mlflow_insecure_tls, - mlflow_secrets_path=mlflow_secrets_path, + mlflow_tracking_uri=final_config["tracking_uri"], + mlflow_experiment=final_config["experiment"], + mlflow_run_id=final_config["run_id"], + mlflow_run_name=final_config["run_name"], + mlflow_insecure_tls=final_config["insecure_tls"], + mlflow_secrets_path=Path(final_config["secrets_path"]) + if final_config["secrets_path"] + else None, mlflow_config_data=final_config, dry_run=dry_run, verbose=verbose, @@ -1263,6 +1266,215 @@ def artifacts_export( click.echo(full_traceback, err=True) sys.exit(3) + # Check return value outside try block to avoid intercepting SystemExit + if exit_code != 0: + sys.exit(exit_code) + + +@click.command("censor") +@click.option( + "--from", + "from_path", + type=click.Path(path_type=Path, exists=True), + required=True, + help="Source directory containing artifacts to censor.", +) +@click.option( + "--output", + "output_path", + type=click.Path(path_type=Path), + default=None, + help="Output directory for cleaned artifacts. If not specified, censoring is done in-place.", +) +@click.option("--dry-run", is_flag=True, help="Show what would be censored without making changes.") +@click.option("-v", "--verbose", is_flag=True, help="Show detailed censoring information.") +@click.option( + "--report", + "report_path", + type=click.Path(path_type=Path), + default=None, + help="Write censoring report to this file.", +) +def artifacts_censor( + from_path: Path, + output_path: Path | None, + dry_run: bool, + verbose: bool, + report_path: Path | None, +) -> None: + """Censor sensitive artifacts by removing files containing secrets or sensitive patterns.""" + import json + import shutil + from datetime import datetime + + if not from_path.exists(): + click.echo(f"Error: Source path does not exist: {from_path}", err=True) + sys.exit(1) + + if not from_path.is_dir(): + click.echo(f"Error: Source path is not a directory: {from_path}", err=True) + sys.exit(1) + + # Collect all artifact files + all_artifact_paths = [p for p in from_path.rglob("*") if p.is_file()] + + if not all_artifact_paths: + click.echo(f"No artifacts found in {from_path}") + return + + click.echo(f"Scanning {len(all_artifact_paths)} artifacts in {from_path}") + + # Determine which files to censor based on output_path + if output_path: + # Copy source tree to output directory first, then censor the copy + if not dry_run: + try: + if output_path.exists(): + import shutil + + shutil.rmtree(output_path) + shutil.copytree(from_path, output_path) + click.echo(f"Copied source tree to {output_path}") + except Exception as e: + click.echo(f"Error copying source tree: {e}", err=True) + sys.exit(1) + else: + click.echo(f"Dry run: Would copy source tree to {output_path}") + + # Get corresponding files in the output directory for censoring + artifact_paths_to_censor = [ + output_path / p.relative_to(from_path) for p in all_artifact_paths + ] + censoring_source_path = output_path + else: + # Use original files for in-place censoring + artifact_paths_to_censor = all_artifact_paths + censoring_source_path = from_path + + # Apply censoring to the target files (copy or original) + filtered_paths, censoring_results = apply_censoring_to_artifacts( + artifact_paths_to_censor, censoring_enabled=True, verbose=verbose, dry_run=dry_run + ) + + # Separate results: excluded (fully censored), sanitized (redacted in-place), and clean + excluded_files = [r.file_path for r in censoring_results if r.censored and not r.sanitized] + sanitized_files = [r.file_path for r in censoring_results if r.sanitized] + clean_files = [r.file_path for r in censoring_results if not r.censored] + + # Print summary + click.echo("\nCensoring Summary:") + click.echo(f" Total files scanned: {len(all_artifact_paths)}") + click.echo(f" Clean files: {len(clean_files)}") + click.echo(f" Sanitized files: {len(sanitized_files)}") + click.echo(f" Excluded files: {len(excluded_files)}") + + if (excluded_files or sanitized_files) and verbose: + if sanitized_files: + click.echo("\nSanitized files (redacted in-place, kept):") + for result in censoring_results: + if result.sanitized: + rel_path = result.file_path.relative_to(censoring_source_path) + click.echo(f" - {rel_path}: {result.reason}") + if excluded_files: + click.echo("\nExcluded files (to be removed):") + for result in censoring_results: + if result.censored and not result.sanitized: + rel_path = result.file_path.relative_to(censoring_source_path) + click.echo(f" - {rel_path}: {result.reason}") + + # Generate report if requested + if report_path: + report_data = { + "timestamp": datetime.now().isoformat(), + "source_directory": str(from_path), + "output_directory": str(output_path) if output_path else None, + "total_files": len(all_artifact_paths), + "clean_files": len(clean_files), + "sanitized_files": len(sanitized_files), + "excluded_files": len(excluded_files), + "sanitized_details": [ + { + "file": str(r.file_path.relative_to(censoring_source_path)), + "reason": r.reason, + } + for r in censoring_results + if r.sanitized + ], + "excluded_details": [ + { + "file": str(r.file_path.relative_to(censoring_source_path)), + "reason": r.reason, + } + for r in censoring_results + if r.censored and not r.sanitized + ], + } + + if dry_run: + click.echo(f"\nDry run: Would write report to {report_path}") + else: + try: + report_path.parent.mkdir(parents=True, exist_ok=True) + with open(report_path, "w") as f: + json.dump(report_data, f, indent=2) + click.echo( + f"\nCensoring report written to {report_path}\n{report_path.read_text()}" + ) + except Exception as e: + click.echo(f"Error writing report: {e}", err=True) + + # Handle output + if dry_run: + click.echo("\nDry run: No files were modified") + if output_path: + kept_count = len(clean_files) + len(sanitized_files) + click.echo( + f"Would copy source tree to {output_path} and remove {len(excluded_files)} excluded files" + ) + else: + click.echo(f"Would remove {len(excluded_files)} excluded files from {from_path}") + + elif output_path: + # Files have already been copied and censored in the output directory + # Now remove excluded files from the output directory + if excluded_files: + try: + for excluded_file in excluded_files: + excluded_file.unlink() + if verbose: + rel_path = excluded_file.relative_to(output_path) + click.echo(f" Removed from output: {rel_path}") + + kept_count = len(clean_files) + len(sanitized_files) + click.echo( + f"\nProcessed artifacts in {output_path}: {kept_count} kept, {len(excluded_files)} removed" + ) + + except Exception as e: + click.echo(f"Error removing excluded files from output: {e}", err=True) + sys.exit(1) + else: + kept_count = len(clean_files) + len(sanitized_files) + click.echo(f"\nProcessed artifacts in {output_path}: {kept_count} files, all clean!") + + else: + # Remove only excluded files in-place (sanitized files are already redacted and kept) + if excluded_files: + try: + for excluded_file in excluded_files: + excluded_file.unlink() + if verbose: + rel_path = excluded_file.relative_to(from_path) + click.echo(f" Removed: {rel_path}") + + click.echo(f"\nRemoved {len(excluded_files)} excluded files from {from_path}") + + except Exception as e: + click.echo(f"Error removing files: {e}", err=True) + sys.exit(1) + else: + click.echo("\nNo files to remove - all artifacts are clean!") + @click.command("import") @click.option("--from-mlflow", "mlflow_run_id", help="MLflow run ID to download artifacts from.") diff --git a/projects/caliper/cli/main.py b/projects/caliper/cli/main.py index 0bfcd2c13..37f397a62 100644 --- a/projects/caliper/cli/main.py +++ b/projects/caliper/cli/main.py @@ -12,6 +12,7 @@ from projects.caliper.cli.commands import ( ai_eval_export, analyse_kpis_cmd, + artifacts_censor, artifacts_export, artifacts_import, kpi_csv_export, @@ -219,6 +220,7 @@ def run_cli() -> None: kpi_group.add_command(s3_export_cmd) # Register artifacts commands +artifacts_group.add_command(artifacts_censor) artifacts_group.add_command(artifacts_export) artifacts_group.add_command(artifacts_import) diff --git a/projects/caliper/engine/file_export/censoring.py b/projects/caliper/engine/file_export/censoring.py new file mode 100644 index 000000000..6aead82db --- /dev/null +++ b/projects/caliper/engine/file_export/censoring.py @@ -0,0 +1,327 @@ +""" +Artifact censoring module for Caliper. + +This module provides censoring capabilities to filter artifacts before upload: +1. Keyword-based censoring - filters files containing specified keywords/patterns +2. Filename-based censoring - filters files with sensitive filename patterns +""" + +from __future__ import annotations + +import logging +import stat +from dataclasses import dataclass +from pathlib import Path + +from .censoring_rules import ( + COMPILED_KEYWORD_PATTERNS, + matches_sensitive_filename, +) + +logger = logging.getLogger(__name__) + + +def _merge_overlapping_spans(spans: list[tuple[int, int]]) -> list[tuple[int, int]]: + """Merge overlapping or adjacent (start, end) spans into non-overlapping spans. + + Args: + spans: List of (start, end) tuples representing character ranges. + + Returns: + Sorted list of merged (start, end) tuples with no overlaps. + """ + if not spans: + return [] + + sorted_spans = sorted(spans) + merged = [sorted_spans[0]] + + for start, end in sorted_spans[1:]: + prev_start, prev_end = merged[-1] + if start <= prev_end: + # Overlapping or adjacent — extend the previous span + merged[-1] = (prev_start, max(prev_end, end)) + else: + merged.append((start, end)) + + return merged + + +@dataclass +class CensoringResult: + """Result of censoring operation on a single file.""" + + file_path: Path + censored: bool + reason: str + sanitized: bool = False # True if content was sanitized in-place + + def __str__(self): + if self.sanitized: + return f"SANITIZED: {self.file_path} ({self.reason})" + elif self.censored: + return f"EXCLUDED: {self.file_path} ({self.reason})" + else: + return f"ALLOWED: {self.file_path} ({self.reason})" + + +class ArtifactCensor: + """Main censoring class that applies keyword and secret filtering.""" + + def __init__( + self, + vault_secrets: set[str] | None = None, + secret_mapping: dict[str, str] | None = None, + verbose: bool = False, + dry_run: bool = False, + ): + """ + Initialize the artifact censor. + + Args: + vault_secrets: Set of secret strings loaded from vaults + secret_mapping: Dict mapping secret strings to vault/content identifiers + verbose: Enable verbose logging + dry_run: Skip file modifications while preserving analysis + """ + self.vault_secrets = vault_secrets or set() + self.secret_mapping = secret_mapping or {} + self.verbose = verbose + self.dry_run = dry_run + + if self.verbose: + logger.info(f"Initialized censoring with {len(self.vault_secrets)} vault secrets") + + def _is_text_file(self, file_path: Path) -> bool: + """Check if file is likely a text file based on extension and content sample.""" + # Common text file extensions + text_extensions = { + ".txt", + ".log", + ".yaml", + ".yml", + ".json", + ".xml", + ".csv", + ".md", + ".rst", + ".py", + ".js", + ".html", + ".css", + ".sql", + ".sh", + ".bash", + ".conf", + ".cfg", + ".ini", + ".properties", + ".out", + ".err", + } + + if file_path.suffix.lower() in text_extensions: + return True + + # For files without clear extension, try to detect if it's text + try: + # Read only first 1KB to avoid loading large files into memory + with open(file_path, "rb") as f: + sample = f.read(1024) + if not sample: + return True # Empty file is text + + # Check if most bytes are printable ASCII or common UTF-8 + printable_ratio = sum(1 for b in sample if 32 <= b <= 126 or b in (9, 10, 13)) / len( + sample + ) + return printable_ratio > 0.7 + + except Exception: + return False # If we can't read it, assume it's binary + + def _sanitize_file_content(self, file_path: Path) -> CensoringResult: + """Sanitize file content by replacing sensitive patterns.""" + try: + # First check if filename itself is sensitive - sanitize these files + if matches_sensitive_filename(str(file_path)): + # Replace content with censoring message + censored_content = ( + f"Content censored by caliper - sensitive filename: {file_path.name}\n" + ) + + # Write sanitized content back to file (skip if dry run) + if not self.dry_run: + file_path.write_text(censored_content, encoding="utf-8") + + return CensoringResult( + file_path, True, f"sensitive filename pattern: {file_path.name}", sanitized=True + ) + + # Skip non-text files to avoid reading binary content + if not self._is_text_file(file_path): + return CensoringResult(file_path, False, "non-text file") + + content = file_path.read_text(encoding="utf-8", errors="ignore") + sanitized = False + reasons = [] + + # Import KEYWORD_PATTERNS for reporting + from .censoring_rules import KEYWORD_PATTERNS + + # Check for keyword patterns - redact matched spans in place + keyword_detected = False + matched_patterns = set() + for i, pattern in enumerate(COMPILED_KEYWORD_PATTERNS): + matches = list(pattern.finditer(content)) + if matches: + keyword_detected = True + matched_patterns.add(i) + + if keyword_detected: + # Collect all match spans from all patterns + all_spans = [] + for pattern in COMPILED_KEYWORD_PATTERNS: + for match in pattern.finditer(content): + all_spans.append((match.start(), match.end())) + + # Merge overlapping spans to avoid corrupting replacements + merged_spans = _merge_overlapping_spans(all_spans) + + # Replace merged spans in reverse order to preserve string indices + for start, end in reversed(merged_spans): + content = content[:start] + "[REDACTED]" + content[end:] + + # Add reasons for all matched patterns + for pattern_idx in sorted(matched_patterns): + reasons.append(f"contains keyword pattern: {KEYWORD_PATTERNS[pattern_idx]}") + + sanitized = True + + # Always check for vault secrets, independent of keyword pattern detection + for secret in self.vault_secrets: + if secret and secret.strip() and secret.strip() in content: + content = content.replace(secret.strip(), "*******") + sanitized = True + vault_identifier = self.secret_mapping.get(secret.strip(), "unknown vault") + reasons.append(f"contains vault secret: {vault_identifier}") + + if sanitized: + # Write sanitized content back to original file (skip if dry run) + if not self.dry_run: + try: + file_path.write_text(content, encoding="utf-8") + except PermissionError: + # Handle read-only files by making them writable + try: + # Make file writable + file_path.chmod(file_path.stat().st_mode | stat.S_IWUSR) + file_path.write_text(content, encoding="utf-8") + except Exception: + # If we still can't write, abort this file export + raise + + reason = reasons[0] if reasons else "sensitive content detected" + return CensoringResult(file_path, True, reason, sanitized=True) + + return CensoringResult(file_path, False, "content check passed") + + except Exception as e: + logger.warning(f"Error sanitizing file {file_path}: {e}") + return CensoringResult(file_path, True, f"sanitization failed: {e}") + + def censor_files(self, file_paths: list[Path]) -> tuple[list[Path], list[CensoringResult]]: + """ + Apply censoring to a list of file paths, sanitizing content where possible. + + Args: + file_paths: List of file paths to check + + Returns: + tuple: (processed_files, censoring_results) + processed_files contains original paths for clean files, + sanitized paths for files with content replacements, + and excludes files with sensitive filenames + """ + processed_files = [] + results = [] + + for file_path in file_paths: + if not file_path.is_file(): + # Skip directories and non-existent files + processed_files.append(file_path) + continue + + result = self._sanitize_file_content(file_path) + results.append(result) + + if not result.censored: + # Clean file, include original + processed_files.append(file_path) + elif result.sanitized: + # File had content sanitized in-place, include original path + processed_files.append(file_path) + if self.verbose: + logger.info(f"Sanitized: {result}") + else: + # File excluded entirely (e.g., sensitive filename) + if self.verbose: + logger.info(f"Excluded: {result}") + + return processed_files, results + + +def apply_censoring_to_artifacts( + artifact_paths: list[Path], + censoring_enabled: bool = True, + verbose: bool = False, + vault_secrets: set[str] | None = None, + secret_mapping: dict[str, str] | None = None, + dry_run: bool = False, +) -> tuple[list[Path], list[CensoringResult]]: + """ + Apply censoring to artifact paths, sanitizing sensitive content in-place. + + This function processes artifacts by: + - Replacing sensitive content patterns with "Content censored by caliper" in-place + - Replacing vault secrets with "*******" in-place + - Excluding files with sensitive filename patterns (.pem, .key, files with "secret" in name, etc.) + + Args: + artifact_paths: List of artifact file paths + censoring_enabled: Whether to apply censoring + verbose: Enable verbose logging + vault_secrets: Set of vault secret strings to censor + secret_mapping: Dict mapping secret strings to vault/content identifiers + dry_run: Skip file modifications while preserving analysis + + Returns: + tuple: (processed_paths, censoring_results) + processed_paths contains original paths for clean and sanitized files, + and excludes files with sensitive filenames + """ + if not censoring_enabled: + if verbose: + logger.info("Censoring disabled, allowing all artifacts") + return artifact_paths, [] + + # Apply censoring using keyword and filename patterns + censor = ArtifactCensor( + vault_secrets=vault_secrets or set(), + secret_mapping=secret_mapping or {}, + verbose=verbose, + dry_run=dry_run, + ) + processed_paths, results = censor.censor_files(artifact_paths) + + # Log summary + sanitized_count = len([r for r in results if r.sanitized]) + excluded_count = len([r for r in results if r.censored and not r.sanitized]) + clean_count = len([r for r in results if not r.censored]) + + if verbose or sanitized_count > 0 or excluded_count > 0: + logger.info( + f"Censoring complete: {clean_count} clean, {sanitized_count} sanitized, {excluded_count} excluded" + ) + + return processed_paths, results diff --git a/projects/caliper/engine/file_export/censoring_rules.py b/projects/caliper/engine/file_export/censoring_rules.py new file mode 100644 index 000000000..1c56b0bf7 --- /dev/null +++ b/projects/caliper/engine/file_export/censoring_rules.py @@ -0,0 +1,107 @@ +""" +Censoring rules and patterns for Caliper artifact filtering. + +This module defines the patterns and rules used to identify sensitive content +in artifacts before upload. +""" + +from __future__ import annotations + +import re + +# Keyword patterns to detect in file content +# These are compiled regex patterns that match common sensitive data patterns +KEYWORD_PATTERNS = [ + # Password patterns + r"password\s*[:=]\s*\S+", + r"passwd\s*[:=]\s*\S+", + # API key patterns + r"api[_-]?key\s*[:=]\s*\S+", + r"apikey\s*[:=]\s*\S+", + r"api[_-]?secret\s*[:=]\s*\S+", + # Token patterns + r"token\s*[:=]\s*\S+", + r"secret[_-]?token\s*[:=]\s*\S+", + r"access[_-]?token\s*[:=]\s*\S+", + r"refresh[_-]?token\s*[:=]\s*\S+", + # Bearer tokens (compiled with IGNORECASE, so one pattern covers both cases) + r"Bearer\s+[A-Za-z0-9+/=]+", + # Specific service API keys + r"sk-[a-zA-Z0-9]{32,}", # OpenAI API keys + r"ghp_[a-zA-Z0-9]{36}", # GitHub personal access tokens + r"gho_[a-zA-Z0-9]{36}", # GitHub OAuth tokens + r"ghu_[a-zA-Z0-9]{36}", # GitHub user-to-server tokens + r"ghs_[a-zA-Z0-9]{36}", # GitHub server-to-server tokens + r"ghr_[a-zA-Z0-9]{36}", # GitHub refresh tokens + # AWS patterns + r"AKIA[0-9A-Z]{16}", # AWS Access Key ID + r"aws[_-]?secret[_-]?access[_-]?key\s*[:=]\s*\S+", + # Database connection strings + r"mongodb://[^/\s]+:[^@\s]+@", + r"mysql://[^/\s]+:[^@\s]+@", + r"postgresql://[^/\s]+:[^@\s]+@", + # Generic credential patterns + r"credential\s*[:=]\s*\S+", + r"private[_-]?key\s*[:=]\s*\S+", +] + +# Compile patterns for better performance +COMPILED_KEYWORD_PATTERNS = [re.compile(pattern, re.IGNORECASE) for pattern in KEYWORD_PATTERNS] + +# File patterns to always censor (by filename) +SENSITIVE_FILE_PATTERNS = [ + r".*\.pem$", # PEM certificate files + r".*\.key$", # Private key files + r".*\.p12$", # PKCS#12 certificate files + r".*\.pfx$", # PKCS#12 certificate files (Windows) + r".*secret.*", # Any file with "secret" in the name + r".*credential.*", # Any file with "credential" in the name + r".*password.*", # Any file with "password" in the name + r".*\.ssh/.*", # SSH directory contents + r".*/\.ssh/.*", # SSH directory contents (with path) + r".*id_rsa.*", # SSH private keys + r".*id_dsa.*", # DSA private keys + r".*id_ecdsa.*", # ECDSA private keys + r".*id_ed25519.*", # Ed25519 private keys + r".*\.env$", # Environment files + r".*\.env\..*", # Environment files with suffixes +] + +# Compile file patterns for better performance +COMPILED_FILE_PATTERNS = [re.compile(pattern, re.IGNORECASE) for pattern in SENSITIVE_FILE_PATTERNS] + + +def matches_sensitive_filename(filename: str) -> bool: + """ + Check if a filename matches any sensitive file pattern. + + Args: + filename: The filename or path to check + + Returns: + bool: True if the filename indicates a sensitive file + """ + from pathlib import Path + + # Filename-only patterns that should match just the basename + filename_only_patterns = [ + r".*secret.*", + r".*credential.*", + r".*password.*", + ] + + # Check filename-only patterns against basename + basename = Path(filename).name + for pattern_str in filename_only_patterns: + pattern = re.compile(pattern_str, re.IGNORECASE) + if pattern.match(basename): + return True + + # Check all other patterns against full path + for pattern in COMPILED_FILE_PATTERNS: + pattern_str = pattern.pattern + if pattern_str not in filename_only_patterns: + if pattern.match(filename): + return True + + return False diff --git a/projects/caliper/orchestration/censoring.py b/projects/caliper/orchestration/censoring.py new file mode 100644 index 000000000..6d7c61ebe --- /dev/null +++ b/projects/caliper/orchestration/censoring.py @@ -0,0 +1,370 @@ +""" +Orchestration-level censoring functionality for Caliper artifact export. + +This module provides high-level censoring operations that integrate with the export +orchestration system, including reporting and notification generation. +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import yaml + +from projects.caliper.engine.file_export.censoring import apply_censoring_to_artifacts +from projects.caliper.orchestration.export_config import CaliperOrchestrationExportConfig +from projects.core.library import ci as ci_lib +from projects.core.library import env +from projects.core.library import vault as vault_lib + +logger = logging.getLogger(__name__) + + +def discover_vault_secrets(verbose: bool = False) -> tuple[set[str], dict[str, str]]: + """ + Discover all vault secrets for censoring (all vault content is treated as sensitive). + + Args: + verbose: Enable verbose logging + + Returns: + Tuple of (secret_strings, secret_mapping) where: + - secret_strings: Set of secret strings loaded from vaults + - secret_mapping: Dict mapping secret string -> vault/content identifier + """ + vault_secrets = set() + secret_mapping = {} + + try: + vault_manager = vault_lib.get_vault_manager() + available_vaults = vault_manager.list_vaults() + + if verbose: + logger.info(f"Discovering vault secrets from {len(available_vaults)} vaults") + + secrets_discovered = 0 + for vault_name in available_vaults: + vault = vault_manager.get_vault(vault_name) + if not vault: + continue + + for content_name, content_def in vault.content.items(): + # Only process content marked as sensible + if not content_def.is_sensible: + if verbose: + logger.info( + f"Skipping non-sensible vault content {vault_name}/{content_name}" + ) + continue + + content_path = vault_manager.get_vault_content_path(vault_name, content_name) + if not (content_path and content_path.exists()): + logger.warning(f"Invalid vault found: {vault_name} {content_name} (missing)") + continue + + try: + # Read vault content (assume it's text) + content_text = content_path.read_text(encoding="utf-8", errors="ignore").strip() + if not content_text: + logger.warning(f"Invalid vault found: {vault_name} {content_name} (empty)") + else: + vault_secrets.add(content_text) + secret_mapping[content_text] = f"{vault_name}/{content_name}" + secrets_discovered += 1 + if verbose: + logger.info(f"Discovered secret from vault {vault_name}/{content_name}") + except Exception as e: + logger.warning(f"Failed to read vault content {vault_name}/{content_name}: {e}") + + if verbose: + logger.info(f"Discovered {secrets_discovered} vault secrets for censoring") + + except Exception as e: + logger.exception(f"Failed to discover vault secrets: {e}") + raise + + return vault_secrets, secret_mapping + + +def censor_text(text: str, verbose: bool = False) -> str: + """ + Censor sensitive content in text using caliper's censoring rules. + + Args: + text: The text to censor + verbose: Enable verbose logging + + Returns: + Censored text with sensitive content replaced + """ + if not text: + return text + + censored_text = text + replacements_made = 0 + + # Import keyword patterns for content censoring (always try this first) + from projects.caliper.engine.file_export.censoring_rules import ( + COMPILED_KEYWORD_PATTERNS, + KEYWORD_PATTERNS, + ) + + # Apply keyword pattern censoring first (preserve this even if vault discovery fails) + for i, pattern in enumerate(COMPILED_KEYWORD_PATTERNS): + matches = list(pattern.finditer(censored_text)) + if matches: + # Replace all matched spans with redacted text, preserving other content + # Process patterns in reverse order by position to maintain string indices + for match in reversed(matches): + censored_text = ( + censored_text[: match.start()] + "[REDACTED]" + censored_text[match.end() :] + ) + replacements_made += 1 + + if verbose: + logger.info( + f"Censored {len(matches)} instances of pattern '{KEYWORD_PATTERNS[i]}' in text" + ) + + # Now try vault secrets discovery and censoring + try: + # Discover vault secrets for censoring + vault_secrets, secret_mapping = discover_vault_secrets(verbose=verbose) + # /!\ secret_mapping contains the secret values. Process with extra care. + + # Replace vault secrets (more specific) + for secret in vault_secrets: + if secret and secret.strip() and secret.strip() in censored_text: + censored_text = censored_text.replace(secret.strip(), "[REDACTED-VAULT]") + replacements_made += 1 + if verbose: + vault_identifier = secret_mapping.get(secret.strip(), "unknown vault") + logger.info(f"Censored vault secret from {vault_identifier} in text") + + if replacements_made > 0: + logger.info(f"Censored {replacements_made} sensitive items from text") + elif verbose: + logger.info("No sensitive content found in text") + + except Exception as e: + logger.error(f"Failed to discover vault secrets during text censoring: {e}") + # Propagate the vault discovery failure after keyword censoring is complete + # This allows _censor_notification_text to handle the failure appropriately + raise + + return censored_text + + +def orchestration_apply_censoring( + from_path: Path, + export_cfg: CaliperOrchestrationExportConfig, + disable_censoring: bool = False, +) -> bool: + """ + Apply censoring for single-run orchestration export. + + Args: + from_path: Source directory containing artifacts + export_cfg: Export configuration + disable_censoring: Whether to skip censoring + + Returns: + True if any files were censored (sanitized or excluded), False otherwise + """ + if disable_censoring: + if export_cfg.verbose: + logger.info("Censoring disabled via --disable-censoring flag") + return False + + # Discover vault secrets + vault_secrets, secret_mapping = discover_vault_secrets(verbose=export_cfg.verbose) + # /!\ secret_mapping contains the secret values. Process with extra care. + + # Collect all artifact files + all_artifact_paths = [p for p in from_path.rglob("*") if p.is_file()] + + logger.info( + f"Starting artifact censoring: {len(all_artifact_paths)} files to scan from {from_path}" + ) + + # Apply in-place censoring + processed_paths, censoring_results = apply_censoring_to_artifacts( + all_artifact_paths, + censoring_enabled=True, + verbose=export_cfg.verbose, + vault_secrets=vault_secrets, + secret_mapping=secret_mapping, + ) + + # Separate results by type + clean_files = [r for r in censoring_results if not r.censored] + sanitized_files = [r for r in censoring_results if r.sanitized] + excluded_files = [r for r in censoring_results if r.censored and not r.sanitized] + + logger.info( + f"Censoring complete: {len(clean_files)} clean, {len(sanitized_files)} sanitized, {len(excluded_files)} excluded" + ) + + _generate_censoring_report(censoring_results, from_path) + + # Abort export if any files were excluded (fail closed) + if excluded_files: + # Create notification about censoring activity blocking export + _create_censoring_notification(censoring_results, from_path, export_blocked=True) + + excluded_file_names = [str(r.file_path.relative_to(from_path)) for r in excluded_files] + logger.error( + f"Export aborted: {len(excluded_files)} files contain sensitive content that could not be sanitized: {excluded_file_names[:5]}{'...' if len(excluded_file_names) > 5 else ''}" + ) + raise RuntimeError( + f"Export blocked due to {len(excluded_files)} unsanitizable sensitive files" + ) + + # Generate censoring report and notification if any files were processed + if sanitized_files: + # Create notification about censoring activity (export proceeds with sanitized files) + _create_censoring_notification(censoring_results, from_path, export_blocked=False) + + if export_cfg.verbose: + logger.info( + f"Export proceeding with {len(sanitized_files)} files sanitized for sensitive content" + ) + return True # Censoring occurred + + elif export_cfg.verbose: + logger.info("No files required censoring") + + return False # No censoring occurred + + +def _generate_censoring_report(censoring_results, from_path: Path) -> None: + """Generate a YAML report of censored files in ARTIFACT_DIR.""" + try: + from datetime import datetime + + # Group censored files by reason + censored_by_reason = {} + for r in censoring_results: + if r.censored: + reason = r.reason + if reason not in censored_by_reason: + censored_by_reason[reason] = [] + + file_path = ( + str(r.file_path.relative_to(from_path)) + if r.file_path.is_relative_to(from_path) + else str(r.file_path) + ) + censored_by_reason[reason].append(file_path) + + # Create report data + report_data = { + "timestamp": datetime.now().isoformat(), + "source_directory": str(from_path), + "total_files": len(censoring_results), + "censored_files": len([r for r in censoring_results if r.censored]), + "clean_files": len([r for r in censoring_results if not r.censored]), + "censored_by_reason": censored_by_reason, + } + + # Write report to ARTIFACT_DIR + if env.ARTIFACT_DIR: + report_path = env.ARTIFACT_DIR / "censoring_report.yaml" + report_path.parent.mkdir(parents=True, exist_ok=True) + + with open(report_path, "w") as f: + yaml.dump(report_data, f, indent=2, default_flow_style=False) + + logger.info(f"Censoring report written to {report_path}") + else: + logger.warning("ARTIFACT_DIR not set, cannot write censoring report") + + except Exception as e: + logger.error(f"Failed to generate censoring report: {e}") + + +def _create_censoring_notification( + censoring_results, from_path: Path, export_blocked: bool = False +) -> None: + """Create a notification file about censoring activity.""" + try: + # Separate results + sanitized_files = [r.file_path for r in censoring_results if r.sanitized] + excluded_files = [r.file_path for r in censoring_results if r.censored and not r.sanitized] + + if not sanitized_files and not excluded_files: + return # No censoring occurred + + # Prepare file lists + def format_file_list(files, limit=10): + file_list = "\n".join( + [ + f" - {file.relative_to(from_path) if file.is_relative_to(from_path) else file}" + for file in files[:limit] + ] + ) + if len(files) > limit: + file_list += f"\n ... and {len(files) - limit} more files" + return file_list + + if export_blocked and excluded_files: + # Export blocked due to excluded files + message = f"""Caliper Export Blocked: Sensitive Files Cannot Be Sanitized + +{len(excluded_files)} file(s) with sensitive filenames were excluded from export: + +{format_file_list(excluded_files)} + +These files have filenames that indicate sensitive content and cannot be safely sanitized: +- Certificate files (.pem, .key, .p12, .pfx) +- Files with 'secret', 'credential', or 'password' in their names +- SSH keys and configuration files + +Please review these artifacts and rename or relocate sensitive files before re-running the export. + +For details, see: $ARTIFACT_DIR/censoring_report.yaml""" + + notification_name = "censoring_blocked" + + else: + # Export proceeded with sanitization + total_censored = len(sanitized_files) + len(excluded_files) + message_parts = [ + f"Caliper Export: {total_censored} file(s) contained sensitive content" + ] + + if sanitized_files: + message_parts.append( + f""" +{len(sanitized_files)} file(s) had sensitive content sanitized and included in export: + +{format_file_list(sanitized_files)} + +Sensitive content (passwords, API keys, tokens) was replaced with placeholder text.""" + ) + + if excluded_files: + message_parts.append( + f""" +{len(excluded_files)} file(s) with sensitive filenames were excluded: + +{format_file_list(excluded_files)} + +These files cannot be safely sanitized due to their filenames.""" + ) + + message_parts.append("\nFor details, see: $ARTIFACT_DIR/censoring_report.yaml") + message = "\n".join(message_parts) + "\n" + notification_name = "censoring_applied" + + # Create notification file using CI library + notification_file = ci_lib.add_notification_file(name=notification_name, message=message) + + if notification_file: + logger.info(f"Censoring notification created: {notification_file}") + else: + logger.error("Failed to create censoring notification file") + + except Exception as e: + logger.error(f"Failed to create censoring notification: {e}") diff --git a/projects/caliper/orchestration/export.py b/projects/caliper/orchestration/export.py index c9a5b4569..39c3dc953 100644 --- a/projects/caliper/orchestration/export.py +++ b/projects/caliper/orchestration/export.py @@ -21,9 +21,14 @@ from projects.caliper.engine.file_export.artifacts_export_run import ( discover_run_dirs, run_artifacts_export, - run_multi_run_artifacts_export, ) -from projects.caliper.engine.file_export.mlflow_config import load_mlflow_config_yaml +from projects.caliper.engine.file_export.mlflow_config import ( + load_mlflow_config_yaml, + project_metadata_fields, +) +from projects.caliper.orchestration.censoring import ( + orchestration_apply_censoring, +) from projects.caliper.orchestration.export_config import ( CaliperOrchestrationExportConfig, ) @@ -34,6 +39,18 @@ logger = logging.getLogger(__name__) +class CaliperExportError(Exception): + """Base exception for Caliper export errors.""" + + pass + + +class ExportFailedException(CaliperExportError): + """Exception raised when export fails.""" + + pass + + # --------------------------------------------------------------------------- # Run naming helpers # --------------------------------------------------------------------------- @@ -146,6 +163,8 @@ def resolve_run_names( def run_from_orchestration_config( caliper_cfg: dict[str, Any] | None, + disable_censoring: bool = False, + disable_file_export: bool = False, ) -> int: """ Run Caliper file export from orchestration config. @@ -231,21 +250,24 @@ def run_from_orchestration_config( logger.info( "dry-run: would export %d run dirs from %s (skipping)", len(run_dirs), from_path ) - ret = 0 + return { + "success": True, + "final_status": "dry-run", + "dry_run": True, + "run_dirs": len(run_dirs), + } else: - ret = run_multi_run_artifacts_export( + _run_multi_run_export( + export_cfg=export_cfg, from_path=from_path, - run_dirs=run_dirs, - backend=backends, - mlflow_experiment=export_cfg.mlflow_experiment, - mlflow_run_name=naming.get("parent_run_name"), + status_yaml=status_yaml, mlflow_secrets_path=mlflow_secrets_path, mlflow_config_data=mlflow_config_data, - mlflow_run_id=mlflow_run_id, + run_dirs=run_dirs, + resolved_parent_name=naming.get("parent_run_name"), child_run_names=naming.get("child_run_names") or {}, - verbose=export_cfg.verbose, - status_yaml_path=status_yaml, - upload_workers=export_cfg.upload_workers, + disable_censoring=disable_censoring, + disable_file_export=disable_file_export, ) else: effective_name = ( @@ -261,21 +283,44 @@ def run_from_orchestration_config( if mlflow_config_data is not None: mlflow_kwargs["mlflow_config_data"] = mlflow_config_data - ret = run_artifacts_export( - from_path=from_path, - status_yaml_path=status_yaml, - dry_run=export_cfg.dry_run, - verbose=export_cfg.verbose, - upload_workers=export_cfg.upload_workers, - backend=backends, - **mlflow_kwargs, - ) - - if ret != 0: - raise RuntimeError(f"Caliper export failed (ret code = {ret})") + # Apply censoring if enabled (in-place modification) + censoring_occurred = orchestration_apply_censoring(from_path, export_cfg, disable_censoring) + + if disable_file_export: + # Create mock status for notifications + mock_status = { + "success": True, + "final_status": "success", + "caliper_artifacts_export": { + "backends": {"mlflow": {"success": True, "run_id": "mock-disabled-export-id"}}, + }, + "duration": "0 seconds (export disabled)", + "censoring_occurred": censoring_occurred, + } + # Write mock status to status file + with open(status_yaml, "w") as f: + yaml.dump(mock_status, f, indent=4) + else: + ret = run_artifacts_export( + from_path=from_path, + status_yaml_path=status_yaml, + dry_run=export_cfg.dry_run, + verbose=export_cfg.verbose, + upload_workers=export_cfg.upload_workers, + backend=backends, + **mlflow_kwargs, + ) + if ret != 0: + raise ExportFailedException(f"Artifacts export failed (ret code = {ret})") with open(status_yaml) as f: - return yaml.safe_load(f.read()) + status = yaml.safe_load(f.read()) + + # Add censoring information to status + if len(run_dirs) == 1: + status["censoring_occurred"] = censoring_occurred + + return status TEST_LABELS_FILENAME = "__test_labels__.yaml" @@ -459,7 +504,7 @@ def build_mlflow_run_url( assert_tracking_uri_has_no_userinfo(tracking_uri) qs = f"?workspace={quote(workspace, safe='')}" if workspace else "" - return f"{tracking_uri}/#/experiments/{experiment_id}/runs/{run_id}/artifacts{qs}" + return f"{tracking_uri}/{qs}#/experiments/{experiment_id}/runs/{run_id}/artifacts" def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: @@ -480,3 +525,182 @@ def _discover_precreated_mlflow_run_id(from_path: Path) -> str | None: logger.warning("Failed to read test labels %s: %s", labels_file, e) return None + + +METRICS_FILE = "metrics.json" +PARAMETERS_FILE = "parameters.json" +TEST_LABELS_MARKER = "__test_labels__.yaml" + + +def _discover_run_dirs(from_path: Path) -> list[Path]: + """Auto-detect test run directories via ``__test_labels__.yaml`` markers.""" + run_dirs: list[Path] = [] + for marker in sorted(from_path.rglob(TEST_LABELS_MARKER)): + if marker.is_file(): + run_dirs.append(marker.parent) + + if run_dirs: + logger.info( + "Auto-detected %d test run director%s via %s", + len(run_dirs), + "y" if len(run_dirs) == 1 else "ies", + TEST_LABELS_MARKER, + ) + return run_dirs + + +def _run_multi_run_export( + *, + export_cfg: CaliperOrchestrationExportConfig, + from_path: Path, + status_yaml: Path, + mlflow_secrets_path: Path, + mlflow_config_data: dict[str, Any] | None, + run_dirs: list[Path], + resolved_parent_name: str | None = None, + child_run_names: dict[Path, str] | None = None, + disable_censoring: bool = False, + disable_file_export: bool = False, +) -> None: + """Export as parent + nested child MLflow runs. + + Raises: + ExportFailedException: If the export fails + """ + import sys + import traceback + + import click + + from projects.caliper.engine.file_export import mlflow_backend + from projects.caliper.engine.file_export.artifacts_export_run import ( + merge_mlflow_files_with_cli, + write_artifacts_status_yaml, + ) + from projects.caliper.engine.file_export.mlflow_secrets import ( + load_mlflow_secrets_yaml, + project_secrets_fields, + validate_mlflow_secrets, + ) + from projects.caliper.engine.model import FileExportBackendResult + + logger.info("Multi-run export: %d test run(s) detected", len(run_dirs)) + + # Apply censoring for multi-run export if enabled + censoring_occurred = orchestration_apply_censoring(from_path, export_cfg, disable_censoring) + + # Collect artifact paths after censoring (files may have been modified in-place) + all_artifact_paths = [p for p in from_path.rglob("*") if p.is_file()] + + secrets_data = None + if mlflow_secrets_path is not None: + secrets_data = load_mlflow_secrets_yaml(mlflow_secrets_path) + validate_mlflow_secrets(secrets_data) + + merged_ml = merge_mlflow_files_with_cli( + None, + secrets_data=secrets_data, + config_data=mlflow_config_data, + cli_tracking_uri=None, + cli_experiment=export_cfg.mlflow_experiment, + cli_run_id=None, + cli_run_name=export_cfg.mlflow_run_name, + ) + + secret_part = project_secrets_fields(merged_ml) + mlflow_connection = secret_part if secret_part else None + + tracking_uri = merged_ml.get("tracking_uri") + experiment = merged_ml.get("experiment") + run_name = resolved_parent_name or merged_ml.get("run_name") + workspace = merged_ml.get("workspace") + if not workspace: + raise ValueError("The export workspace must be specified") + + meta = project_metadata_fields(merged_ml) + run_metadata = meta if meta else None + + insecure_tls = bool(mlflow_connection and mlflow_connection.get("insecure_tls")) + + if export_cfg.verbose: + click.echo("caliper multi-run export (verbose)", err=True) + click.echo(f" Source: {from_path}", err=True) + click.echo(f" Total artifact files: {len(all_artifact_paths)}", err=True) + click.echo(f" Run directories: {len(run_dirs)}", err=True) + click.echo(f" Workspace: {workspace}", err=True) + for rd in run_dirs: + click.echo(f" - {rd.name}", err=True) + click.echo("", err=True) + + try: + if disable_file_export: + # Create mock results for notifications + detail = "" + ml_meta = { + "run_id": "mock-multi-run-disabled-export", + "experiment_url": "http://DRY_RUN_MLFLOW_FAKE_URL/#/experiments/disabled", + "run_url": "http://DRY_RUN_MLFLOW_FAKE_URL/#/experiments/disabled/runs/mock-multi-run-disabled-export", + "tracking_uri": "http://DRY_RUN_MLFLOW_FAKE_URL", + } + results = [ + FileExportBackendResult( + backend="mlflow", + status="success", + detail=detail, + metadata=ml_meta, + ) + ] + else: + detail, ml_meta = mlflow_backend.log_multi_run_artifacts( + all_artifact_paths=all_artifact_paths, + artifact_root=from_path, + run_dirs=run_dirs, + metrics_file=METRICS_FILE, + parameters_file=PARAMETERS_FILE, + tracking_uri=tracking_uri, + experiment=experiment, + parent_run_name=run_name, + insecure_tls=insecure_tls, + connection=mlflow_connection, + verbose=export_cfg.verbose, + upload_workers=export_cfg.upload_workers, + run_metadata=run_metadata, + workspace=workspace, + child_run_names=child_run_names or None, + ) + results = [ + FileExportBackendResult( + backend="mlflow", + status="success", + detail=detail, + metadata=ml_meta, + ) + ] + except Exception as e: + traceback.print_exception(e, file=sys.stderr) + click.echo(f"multi-run export failed: {e}", err=True) + results = [FileExportBackendResult(backend="mlflow", status="failure", detail=str(e))] + + if not disable_file_export: + for r in results: + click.echo(f"{r.backend}: {r.status} {r.detail}") + + if status_yaml is not None: + try: + write_artifacts_status_yaml(status_yaml, results) + + # Add censoring information to the status file + with open(status_yaml) as f: + status_data = yaml.safe_load(f) + status_data["censoring_occurred"] = censoring_occurred + with open(status_yaml, "w") as f: + yaml.dump(status_data, f, indent=4) + + if not disable_file_export: + click.echo(f"Wrote status YAML to {status_yaml}") + except OSError as e: + click.echo(f"Failed to write status YAML ({status_yaml}): {e}", err=True) + raise ExportFailedException(f"Failed to write status YAML: {e}") from None + + if any(r.status == "failure" for r in results): + raise ExportFailedException("MLflow backend export failed") diff --git a/projects/caliper/orchestration/notification.py b/projects/caliper/orchestration/notification.py index d2d8558a4..c3f928392 100644 --- a/projects/caliper/orchestration/notification.py +++ b/projects/caliper/orchestration/notification.py @@ -62,8 +62,14 @@ def format_postprocess_status_notification( # Check overall status (keep unchanged regardless of abort status) status_emoji = "✅" if result.success else "❌" + lines.append("") + lines.append("---") lines.append(f"**Post-processing Status** {status_emoji}") + # Add directory path if available + if result.base_directory: + lines.append(f"`{result.base_directory}`") + # Add steps information if available, sorted by completion time if result.steps: # Sort steps by completion timestamp (completed_at), with fallback to step name for stable ordering diff --git a/projects/caliper/tests/test_censoring.py b/projects/caliper/tests/test_censoring.py new file mode 100644 index 000000000..4bd5d6333 --- /dev/null +++ b/projects/caliper/tests/test_censoring.py @@ -0,0 +1,552 @@ +""" +Tests for the caliper artifact censoring module. + +Covers: +- Keyword pattern matching and redaction +- Filename-based sensitive file detection +- Content sanitization with in-place replacement +- Overlapping regex span merging +- Dry-run mode behavior +- Separation of censored vs sanitized files +- File exclusion vs sanitization classification +- End-to-end censoring with temporary directories +- Bearer token pattern deduplication +- Vault secret replacement +""" + +from __future__ import annotations + +from pathlib import Path + +from projects.caliper.engine.file_export.censoring import ( + ArtifactCensor, + CensoringResult, + _merge_overlapping_spans, + apply_censoring_to_artifacts, +) +from projects.caliper.engine.file_export.censoring_rules import ( + COMPILED_KEYWORD_PATTERNS, + KEYWORD_PATTERNS, + matches_sensitive_filename, +) + +# --------------------------------------------------------------------------- +# _merge_overlapping_spans +# --------------------------------------------------------------------------- + + +class TestMergeOverlappingSpans: + def test_empty_list(self): + assert _merge_overlapping_spans([]) == [] + + def test_single_span(self): + assert _merge_overlapping_spans([(0, 5)]) == [(0, 5)] + + def test_non_overlapping_spans(self): + spans = [(0, 5), (10, 15), (20, 25)] + assert _merge_overlapping_spans(spans) == [(0, 5), (10, 15), (20, 25)] + + def test_overlapping_spans_are_merged(self): + spans = [(0, 10), (5, 15)] + assert _merge_overlapping_spans(spans) == [(0, 15)] + + def test_adjacent_spans_are_merged(self): + spans = [(0, 5), (5, 10)] + assert _merge_overlapping_spans(spans) == [(0, 10)] + + def test_nested_spans_are_merged(self): + spans = [(0, 20), (5, 10)] + assert _merge_overlapping_spans(spans) == [(0, 20)] + + def test_multiple_overlapping_groups(self): + spans = [(0, 5), (3, 8), (10, 15), (12, 18)] + assert _merge_overlapping_spans(spans) == [(0, 8), (10, 18)] + + def test_unsorted_input_is_handled(self): + spans = [(10, 15), (0, 5), (3, 8)] + assert _merge_overlapping_spans(spans) == [(0, 8), (10, 15)] + + def test_fully_contained_spans(self): + spans = [(0, 100), (10, 20), (30, 40), (50, 60)] + assert _merge_overlapping_spans(spans) == [(0, 100)] + + +# --------------------------------------------------------------------------- +# matches_sensitive_filename +# --------------------------------------------------------------------------- + + +class TestMatchesSensitiveFilename: + def test_pem_file(self): + assert matches_sensitive_filename("server.pem") + + def test_key_file(self): + assert matches_sensitive_filename("private.key") + + def test_p12_file(self): + assert matches_sensitive_filename("cert.p12") + + def test_pfx_file(self): + assert matches_sensitive_filename("cert.pfx") + + def test_env_file(self): + assert matches_sensitive_filename(".env") + + def test_env_with_suffix(self): + assert matches_sensitive_filename(".env.production") + + def test_secret_in_name(self): + assert matches_sensitive_filename("my_secret_config.yaml") + + def test_credential_in_name(self): + assert matches_sensitive_filename("credential_store.json") + + def test_password_in_name(self): + assert matches_sensitive_filename("password_file.txt") + + def test_ssh_key(self): + assert matches_sensitive_filename("id_rsa") + + def test_ecdsa_key(self): + assert matches_sensitive_filename("id_ecdsa") + + def test_ed25519_key(self): + assert matches_sensitive_filename("id_ed25519") + + def test_normal_text_file(self): + assert not matches_sensitive_filename("readme.txt") + + def test_normal_yaml_file(self): + assert not matches_sensitive_filename("config.yaml") + + def test_normal_log_file(self): + assert not matches_sensitive_filename("output.log") + + def test_path_with_secret_in_basename(self): + assert matches_sensitive_filename("/some/path/secret_config.yaml") + + def test_case_insensitive_pem(self): + assert matches_sensitive_filename("CERT.PEM") + + def test_case_insensitive_secret(self): + assert matches_sensitive_filename("MY_SECRET.txt") + + +# --------------------------------------------------------------------------- +# Keyword pattern matching +# --------------------------------------------------------------------------- + + +class TestKeywordPatterns: + """Test that compiled keyword patterns detect sensitive content.""" + + def _matches_any_pattern(self, text: str) -> bool: + return any(p.search(text) for p in COMPILED_KEYWORD_PATTERNS) + + def test_password_pattern(self): + assert self._matches_any_pattern("password=hunter2") + + def test_password_colon(self): + assert self._matches_any_pattern("password: my_password") + + def test_api_key_pattern(self): + assert self._matches_any_pattern("api_key=abc123xyz") + + def test_api_secret_pattern(self): + assert self._matches_any_pattern("api-secret=supersecret") + + def test_token_pattern(self): + assert self._matches_any_pattern("token=eyJhbGciOiJIUzI1NiJ9") + + def test_bearer_token(self): + assert self._matches_any_pattern("Bearer eyJhbGciOiJIUzI1NiJ9") + + def test_bearer_case_insensitive(self): + # Since patterns are compiled with IGNORECASE, lowercase should match too + assert self._matches_any_pattern("bearer eyJhbGciOiJIUzI1NiJ9") + + def test_openai_key(self): + assert self._matches_any_pattern("sk-abcdefghijklmnopqrstuvwxyz123456") + + def test_github_token(self): + assert self._matches_any_pattern("ghp_ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghij") + + def test_aws_access_key(self): + assert self._matches_any_pattern("AKIAIOSFODNN7EXAMPLE") + + def test_mongodb_uri(self): + assert self._matches_any_pattern("mongodb://user:pass@host") + + def test_postgresql_uri(self): + assert self._matches_any_pattern("postgresql://admin:secret@db.example.com") + + def test_clean_text(self): + assert not self._matches_any_pattern("This is a normal log line with no secrets") + + def test_clean_json(self): + assert not self._matches_any_pattern('{"name": "test", "value": 42}') + + def test_no_duplicate_bearer_pattern(self): + # After removing the redundant lowercase bearer pattern, + # verify the single pattern still works for both cases + bearer_patterns = [p for p in KEYWORD_PATTERNS if "earer" in p] + assert len(bearer_patterns) == 1, ( + f"Expected exactly 1 bearer pattern, found {len(bearer_patterns)}: {bearer_patterns}" + ) + + +# --------------------------------------------------------------------------- +# CensoringResult +# --------------------------------------------------------------------------- + + +class TestCensoringResult: + def test_sanitized_str(self): + r = CensoringResult(Path("test.txt"), censored=True, reason="keyword", sanitized=True) + assert "SANITIZED" in str(r) + + def test_excluded_str(self): + r = CensoringResult(Path("test.txt"), censored=True, reason="keyword", sanitized=False) + assert "EXCLUDED" in str(r) + + def test_allowed_str(self): + r = CensoringResult(Path("test.txt"), censored=False, reason="clean", sanitized=False) + assert "ALLOWED" in str(r) + + +# --------------------------------------------------------------------------- +# ArtifactCensor._sanitize_file_content +# --------------------------------------------------------------------------- + + +class TestSanitizeFileContent: + def test_clean_file_passes(self, tmp_path): + f = tmp_path / "clean.txt" + f.write_text("Nothing sensitive here\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert not result.censored + assert not result.sanitized + + def test_password_is_redacted(self, tmp_path): + f = tmp_path / "config.txt" + f.write_text("password=hunter2\nother_line\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.censored + assert result.sanitized + content = f.read_text() + assert "hunter2" not in content + assert "[REDACTED]" in content + # Other content should be preserved + assert "other_line" in content + + def test_sensitive_filename_is_sanitized(self, tmp_path): + f = tmp_path / "secret_config.yaml" + f.write_text("important: data\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.censored + assert result.sanitized + content = f.read_text() + assert "Content censored by caliper" in content + assert "important: data" not in content + + def test_vault_secret_is_replaced(self, tmp_path): + f = tmp_path / "log.txt" + f.write_text("connecting with supersecrettoken123\n", encoding="utf-8") + + censor = ArtifactCensor( + vault_secrets={"supersecrettoken123"}, + secret_mapping={"supersecrettoken123": "vault/token"}, + ) + result = censor._sanitize_file_content(f) + + assert result.censored + assert result.sanitized + content = f.read_text() + assert "supersecrettoken123" not in content + assert "*******" in content + + def test_binary_file_skipped(self, tmp_path): + f = tmp_path / "image.dat" + f.write_bytes(bytes(range(256))) + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert not result.censored + assert not result.sanitized + + def test_dry_run_does_not_modify_file(self, tmp_path): + original_content = "password=hunter2\n" + f = tmp_path / "config.txt" + f.write_text(original_content, encoding="utf-8") + + censor = ArtifactCensor(dry_run=True) + result = censor._sanitize_file_content(f) + + assert result.censored + assert result.sanitized + # File should NOT be modified in dry run + assert f.read_text() == original_content + + def test_overlapping_patterns_produce_correct_output(self, tmp_path): + # token=xxx matches "token" pattern. "access_token=xxx" also matches. + # These could overlap if the text has "access_token=myvalue" + f = tmp_path / "config.txt" + f.write_text("access_token=myvalue123\nother content\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.sanitized + content = f.read_text() + # The redacted content should not be corrupted + assert "[REDACTED]" in content + assert "other content" in content + # Original secret should be gone + assert "myvalue123" not in content + + def test_multiple_secrets_on_same_line(self, tmp_path): + f = tmp_path / "multi.txt" + f.write_text("password=abc api_key=xyz\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.sanitized + content = f.read_text() + assert "abc" not in content + assert "xyz" not in content + assert content.count("[REDACTED]") >= 2 + + +# --------------------------------------------------------------------------- +# ArtifactCensor.censor_files +# --------------------------------------------------------------------------- + + +class TestCensorFiles: + def test_separates_clean_sanitized_excluded(self, tmp_path): + """Verify that clean, sanitized, and excluded files are handled correctly.""" + clean = tmp_path / "readme.txt" + clean.write_text("Hello world\n", encoding="utf-8") + + sensitive_content = tmp_path / "config.yaml" + sensitive_content.write_text("password=mysecret\n", encoding="utf-8") + + sensitive_name = tmp_path / "secret_data.yaml" + sensitive_name.write_text("key: value\n", encoding="utf-8") + + censor = ArtifactCensor() + processed, results = censor.censor_files([clean, sensitive_content, sensitive_name]) + + # All three should be in processed (clean + 2 sanitized) + assert len(processed) == 3 + assert clean in processed + assert sensitive_content in processed + assert sensitive_name in processed + + # Check result classifications + clean_results = [r for r in results if not r.censored] + sanitized_results = [r for r in results if r.sanitized] + + assert len(clean_results) == 1 + assert len(sanitized_results) == 2 + + def test_nonexistent_file_is_passed_through(self, tmp_path): + nonexistent = tmp_path / "does_not_exist.txt" + censor = ArtifactCensor() + processed, results = censor.censor_files([nonexistent]) + + assert nonexistent in processed + assert len(results) == 0 + + def test_verbose_logging(self, tmp_path): + f = tmp_path / "config.txt" + f.write_text("password=test\n", encoding="utf-8") + + censor = ArtifactCensor(verbose=True) + processed, results = censor.censor_files([f]) + + assert len(processed) == 1 + assert results[0].sanitized + + +# --------------------------------------------------------------------------- +# apply_censoring_to_artifacts +# --------------------------------------------------------------------------- + + +class TestApplyCensoringToArtifacts: + def test_censoring_disabled_returns_all(self, tmp_path): + f = tmp_path / "secret.pem" + f.write_text("private key data\n", encoding="utf-8") + + paths, results = apply_censoring_to_artifacts([f], censoring_enabled=False, verbose=True) + + assert f in paths + assert len(results) == 0 + + def test_end_to_end_with_mixed_files(self, tmp_path): + """End-to-end test with a directory containing various file types.""" + # Clean file + clean = tmp_path / "output.log" + clean.write_text("Test completed successfully\n", encoding="utf-8") + + # File with password + with_secret = tmp_path / "app.conf" + with_secret.write_text("database_password=abc123\nhost=localhost\n", encoding="utf-8") + + # File with sensitive filename + sensitive_name = tmp_path / "credentials.json" + sensitive_name.write_text('{"user": "admin"}\n', encoding="utf-8") + + # Binary file + binary = tmp_path / "image.png" + binary.write_bytes(b"\x89PNG\r\n\x1a\n" + bytes(100)) + + all_paths = [clean, with_secret, sensitive_name, binary] + processed, results = apply_censoring_to_artifacts(all_paths, censoring_enabled=True) + + # Clean and binary files pass through; secret-content and sensitive-name are sanitized + assert clean in processed + assert binary in processed + + # Verify content was sanitized + assert "abc123" not in with_secret.read_text() + assert "[REDACTED]" in with_secret.read_text() + + # Verify sensitive filename content was replaced + assert "Content censored by caliper" in sensitive_name.read_text() + + def test_dry_run_preserves_all_files(self, tmp_path): + f = tmp_path / "config.txt" + original = "password=secret123\n" + f.write_text(original, encoding="utf-8") + + processed, results = apply_censoring_to_artifacts([f], censoring_enabled=True, dry_run=True) + + # File content should be unchanged in dry run + assert f.read_text() == original + # But results should still report what would be censored + assert len(results) == 1 + assert results[0].sanitized + + def test_vault_secrets_are_censored(self, tmp_path): + f = tmp_path / "log.txt" + # Use content that won't match keyword patterns but contains a vault secret + f.write_text("Connecting to host with my-vault-secret-value as auth\n", encoding="utf-8") + + processed, results = apply_censoring_to_artifacts( + [f], + censoring_enabled=True, + vault_secrets={"my-vault-secret-value"}, + secret_mapping={"my-vault-secret-value": "test-vault/api-key"}, + ) + + content = f.read_text() + assert "my-vault-secret-value" not in content + assert "*******" in content + + def test_summary_counts_are_correct(self, tmp_path): + # Create files of each type + clean = tmp_path / "clean.txt" + clean.write_text("all good\n", encoding="utf-8") + + sensitive = tmp_path / "app.cfg" + sensitive.write_text("api_key=abc\n", encoding="utf-8") + + _, results = apply_censoring_to_artifacts([clean, sensitive], censoring_enabled=True) + + sanitized_count = len([r for r in results if r.sanitized]) + excluded_count = len([r for r in results if r.censored and not r.sanitized]) + clean_count = len([r for r in results if not r.censored]) + + assert clean_count == 1 + assert sanitized_count == 1 + assert excluded_count == 0 + + +# --------------------------------------------------------------------------- +# End-to-end directory censoring +# --------------------------------------------------------------------------- + + +class TestEndToEndDirectoryCensoring: + def test_full_directory_scan(self, tmp_path): + """Simulate the full censoring workflow on a directory tree.""" + # Create a realistic artifact tree + subdir = tmp_path / "test_run" / "artifacts" + subdir.mkdir(parents=True) + + (subdir / "pod_logs.txt").write_text("INFO: Pod started\nINFO: Pod ready\n") + (subdir / "config.yaml").write_text("password: supersecret\nhost: example.com\n") + (subdir / "results.json").write_text('{"score": 95, "status": "pass"}\n') + (subdir / "server.pem").write_text("-----BEGIN CERTIFICATE-----\nfake\n") + + # Collect all files + all_paths = list(tmp_path.rglob("*")) + file_paths = [p for p in all_paths if p.is_file()] + + processed, results = apply_censoring_to_artifacts(file_paths, censoring_enabled=True) + + # All files should be in processed (all get sanitized, none fully excluded) + assert len(processed) == 4 + + # Check specific files + config_content = (subdir / "config.yaml").read_text() + assert "supersecret" not in config_content + assert "[REDACTED]" in config_content + assert "host: example.com" in config_content + + pem_content = (subdir / "server.pem").read_text() + assert "Content censored by caliper" in pem_content + + pod_content = (subdir / "pod_logs.txt").read_text() + assert "Pod started" in pod_content # Clean content preserved + + def test_read_only_file_handling(self, tmp_path): + """Test that read-only files can be sanitized.""" + import stat + + f = tmp_path / "readonly_config.txt" + f.write_text("password=readonly_secret\n", encoding="utf-8") + f.chmod(stat.S_IRUSR | stat.S_IRGRP | stat.S_IROTH) + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.sanitized + content = f.read_text() + assert "readonly_secret" not in content + assert "[REDACTED]" in content + + def test_empty_file_passes_clean(self, tmp_path): + f = tmp_path / "empty.txt" + f.write_text("", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert not result.censored + assert not result.sanitized + + def test_utf8_content_preserved(self, tmp_path): + f = tmp_path / "unicode.txt" + f.write_text("Ünîcödé content: password=test123\nMore: über cool\n", encoding="utf-8") + + censor = ArtifactCensor() + result = censor._sanitize_file_content(f) + + assert result.sanitized + content = f.read_text() + assert "test123" not in content + assert "über cool" in content diff --git a/projects/core/ci_entrypoint/prepare_ci.py b/projects/core/ci_entrypoint/prepare_ci.py index d8ebfa7c3..0d91c4d05 100755 --- a/projects/core/ci_entrypoint/prepare_ci.py +++ b/projects/core/ci_entrypoint/prepare_ci.py @@ -325,10 +325,10 @@ def ci_banner(project: str, operation: str, args: list[str]): base_sha = os.environ.get("PULL_BASE_SHA", "main") if base_sha == "main": - logger.warning("PULL_BASE_SHA not set. Showing the last commits from main.") + logger.info("PULL_BASE_SHA not set. Showing the last commits from main.") pull_sha = os.environ.get("PULL_PULL_SHA", "") if not pull_sha: - logger.warning("PULL_PULL_SHA not set. Showing the last commits from main.") + logger.info("PULL_PULL_SHA not set. Showing the last commits from main.") logger.info(f"Git command will be: git show --quiet --oneline {base_sha}..{pull_sha}") diff --git a/projects/core/library/ci.py b/projects/core/library/ci.py index d58ec9e04..3abea996d 100644 --- a/projects/core/library/ci.py +++ b/projects/core/library/ci.py @@ -169,12 +169,19 @@ def safe_ci_entrypoint(command_func): @functools.wraps(command_func) def wrapper(*args, **kwargs): exit_code = 0 + reason = None try: result = command_func(*args, **kwargs) - exit_code = result if result is not None else 0 + if result is None: + exit_code = 0 + elif isinstance(result, tuple) and len(result) == 2: + exit_code, reason = result + else: + exit_code = result except Exception as e: handle_ci_exception(e) exit_code = 1 + reason = str(e) # Save exit status to YAML file try: @@ -183,11 +190,18 @@ def wrapper(*args, **kwargs): exit_status_file = metadata_dir / "exit_status.yaml" exit_status_data = {"return_code": exit_code} + if reason is not None: + exit_status_data["reason"] = reason with open(exit_status_file, "w", encoding="utf-8") as f: yaml.dump(exit_status_data, f, default_flow_style=False) - logger.info(f"Exit status saved: {exit_status_file} (return_code: {exit_code})") + if reason: + logger.info( + f"Exit status saved: {exit_status_file} (return_code: {exit_code}, reason: {reason})" + ) + else: + logger.info(f"Exit status saved: {exit_status_file} (return_code: {exit_code})") except Exception as save_error: logger.warning(f"Failed to save exit status: {save_error}") @@ -232,6 +246,7 @@ def add_notification_file( f.write(message) logger.info(f"Created notification file: {file_path}") + logger.info(f"{message}") return str(file_path) except Exception as e: diff --git a/projects/core/library/config.py b/projects/core/library/config.py index 311d480d5..c4a7f5bd8 100644 --- a/projects/core/library/config.py +++ b/projects/core/library/config.py @@ -164,40 +164,60 @@ def apply_config_overrides( logger.fatal(msg) raise ValueError(msg) - for key, value in variable_overrides.items(): - MAGIC_DEFAULT_VALUE = object() - handled_secretly = True # current_value MUST NOT be printed below. - current_value = self.get_config( - key, - MAGIC_DEFAULT_VALUE, - print=False, - warn=False, - handled_secretly=handled_secretly, - ) - if current_value == MAGIC_DEFAULT_VALUE: - try: - # Try to create the key if parent exists and is a dict - self._create_first_parent_config_key(key, value) - self.save_config() - except ValueError: - if not ignore_not_found: - raise + # Setup presets_applied.txt file for writing variable overrides + dest_txt = env.ARTIFACT_DIR / CI_METADATA_DIRNAME / "presets_applied.txt" + dest_txt.parent.mkdir(parents=True, exist_ok=True) + + # Collect all override messages to write to file once + file_messages = [] + try: + for key, value in variable_overrides.items(): + MAGIC_DEFAULT_VALUE = object() + handled_secretly = True # current_value MUST NOT be printed below. + current_value = self.get_config( + key, + MAGIC_DEFAULT_VALUE, + print=False, + warn=False, + handled_secretly=handled_secretly, + ) + if current_value == MAGIC_DEFAULT_VALUE: + try: + # Try to create the key if parent exists and is a dict + self._create_first_parent_config_key(key, value) + self.save_config() + except ValueError: + if not ignore_not_found: + raise + + if log: + msg = f"config override IGNORED: {key} --> {value}" + logger.info(msg) + file_messages.append(msg) + continue + + self.save_config() if log: - logger.info(f"config override IGNORED: {key} --> {value}") + msg = f"config override (new key): {key} --> {value}" + logger.info(msg) + file_messages.append(msg) continue - self.save_config() + self.set_config(key, value, print=False) + actual_value = self.get_config( + key, print=False + ) # ensure that key has been set, raises an exception otherwise if log: - logger.info(f"config override (new key): {key} --> {value}") - continue - - self.set_config(key, value, print=False) - actual_value = self.get_config( - key, print=False - ) # ensure that key has been set, raises an exception otherwise - if log: - logger.info(f"config override: {key} --> {actual_value}") + msg = f"config override: {key} --> {actual_value}" + logger.info(msg) + file_messages.append(msg) + finally: + # Write all collected messages to file, even if processing failed + if file_messages: + with open(dest_txt, "a") as f: + for msg in file_messages: + print(msg, file=f) def apply_preset(self, name): values = self.get_preset(name) @@ -208,6 +228,9 @@ def apply_preset(self, name): dest_txt = env.ARTIFACT_DIR / CI_METADATA_DIRNAME / "presets_applied.txt" dest_txt.parent.mkdir(parents=True, exist_ok=True) + # Collect preset messages to write to file once + preset_messages = [] + for key, value in values.items(): if key == "extends": for extend_name in value or []: @@ -216,12 +239,16 @@ def apply_preset(self, name): msg = f"preset[{name}] {key} --> {value}" logger.info(msg) - - with open(dest_txt, "a") as f: - print(msg, file=f) + preset_messages.append(msg) self.set_config(key, value, print=False) + # Write all collected preset messages to file once + if preset_messages: + with open(dest_txt, "a") as f: + for msg in preset_messages: + print(msg, file=f) + def load_presets(self, preset_dir): for preset_file in preset_dir.glob("*.yaml"): with open(preset_file) as preset_f: diff --git a/projects/core/library/export.py b/projects/core/library/export.py index bbdb4a3fd..18a567f4f 100644 --- a/projects/core/library/export.py +++ b/projects/core/library/export.py @@ -16,21 +16,17 @@ import click import yaml -from projects.caliper.orchestration.export import run_from_orchestration_config -from projects.core.ci_entrypoint.prepare_ci import CI_METADATA_DIRNAME +from projects.caliper.orchestration.export import ( + ExportFailedException, + run_from_orchestration_config, +) from projects.core.library import ci as ci_lib from projects.core.library import config, env, run - - -class StepStatus(StrEnum): - """Status of a step execution.""" - - SUCCESS = "success" - FAILURE = "failure" - ONGOING = "ongoing" - UNKNOWN = "unknown" - WARNING = "warning" - +from projects.core.library.export_notifications import ( + _check_job_shutdown_status, + _create_mlflow_file_url_for_step, + send_notification, +) logger = logging.getLogger(__name__) @@ -98,329 +94,6 @@ def _update_fjob_export_status(status: dict): os.environ["KUBECONFIG"] = original_kubeconfig -def send_notification(status: dict[str, Any], notification_provider=None) -> bool: - """Send job completion notifications based on caliper export status. - - Args: - status: Caliper export status object containing backend results and metadata - notification_provider: Optional per-project SlackNotificationProvider instance - - Returns: - bool: True if notifications were sent successfully, False otherwise - """ - # Extract notification parameters from status object - project = _extract_project_from_status(status) - operation = _extract_operation_from_status(status) - finish_reason = _extract_finish_reason_from_status(status) - duration_str = _extract_duration_from_status(status) - - # Apply minimal filtering logic - if _should_skip_notification(project, operation, finish_reason): - logger.info(f"Skipping notification for {project} {operation}") - return True # Skipped is considered success - - # Send actual notifications - notification_success = True - logger.info(f"Sending notification: {project} {operation} {finish_reason}{duration_str}") - - # Build enhanced notification with fournos job info and artifact links - notification_status = _build_enhanced_notification(project, finish_reason, duration_str, status) - - # Write notification to file for GitHub pickup - try: - if env.ARTIFACT_DIR: - notification_file = Path(env.ARTIFACT_DIR) / "NOTIFICATION-github.md" - with open(notification_file, "w", encoding="utf-8") as f: - f.write(notification_status) - logger.info(f"Wrote export notification file {notification_file}") - else: - logger.warning("ARTIFACT_DIR not available, skipping notification file") - except Exception as e: - logger.warning(f"Failed to write notification file: {e}") - - # Actually send notification through GitHub API - try: - from projects.core.notifications.send import send_notification as send_github_notification - - # Get notification vault from configuration - notification_vault = None - try: - from projects.core.library import config - - notification_config = config.project.get_config("caliper.export.notifications", {}) - notification_vault = notification_config.get("vault") - if notification_vault: - logger.info(f"Using notification vault from config: {notification_vault}") - except Exception as e: - logger.warning(f"Failed to get notification vault from config: {e}") - - success = send_github_notification( - message=notification_status, - github=True, - slack=False, - dry_run=False, - notification_vault=notification_vault, - ) - if success: - logger.info("Successfully sent GitHub notification") - else: - logger.error("GitHub notification sending failed") - notification_success = False - except Exception as e: - logger.error(f"Failed to send GitHub notification: {e}") - notification_success = False - - # Per-project Slack notification via provider - if notification_provider: - try: - from projects.core.notifications.provider import NotificationContext - - artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None - context = NotificationContext( - status=status, - finish_reason=str(finish_reason), - project_name=project or "unknown", - pr_number=os.environ.get("PULL_NUMBER"), - job_type=os.environ.get("JOB_TYPE"), - artifact_dir=artifact_dir, - ) - ok = notification_provider.notify(context) - if ok: - logger.info("Successfully sent per-project Slack notification") - else: - logger.warning("Per-project Slack notification failed") - notification_success = False - except Exception as e: - logger.warning(f"Failed to send per-project Slack notification: {e}") - notification_success = False - - return notification_success - - -def _get_project_and_args(project: str) -> tuple[str, str]: - """Extract project name and args from fournos job or config.""" - fjob_project = project - fjob_args_str = "" - - try: - metadata_dir = ci_lib.get_ci_metadata_dir() - fournos_fjob_path = metadata_dir / "fournos_fjob.yaml" - if not fournos_fjob_path.exists(): - return fjob_project, fjob_args_str - - with open(fournos_fjob_path, encoding="utf-8") as f: - fjob_data = yaml.safe_load(f) - - display_name = fjob_data.get("spec", {}).get("displayName", "") - if not display_name: - return fjob_project, fjob_args_str - - parts = display_name.split() - if not parts: - return fjob_project, fjob_args_str - - fjob_project = parts[0] - fjob_args_str = " ".join(parts[1:]) if len(parts) > 1 else "" - except Exception as e: - logger.warning(f"Failed to read fournos job for project/args: {e}") - - if fjob_args_str: - return fjob_project, fjob_args_str - - try: - from projects.core.library import config - - job_args = config.project.get_config("ci_job.args", [], warn=False) - fjob_args_str = " ".join(job_args) if job_args else "" - except Exception as e: - logger.warning(f"Failed to get args from config: {e}") - - return fjob_project, fjob_args_str - - -def _get_execution_engine_config() -> str | None: - """Read and format execution engine configuration.""" - try: - metadata_dir = ci_lib.get_ci_metadata_dir() - fournos_fjob_path = metadata_dir / "fournos_fjob.yaml" - if not fournos_fjob_path.exists(): - return None - - with open(fournos_fjob_path, encoding="utf-8") as f: - fjob_data = yaml.safe_load(f) - - execution_engine = fjob_data.get("spec", {}).get("executionEngine", {}) - if not execution_engine: - return None - - engine_yaml = yaml.dump(execution_engine, default_flow_style=False, sort_keys=True) - return f"```yaml\n{engine_yaml.strip()}\n```" - except Exception as e: - logger.warning(f"Failed to read fournos job config: {e}") - return None - - -def _check_job_shutdown_status() -> dict[str, Any] | None: - """Check if the job has been aborted via spec.shutdown field.""" - try: - metadata_dir = ci_lib.get_ci_metadata_dir() - fournos_fjob_path = metadata_dir / "fournos_fjob.yaml" - if not fournos_fjob_path.exists(): - return None - - with open(fournos_fjob_path, encoding="utf-8") as f: - fjob_data = yaml.safe_load(f) - - shutdown_value = fjob_data.get("spec", {}).get("shutdown") - if shutdown_value: - return { - "shutdown_detected": True, - "shutdown_value": shutdown_value, - "is_aborted": shutdown_value.lower() == "stop", - } - - return {"shutdown_detected": False, "shutdown_value": None, "is_aborted": False} - except Exception as e: - logger.warning(f"Failed to check job shutdown status: {e}") - return None - - -def _extract_artifact_links(status: dict[str, Any]) -> tuple[list[str], str | None]: - """Extract artifact links and MLflow URL from status.""" - artifact_links = [] - mlflow_run_url = None - - caliper_export = status.get("caliper_artifacts_export", {}) - backends = caliper_export.get("backends", {}) - - for backend_name, backend_result in backends.items(): - if not isinstance(backend_result, dict): - continue - - if backend_result.get("experiment_url"): - artifact_links.append( - f"[{backend_name} Experiment]({backend_result['experiment_url']})" - ) - - if backend_result.get("run_url"): - mlflow_run_url = backend_result["run_url"] - artifact_links.append(f"[{backend_name} Results]({mlflow_run_url})") - elif backend_result.get("artifact_url"): - artifact_links.append(f"[{backend_name} Artifacts]({backend_result['artifact_url']})") - elif backend_result.get("dashboard_url"): - artifact_links.append(f"[{backend_name} Dashboard]({backend_result['dashboard_url']})") - - if status.get("artifact_url"): - artifact_links.append(f"[Artifacts]({status['artifact_url']})") - - return artifact_links, mlflow_run_url - - -def _create_mlflow_url(mlflow_run_url: str, step_dir_name: str) -> str | None: - """Create MLflow URL for step logs.""" - if "/artifacts" not in mlflow_run_url: - logger.warning(f"Unexpected MLflow URL format: {mlflow_run_url}") - return None - - if "#" in mlflow_run_url: - base_domain, hash_fragment = mlflow_run_url.split("#", 1) - if "/artifacts" not in hash_fragment: - raise ValueError("Artifacts not found in hash fragment") - - hash_base, params = hash_fragment.split("/artifacts", 1) - workspace_param = params if "?workspace=" in params else "" - return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}/run.log{workspace_param}" - else: - base_url, params = mlflow_run_url.split("/artifacts", 1) - workspace_param = params if "?workspace=" in params else "" - return f"{base_url}/artifacts/{step_dir_name}/run.log{workspace_param}" - - -def _create_mlflow_step_url(mlflow_run_url: str, step_dir_name: str) -> str | None: - """Create MLflow URL for step directory (for file access).""" - if "/artifacts" not in mlflow_run_url: - logger.warning(f"Unexpected MLflow URL format: {mlflow_run_url}") - return None - - if "#" in mlflow_run_url: - base_domain, hash_fragment = mlflow_run_url.split("#", 1) - if "/artifacts" not in hash_fragment: - raise ValueError("Artifacts not found in hash fragment") - - hash_base, artifacts_part = hash_fragment.split("/artifacts", 1) - # Extract workspace parameter if present, ignoring existing path - workspace_param = "" - if "?workspace=" in artifacts_part: - workspace_param = artifacts_part[artifacts_part.find("?") :] - return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}{workspace_param}" - else: - base_url, artifacts_part = mlflow_run_url.split("/artifacts", 1) - # Extract workspace parameter if present, ignoring existing path - workspace_param = "" - if "?workspace=" in artifacts_part: - workspace_param = artifacts_part[artifacts_part.find("?") :] - return f"{base_url}/artifacts/{step_dir_name}{workspace_param}" - - -def _create_mlflow_file_url_for_step( - mlflow_run_url: str, step_dir_name: str, file_path: str -) -> str: - """Create MLflow URL for a specific file within a step directory. - - Args: - mlflow_run_url: Base MLflow run URL - step_dir_name: Name of the step directory - file_path: Relative path to file from step directory - - Returns: - Full MLflow URL to the file - - Raises: - ValueError: If URL format is unexpected - """ - if "/artifacts" not in mlflow_run_url: - raise ValueError(f"Unexpected MLflow URL format: {mlflow_run_url}") - - # Clean file path - file_clean = file_path.lstrip("/") - - if "#" in mlflow_run_url: - base_domain, hash_fragment = mlflow_run_url.split("#", 1) - if "/artifacts" not in hash_fragment: - raise ValueError("Artifacts not found in hash fragment") - - hash_base, artifacts_part = hash_fragment.split("/artifacts", 1) - # Extract workspace parameter if present, ignoring existing path - workspace_param = "" - if "?workspace=" in artifacts_part: - workspace_param = artifacts_part[artifacts_part.find("?") :] - return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}/{file_clean}{workspace_param}" - else: - base_url, artifacts_part = mlflow_run_url.split("/artifacts", 1) - # Extract workspace parameter if present, ignoring existing path - workspace_param = "" - if "?workspace=" in artifacts_part: - workspace_param = artifacts_part[artifacts_part.find("?") :] - return f"{base_url}/artifacts/{step_dir_name}/{file_clean}{workspace_param}" - - -def _read_step_duration(step_dir: Path) -> str: - """Read step duration from timing file.""" - timing_file = step_dir / CI_METADATA_DIRNAME / "test_duration.yaml" - if not timing_file.exists(): - return "" - - try: - with open(timing_file, encoding="utf-8") as f: - timing_data = yaml.safe_load(f) - - formatted_duration = timing_data.get("duration", {}).get("formatted") - return formatted_duration or "" - except Exception as timing_error: - logger.warning(f"Failed to read timing file {timing_file}: {timing_error}") - return "" - - def _process_caliper_postprocess_status( step_dir: Path, step_log_links: list[str], mlflow_run_url: str | None = None ) -> None: @@ -457,6 +130,7 @@ def _process_caliper_postprocess_status( if mlflow_run_url: # Use base_directory from status data for MLflow URL construction base_directory = result.base_directory + if base_directory: # Calculate path relative to BASE_ARTIFACT_DIR.parent # e.g., "/workspace/artifacts/000__replot/postprocess_output" -> "000__replot/postprocess_output" @@ -466,7 +140,15 @@ def _process_caliper_postprocess_status( # Calculate step subdirectory relative to BASE_ARTIFACT_DIR.parent # e.g., "/workspace/artifacts/000__replot/postprocess_output" relative to "/workspace/artifacts" = "000__replot/postprocess_output" - step_subdir = str(base_path.relative_to(env.BASE_ARTIFACT_DIR.parent)) + try: + step_subdir = str(base_path.relative_to(env.BASE_ARTIFACT_DIR.parent)) + except ValueError as e: + # Path resolution failed (common in dry-run or different directory contexts) + logger.warning( + f"Failed to resolve path {base_path} relative to {env.BASE_ARTIFACT_DIR.parent}: {e}" + ) + # Fallback: use the step directory name + step_subdir = step_dir.name else: # Fallback to step_dir.name for backward compatibility step_subdir = step_dir.name @@ -484,368 +166,9 @@ def get_file_link(file_path: str, step_subdir=step_subdir) -> str: raise -def _process_notification_files(step_dir: Path, step_log_links: list[str]) -> None: - """Process notification files from step directory.""" - notifications_dir = step_dir / CI_METADATA_DIRNAME / "notifications" - if not (notifications_dir.exists() and notifications_dir.is_dir()): - return - - import re - - for notification_file in sorted(notifications_dir.glob("*.txt")): - try: - with open(notification_file, encoding="utf-8") as f: - content = f.read().strip() - - if not content: - continue - - subtitle = notification_file.stem.replace("__", " ").replace("_", " ").title() - subtitle = re.sub(r"^\d+\s+", "", subtitle) - step_log_links.append(f"##### {subtitle}") - - for line in content.splitlines(): - step_log_links.append(f"> {line}") - - except Exception as file_error: - logger.warning(f"Failed to read notification file {notification_file}: {file_error}") - continue - - -def _process_step_logs(mlflow_run_url: str) -> list[str]: - """Process step logs from parent directory.""" - if not mlflow_run_url: - logging.warning("mlflow_run_url not set, skipping step log browsing") - return [] - - step_log_links = [] - parent_dir = Path(env.BASE_ARTIFACT_DIR).parent - current_step_name = Path(env.BASE_ARTIFACT_DIR).name - - for step_dir in sorted(parent_dir.iterdir()): - if not step_dir.is_dir(): - continue - if step_dir.name.startswith("."): - continue - - run_log = step_dir / "run.log" - if not run_log.exists(): - continue - - try: - mlflow_log_url = _create_mlflow_url(mlflow_run_url, step_dir.name) - if not mlflow_log_url: - continue - - step_name = step_dir.name.replace("__", " ").replace("_", " ").title() - duration_str = _read_step_duration(step_dir) - exit_status_emoji, exit_status = _read_step_exit_status(step_dir, current_step_name) - - if duration_str: - step_log_links.append( - f"#### {exit_status_emoji} [{step_name}]({mlflow_log_url}) `{duration_str}`" - ) - else: - step_log_links.append(f"#### {exit_status_emoji} [{step_name}]({mlflow_log_url})") - - _process_notification_files(step_dir, step_log_links) - - except Exception as e: - logger.warning(f"Failed to create MLflow link for {run_log}: {e}") - continue - - return step_log_links - - -def _process_postprocess_status(mlflow_run_url: str | None = None) -> list[str]: - """Process post-processing status from all step directories.""" - if not mlflow_run_url: - return [] - - postprocess_links = [] - parent_dir = Path(env.BASE_ARTIFACT_DIR).parent - - for step_dir in sorted(parent_dir.iterdir()): - if not step_dir.is_dir(): - continue - if step_dir.name.startswith("."): - continue - - try: - _process_caliper_postprocess_status(step_dir, postprocess_links, mlflow_run_url) - except Exception as e: - logger.error(f"Failed to process postprocess status for {step_dir.name}: {e}") - raise - - return postprocess_links - - -def _read_step_exit_status( - step_dir: Path, current_step_name: str | None = None -) -> tuple[str, StepStatus]: - """Read exit status from step directory and return emoji and status enum.""" - try: - exit_status_file = step_dir / CI_METADATA_DIRNAME / "exit_status.yaml" - if not exit_status_file.exists(): - # Check if this is the current ongoing step - if current_step_name and step_dir.name == current_step_name: - return "🔄", StepStatus.ONGOING # Ongoing step - return "❓", StepStatus.UNKNOWN # Unknown status if file doesn't exist - - with open(exit_status_file, encoding="utf-8") as f: - exit_data = yaml.safe_load(f) - - return_code = exit_data.get("return_code") - if return_code is None or return_code == 0: - return "✅", StepStatus.SUCCESS - else: - return "❌", StepStatus.FAILURE - except Exception as e: - logger.warning(f"Failed to read exit status from {step_dir}: {e}") - # Check if this is the current ongoing step even on error - if current_step_name and step_dir.name == current_step_name: - return "🔄", StepStatus.ONGOING # Ongoing step - return "❓", StepStatus.UNKNOWN # Unknown status on error - - -def _check_postprocess_warnings(step_dir: Path) -> StepStatus: - """Check for warning status in postprocess status file.""" - - status = StepStatus.SUCCESS # No postprocess warning/error, assume no warnings - for status_file in step_dir.glob("**/postprocess_status.yaml"): - try: - with open(status_file, encoding="utf-8") as f: - status_data = yaml.safe_load(f) - except Exception as e: - logging.error(f"Failed to read {status_file} as yaml: {e}") - status = StepStatus.WARNING - continue - - if not status_data: - continue - - # Check top-level success field for warning value - success_value = status_data.get("success") - if success_value == "warning": - logging.warning( - f"Post-process warning detected in {status_file}, setting the WARNING flag" - ) - status = StepStatus.WARNING - - if success_value in ("failure", "error"): - logging.error(f"Post-process {success_value} detected, raising the FAILURE flag") - return StepStatus.FAILURE - - return status - - -def _get_overall_status_from_steps() -> str: - """Check all step exit statuses and return overall status emoji.""" - try: - parent_dir = Path(env.BASE_ARTIFACT_DIR).parent - current_step_name = Path(env.BASE_ARTIFACT_DIR).name - - step_statuses = [] - - for step_dir in sorted(parent_dir.iterdir()): - if not step_dir.is_dir(): - continue - if step_dir.name.startswith("."): - continue - - # Only check directories that have run.log (actual steps) - run_log = step_dir / "run.log" - if not run_log.exists(): - continue - - _emoji, status = _read_step_exit_status(step_dir, current_step_name) - - step_statuses.append(status) - - # Check for postprocess warnings in this step (always check, regardless of exit status) - postprocess_status = _check_postprocess_warnings(step_dir) - step_statuses.append(postprocess_status) - - # Priority: failure > ongoing > warning > unknown > success - if StepStatus.FAILURE in step_statuses: - return "🔴" # Any failure = red - elif StepStatus.WARNING in step_statuses: - return "🟠" # Warning = orange - elif StepStatus.UNKNOWN in step_statuses: - return "🟠" # Unknown = orange - elif StepStatus.ONGOING in step_statuses: - return "🟢" # Ongoing --> success - else: - return "🟢" # All successful = green - - except Exception as e: - logger.warning(f"Failed to check step statuses: {e}") - return "🔴" # Error checking = red - - -def _build_enhanced_notification( - project: str, finish_reason: FinishReason, duration_str: str, status: dict[str, Any] -) -> str: - """Build enhanced notification with fournos job config and artifact links.""" - fjob_project, fjob_args_str = _get_project_and_args(project) - - # Check for job shutdown first (takes highest priority) - shutdown_status = _check_job_shutdown_status() - if shutdown_status and shutdown_status.get("is_aborted"): - status_emoji = "🛑" # Abort status overrides everything - else: - # Check all step statuses for overall status emoji (takes priority over finish_reason) - status_emoji = _get_overall_status_from_steps() - - base_status = f"**{status_emoji} Execution of `{fjob_project}` {fjob_args_str} {status_emoji}**" - - notification_parts = [base_status, "---"] - - # Add job abort message right below overall status if applicable - if shutdown_status and shutdown_status.get("is_aborted"): - shutdown_value = shutdown_status.get("shutdown_value", "Stop") - notification_parts.append(f"🛑 **JOB ABORTED** - `spec.shutdown={shutdown_value}`") - notification_parts.append("") - - execution_engine_config = _get_execution_engine_config() - if execution_engine_config: - notification_parts.append("**Execution Engine Configuration**") - notification_parts.append(execution_engine_config) - - try: - artifact_links, mlflow_run_url = _extract_artifact_links(status) - step_log_links = _process_step_logs(mlflow_run_url) - postprocess_status_links = _process_postprocess_status(mlflow_run_url) - - if artifact_links: - notification_parts.append("") - notification_parts.append("**Artifact Links**") - notification_parts.extend([f"* {link}" for link in artifact_links]) - else: - notification_parts.append("**Artifact Links:** No direct links available") - - if step_log_links: - notification_parts.append("") - notification_parts.append("**Test Logs**") - notification_parts.extend(step_log_links) - - # Add distinct test and post-processing status right under Test Logs - test_status_section = _build_test_status_section(status) - if test_status_section: - notification_parts.append("") - notification_parts.extend(test_status_section) - - if postprocess_status_links: - notification_parts.append("") - notification_parts.extend(postprocess_status_links) - - except Exception as e: - logger.warning(f"Failed to extract artifact links: {e}") - notification_parts.append("**Artifact Links:** Error extracting links") - - return "\n".join(notification_parts) - - -def _build_test_status_section(status: dict[str, Any]) -> list[str]: - """Build distinct test and post-processing status section.""" - try: - test_phase = status.get("test_phase", {}) - if not test_phase: - return [] - - test_status = test_phase.get("phase", "UNKNOWN") - test_message = test_phase.get("message", "") - - # Determine post-processing status based on final status and test outcome - final_status = status.get("final_status", "unknown") - if test_status == "FAILED": - post_processing_status = "skipped" # Don't run post-processing if test failed - elif final_status == "success": - post_processing_status = "success" - elif "failed" in final_status.lower(): - post_processing_status = "failed" - else: - post_processing_status = "unknown" - - status_lines = [f"**test:** {test_status}"] - - if test_message: - # Format message with blockquote-style prefix - status_lines.append(f"> {test_message}") - - status_lines.append(f"**post-processing:** {post_processing_status}") - - return status_lines - - except Exception as e: - logger.warning(f"Failed to build test status section: {e}") - return [] - - -def _extract_project_from_status(status: dict[str, Any]) -> str: - """Extract project name from status object or environment.""" - # Try to get project from environment variables - project = os.environ.get("PROJECT_NAME") - if project: - return project - - # Fallback to JOB_NAME parsing (common in CI environments) - job_name = os.environ.get("JOB_NAME", "") - if job_name and "-" in job_name: - # Extract project from job name pattern like "project-operation-variant" - return job_name.split("-")[0] - - return "unknown" - - -def _extract_operation_from_status(status: dict[str, Any]) -> str: - """Extract operation name from status object.""" - return "export-artifacts" - - -def _extract_finish_reason_from_status(status: dict[str, Any]) -> FinishReason: - """Extract finish reason from status object.""" - # Check if any backend failed in the status - if not status: - return FinishReason.ERROR - - # Look for backend results - backends = status.get("backends", {}) - for backend_name, backend_result in backends.items(): - # Check both explicit success flag and status field - if backend_result.get("success") is False or backend_result.get("status") not in ( - None, - "success", - ): - logger.info(f"Backend {backend_name} failed, marking as error") - return FinishReason.ERROR - - return FinishReason.SUCCESS - - -def _extract_duration_from_status(status: dict[str, Any]) -> str: - """Extract duration from status object.""" - # Look for duration in status - duration = status.get("duration") - if duration: - return f" after {duration}" - return "" - - -def _should_skip_notification(project: str, operation: str, finish_reason: FinishReason) -> bool: - """Apply minimal filtering logic to determine if notification should be skipped.""" - # Minimal filtering - no special cases for now - return False - - -def run_caliper_orchestration_export(*, artifact_directory: Path | None): - """Set optional ``caliper.export.from`` and run orchestration export.""" - - if artifact_directory is None and "ARTIFACT_BASE_DIR" in os.environ: - artifact_directory = os.environ["ARTIFACT_BASE_DIR"] - - if artifact_directory is not None: - config.project.set_config("caliper.export.from", str(artifact_directory)) +def run_caliper_orchestration_export( + *, artifact_dir: Path, disable_censoring: bool = False, disable_file_export: bool = False +): # Use FJOB_NAME as fallback for mlflow run_name if not configured run_name = config.project.get_config( @@ -857,35 +180,15 @@ def run_caliper_orchestration_export(*, artifact_directory: Path | None): ) # Initialize vaults needed for export operations - logger.info("Checking vaults for export operations") try: + from projects.core.library import vault + # Get export-specific vaults (MLflow, S3, notifications) export_vaults = caliper_export_list_vaults() - logger.info(f"Export vaults needed: {len(export_vaults)} - {export_vaults}") - # Initialize vaults if any are needed if export_vaults: - from projects.core.library import vault - - # Check if vault manager is already initialized - try: - vault.get_vault_manager() - logger.info( - f"Vault manager already initialized, checking {len(export_vaults)} export vaults" - ) - manager_already_initialized = True - except RuntimeError: - logger.info(f"Initializing vault manager with {len(export_vaults)} export vaults") - manager_already_initialized = False - vault.init(vaults=export_vaults) - - if manager_already_initialized: - logger.info(f"Export vault check completed for {len(export_vaults)} vaults") - else: - logger.info( - f"Successfully initialized vault manager with {len(export_vaults)} vaults for export" - ) + logger.info(f"Initialized vault manager with {len(export_vaults)} vaults for export") else: logger.info("No vaults needed for export operation") @@ -895,20 +198,57 @@ def run_caliper_orchestration_export(*, artifact_directory: Path | None): caliper_cfg = config.project.get_config("caliper", print=False) - return run_from_orchestration_config(caliper_cfg) + return run_from_orchestration_config( + caliper_cfg, disable_censoring=disable_censoring, disable_file_export=disable_file_export + ) @click.command("export-artifacts") @click.option( - "--artifact-directory", - "artifact_directory", + "--artifact-dir", + "artifact_dir", type=click.Path(path_type=Path, exists=False, file_okay=True, dir_okay=True), default=None, help="If set, overrides caliper.export.from (artifact root directory).", ) +@click.option( + "--dry-run", + "dry_run", + is_flag=True, + default=False, + help="Show what would be exported and notified without actually performing operations.", +) +@click.option( + "--disable-notification", + "disable_notification", + is_flag=True, + default=False, + help="Skip sending completion notifications.", +) +@click.option( + "--disable-censoring", + "disable_censoring", + is_flag=True, + default=False, + help="Skip censoring sensitive artifacts before export.", +) +@click.option( + "--disable-file-export", + "disable_file_export", + is_flag=True, + default=False, + help="Skip artifact file upload but still run notifications with mock status.", +) @click.pass_context @ci_lib.safe_ci_entrypoint -def caliper_export_entrypoint(_ctx, artifact_directory: Path | None): +def caliper_export_entrypoint( + _ctx, + artifact_dir: Path | None, + dry_run: bool, + disable_notification: bool, + disable_censoring: bool, + disable_file_export: bool, +): """Export the file artifacts.""" notification_provider = getattr(getattr(_ctx, "obj", None), "notification_provider", None) @@ -917,25 +257,100 @@ def caliper_export_entrypoint(_ctx, artifact_directory: Path | None): export_failed = False notification_failed = False + # Determine artifact directory with proper precedence and FOURNOS_CI handling + if not artifact_dir: + # First try the config field + artifact_dir = config.project.get_config( + "caliper.export.from", None, print=False, warn=False + ) + + if not artifact_dir and env.ARTIFACT_DIR: + artifact_dir = env.ARTIFACT_DIR + logger.info(f"Using ARTIFACT_DIR from environment: {artifact_dir}") + # Apply FOURNOS_CI logic only when using ARTIFACT_DIR + if os.environ.get("FOURNOS_CI") == "true": + artifact_dir = Path(artifact_dir).parent + logger.info(f"FOURNOS_CI=true: using parent directory: {artifact_dir}") + + if not artifact_dir: + logger.error( + "No artifact directory found. Please set --artifact-dir parameter, " + "caliper.export.from config, ARTIFACT_DIR, or ARTIFACT_BASE_DIR environment variable." + ) + return 1 + + # Normalize artifact_dir to a pathlib.Path after precedence resolution + artifact_dir = Path(artifact_dir) + + if dry_run: + logging.info(f"DRY RUN: Building caliper notification from {artifact_dir}") + else: + logging.info(f"Building caliper notification from {artifact_dir}") + + # Set the config so other functions can access it + config.project.set_config("caliper.export.from", str(artifact_dir)) + try: - status = run_caliper_orchestration_export(artifact_directory=artifact_directory) - logger.info("Export status:\n" + yaml.dump(status, indent=4)) + if dry_run: + logger.info( + "DRY RUN: Skipping actual caliper export, creating mock status for notification" + ) + # Create a realistic mock status for notification testing + status = { + "success": True, + "final_status": "success", + "backends": {}, + "caliper_artifacts_export": { + "backends": { + "mlflow": { + "success": True, + "run_id": "dry-run-mock-id", + "experiment_url": "http://DRY_RUN_MLFLOW_FAKE_URL/#/experiments/123", + "run_url": "http://DRY_RUN_MLFLOW_FAKE_URL/#/experiments/123/runs/dry-run-mock-id/artifacts?workspace=forge-dry-run", + "tracking_uri": "http://DRY_RUN_MLFLOW_FAKE_URL", + } + } + }, + "duration": "15 minutes, 30 seconds", + "test_phase": { + "phase": "FAILED", + "message": "Test execution completed with failures", + }, + } + else: + status = run_caliper_orchestration_export( + artifact_dir=artifact_dir, + disable_censoring=disable_censoring, + disable_file_export=disable_file_export, + ) + logger.info("Export status:\n" + yaml.dump(status, indent=4)) - # Update fjob status with export results - _update_fjob_export_status(status) + # Update fjob status with export results (only if file export is not disabled) + if not disable_file_export: + _update_fjob_export_status(status) + else: + logger.info("Skipping fjob status update due to --disable-file-export flag") + except ExportFailedException as e: + logger.exception(f"Export failed: {e}") + export_failed = True + # Create failure status for notification + status = {"success": False, "error": str(e), "backends": {}} except Exception as e: - logger.error(f"Export failed: {e}") + logger.exception(f"Export failed with unexpected error: {e}") export_failed = True # Create failure status for notification status = {"success": False, "error": str(e), "backends": {}} finally: # Send completion notifications regardless of success/failure - if status: + if status and not disable_notification: try: notification_success = send_notification( - status, notification_provider=notification_provider + artifact_dir, + status, + notification_provider=notification_provider, + dry_run=dry_run, ) if not notification_success: logger.error("Notification sending failed") @@ -943,16 +358,28 @@ def caliper_export_entrypoint(_ctx, artifact_directory: Path | None): except Exception as e: logger.exception(f"Failed to send notifications: {e}") notification_failed = True + elif disable_notification: + logger.info("Notifications disabled via --disable-notification flag") - _update_final_artifacts(status) + if not disable_file_export: + if not dry_run: + _update_final_artifacts(artifact_dir, status) + else: + logger.info("DRY RUN: Skipping final artifacts update to MLflow") + else: + logger.info("Skipping final artifacts upload due to --disable-file-export flag") - # Return proper exit code if export_failed or notification_failed: - return 1 + return 1, "failed" + + # Check if censoring occurred and return exit code 1 if so + if status and status.get("censoring_occurred", False): + return 1, "censoring_occurred" + return 0 -def _update_final_artifacts(export_status: dict[str, Any] | None) -> None: +def _update_final_artifacts(artifact_dir, export_status: dict[str, Any] | None) -> None: """Update the final artifacts (run.log, notifications) to MLflow after all post-export work is done.""" if not export_status: logger.warning("No export status received, cannot update the final artifacts") @@ -972,16 +399,7 @@ def _update_final_artifacts(export_status: dict[str, Any] | None) -> None: if not run_id: return - artifact_from = config.project.get_config( - "caliper.export.from", None, print=False, warn=False - ) - if not artifact_from: - logger.warning( - "Export status don't have the caliper.export.from field, cannot update the final artifacts" - ) - return - - artifact_root = Path(artifact_from) + artifact_root = Path(artifact_dir) artifact_path = str(env.ARTIFACT_DIR.relative_to(artifact_root)) tracking_uri = mlflow_meta.get("tracking_uri") diff --git a/projects/core/library/export_notifications.py b/projects/core/library/export_notifications.py new file mode 100644 index 000000000..cfea6c881 --- /dev/null +++ b/projects/core/library/export_notifications.py @@ -0,0 +1,1009 @@ +""" +Notification handling for Caliper export operations. + +This module provides notification functionality for export completion, +including GitHub notifications and Slack notifications via project providers. +""" + +import logging +import os +import re +from pathlib import Path +from typing import Any + +import yaml + +from projects.caliper.orchestration.censoring import censor_text +from projects.core.ci_entrypoint.prepare_ci import CI_METADATA_DIRNAME +from projects.core.library import ci as ci_lib +from projects.core.library import config, env +from projects.core.library.step_status import StepStatus + +logger = logging.getLogger(__name__) + + +def _censor_notification_text(text: str, verbose: bool = False) -> str: + """ + Censor sensitive content in notification text using caliper orchestration. + + Args: + text: The notification text to censor + verbose: Enable verbose logging + + Returns: + Censored notification text with sensitive content replaced + + Raises: + Exception: If vault discovery fails, preventing notification delivery + """ + try: + return censor_text(text, verbose=verbose) + except Exception as e: + logger.error(f"Censoring failed during notification preparation: {e}") + # Re-raise to abort GitHub and Slack notification delivery + # rather than sending potentially uncensored text + raise + + +def send_notification( + artifact_dir: Path | None, + status: dict[str, Any], + notification_provider=None, + dry_run: bool = False, +) -> bool: + """Send job completion notifications based on caliper export status. + + Args: + artifact_dir: Directory to browse to find the artifacts + status: Caliper export status object containing backend results and metadata + notification_provider: Optional per-project SlackNotificationProvider instance + dry_run: If True, only build and log notification content without sending + + Returns: + bool: True if notifications were sent successfully, False otherwise + """ + # Extract notification parameters from status object + project = config.project.get_config("project.name") + finish_reason = _extract_finish_reason_from_status(status) + + # Build enhanced notification with fournos job info and artifact links + notification_status, notification_success = _build_enhanced_notification( + artifact_dir, project, finish_reason, status + ) + + # Apply censoring to notification content before sending + try: + notification_status = _censor_notification_text(notification_status, verbose=dry_run) + except Exception as e: + logger.error(f"Notification censoring failed, aborting notification delivery: {e}") + return False + + # Send actual notifications + if dry_run: + logger.info("DRY RUN: Would send notification") + logger.info(f"DRY RUN: Notification content:\n{notification_status}") + else: + logger.info("Sending notification ...") + + # Write notification to file for GitHub pickup (always generate, even in dry-run) + try: + if env.ARTIFACT_DIR: + notification_file = Path(env.ARTIFACT_DIR) / "NOTIFICATION-github.md" + with open(notification_file, "w", encoding="utf-8") as f: + f.write(notification_status + "\n") + if dry_run: + logger.info(f"DRY RUN: Generated notification file {notification_file}") + else: + logger.info(f"Wrote export notification file {notification_file}") + else: + logger.warning("ARTIFACT_DIR not available, skipping notification file") + except Exception as e: + logger.exception(f"Failed to write notification file: {e}") + + # Actually send notification through GitHub API + try: + from projects.core.notifications.send import send_notification as send_github_notification + + # Get notification vault from configuration + notification_vault = None + try: + notification_config = config.project.get_config("caliper.export.notifications", {}) + notification_vault = notification_config.get("vault") + if notification_vault: + logger.info(f"Using notification vault from config: {notification_vault}") + except Exception as e: + logger.warning(f"Failed to get notification vault from config: {e}") + + success = send_github_notification( + message=notification_status, + github=True, + slack=False, + dry_run=dry_run, + notification_vault=notification_vault, + ) + if success: + logger.info("Successfully sent GitHub notification") + else: + logger.error("GitHub notification sending failed") + notification_success = False + except Exception as e: + logger.error(f"Failed to send GitHub notification: {e}") + notification_success = False + + # Per-project Slack notification via provider + if notification_provider: + if not dry_run: + try: + from projects.core.notifications.provider import NotificationContext + + artifact_dir = Path(env.ARTIFACT_DIR) if env.ARTIFACT_DIR else None + # Censor status fields before creating NotificationContext + try: + censored_status = yaml.safe_load( + _censor_notification_text(yaml.dump(status), verbose=dry_run) + ) + except Exception as e: + logger.error( + f"Slack notification censoring failed, aborting Slack notification: {e}" + ) + notification_success = False + else: + # Only proceed with Slack notification if censoring succeeded + context = NotificationContext( + status=censored_status, + finish_reason=str(finish_reason), + project_name=project or "unknown", + pr_number=os.environ.get("PULL_NUMBER"), + job_type=os.environ.get("JOB_TYPE"), + artifact_dir=artifact_dir, + ) + ok = notification_provider.notify(context) + if ok: + logger.info("Successfully sent per-project Slack notification") + else: + logger.warning("Per-project Slack notification failed") + notification_success = False + except Exception as e: + logger.warning(f"Failed to send per-project Slack notification: {e}") + notification_success = False + else: + logger.info("DRY RUN: Would send per-project Slack notification") + + return notification_success + + +def _get_project_and_args(project: str) -> tuple[str, str]: + """Extract project name and args from fournos job or config.""" + fjob_project = project + fjob_args_str = "" + + try: + metadata_dir = ci_lib.get_ci_metadata_dir() + fournos_fjob_path = metadata_dir / "fournos_fjob.yaml" + if not fournos_fjob_path.exists(): + return fjob_project, fjob_args_str + + with open(fournos_fjob_path, encoding="utf-8") as f: + fjob_data = yaml.safe_load(f) + + display_name = fjob_data.get("spec", {}).get("displayName", "") + if not display_name: + return fjob_project, fjob_args_str + + parts = display_name.split() + if not parts: + return fjob_project, fjob_args_str + + fjob_project = parts[0] + fjob_args_str = " ".join(parts[1:]) if len(parts) > 1 else "" + except Exception as e: + logger.warning(f"Failed to read fournos job for project/args: {e}") + + if fjob_args_str: + fjob_args_str = f"with `{fjob_args_str}`" + + return fjob_project, fjob_args_str + + +def _extract_finish_reason_from_status(status: dict[str, Any]) -> str: + """Extract finish reason from status.""" + if not status.get("success", False): + return "failed" + elif status.get("censoring_occurred", False): + return "completed with censoring" + else: + return "completed" + + +def _build_enhanced_notification( + artifact_dir: Path, + project: str, + finish_reason: str, + status: dict[str, Any], +) -> tuple[str, bool]: + """Build enhanced notification with fournos job config and artifact links.""" + fjob_project, fjob_args_str = _get_project_and_args(project) + + status_emoji = "✅" if status.get("success", False) else "❌" + if status.get("censoring_occurred", False): + status_emoji = "⚠️" + + if finish_reason == "failed": + status_emoji = "❌" + + base_status = f"**{status_emoji} Execution of `{fjob_project}` {fjob_args_str} {status_emoji}**" + notification_parts = [base_status] + + # Add job abort message right below overall status if applicable + shutdown_status = status.get("job_shutdown") + if shutdown_status and shutdown_status.get("is_aborted"): + shutdown_value = shutdown_status.get("shutdown_value", "Stop") + notification_parts.append(f"🛑 **JOB ABORTED** - `spec.shutdown={shutdown_value}`") + + notification_parts.append("---") + notification_parts.append("") + + execution_engine_config = _get_execution_engine_config() + if execution_engine_config: + notification_parts.append("**Execution Engine Configuration**") + notification_parts.append(execution_engine_config) + + notification_success = True + try: + artifact_links, mlflow_run_url = _extract_artifact_links(status) + + test_status_section = _extract_test_status_section(status) + step_status = _get_step_status_section(artifact_dir, mlflow_run_url) + postprocess_status_links = _get_postprocess_status_links(artifact_dir, mlflow_run_url) + + if test_status_section: + notification_parts.append("") + notification_parts.append("---") + notification_parts.extend(test_status_section) + + if artifact_links: + notification_parts.append("") + notification_parts.append("---") + notification_parts.append("**Artifact Links**") + notification_parts.extend([f"* {link}" for link in artifact_links]) + else: + notification_parts.append("**Artifact Links:** No direct links available") + + if step_status: + notification_parts.append("") + notification_parts.append("---") + notification_parts.append("**Step details**") + for link in step_status: + notification_parts.append(link) + + if postprocess_status_links: + notification_parts.append("") + notification_parts.extend(postprocess_status_links) + + except Exception as e: + logger.exception(f"Failed to build the extended notifications: {e}") + notification_parts.append("**Artifact Links:** Error extracting links") + notification_success = False + + notification_parts.append("") + notification_parts.append("---") + + return "\n".join(notification_parts), notification_success + + +def _get_execution_engine_config() -> str | None: + """Get execution engine configuration for notification.""" + try: + cluster_config = config.project.get_config("cluster", None, warn=False) + if not cluster_config: + return None + + config_parts = [] + for key, value in cluster_config.items(): + if value: + config_parts.append(f"`{key}`: {value}") + + if config_parts: + return "* " + " \n* ".join(config_parts) + except Exception: + pass + return None + + +def _extract_test_status_section(status: dict[str, Any]) -> list[str] | None: + """Extract test status section from status.""" + test_phase = status.get("test_phase") + if not test_phase: + return None + + phase = test_phase.get("phase", "").upper() + message = test_phase.get("message", "") + test_status_emoji = "✅" if phase == "PASSED" else "❌" if phase == "FAILED" else "⚠️" + + return [ + f"**{test_status_emoji} Test Status: {phase} {test_status_emoji}**", + f"**Message:** {message}", + ] + + +def _get_step_status_section(artifact_dir: Path | None, mlflow_run_url: str | None) -> list[str]: + """Get step status section with links.""" + if not artifact_dir or not artifact_dir.exists(): + return [] + + step_status = [] + for step_dir in sorted(artifact_dir.glob("*")): + if not step_dir.is_dir() or step_dir.name.startswith("."): + continue + + step_name = step_dir.name + + # Skip special directories + if step_name == CI_METADATA_DIRNAME: + continue + + exit_status_file = step_dir / CI_METADATA_DIRNAME / "exit_status.yaml" + exit_status_emoji = "❓" + + if exit_status_file.exists(): + try: + with open(exit_status_file, encoding="utf-8") as f: + exit_data = yaml.safe_load(f) + exit_code = exit_data.get("return_code", 999) + if exit_code == 0: + exit_status_emoji = "✅" + else: + exit_status_emoji = "❌" + except Exception: + exit_status_emoji = "❓" + + # Count ERROR and WARNING messages in run.log + log_counts = _count_log_messages(step_dir) + log_summary = _format_log_summary(log_counts) + + # Create step title - linked if MLflow URL available, plain-text otherwise + if mlflow_run_url: + mlflow_log_url = _create_mlflow_url(mlflow_run_url, step_name) + step_title = f"#### {exit_status_emoji} [{step_name}]({mlflow_log_url}){log_summary}" + else: + step_title = f"#### {exit_status_emoji} {step_name}{log_summary}" + + step_status.append(step_title) + + step_status.extend(_process_notification_files(step_dir)) + + step_details = _process_step_details(step_dir, mlflow_run_url) + if step_details: + step_status.extend(step_details) + + return step_status + + +def _count_log_messages(step_dir: Path) -> dict[str, int]: + """Count ERROR and WARNING messages in run.log file. + + Args: + step_dir: Directory containing the run.log file + + Returns: + Dict with 'errors' and 'warnings' counts + """ + log_file = step_dir / "run.log" + counts = {"errors": 0, "warnings": 0} + + if not log_file.exists(): + return counts + + try: + with open(log_file, encoding="utf-8", errors="ignore") as f: + for line in f: + line = line.strip() + if line.startswith("ERROR:"): + counts["errors"] += 1 + elif line.startswith("WARNING:"): + counts["warnings"] += 1 + except Exception as e: + logger.warning(f"Failed to read log file {log_file}: {e}") + + return counts + + +def _format_log_summary(log_counts: dict[str, int]) -> str: + """Format log counts for display in step title. + + Args: + log_counts: Dict with 'errors' and 'warnings' counts + + Returns: + Formatted string to append to step title, or empty string if no issues + """ + errors = log_counts.get("errors", 0) + warnings = log_counts.get("warnings", 0) + + if errors == 0 and warnings == 0: + return "" + + parts = [] + if errors > 0: + parts.append(f"🔴 {errors}E") + if warnings > 0: + parts.append(f"🟡 {warnings}W") + + return f" ({', '.join(parts)})" + + +def _get_postprocess_status_links( + artifact_dir: Path | None, mlflow_run_url: str | None +) -> list[str]: + """Get postprocess status links.""" + if not artifact_dir or not artifact_dir.exists(): + return [] + + step_log_links = [] + + # Look for postprocess results in step directories + for step_dir in sorted(artifact_dir.glob("*")): + if not step_dir.is_dir() or step_dir.name.startswith("."): + continue + + step_name = step_dir.name + + # Skip special directories + if step_name == CI_METADATA_DIRNAME: + continue + + try: + postprocess_status_file = step_dir / "postprocess_status.yaml" + if not postprocess_status_file.exists(): + continue + + with open(postprocess_status_file, encoding="utf-8") as f: + status_data = yaml.safe_load(f.read()) + + if not status_data: + continue + + # Add job shutdown status if available + if "job_shutdown" in status_data: + shutdown_status = status_data["job_shutdown"] + status_data["job_shutdown"] = shutdown_status + + # Import notification functions from caliper + from projects.caliper.orchestration.notification import ( + format_postprocess_status_notification, + parse_postprocess_result, + ) + + # Parse postprocess result + result = parse_postprocess_result(status_data) + if not result: + continue + + # Create file link function for this step + def get_file_link(file_path: Path, step_subdir: str = step_name) -> str: + if mlflow_run_url: + # Create MLflow artifact URL + return _create_mlflow_file_url_for_step( + mlflow_run_url, step_subdir, str(file_path) + ) + + # Generate notification text from the structured result + notification_text = format_postprocess_status_notification(result, get_file_link) + if notification_text: + step_log_links.append(notification_text) + + except Exception as e: + logger.warning(f"Failed to process postprocess status for {step_name}: {e}") + + return step_log_links + + +def _extract_artifact_links(status: dict[str, Any]) -> tuple[list[str], str | None]: + """Extract artifact links and MLflow URL from status.""" + artifact_links = [] + mlflow_run_url = None + + caliper_export = status.get("caliper_artifacts_export", {}) + backends = caliper_export.get("backends", {}) + + for backend_name, backend_result in backends.items(): + if not isinstance(backend_result, dict): + continue + + if backend_result.get("experiment_url"): + artifact_links.append( + f"[{backend_name} Experiment]({backend_result['experiment_url']})" + ) + + if backend_result.get("run_url"): + mlflow_run_url = backend_result["run_url"] + artifact_links.append(f"[{backend_name} Results]({mlflow_run_url})") + elif backend_result.get("artifact_url"): + artifact_links.append(f"[{backend_name} Artifacts]({backend_result['artifact_url']})") + elif backend_result.get("dashboard_url"): + artifact_links.append(f"[{backend_name} Dashboard]({backend_result['dashboard_url']})") + + if status.get("artifact_url"): + artifact_links.append(f"[Artifacts]({status['artifact_url']})") + + return artifact_links, mlflow_run_url + + +def _create_mlflow_file_url_for_step( + mlflow_run_url: str, step_dir_name: str, file_path: str +) -> str: + """Create MLflow URL for a specific file within a step directory. + + Args: + mlflow_run_url: Base MLflow run URL + step_dir_name: Name of the step directory + file_path: Relative path to file from step directory + + Returns: + Full MLflow URL to the file + + Raises: + ValueError: If URL format is unexpected + """ + if "/artifacts" not in mlflow_run_url: + raise ValueError(f"Unexpected MLflow URL format: {mlflow_run_url}") + + # Clean file path + file_clean = file_path.lstrip("/") + + if "#" in mlflow_run_url: + base_domain, hash_fragment = mlflow_run_url.split("#", 1) + if "/artifacts" not in hash_fragment: + raise ValueError("Artifacts not found in hash fragment") + + hash_base, artifacts_part = hash_fragment.split("/artifacts", 1) + # Extract workspace parameter if present, ignoring existing path + workspace_param = "" + if "?workspace=" in artifacts_part: + workspace_param = artifacts_part[artifacts_part.find("?") :] + return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}/{file_clean}{workspace_param}" + else: + base_url, artifacts_part = mlflow_run_url.split("/artifacts", 1) + # Extract workspace parameter if present, ignoring existing path + workspace_param = "" + if "?workspace=" in artifacts_part: + workspace_param = artifacts_part[artifacts_part.find("?") :] + return f"{base_url}/artifacts/{step_dir_name}/{file_clean}{workspace_param}" + + +def _process_notification_files(step_dir: Path) -> list[str]: + """Process notification files from step directory.""" + notifications_dir = step_dir / CI_METADATA_DIRNAME / "notifications" + if not (notifications_dir.exists() and notifications_dir.is_dir()): + return [] + + notifications_from_files = [] + for notification_file in sorted(notifications_dir.glob("*.txt")): + with open(notification_file, encoding="utf-8") as f: + content = f.read().strip() + + if not content: + continue + + subtitle = notification_file.stem.replace("__", " ").replace("_", " ").title() + subtitle = re.sub(r"^\d+\s+", "", subtitle) + notifications_from_files.append(f"##### {subtitle}") + + for line in content.splitlines(): + notifications_from_files.append(f"> {line}") + + return notifications_from_files + + +def _extract_test_labels_info(artifact_dir: Path, mlflow_run_url: str | None = None) -> list[str]: + """Extract test execution information from __test_labels__.yaml files. + + Args: + artifact_dir: Directory to search for __test_labels__.yaml files + mlflow_run_url: Optional MLflow run URL for creating links + + Returns: + List of formatted strings with test information (directory, labels, success, message) + """ + test_info_lines = [] + + # Search for __test_labels__.yaml files recursively + test_labels_files = list(artifact_dir.glob("**/__test_labels__.yaml")) + + if not test_labels_files: + return [] + + for test_labels_file in test_labels_files: + try: + with open(test_labels_file, encoding="utf-8") as f: + test_data = yaml.safe_load(f) or {} + + # Extract directory relative to artifact_dir - use just the immediate directory name + relative_dir = test_labels_file.parent.relative_to(artifact_dir) + dir_name = relative_dir.name if relative_dir != Path(".") else "root" + + # Extract completion info + completion = test_data.get("completion", {}) + success = completion.get("success") + message = completion.get("message") + + if message: + message = f": `{message}`" + else: + message = "" + + # Format status + if success: + status_emoji = "✅" + elif success is False: + status_emoji = "❌" + else: + status_emoji = "❓" + + # Create link to __test_labels__.yaml file if MLflow URL is available + if mlflow_run_url: + try: + # Get the step directory name (relative to parent) + step_dir_name = str(relative_dir) + test_labels_url = _create_mlflow_file_url_for_step( + mlflow_run_url, step_dir_name, "__test_labels__.yaml" + ) + dir_link = f"[**{dir_name}**]({test_labels_url})" + except Exception as e: + logger.warning(f"Failed to create MLflow link for {test_labels_file}: {e}") + dir_link = f"**{dir_name}**" + else: + dir_link = f"**{dir_name}**" + + test_info_lines.append(f"* {status_emoji} {dir_link}{message}") + + except Exception as e: + test_info_lines.append(f"**{test_labels_file.name}**: Error reading file - {e}") + + return test_info_lines + + +def _extract_postprocess_status_info(artifact_dir: Path) -> list[str]: + """Extract post-processing status information from postprocess_status.yaml files. + + Returns: + List of formatted strings with postprocess step status (success only, no details) + """ + postprocess_info_lines = [] + + # Search for postprocess_status.yaml files recursively + postprocess_files = list(artifact_dir.glob("**/postprocess_status.yaml")) + + if not postprocess_files: + return [] + + for postprocess_file in postprocess_files: + try: + with open(postprocess_file, encoding="utf-8") as f: + postprocess_data = yaml.safe_load(f) or {} + + # Extract directory relative to artifact_dir + relative_dir = postprocess_file.parent.relative_to(artifact_dir) + dir_name = str(relative_dir) if relative_dir != Path(".") else "root" + + # Extract overall status + overall_success = postprocess_data.get("success", False) + final_status = postprocess_data.get("final_status", "unknown") + + # Extract individual step statuses + steps = postprocess_data.get("steps", []) + step_statuses = [] + + for step_dict in steps: + for step_name, step_data in step_dict.items(): + if isinstance(step_data, dict): + status = step_data.get("status", "unknown") + status_emoji = ( + "✅" if status == "success" else "❌" if status == "failed" else "⚪" + ) + step_statuses.append(f"{step_name}:{status_emoji}") + + # Format overall line + overall_emoji = "✅" if overall_success else "❌" + if step_statuses: + steps_str = " " + " ".join(step_statuses) + else: + steps_str = f" {final_status}" + + postprocess_info_lines.append(f"**{dir_name}**: {overall_emoji}{steps_str}") + + except Exception as e: + postprocess_info_lines.append(f"**{postprocess_file.name}**: Error reading file - {e}") + + return postprocess_info_lines + + +def _process_step_details(step_dir: Path, mlflow_run_url: str | None = None) -> list[str]: + """Process test labels and postprocess status for a single step directory.""" + step_details = [] + + # Extract test labels for this specific step + try: + test_labels_info = _extract_test_labels_info(step_dir, mlflow_run_url) + if test_labels_info: + step_details.extend(test_labels_info) + except Exception as e: + logger.warning(f"Failed to extract test labels for step {step_dir.name}: {e}") + + # Extract postprocess status for this specific step + try: + postprocess_info = _extract_postprocess_status_info(step_dir) + if postprocess_info: + step_details.extend(postprocess_info) + except Exception as e: + logger.warning(f"Failed to extract postprocess status for step {step_dir.name}: {e}") + + return step_details + + +def _check_job_shutdown_status() -> dict[str, Any] | None: + """Check if the job has been aborted via spec.shutdown field.""" + try: + from projects.core.library import ci as ci_lib + + metadata_dir = ci_lib.get_ci_metadata_dir() + fournos_fjob_path = metadata_dir / "fournos_fjob.yaml" + if not fournos_fjob_path.exists(): + return None + + with open(fournos_fjob_path, encoding="utf-8") as f: + fjob_data = yaml.safe_load(f) + + shutdown_value = fjob_data.get("spec", {}).get("shutdown") + if shutdown_value: + return { + "shutdown_detected": True, + "shutdown_value": shutdown_value, + "is_aborted": shutdown_value.lower() == "stop", + } + + return {"shutdown_detected": False, "shutdown_value": None, "is_aborted": False} + except Exception as e: + logger.warning(f"Failed to check job shutdown status: {e}") + return None + + +def _read_step_exit_status( + step_dir: Path, current_step_name: str | None = None +) -> tuple[str, StepStatus]: + """Read exit status from step directory and return emoji and status enum.""" + from projects.core.ci_entrypoint.prepare_ci import CI_METADATA_DIRNAME + + try: + exit_status_file = step_dir / CI_METADATA_DIRNAME / "exit_status.yaml" + if not exit_status_file.exists(): + # Check if this is the current ongoing step + if current_step_name and step_dir.name == current_step_name: + return "🔄", StepStatus.ONGOING # Ongoing step + return "❓", StepStatus.UNKNOWN # Unknown status if file doesn't exist + + with open(exit_status_file, encoding="utf-8") as f: + exit_data = yaml.safe_load(f) + + return_code = exit_data.get("return_code") + if return_code is None or return_code == 0: + return "✅", StepStatus.SUCCESS + else: + return "❌", StepStatus.FAILURE + except Exception as e: + logger.warning(f"Failed to read exit status from {step_dir}: {e}") + # Check if this is the current ongoing step even on error + if current_step_name and step_dir.name == current_step_name: + return "🔄", StepStatus.ONGOING # Ongoing step + return "❓", StepStatus.UNKNOWN # Unknown status on error + + +def _check_postprocess_warnings(step_dir: Path) -> StepStatus: + """Check for warning status in postprocess status file.""" + + status = StepStatus.SUCCESS # No postprocess warning/error, assume no warnings + for status_file in step_dir.glob("**/postprocess_status.yaml"): + try: + with open(status_file, encoding="utf-8") as f: + status_data = yaml.safe_load(f) + except Exception as e: + logging.error(f"Failed to read {status_file} as yaml: {e}") + status = StepStatus.WARNING + continue + + if not status_data: + continue + + # Check top-level success field for warning value + success_value = status_data.get("success") + if success_value == "warning": + logging.warning( + f"Post-process warning detected in {status_file}, setting the WARNING flag" + ) + status = StepStatus.WARNING + + if success_value in ("failure", "error"): + logging.error(f"Post-process {success_value} detected, raising the FAILURE flag") + return StepStatus.FAILURE + + return status + + +def _get_overall_status_from_steps(artifact_dir: Path) -> str: + """Check all step exit statuses and return overall status emoji.""" + from projects.core.library import env + from projects.core.library.export import StepStatus + + try: + current_step_name = Path(env.BASE_ARTIFACT_DIR).name + + step_statuses = [] + + for step_dir in sorted(artifact_dir.iterdir()): + if not step_dir.is_dir(): + continue + if step_dir.name.startswith("."): + continue + + # Only check directories that have run.log (actual steps) + run_log = step_dir / "run.log" + if not run_log.exists(): + continue + + _emoji, status = _read_step_exit_status(step_dir, current_step_name) + + step_statuses.append(status) + + # Check for postprocess warnings in this step (always check, regardless of exit status) + postprocess_status = _check_postprocess_warnings(step_dir) + step_statuses.append(postprocess_status) + + # Priority: failure > ongoing > warning > unknown > success + if StepStatus.FAILURE in step_statuses: + return "🔴" # Any failure = red + elif StepStatus.WARNING in step_statuses: + return "🟠" # Warning = orange + elif StepStatus.UNKNOWN in step_statuses: + return "🟠" # Unknown = orange + elif StepStatus.ONGOING in step_statuses: + return "🟢" # Ongoing --> success + else: + return "🟢" # All successful = green + + except Exception as e: + logger.exception(f"Failed to check step statuses: {e}") + return "🔴" # Error checking = red + + +def _create_mlflow_url(mlflow_run_url: str, step_dir_name: str) -> str | None: + """Create MLflow URL for step logs.""" + + if not mlflow_run_url: + return f"BASE_URL_MISSING/{step_dir_name}" + + if "/artifacts" not in mlflow_run_url: + logger.warning(f"Unexpected MLflow URL format: {mlflow_run_url}") + return None + + if "#" in mlflow_run_url: + base_domain, hash_fragment = mlflow_run_url.split("#", 1) + if "/artifacts" not in hash_fragment: + raise ValueError("Artifacts not found in hash fragment") + + hash_base, params = hash_fragment.split("/artifacts", 1) + workspace_param = params if "?workspace=" in params else "" + return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}/run.log{workspace_param}" + else: + base_url, params = mlflow_run_url.split("/artifacts", 1) + workspace_param = params if "?workspace=" in params else "" + return f"{base_url}/artifacts/{step_dir_name}/run.log{workspace_param}" + + +def _create_mlflow_step_url(mlflow_run_url: str, step_dir_name: str) -> str | None: + """Create MLflow URL for step directory (for file access).""" + if "/artifacts" not in mlflow_run_url: + logger.warning(f"Unexpected MLflow URL format: {mlflow_run_url}") + return None + + if "#" in mlflow_run_url: + base_domain, hash_fragment = mlflow_run_url.split("#", 1) + if "/artifacts" not in hash_fragment: + raise ValueError("Artifacts not found in hash fragment") + + hash_base, artifacts_part = hash_fragment.split("/artifacts", 1) + # Extract workspace parameter if present, ignoring existing path + workspace_param = "" + if "?workspace=" in artifacts_part: + workspace_param = artifacts_part[artifacts_part.find("?") :] + return f"{base_domain}#{hash_base}/artifacts/{step_dir_name}{workspace_param}" + else: + base_url, artifacts_part = mlflow_run_url.split("/artifacts", 1) + # Extract workspace parameter if present, ignoring existing path + workspace_param = "" + if "?workspace=" in artifacts_part: + workspace_param = artifacts_part[artifacts_part.find("?") :] + return f"{base_url}/artifacts/{step_dir_name}{workspace_param}" + + +def _read_step_duration(step_dir: Path) -> str: + """Read step duration from timing file.""" + from projects.core.ci_entrypoint.prepare_ci import CI_METADATA_DIRNAME + + timing_file = step_dir / CI_METADATA_DIRNAME / "test_duration.yaml" + if not timing_file.exists(): + return "" + + try: + with open(timing_file, encoding="utf-8") as f: + timing_data = yaml.safe_load(f) + + formatted_duration = timing_data.get("duration", {}).get("formatted") + return formatted_duration or "" + except Exception as timing_error: + logger.warning(f"Failed to read timing file {timing_file}: {timing_error}") + return "" + + +def _process_step_status(artifact_dir: Path, mlflow_run_url: str) -> list[str]: + """Process step logs from parent directory.""" + from projects.core.library import env + + if not mlflow_run_url: + logging.warning("mlflow_run_url not set. Will generate dummy links.") + + step_status = [] + + current_step_name = Path(env.BASE_ARTIFACT_DIR).name + + for step_dir in sorted(artifact_dir.iterdir()): + if not step_dir.is_dir(): + continue + if step_dir.name.startswith("."): + continue + + run_log = step_dir / "run.log" + if not run_log.exists(): + continue + + mlflow_log_url = _create_mlflow_url(mlflow_run_url, step_dir.name) + if not mlflow_log_url: + mlflow_log_url = "NO_URL" + + step_name = step_dir.name.replace("__", " ").replace("_", " ").title() + duration_str = _read_step_duration(step_dir) + exit_status_emoji, exit_status = _read_step_exit_status(step_dir, current_step_name) + + step_status.append("") + if duration_str: + step_status.append( + f"#### {exit_status_emoji} [{step_name}]({mlflow_log_url}) `{duration_str}`" + ) + else: + step_status.append(f"#### {exit_status_emoji} [{step_name}]({mlflow_log_url})") + + step_status.extend(_process_notification_files(step_dir)) + + step_details = _process_step_details(step_dir, mlflow_run_url) + + # Add test execution info for this step + if step_details["test_labels_info"]: + # Add Test Execution Overview if we have any step info + step_status.append("") + step_status.append("**Test Execution Overview**") + + step_status.extend([f"* {info}" for info in step_details["test_labels_info"]]) + + # Add postprocess info for this step + if step_details["postprocess_info"]: + step_status.extend([f"* {info}" for info in step_details["postprocess_info"]]) + + return step_status + + +def _extract_duration_from_status(status: dict[str, Any]) -> str: + """Extract duration from status object.""" + # Look for duration in status + duration = status.get("duration") + if duration: + return f" after {duration}" + return "" diff --git a/projects/core/library/step_status.py b/projects/core/library/step_status.py new file mode 100644 index 000000000..2325385d9 --- /dev/null +++ b/projects/core/library/step_status.py @@ -0,0 +1,15 @@ +"""Step status enumeration for export operations.""" + +from __future__ import annotations + +from enum import StrEnum + + +class StepStatus(StrEnum): + """Status of a step execution.""" + + SUCCESS = "success" + FAILURE = "failure" + ONGOING = "ongoing" + UNKNOWN = "unknown" + WARNING = "warning" diff --git a/projects/core/library/vault.py b/projects/core/library/vault.py index 18232702b..bf270501e 100644 --- a/projects/core/library/vault.py +++ b/projects/core/library/vault.py @@ -31,6 +31,7 @@ class VaultContent: name: str description: str filename: str | None = None + sensible: bool = True _vault: Optional["VaultDefinition"] = None def __post_init__(self): @@ -38,6 +39,11 @@ def __post_init__(self): if self.filename is None: self.filename = self.name + @property + def is_sensible(self) -> bool: + """Whether this content should be considered sensitive for censoring purposes""" + return self.sensible + @property def file_path(self) -> Path | None: """Get the full absolute path to this content file""" @@ -107,13 +113,15 @@ def _load_vault_definition(self, vault_file: Path) -> VaultDefinition: # New format with file mapping and description filename = content_def.get("file", content_name) description = content_def.get("description", "") # Don't provide default + sensible = content_def.get("sensible", True) # Default to True else: # Legacy format - content_def is the description filename = content_name description = content_def if content_def else "" + sensible = True # Sensible by default content[content_name] = VaultContent( - name=content_name, description=description, filename=filename + name=content_name, description=description, filename=filename, sensible=sensible ) vault_def = VaultDefinition( @@ -435,7 +443,7 @@ def init( global _vault_manager, _strict_validation_enabled if _vault_manager is not None: - logger.warning("VaultManager already initialized") + logger.warning("VaultManager already initialized", stack_info=True) return _vault_manager = VaultManager() diff --git a/projects/skeleton/orchestration/test_skeleton.py b/projects/skeleton/orchestration/test_skeleton.py index 672fedc04..0f8ddcfe7 100644 --- a/projects/skeleton/orchestration/test_skeleton.py +++ b/projects/skeleton/orchestration/test_skeleton.py @@ -100,15 +100,15 @@ def do_test(): logger.info("=== Skeleton Project Test Phase ===") if config.project.get_config("skeleton.deep_testing"): - logger.warning("Running the (fake) deep testing ...") + logger.info("Running the (fake) deep testing ...") else: - logger.warning("Running the (fake) light testing ...") + logger.info("Running the (fake) light testing ...") client_id = vault.get_vault_content_path("psap-forge-notifications", "topsail-bot.clientid") if not client_id: - logger.warning("`client_id` secret not available.") + logger.error("`client_id` secret not available.") else: - logger.warning(f"`client_id` secret available. Size: {client_id.stat().st_size}b") + logger.info(f"`client_id` secret available. Size: {client_id.stat().st_size}b") del client_id skeleton_config = config.project.get_config("skeleton", print=False) @@ -128,7 +128,7 @@ def do_test(): seed_skeleton_caliper_artifacts() if not config.project.get_config("skeleton.collect_cluster_info"): - logger.warning("⚠️ Cluster information gathering not enabled. Returning early.") + logger.info("⚠️ Cluster information gathering not enabled. Returning early.") return 0 # Demonstrate calling a toolbox from orchestration diff --git a/vaults/psap-forge-mlflow-export.yaml b/vaults/psap-forge-mlflow-export.yaml index 1538e8d2f..c4c2425e1 100644 --- a/vaults/psap-forge-mlflow-export.yaml +++ b/vaults/psap-forge-mlflow-export.yaml @@ -17,8 +17,9 @@ content: the URI of the MLFlow endpoint, so that the CI engine can censor it in the logs username: + sensible: false description: | - the mlflow username, so that the CI engine can censor it in the logs + the mlflow username password: description: | diff --git a/vaults/psap-forge-notifications.yaml b/vaults/psap-forge-notifications.yaml index 12ae01cc8..73eb1ce5b 100644 --- a/vaults/psap-forge-notifications.yaml +++ b/vaults/psap-forge-notifications.yaml @@ -7,8 +7,10 @@ content: description: | Private key to send github notifications topsail-bot.clientid: + sensible: false description: | - Token to identify the TOPSAIL-bot client + Public ID of the TOPSAIL-bot client + topsail-bot.slack-token: description: | Token to send slack notification from the TOPSAIL-bot