Purpose

Implement metrics collection, logging, error tracking, health checks, and profiling for pipeline observability.

When To Use

• •Adding monitoring/observability to a data pipeline framework
• •Tracking pipeline execution metrics
• •Implementing structured logging
• •Setting up health checks
• •Adding performance profiling

Preconditions

• •Existing src/vibe_piper/ package structure
• •Python 3.12+ environment
• •UV package manager

Steps

1. Create monitoring module structure

mkdir -p src/vibe_piper/monitoring

2. Implement metrics collection (metrics.py)

• •Create MetricType enum (COUNTER, GAUGE, HISTOGRAM, TIMER, SUMMARY)
• •Create Metric dataclass (name, value, type, timestamp, labels, unit)
• •Create MetricsSnapshot dataclass with filtering methods
• •Create MetricsCollector class with:
  • •start_execution() / end_execution() for pipeline-level metrics
  • •record_metric() for custom metrics
  • •record_asset_execution() for AssetResult integration
  • •record_execution_result() for ExecutionResult integration
  • •get_snapshot() / to_dict() for export
  • •Thread-safe implementation with _lock

3. Implement structured logging (logging.py)

• •Create LogLevel enum (TRACE, DEBUG, INFO, WARNING, ERROR, CRITICAL)
• •Create JSONFormatter for machine-parsable logs
• •Create ColoredFormatter for console output
• •Create StructuredLogger wrapper with context support
• •Create log_execution() context manager for pipeline tracing
• •Create configure_logging() for setup

4. Implement health checks (health.py)

• •Create HealthStatus enum (HEALTHY, DEGRADED, UNHEALTHY, UNKNOWN)
• •Create HealthCheckResult dataclass
• •Create HealthChecker class with:
  • •register_check() / unregister_check() for dynamic checks
  • •run_check() / run_all_checks() for execution
  • •get_overall_health() for aggregate status
• •Create factory functions: create_disk_space_check(), create_memory_check()

5. Implement error aggregation (errors.py)

• •Create ErrorSeverity enum (LOW, MEDIUM, HIGH, CRITICAL)
• •Create ErrorCategory enum (VALIDATION, CONNECTION, TRANSFORMATION, IO, TIMEOUT, SYSTEM, UNKNOWN)
• •Create ErrorRecord dataclass with aggregation support
• •Create ErrorAggregator class with:
  • •add_error() for recording errors
  • •Aggregation window for similar errors
  • •Filtering methods (by severity, category, asset)
  • •get_summary() for analytics

6. Implement profiling (profiling.py)

• •Create ProfileData dataclass
• •Create Profiler class with:
  • •@profile decorator
  • •get_stats() / get_history() for analysis
  • •Optional psutil integration for memory tracking
• •Create profile_execution() context manager

7. Update package exports

# Edit src/vibe_piper/__init__.py
# Add monitoring imports to __all__

8. Type checking and linting

uv run mypy src/vibe_piper/monitoring/
uv run ruff check src/vibe_piper/monitoring/ --fix
uv run ruff format src/vibe_piper/monitoring/

9. Create test suite

mkdir -p tests/monitoring

Create tests for:

• •test_metrics.py: MetricsCollector, MetricsSnapshot, Metric
• •test_logging.py: LogLevel, formatters, StructuredLogger, log_execution
• •test_health.py: HealthChecker, health check functions
• •test_errors.py: ErrorAggregator, ErrorRecord
• •test_profiling.py: Profiler, ProfileData

10. Integration points

• •MetricsCollector integrates with ExecutionEngine via record_execution_result()
• •StructuredLogger can be used throughout codebase via get_logger()
• •HealthChecker for system/resource health monitoring
• •ErrorAggregator for tracking and alerting on errors

Examples

from vibe_piper.monitoring import (
    MetricsCollector,
    StructuredLogger,
    HealthChecker,
    ErrorAggregator,
    configure_logging,
)

# Configure logging
configure_logging(level=LogLevel.INFO, format_type="json")

# Collect metrics
metrics = MetricsCollector()
metrics.start_execution("my_pipeline", "run_123")
metrics.record_metric("custom_metric", 42)
metrics.end_execution()

# Health checks
health_checker = HealthChecker()
health_checker.register_check("disk", create_disk_space_check("/tmp"))
results = health_checker.run_all_checks()

Gotchas

• •Optional psutil dependency: handle ImportError gracefully
• •Thread-safety: use threading.Lock for shared state
• •Type safety: use Optional[T] with proper None handling
• •MyPy errors: use type: ignore[import-untyped] for untyped deps
• •Formatter type mismatch: use separate variables for each formatter type
• •Datetime arithmetic: coalesce None with datetime.utcnow()

Verification

uv run pytest tests/monitoring/ -v
uv run mypy src/vibe_piper/monitoring/ strict

Manual notes

This section is preserved when the skill is updated. Put human notes, caveats, and exceptions here.