Shell Scripting and Automation Questions

Clarifying assumptions

• Instances periodically push health reports (heartbeats + boolean/score)
• Services = groups of N instances; quorum rule parameterized per service (M of N)
• Reports may be delayed, lost, or arrive out-of-order

High-level components

• Collectors (agents or push endpoints) → validate and timestamp incoming reports
• Ingest queue (Kafka) → buffers bursts and tolerates spikes
• Stateful Aggregator (stream processor) → computes service health, hysteresis, emits events
• Time-series DB (Prometheus/Influx) + OLAP store (ClickHouse) → real-time metrics + historical views
• Rules engine / config store → per-service M, N, thresholds, window sizes
• API / UI → real-time status, history, alerts
• Alerting & downstream (PagerDuty, runbooks)

Data model

• InstanceReport: { instance_id, service_id, reported_at, status (UP/DEGRADED/DOWN), score, metadata }
• ServiceState: { service_id, evaluated_at, computed_state, confidence, contributing_instances[], reason }
• StateHistory (time-series): timestamp, service_id, state, confidence

Core algorithm (streaming, per service)

1. Windowing: maintain a sliding evaluation window W (e.g., 30s) of latest reports per instance; drop older than TTL
2. Quorum eval: count healthy reports H among active instances A. If H >= M → candidate HEALTHY else UNHEALTHY/DEGRADED
3. Tolerate missing: treat instance as UNKNOWN if no report within TTL; quorum counts only active instances or use configured missing-instance policy
4. Confidence score: compute confidence = H / max(M, A) and factor in report recency and instance weights
5. Hysteresis (debounce & anti-flap):
   • Temporal hysteresis: require candidate state to persist for T_stable (e.g., 60s) before flipping
   • Count hysteresis: require K consecutive evaluations (or sliding-window majority) agreeing
   • Weighted dampening: on flips within short interval, increase required T_stable exponentially (backoff)
6. Emit state change only when debounced; always store instantaneous candidates to TSDB for visibility

Minimize flapping

• Use both time and count hysteresis
• Add jitter to evaluation scheduling to avoid synchronized oscillation
• Implement backoff: each flip doubles T_stable up to cap
• Persistent incidents: if confidence low but unstable, mark as PARTIAL/DEGRADED instead of toggling

Real-time & historical views

• Real-time: materialized current ServiceState in fast cache (Redis) and push via websocket/events
• Historical: append-to TSDB; provide rollups (1m/5m/1h) and query APIs
• Audit trail: store contributing instance reports for each state transition for forensic analysis

Scalability & reliability

• Partition aggregation by service_id across consumer group
• Checkpoint offsets for exactly-once or at-least-once processing with idempotent state writes
• Configurable per-service parameters to trade sensitivity vs stability

Trade-offs

• Longer hysteresis reduces false positives but delays detection
• Treating missing instances as unhealthy is safer but noisy for flaky networks
• Weighted voting lets critical instances dominate; increases config complexity

This design balances quorum correctness, resilience to missing data, anti-flap stability, and provides both real-time and historical observability suitable for cloud-scale operations.

Approach

• Compute exponential backoff = base * 2**attempt, cap to cap. If jitter apply full jitter by returning a uniform random in [0, capped_backoff]. To make jitter deterministic given a seed, provide a module-level RNG you can seed (helper shown). Validate inputs.

Code

import random

_rng = random.Random()  # module-level RNG; deterministic if seeded via set_backoff_seed

def set_backoff_seed(seed: int) -> None:
    """Seed the backoff RNG for deterministic jitter across calls."""
    _rng.seed(seed)

def get_backoff(attempt: int, base: float = 0.5, cap: float = 60.0, jitter: bool = True) -> float:
    """
    Return sleep time in seconds for a given retry attempt.

    - attempt: non-negative integer (0 => first retry uses base)
    - base: base delay in seconds
    - cap: maximum delay in seconds
    - jitter: if True, use full jitter (uniform(0, backoff))
    """
    if attempt < 0:
        raise ValueError("attempt must be non-negative")
    if base <= 0 or cap <= 0:
        raise ValueError("base and cap must be positive")

    backoff = base * (2 ** attempt)
    backoff = min(backoff, cap)

    if not jitter:
        return backoff
    return _rng.uniform(0.0, backoff)

