Graceful shutdown and connection draining.

apiVersion: apps/v1
kind: Deployment
metadata:
  name: reports-api
spec:
  replicas: 3
  selector:
    matchLabels:
      app: reports-api
  template:
    metadata:
      labels:
        app: reports-api
    spec:
      # 5s preStop allowance + up to 25s active-request drain + 10s margin.
      terminationGracePeriodSeconds: 40
      containers:
        - name: api
          image: reports-api:reviewed-release
          ports:
            - name: http
              containerPort: 8080
          lifecycle:
            preStop:
              httpGet:
                path: /drain
                port: http
          readinessProbe:
            httpGet:
              path: /readyz
              port: http
            periodSeconds: 3
            timeoutSeconds: 1
            failureThreshold: 1

import { createServer } from 'node:http';
import { setTimeout as delay } from 'node:timers/promises';

// Shorter values are useful for local drills; defaults match the Deployment.
function milliseconds(name: string, fallback: number) {
	const value = Number(process.env[name] ?? fallback);
	if (!Number.isSafeInteger(value) || value < 1 || value > 120_000)
		throw new Error(`Invalid ${name}`);
	return value;
}
const drainMs = milliseconds('DRAIN_MS', 25_000);
const hookMs = milliseconds('HOOK_MS', 5_000);
const workMs = milliseconds('WORK_MS', 3_000);
const port = Number(process.env.PORT ?? 8080);
let draining = false;
let hookWait: Promise<void> | undefined;
let stopping = false;
const activeWork = new Set<AbortController>();
const log = (event: string) => console.log(new Date().toISOString(), event);

const server = createServer(async (request, response) => {
	const path = new URL(request.url ?? '/', 'http://local').pathname;
	if (path === '/readyz') {
		response.writeHead(draining ? 503 : 200).end();
		return;
	}
	if (path === '/drain') {
		draining = true;
		log('drain requested');
		// Repeat hooks share the first wait; they never restart the allowance.
		// A hook timer must not keep the process alive after forced socket closure.
		hookWait ??= delay(hookMs, undefined, { ref: false });
		await hookWait;
		if (!response.destroyed) response.writeHead(200).end('draining');
		return;
	}
	if (path !== '/work') {
		response.writeHead(404).end();
		return;
	}
	if (draining) {
		response.writeHead(503, { connection: 'close' }).end('instance is draining');
		return;
	}
	const controller = new AbortController();
	activeWork.add(controller);
	response.on('close', () => controller.abort());
	log('work accepted');
	try {
		await delay(workMs, undefined, { signal: controller.signal });
		if (!response.destroyed)
			response.writeHead(200, draining ? { connection: 'close' } : {}).end('request complete');
		log('work complete');
	} catch (error) {
		if (!(error instanceof Error) || error.name !== 'AbortError') throw error;
		log('work cancelled');
	} finally {
		activeWork.delete(controller);
	}
});

server.listen(port, () => log(`listening ${port}`));

function shutdown() {
	if (stopping) return;
	stopping = true;
	draining = true; // Signals must work even when the preStop hook did not run.
	log('signal received; stop admission');
	const forceStop = setTimeout(() => {
		log('drain deadline expired');
		for (const controller of activeWork) controller.abort();
		server.closeAllConnections();
	}, drainMs);
	server.close(() => {
		clearTimeout(forceStop);
		log('server closed');
		process.exitCode = 0;
	});
}
process.on('SIGTERM', shutdown);
process.on('SIGINT', shutdown);

package main

import (
	"context"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strconv"
	"sync"
	"sync/atomic"
	"syscall"
	"time"
)

func main() {
	log.SetFlags(log.LstdFlags | log.Lmicroseconds)
	var draining atomic.Bool
	var hookOnce sync.Once
	hookDone := make(chan struct{})
	readMS := func(name string, fallback int) time.Duration {
		value := fallback
		if raw := os.Getenv(name); raw != "" {
			parsed, err := strconv.Atoi(raw)
			if err != nil || parsed < 1 || parsed > 120000 {
				log.Fatalf("invalid %s", name)
			}
			value = parsed
		}
		return time.Duration(value) * time.Millisecond
	}
	hookDuration := readMS("HOOK_MS", 5000)
	drainDuration := readMS("DRAIN_MS", 25000)
	workDuration := readMS("WORK_MS", 3000)
	port := os.Getenv("PORT")
	if port == "" {
		port = "8080"
	}
	mux := http.NewServeMux()
	mux.HandleFunc("GET /readyz", func(w http.ResponseWriter, r *http.Request) {
		if draining.Load() {
			http.Error(w, "draining", http.StatusServiceUnavailable)
			return
		}
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("GET /drain", func(w http.ResponseWriter, r *http.Request) {
		draining.Store(true)
		hookOnce.Do(func() {
			log.Print("drain requested")
			go func() { time.Sleep(hookDuration); close(hookDone) }()
		})
		<-hookDone
		w.WriteHeader(http.StatusOK)
	})
	mux.HandleFunc("GET /work", func(w http.ResponseWriter, r *http.Request) {
		if draining.Load() {
			http.Error(w, "instance is draining", http.StatusServiceUnavailable)
			return
		}
		log.Print("work accepted")
		select {
		case <-time.After(workDuration):
			_, _ = w.Write([]byte("request complete"))
			log.Print("work complete")
		case <-r.Context().Done():
			log.Print("work cancelled")
			return
		}
	})

	server := &http.Server{Addr: ":" + port, Handler: mux}
	serveErr := make(chan error, 1)
	go func() { serveErr <- server.ListenAndServe() }()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGTERM, os.Interrupt)
	select {
	case <-signals:
	case err := <-serveErr:
		if err != nil && err != http.ErrServerClosed {
			log.Fatal(err)
		}
		return
	}
	draining.Store(true)
	log.Print("signal received; stop admission")

	ctx, cancel := context.WithTimeout(context.Background(), drainDuration)
	defer cancel()
	if err := server.Shutdown(ctx); err != nil {
		log.Printf("grace period expired; close remaining connections: %v", err)
		_ = server.Close()
	}
	if err := <-serveErr; err != nil && err != http.ErrServerClosed {
		log.Fatal(err)
	}
	log.Print("server closed")
}