Python Automation and Monitoring Questions

Focuses on using the Python programming language to automate operational tasks and implement monitoring workflows. Includes writing robust scripts and small applications for file input and output, subprocess management, making HTTP requests and handling responses, parsing and producing JSON, structured error handling and retries, logging and alerting, and clear code organization and packaging. Discuss when to choose Python over shell scripts for complexity, portability, error handling, and maintainability. Covers libraries and tooling for scheduling and background jobs, interacting with cloud provider software development kits, instrumenting applications for metrics and traces, integrating with monitoring stacks, and basic concurrency and asynchronous programming patterns when needed. Also includes testing and validation of automation scripts, secure handling of credentials and configuration, and deployment practices for operational scripts.

HardTechnical

37 practiced

Write a Python utility that watches Docker events, attaches to container logs, and streams them to a remote ingestion service. The utility should handle container restarts, resume streaming without duplicating logs (where possible), apply backpressure to the ingestion endpoint, and buffer locally during network outages. Outline the main components and show key code snippets using docker-py.

Sample Answer

Approach (brief)- Build an async watcher that subscribes to Docker events, attaches to container log streams, and forwards records to a remote ingestion service.- Use per-container checkpointing (last-seen timestamp + seq) persisted to disk to avoid duplicates after restarts.- Flow-control with an asyncio bounded queue and an async HTTP client that applies retry/backoff; if queue fills, buffer to local disk (sqlite) and pause reading.- Rehydrate buffer on network recovery and resume.

Main components1. DockerEventWatcher: listens to docker.events and starts/stops LogStreams.2. LogStream: attaches to container logs with docker-py, uses since timestamp from checkpoint, emits structured records.3. Forwarder: consumes queue, posts to ingestion API with backpressure, retries, and moves overflow to LocalBuffer.4. LocalBuffer: durable SQLite-backed queue for outages.5. Checkpointer: per-container persistence of last timestamp/offset.

Key code snippets

Docker event watcher + starting log stream:

python

import asyncio
import docker, json, time
from datetime import datetime

client = docker.from_env()

async def watch_events(queue, checkpoint):
    loop = asyncio.get_event_loop()
    for event in client.events(decode=True):
        if event.get("Type")=="container" and event.get("Action") in ("start","restart"):
            cid = event["id"][:12]
            asyncio.create_task(stream_container_logs(cid, queue, checkpoint))

Attaching logs with resume (uses since checkpoint timestamp):

python

def stream_container_logs(container_id, queue, checkpoint):
    c = client.containers.get(container_id)
    ck = checkpoint.get(container_id) or {"since": 0}
    since = ck["since"]
    # docker-py blocking generator
    for raw in c.logs(stream=True, since=since, timestamps=True, follow=True):
        # raw is bytes like b'2025-11-22T... log line'
        ts_str, _, line = raw.decode().partition(' ')
        ts = datetime.fromisoformat(ts_str).timestamp()
        record = {"container": container_id, "ts": ts, "line": line.rstrip()}
        # synchronous put into asyncio queue
        asyncio.get_event_loop().call_soon_threadsafe(asyncio.create_task, queue.put(record))
        checkpoint[container_id] = {"since": ts}

Async forwarder with backpressure and local buffering:

python

import aiohttp, aiosqlite

