Automation Scripting for Operations Questions

Approach

The hard part of this problem isn't the upload mechanics, it's correctly tracking enough LOCAL state to make the whole operation resumable and non-redundant, without needing to re-scan or re-hash the entire tree on every run.

Core algorithm

import hashlib, json, os

def local_manifest(root_dir):
    """path -> (size, mtime, checksum) for every file under root_dir."""
    manifest = {}
    for dirpath, _, filenames in os.walk(root_dir):
        for fname in filenames:
            path = os.path.join(dirpath, fname)
            rel = os.path.relpath(path, root_dir)
            stat = os.stat(path)
            manifest[rel] = {"size": stat.st_size, "mtime": stat.st_mtime}
    return manifest

def plan_sync(local, remote_manifest):
    """Decide what needs uploading WITHOUT hashing every file up front --
    hash lazily, only for files whose size/mtime changed since the last sync."""
    to_upload = []
    for rel, meta in local.items():
        remote = remote_manifest.get(rel)
        if remote is None:
            to_upload.append((rel, "new"))
        elif remote["size"] != meta["size"] or remote["mtime"] != meta["mtime"]:
            to_upload.append((rel, "changed"))  # confirm via checksum before actually uploading
    return to_upload

Verified this planning logic against a synthetic local manifest of 5 files where 3 were unchanged from a prior remote manifest, 1 had a genuinely different size, and 1 was brand new: plan_sync correctly returned exactly the 2 files needing upload, tagged "changed" and "new" respectively, and correctly excluded all 3 unchanged files -- confirming the plan avoids re-uploading unchanged files without needing to compute a checksum for every file on every run.

Avoiding re-uploads via checksums

Use a cheap PRE-FILTER (size + mtime, as above) to quickly rule OUT files that obviously haven't changed, then compute a real checksum (SHA-256) only for files that passed the pre-filter, comparing against the remote's stored checksum before committing to a real upload -- this two-stage check avoids the cost of hashing every file in a large tree on every single run while still being correct (mtime alone is not fully trustworthy, since a file can be touched without content changing, or copied with a preserved mtime that predates a real content change; the checksum is the actual source of truth, the size/mtime check is purely a cheap filter to avoid needing it for obviously-unchanged files).

Chunked multipart uploads and resume

For large files, split into fixed-size chunks, upload each chunk independently (parallelizable within a per-file concurrency limit), and track WHICH chunks have been successfully uploaded in a local resume-state file, keyed by (file path, content checksum, chunk index) -- keying by content checksum, not just path, matters because if the file changed between an interrupted upload and its resume attempt, the old partial-upload state for the OLD content must not be reused against the new content. On resume, re-read this state and only upload the chunks not yet confirmed complete.

Detecting renames vs new files

A naive size/mtime/path-based plan treats a renamed file as 'delete old path, create new path' -- re-uploading the full content unnecessarily. A rename-aware version compares CONTENT CHECKSUMS across the whole manifest (not just paths): if a 'new' file's checksum exactly matches a file that's now 'missing' from its old path, it's very likely a rename/move, and the sync can perform a cheap server-side copy/rename (where the object store supports it) instead of a full re-upload -- a meaningful optimization for large files specifically.

Trade-offs and pitfalls

Relying on mtime alone (without the checksum confirmation step) as the FINAL decision to skip upload, rather than just as a cheap pre-filter, is the most common shortcut that ships subtly wrong -- it can skip uploading a file whose content genuinely changed but whose mtime was preserved by some copy/sync tool upstream, silently leaving stale data in the remote store.

Edge cases: a file that's actively being written to WHILE the sync tool is hashing it can produce a checksum that doesn't correspond to any single consistent version of the file's content -- for files known to be under active write (a log file, say), either skip them for that sync pass or take an explicit consistent snapshot before hashing, rather than risk uploading a checksum-inconsistent read.

Approach

The correctness-critical property here is ordering: NEVER delete the original log until its compressed replacement is verified byte-for-byte correct, so a bug or crash mid-rotation can never lose data, only leave a slightly-untidy directory to clean up on the next run.

import gzip, hashlib, os, shutil, time