Examples

• Deterministic jitter:
  • set_backoff_seed(42); get_backoff(3) -> same value across runs for seed 42.
• No jitter:
  • get_backoff(3, jitter=False) -> deterministic pure exponential capped.

Complexity

• Time: O(1)
• Space: O(1)

Assumptions & Notes

• attempt 0 returns base (or jitter in [0, base]).
• Full jitter chosen per spec (random 0..backoff). Alternative: equal jitter, decorrelated jitter exist.
• Deterministic jitter: caller must call set_backoff_seed(seed) once before calls to make behavior reproducible.

Runbook — Immediate Steps (On-call)

1. Assess & Triage

   • Confirm alerts (latency/error rate dashboards, SLOs).
   • Identify scope: affected services, regions, zones, downstream dependencies.
   • Capture timestamps, logs, and recent deploys.

2. Contain

   • Enable circuit breaker (prevent new requests to failing components).
   • Shift traffic away from affected region to healthy region(s) at low volume (10-20%) for canary.
   • Notify on-call rotation, incident commander, stakeholders.

3. Mitigate

   • Gradually increase traffic to healthy region if metrics improve.
   • If errors persist, maintain traffic shift and investigate root cause.

4. Recover & Learn

   • Once healthy, restore normal routing, disable circuit breaker, run postmortem.

Automated Mitigation Script (outline, Python using cloud SDKs)

#!/usr/bin/env python3
# deploy_circuit_breaker_and_shift.py
# Requires: cloud SDK, kubernetes client, or LB API credentials

import time
from cloudsdk import load_client  # placeholder
from monitoring import get_metric

CLIENT = load_client()

def safety_checks():
    # Ensure runbook owner approval flag file or 2FA token exists
    assert CLIENT.get_flag("oncall_auto_enable") == True
    # Ensure healthy region has capacity
    healthy_capacity = CLIENT.check_capacity("region-b")
    assert healthy_capacity > 0.2

def enable_circuit_breaker(service):
    # Set CB to reject new requests / return 503 or route to fallback
    CLIENT.update_service(service, {"circuit_breaker": {"maxConnections":0, "enabled":True}})

def shift_traffic(from_region, to_region, percent):
    # Modify traffic split on global LB / service mesh
    CLIENT.update_traffic_split(service="svc", from=from_region, to=to_region, percent=percent)

def rollback():
    CLIENT.update_service("svc", {"circuit_breaker":{"enabled":False}})
    CLIENT.update_traffic_split(service="svc", restore_default=True)

def main():
    try:
        safety_checks()
        enable_circuit_breaker("backend-pool-affected")
        for p in [10,25,50,100]:
            shift_traffic("region-a","region-b", p)
            time.sleep(60)  # wait for metrics
            if get_metric("error_rate", "svc") > 0.05:
                rollback()
                raise Exception("Error rate high after shift — rolled back")
        # final verification
    except Exception as e:
        rollback()
        raise

if __name__ == "__main__":
    main()

Safety checks

• Require on-call flag/approval token.
• Verify target region capacity and service health.
• Rate-limit percentage changes and wait for stabilization windows.

Rollback path

• Script includes immediate rollback() to restore traffic and disable circuit breaker.
• Manual abort allowed; escalation to SRE lead if rollback fails.

Why this works

• Circuit breaker isolates failing components quickly; progressive traffic shifting minimizes blast radius while validating healthy region capacity; safety checks and automated rollback reduce human error.

Approach (brief)
Monitor filesystems via df -P. For each mount track recent state in a persistent store (SQLite). Alert only when usage exceeds threshold for N consecutive checks and ensure cooldown per filesystem to deduplicate alerts.

Code (Python)

#!/usr/bin/env python3
import subprocess, sqlite3, time, json, os
THRESH=85          # percent
CONSECUTIVE=3      # require 3 consecutive breaches
COOLDOWN=3600      # seconds
DB='/var/lib/diskmon/state.db'
CHECK_INTERVAL=60