class Forwarder:
    def __init__(self, queue, buffer_db='buffer.db', max_q=1000):
        self.queue = queue
        self.sem = asyncio.Semaphore(10)  # concurrency to ingestion
        self.buffer_db = buffer_db

    async def start(self):
        self.session = aiohttp.ClientSession()
        await self._init_db()
        asyncio.create_task(self._drain_buffer())
        while True:
            rec = await self.queue.get()
            try:
                await self._send_with_backoff(rec)
            except Exception:
                await self._persist(rec)
            self.queue.task_done()

    async def _send_with_backoff(self, rec):
        async with self.sem:
            for attempt in range(6):
                try:
                    async with self.session.post("https://ingest.example/api", json=rec, timeout=10) as r:
                        if r.status==200:
                            return
                        elif 429 == r.status:
                            await asyncio.sleep(2**attempt)
                        else:
                            r.raise_for_status()
                except Exception:
                    await asyncio.sleep(2**attempt)
            raise RuntimeError("failed after retries")

    async def _init_db(self):
        async with aiosqlite.connect(self.buffer_db) as db:
            await db.execute("CREATE TABLE IF NOT EXISTS buffer(id INTEGER PRIMARY KEY, payload TEXT, ts REAL)")
            await db.commit()

    async def _persist(self, rec):
        async with aiosqlite.connect(self.buffer_db) as db:
            await db.execute("INSERT INTO buffer(payload,ts) VALUES (?,?)", (json.dumps(rec), rec["ts"]))
            await db.commit()

    async def _drain_buffer(self):
        while True:
            async with aiosqlite.connect(self.buffer_db) as db:
                async for row in db.execute("SELECT id,payload FROM buffer ORDER BY id LIMIT 100"):
                    id_, payload = row
                    rec = json.loads(payload)
                    try:
                        await self._send_with_backoff(rec)
                        await db.execute("DELETE FROM buffer WHERE id=?", (id_,))
                        await db.commit()
                    except Exception:
                        break
            await asyncio.sleep(5)

Checkpoint persistence (simple JSON or sqlite):

python

import json
def save_checkpoint(path, checkpoint):
    with open(path,'w') as f: json.dump(checkpoint,f)
def load_checkpoint(path):
    try:
        with open(path) as f: return json.load(f)
    except: return {}

Why this works / trade-offs- Using docker.events lets you react to restarts and new containers. Resuming by timestamp minimizes duplicates; exact-once is hard without application-level offsets, so we aim for at-least-once with minimal duplication.- Backpressure: bounded asyncio queue + semaphore prevents overwhelming ingestion; overflow persisted to SQLite avoids memory blowup.- Local buffering ensures durability during outages; draining asynchronously recovers.- Trade-offs: timestamps can duplicate lines if container clocks or log timestamps are coarse; for stronger guarantees consider embedding per-line monotonic offsets (if producer provides) or using a sidecar that writes offsets to shared storage.

Edge cases- Containers logging before watcher starts: initial since should be 0 or container.StartedAt.- Log timestamps missing or non-ISO formats — parse robustly.- High throughput: tune queue size, semaphore, and batching to ingestion API (POST batches to improve throughput).

This design balances reliability, backpressure, and practical resumability for SRE-grade log forwarding.

EasyTechnical

36 practiced

When should you choose Python instead of shell scripts for operational automation? Provide concrete criteria (complexity, portability, structured data, error handling, testing, third-party libraries) and a short real-world example for each criterion illustrating why Python is a better choice.

Sample Answer

Use Python instead of shell scripts when the task demands clarity, robustness, portability, or advanced logic. Below are concrete criteria with short real-world SRE examples.

1) Complexity of logic- Criteria: Multiple branching, loops, data structures, or modular code -> choose Python.- Example: Rolling-update that computes dependency graphs and parallelizes hosts. Python’s functions and data structures keep logic readable and maintainable vs bash with nested conditionals.

2) Portability- Criteria: Must run reliably across Linux, macOS, and BSD or inside containers with varying shells -> choose Python.- Example: A health-check agent deployed to diverse customer VMs; Python’s consistent runtime avoids POSIX shell differences and varying sed/awk behavior.

3) Structured data handling- Criteria: Parsing/generating JSON, YAML, protobufs -> choose Python.- Example: Aggregating JSON from multiple REST endpoints and emitting Prometheus metrics: Python’s json/yaml libraries and dicts simplify parsing and transformation.

4) Error handling & resilience- Criteria: Need fine-grained exceptions, retries, backoff, resource cleanup -> choose Python.- Example: An automated DB migration script that must rollback on partial failure; Python’s try/except/finally and context managers make safe cleanup straightforward.