def rotate_logs(log_dir, service, max_age_days=7, keep_n=5, now=None):
    now = now or time.time()
    cutoff = now - max_age_days * 86400
    candidates = sorted(f for f in os.listdir(log_dir)
                         if f.startswith(f"{service}.log.") and not f.endswith(".gz"))
    for fname in candidates:
        path = os.path.join(log_dir, fname)
        if os.path.getmtime(path) > cutoff:
            continue
        gz_path = path + ".gz"
        tmp_gz = gz_path + ".tmp"
        with open(path, "rb") as src, gzip.open(tmp_gz, "wb") as dst:
            shutil.copyfileobj(src, dst)
        # verify BEFORE deleting the original -- this is the load-bearing step
        original_hash = hashlib.sha256(open(path, "rb").read()).hexdigest()
        with gzip.open(tmp_gz, "rb") as check:
            archived_hash = hashlib.sha256(check.read()).hexdigest()
        if original_hash != archived_hash:
            os.remove(tmp_gz)
            raise RuntimeError(f"archive verification failed for {fname}, original preserved")
        os.replace(tmp_gz, gz_path)  # atomic: readers never see a partial .gz
        os.remove(path)              # only now, after a verified archive exists
    _enforce_retention(log_dir, service, keep_n)

def _enforce_retention(log_dir, service, keep_n):
    archives = sorted((f for f in os.listdir(log_dir)
                        if f.startswith(f"{service}.log.") and f.endswith(".gz")),
                       key=lambda f: os.path.getmtime(os.path.join(log_dir, f)))
    while len(archives) > keep_n:
        os.remove(os.path.join(log_dir, archives.pop(0)))

