FAANG-Standard Interview Preparation Guide for Junior SRE

# awk -f to_jsonl.awk
function jesc(s,    t) {
    # escape backslash and double-quote first
    gsub(/\\/,"\\\\",s)
    gsub(/\"/,"\\\"",s)
    # common control escapes
    gsub(/\r/,"\\r",s)
    gsub(/\n/,"\\n",s)
    gsub(/\t/,"\\t",s)
    return s
}

{
    # split first three space-separated fields; message is remainder
    timestamp = $1
    level     = $2
    module    = $3

    # build message from $4..$NF preserving spaces
    msg = ""
    for(i=4;i<=NF;i++){
        msg = msg (i==4 ? "" : " ") $i
    }

    msg = jesc(msg)
    printf("{\"timestamp\":\"%s\",\"level\":\"%s\",\"module\":\"%s\",\"message\":\"%s\"}\n", jesc(timestamp), jesc(level), jesc(module), msg)
}

#!/usr/bin/env bash
set -euo pipefail
IFS=$'\n\t'

# Step identity (helps logs)
STEP_NAME="${STEP_NAME:-deployment-step}"

# List required env vars here
required_env_vars=(DEPLOY_TARGET AUTH_TOKEN)

# Validate required environment variables (fail fast if missing)
for var in "${required_env_vars[@]}"; do
  if [[ -z "${!var-}" ]]; then
    echo "ERROR: Required environment variable '$var' is not set. Aborting ${STEP_NAME}." >&2
    exit 2
  fi
done

# Cleanup function (idempotent)
cleanup() {
  # Put idempotent teardown here, safe to run on success or failure
  echo "Cleanup: tearing down temporary resources for ${STEP_NAME}..."
  # e.g., rm -f /tmp/my-deploy-*
}

# Error handler: captures exit code and the command that failed
on_error() {
  local exit_code=${1:-$?}
  local failed_cmd=${2:-'unknown'}
  echo "ERROR: ${STEP_NAME} failed. Command: '${failed_cmd}' Exit code: ${exit_code}" >&2
  # perform cleanup actions
  cleanup
  # propagate failure
  exit "${exit_code}"
}

# Trap ERR to capture the failing command and its exit code
trap 'on_error $? "$BASH_COMMAND"' ERR

# Optional: ensure cleanup runs on normal exit too (no-op if on_error already exited)
trap 'cleanup' EXIT

# -------------------------
# Actual deployment steps
# -------------------------
echo "Starting ${STEP_NAME}..."
# Example commands; any failing command triggers ERR trap
# e.g., build, push, or apply
# build step
# make build

# push artifacts
# ./push-artifacts.sh

# deploy
# kubectl apply -f k8s/ --namespace="${DEPLOY_TARGET}"

echo "${STEP_NAME} completed successfully."

# python pseudocode

import time, json
from enum import Enum

class Status(Enum): PENDING = "PENDING"; DONE = "DONE"; FAILED = "FAILED"; COMPENSATING = "COMPENSATING"; COMPENSATED = "COMPENSATED"

class Step:
    def __init__(self, name):
        self.name = name
    def execute(self, ctx): raise NotImplementedError
    def compensate(self, ctx): raise NotImplementedError

# simple durable store interface (wrap Redis/RDB)
class Store:
    def save(self, saga_id, state_dict): ...
    def load(self, saga_id): ...

class Orchestrator:
    def __init__(self, steps, store, max_retries=5):
        self.steps = steps
        self.store = store
        self.max_retries = max_retries

    def start(self, saga_id, initial_ctx):
        state = self.store.load(saga_id) or self._init_state(saga_id, initial_ctx)
        try:
            self._run(state)
        finally:
            self.store.save(saga_id, state)

    def _init_state(self, saga_id, ctx):
        state = {
            "saga_id": saga_id,
            "ctx": ctx,
            "steps": [{ "name": s.name, "status": Status.PENDING.value, "retries":0 } for s in self.steps]
        }
        self.store.save(saga_id, state)
        return state

    def _run(self, state):
        for idx, step_meta in enumerate(state["steps"]):
            step = self.steps[idx]
            if step_meta["status"] == Status.DONE.value: continue
            if step_meta["status"] in (Status.FAILED.value, Status.PENDING.value):
                success = self._execute_with_retries(step, state, idx)
                if not success:
                    # fatal -> compensate completed steps in reverse
                    self._compensate_completed(state, idx-1)
                    raise Exception(f"Fatal step failure: {step.name}")
        # all done
        self.store.save(state["saga_id"], state)

    def _execute_with_retries(self, step, state, idx):
        meta = state["steps"][idx]
        while meta["retries"] < self.max_retries:
            try:
                # idempotency: step.execute should be safe to call multiple times using state["ctx"]
                result = step.execute(state["ctx"])
                meta["status"] = Status.DONE.value
                meta["result"] = result
                self.store.save(state["saga_id"], state)
                return True
            except TransientError:
                meta["retries"] += 1
                self.store.save(state["saga_id"], state)
                time.sleep(2 ** meta["retries"])
                continue
            except FatalError:
                meta["status"] = Status.FAILED.value
                self.store.save(state["saga_id"], state)
                return False
        # exceeded retries -> treat as fatal
        meta["status"] = Status.FAILED.value
        self.store.save(state["saga_id"], state)
        return False

    def _compensate_completed(self, state, last_completed_idx):
        for idx in range(last_completed_idx, -1, -1):
            meta = state["steps"][idx]
            if meta["status"] != Status.DONE.value: continue
            step = self.steps[idx]
            try:
                meta["status"] = Status.COMPENSATING.value
                self.store.save(state["saga_id"], state)
                step.compensate(state["ctx"])
                meta["status"] = Status.COMPENSATED.value
                self.store.save(state["saga_id"], state)
            except Exception:
                # log and continue – compensations should be retriable and idempotent
                meta["status"] = Status.FAILED.value
                self.store.save(state["saga_id"], state)

import json
import logging
from typing import Dict, Optional

logger = logging.getLogger(__name__)

class ConfigAccessError(PermissionError):
    """Raised when config file exists but isn't readable due to permissions."""
    pass

def load_json_config(path: str, defaults: Dict, correlation_id: Optional[str] = None) -> Dict:
    """
    Load JSON config from path and merge with defaults.
    File values override defaults.
    - FileNotFoundError: return defaults
    - PermissionError: raise ConfigAccessError
    - json.JSONDecodeError: log structured error and return defaults
    """
    extra = {"correlation_id": correlation_id}
    try:
        with open(path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        logger.info("Config file not found; using defaults.", extra=extra)
        return dict(defaults)
    except PermissionError as e:
        logger.error("Permission denied reading config file.", exc_info=True, extra=extra)
        raise ConfigAccessError(str(e)) from e
    except json.JSONDecodeError as e:
        # structured log containing file path, error position, and optional correlation id
        logger.error(
            "Malformed JSON in config file.",
            extra={**extra, "path": path, "msg": e.msg, "lineno": e.lineno, "colno": e.colno},
            exc_info=False,
        )
        return dict(defaults)

    if not isinstance(data, dict):
        logger.error("Config JSON must be an object/dict; ignoring file and using defaults.",
                     extra={**extra, "path": path})
        return dict(defaults)

    merged = dict(defaults)
    merged.update(data)  # file overrides defaults
    return merged