Observability-Driven Development (ODD), Telemetry-First Coding Practices, and Preemptive Debugging Architecture: Best Practices¶

# Logging from day one
import logging
from opentelemetry import trace

logger = logging.getLogger(__name__)
tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("process_data")
def process_data(data: dict):
    """Process data with observability"""
    logger.info("Processing data", extra={"data_id": data.get("id")})

    try:
        result = transform(data)
        logger.info("Data processed successfully", extra={"result_id": result.id})
        return result
    except Exception as e:
        logger.error("Data processing failed", exc_info=True, extra={"data_id": data.get("id")})
        raise

# Tracing from day one
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Setup tracing
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Add span processor
otlp_exporter = OTLPSpanExporter(endpoint="http://collector:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

@tracer.start_as_current_span("api_request")
def handle_request(request):
    """Handle request with tracing"""
    with tracer.start_as_current_span("database_query") as span:
        span.set_attribute("db.query", "SELECT * FROM users")
        result = db.query("SELECT * FROM users")
        span.set_attribute("db.rows", len(result))
    return result

# Metrics from day one
from prometheus_client import Counter, Histogram, Gauge

# Define metrics
request_count = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint'])
request_duration = Histogram('http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])
active_connections = Gauge('active_connections', 'Active connections')

@app.route('/api/users')
def get_users():
    """API endpoint with metrics"""
    start_time = time.time()

    try:
        result = fetch_users()
        request_count.labels(method='GET', endpoint='/api/users').inc()
        return result
    finally:
        request_duration.labels(method='GET', endpoint='/api/users').observe(time.time() - start_time)

# Python instrumentation standard
from opentelemetry import trace, metrics
from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
from opentelemetry.instrumentation.psycopg2 import Psycopg2Instrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# Auto-instrumentation
FastAPIInstrumentor.instrument_app(app)
Psycopg2Instrumentor.instrument()
RequestsInstrumentor.instrument()

# Observability scaffold for new repo
# observability.py
from opentelemetry import trace, metrics
from prometheus_client import Counter, Histogram
import logging

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# Setup tracing
tracer = trace.get_tracer(__name__)

# Setup metrics
request_count = Counter('requests_total', 'Total requests')
request_duration = Histogram('request_duration_seconds', 'Request duration')

# Geospatial pipeline observability
from opentelemetry import trace

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("process_raster")
def process_raster(raster_path: str):
    """Process raster with observability"""
    span = trace.get_current_span()

    # Add geospatial context
    span.set_attribute("raster.path", raster_path)
    span.set_attribute("raster.size", get_raster_size(raster_path))
    span.set_attribute("raster.crs", get_raster_crs(raster_path))
    span.set_attribute("raster.bounds", get_raster_bounds(raster_path))

    # Process raster
    result = process(raster_path)

    # Add result context
    span.set_attribute("result.size", result.size)
    span.set_attribute("result.duration", result.duration)

    return result

# Signal coverage fitness function
class SignalCoverageFitnessFunction:
    def evaluate(self, codebase: Codebase) -> float:
        """Evaluate signal coverage"""
        # Count instrumented functions
        instrumented = self.count_instrumented_functions(codebase)

        # Count total functions
        total = self.count_total_functions(codebase)

        # Calculate coverage
        if total == 0:
            coverage = 1.0
        else:
            coverage = instrumented / total

        return coverage

# Trace completeness fitness function
class TraceCompletenessFitnessFunction:
    def evaluate(self, system: System) -> float:
        """Evaluate trace completeness"""
        # Check trace coverage
        trace_coverage = self.check_trace_coverage(system)

        # Check span attributes
        span_attributes = self.check_span_attributes(system)

        # Check trace correlation
        trace_correlation = self.check_trace_correlation(system)

        # Calculate fitness
        fitness = (trace_coverage * 0.4) + \
                  (span_attributes * 0.3) + \
                  (trace_correlation * 0.3)

        return fitness

# Bad: Unobservable code path
def process_data(data):
    return transform(data)  # No logging, tracing, or metrics!

# Good: Observable code path
@tracer.start_as_current_span("process_data")
def process_data(data):
    logger.info("Processing data", extra={"data_id": data.id})
    try:
        result = transform(data)
        logger.info("Data processed", extra={"result_id": result.id})
        return result
    except Exception as e:
        logger.error("Processing failed", exc_info=True)
        raise