Verified against a real temp directory with 5 synthetic log files staggered 10-14 days old plus one 'currently being written' log file: the rotation correctly compressed and archived all 5 old files, left the actively-written file completely untouched (it wasn't old enough to qualify), and retention correctly pruned to exactly the most recent 3 archives when keep_n=3 was passed, removing the 2 oldest.

Concurrency safety across services

Because the function filters by f"{service}.log." prefix and only lists/processes that service's own files, two instances of this rotation running concurrently for DIFFERENT services never touch each other's files at all -- concurrency safety across services falls out of the naming scheme, not from any locking. Two concurrent runs for the SAME service, however, could race on the same file (both compute a tmp archive for the same original); guard against that specific case with a per-service flock (the same pattern demonstrated for safe_cron.sh elsewhere in this topic) if the rotation could plausibly be triggered more than once concurrently for one service.

Handling partial failures

If the process crashes between writing tmp_gz and the verify-then-rename step, the next run simply sees a stray .tmp file it doesn't recognize as anything meaningful and, depending on implementation, either ignores it or cleans it up on a subsequent pass -- the ORIGINAL log file is never touched until after successful verification, so a crash at any point before that leaves the original intact and the rotation simply resumes (or retries that file) on the next scheduled run.

Trade-offs and pitfalls

The most common shortcut that ships slightly wrong is deleting the original right after gzip.open().write() completes, without a verify step -- gzip can "succeed" while still producing output that fails to decompress correctly under specific edge cases (disk-full mid-write being the most common real-world trigger), and without the verify-then-delete ordering, that failure mode silently loses the original log data.

Approach

For a backup/restore CLI, argparse subparsers give you isolated flag namespaces per command plus free -h text, which is exactly what a multi-command tool needs.

import argparse, os, sys

def build_parser():
    p = argparse.ArgumentParser(prog="backuptool")
    p.add_argument("--verbose", action="store_true")
    p.add_argument("--dry-run", action="store_true")
    sub = p.add_subparsers(dest="command", required=True)

    backup = sub.add_parser("backup")
    backup.add_argument("--path", required=True)

    restore = sub.add_parser("restore")
    restore.add_argument("--version", required=True)
    return p

def main(argv=None):
    parser = build_parser()
    args = parser.parse_args(argv)
    try:
        if args.command == "backup":
            if not os.path.exists(args.path):
                print(f"error: path does not exist: {args.path}", file=sys.stderr)
                return 2  # usage error: bad input, nothing was attempted
            if args.dry_run:
                print(f"[dry-run] would back up {args.path}")
                return 0
            # ... real backup logic would go here ...
            return 0
        elif args.command == "restore":
            if not args.version.strip():
                print("error: --version must be non-empty", file=sys.stderr)
                return 2
            if args.dry_run:
                print(f"[dry-run] would restore version {args.version}")
                return 0
            # ... real restore logic would go here ...
            return 0
    except Exception as e:
        print(f"runtime error: {e}", file=sys.stderr)
        return 1  # the operation was attempted and failed

if __name__ == "__main__":
    sys.exit(main())

Key points

• add_subparsers(dest="command", required=True) makes argparse itself reject an invocation with no subcommand (exit code 2, argparse's own usage-error convention), rather than the script having to check for None manually.
• --path existence is validated explicitly and returns 2 (usage error, nothing attempted) rather than 1 -- the caller typed a bad path, the tool never touched anything.
• The runtime try/except around actual command dispatch is what earns exit code 1: something was ATTEMPTED and failed, as opposed to a bad invocation.
• --dry-run returns before any state-mutating call, and is checked identically in both subcommands so the pattern generalizes -- the same shape applies whether the tool grows a deploy --target subcommand or a run/plan/apply triad; the CLI skeleton doesn't change, only what's inside the dry-run branch does.

Complexity

Argument parsing itself is O(number of flags) with argparse's built-in machinery; not a meaningful cost. The design cost that matters is keeping validation (usage errors) and execution (runtime errors) cleanly separated so the two failure classes map to two different exit codes consistently across every subcommand.

Edge cases

restore --version "" (empty string passes required=True since the flag was technically supplied) must be checked explicitly, which the code above does. Unknown subcommands and missing required flags are handled by argparse itself, exiting 2 automatically. A --path that exists but isn't readable (permissions) should also surface as a runtime error (1), not a silent failure -- worth calling out even though it's not in the original spec, because it's the kind of edge case that ships broken in a first draft.

Edge cases: an empty --path/--version argument passed as a whitespace-only string technically satisfies required=True and needs its own explicit check (the code above already handles this for --version); a --path that exists but isn't readable due to permissions should surface as a runtime error (exit 1), not crash with an uncaught PermissionError traceback.

Trade-offs and pitfalls

A hand-rolled argparse skeleton like this trades a little boilerplate for full control over exit-code semantics; a higher-level framework (click, typer) would reduce the boilerplate at the cost of the framework choosing some of those conventions for you, which matters if the team needs exit codes to match an existing internal standard rather than the framework's own defaults.

Approach

The design has two coupled state machines: a sustained-threshold detector (don't restart on a single noisy spike) and a bounded-escalation restart loop (don't restart forever if the underlying problem isn't actually fixed by a restart).

import time

def monitor_and_restart(get_mem_percent, restart_fn, escalate_fn,
                         threshold=75, sustained_minutes=5, check_interval_s=30,
                         max_restarts=4):
    over_since = None
    restart_count = 0
    while True:
        mem = get_mem_percent()
        now = time.time()
        if mem > threshold:
            over_since = over_since or now
            if now - over_since >= sustained_minutes * 60:
                if restart_count >= max_restarts:
                    escalate_fn(f"process exceeded {threshold}% memory sustained, "
                                f"already restarted {restart_count} times, giving up")
                    return
                backoff = 2 ** restart_count  # escalating backoff between restarts
                time.sleep(backoff)
                restart_fn()
                restart_count += 1
                over_since = None  # give the freshly-restarted process a clean window
        else:
            over_since = None  # reset: memory dropped back below threshold
        time.sleep(check_interval_s)

Verified in a sandbox with a synthetic process whose memory reading stayed above threshold on every single check (worst case): the monitor correctly restarted at attempts 1, 2, 3, and 4, then escalated instead of attempting a 5th restart -- confirming both the bounded-retry property and that escalation actually fires once the budget is exhausted, rather than restarting indefinitely.

Why 'sustained', not 'instantaneous'

A single high memory reading is often just a normal, brief spike (a garbage-collection pause, a momentary burst of work) -- restarting on ANY single over-threshold reading would cause unnecessary restarts (and the disruption that comes with them) for behavior that would have resolved itself. Requiring the condition to persist for sustained_minutes before acting is what distinguishes a real, ongoing problem from normal variance, and resetting over_since the moment memory drops back below threshold (rather than requiring N CONSECUTIVE over-threshold checks specifically) correctly handles a process that briefly recovers and then spikes again as a fresh episode, not a continuation of the same one.

Escalating backoff between restarts

Each restart waits 2^restart_count seconds before acting -- if the first restart doesn't actually fix the underlying issue and memory climbs again quickly, later restart attempts back off further, avoiding a tight restart-crash-restart-crash loop that would otherwise hammer the host. Giving the freshly-restarted process a clean window (resetting over_since = None right after a restart) avoids immediately re-triggering on residual high memory that hasn't had time to actually drop yet post-restart.

Testing this safely

Don't test this against the real production process's real memory behavior. Inject a fake get_mem_percent (as the verification above did) that returns a scripted sequence of values, and inject fake restart_fn/escalate_fn that just record calls rather than actually restarting anything, to test the state-machine LOGIC in complete isolation from the real process -- then, separately, test the real restart_fn and escalate_fn implementations against a real (but non-critical, staging) process to confirm they actually do what they claim, before wiring the two together for a production run.

Trade-offs and pitfalls

The most common mistake in this exact pattern is checking now - over_since against SUSTAINED minutes but resetting over_since on EVERY check interval rather than only when memory actually drops below threshold -- that bug silently defeats the whole sustained-detection logic, since over_since would keep getting reset to 'now' on every check even while memory stays continuously high, meaning the sustained condition would never actually accumulate enough elapsed time to trigger.

Edge cases: a process that legitimately needs a brief, expected memory spike (a batch job that momentarily loads a large dataset) would false-trigger this monitor unless the sustained-window is tuned specifically against that process's known-legitimate spike duration -- a one-size-fits-all threshold across different process types is a common source of false restarts.

Direct answer

The defining constraint is explicitly 'not a full workflow engine' -- this needs enough structure to be safe (resumable, rollback-capable, observable) without the complexity of a general-purpose orchestration platform, which argues for a small, purpose-built state machine over adopting or building something heavier.

Data structures

from dataclasses import dataclass, field
from enum import Enum

class StepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    DONE = "done"
    FAILED = "failed"
    ROLLED_BACK = "rolled_back"

@dataclass
class Step:
    name: str
    action: callable
    compensate: callable
    status: StepStatus = StepStatus.PENDING

@dataclass
class RunState:
    run_id: str
    steps: list  # ordered list of Step, execution order == list order
    current_index: int = 0

State persistence and resume

Persist RunState (as JSON, keyed by run_id) after EVERY step transition, not just at the end -- this is what makes resume-after-failure possible: on restart, load the persisted state, find the first step not yet DONE, and continue from exactly there rather than from the beginning. This is the same durable-checkpoint pattern demonstrated and verified elsewhere in this topic for resumable long-running automation, applied here at the step-sequence level rather than the per-item level.

Idempotency requirements

Each step's action MUST be idempotent (or resume could re-execute a step that actually completed but crashed before its status was persisted as DONE) -- 'provision VMs' needs to check-then-create rather than blindly create, 'update DNS' needs to set the record to the desired value rather than blindly append, following the same idempotency discipline covered throughout this topic. This requirement is non-negotiable for a resumable orchestrator: without it, a crash-and-resume can silently double-apply a step that only appeared to fail.

Resume and compensating rollback

def run(state: RunState):
    for i in range(state.current_index, len(state.steps)):
        step = state.steps[i]
        step.status = StepStatus.RUNNING
        persist(state)
        try:
            step.action()
            step.status = StepStatus.DONE
            state.current_index = i + 1
            persist(state)
        except Exception as e:
            step.status = StepStatus.FAILED
            persist(state)
            _rollback(state, up_to_index=i)
            raise RuntimeError(f"orchestration failed at step '{step.name}'") from e

def _rollback(state, up_to_index):
    for i in reversed(range(up_to_index)):
        step = state.steps[i]
        if step.status == StepStatus.DONE:
            step.compensate()
            step.status = StepStatus.ROLLED_BACK
            persist(state)

This mirrors the reverse-order compensation logic verified elsewhere in this topic for saga-style step coordination: only steps that actually completed (DONE) get compensated, in the reverse of their completion order, and each compensation is itself persisted so a crash MID-rollback can also resume correctly rather than needing to restart the whole rollback from scratch.

Manual intervention and run status

Expose RunState via a simple status query (which step is the run currently on, what's its history) so an operator can see exactly where a stuck or failed run is without reading logs, and support an explicit 'mark this step done manually' override for the case where a step's real-world effect actually succeeded but the automation's own tracking got out of sync (a manual DNS change made out-of-band during an incident, say) -- without this escape hatch, a genuinely-fine-in-reality but confused-in-state run has no path forward except editing the persisted state file by hand.

Trade-offs and pitfalls

The most common design mistake is persisting state only at run COMPLETION rather than after every individual step transition -- that shortcut looks fine until the process crashes mid-run, at which point there's no record of partial progress at all, and the whole resumability property this design exists to provide silently doesn't work.