# initialize DB
conn=sqlite3.connect(DB)
c=conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS fs_state(
  filesystem TEXT PRIMARY KEY, consecutive INT, last_alert INT)''')
conn.commit()

def get_fs_usage():
  out=subprocess.check_output(['df','-P']).decode()
  lines=out.strip().splitlines()[1:]
  res=[]
  for ln in lines:
    cols=ln.split()
    fs=cols[5]; pct=int(cols[4].rstrip('%'))
    res.append((fs,pct))
  return res

def send_alert(fs, pct):
  # replace with pager/email integration
  print(f'ALERT {fs} usage {pct}%')

while True:
  now=int(time.time())
  for fs,pct in get_fs_usage():
    row=c.execute('SELECT consecutive,last_alert FROM fs_state WHERE filesystem=?',(fs,)).fetchone()
    consecutive,last_alert=(row if row else (0,0))
    if pct>=THRESH:
      consecutive+=1
      if consecutive>=CONSECUTIVE and now - last_alert >= COOLDOWN:
        send_alert(fs,pct)
        last_alert=now
        consecutive=0   # reset after alert to avoid repeated alerts
    else:
      consecutive=0
    c.execute('REPLACE INTO fs_state(filesystem,consecutive,last_alert) VALUES (?,?,?)',(fs,consecutive,last_alert))
  conn.commit()
  time.sleep(CHECK_INTERVAL)

Persistence & Robustness

• SQLite stores consecutive counts and last alert timestamps across restarts.
• Requiring CONSECUTIVE breaches handles transient spikes.
• Cooldown prevents alert spam; resetting consecutive after alert avoids duplicate alerts.
• For large fleets, replace SQLite with etcd/Redis and use leader election if running multiple instances.

Edge cases

• Handle removed mounts by pruning DB entries.
• Permission issues for DB path.
• Integrate with metric/alerting backends and add exponential backoff on alert failures.

Approach (brief)

• Use dateutil.parser for flexible parsing, with explicit format fallbacks for ambiguous inputs. Accept optional reference datetime to infer missing year and a dayfirst flag to resolve MM/DD vs DD/MM. Normalize to ISO8601 with timezone (UTC by default or preserved if present).

Code

from datetime import datetime
from dateutil import parser, tz
from typing import Optional

def parse_log_ts(line: str,
                 dayfirst: bool = False,
                 default_tz: str = "UTC",
                 ref_dt: Optional[datetime] = None) -> str:
    """
    Parse timestamp from a log line and return ISO8601 with timezone.
    - dayfirst: interpret ambiguous dates with day first.
    - default_tz: fallback timezone name.
    - ref_dt: reference datetime used to infer missing year (defaults to now).
    """
    ref_dt = ref_dt or datetime.now(tz=tz.gettz(default_tz))
    # extract token(s) likely to be timestamp (simple heuristic)
    tok_candidates = line.split()
    for tok in tok_candidates:
        try:
            # allow parser to use dayfirst and default year from ref_dt if missing
            dt = parser.parse(tok, default=ref_dt, dayfirst=dayfirst)
            # ensure timezone
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=tz.gettz(default_tz))
            return dt.astimezone(tz.UTC).isoformat()
        except (ValueError, OverflowError):
            continue
    # fallback: try parsing whole line
    try:
        dt = parser.parse(line, default=ref_dt, dayfirst=dayfirst)
        if dt.tzinfo is None:
            dt = dt.replace(tzinfo=tz.gettz(default_tz))
        return dt.astimezone(tz.UTC).isoformat()
    except Exception as e:
        raise ValueError("No parsable timestamp found") from e

Key points & reasoning

• dateutil handles many formats; default year is provided via default parameter so missing year is inferred from ref_dt.
• dayfirst resolves ambiguous month/day ordering.
• Return normalized ISO8601 in UTC for consistency across systems.

Complexity

• Time: O(n) where n = number of tokens checked (small constant).
• Space: O(1).

Edge cases

• Multiple timestamps in a line — current function returns first parsed; could be extended to return all.
• Locale-specific month names or rare formats may need custom regex/formatters.
• Logs with no year and crossing year boundaries: pass an appropriate ref_dt (e.g., log file creation time).