Spaces:

calebdevans
/

cordon

Paused

cordon / app.py

Enable ZeroGPU (A10G) and increase batch size

16d7c54 unverified 11 months ago

11.8 kB

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

import tempfile
from pathlib import Path

import gradio as gr
import torch

from cordon import AnalysisConfig, SemanticLogAnalyzer

# ZeroGPU support with fallback for local development
try:
    import spaces
    ZEROGPU_AVAILABLE = True
except ImportError:
    ZEROGPU_AVAILABLE = False

# Detect available device
def get_device() -> str:
    """Detect best available device."""
    if torch.cuda.is_available():
        return "cuda"
    elif torch.backends.mps.is_available():
        return "mps"
    return "cpu"

# Global model cache (keyed by model_name:device)
_analyzer_cache: dict[str, SemanticLogAnalyzer] = {}

def get_cached_analyzer(config: AnalysisConfig) -> SemanticLogAnalyzer:
    """Get or create a cached analyzer instance.
    
    Caches the analyzer to avoid repeated model loading.
    The config is updated on the cached instance for each call.
    """
    cache_key = f"{config.model_name}:{config.device}"
    if cache_key not in _analyzer_cache:
        _analyzer_cache[cache_key] = SemanticLogAnalyzer(config)
    else:
        # Update config params (window_size, k_neighbors, etc.)
        _analyzer_cache[cache_key].config = config
    return _analyzer_cache[cache_key]

# 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


def _analyze_logs_impl(
    log_content: str,
    window_size: int,
    k_neighbors: int,
    mode: str,
    anomaly_percentile: float,
    range_min: float,
    range_max: float,
    progress: gr.Progress = gr.Progress(),
) -> tuple[str, str]:
    """Analyze log content and return results."""
    
    if not log_content.strip():
        return "", "Anomalous Sections"
    
    progress(0.05, desc="Preparing log file...")
    
    # Write content to temp file
    with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
        f.write(log_content)
        temp_path = Path(f.name)
    
    try:
        progress(0.1, desc="Configuring analyzer...")
        
        device = get_device()
        # Use larger batch size on GPU for better throughput
        batch_size = 128 if device == "cuda" else 32
        
        if mode == "Percentile":
            config = AnalysisConfig(
                window_size=window_size,
                k_neighbors=k_neighbors,
                anomaly_percentile=anomaly_percentile,
                device=device,
                batch_size=batch_size,
            )
        else:
            config = AnalysisConfig(
                window_size=window_size,
                k_neighbors=k_neighbors,
                anomaly_range_min=range_min,
                anomaly_range_max=range_max,
                device=device,
                batch_size=batch_size,
            )
        
        progress(0.15, desc="Loading embedding model...")
        analyzer = get_cached_analyzer(config)
        
        progress(0.25, desc="Segmenting log into windows...")
        # Small delay to show progress
        import time
        time.sleep(0.1)
        
        progress(0.4, desc="Generating embeddings...")
        time.sleep(0.1)
        
        progress(0.6, desc="Scoring anomalies...")
        time.sleep(0.1)
        
        progress(0.75, desc="Analyzing patterns...")
        result = analyzer.analyze_file_detailed(temp_path)
        
        progress(0.9, desc="Formatting output...")
        time.sleep(0.05)
        
        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
        
        label = f"Anomalous Sections — {input_tokens:,} → {output_tokens:,} tokens ({reduction:.0f}% reduction)"
        
        return result.output, label
        
    finally:
        temp_path.unlink(missing_ok=True)


# Apply ZeroGPU decorator if available, otherwise use the function directly
if ZEROGPU_AVAILABLE:
    analyze_logs = spaces.GPU(duration=120)(_analyze_logs_impl)
else:
    analyze_logs = _analyze_logs_impl


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", fill_width=True, 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]
                )
                
                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
