app.py

"""
Cordon - Semantic Anomaly Detection for Log Files
HuggingFace Space Demo (ZeroGPU)
"""

import logging
import os
import sys
import time
from pathlib import Path

import gradio as gr

# ZeroGPU support - graceful fallback for local development
try:
    import spaces
    ZEROGPU_AVAILABLE = True
except ImportError:
    ZEROGPU_AVAILABLE = False
    # Create a no-op decorator for local development
    class spaces:
        @staticmethod
        def GPU(duration=60):
            def decorator(fn):
                return fn
            return decorator

# NOTE: Do NOT import torch or cordon at module level!
# ZeroGPU requires all CUDA-related imports to happen inside @spaces.GPU functions.
# Importing torch at module level can trigger CUDA initialization and hang the space.

# Configure logging for HuggingFace Spaces visibility
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Load sample log file
SAMPLE_LOG_PATH = Path(__file__).parent / "sample_apache.log"
FALLBACK_LOG_PATH = Path(__file__).parent.parent / "examples" / "apache_sample.log"


def load_sample_log() -> str:
    """Load the sample Apache log file."""
    if SAMPLE_LOG_PATH.exists():
        return SAMPLE_LOG_PATH.read_text()
    if FALLBACK_LOG_PATH.exists():
        return FALLBACK_LOG_PATH.read_text()
    return "Sample log file not found. Please paste your log content here."


def estimate_tokens(text: str) -> int:
    """Rough token estimate (chars / 4)."""
    return len(text) // 4 if text else 0


@spaces.GPU(duration=120)
def analyze_logs(
    log_content: str,
    window_size: int,
    k_neighbors: int,
    mode: str,
    anomaly_percentile: float,
    range_min: float,
    range_max: float,
    output_format: str = "xml",
    token_budget: float | None = None,
    max_blocks: float | None = None,
    min_score: float = 0.0,
    progress=gr.Progress(),
):
    """Analyze log content and return results.

    This function is decorated with @spaces.GPU to run on ZeroGPU.
    The GPU is allocated when this function is called.

    IMPORTANT: torch and cordon are imported here (not at module level)
    because ZeroGPU only allocates GPU resources inside @spaces.GPU functions.
    """
    # Lazy imports - only load when GPU is allocated
    import torch
    from cordon import AnalysisConfig, SemanticLogAnalyzer

    start_time = time.time()
    logger.info("=" * 60)
    logger.info("Analysis started - GPU allocated by ZeroGPU")

    if not log_content.strip():
        logger.warning("Empty log content provided")
        return "", "Anomalous Sections"

    logger.info(
        f"Log content size: {len(log_content)} characters "
        f"({len(log_content.splitlines())} lines)"
    )

    progress(0.1, desc="Configuring analyzer...")

    # Check if CUDA is available (should be True on ZeroGPU)
    device = "cuda" if torch.cuda.is_available() else "cpu"
    batch_size = 128 if device == "cuda" else 32

    logger.info(f"Device: {device}")
    logger.info(f"Batch size: {batch_size}")

    config_kwargs: dict = {
        "window_size": window_size,
        "k_neighbors": k_neighbors,
        "device": device,
        "batch_size": batch_size,
        "show_progress": False,
        "output_format": output_format,
    }

    if token_budget is not None and token_budget > 0:
        config_kwargs["token_budget"] = int(token_budget)
    if max_blocks is not None and max_blocks > 0:
        config_kwargs["max_blocks"] = int(max_blocks)
    if min_score > 0:
        config_kwargs["min_score"] = min_score

    if mode == "Percentile":
        config_kwargs["anomaly_percentile"] = anomaly_percentile
    else:
        config_kwargs["anomaly_range_min"] = range_min
        config_kwargs["anomaly_range_max"] = range_max

    config = AnalysisConfig(**config_kwargs)
    logger.info(f"Configuration: mode={mode}, {config_kwargs}")

    try:
        progress(0.15, desc="Loading embedding model...")
        logger.info("Loading embedding model...")
        model_load_start = time.time()
        analyzer = SemanticLogAnalyzer(config)
        logger.info(f"Embedding model loaded in {time.time() - model_load_start:.2f}s")

        progress(0.4, desc="Analyzing patterns...")
        logger.info("Starting log analysis...")
        analysis_start = time.time()
        result = analyzer.analyze_text_detailed(log_content)
        logger.info(f"Analysis completed in {time.time() - analysis_start:.2f}s")

        progress(0.9, desc="Formatting output...")
        progress(1.0, desc="Complete!")

        # Calculate token counts
        input_tokens = estimate_tokens(log_content)
        output_tokens = estimate_tokens(result.output)
        reduction = (
            (input_tokens - output_tokens) / input_tokens * 100
            if input_tokens > 0 else 0
        )

        block_count = len(result.blocks)
        if block_count > 0:
            scores = [b.max_score for b in result.blocks]
            score_range = f"scores: {min(scores):.3f}-{max(scores):.3f}"
        else:
            score_range = "no anomalies"

        label = (
            f"Anomalous Sections — {input_tokens:,} → {output_tokens:,} tokens "
            f"({reduction:.0f}% reduction) | {block_count} blocks, {score_range}"
        )

        total_time = time.time() - start_time
        logger.info(f"Total analysis time: {total_time:.2f}s")
        logger.info(f"Token reduction: {input_tokens:,} → {output_tokens:,} ({reduction:.0f}%)")
        logger.info("=" * 60)

        return gr.update(value=result.output, label=label)

    except Exception as e:
        logger.error(f"Error during analysis: {str(e)}", exc_info=True)
        error_msg = f"Analysis failed: {str(e)}"
        return gr.update(value=error_msg, label="Error")


