akaszubski/observability

Observability Skill

Developer-focused observability: logging, debugging, profiling, and performance monitoring.

When This Skill Activates

• Adding logging to code
• Debugging errors or issues
• Profiling performance bottlenecks
• Analyzing memory usage
• Adding instrumentation
• Setting up structured logging
• Keywords: "logging", "debug", "profiling", "performance", "trace"

---

Structured Logging

Why Structured Logging?

Traditional (unstructured):

print(f"User {user_id} logged in from {ip_address}")
# Hard to parse, search, filter

Structured (JSON):

logger.info("user_login", extra={
    "user_id": user_id,
    "ip_address": ip_address,
    "timestamp": datetime.utcnow().isoformat()
})
# Easy to parse, search, filter, aggregate

Benefits:

• ✅ Easy to search/filter
• ✅ Machine-readable
• ✅ Aggregate metrics
• ✅ Correlate events
• ✅ Feed to log aggregators (ELK, Splunk, Datadog)

---

Python Logging Setup

Basic configuration:

import logging
import sys

# Configure root logger
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),  # Console
        logging.FileHandler('app.log')      # File
    ]
)

logger = logging.getLogger(__name__)

# Use it
logger.info("Application started")
logger.warning("Database connection slow")
logger.error("Failed to process request", exc_info=True)

---

Log Levels

| Level | Value | Use When | Example | | ------------ | ----- | ----------------------------- | --------------------------------- | | DEBUG | 10 | Development, detailed tracing | Variable values, function calls | | INFO | 20 | Normal operations | User logged in, request processed | | WARNING | 30 | Unexpected but handled | Deprecated API used, slow query | | ERROR | 40 | Error occurred | Failed to save file, API error | | CRITICAL | 50 | System failure | Database down, disk full |

Usage:

logger.debug(f"Processing item: {item}")           # Development only
logger.info(f"User {user_id} logged in")           # Normal event
logger.warning(f"Query took {duration}s (slow!)")  # Unexpected
logger.error(f"Failed to save: {error}")           # Error
logger.critical(f"Database unreachable!")          # System failure

---

Structured JSON Logging

Using python-json-logger:

from pythonjsonlogger import jsonlogger
import logging

# Setup JSON formatter
logHandler = logging.StreamHandler()
formatter = jsonlogger.JsonFormatter(
    '%(asctime)s %(name)s %(levelname)s %(message)s'
)
logHandler.setFormatter(formatter)

logger = logging.getLogger()
logger.addHandler(logHandler)
logger.setLevel(logging.INFO)

# Log with extra fields
logger.info("user_login", extra={
    "user_id": 123,
    "ip_address": "192.168.1.1",
    "user_agent": "Mozilla/5.0..."
})

Output:

{
  "asctime": "2025-10-24 12:00:00,000",
  "name": "root",
  "levelname": "INFO",
  "message": "user_login",
  "user_id": 123,
  "ip_address": "192.168.1.1",
  "user_agent": "Mozilla/5.0..."
}

---

Context Logging

Add context to all logs in a function:

import logging
from contextvars import ContextVar

# Context variable
request_id_var: ContextVar[str] = ContextVar('request_id', default=None)

class ContextFilter(logging.Filter):
    def filter(self, record):
        record.request_id = request_id_var.get()
        return True

# Add filter
logger.addFilter(ContextFilter())

# Use in request handler
def handle_request(request):
    request_id = generate_request_id()
    request_id_var.set(request_id)

    logger.info("Processing request")  # Includes request_id automatically
    process_data()
    logger.info("Request complete")    # Includes request_id automatically

---

Best Practices for Logging

✅ DO:

# 1. Use appropriate log levels
logger.debug("Variable value: {value}")      # Debug details
logger.info("User action completed")         # Normal operations
logger.error("Failed to process", exc_info=True)  # Errors

# 2. Include context
logger.info("Processing payment", extra={
    "user_id": user_id,
    "amount": amount,
    "currency": currency
})

# 3. Log exceptions with stack traces
try:
    process_data()
except Exception as e:
    logger.error("Processing failed", exc_info=True)  # Includes stack trace

