How to Flush Metrics on Graceful Shutdown in Python

Easy Python 3.9+ Aug 9, 2026 Observability & SRE 14 views 0 copies

atexit metrics graceful-shutdown monitoring stdlib

Python code

28 lines

Python 3.9+

import atexit
import time
import random


class MetricsCollector:
    def __init__(self):
        self._metrics = []
        atexit.register(self.flush)

    def record(self, name, value):
        self._metrics.append((name, value, time.time()))

    def flush(self):
        print(f"Flushing {len(self._metrics)} metrics")
        for name, value, ts in self._metrics:
            print(f"  {name}: {value} @ {ts:.2f}")
        self._metrics.clear()


if __name__ == "__main__":
    collector = MetricsCollector()
    collector.record("request_count", 42)
    collector.record("latency_ms", random.randint(50, 200))

    print("Simulating running process...")
    time.sleep(0.5)
    print("Exiting...")

Output

stdout

Simulating running process...
Exiting...
Flushing 2 metrics
  request_count: 42 @ 1712345678.90
  latency_ms: 153 @ 1712345678.90

How it works

The atexit.register call wires the flush method to run when the interpreter shuts down normally. Each record call stores a timestamped tuple, and flush prints the buffered metrics then clears the list. This pattern ensures no data is lost on a clean sys.exit() or when the script reaches the end of its if __name__ == "__main__" block. The mock time values come from time.time() since the example doesn't inject a clock, so the exact numbers will vary on every run.

Common mistakes

Using `sys.exit()` or raising exceptions, which can skip atexit handlers in some edge cases
Forgetting that `atexit` doesn't fire on SIGKILL or hard crashes
Not clearing the metrics list after flushing, causing duplicate flushes on repeated calls
Registering the method bound to a local instance that goes out of scope before exit

Variations

Use `signal.signal(signal.SIGTERM, handler)` to catch termination signals explicitly
Flush to a file or remote endpoint instead of stdout in real collectors
Wrap cleanup in a `try/finally` block for more explicit control flow

Real-world use cases

Batching application metrics in memory and pushing them to a monitoring agent at process exit to avoid tiny network calls per request.
Flushing a local buffer of log events to the central aggregator when a worker pod shuts down during a rolling deploy.
Ensuring a report generator writes out its final accumulated counters before a scheduled batch job terminates on a timer.

How to Flush Metrics on Graceful Shutdown in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Keep learning

Tutorials

Quizzes