# Save to file
cordon --output anomalies.xml system.log
```
### Python API
```python
from cordon import SemanticLogAnalyzer, AnalysisConfig
config = AnalysisConfig(window_size=4, k_neighbors=5, anomaly_percentile=0.1)
analyzer = SemanticLogAnalyzer(config)
result = analyzer.analyze_file(Path("system.log"))
```
### 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])
        
        def run_analysis(log_content, window_size, k_neighbors, mode, anomaly_percentile, range_min, range_max, progress=gr.Progress()):
            output, label = analyze_logs(log_content, window_size, k_neighbors, mode, anomaly_percentile, range_min, range_max, progress)
            return gr.update(value=output, label=label)
        
        analyze_btn.click(
            fn=run_analysis,
            inputs=[log_input, window_size, k_neighbors, mode_select, anomaly_percentile, range_min, range_max],
            outputs=[output_text],
        )
    
    return demo


if __name__ == "__main__":
    demo = create_interface()
    demo.launch()

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

	import tempfile
	from pathlib import Path

	import gradio as gr
	import torch

	from cordon import AnalysisConfig, SemanticLogAnalyzer

	# ZeroGPU support with fallback for local development
	try:
	import spaces
	ZEROGPU_AVAILABLE = True
	except ImportError:
	ZEROGPU_AVAILABLE = False

	# Detect available device
	def get_device() -> str:
	"""Detect best available device."""
	if torch.cuda.is_available():
	return "cuda"
	elif torch.backends.mps.is_available():
	return "mps"
	return "cpu"

	# Global model cache (keyed by model_name:device)
	_analyzer_cache: dict[str, SemanticLogAnalyzer] = {}

	def get_cached_analyzer(config: AnalysisConfig) -> SemanticLogAnalyzer:
	"""Get or create a cached analyzer instance.

	Caches the analyzer to avoid repeated model loading.
	The config is updated on the cached instance for each call.
	"""
	cache_key = f"{config.model_name}:{config.device}"
	if cache_key not in _analyzer_cache:
	_analyzer_cache[cache_key] = SemanticLogAnalyzer(config)
	else:
	# Update config params (window_size, k_neighbors, etc.)
	_analyzer_cache[cache_key].config = config
	return _analyzer_cache[cache_key]

	# 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


	def _analyze_logs_impl(
	log_content: str,
	window_size: int,
	k_neighbors: int,
	mode: str,
	anomaly_percentile: float,
	range_min: float,
	range_max: float,
	progress: gr.Progress = gr.Progress(),
	) -> tuple[str, str]:
	"""Analyze log content and return results."""

	if not log_content.strip():
	return "", "Anomalous Sections"

	progress(0.05, desc="Preparing log file...")

	# Write content to temp file
	with tempfile.NamedTemporaryFile(mode="w", suffix=".log", delete=False) as f:
	f.write(log_content)
	temp_path = Path(f.name)

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

	device = get_device()
	# Use larger batch size on GPU for better throughput
	batch_size = 128 if device == "cuda" else 32

	if mode == "Percentile":
	config = AnalysisConfig(
	window_size=window_size,
	k_neighbors=k_neighbors,
	anomaly_percentile=anomaly_percentile,
	device=device,
	batch_size=batch_size,
	)
	else:
	config = AnalysisConfig(
	window_size=window_size,
	k_neighbors=k_neighbors,
	anomaly_range_min=range_min,
	anomaly_range_max=range_max,
	device=device,
	batch_size=batch_size,
	)

	progress(0.15, desc="Loading embedding model...")
	analyzer = get_cached_analyzer(config)

	progress(0.25, desc="Segmenting log into windows...")
	# Small delay to show progress
	import time
	time.sleep(0.1)

	progress(0.4, desc="Generating embeddings...")
	time.sleep(0.1)

	progress(0.6, desc="Scoring anomalies...")
	time.sleep(0.1)

	progress(0.75, desc="Analyzing patterns...")
	result = analyzer.analyze_file_detailed(temp_path)

	progress(0.9, desc="Formatting output...")
	time.sleep(0.05)

	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

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

	return result.output, label

	finally:
	temp_path.unlink(missing_ok=True)


	# Apply ZeroGPU decorator if available, otherwise use the function directly
	if ZEROGPU_AVAILABLE:
	analyze_logs = spaces.GPU(duration=120)(_analyze_logs_impl)
	else:
	analyze_logs = _analyze_logs_impl


	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", fill_width=True, 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]
	)

	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

	# Save to file
	cordon --output anomalies.xml system.log
	```

	### Python API

	```python
	from cordon import SemanticLogAnalyzer, AnalysisConfig

	config = AnalysisConfig(window_size=4, k_neighbors=5, anomaly_percentile=0.1)
	analyzer = SemanticLogAnalyzer(config)
	result = analyzer.analyze_file(Path("system.log"))
	```

	### 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])

	def run_analysis(log_content, window_size, k_neighbors, mode, anomaly_percentile, range_min, range_max, progress=gr.Progress()):
	output, label = analyze_logs(log_content, window_size, k_neighbors, mode, anomaly_percentile, range_min, range_max, progress)
	return gr.update(value=output, label=label)

	analyze_btn.click(
	fn=run_analysis,
	inputs=[log_input, window_size, k_neighbors, mode_select, anomaly_percentile, range_min, range_max],
	outputs=[output_text],
	)

	return demo


	if __name__ == "__main__":
	demo = create_interface()
	demo.launch()