5) Testing & maintainability- Criteria: Requires unit/integration tests, CI validation, or long-term maintenance -> choose Python.- Example: Alert-processing pipeline with many edge cases — pytest unit tests and mocks verify behavior before deployment.

6) Third-party libraries & ecosystem- Criteria: Need HTTP clients, cloud SDKs, cryptography, or async I/O -> choose Python.- Example: Provisioning cloud resources via AWS/GCP SDKs: boto3/google-cloud libraries in Python provide vetted APIs; reimplementing in shell is fragile.

When to stick with shell: trivial glue work (simple file ops, launching commands, quick one-liners) where invoking core Unix tools is simpler. For anything above “throwaway” automation in production SRE work, favor Python for correctness and long-term reliability.

HardTechnical

50 practiced

A Python automation's retry logic caused a retry storm and overwhelmed a downstream API. Walk through a corrective plan: identify and fix the bug, implement safeguards to prevent future retry storms (max retries, global rate limits, circuit breakers, jitter), and describe tests (unit, integration, chaos) to validate behavior under intermittent failures.

Sample Answer

Situation: A Python automation caused a retry storm that overwhelmed a downstream API. Below is a corrective plan covering root-cause fix, safeguards, and tests.

1) Identify & fix the bug- Reproduce locally with captured logs and timestamps to confirm simultaneous retries.- Likely cause: exponential backoff implemented incorrectly (e.g., backoff reset or zero delay) or shared state leading to synchronized retry loops.- Fix example: ensure per-request retry state, correct backoff math, and bounded sleeps.

python

import random, time
def backoff(attempt, base=0.5, cap=30):
    return min(cap, base * (2**attempt)) * (1 + random.uniform(-0.1, 0.1))
# use per-request attempt counter; never reset erroneously

2) Safeguards to prevent future retry storms- Max retries per request (e.g., 3-5) with clear failure path.- Global rate limiter shared across workers (token bucket or leaky bucket) to cap requests/s to downstream.- Circuit breaker per downstream service: open on error rate threshold or latency spike, with exponential retry of closed state.- Add jitter (randomized) to backoff to avoid thundering herd.- Centralized coordination: leader election or distributed rate limit via Redis to avoid synchronized clients.

3) Tests to validate behavior- Unit tests: validate backoff math, jitter bounds, max-retry enforcement, circuit state transitions.- Integration tests: run automation against a test stub that simulates intermittent 5xx and latency; assert rate limits and circuit breaker open as expected.- Chaos tests: in staging, simulate downstream outages/high latency and verify global rate limiter, CB behavior, and that incident alerts fire; run load tests to confirm no retry amplification.- SLO-based smoke: verify alerts when downstream error rate > threshold and that automation respects error budget.

4) Monitoring & ops- Add metrics: retries/sec, retry failures, rate-limiter rejections, circuit state, latency histograms.- Alerts: high retry rate, sudden spike in downstream 5xx, circuit breaker open > X minutes.- Run post-incident blameless RCA; deploy fix behind feature flag with canary rollout.

This plan fixes the bug, adds layered defenses (client-side limits, global coordination, circuit breaker), and validates via unit/integration/chaos tests plus monitoring and alerting.

HardTechnical

36 practiced

Implement a safe log-rotation utility in Python that rotates logs based on size and time, compresses rotated logs, and handles concurrent writers (multiple processes writing to the same log file). Discuss file-locking strategies (fcntl), rename vs copytruncate trade-offs, and how to ensure no log lines are lost during rotation.

Sample Answer

Approach: implement a RotatingLogger that checks size and time thresholds, uses O_APPEND for writers, acquires an exclusive fcntl lock when performing rotation, performs an atomic rename of the current file to a rotated filename, creates a new file, compresses the rotated file asynchronously, and preserves lines by ensuring writers always open in O_APPEND and rotation holds the lock so no interleaving/partial writes occur.

python

import os, time, gzip, shutil, fcntl, threading
from datetime import datetime

