Kubernetes Pod Lifecycle Guide — From Pending to Running

import time
import kubernetes
from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

def watch_pod_lifecycle(namespace: str, pod_name: str, timeout: int = 120):
    start = time.time()
    w = kubernetes.watch.Watch()
    for event in w.stream(
        v1.list_namespaced_pod,
        namespace,
        field_selector=f"metadata.name={pod_name}",
        timeout_seconds=timeout
    ):
        pod = event["object"]
        phase = pod.status.phase
        conditions = {
            c.type: c.status for c in (pod.status.conditions or [])
        }
        container_statuses = pod.status.container_statuses or []
        container_states = {}
        for cs in container_statuses:
            state = cs.state
            if state.running:
                container_states[cs.name] = "running"
            elif state.waiting:
                container_states[cs.name] = f"waiting({state.waiting.reason})"
            elif state.terminated:
                container_states[cs.name] = f"terminated({state.terminated.exit_code})"

        elapsed = time.time() - start
        print(f"[{elapsed:5.1f}s] Phase: {phase:>10} | "
              f"Conditions: {conditions} | "
              f"Containers: {container_states}")

        if phase in ("Succeeded", "Failed"):
            break
    w.stop()

# Run a simple pod and watch its lifecycle
pod_manifest = {
    "apiVersion": "v1",
    "kind": "Pod",
    "metadata": {"name": "lifecycle-demo"},
    "spec": {
        "restartPolicy": "Never",
        "containers": [{
            "name": "main",
            "image": "alpine",
            "command": ["sh", "-c", "echo 'Starting...'; sleep 3; echo 'Done!'"]
        }]
    }
}

v1.create_namespaced_pod("default", pod_manifest)
watch_pod_lifecycle("default", "lifecycle-demo")
v1.delete_namespaced_pod("default", "lifecycle-demo")

# init-container-pod.yaml
apiVersion: v1
kind: Pod
metadata:
  name: init-demo
spec:
  initContainers:
    - name: init-download
      image: busybox:1.36
      command:
        - wget
        - "-O"
        - "/data/signatures.db"
        - "https://cdn.dodatech.com/signatures/latest.db"
      volumeMounts:
        - name: data
          mountPath: /data
    - name: init-check
      image: alpine:3.20
      command:
        - sh
        - "-c"
        - "test -s /data/signatures.db && echo 'Signatures verified'"
      volumeMounts:
        - name: data
          mountPath: /data
  containers:
    - name: main
      image: alpine:3.20
      command: ["sh", "-c", "cat /data/signatures.db | head -5"]
      volumeMounts:
        - name: data
          mountPath: /data
  volumes:
    - name: data
      emptyDir: {}

kubectl apply -f init-container-pod.yaml
kubectl get pod init-demo -w

apiVersion: v1
kind: Pod
metadata:
  name: probe-demo
spec:
  containers:
    - name: app
      image: python:3.12-slim
      command:
        - python
        - "-c"
        - |
          import http.server, time
          class H(http.server.BaseHTTPRequestHandler):
              ready = False
              def do_GET(self):
                  if self.path == "/healthz":
                      if H.ready:
                          self.send_response(200)
                      else:
                          self.send_response(503)
                  else:
                      self.send_response(200)
                  self.end_headers()
          time.sleep(5)  # Simulate slow startup
          H.ready = True
          http.server.HTTPServer(("0.0.0.0", 8080), H).serve_forever()
      ports:
        - containerPort: 8080
      startupProbe:
        httpGet:
          path: /healthz
          port: 8080
        initialDelaySeconds: 0
        periodSeconds: 2
        failureThreshold: 30
      livenessProbe:
        httpGet:
          path: /healthz
          port: 8080
        periodSeconds: 10
        failureThreshold: 3
      readinessProbe:
        httpGet:
          path: /healthz
          port: 8080
        periodSeconds: 5
        initialDelaySeconds: 0

class ContainerStateMachine:
    def __init__(self):
        self.state = "waiting"
        self.reason = "ContainerCreating"
        self.restart_count = 0
        self.exit_code = None

    def transition(self):
        transitions = {
            "waiting": self._waiting_transition,
            "running": self._running_transition,
            "terminated": self._terminated_transition,
        }
        return transitions.get(self.state, lambda: self.state)()

    def _waiting_transition(self):
        print(f"Waiting: {self.reason}")
        if self.reason == "ContainerCreating":
            self.state = "running"
            self.reason = None
        elif self.reason == "CrashLoopBackOff":
            self.restart_count += 1
            self.state = "waiting"
            self.reason = f"CrashLoopBackOff (backoff {2**self.restart_count}s)"

    def _running_transition(self):
        print("Running")
        import random
        if random.random() < 0.2:  # 20% chance of crash
            self.state = "terminated"
            self.exit_code = 1
            self.reason = "Error"

    def _terminated_transition(self):
        print(f"Terminated: exit code {self.exit_code}")
        if self.exit_code != 0 and self.restart_count < 3:
            self.state = "waiting"
            self.reason = "CrashLoopBackOff"
        else:
            print("Pod failed permanently")

fsm = ContainerStateMachine()
for i in range(10):
    print(f"Step {i}: ", end="")
    fsm.transition()
    if fsm.state == "terminated" and fsm.restart_count >= 3:
        break

apiVersion: v1
kind: Pod
metadata:
  name: graceful-shutdown
spec:
  terminationGracePeriodSeconds: 60
  containers:
    - name: app
      image: python:3.12-slim
      command:
        - python
        - "-c"
        - |
          import signal, time, sys
          def handler(signum, frame):
              print("Received SIGTERM, draining connections...")
              time.sleep(3)  # Complete in-flight requests
              print("Shutdown complete")
              sys.exit(0)
          signal.signal(signal.SIGTERM, handler)
          print("Server started")
          while True:
              time.sleep(1)
      lifecycle:
        preStop:
          exec:
            command: ["sh", "-c", "echo 'PreStop: removing from LB'; sleep 2"]

import time
import random

class Pod:
    def __init__(self, name: str):
        self.name = name
        self.phase = "Pending"
        self.ready = False
        self.init_complete = False
        self.startup_ok = False
        self.restart_count = 0
        self.deleted = False

    def tick(self):
        if self.deleted and self.phase != "Terminating":
            self.phase = "Terminating"
            return

        if self.phase == "Pending" and self.init_complete:
            self.phase = "Running"

        if self.phase == "Running":
            if random.random() < 0.05:
                self.restart_count += 1
                if self.restart_count > 3:
                    self.phase = "CrashLoopBackOff"
                else:
                    self.phase = "Running"

    def show(self):
        status = f"{self.phase:>15}"
        if self.phase == "Running":
            status += f" (restarts: {self.restart_count})"
        return f"Pod {self.name:>10}: {status} | Ready: {str(self.ready):>5}"

pods = [Pod(f"web-{i}") for i in range(3)]
for _ in range(20):
    for p in pods:
        p.tick()
    for p in pods:
        print(p.show())
    print()
    time.sleep(0.3)