# 4. Use lazy formatting
logger.debug("Processing %s items", len(items))  # Only formats if DEBUG enabled

❌ DON'T:

# 1. Don't log sensitive data
logger.info(f"User password: {password}")  # NEVER!
logger.info(f"Credit card: {card_number}")  # NEVER!

# 2. Don't log in loops (too verbose)
for item in items:
    logger.info(f"Processing {item}")  # 1M items = 1M log lines!

# 3. Don't use print() in production
print("Debug info")  # Use logger.debug() instead

# 4. Don't catch and ignore
try:
    process()
except:
    pass  # Silent failure! At least log it!

---

Debugging

Print Debugging

Strategic print statements:

def process_data(data):
    print(f"[DEBUG] Input: {data}")  # See input

    result = transform(data)
    print(f"[DEBUG] After transform: {result}")  # See intermediate

    validated = validate(result)
    print(f"[DEBUG] After validate: {validated}")  # See output

    return validated

Use temporary, remove before commit:

# ✅ GOOD: Temporary debug
print(f"[DEBUG] value={value}")  # Remove before commit

# ❌ BAD: Permanent debug prints
print("Processing...")  # Clutters output forever

---

Interactive Debugger (pdb)

Built-in Python debugger:

import pdb

def buggy_function(x, y):
    result = x + y
    pdb.set_trace()  # Debugger starts here
    return result * 2

# When debugger starts:
# (Pdb) p x          # Print variable x
# (Pdb) p y          # Print variable y
# (Pdb) p result     # Print result
# (Pdb) n            # Next line
# (Pdb) s            # Step into function
# (Pdb) c            # Continue execution
# (Pdb) q            # Quit debugger

Breakpoint (Python 3.7+):

def buggy_function(x, y):
    result = x + y
    breakpoint()  # Better than pdb.set_trace()
    return result * 2

---

IPython Debugger (ipdb)

Enhanced debugger with syntax highlighting:

pip install ipdb

import ipdb

def buggy_function(x, y):
    result = x + y
    ipdb.set_trace()  # Colorized, tab completion
    return result * 2

Commands:

(ipdb) p x              # Print x
(ipdb) pp x             # Pretty-print x
(ipdb) l                # List code around current line
(ipdb) w                # Where (stack trace)
(ipdb) u                # Up stack frame
(ipdb) d                # Down stack frame
(ipdb) h                # Help

---

Post-Mortem Debugging

Debug after exception:

import pdb
import sys

def buggy_function():
    x = 1
    y = 0
    return x / y  # ZeroDivisionError

try:
    buggy_function()
except:
    pdb.post_mortem(sys.exc_info()[2])
    # Debugger starts at exception point

Automatic on uncaught exception:

import sys
import pdb

def exception_handler(exc_type, exc_value, exc_traceback):
    if exc_type is KeyboardInterrupt:
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    print("Uncaught exception, starting debugger:")
    pdb.post_mortem(exc_traceback)

sys.excepthook = exception_handler

# Now any uncaught exception starts debugger

---

Profiling

CPU Profiling (cProfile)

Find slow functions:

import cProfile
import pstats

def slow_function():
    # Your code here
    pass

# Profile
profiler = cProfile.Profile()
profiler.enable()
slow_function()
profiler.disable()

# Print stats
stats = pstats.Stats(profiler)
stats.sort_stats('cumulative')
stats.print_stats(10)  # Top 10 slowest functions

Output:

   ncalls  tottime  percall  cumtime  percall filename:lineno(function)
       10    0.500    0.050    0.800    0.080 mymodule.py:10(slow_function)
      100    0.300    0.003    0.300    0.003 mymodule.py:20(helper)

Sort options:

• cumulative - Total time (includes subfunctions)
• time - Internal time (excludes subfunctions)
• calls - Number of calls

---

Profile Decorator

Profile specific functions:

import cProfile
import pstats
from functools import wraps