class RotatingLogger:
    def __init__(self, path, max_bytes=10*1024*1024, interval_secs=3600, backup_count=5):
        self.path = path
        self.max_bytes = max_bytes
        self.interval = interval_secs
        self.backup = backup_count
        self._roll_at = time.time() + self.interval
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)

    def _acquire_lock(self, fd):
        # exclusive, blocking
        fcntl.flock(fd, fcntl.LOCK_EX)

    def _release_lock(self, fd):
        fcntl.flock(fd, fcntl.LOCK_UN)

    def _do_rotate(self):
        # open file descriptor to lock rotation across processes
        with open(self.path, 'a+') as f:
            self._acquire_lock(f.fileno())
            try:
                f.flush()
                size = os.stat(self.path).st_size
                if size < self.max_bytes and time.time() < self._roll_at:
                    return
                # compute rotated name
                ts = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
                rotated = f"{self.path}.{ts}"
                # atomic rename (same filesystem)
                os.rename(self.path, rotated)
                # recreate the original file for writers
                open(self.path, 'a').close()
                # update next roll time
                self._roll_at = time.time() + self.interval
            finally:
                self._release_lock(f.fileno())

        # compress rotated file in background to avoid blocking writers
        threading.Thread(target=self._compress, args=(rotated,), daemon=True).start()
        self._prune_backups()

    def _compress(self, fname):
        gzname = fname + ".gz"
        with open(fname, 'rb') as src, gzip.open(gzname, 'wb') as dst:
            shutil.copyfileobj(src, dst)
        os.remove(fname)

    def _prune_backups(self):
        logs = sorted([p for p in os.listdir(os.path.dirname(self.path) or ".")
                       if p.startswith(os.path.basename(self.path)+".")])[-self.backup:]
        # remove older gz files beyond backup_count
        for p in sorted([p for p in os.listdir(os.path.dirname(self.path) or ".")
                         if p.startswith(os.path.basename(self.path)+".") and not p in logs]):
            try: os.remove(os.path.join(os.path.dirname(self.path) or ".", p))
            except: pass

    def write(self, data):
        # writers should open, lock-free, with O_APPEND so writes are atomic (POSIX)
        # but small writes are atomic only up to PIPE_BUF; for long lines use file lock briefly
        mode = os.O_WRONLY | os.O_APPEND | os.O_CREAT
        fd = os.open(self.path, mode, 0o644)
        try:
            # optional tiny lock to ensure atomic multi-write lines
            # fcntl.flock(fd, fcntl.LOCK_SH)
            os.write(fd, data.encode())
        finally:
            os.close(fd)
        # check rotate opportunistically
        if os.path.getsize(self.path) >= self.max_bytes or time.time() >= self._roll_at:
            self._do_rotate()

Key concepts and reasoning:- Use O_APPEND so writers append atomically at OS level; small writes are atomic. For multi-part lines, acquire an advisory lock (fcntl) briefly.- For rotation, use an exclusive flock on the file descriptor to serialize rotation across processes. Holding the lock during rename prevents race conditions.- Prefer rename over copytruncate: rename is atomic and preserves file inode for safety when writers reopen; copytruncate avoids rename but risks losing lines written between copy and truncate and is costlier for large files.- Compress asynchronously to avoid blocking rotation/writers.- On NFS, fcntl/flock semantics vary — consider a lock file on a shared, reliable metadata store (e.g., Redis or etcd) for cluster-wide locking.

Time/space:- Rotation time dominated by compression: O(n) I/O where n is file size; writes O(1).- Extra space: one copy during compression (gz file until original removed).

Edge cases:- NFS/cooperative locking differences — test environment-specific.- Crash between rename and recreation: we recreate immediately after rename while holding lock.- Large single-line writes > PIPE_BUF — ensure writers use single write calls or use application-level buffering with locks.- Filesystem boundaries: os.rename requires same FS.

Alternatives:- Use system tools like logrotate or use structured logging with centralized collectors (Fluentd/Logstash) to avoid complex local rotation.