CUSTOM_CSS = """
.main, .wrap, .contain, .gradio-container, .app {
    max-width: 100% !important;
    width: 100% !important;
    padding-left: 2rem !important;
    padding-right: 2rem !important;
}

/* Monospace font for log input/output */
textarea {
    font-family: "JetBrains Mono", "Fira Code", "SF Mono", "Cascadia Code", Menlo, Monaco, Consolas, "Liberation Mono", monospace !important;
    font-size: 13px !important;
    line-height: 1.5 !important;
}
"""

THEME = gr.themes.Base(
    primary_hue="teal",
    secondary_hue="cyan",
    neutral_hue="slate",
).set(
    body_background_fill="#0f172a",
    body_text_color="#e2e8f0",
    block_background_fill="#1e293b",
    block_border_color="#334155",
    block_title_text_color="#5eead4",
    input_background_fill="#334155",
    input_border_color="#475569",
    button_primary_background_fill="#14b8a6",
    button_primary_background_fill_hover="#2dd4bf",
    button_primary_text_color="#ffffff",
)


def create_interface():
    """Create the Gradio interface."""

    with gr.Blocks(title="Cordon", theme=THEME, css=CUSTOM_CSS) as demo:

        # Header
        gr.HTML("""
        <div style="text-align: center; padding: 16px 0 24px 0;">
            <a href="https://github.com/calebevans/cordon" target="_blank" style="text-decoration: none;">
                <h1 style="font-size: 2.5rem; letter-spacing: 0.2em; margin: 0 0 8px 0; color: #2dd4bf; cursor: pointer;">CORDON</h1>
            </a>
            <p style="color: #94a3b8; margin: 0;">Reduce logs to their semantically anomalous parts.</p>
        </div>
        """)

        # Main content - two equal columns
        with gr.Row():

            # Left: Input
            with gr.Column(scale=1):
                log_input = gr.Textbox(
                    label="Input Log",
                    placeholder="Paste your log content here...",
                    lines=22,
                )

                with gr.Row():
                    load_sample_btn = gr.Button("📂 Load Sample", size="sm")
                    clear_btn = gr.Button("Clear", size="sm")

                with gr.Row():
                    window_size = gr.Slider(2, 20, value=4, step=1, label="Window Size")
                    k_neighbors = gr.Slider(2, 20, value=5, step=1, label="k Neighbors")

                with gr.Row():
                    mode_select = gr.Dropdown(
                        choices=["Percentile", "Range"],
                        value="Percentile",
                        label="Mode",
                        scale=1
                    )
                    anomaly_percentile = gr.Slider(0.01, 0.5, value=0.1, step=0.01, label="Top N%", scale=2)
                    range_min = gr.Slider(0.0, 0.49, value=0.05, step=0.01, label="Min %", visible=False, scale=1)
                    range_max = gr.Slider(0.01, 0.5, value=0.15, step=0.01, label="Max %", visible=False, scale=1)

                def toggle_mode(mode):
                    if mode == "Percentile":
                        return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)
                    else:
                        return gr.update(visible=False), gr.update(visible=True), gr.update(visible=True)

                mode_select.change(
                    fn=toggle_mode,
                    inputs=[mode_select],
                    outputs=[anomaly_percentile, range_min, range_max]
                )

                with gr.Accordion("Advanced Options", open=False):
                    with gr.Row():
                        output_format = gr.Dropdown(
                            choices=["xml", "json"],
                            value="xml",
                            label="Output Format",
                            scale=1,
                        )
                        token_budget = gr.Number(
                            value=None,
                            label="Token Budget",
                            precision=0,
                            minimum=1,
                            info="Max tokens for output (auto-adjusts percentile)",
                            scale=1,
                        )
                    with gr.Row():
                        max_blocks = gr.Number(
                            value=None,
                            label="Max Blocks",
                            precision=0,
                            minimum=1,
                            info="Limit to top N blocks by score",
                            scale=1,
                        )
                        min_score = gr.Slider(
                            0.0, 1.0, value=0.0, step=0.01,
                            label="Min Score",
                            info="Filter blocks below this score",
                            scale=1,
                        )

                analyze_btn = gr.Button("🔍 Analyze", variant="primary", size="lg")

            # Right: Output
            with gr.Column(scale=1):
                output_text = gr.Textbox(
                    label="Anomalous Sections",
                    placeholder="Results will appear here...",
                    lines=37,
                )


        # Documentation section
        gr.Markdown("""
---

## Documentation

**Cordon** is a semantic anomaly detection tool for system log files. It uses transformer embeddings and k-NN density scoring to identify semantically unusual patterns, reducing massive logs to their most anomalous sections.

### Why Cordon?

Traditional log analysis tools use keyword matching or regex patterns. Cordon understands the *meaning* of log content. This means:

- **Repetitive errors are filtered out** — The same error 1000 times? That's "normal background noise."
- **Rare patterns surface** — Unusual state transitions, unexpected clusters, one-off events.
- **No rules to maintain** — The model learns what's normal from your logs.

### How It Works

```
Log File → Segmentation → Embedding → Scoring → Thresholding → Output
```

1. **Segmentation** — Split logs into windows of N lines
2. **Embedding** — Convert each window to a vector using transformer models
3. **Scoring** — For each window, find its k nearest neighbors in the embedding space and calculate the average distance. Windows that are *far* from their neighbors are semantically unique (anomalous). Windows that are *close* to many others are repetitive (normal).
4. **Thresholding** — Keep top N% highest-scoring windows
5. **Output** — XML-tagged blocks with line numbers and scores

### Installation

```bash
# With uv (recommended)
uv pip install cordon

# With pip
pip install cordon
```

### Command Line Usage

```bash
# Basic usage
cordon system.log

# Customize parameters
cordon --window-size 10 --k-neighbors 10 --anomaly-percentile 0.05 app.log

# Use range mode (exclude top 5%, keep next 10%)
cordon --anomaly-range 0.05 0.15 app.log

# JSON output
cordon --format json system.log

# Quiet mode for CI
cordon --quiet -o anomalies.json --format json app.log

# Token budget (fit output in 2000 tokens)
cordon --token-budget 2000 system.log

# Filter output
cordon --max-blocks 10 --min-score 0.1 system.log

# Read from stdin
kubectl logs pod-name | cordon -
```

### Python API

```python
from cordon import SemanticLogAnalyzer, AnalysisConfig

config = AnalysisConfig(window_size=4, k_neighbors=5, anomaly_percentile=0.1)
analyzer = SemanticLogAnalyzer(config)

# From file
result = analyzer.analyze_file_detailed(Path("system.log"))

# From text (no temp file needed)
result = analyzer.analyze_text_detailed(log_text)

# Access structured blocks
for block in result.blocks:
    print(f"Lines {block.start_line}-{block.end_line}: score={block.max_score:.4f}")
```

### Links

- [GitHub Repository](https://github.com/calebevans/cordon)
- [PyPI Package](https://pypi.org/project/cordon/)
- [Red Hat Developer Article](https://developers.redhat.com/articles/2025/12/09/semantic-anomaly-detection-log-files-cordon)
        """)

        # Events
        load_sample_btn.click(fn=load_sample_log, outputs=[log_input])
        clear_btn.click(fn=lambda: "", outputs=[log_input])

        analyze_btn.click(
            fn=analyze_logs,
            inputs=[
                log_input, window_size, k_neighbors, mode_select,
                anomaly_percentile, range_min, range_max,
                output_format, token_budget, max_blocks, min_score,
            ],
            outputs=[output_text],
        )

    return demo


# Startup logging
# NOTE: torch is imported lazily inside @spaces.GPU function for ZeroGPU compatibility
logger.info("=" * 60)
logger.info("Cordon Space starting...")
logger.info(f"Python version: {sys.version}")
logger.info(f"Gradio version: {gr.__version__}")
logger.info(f"ZeroGPU available: {ZEROGPU_AVAILABLE}")
logger.info("PyTorch/Cordon loaded lazily inside @spaces.GPU function")
logger.info("=" * 60)

# Create and launch the demo
try:
    logger.info("Creating Gradio interface...")
    demo = create_interface()
    logger.info("Gradio interface created successfully")

    # Detect if running on HuggingFace Spaces
    is_hf_space = os.getenv("SPACE_ID") is not None

    if is_hf_space:
        # On HuggingFace: use strict port binding for health checks
        logger.info("Running on HuggingFace Spaces - using port 7860")
        demo.launch(
            server_name="0.0.0.0",
            server_port=7860,
        )
    else:
        # Local development: let Gradio find an available port
        logger.info("Running locally - auto-detecting available port")
        demo.launch()
    logger.info("Demo launched successfully")
except Exception as e:
    logger.error(f"Failed to launch demo: {str(e)}", exc_info=True)
    raise