def profile(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()
        result = func(*args, **kwargs)
        profiler.disable()

        stats = pstats.Stats(profiler)
        stats.sort_stats('cumulative')
        stats.print_stats(10)

        return result
    return wrapper

@profile
def slow_function():
    # Your code
    pass

---

Line Profiler (line_profiler)

Profile line-by-line:

pip install line_profiler

from line_profiler import LineProfiler

def slow_function():
    total = 0
    for i in range(1000000):
        total += i
    return total

profiler = LineProfiler()
profiler.add_function(slow_function)
profiler.enable()
slow_function()
profiler.disable()
profiler.print_stats()

Output:

Line #      Hits         Time  Per Hit   % Time  Line Contents
==============================================================
     2                                           def slow_function():
     3         1          2.0      2.0      0.0      total = 0
     4   1000001     150000.0      0.1     30.0      for i in range(1000000):
     5   1000000     350000.0      0.4     70.0          total += i
     6         1          1.0      1.0      0.0      return total

---

Memory Profiling (memory_profiler)

Find memory leaks:

pip install memory_profiler

from memory_profiler import profile

@profile
def memory_hog():
    big_list = [0] * (10 ** 7)  # 10M items
    return sum(big_list)

memory_hog()

Output:

Line #    Mem usage    Increment  Occurrences   Line Contents
=============================================================
     2     40.0 MiB     40.0 MiB           1   @profile
     3                                         def memory_hog():
     4    116.4 MiB     76.4 MiB           1       big_list = [0] * (10 ** 7)
     5    116.4 MiB      0.0 MiB           1       return sum(big_list)

---

Sampling Profiler (py-spy)

Profile running processes (no code changes needed):

pip install py-spy

# Profile running Python process
py-spy top --pid 12345

# Generate flamegraph
py-spy record -o profile.svg --pid 12345

Advantages:

• ✅ No code modification
• ✅ Profile production code
• ✅ Low overhead
• ✅ Visualizations (flamegraphs)

---

Stack Traces

Print Stack Trace

import traceback

try:
    risky_operation()
except Exception as e:
    traceback.print_exc()  # Print stack trace to stderr

Capture to string:

import traceback

try:
    risky_operation()
except Exception as e:
    trace = traceback.format_exc()
    logger.error(f"Error occurred: {trace}")

---

Rich Tracebacks (rich)

Beautiful, informative tracebacks:

pip install rich

from rich.traceback import install
install(show_locals=True)

def buggy_function():
    x = 42
    y = None
    return x + y  # TypeError with beautiful traceback

buggy_function()

Shows:

• ✅ Syntax highlighted code
• ✅ Local variables at each frame
• ✅ Clear error message

---

Performance Monitoring

Timing Decorator

Measure function execution time:

import time
from functools import wraps

def timer(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        result = func(*args, **kwargs)
        end = time.perf_counter()

        duration = end - start
        print(f"{func.__name__} took {duration:.4f}s")

        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)
    return "done"

slow_function()
# Output: slow_function took 1.0001s

---

Context Manager Timer

import time
from contextlib import contextmanager

@contextmanager
def timer(name):
    start = time.perf_counter()
    yield
    end = time.perf_counter()
    print(f"{name} took {end - start:.4f}s")

# Usage
with timer("Database query"):
    result = db.query(...)

with timer("File processing"):
    process_file(path)

---

Performance Assertions

Fail if too slow:

import time

def performance_test():
    start = time.perf_counter()

    # Your code
    process_data()

    duration = time.perf_counter() - start
    assert duration < 0.1, f"Too slow: {duration:.4f}s"

---

Simple Metrics

Counter

class Metrics:
    def __init__(self):
        self.counters = {}

    def increment(self, name, value=1):
        self.counters[name] = self.counters.get(name, 0) + value

    def get(self, name):
        return self.counters.get(name, 0)

    def reset(self):
        self.counters = {}

# Usage
metrics = Metrics()

def process_item(item):
    metrics.increment("items_processed")
    if item.is_valid():
        metrics.increment("items_valid")
    else:
        metrics.increment("items_invalid")

# Later
print(f"Processed: {metrics.get('items_processed')}")
print(f"Valid: {metrics.get('items_valid')}")
print(f"Invalid: {metrics.get('items_invalid')}")

---

Histogram

Track value distributions:

from collections import defaultdict

class Histogram:
    def __init__(self):
        self.buckets = defaultdict(int)

    def record(self, value):
        # Bucket by value range
        if value < 1:
            bucket = "< 1s"
        elif value < 5:
            bucket = "1-5s"
        elif value < 10:
            bucket = "5-10s"
        else:
            bucket = "> 10s"

        self.buckets[bucket] += 1

    def report(self):
        for bucket, count in sorted(self.buckets.items()):
            print(f"{bucket}: {count}")

# Usage
histogram = Histogram()

for duration in query_durations:
    histogram.record(duration)

histogram.report()
# Output:
# < 1s: 850
# 1-5s: 120
# 5-10s: 25
# > 10s: 5

---

Debugging Best Practices

1. Binary Search Debugging

Narrow down the problem:

# Problem: Something breaks between start and end

def debug_binary_search():
    # Step 1: Check middle
    middle_result = process_half()
    print(f"Middle result: {middle_result}")

    # Step 2: Narrow to first or second half
    if middle_result == expected:
        # Problem in second half
        debug_second_half()
    else:
        # Problem in first half
        debug_first_half()

---

2. Rubber Duck Debugging

Explain code out loud (or to a rubber duck):

1. Describe what code should do
2. Explain what code actually does
3. Often you'll spot the bug while explaining!

---

3. Add Assertions

Catch bugs early:

def divide(a, b):
    assert b != 0, "Divisor cannot be zero"
    assert isinstance(a, (int, float)), "a must be numeric"
    assert isinstance(b, (int, float)), "b must be numeric"

    result = a / b

    assert isinstance(result, (int, float)), "Result should be numeric"
    return result

---

4. Simplify and Isolate

Reduce problem to minimal case:

# ❌ BAD: Complex function with bug
def complex_function(data, config, options, flags):
    # 100 lines of code
    pass

# ✅ GOOD: Extract minimal failing case
def minimal_failing_case():
    data = [1, 2, 3]
    result = transform(data)  # Fails here
    return result

---

Logging Anti-Patterns

❌ DON'T:

# 1. Log sensitive data
logger.info(f"Password: {password}")

# 2. Log in tight loops
for item in million_items:
    logger.info(f"Processing {item}")

# 3. Catch and ignore silently
try:
    critical_operation()
except:
    pass  # Silent failure!

# 4. Over-log (log everything)
logger.debug("Starting function")
logger.debug("Variable x = 1")
logger.debug("Variable y = 2")
logger.debug("Calling helper")
logger.debug("Helper returned")

# 5. Use wrong log level
logger.error("User logged in")  # Should be INFO
logger.info("Database crashed")  # Should be CRITICAL

---

Quick Reference

Logging

logger.debug("Detailed trace")
logger.info("Normal event")
logger.warning("Unexpected")
logger.error("Error occurred", exc_info=True)
logger.critical("System failure")

Debugging

breakpoint()              # Start debugger
pdb.set_trace()          # Start debugger (old way)
traceback.print_exc()    # Print stack trace

Profiling

# CPU
python -m cProfile -s cumulative script.py

# Memory
@profile  # from memory_profiler
def func(): pass

# Live process
py-spy top --pid 12345

Timing

@timer
def func(): pass

with timer("operation"):
    do_work()

---

Key Takeaways

1. Structured logging - Use JSON for machine-readable logs
2. Appropriate log levels - DEBUG/INFO/WARNING/ERROR/CRITICAL
3. Don't log secrets - Never log passwords, API keys, etc.
4. Use debugger - breakpoint() or ipdb for interactive debugging
5. Profile first - Don't optimize without profiling
6. cProfile for CPU - Find slow functions
7. memory_profiler for memory - Find memory leaks
8. Stack traces on errors - Always log exc_info=True
9. Time critical paths - Use @timer decorator
10. Simple metrics - Track counters and histograms

---

Version: 1.0.0 Type: Knowledge skill (no scripts) See Also: python-standards (best practices), testing-guide (test debugging), security-patterns (safe logging)