HardTechnical

37 practiced

You must correlate traces produced by Python automation with logs and metrics across services. Explain how to propagate trace context across process boundaries (HTTP headers, message queues, subprocesses), provide code snippets for injecting/extracting W3C Trace Context using OpenTelemetry, and discuss sampling strategies and storage implications for high-volume tracing.

Sample Answer

Start by treating trace context as first-class telemetry: propagate the W3C Trace Context (traceparent + tracestate) across HTTP, MQ messages, and subprocesses so traces, logs and metrics can be correlated by trace_id/span_id.

HTTP (headers) — inject/extract with OpenTelemetry:

python

from opentelemetry import trace, propagators
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator
from opentelemetry.sdk.trace import TracerProvider

tracer = trace.get_tracer(__name__)
propagator = TraceContextTextMapPropagator()

# Inject into outgoing HTTP headers
with tracer.start_as_current_span("outgoing-call") as span:
    headers = {}
    propagator.inject(headers)
    # use headers with requests or httpx
    # requests.get(url, headers=headers)

# Extract incoming HTTP headers
def handle_request(request_headers):
    ctx = propagator.extract(carrier=request_headers)
    with tracer.start_as_current_span("server-handler", context=ctx):
        # handle request
        pass

Message queues — put context onto message attributes/headers:

python

# When producing
msg_headers = {}
propagator.inject(msg_headers)
producer.send(topic, value=payload, headers=list(msg_headers.items()))

# When consuming
def on_message(msg):
    headers = dict(msg.headers)
    ctx = propagator.extract(headers)
    with tracer.start_as_current_span("consumer-process", context=ctx):
        process(msg)

Subprocesses — pass traceparent in environment or CLI arg:

python

# Parent
env = os.environ.copy()
propagator.inject(env)  # inject into env keys like TRACEPARENT
subprocess.run(["python", "worker.py"], env=env)

# Child: extract from os.environ
ctx = propagator.extract(os.environ)
with tracer.start_as_current_span("child-work", context=ctx):
    ...

Correlating logs and metrics: capture trace_id/span_id into structured logs and metric labels. Configure your logging formatter to read current span:

python

from opentelemetry.trace import get_current_span
span = get_current_span()
trace_id = format(span.get_span_context().trace_id, '032x')
logger.info("step", extra={"trace_id": trace_id})

Sampling strategies and trade-offs:- Head-based sampling (probabilistic at SDK/agent): cheap, prevents overload but loses late errors. Use dynamic rates per service.- Tail-based sampling (store spans temporarily, decide after seeing error/latency): higher fidelity for anomalies with higher storage/compute cost (requires a tail sampler/collector).- Adaptive sampling: change rates based on traffic, SLOs, or error budget.Recommendation: sample a small percent (0.1–1%) of successful traces, keep 100% of errors/high-latency traces via tail/adaptive sampling. Tag sampled vs unsampled in logs to avoid confusion.

Storage implications:- Trace volume ≈ spans_per_trace * traces/sec. High-cardinality attributes and per-span logs massively increase storage. Limit attribute cardinality, use sampling and aggregation (span counts, histograms) for metrics.- Retention: store full traces shorter (days-weeks), derive long-term metrics/alerts from aggregates.- Cost controls: enforce max spans per trace, drop verbose attributes, use pre-ingest processors to redact PII and compress/span-merge.

Operational best practices:- Enforce a company-wide propagation standard (W3C) and instrument libraries centrally.- Add health dashboards: trace ingestion rate, sampled vs unsampled ratio, span size histograms.- Test propagation via integration tests that assert trace_id flows across HTTP, MQ, and subprocess boundaries.

This approach ensures traceable automation, consistent correlation with logs/metrics, and manageable storage/costs in high-volume environments.

Unlock Full Question Bank

Get access to hundreds of Python Automation and Monitoring interview questions and detailed answers.

Join thousands of developers preparing for their dream job.