Forem: James Lee

Full Observability in Istio: Metrics with Prometheus/Grafana + Distributed Tracing with Jaeger

James Lee — Fri, 22 May 2026 10:16:22 +0000

Observability is the ability to understand the internal state of a system from its external outputs. In a microservices architecture — where a single API call may fan out to dozens of services and hundreds of DB queries — observability is not optional. It's survival.

Istio provides three pillars of observability out of the box:

Pillar	Tool	Best For
Metrics	Prometheus + Grafana	Trend analysis, QPS, latency, error rate
Distributed Tracing	Jaeger	Root cause analysis of slow/failed requests
Logging	ELK Stack	Detailed per-request log investigation

This article covers the first two in depth.

Part 1: Metrics with Prometheus + Grafana

Why Prometheus Over StatsD or InfluxDB?

Three metrics systems are commonly used in modern infrastructure:

StatsD + Graphite    →  UDP push model, lightweight, limited ecosystem
InfluxDB + Telegraf  →  Rich system-level plugins, custom app metrics need manual work
Prometheus           →  Pull model, native K8s/Consul service discovery, full alerting stack

Prometheus wins in cloud-native environments for two reasons:

Pull model — Prometheus scrapes targets over HTTP. This simplifies client code and makes debugging trivial (just curl the /metrics endpoint).
Native service discovery — No static IP lists. Prometheus discovers targets from Kubernetes, Consul, or DNS automatically.

Prometheus Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Prometheus Server                        │
│   ┌──────────────┐    ┌──────────────┐    ┌─────────────┐  │
│   │   Scraper    │    │    TSDB       │    │ Rule Engine │  │
│   │  (pull HTTP) │    │  (storage)   │    │  (alerts)   │  │
│   └──────┬───────┘    └──────────────┘    └──────┬──────┘  │
└──────────┼────────────────────────────────────────┼─────────┘
           │ scrape /metrics                        │ fire alerts
           ▼                                        ▼
┌─────────────────────┐                  ┌──────────────────────┐
│  Service Discovery  │                  │    Alertmanager      │
│  (K8s / Consul/DNS) │                  │  (dedup, group,      │
└─────────────────────┘                  │   notify: DingTalk / │
                                         │   WeCom / Email)     │
┌─────────────────────┐                  └──────────────────────┘
│    PushGateway      │ ← for short-lived jobs (batch, PHP scripts)
└─────────────────────┘
┌─────────────────────┐
│     Exporters       │ ← Node Exporter, Blackbox Exporter, etc.
└─────────────────────┘

PushGateway exists specifically for short-lived processes (batch jobs, PHP scripts) that won't survive long enough to be scraped.

Prometheus Data Model

Every metric is a combination of a name + labels:

negri_http_request_total{client="serviceA", code="200", exported_service="serviceB", path="/ping"}

negri_http_request_total — metric name (describes what it measures)
{client, code, exported_service, path} — labels (dimensions for filtering/aggregation)

The Four Metric Types

Type	Computed By	Best For	Example
Counter	Server-side	QPS, total requests (monotonically increasing)	`negri_http_request_total`
Gauge	Client-side	Instantaneous values, event flags (circuit breaker open/closed)	`degrade_events{event="eventBreakerOpenStatus"}`
Histogram	Server-side	Latency percentiles (p90/p95/p99), response size distribution	`negri_http_response_time_us_bucket{le="0.5"}`
Summary	Client-side	Precise percentiles, but less flexible at query time	Similar to Histogram

Counter vs Gauge: If your value only goes up → use Counter (cheaper, server-side). If it goes up and down → use Gauge.

Histogram vs Summary: Histogram is more flexible (percentiles calculated at query time in Grafana). Summary is more precise but percentiles are fixed at instrumentation time.

Grafana Dashboard: What to Look At

Istio ships with pre-built Grafana dashboards. Here's how to navigate them effectively.

Services Dashboard

Key filter parameters:

Parameter	What It Means
`Service`	The Kubernetes service name (e.g. `details.default.svc.cluster.local`)
`Client Workload Namespace`	The K8s namespace where the client lives
`Client Workload`	The upstream caller — e.g. `productpage-v1` calling `details`
`Service Workload`	The specific version of the service — e.g. `reviews-v1`, `reviews-v2`, `reviews-v3`

Why Client Workload matters: Traditional microservice SDKs often struggle to reliably inject the caller's service name into metrics. Developers might hardcode the wrong name or copy-paste from another service. Istio eliminates this entirely — the control plane injects the caller identity automatically via the sidecar. No human error possible.

Key panels in the dashboard:

QPS (client-side vs server-side) — client-side is more accurate as it includes network latency
Success Rate — percentage of non-5xx responses
Latency — p50/p90/p99 breakdown

Workload Dashboard

The Workload dashboard adds a critical dimension: inbound vs outbound traffic.

reviews-v3 Workload
├── INBOUND  ← traffic FROM productpage-v1
└── OUTBOUND ← traffic TO ratings.default.svc.cluster.local

This dual-direction view is invaluable for root cause analysis. When reviews-v3 is slow, you can immediately see:

Is it slow because incoming traffic is high? (upstream pressure)
Or is it slow because outgoing calls to ratings are slow? (downstream dependency)

Part 2: Distributed Tracing with Jaeger

The Three Pillars of Observability — Compared

Metrics  →  "Something is wrong with reviews-v3 (p99 latency spiked to 2s)"
Tracing  →  "The spike is caused by the ratings service call at step 4 of this specific request"
Logging  →  "Here's the exact SQL query and stack trace that caused it"

Each tool answers a different question. Tracing is the bridge between "something is wrong" and "here's exactly why."

How Distributed Tracing Works (Dapper Model)

Tracing originates from Google's Dapper paper. The core concepts:

TraceId: 5f1db306ef459b2f  (unique per request, generated at the gateway)
│
├── Span A  (SpanId: aaa, ParentSpanId: null)  ← root span (productpage)
│     │
│     ├── Span B  (SpanId: bbb, ParentSpanId: aaa)  ← details service call
│     │
│     └── Span C  (SpanId: ccc, ParentSpanId: aaa)  ← reviews service call
│           │
│           └── Span D  (SpanId: ddd, ParentSpanId: ccc)  ← ratings service call

Trace context is propagated via HTTP headers:

Header	Purpose
`x-request-id`	Unique request ID for log correlation
`x-b3-traceid`	64-bit global trace identifier
`x-b3-spanid`	Current span position in the trace tree
`x-b3-parentspanid`	Parent span (absent = root node)
`x-b3-sampled`	Sampling flag (`1` = record this trace)

A Real Trace Record

{
  "traceID": "5f1db306ef459b2f",
  "spanID": "5f1db306ef459b2f",
  "parentSpanID": "0",
  "operationName": "/ping",
  "duration": 2065,
  "startTime": 1609241265147010,
  "process": {
    "serviceName": "negri.sidecarserverlistener.myapp",
    "tags": { "hostname": "MacBook-Pro-3.local", "ip": "192.168.1.88" }
  },
  "tags": {
    "http.method": "GET",
    "http.status_code": "200",
    "http.url": "/ping",
    "peer.address": "http://127.0.0.1:8888",
    "span.kind": "server"
  }
}

Jaeger Architecture

Application Code
     │ UDP
     ▼
┌──────────────┐     ┌──────────────────┐     ┌──────────────┐
│ jaeger-client│────▶│  jaeger-agent    │────▶│jaeger-       │
│ (SDK)        │     │  (per-host       │     │collector     │
└──────────────┘     │   daemon)        │     │(validate +   │
                     └──────────────────┘     │ process)     │
                                              └──────┬───────┘
                                                     │
                                                     ▼
                                             ┌──────────────┐
                                             │  jaeger-db   │
                                             │ (Cassandra / │
                                             │  Elasticsearch│
                                             └──────┬───────┘
                                                    │
                                                    ▼
                                             ┌──────────────┐
                                             │ jaeger-query │
                                             │ (UI + API)   │
                                             └──────────────┘

Why jaeger-agent? Same philosophy as Istio's sidecar — it offloads service discovery and routing complexity from the client SDK. The app just sends UDP to localhost and forgets about it.

The One Thing Your Code Must Do

Envoy automatically generates spans and propagates B3 headers to upstream services. But — it cannot read the incoming headers and pass them downstream on your behalf. Your application code must forward the trace headers manually.

Here's the pattern from the Bookinfo sample app (Python):

@app.route('/productpage')
@trace()
def front():
    headers = getForwardHeaders(request)  # extract + forward trace headers
    details = getProductDetails(product_id, headers)  # pass them downstream

def getForwardHeaders(request):
    headers = {}

    # B3 headers — extracted automatically via OpenTracing library
    span = get_current_span()
    carrier = {}
    tracer.inject(span_context=span.context, format=Format.HTTP_HEADERS, carrier=carrier)
    headers.update(carrier)

    # Other headers that must be forwarded manually
    incoming_headers = [
        'x-request-id',
        'x-ot-span-context',
        'traceparent',
        'tracestate',
        'x-cloud-trace-context',
        'grpc-trace-bin',
        'user-agent',
    ]
    for h in incoming_headers:
        val = request.headers.get(h)
        if val is not None:
            headers[h] = val

    return headers

Key insight: Envoy handles span creation and injection automatically. Your app only needs to forward the headers it receives. This is a much lighter instrumentation burden than traditional APM agents.

What Jaeger Shows You

Trace List View — All traces for a service, sortable by duration. Immediately surfaces the slowest requests.

Trace Detail View — Full waterfall of every span in a single request:

productpage  ──────────────────────────────  450ms
  details      ───  45ms
  reviews        ──────────────────  380ms
    ratings        ────────────  310ms  ← bottleneck identified

System Architecture View — Auto-generated service dependency graph derived from trace data. No manual diagram maintenance needed.

Summary: Metrics vs Tracing — When to Use Which

Scenario	Use
"Is the service healthy right now?"	Metrics (Grafana dashboard)
"Has latency been trending up this week?"	Metrics (Prometheus query)
"Why did this specific request take 3 seconds?"	Tracing (Jaeger)
"Which service is the bottleneck in my call chain?"	Tracing (Jaeger waterfall)
"What exact error did service X return at 14:32?"	Logging (ELK)

The power of Istio is that you get all of this without modifying your application — except for the minimal header forwarding shown above. The sidecar handles metric collection, span generation, and header injection automatically.

💻 Explore the full implementation:
github.com/muzinan123/servicemesh

Traffic Management in Istio: Circuit Breaking & Rate Limiting with DestinationRule

James Lee — Fri, 22 May 2026 10:15:40 +0000

One of Istio's most powerful features is the ability to configure circuit breaking and rate limiting declaratively — without touching a single line of application code. Everything is controlled through a single CRD: DestinationRule.

In this article, I'll break down exactly how connectionPool (rate limiting) and outlierDetection (circuit breaking) work, with real YAML examples.

The Entry Point: TrafficPolicy in DestinationRule

Both circuit breaking and rate limiting are configured under the trafficPolicy field of a DestinationRule:

apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: my-service-dr
spec:
  host: my-service
  trafficPolicy:
    connectionPool:    # ← Rate limiting config
      tcp:
        maxConnections: 100
      http:
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
    outlierDetection:  # ← Circuit breaking config
      consecutiveErrors: 5
      interval: 10s
      baseEjectionTime: 30s
      maxEjectionPercent: 10

Part 1: Rate Limiting with `connectionPool`

connectionPool controls how many concurrent connections and requests are allowed to reach an upstream service. It has two sub-sections: tcp and http.

TCP Settings

Field	Description	Default
`maxConnections`	Max number of HTTP1/TCP connections to the destination	unlimited
`connectTimeout`	TCP connection timeout	10s

⚠️ Important: maxConnections only limits HTTP/1.1 and TCP connections. It does not affect HTTP/2, because HTTP/2 multiplexes all requests over a single connection.

HTTP Settings

Field	Description	Default
`http1MaxPendingRequests`	Max queued HTTP/1.1 requests waiting to be processed	1024
`http2MaxRequests`	Max concurrent HTTP/2 requests	1024
`maxRequestsPerConnection`	How many requests can reuse one TCP connection. Set to `1` to disable keepalive	unlimited

Full connectionPool Example

trafficPolicy:
  connectionPool:
    tcp:
      maxConnections: 100
      connectTimeout: 3s
    http:
      http1MaxPendingRequests: 100
      http2MaxRequests: 1000
      maxRequestsPerConnection: 10

What this does: Limits the service to 100 concurrent TCP connections, queues at most 100 pending HTTP/1.1 requests, and allows up to 1000 concurrent HTTP/2 requests. Each TCP connection can serve up to 10 requests before being recycled.

Part 2: Circuit Breaking with `outlierDetection`

outlierDetection implements the circuit breaker pattern at the Envoy level. When a service instance starts returning errors, Istio automatically ejects it from the load balancing pool.

How It Works

Request → Envoy → Load Balancer Pool
                   ├── Instance A (healthy) ✅
                   ├── Instance B (healthy) ✅
                   └── Instance C (5xx × 5) ❌ → Ejected for 30s

When Instance C is ejected, traffic is only routed to A and B. After baseEjectionTime, Instance C gets a chance to re-enter the pool.

Configuration Fields

Field	Description	Default
`consecutiveErrors`	Number of consecutive 5xx errors (502/503/504) before ejection	5
`interval`	Time window for counting errors	10s
`baseEjectionTime`	Minimum ejection duration. Actual time = `baseEjectionTime × ejection count`	30s
`maxEjectionPercent`	Max % of instances that can be ejected at once	10%
`minHealthPercent`	If healthy instances drop below this %, circuit breaking is disabled entirely	50%

Full outlierDetection Example

trafficPolicy:
  outlierDetection:
    consecutiveErrors: 5
    interval: 10s
    baseEjectionTime: 30s
    maxEjectionPercent: 10
    minHealthPercent: 50

Key Design Decisions Worth Understanding

1. Progressive Ejection Time

The actual ejection time is not fixed — it grows with each ejection:

1st ejection: 30s × 1 = 30s
2nd ejection: 30s × 2 = 60s
3rd ejection: 30s × 3 = 90s

This is intentional: a repeatedly failing instance gets longer and longer timeouts, giving it more time to recover.

2. The Safety Net: `maxEjectionPercent` + `minHealthPercent`

These two fields work together to prevent a cascade failure from taking down your entire service:

maxEjectionPercent: 10 — Even if 50 instances are all returning 5xx errors, only 10% will be ejected at any one time.
minHealthPercent: 50 — If healthy instances drop below 50%, Istio disables outlier detection entirely and allows all instances (including unhealthy ones) to receive traffic.

Why disable circuit breaking when too many instances are unhealthy?
Because at that point, the problem is likely systemic (e.g., a bad deployment, downstream dependency failure). Ejecting more instances would only make things worse. It's better to let all instances handle traffic and surface the real error.

Putting It All Together

Here's a production-ready DestinationRule combining both features:

apiVersion: networking.istio.io/v1alpha3
kind: DestinationRule
metadata:
  name: reviews-dr
spec:
  host: reviews
  trafficPolicy:
    connectionPool:
      tcp:
        maxConnections: 100
        connectTimeout: 3s
      http:
        http1MaxPendingRequests: 100
        http2MaxRequests: 1000
        maxRequestsPerConnection: 10
    outlierDetection:
      consecutiveErrors: 5
      interval: 10s
      baseEjectionTime: 30s
      maxEjectionPercent: 10
      minHealthPercent: 50

Summary

Feature	Config Field	Mechanism
Rate Limiting	`connectionPool.tcp`	Limit TCP connections
Rate Limiting	`connectionPool.http`	Limit concurrent HTTP requests
Circuit Breaking	`outlierDetection.consecutiveErrors`	Eject on N consecutive errors
Safety Net	`outlierDetection.maxEjectionPercent`	Cap max ejected instances
Safety Net	`outlierDetection.minHealthPercent`	Disable CB when cluster is too unhealthy

The beauty of Istio's approach: all of this happens at the proxy layer. Your application code doesn't need to implement Hystrix, Resilience4j, or any circuit breaker library. The infrastructure handles it.

💻 Explore the full implementation:
github.com/muzinan123/servicemesh

📖 Next in this series: Full Observability in Istio: Metrics + Distributed Tracing

Istio & Envoy Service Mesh Architecture: How the Control Plane and Data Plane Work Together

James Lee — Fri, 22 May 2026 10:14:51 +0000

If you've worked with Istio, you've probably heard terms like "control plane," "data plane," "Pilot," and "Envoy sidecar." But how do they actually fit together? In this article, I'll walk through the full architecture — from how Pilot discovers services to how a client request is processed inside Envoy.

Part 1: How Pilot and Envoy Work Together

The Istio control plane component Pilot is responsible for bridging Kubernetes and Envoy. Here's the three-step flow:

┌─────────────────────────────────────────────────────┐
│                    API Server                       │
└──────────────────────┬──────────────────────────────┘
                       │ 1. Service discovery
                       │    (Services + Endpoints via client-go Informer)
                       │ 2. Fetch Istio CRDs
                       │    (VirtualService, DestinationRule, etc.)
                       ▼
┌─────────────────────────────────────────────────────┐
│                   Istio Pilot                       │
│                                                     │
│  Convert: Endpoints + governance policies           │
│        → Envoy-compatible xDS format                │
└──────────────────────┬──────────────────────────────┘
                       │ 3. Push via xDS
                       ▼
┌─────────────────────────────────────────────────────┐
│                  Envoy Sidecar                      │
│         (Listener → Route → Cluster → Endpoint)    │
└─────────────────────────────────────────────────────┘

Step 1 — Service Discovery:
Pilot uses the client-go Informer to watch the Kubernetes API Server, collecting all Service and Endpoints resources across the cluster.

Step 2 — Policy Discovery:
Pilot also watches for user-defined Istio CRD objects: VirtualService, DestinationRule, Gateway, etc. These define the traffic governance policies.

Step 3 — xDS Push:
Pilot converts the combined service topology and governance policies into Envoy's native xDS format, then pushes the config to each Envoy sidecar via gRPC streaming.

Part 2: The Four Core Resources in Envoy

Once config is pushed to Envoy, how does a client request actually get processed? It flows through four core resource types:

Client Request
     │
     ▼
┌──────────┐     ┌──────────┐     ┌──────────┐     ┌────────────┐
│ Listener │────▶│  Route   │────▶│ Cluster  │──LB▶│  Endpoint  │
│  (LDS)   │     │  (RDS)   │     │  (CDS)   │     │   (EDS)    │
└──────────┘     └──────────┘     └──────────┘     └────────────┘
     │                                                     │
L3/L4 Filters                                     Upstream Service

1. Listener (LDS)

A Listener is a port that Envoy opens to accept incoming connections. Multiple Listeners are fully isolated from each other. Beyond port binding, each Listener is configured with L3/L4 filters.

Config is managed via LDS (Listener Discovery Service).

2. Cluster (CDS + EDS)

A Cluster is the abstraction for an upstream service. Every upstream service maps to exactly one Cluster. Cluster config includes:

Connection timeout
Connection pool settings
Load balancing policy
Health check config
Endpoint list

Config is managed via CDS (Cluster Discovery Service) + EDS (Endpoint Discovery Service).

3. Route (RDS)

Route is the bridge between Listeners and Clusters:

Listener receives the request
Route decides which Cluster to forward it to
Route also handles: Virtual Host definitions, HTTP header manipulation (add/remove/modify), timeout & retry policies

Config is managed via RDS (Route Discovery Service).

4. Filter (Envoy Filter)

Filters are Envoy's plugin mechanism — they allow powerful extensions without modifying source code. There are three layers:

Filter Layer	Scope	Examples
Listener Filter	Pre-routing, L3/L4	TLS Inspector, HTTP Inspector, Original Destination
Network Filter	L3/L4 connection handling	Dubbo proxy, Kafka filter, MySQL proxy, Redis proxy, HTTP Connection Manager
HTTP Filter	L7 HTTP traffic	Route matching, Health check, Lua, CSRF, VHDS

Key insight: The HTTP Connection Manager (HCM) is itself a special Network Filter that manages all HTTP Filters. This layered design is what gives Envoy the ability to support virtually any protocol — and even perform protocol conversion.

Full Architecture: Putting It All Together

Here's how the complete picture looks — from API Server down to upstream services:

┌──────────────────────────────────────────────────────────────────┐
│                         API Server                               │
└────────────────────────────┬─────────────────────────────────────┘
                             │ Services, Endpoints, Istio CRDs
                             ▼
┌──────────────────────────────────────────────────────────────────┐
│                        Istio Pilot                               │
│  ① Service discovery    ② Fetch CRDs    ③ Convert & push xDS    │
└────────────────────────────┬─────────────────────────────────────┘
                             │ xDS (gRPC streaming)
                             ▼
┌──────────────────────────────────────────────────────────────────┐
│                          Envoy                                   │
│                                                                  │
│  Listener0 ┐                  Cluster0 ┐         Endpoint0      │
│  Listener1 ├──Route──────────▶Cluster1 ├──LB────▶Endpoint1      │
│  Listener2 ┘                  Cluster2 ┘         Endpoint2      │
│                                                  Endpoint3      │
│                                                                  │
│  [Listener Filter] → [Network Filter] → [HTTP Filter]           │
└──────────────────────────────────────────────────────────────────┘
                             │
                             ▼
                   Upstream Application Services

Why This Architecture Matters

This design gives Istio several powerful properties:

Zero-touch config updates — Envoy never needs to restart. All config changes are pushed via xDS streaming.
Protocol agnostic — Through Network Filters, Envoy natively supports HTTP, gRPC, Dubbo, Kafka, MySQL, Redis, and more.
Policy separation — Traffic governance rules (VirtualService, DestinationRule) live in the control plane. The data plane (Envoy) just executes them.
Extensibility — Custom Envoy Filters allow teams to add any behavior (auth, rate limiting, tracing) without touching application code.

Summary

Component	Role
Pilot	Discovers services + policies, converts to xDS, pushes to Envoy
Envoy Listener	Accepts client connections on configured ports
Envoy Route	Decides which Cluster handles each request
Envoy Cluster	Abstracts upstream services with LB + health check
Envoy Endpoint	The actual upstream instance IP:port
Envoy Filter	Plugin mechanism for L4/L7 traffic manipulation

Understanding this architecture is the foundation for everything else in Istio — circuit breaking, canary releases, observability. They all build on top of these primitives.

💻 Explore the full implementation:
github.com/muzinan123/servicemesh

📖 Next in this series: Traffic Management in Istio: Circuit Breaking & Rate Limiting

xDS Protocol Deep Dive: The Universal Control Plane API Behind Envoy and Istio

James Lee — Fri, 22 May 2026 10:13:12 +0000

When we talk about Istio or Envoy, we often hear terms like "dynamic configuration" and "control plane push." But what exactly is the underlying protocol that makes all of this work? The answer is xDS.

In this article, I'll do a deep dive into the xDS protocol — what it is, how it works, and why it matters beyond just Service Mesh.

What is xDS?

xDS is a family of discovery service APIs that allow a data plane proxy (like Envoy) to dynamically fetch configuration from a management server — without any restart or file reload.

The "x" in xDS is a wildcard. It covers:

API	Full Name	Purpose
LDS	Listener Discovery Service	Dynamically manage listeners (ports)
RDS	Route Discovery Service	Dynamically manage routing rules
CDS	Cluster Discovery Service	Dynamically manage upstream clusters
EDS	Endpoint Discovery Service	Dynamically manage cluster endpoints
SDS	Secret Discovery Service	Dynamically manage TLS certificates

Key insight: xDS is not just for Service Mesh. gRPC also uses xDS for service discovery. It defines a universal, extensible control API for microservices — any configuration can be resolved through discovery.

The Four Core Resources in Envoy

Each xDS type corresponds to a specific resource type. The type is stored in the TypeUrl field of every DiscoveryRequest and DiscoveryResponse, in the format:

type.googleapis.com/<resource type>

For example: type.googleapis.com/envoy.api.v2.Cluster means this is a CDS (Cluster) resource.

1. LDS — Listener Discovery Service

A Listener is a port that Envoy opens to accept incoming connections. It can be configured with L3/L4 filters.

{
  "name": "...",
  "address": "{...}",
  "filter_chains": [],
  "listener_filters": [],
  "traffic_direction": "...",
  "access_log": []
}

2. RDS — Route Discovery Service

Routes act as the bridge between Listeners and Clusters. They define traffic distribution rules, virtual hosts, header manipulation, timeouts, and retries.

{
  "name": "...",
  "virtual_hosts": [],
  "response_headers_to_add": [],
  "request_headers_to_add": [],
  "validate_clusters": "{...}"
}

3. CDS — Cluster Discovery Service

A Cluster is an abstraction of an upstream service. It includes load balancing policy, health checks, circuit breaker config, and TLS settings.

{
  "name": "...",
  "type": "...",
  "eds_cluster_config": "{...}",
  "lb_policy": "...",
  "health_checks": [],
  "circuit_breakers": "{...}",
  "outlier_detection": "{...}"
}

4. EDS — Endpoint Discovery Service

EDS is the actual service discovery layer. It returns the live endpoints (IP + port) for a given cluster.

{
  "cluster_name": "...",
  "endpoints": [],
  "policy": "{...}"
}

5. SDS — Secret Discovery Service

SDS enables dynamic TLS certificate rotation without restarting Envoy. In early Istio versions, certificate updates required a hot restart — SDS eliminated that entirely.

{
  "name": "...",
  "tls_certificate": "{...}",
  "validation_context": "{...}"
}

How xDS Works: gRPC Streaming Subscription

Early xDS used REST/JSON polling. Starting from v2, it switched to gRPC bidirectional streaming, which provides:

Lower latency for config updates
Better performance under high churn
Native support for ACK/NACK flow control

Request Flow for a Typical HTTP Route

The API request order follows the dependency chain:

LDS → RDS → CDS → EDS

Envoy fetches Listeners (LDS) to know which ports to open
From the Listener config, it gets the Route name → fetches Routes (RDS)
Routes reference Clusters → fetches Clusters (CDS)
Clusters need live Endpoints → fetches Endpoints (EDS)

Full Subscription vs. Delta (Incremental) Subscription

Mode	Behavior
Full (SotW)	Management server returns all subscribed resources on every update
Delta (Incremental)	Only changed resources are sent — much more efficient at scale

xDS Protocol Deep Dive

Request Message Example

version_info: ""          # empty = first request
node:
  id: envoy               # unique node identifier (e.g. hostname)
resource_names:
  - foo
  - bar
type_url: type.googleapis.com/envoy.api.v2.ClusterLoadAssignment
response_nonce: ""        # empty = first request

Key fields:

version_info: empty on first request; subsequent requests echo the server's version
node.id: only required on the first message per stream
resource_names: the specific resources being subscribed to
response_nonce: used to correlate ACK/NACK with a specific server push

Response Message Example

version_info: X
resources:
  - foo ClusterLoadAssignment proto encoding
  - bar ClusterLoadAssignment proto encoding
type_url: type.googleapis.com/envoy.api.v2.ClusterLoadAssignment
nonce: A

ACK and NACK

After receiving a push from the management server, Envoy responds with either:

✅ ACK: Config applied successfully → sends back the new version_info
❌ NACK: Config failed to apply → sends back the old version_info with an error detail

This gives the control plane full visibility into whether each Envoy instance has successfully adopted a new config.

Resource Update & The Nonce Problem

Here's where it gets interesting. Consider this scenario:

Envoy subscribes to cluster foo (version X, nonce A)
Management server pushes a new version Y (nonce B) because foo's endpoints changed
At the same time, Envoy wants to add a new subscription to cluster bar

If Envoy sends a new DiscoveryRequest with version_info=X and resource_names=[foo, bar], the management server might misinterpret this as a NACK for version Y.

Solution: Nonce

The nonce uniquely identifies each push. Envoy's new subscription request carries nonce=A (from its last ACK), while the version Y push has nonce=B. The management server can distinguish between them unambiguously.

Istio's pragmatic approach: Istio's control plane (Pilot) doesn't strictly follow the nonce/version_info spec. Instead, it checks whether the resource_names (Clusters list) has actually changed. If yes, it treats the request as a resource update rather than ACK/NACK. Simpler and easier to reason about.

Summary

xDS is the backbone of Envoy's dynamic configuration system. Understanding it is essential for anyone working with Istio, Envoy-based gateways, or building custom control planes.

Key takeaways:

LDS → RDS → CDS → EDS is the standard dependency chain
gRPC streaming replaced REST polling for real-time, low-latency updates
Nonce solves the ACK/NACK ambiguity problem in concurrent update scenarios
SDS enables zero-downtime certificate rotation
xDS is not Istio-specific — gRPC and other frameworks use it too

💻 Explore the full Istio + Envoy implementation:
github.com/muzinan123/servicemesh

📖 Next in this series: Istio & Envoy Service Mesh Architecture

client-go Deep Dive: Operator in Practice — Building a Canary Release Controller

James Lee — Tue, 19 May 2026 10:10:38 +0000

In the previous article we defined the Canary CRD. Now we build the Operator: the controller that watches Canary resources and drives the actual release process. This is the final article in the series — everything we've covered comes together here.

1. What Is an Operator?

Operator = CRD (schema) + Controller (reconciliation logic)

┌──────────────────────────────────────────────────────┐
│  Kubernetes Controller Manager                       │
│  manages built-in controllers:                       │
│  DeploymentController, ReplicaSetController, ...     │
└──────────────────────────────────────────────────────┘
                    +
┌──────────────────────────────────────────────────────┐
│  Your Operator                                       │
│  = CRD (Canary) + CanaryController                  │
│  encodes domain knowledge:                           │
│  "how to safely roll out a new version in batches"   │
└──────────────────────────────────────────────────────┘

Operator = a Kubernetes extension that solves domain-specific problems the platform itself doesn't know how to solve.

2. Project Structure

canary-controller/
├── main.go                          ← controller entrypoint
├── go.mod
├── manifests/
│   └── crd-canary.yaml              ← CRD definition (Part 8)
└── pkg/
    ├── apis/
    │   └── canarycontroller/
    │       ├── register.go          ← GroupName constant
    │       └── v1alpha1/
    │           ├── doc.go           ← code-gen annotations
    │           ├── types.go         ← Canary/CanarySpec/CanaryStatus
    │           └── register.go      ← register types with runtime.Scheme
    ├── controllers/
    │   └── canary_controller.go     ← all reconciliation logic
    ├── generated/                   ← auto-generated by code-generator
    │   ├── clientset/               ← typed Canary clientset
    │   ├── informers/               ← typed Canary informer
    │   └── listers/                 ← typed Canary lister
    └── signals/                     ← OS signal handling (SIGTERM/SIGINT)

3. Generating the Typed Client Code

The built-in kubernetes.Clientset only supports built-in resources. For your CRD, you need a generated typed client:

git clone https://github.com/kubernetes/code-generator $GOPATH/src/k8s.io/code-generator

cd $GOPATH/src/k8s.io/code-generator && ./generate-groups.sh all \
  canary-controller/pkg/client \      ← output
  canary-controller/pkg/apis \        ← input: your types
  canarycontroller:v1alpha1

Generated output	Purpose
`zz_generated.deepcopy.go`	`DeepCopy()` for all types
`generated/clientset/`	`CanarycontrollerV1alpha1().Canaries(ns)` — typed CRUD
`generated/informers/`	Typed `CanaryInformer` with `Lister()` and `HasSynced()`
`generated/listers/`	`CanaryLister` — fast local cache queries

4. Controller Struct & Wiring

type Controller struct {
    kubeclientset   kubernetes.Interface   // for built-in resources (Deployment)
    canaryclientset clientset.Interface    // for Canary resources (generated)

    deploymentsLister appslisters.DeploymentLister
    deploymentsSynced cache.InformerSynced
    canariesLister    listers.CanaryLister
    canariesSynced    cache.InformerSynced

    workqueue workqueue.RateLimitingInterface
    recorder  record.EventRecorder
}

In NewController, two event subscriptions are registered:

① Canary Add/Update  → enqueueCanary()  → workqueue.Add(key)
  Direct: user changed Canary spec → reconcile immediately

② Deployment Add/Update/Delete → handleObject()
  → find owning Canary by name prefix → enqueueCanary()
  Indirect: managed Deployment changed state → re-check Canary

5. Run → Worker → syncHandler

controller.Run(threadiness=2, stopCh)
     │
     ├── cache.WaitForCacheSync()     ← block until Indexer ready
     │
     └── 2× goroutine: runWorker()
             └── processNextWorkItem()
                   ├── workqueue.Get(key)
                   ├── syncHandler(key)
                   │     ├── OK  → workqueue.Forget(key)
                   │     └── ERR → workqueue.AddRateLimited(key)  ← backoff retry
                   └── workqueue.Done(key)

syncHandler is the reconciliation core — it reads the Canary spec and routes to the correct release strategy:

func (c *Controller) syncHandler(key string) error {
    // ① Fetch from local cache — no API Server call
    canary, err := c.canariesLister.Canaries(namespace).Get(name)

    // ② Route to release strategy
    switch canary.Spec.Info["type"] {
    case "NormalDeploy":   c.normalDeploy(...)
    case "CanaryDeploy":   c.firstCanaryDeploy(...) or c.notFirstCanaryDeploy(...)
    case "CanaryRollback": c.canaryRollback(...)
    }

    // ③ Update status + emit Event
    c.updateCanaryStatus(canary, deployment)
    c.recorder.Event(canary, corev1.EventTypeNormal, SuccessSynced, ...)
}

6. The Three Release Strategies

NormalDeploy — Rolling Update

Deployment exists?
├── No  → Create
└── Yes → image / replicas / CPU / memory drifted? → Update

Standard Kubernetes rolling update. No batching, no pause. The controller simply ensures the Deployment matches what's declared in the Canary spec.

CanaryDeploy — Batched Release

Splits the rollout into N batches. Batch 1 always pauses for human approval; subsequent batches auto-advance (when pauseType=First).

Example: 4 replicas, 2 batches, pauseType=First

Start:    [v1][v1][v1][v1]   old=4, new=0

Batch 1:  [v2][v1][v1][v1]   old=3, new=1
               │
               └── PAUSE — wait for: kubectl patch canary ... currentBatch=2

Batch 2:  [v2][v2][v2][v2]   old=0, new=4  → old Deployment deleted ✅

Replica calculation per batch:

everyAddReplicas = round(totalReplicas / totalBatches)

batch N target replicas = 1 + (N-1) × everyAddReplicas
final batch             = totalReplicas (full rollout)

old replicas = totalReplicas - new.AvailableReplicas

The controller continuously reconciles both Deployments — if anything drifts (pod crash, manual edit), the next reconcile loop corrects it automatically.

CanaryRollback — Reverse the Canary

Rollback type?
├── NormalDeploy rollback:
│     old.Name == new.Name → just update image back, no second Deployment
│
└── CanaryDeploy rollback:
      Restore old version replicas while draining new version
      new.AvailableReplicas → 0 → delete new Deployment ✅

7. Status Reporting: updateCanaryStatus

After every reconcile, the controller updates the Canary's status subresource to reflect real-time progress:

CanaryDeploy:
  batch N running  → status.info["batchNStatus"] = "Ing"
  batch N complete → status.info["batchNStatus"] = "Finished"
  availableReplicas updated on every loop

CanaryRollback:
  old Deployment still exists → status.info["rollbackStatus"] = "Ing"
  old Deployment deleted      → status.info["rollbackStatus"] = "Finished"

NormalDeploy:
  no status update needed

Always DeepCopy() before mutating — the object from Lister is a read-only reference to the Indexer cache. Modifying it directly corrupts the local cache.

canaryCopy := canary.DeepCopy()   // ✅ safe to mutate
canaryCopy.Status.Info[...] = "Finished"
c.canaryclientset.CanarycontrollerV1alpha1().
    Canaries(ns).UpdateStatus(canaryCopy, ...)

8. Complete Execution Flow

main.go
  └── NewController(kubeclientset, canaryclientset, deployInformer, canaryInformer)
        ├── AddEventHandler(Canary)     → enqueueCanary
        └── AddEventHandler(Deployment) → handleObject → enqueueCanary

factory.Start(stopCh)          → Reflector List/Watch (Deployment + Canary)
cache.WaitForCacheSync(...)    → wait for Indexer fully populated

controller.Run(2, stopCh)
  └── syncHandler("tech-prod/go-hello-canary")
        │
        ├── canariesLister.Get()        ← Indexer (no API call)
        ├── switch type:
        │     NormalDeploy   → Create or Update Deployment
        │     CanaryDeploy   → Batch 1: new=1, old=N-1, PAUSE
        │                      Batch N: new=total, old=0, DELETE old
        │     CanaryRollback → drain new, restore old
        │
        ├── updateCanaryStatus()        ← PATCH /status
        └── recorder.Event(SuccessSynced)

9. Key Design Principles

Principle	How
Never mutate cache objects	Always `DeepCopy()` before modifying
Read cache, write API	`Lister.Get()` for reads; `clientset.Update()` for writes
Idempotent reconciliation	Every loop checks actual vs desired — safe to re-run
Ownership	`OwnerReferences` ensures Deployments are GC'd with their Canary
Error → retry	`workqueue.AddRateLimited(key)` — exponential backoff on failure

Series Complete 🎉

#	Article	Key concept
1	Informer	List/Watch, SharedInformerFactory
2	Reflector	ListerWatcher, ListAndWatch, ResourceVersion
3	DeltaFIFO	queue+items, produce/consume, HandleDeltas
4	Indexer	ThreadSafeMap, IndexFunc, ByIndex
5	WorkQueue	Deduplication, Exponential Backoff, Token Bucket
6	EventBroadcaster	Event resource, Recorder, Sink
7	Controller (internal)	Run, processLoop, WaitForCacheSync
8	CRD	Schema, Go types, spec/status contract
9	Operator in Practice	Full canary release controller end-to-end

Thank you for following the series. If you found it useful, share it with your team!

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: CRD — Extending the Kubernetes API with Custom Resources

James Lee — Tue, 19 May 2026 10:10:13 +0000

In the previous seven articles we've thoroughly explored the Informer framework — Reflector, DeltaFIFO, Indexer, WorkQueue, EventBroadcaster, and the internal Controller. Now we shift from framework internals to application development: building your own Kubernetes extensions with CRD and Operator.

This article covers CRD — the foundation. The next article will build the Operator (controller) on top of it.

1. The Kubernetes Declarative Model

Before defining a CRD, it's worth understanding why CRDs exist.

Kubernetes is built on a declarative model:

┌─────────────────────────────────────────────────────────────┐
│               Kubernetes Declarative Model                  │
│                                                             │
│  User declares desired state:                               │
│    "I want 3 replicas of nginx running"                     │
│                          │                                  │
│                          ▼                                  │
│  API Server stores it in etcd                               │
│                          │                                  │
│                          ▼                                  │
│  Controller Manager watches for drift:                      │
│    actual state ≠ desired state → take action               │
│                          │                                  │
│                          ▼                                  │
│  Actual state converges to desired state                    │
│  (user never specifies HOW — only WHAT)                     │
└─────────────────────────────────────────────────────────────┘

Built-in resources (Deployment, StatefulSet, DaemonSet, Service, Ingress, ConfigMap…) cover most use cases. But when they don't, Kubernetes provides two extension mechanisms:

Mechanism	What it provides
CRD (Custom Resource Definition)	A new resource type — defines the schema
Operator (Custom Controller)	The reconciliation logic — watches the CRD and acts on it

CRD = the "what". Operator = the "how".

2. Anatomy of a CRD YAML

Here's a complete CRD definition for a canary deployment resource:

apiVersion: apiextensions.k8s.io/v1beta1
kind: CustomResourceDefinition
metadata:
  # Format: {plural}.{group}
  name: canaries.canarycontroller.tech.com
  annotations:
    "api-approved.kubernetes.io": "https://github.com/kubernetes/kubernetes/pull/78458"

spec:
  # API group — appears in REST path: /apis/{group}/{version}
  group: canarycontroller.tech.com

  versions:
    - name: v1alpha1
      served: true    # this version is active and served by the API Server
      storage: true   # this version is used for storage in etcd (only one allowed)

  # Namespaced: resource belongs to a specific namespace
  # Cluster: resource is cluster-wide (like Node, PersistentVolume)
  scope: Namespaced

  names:
    plural:    canaries   # kubectl get canaries
    singular:  canary     # kubectl get canary
    kind:      Canary     # used in YAML: kind: Canary
    shortNames:
      - can               # kubectl get can

  subresources:
    status: {}            # enables /status subresource (separate update endpoint)

Key fields explained

┌──────────────────────────────────────────────────────────────────┐
│  metadata.name = "canaries.canarycontroller.tech.com"            │
│                   ────────  ──────────────────────               │
│                   plural    group                                 │
│                   name      name                                  │
└──────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│  REST API path after registration:                               │
│  /apis/canarycontroller.tech.com/v1alpha1/namespaces/{ns}/canaries│
│         ──────────────────────── ────────                        │
│         group                    version                         │
└──────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│  subresources.status: {}                                         │
│  → enables separate /status endpoint                             │
│  → controller updates status via PATCH /status                   │
│  → user updates spec via PATCH /spec                             │
│  → prevents spec/status update conflicts                         │
└──────────────────────────────────────────────────────────────────┘

Register the CRD with the cluster

# Apply the CRD definition
kubectl create -f crd-canary.yaml

# Verify it's registered
kubectl get crd canaries.canarycontroller.tech.com

# The new resource type is now available
kubectl get canaries --all-namespaces
kubectl get can -n tech-prod

3. A Custom Resource Instance

Once the CRD is registered, you can create instances of it — just like built-in resources:

apiVersion: canarycontroller.tech.com/v1alpha1
kind: Canary
metadata:
  name: go-hello-canary
  namespace: tech-prod
  labels:
    app: go-hello
    tech.com/clusterId: prod
  # Auto-populated by API Server:
  # resourceVersion: "430698608"
  # uid: c5bb13a2-b2be-11ea-8fef-e4434b7c7170

spec:
  info:
    type: CanaryDeploy
    totalBatches: "2"       # deploy in 2 batches
    currentBatch: "2"
    pauseType: First        # pause after first batch — wait for human approval
    newDeploymentYaml: |    # full YAML of the new version deployment
      ...
    oldDeploymentYaml: |    # full YAML of the current version deployment
      ...

status:
  info:
    availableReplicas: "4"
    batch2Status: Finished

What this describes:

go-hello application canary release strategy:
┌─────────────────────────────────────────────────────────┐
│  Total instances: 4                                     │
│  Release strategy: 2 batches                            │
│  Batch 1: deploy 2 new instances → PAUSE                │
│           (wait for human approval via kubectl patch)   │
│  Batch 2: deploy remaining 2 instances → DONE           │
└─────────────────────────────────────────────────────────┘

4. Go Type Definitions for the CRD

To build a controller for this CRD, you need to define the corresponding Go types. These are what code-generator will use to generate the typed client, informer, and lister code.

Directory structure

pkg/
└── apis/
    └── canarycontroller/
        ├── register.go          ← group name constant
        └── v1alpha1/
            ├── doc.go           ← code-gen annotations
            ├── types.go         ← CRD Go type definitions  ← main file
            └── register.go      ← register types with scheme

register.go — group name

// pkg/apis/canarycontroller/register.go
package canarycontroller

const (
    GroupName = "canarycontroller.tech.com"
)

doc.go — code generation annotations

// pkg/apis/canarycontroller/v1alpha1/doc.go

// +k8s:deepcopy-gen=package
// +groupName=canarycontroller.tech.com

package v1alpha1

types.go — the core type definitions

// pkg/apis/canarycontroller/v1alpha1/types.go
package v1alpha1

import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

// +genclient
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// Canary is the Schema for the canaries API
type Canary struct {
    metav1.TypeMeta   `json:",inline"`
    metav1.ObjectMeta `json:"metadata,omitempty"`

    Spec   CanarySpec   `json:"spec,omitempty"`
    Status CanaryStatus `json:"status,omitempty"`
}

// CanarySpec defines the desired state of a Canary release
type CanarySpec struct {
    Info CanaryInfo `json:"info"`
}

type CanaryInfo struct {
    // "CanaryDeploy" or "ABTest"
    Type string `json:"type"`

    // Total number of batches to roll out
    TotalBatches string `json:"totalBatches"`

    // Current batch being processed
    CurrentBatch string `json:"currentBatch"`

    // "First": pause after first batch and wait for approval
    // "None": roll out all batches automatically
    PauseType string `json:"pauseType"`

    // Full YAML of the new version Deployment
    NewDeploymentYaml string `json:"newDeploymentYaml"`

    // Full YAML of the current version Deployment
    OldDeploymentYaml string `json:"oldDeploymentYaml"`
}

// CanaryStatus defines the observed state (written by the controller)
type CanaryStatus struct {
    Info CanaryStatusInfo `json:"info,omitempty"`
}

type CanaryStatusInfo struct {
    AvailableReplicas string `json:"availableReplicas,omitempty"`
    Batch1Status      string `json:"batch1Status,omitempty"`
    Batch2Status      string `json:"batch2Status,omitempty"`
}

// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object

// CanaryList is required by the API machinery for List operations
type CanaryList struct {
    metav1.TypeMeta `json:",inline"`
    metav1.ListMeta `json:"metadata"`
    Items []Canary  `json:"items"`
}

Code generation annotations explained

Annotation	Effect
`// +genclient`	Generate a typed client (`CanaryInterface`) for this type
`// +k8s:deepcopy-gen:interfaces=...`	Generate `DeepCopyObject()` — required by `runtime.Object`
`// +k8s:deepcopy-gen=package`	Generate `DeepCopy()` for all types in this package
`// +groupName=...`	Set the API group for generated code

5. Spec vs Status: The Controller Contract

The separation of spec and status is a fundamental Kubernetes design pattern:

┌─────────────────────────────────────────────────────────────┐
│  spec   — desired state                                     │
│  Written by: the user (kubectl apply)                       │
│  Read by:    the controller                                 │
│                                                             │
│  status — observed state                                    │
│  Written by: the controller (via /status subresource)       │
│  Read by:    the user (kubectl get/describe)                │
└─────────────────────────────────────────────────────────────┘

Controller reconciliation loop:
┌──────────────────────────────────────────────────────────────┐
│                                                              │
│  Read spec  →  compare with status  →  take action          │
│                                              │               │
│                                              ▼               │
│                                       update status          │
│                                       (reflect new reality)  │
└──────────────────────────────────────────────────────────────┘

Enabling subresources.status: {} in the CRD YAML is critical — it creates a separate /status API endpoint, which means:

Users updating spec won't accidentally overwrite status
Controllers updating status won't accidentally overwrite spec
RBAC can grant different permissions for spec vs status updates

6. The CRD Lifecycle

Step 1: Define the CRD schema (YAML)
        kubectl create -f crd-canary.yaml
        → API Server registers new resource type
        → REST endpoint /apis/canarycontroller.tech.com/v1alpha1/... becomes available

Step 2: Define Go types (types.go)
        → mirrors the CRD schema in Go structs

Step 3: Run code-generator
        → generates typed client (Clientset)
        → generates typed Informer + Lister
        → generates DeepCopy methods

Step 4: Create resource instances
        kubectl apply -f canary-go-hello.yaml
        → Canary object stored in etcd

Step 5: Controller watches and reconciles
        → Informer watches Canary resources
        → Controller reads spec, takes action, updates status
        → Desired state converges to actual state

7. Summary

Concept	Detail
CRD	Registers a new resource type with the API Server — no code required
Custom Resource	An instance of a CRD — stored in etcd, managed via kubectl
spec	Desired state — written by users, read by controllers
status	Observed state — written by controllers, read by users
subresources.status	Separates spec/status update paths — prevents conflicts
scope	`Namespaced` or `Cluster` — determines resource visibility
group/version	Forms the REST API path: `/apis/{group}/{version}/...`
code-generator	Generates typed client, Informer, Lister, and DeepCopy from Go types

CRD is the "schema" half of the Operator pattern. It extends the Kubernetes API with your domain-specific resource types — giving you the same declarative, watch-driven, etcd-backed machinery that built-in resources enjoy. The controller (Operator) is what brings those resources to life.

Next in this series: Operator in Practice: Building a Canary Release Controller (Part 9 — Final)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: Controller — The Central Hub That Wires the Informer Framework Together

James Lee — Tue, 19 May 2026 10:09:48 +0000

Over the past six articles we've examined each component of the Informer framework individually. Now it's time to look at the component that wires them all together: the Informer-internal Controller.

⚠️ Naming clarification: The "Controller" in this article is k8s.io/client-go/tools/cache/controller.go — the Informer-internal orchestrator. It is NOT the same as the business-logic controller you write to reconcile custom resources. We'll call it the Informer Controller throughout this article to avoid confusion.

Source: k8s.io/client-go/tools/cache/controller.go

1. The Two Types of "Controller" in Kubernetes

Before diving in, let's be precise about terminology:

┌──────────────────────────────────────────────────────────────────┐
│  Type A: Informer Controller (this article)                      │
│  Package: k8s.io/client-go/tools/cache                          │
│  Role: internal orchestrator — starts Reflector, drives          │
│        processLoop, connects DeltaFIFO → Indexer + callbacks     │
│  You never instantiate this directly.                            │
└──────────────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────────────┐
│  Type B: Your Application Controller                             │
│  Package: your codebase                                          │
│  Role: business logic — reconciles desired vs actual state       │
│        using WorkQueue + Lister + Clientset                      │
│  You write and own this.                                         │
└──────────────────────────────────────────────────────────────────┘

2. The Controller Struct

// k8s.io/client-go/tools/cache/controller.go
type controller struct {
    config         Config          // all dependencies injected here
    reflector      *Reflector      // created from config at startup
    reflectorMutex sync.RWMutex
    clock          clock.Clock
}

The controller itself is intentionally thin — all meaningful configuration lives in Config:

type Config struct {
    // DeltaFIFO — the event queue (producer: Reflector, consumer: processLoop)
    Queue

    // Provides List() and Watch() — injected into Reflector at startup
    ListerWatcher

    // The callback invoked by processLoop for each item popped from DeltaFIFO
    // In practice: sharedIndexInformer.HandleDeltas
    Process ProcessFunc

    // The resource type being watched (e.g. *v1.Pod)
    ObjectType runtime.Object

    // How often to force a full resync (0 = never)
    FullResyncPeriod time.Duration

    // Optional: custom function to decide whether to resync on each tick
    ShouldResync ShouldResyncFunc

    // If true: when Process() returns an error, re-enqueue the item
    RetryOnError bool

    // Called when Watch() returns an error
    WatchErrorHandler WatchErrorHandler

    // Pagination size for Watch requests
    WatchListPageSize int64
}

Config field map

Field	Provided by	Used for
`Queue`	DeltaFIFO	Event buffer between Reflector and processLoop
`ListerWatcher`	Pod/Deployment Informer's ListFunc+WatchFunc	Injected into Reflector
`Process`	`sharedIndexInformer.HandleDeltas`	Callback for each popped event batch
`ObjectType`	e.g. `&v1.Pod{}`	Tells Reflector what type to deserialize
`FullResyncPeriod`	User config (e.g. `30*time.Minute`)	Periodic forced re-List
`RetryOnError`	Usually `false`	Re-enqueue on Process() failure
`WatchErrorHandler`	Optional custom handler	React to Watch connection errors

3. How the Informer Controller Starts: Run()

func (c *controller) Run(stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    go func() {
        <-stopCh
        c.config.Queue.Close()
    }()

    // ① Create Reflector from Config
    r := NewReflector(
        c.config.ListerWatcher,
        c.config.ObjectType,
        c.config.Queue,       // DeltaFIFO is passed as the Store
        c.config.FullResyncPeriod,
    )
    r.ShouldResync   = c.config.ShouldResync
    r.WatchListPageSize = c.config.WatchListPageSize
    r.clock          = c.clock
    if c.config.WatchErrorHandler != nil {
        r.watchErrorHandler = c.config.WatchErrorHandler
    }

    c.reflectorMutex.Lock()
    c.reflector = r
    c.reflectorMutex.Unlock()

    var wg wait.Group
    defer wg.Wait()

    // ② Start Reflector in a goroutine — runs List/Watch forever
    wg.StartWithChannel(stopCh, r.Run)

    // ③ Start processLoop — consumes DeltaFIFO forever
    wait.Until(c.processLoop, time.Second, stopCh)
}

Startup sequence:

controller.Run(stopCh)
     │
     ├── ① NewReflector(ListerWatcher, ObjectType, DeltaFIFO, resyncPeriod)
     │         → Reflector is wired to DeltaFIFO as its Store
     │
     ├── ② go r.Run(stopCh)
     │         → Reflector.ListAndWatch() starts in background goroutine
     │         → Phase 1: List → syncWith() → DeltaFIFO seeded
     │         → Phase 2: Watch → watchHandler() → events stream into DeltaFIFO
     │
     └── ③ wait.Until(c.processLoop, 1s, stopCh)
               → processLoop runs in foreground, restarted every 1s if it exits

4. The processLoop: Consuming DeltaFIFO

func (c *controller) processLoop() {
    for {
        // Pop() blocks until DeltaFIFO has an item
        // Passes each item to c.config.Process (= HandleDeltas)
        obj, err := c.config.Queue.Pop(PopProcessFunc(c.config.Process))
        if err != nil {
            if err == ErrFIFOClosed {
                return  // queue was shut down — exit cleanly
            }
            if c.config.RetryOnError {
                // Re-enqueue the item if RetryOnError is set
                c.config.Queue.AddIfNotPresent(obj)
            }
        }
    }
}

processLoop flow:

processLoop (runs forever)
     │
     └── DeltaFIFO.Pop(HandleDeltas)
           │
           ├── blocks on cond.Wait() if queue is empty
           ├── dequeues next key (FIFO order)
           └── calls HandleDeltas([]Delta)
                      │
                      ├── Indexer.Add/Update/Delete   ← keep local cache in sync
                      └── sharedProcessor.distribute() ← fan-out to user callbacks

5. The Complete Informer Framework — All Components Together

Now that we've covered every component, here is the complete picture:

┌──────────────────────────────────────────────────────────────────────────┐
│                    Informer Framework (full picture)                     │
│                                                                          │
│  ┌──────────────────────────────────────────────────────────────────┐    │
│  │                    sharedIndexInformer                           │    │
│  │                                                                  │    │
│  │  ┌─────────────────────────────────────────────────────────┐    │    │
│  │  │              Informer Controller (internal)             │    │    │
│  │  │                                                         │    │    │
│  │  │   API Server                                            │    │    │
│  │  │       │  List/Watch (HTTP chunked)                      │    │    │
│  │  │       ▼                                                 │    │    │
│  │  │  ┌──────────┐  events   ┌──────────────┐               │    │    │
│  │  │  │ Reflector│ ────────► │  DeltaFIFO   │               │    │    │
│  │  │  │(goroutine│           │  queue+items │               │    │    │
│  │  │  └──────────┘           └──────┬───────┘               │    │    │
│  │  │                                │ Pop()                  │    │    │
│  │  │                                ▼                        │    │    │
│  │  │                        processLoop()                    │    │    │
│  │  │                                │                        │    │    │
│  │  └────────────────────────────────┼────────────────────────┘    │    │
│  │                                   │ HandleDeltas                 │    │
│  │                    ┌──────────────┴──────────────┐              │    │
│  │                    ▼                             ▼              │    │
│  │             ┌────────────┐            ┌──────────────────┐      │    │
│  │             │  Indexer   │            │ sharedProcessor  │      │    │
│  │             │(ThreadSafe │            │  distribute()    │      │    │
│  │             │   Map)     │            └────────┬─────────┘      │    │
│  │             └────────────┘                     │                │    │
│  │                    ▲                           ▼                │    │
│  │                    │ Lister             AddEventHandler         │    │
│  └────────────────────┼───────────────────────────┼───────────────┘    │
│                       │                           │                     │
│              ┌────────┴──────────────────────────▼──────────────┐      │
│              │              Your Application Controller          │      │
│              │                                                   │      │
│              │  podsLister.Pods(ns).Get(name)  ← read from cache │      │
│              │  clientset.CoreV1().Pods().Update() ← write to API│      │
│              │  workqueue.AddRateLimited(key)   ← retry on error │      │
│              │  recorder.Eventf(obj, ...)       ← emit events    │      │
│              └───────────────────────────────────────────────────┘      │
└──────────────────────────────────────────────────────────────────────────┘

6. Data Flow: End to End

① factory.Start(stopCh)
   └── for each registered informer:
         go sharedIndexInformer.Run(stopCh)
              └── controller.Run(stopCh)
                    ├── go Reflector.Run(stopCh)   [goroutine A]
                    └── processLoop()              [goroutine B, main]

② Goroutine A — Reflector
   └── ListAndWatch()
         ├── Phase 1: List → syncWith() → DeltaFIFO.Replace()
         │     all existing objects enqueued as "Added" or "Replaced"
         └── Phase 2: Watch → watchHandler() → DeltaFIFO.Add/Update/Delete
               incremental events stream in continuously

③ Goroutine B — processLoop
   └── DeltaFIFO.Pop(HandleDeltas)  [blocks when empty]
         └── HandleDeltas([]Delta)
               for each Delta:
               ├── Indexer.Add/Update/Delete   → local cache updated
               └── sharedProcessor.distribute()
                     └── for each listener (AddEventHandler):
                           ├── OnAdd(newObj)
                           ├── OnUpdate(oldObj, newObj)
                           └── OnDelete(oldObj)
                                 └── workqueue.Add(key)

④ Your controller worker goroutine(s)
   └── workqueue.Get(key)
         └── reconcile(key)
               ├── lister.Get(key)              → read from Indexer (no API call)
               ├── clientset.Update/Patch/Delete → write to API Server
               ├── recorder.Eventf(...)          → emit Event to etcd
               └── on error: workqueue.AddRateLimited(key)

7. WaitForCacheSync: Why It Matters

Before your controller starts processing, the local cache must be fully populated from the initial List. WaitForCacheSync blocks until this is complete:

func (kc *KubeController) Run(stopCh chan struct{}) {
    // Start all informers
    kc.factory.Start(stopCh)

    // Block until initial List is complete and DeltaFIFO is drained
    // HasSynced() returns true when initialPopulationCount reaches 0
    if !cache.WaitForCacheSync(stopCh,
        kc.podsSynced,
        kc.deploymentsSynced,
    ) {
        utilruntime.HandleError(fmt.Errorf("timed out waiting for caches to sync"))
        return
    }

    // Safe to start workers now — Indexer has a complete snapshot
    for i := 0; i < workers; i++ {
        go wait.Until(kc.runWorker, time.Second, stopCh)
    }

    <-stopCh
}

What HasSynced() tracks internally:

DeltaFIFO.initialPopulationCount
     │
     ├── set to N during Replace() (initial List result: N objects)
     ├── decremented by 1 on each Pop()
     └── HasSynced() = true when initialPopulationCount == 0
                        AND at least one Replace() has been called

Never start reconciliation workers before WaitForCacheSync returns. If you do, your Lister queries will return incomplete results and your controller will make incorrect decisions based on a partial view of cluster state.

8. Series Summary: The Complete Component Map

We've now covered every component in the Informer framework. Here's the complete reference:

Component	Package	Role	Article
Reflector	`tools/cache`	List/Watch against API Server; feeds DeltaFIFO	Part 2
DeltaFIFO	`tools/cache`	FIFO event queue; stores `{type, object}` deltas	Part 3
Indexer	`tools/cache`	Thread-safe in-memory cache with custom indexing	Part 4
WorkQueue	`util/workqueue`	Rate-limited, deduplicated task queue for workers	Part 5
EventBroadcaster	`tools/record`	Emits structured Events to API Server / logs	Part 6
Controller (internal)	`tools/cache`	Orchestrates Reflector + DeltaFIFO + processLoop	Part 7 (this)
sharedProcessor	`tools/cache`	Fan-out: delivers events to all AddEventHandler listeners	—
SharedInformerFactory	`informers`	Ensures one Reflector per resource type across all controllers	Part 1

9. Summary

Informer Controller = orchestrator, not business logic

controller.Run(stopCh)
     │
     ├── NewReflector(ListerWatcher, ObjectType, DeltaFIFO)
     │         → wires API Server connection to event queue
     │
     ├── go Reflector.Run()
     │         → List (seed cache) + Watch (stream events) forever
     │
     └── processLoop()
               → Pop from DeltaFIFO
               → HandleDeltas:
                     Indexer (storage) + sharedProcessor (callbacks)

Concept	Detail
controller struct	Thin wrapper — holds Config + Reflector reference
Config.Queue	DeltaFIFO — the handoff point between Reflector and processLoop
Config.Process	`HandleDeltas` — routes events to Indexer and sharedProcessor
Config.RetryOnError	Re-enqueues items when HandleDeltas returns an error
processLoop	Infinite loop: Pop → HandleDeltas → repeat
WaitForCacheSync	Safety gate — ensures Indexer is fully populated before workers start

The Informer Controller is the invisible backbone of every Kubernetes controller. By orchestrating Reflector, DeltaFIFO, Indexer, and sharedProcessor into a single coherent pipeline, it gives you a reliable, consistent, and efficient view of cluster state — so your application controller can focus entirely on reconciliation logic.

Next in this series: CRD: Extending the Kubernetes API with Custom Resources (Part 8)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: EventBroadcaster — How Kubernetes Records and Distributes Cluster Events

James Lee — Tue, 19 May 2026 10:09:20 +0000

In the previous article we covered WorkQueue — the retry and rate-limiting backbone of controllers. In this article we look at a different but equally important mechanism: EventBroadcaster. It's how Kubernetes components record what they're doing, and how you can do the same in your own controllers.

Source: k8s.io/client-go/tools/record/

1. What Is a Kubernetes Event?

A Kubernetes Event is a first-class resource object — stored in etcd, queryable via kubectl, and subject to the same API machinery as Pods or Deployments.

# View recent events in a namespace
kubectl get events -n default

# View events for a specific Pod (shown in describe output)
kubectl describe pod nginx-deployment-7d8ff89d87-25kb4

LAST SEEN   TYPE      REASON      OBJECT                          MESSAGE
2m          Normal    Scheduled   pod/nginx-7d8ff89d87-25kb4      Successfully assigned default/nginx to node-1
2m          Normal    Pulling     pod/nginx-7d8ff89d87-25kb4      Pulling image "nginx:1.21"
1m          Normal    Pulled      pod/nginx-7d8ff89d87-25kb4      Successfully pulled image in 3.2s
1m          Normal    Created     pod/nginx-7d8ff89d87-25kb4      Created container nginx
1m          Normal    Started     pod/nginx-7d8ff89d87-25kb4      Started container nginx

⚠️ Important distinction: Kubernetes Events are resource objects managed by the API Server. They are NOT the same as the etcd watch callback events used internally by Informer/Reflector. Don't confuse the two.

2. The Event Resource Structure

// k8s.io/api/core/v1/types.go
type Event struct {
    metav1.TypeMeta
    metav1.ObjectMeta

    // The object this event is about (e.g. the Pod that was scheduled)
    InvolvedObject ObjectReference

    // Short machine-readable reason string (e.g. "Scheduled", "Pulling", "Failed")
    Reason  string

    // Human-readable description of what happened
    Message string

    // Component that generated this event
    Source EventSource   // {Component: "scheduler", Host: "master-1"}

    // Timing
    FirstTimestamp metav1.Time      // when this event first occurred
    LastTimestamp  metav1.Time      // most recent occurrence
    Count          int32            // how many times this event has occurred

    // "Normal" or "Warning"
    Type string

    // High-precision timestamp (for newer event API)
    EventTime metav1.MicroTime

    // For aggregated/series events
    Series *EventSeries

    // What action was taken
    Action string

    // Secondary object involved (e.g. the Node a Pod was scheduled to)
    Related *ObjectReference

    // For audit: which controller generated this event
    ReportingController string
    ReportingInstance   string
}

Key fields explained

Field	Example value	Purpose
`InvolvedObject`	`{Kind:"Pod", Name:"nginx-xxx"}`	The resource this event describes
`Reason`	`"Scheduled"`, `"Pulling"`, `"BackOff"`	Machine-readable event category
`Message`	`"Successfully assigned to node-1"`	Human-readable detail
`Type`	`"Normal"` or `"Warning"`	Severity indicator
`Count`	`5`	Deduplication counter — same event aggregated
`FirstTimestamp` / `LastTimestamp`	timestamps	When it first/last occurred
`Source.Component`	`"kubelet"`, `"scheduler"`	Which Kubernetes component fired it

3. Event Retention Policy

Since Events are stored in etcd, unchecked growth would fill disk space. Kubernetes enforces a strict retention policy:

┌─────────────────────────────────────────────────────────┐
│               Event Retention Rules                     │
│                                                         │
│  • Default TTL: 1 hour after last occurrence            │
│  • kubectl get events shows only last-hour events       │
│  • Identical events are AGGREGATED (Count++ instead     │
│    of creating duplicate objects)                       │
│  • etcd stores events in a separate keyspace to         │
│    prevent them from crowding out other resources       │
└─────────────────────────────────────────────────────────┘

This is why kubectl describe pod shows "Events: " for Pods that have been running stably for more than an hour — the events existed, but have been garbage collected.

4. The EventBroadcaster Pipeline

EventBroadcaster is the mechanism that takes an event generated by a component and delivers it to one or more sinks (e.g. the API Server, a log file, stdout).

┌──────────────────────────────────────────────────────────────────┐
│                  EventBroadcaster Pipeline                       │
│                                                                  │
│  Your Controller                                                 │
│       │                                                          │
│       │  recorder.Eventf(obj, type, reason, message)             │
│       ▼                                                          │
│  ┌──────────────┐                                                │
│  │EventRecorder │  generates Event object                        │
│  └──────┬───────┘                                                │
│         │                                                        │
│         ▼                                                        │
│  ┌──────────────────┐                                            │
│  │ EventBroadcaster │  fan-out to multiple sinks                 │
│  └──────┬───────────┘                                            │
│         │                                                        │
│    ┌────┴──────────────────────────┐                             │
│    ▼                               ▼                             │
│  ┌──────────────────┐   ┌──────────────────────────┐            │
│  │  API Server Sink │   │  Logging Sink             │            │
│  │  (writes Event   │   │  (prints to stdout/log)   │            │
│  │   to etcd)       │   │                           │            │
│  └──────────────────┘   └──────────────────────────┘            │
└──────────────────────────────────────────────────────────────────┘

The three actors

Actor	Role
EventRecorder	The interface your code calls — `Event()`, `Eventf()`, `AnnotatedEventf()`
EventBroadcaster	Receives events from recorders, fans out to registered sinks
EventSink	The destination — API Server (etcd), log output, or custom handler

5. Using EventBroadcaster in a Custom Controller

Here's the complete setup pattern used in real Kubernetes controllers:

Step 1 — Create the broadcaster and recorder

import (
    "k8s.io/client-go/tools/record"
    "k8s.io/client-go/kubernetes/scheme"
    corev1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/runtime"
)

func NewController(clientset kubernetes.Interface, ...) *Controller {
    // Create the broadcaster
    eventBroadcaster := record.NewBroadcaster()

    // Sink 1: write events to the API Server (persisted in etcd)
    eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{
        Interface: clientset.CoreV1().Events(""),
    })

    // Sink 2: also log events to stdout (useful for debugging)
    eventBroadcaster.StartEventWatcher(func(e *corev1.Event) {
        klog.V(4).Infof("Event: %s %s %s", e.Type, e.Reason, e.Message)
    })

    // Create a recorder scoped to this controller
    // "my-canary-controller" will appear in Event.Source.Component
    recorder := eventBroadcaster.NewRecorder(
        scheme.Scheme,
        corev1.EventSource{Component: "my-canary-controller"},
    )

    return &Controller{
        recorder: recorder,
        ...
    }
}

Step 2 — Emit events during reconciliation

func (c *Controller) reconcile(key string) error {
    namespace, name, _ := cache.SplitMetaNamespaceKey(key)

    // Fetch the resource from local cache
    canary, err := c.canaryLister.Canaries(namespace).Get(name)
    if err != nil {
        return err
    }

    // --- Normal event: record a successful action ---
    c.recorder.Event(
        canary,           // InvolvedObject
        corev1.EventTypeNormal,
        "TrafficShifted", // Reason
        "Shifted 10% traffic to canary deployment",
    )

    // --- Warning event: record a problem ---
    if canary.Status.ErrorRate > threshold {
        c.recorder.Eventf(
            canary,
            corev1.EventTypeWarning,
            "HighErrorRate",
            "Canary error rate %.2f%% exceeds threshold %.2f%%, rolling back",
            canary.Status.ErrorRate,
            threshold,
        )
        return c.rollback(canary)
    }

    return nil
}

What the emitted Event looks like in kubectl

LAST SEEN   TYPE      REASON           OBJECT                  MESSAGE
30s         Normal    TrafficShifted   canary/my-app-canary    Shifted 10% traffic to canary deployment
10s         Warning   HighErrorRate    canary/my-app-canary    Canary error rate 5.23% exceeds threshold 1.00%, rolling back

6. Pod Lifecycle Events: The Most Important Events in Practice

Since Kubernetes is Pod-centric — Deployments, StatefulSets, DaemonSets, CronJobs all ultimately create Pods — Pod events are the most frequently used events in production operations.

Pod lifecycle → key events:

┌─────────────────────────────────────────────────────────────────┐
│  Phase          │ Reason           │ Type    │ Source           │
├─────────────────┼──────────────────┼─────────┼──────────────────┤
│  Scheduling     │ Scheduled        │ Normal  │ scheduler        │
│                 │ FailedScheduling │ Warning │ scheduler        │
├─────────────────┼──────────────────┼─────────┼──────────────────┤
│  Image pull     │ Pulling          │ Normal  │ kubelet          │
│                 │ Pulled           │ Normal  │ kubelet          │
│                 │ Failed           │ Warning │ kubelet          │
│                 │ ErrImagePull     │ Warning │ kubelet          │
├─────────────────┼──────────────────┼─────────┼──────────────────┤
│  Container      │ Created          │ Normal  │ kubelet          │
│  lifecycle      │ Started          │ Normal  │ kubelet          │
│                 │ Killing          │ Normal  │ kubelet          │
│                 │ BackOff          │ Warning │ kubelet          │
├─────────────────┼──────────────────┼─────────┼──────────────────┤
│  Node           │ Evicted          │ Warning │ kubelet          │
│                 │ OOMKilling       │ Warning │ kernel/kubelet   │
└─────────────────┴──────────────────┴─────────┴──────────────────┘

Practical use cases:

Slow startup diagnosis: Compare Pulling → Pulled timestamps to measure image pull time
Crash analysis: BackOff events with increasing intervals signal CrashLoopBackOff
Scheduling failures: FailedScheduling with reason message explains why a Pod is Pending
Eviction tracking: Evicted events identify nodes under memory pressure

7. Summary

EventBroadcaster flow:

recorder.Eventf(obj, type, reason, msg)
     │
     ▼
EventBroadcaster.ActionOrDrop(event)
     │
     ├── StartRecordingToSink()  → clientset.CoreV1().Events().Create/Patch
     │                             (aggregates duplicate events via Count++)
     │
     └── StartEventWatcher()    → custom handler (logging, metrics, alerting)

Concept	Detail
Event	A core API resource — stored in etcd, queryable via kubectl
Type	`Normal` (informational) or `Warning` (something needs attention)
Reason	Short machine-readable string — used for filtering and alerting
Count	Identical events are aggregated — prevents etcd flooding
Retention	Events are deleted 1 hour after last occurrence
EventRecorder	Your controller's interface for emitting events
EventBroadcaster	Fan-out hub — delivers events to all registered sinks
EventSink	Destination: API Server (etcd), stdout log, or custom handler

EventBroadcaster closes the observability loop for Kubernetes controllers. By emitting structured events at key decision points in your reconciliation logic, you give operators the same visibility into your custom controller that they have for built-in Kubernetes components — making your controller production-ready and debuggable.

Next in this series: Controller: The Central Hub of the Informer Framework (Part 7)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: WorkQueue — The Reliable Task Queue for Kubernetes Controllers

James Lee — Tue, 19 May 2026 10:08:43 +0000

In the previous article we saw how Indexer stores and indexes resource objects for fast local queries. Now we look at the other half of the controller pattern: WorkQueue — the component that bridges event callbacks and controller reconciliation workers.

Source: k8s.io/client-go/util/workqueue/

1. Where WorkQueue Fits

informer.AddEventHandler(ResourceEventHandlerFuncs{
    AddFunc:    func(obj) { queue.Add(keyOf(obj)) },
    UpdateFunc: func(old, new) { queue.Add(keyOf(new)) },
    DeleteFunc: func(obj) { queue.Add(keyOf(obj)) },
})
      │
      │  event fires → key enqueued
      ▼
┌─────────────────────────────────────────────────┐
│                  WorkQueue                      │
│  - deduplicates keys                            │
│  - rate limits re-enqueues on error             │
│  - supports multiple concurrent workers         │
└─────────────────────────────────────────────────┘
      │
      │  worker goroutine calls queue.Get()
      ▼
Controller reconciliation logic
├── success → queue.Done(key)
└── error   → queue.AddRateLimited(key)  ← retry with backoff

Key pattern: Event handlers only enqueue the key (e.g. "default/nginx"), never the full object. The worker fetches the latest object from Indexer at processing time — ensuring it always works with the most current state.

2. WorkQueue vs Plain FIFO

WorkQueue is not a simple queue. It adds critical properties that make it production-safe for controller development:

Property	Plain FIFO	WorkQueue
Ordered (FIFO)	✅	✅
Concurrent producers & consumers	❌	✅
Deduplication	❌	✅
Rate limiting on retry	❌	✅
Prometheus metrics	❌	✅
Graceful shutdown (SIGTERM)	❌	✅

Deduplication in detail

Event burst: Pod "nginx" Updated 5 times in 100ms

Plain FIFO:   [nginx, nginx, nginx, nginx, nginx]  ← processed 5 times
WorkQueue:    [nginx]                               ← processed once  ✅

Rule: if a key is already in the queue (not yet processed),
      subsequent Add() calls for the same key are silently dropped.
      The key is only re-eligible after Done() is called.

3. Three Queue Types

client-go ships three queue implementations, each building on the previous:

┌─────────────────────────────────────────────────────────┐
│  Type 1: FIFO Queue                                     │
│  k8s.io/client-go/util/workqueue/queue.go               │
│  Basic ordered queue with deduplication                 │
└─────────────────────────────┬───────────────────────────┘
                              │ extends
┌─────────────────────────────▼───────────────────────────┐
│  Type 2: Delaying Queue                                 │
│  k8s.io/client-go/util/workqueue/delaying_queue.go      │
│  Adds: AddAfter(item, duration) — insert after a delay  │
└─────────────────────────────┬───────────────────────────┘
                              │ extends
┌─────────────────────────────▼───────────────────────────┐
│  Type 3: RateLimiting Queue  ← used in custom controllers│
│  k8s.io/client-go/util/workqueue/rate_limiting_queue.go  │
│  Adds: AddRateLimited, Forget, NumRequeues               │
│  Uses a RateLimiter to compute delay before re-insert    │
└─────────────────────────────────────────────────────────┘

In practice, custom controllers always use the RateLimiting Queue.

4. The RateLimiting Queue Interface

// The rate limiter: decides HOW LONG to delay a re-enqueue
type RateLimiter interface {
    When(item interface{}) time.Duration  // return delay for this item
    Forget(item interface{})              // reset the item's failure history
    NumRequeues(item interface{}) int     // how many times has this item failed
}

// The queue interface: wraps DelayingInterface + rate limiting methods
type RateLimitingInterface interface {
    DelayingInterface                     // inherits Add, Get, Done, Len, etc.
    AddRateLimited(item interface{})      // enqueue with rate-limited delay
    Forget(item interface{})              // clear failure history for this item
    NumRequeues(item interface{}) int     // query failure count
}

Typical controller worker pattern

func (c *Controller) runWorker() {
    for c.processNextItem() {}
}

func (c *Controller) processNextItem() bool {
    // Block until an item is available
    key, quit := c.queue.Get()
    if quit {
        return false
    }
    defer c.queue.Done(key)   // MUST call Done() when finished

    err := c.reconcile(key.(string))
    if err == nil {
        // Success: clear the failure history
        c.queue.Forget(key)
        return true
    }

    if c.queue.NumRequeues(key) < maxRetries {
        // Transient error: re-enqueue with rate-limited delay
        c.queue.AddRateLimited(key)
        return true
    }

    // Too many retries: give up and clear history
    c.queue.Forget(key)
    utilruntime.HandleError(err)
    return true
}

5. Three Rate-Limiting Algorithms

Algorithm 1: Exponential Backoff (ItemExponentialFailureRateLimiter)

When the same item fails repeatedly, the delay grows exponentially with each failure.

Failure count → delay:
1st failure:  base delay × 2^0  =  5ms
2nd failure:  base delay × 2^1  = 10ms
3rd failure:  base delay × 2^2  = 20ms
4th failure:  base delay × 2^3  = 40ms
...
nth failure:  min(base × 2^(n-1), maxDelay)   ← capped at maxDelay

// Default values used in most controllers
workqueue.NewItemExponentialFailureRateLimiter(
    5*time.Millisecond,   // base delay
    1000*time.Second,     // max delay cap
)

Rate-limiting period: starts at AddRateLimited(), ends at Forget().

Timeline:
t=0ms    AddRateLimited("nginx")  → delay 5ms  → enqueued at t=5ms
t=5ms    Get("nginx") → reconcile → fails
t=5ms    AddRateLimited("nginx")  → delay 10ms → enqueued at t=15ms
t=15ms   Get("nginx") → reconcile → fails
t=15ms   AddRateLimited("nginx")  → delay 20ms → enqueued at t=35ms
...
t=Xms    Get("nginx") → reconcile → SUCCESS
         Forget("nginx")          → failure count reset to 0

Real-world example: Kubernetes Pod restart policy Always uses exponential backoff — that's why a crash-looping Pod shows increasing restart intervals (the familiar CrashLoopBackOff state).

Algorithm 2: Token Bucket (BucketRateLimiter)

The token bucket algorithm controls the overall throughput of the queue, regardless of which specific items are being enqueued.

┌─────────────────────────────────────────────────────────┐
│                    Token Bucket                         │
│                                                         │
│  ┌──┬──┬──┬──┬──┐                                       │
│  │🪙│🪙│🪙│🪙│🪙│  ← bucket (fixed capacity)            │
│  └──┴──┴──┴──┴──┘                                       │
│        ▲                    ▲                           │
│  tokens refilled        bucket full:                    │
│  at fixed rate r/s      stop refilling                  │
│                                                         │
│  Each AddRateLimited() consumes one token               │
│  No token available → item must wait                    │
└─────────────────────────────────────────────────────────┘

// Uses golang.org/x/time/rate under the hood
workqueue.NewMaxOfRateLimiter(
    workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second),
    &workqueue.BucketRateLimiter{
        Limiter: rate.NewLimiter(rate.Limit(10), 100),
        // 10 tokens/second refill rate, bucket capacity 100
    },
)

How it works:

Tokens are added to the bucket at a fixed rate (e.g. 10/sec)
Bucket has a maximum capacity — once full, new tokens are discarded
Each AddRateLimited() call requires one token
If no token is available, the item is delayed until a token is refilled

Used in: Nginx rate limiting, Envoy Ratelimit, Sentinel — token bucket is one of the most widely adopted rate-limiting algorithms in distributed systems.

Algorithm 3: Counter with Fast/Slow Lanes (ItemFastSlowRateLimiter)

The counter algorithm limits how many items can be enqueued within a time window. WorkQueue extends the basic counter with two insertion rates: a fast lane and a slow lane.

Configuration:
  fastDelay       = 5ms      ← interval between items in fast lane
  slowDelay       = 1000ms   ← interval between items in slow lane
  maxFastAttempts = 3        ← how many failures use fast lane before switching

Failure timeline for one item:
┌──────────────────────────────────────────────────────────┐
│  Attempt 1 → delay 5ms    (fast lane, attempt 1/3)       │
│  Attempt 2 → delay 5ms    (fast lane, attempt 2/3)       │
│  Attempt 3 → delay 5ms    (fast lane, attempt 3/3)       │
│  Attempt 4 → delay 1000ms (slow lane — maxFastAttempts   │
│                             exceeded)                    │
│  Attempt 5 → delay 1000ms (slow lane)                    │
│  ...                                                     │
└──────────────────────────────────────────────────────────┘

workqueue.NewItemFastSlowRateLimiter(
    5*time.Millisecond,   // fastDelay
    1000*time.Millisecond,// slowDelay
    3,                    // maxFastAttempts
)

Use case: Suitable for scenarios where a few quick retries are acceptable (transient network blips), but sustained failures should back off aggressively to avoid resource exhaustion.

6. Choosing a Rate Limiter

Algorithm	Best for	Behavior
Exponential Backoff	Per-item retry (most common)	Delay doubles with each failure; resets on success
Token Bucket	Global throughput cap	Smooth rate limit across all items; burst-friendly
Fast/Slow Counter	Tiered retry strategy	Quick retries first, then aggressive backoff

In practice, the default rate limiter used by controller-runtime and most Kubernetes controllers combines exponential backoff and token bucket via NewMaxOfRateLimiter — taking the maximum delay from both algorithms:

// Default: use whichever algorithm gives the LONGER delay
// This ensures both per-item backoff AND global rate cap are respected
workqueue.DefaultControllerRateLimiter()
// = NewMaxOfRateLimiter(
//     NewItemExponentialFailureRateLimiter(5ms, 1000s),
//     &BucketRateLimiter{rate.NewLimiter(rate.Limit(10), 100)},
//   )

7. Summary

WorkQueue lifecycle in a controller:

Event fires (Add/Update/Delete)
     │
     ▼
queue.Add(key)          ← deduplicated: same key added once
     │
     ▼
worker: queue.Get(key)  ← blocks until item available
     │
     ├── reconcile OK  → queue.Forget(key) + queue.Done(key)
     │
     └── reconcile ERR → queue.AddRateLimited(key)
                            │
                            └── RateLimiter.When(key) → delay
                                AddAfter(key, delay)  → DelayingQueue
                                → re-enters queue after delay

Component	Role
FIFO Queue	Base: ordered, deduplicated, concurrent-safe
Delaying Queue	Adds `AddAfter(item, duration)` for time-delayed insertion
RateLimiting Queue	Adds `AddRateLimited` — computes delay via RateLimiter before calling AddAfter
Exponential Backoff	Per-item delay that doubles on each failure — prevents hammering a broken resource
Token Bucket	Global throughput cap — prevents queue from overwhelming the API Server
Fast/Slow Counter	Tiered retry — fast initial retries, slow sustained retries

WorkQueue is the safety net of every Kubernetes controller. Its combination of deduplication, rate limiting, and retry semantics ensures that controllers converge to desired state reliably — even under API Server failures, network partitions, or event storms.

Next in this series: EventBroadcaster: How Kubernetes Records and Distributes Cluster Events (Part 6)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: Indexer — Fast In-Memory Resource Storage with Custom Indexing

James Lee — Tue, 19 May 2026 10:08:05 +0000

In the previous article we saw that HandleDeltas routes every event to two destinations — one of them is Indexer. Indexer is the local in-memory cache that makes resource queries fast and keeps the API Server load-free. In this article we'll look at how it's built and how to extend it with custom index functions.

Source: k8s.io/client-go/tools/cache/index.go

1. What Is Indexer?

Indexer is the local read cache of the Informer framework. Once the initial List completes and the cache is synced, all resource queries go through Indexer — never hitting the API Server.

┌─────────────────────────────────────────────────────┐
│                   Indexer                           │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │              ThreadSafeMap                    │  │
│  │  (concurrent-safe storage backend)            │  │
│  │                                               │  │
│  │  items: map[string]interface{}                │  │
│  │  key = "namespace/name"  value = resource obj │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │           Index Layer (on top)                │  │
│  │  Indexers: map[indexName]IndexFunc            │  │
│  │  Indices:  map[indexName]Index                │  │
│  │  Index:    map[indexKey]sets.String           │  │
│  └───────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────┘

Indexer has two layers:

ThreadSafeMap — the raw concurrent-safe key/value store
Index layer — pluggable custom index functions for efficient filtered queries

2. ThreadSafeMap: The Storage Backend

Source: k8s.io/client-go/tools/cache/thread_safe_store.go

type threadSafeMap struct {
    lock     sync.RWMutex          // read-write lock for concurrent safety
    items    map[string]interface{} // primary storage: key → resource object
    indexers Indexers               // registered index functions
    indices  Indices                // computed index data
}

Key format

The default key function is MetaNamespaceKeyFunc, which produces keys in the format:

namespace/name    →  "default/nginx-pod"
cluster-scoped    →  "my-node"   (no namespace prefix)

CRUD operations

ThreadSafeMap exposes standard storage operations, all protected by sync.RWMutex:

Operation	Lock type	What it does
`Add(key, obj)`	Write	Insert object + update all index entries
`Update(key, obj)`	Write	Replace object + rebuild index entries
`Delete(key)`	Write	Remove object + clean up index entries
`Get(key)`	Read	Fetch single object by key
`List()`	Read	Return all objects
`ListKeys()`	Read	Return all keys
`ByIndex(indexName, indexKey)`	Read	Return all objects matching an index query

Write path (Add/Update/Delete):
┌─────────────────────────────────────────────────┐
│  lock.Lock()                                    │
│  ├── modify items map                           │
│  ├── updateIndices() or deleteFromIndices()     │
│  └── lock.Unlock()                              │
└─────────────────────────────────────────────────┘

Read path (Get/List/ByIndex):
┌─────────────────────────────────────────────────┐
│  lock.RLock()   ← multiple readers in parallel  │
│  ├── read from items or indices                 │
│  └── lock.RUnlock()                             │
└─────────────────────────────────────────────────┘

3. The Four Index Data Structures

Indexer's power comes from its pluggable index system. Four types work together:

// A registry of named index functions
type Indexers map[string]IndexFunc

// An index function: given an object, return the index keys it belongs to
type IndexFunc func(obj interface{}) ([]string, error)

// A registry of computed index data (one per registered IndexFunc)
type Indices map[string]Index

// The actual index: maps an index key to the set of object keys that match
type Index map[string]sets.String

How they relate

Indexers["byNamespace"] = namespaceIndexFunc
Indexers["byLabel"]     = labelIndexFunc
      │
      │  when an object is added/updated, each IndexFunc is called
      ▼
Indices["byNamespace"] = Index{
    "default":     {"default/nginx", "default/redis"},
    "kube-system": {"kube-system/coredns"},
}
Indices["byLabel"] = Index{
    "app=web":   {"default/nginx", "default/frontend"},
    "app=cache": {"default/redis"},
}
      │
      │  query: ByIndex("byLabel", "app=web")
      ▼
returns: [nginx object, frontend object]   ← O(1) lookup via set

Visual summary

┌──────────────────────────────────────────────────────────────┐
│  Indexers (function registry)                                │
│  ┌────────────────┬──────────────────────────────────────┐   │
│  │ "byNamespace"  │ func(obj) → [namespace]              │   │
│  │ "byLabel/foo"  │ func(obj) → [label value of "foo"]   │   │
│  └────────────────┴──────────────────────────────────────┘   │
│                          │ applied on every Add/Update        │
│                          ▼                                    │
│  Indices (computed data)                                      │
│  ┌────────────────┬──────────────────────────────────────┐   │
│  │ "byNamespace"  │ {"default": {"ns/pod1","ns/pod2"}}   │   │
│  │ "byLabel/foo"  │ {"bar": {"ns/pod1"}, "biz":{"ns/pod3"}}  │
│  └────────────────┴──────────────────────────────────────┘   │
└──────────────────────────────────────────────────────────────┘

4. Custom Index Functions in Practice

The real power of Indexer is that you can register any index function at initialization time. Here's the complete example from the client-go test suite:

// Custom index function: index Pods by the value of their "foo" label
func testIndexFunc(obj interface{}) ([]string, error) {
    pod := obj.(*v1.Pod)
    return []string{pod.Labels["foo"]}, nil
}

func TestGetIndexFuncValues(t *testing.T) {
    // Initialize Indexer with:
    //   key function:   MetaNamespaceKeyFunc  → "namespace/name"
    //   index function: testIndexFunc         → registered as "testmodes"
    index := NewIndexer(MetaNamespaceKeyFunc, Indexers{"testmodes": testIndexFunc})

    pod1 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{
        Name: "one", Labels: map[string]string{"foo": "bar"}}}
    pod2 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{
        Name: "two", Labels: map[string]string{"foo": "bar"}}}
    pod3 := &v1.Pod{ObjectMeta: metav1.ObjectMeta{
        Name: "tre", Labels: map[string]string{"foo": "biz"}}}

    index.Add(pod1)
    index.Add(pod2)
    index.Add(pod3)

    // List all distinct values produced by the "testmodes" index function
    keys := index.ListIndexFuncValues("testmodes")
    // keys = ["bar", "biz"]  — the two distinct label values

    // Query all Pods where label "foo" == "bar"
    pods, _ := index.ByIndex("testmodes", "bar")
    // pods = [pod1, pod2]
}

What happens internally when `index.Add(pod1)` is called

index.Add(pod1)
     │
     ├── items["default/one"] = pod1        ← store in primary map
     │
     └── for each registered IndexFunc:
           testIndexFunc(pod1) → ["bar"]
           Indices["testmodes"]["bar"].Insert("default/one")

After adding all 3 pods:

items = {
    "default/one": pod1,
    "default/two": pod2,
    "default/tre": pod3,
}

Indices["testmodes"] = {
    "bar": {"default/one", "default/two"},
    "biz": {"default/tre"},
}

Query flow: `ByIndex("testmodes", "bar")`

ByIndex("testmodes", "bar")
     │
     ├── look up Indices["testmodes"]["bar"]
     │   → set: {"default/one", "default/two"}
     │
     └── for each key in set:
           items["default/one"] → pod1
           items["default/two"] → pod2

Result: [pod1, pod2]   ← O(1) index lookup + O(k) object fetch

5. Built-in Index Functions

client-go ships with a standard index function for the most common query pattern:

// MetaNamespaceIndexFunc — index objects by namespace
// This is the default index registered in most Informers
func MetaNamespaceIndexFunc(obj interface{}) ([]string, error) {
    meta, err := meta.Accessor(obj)
    if err != nil {
        return []string{""}, fmt.Errorf("object has no meta: %v", err)
    }
    return []string{meta.GetNamespace()}, nil
}

Usage:

// Get all Pods in the "production" namespace — from local cache, no API call
pods, err := podsLister.Pods("production").List(labels.Everything())

// Internally this calls:
// indexer.ByIndex("namespace", "production")

6. Indexer vs Direct API Query

	Indexer (local cache)	Direct API Server call
Speed	Microseconds (in-memory)	Milliseconds (network + etcd)
API Server load	Zero	One request per query
Consistency	Eventually consistent (synced via Watch)	Strongly consistent
Best for	Controller reconciliation loops	One-time admin queries

Rule of thumb: In any controller hot path — reconciliation loops, event handlers, health checks — always use the Lister (backed by Indexer). Only use Clientset for writes (Create/Update/Delete/Patch).

7. Summary

Indexer = ThreadSafeMap + pluggable Index functions

┌──────────────────────────────────────────────────────────┐
│  Write path (from HandleDeltas):                         │
│  Add/Update/Delete → update items + rebuild index data   │
│                                                          │
│  Read path (from your controller via Lister):            │
│  Get(key)              → O(1) direct lookup              │
│  List()                → O(n) full scan                  │
│  ByIndex(name, value)  → O(1) index lookup + O(k) fetch  │
└──────────────────────────────────────────────────────────┘

Concept	Detail
ThreadSafeMap	`sync.RWMutex` + `map[string]interface{}` — the raw storage layer
MetaNamespaceKeyFunc	Default key function: produces `"namespace/name"` keys
Indexers	Registry of named index functions — define what you can query by
IndexFunc	User-defined function: `obj → []indexKey`
Indices	Computed index data: `indexName → indexKey → set of object keys`
ByIndex	Efficient filtered query: O(1) index lookup + O(k) object retrieval

Indexer is what makes Kubernetes controllers fast. By keeping a fully indexed in-memory snapshot of cluster state, it eliminates the need for controllers to poll the API Server — turning what would be thousands of network calls per second into zero-latency local memory reads.

Next in this series: WorkQueue: The Reliable Task Queue for Kubernetes Controllers (Part 5)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: DeltaFIFO — The Event Queue Behind Informer

James Lee — Tue, 19 May 2026 10:07:34 +0000

In the previous article we saw how Reflector calls List/Watch and feeds events downstream. The immediate destination of those events is DeltaFIFO — the event queue that sits at the heart of the Informer framework.

Source: k8s.io/client-go/tools/cache/delta_fifo.go

1. What Is DeltaFIFO?

The name has two parts:

Part	Meaning
FIFO	First-In-First-Out queue — supports `Add`, `Update`, `Delete`, `List`, `Pop` operations
Delta	Each entry stores not just the object, but also the type of change (Added / Updated / Deleted / Sync / Replaced)

Together, DeltaFIFO is a queue where each item carries both what changed and how it changed — giving downstream consumers the full context they need to act correctly.

2. The Data Structure

// k8s.io/client-go/tools/cache/delta_fifo.go
type DeltaFIFO struct {
    lock sync.RWMutex
    cond sync.Cond             // used to block/unblock consumers

    items map[string]Deltas    // key → list of deltas for that object
    queue []string             // ordered list of keys (FIFO ordering)

    populated              bool
    initialPopulationCount int  // tracks how many items came from the initial List

    keyFunc      KeyFunc         // computes the key for a given object
    knownObjects KeyListerGetter // reference to Indexer (for deletion detection)

    closed                bool
    emitDeltaTypeReplaced bool
}

type Deltas []Delta

type Delta struct {
    Type   DeltaType    // Added | Updated | Deleted | Replaced | Sync
    Object interface{}  // the actual Kubernetes resource object
}

The Three Core Fields

queue: ["pod/default/nginx", "pod/default/redis", "deploy/default/app"]
         ↑ ordered slice of object keys (FIFO order)

items: {
  "pod/default/nginx":  [ {Added, PodObj_v1}, {Updated, PodObj_v2} ],
  "pod/default/redis":  [ {Deleted, PodObj_v1} ],
  "deploy/default/app": [ {Added, DeployObj_v1} ],
}
         ↑ map: key → []Delta (all pending changes for that object)

Delta: { Type: "Updated", Object: <Pod nginx v2> }
         ↑ a single change event: what happened + the object snapshot

Visualizing the Structure

┌─────────────────────────────────────────────────────────────┐
│                        DeltaFIFO                            │
│                                                             │
│  queue (FIFO order):                                        │
│  ┌──────────────────┬──────────────────┬─────────────────┐  │
│  │ "pod/default/    │ "pod/default/    │ "deploy/default │  │
│  │  nginx"          │  redis"          │  /app"          │  │
│  └──────────────────┴──────────────────┴─────────────────┘  │
│          │                   │                  │            │
│          ▼                   ▼                  ▼            │
│  items (map):                                               │
│  ┌──────────────────┐ ┌────────────────┐ ┌──────────────┐  │
│  │ [{Added, nginx}  │ │[{Deleted,      │ │[{Added,      │  │
│  │  {Updated,nginx}]│ │  redis}]       │ │  deploy}]    │  │
│  └──────────────────┘ └────────────────┘ └──────────────┘  │
└─────────────────────────────────────────────────────────────┘

Key design insight: queue maintains ordering (which object to process next), while items accumulates all pending deltas per object. Multiple changes to the same object are batched — but never reordered relative to other objects.

3. Producing Messages: queueActionLocked

Every time Reflector calls syncWith() (from List) or watchHandler() (from Watch), it ultimately calls queueActionLocked to enqueue an event:

func (f *DeltaFIFO) queueActionLocked(actionType DeltaType, obj interface{}) error {
    // ① Compute the object's key (e.g. "default/nginx")
    id, err := f.KeyOf(obj)
    if err != nil {
        return KeyError{obj, err}
    }

    // ② Append new Delta to the existing list for this key
    oldDeltas := f.items[id]
    newDeltas := append(oldDeltas, Delta{actionType, obj})

    // ③ Deduplicate consecutive identical events
    newDeltas = dedupDeltas(newDeltas)

    if len(newDeltas) > 0 {
        // ④ If this key is new, add it to the queue (maintain FIFO order)
        if _, exists := f.items[id]; !exists {
            f.queue = append(f.queue, id)
        }
        // ⑤ Update items map with the new delta list
        f.items[id] = newDeltas

        // ⑥ Wake up all blocked consumers
        f.cond.Broadcast()
    }
    return nil
}

Production flow:

Reflector (r.syncWith / r.watchHandler)
     │
     ▼
queueActionLocked(actionType, obj)
     │
     ├── ① KeyOf(obj)              → compute key (e.g. "default/nginx")
     ├── ② append Delta            → add {type, obj} to existing delta list
     ├── ③ dedupDeltas()           → remove redundant consecutive events
     ├── ④ queue = append(key)     → add key to FIFO queue (if new)
     ├── ⑤ items[key] = newDeltas  → store updated delta list
     └── ⑥ cond.Broadcast()        → wake up Pop() consumers

Delta Types

const (
    Added    DeltaType = "Added"     // object newly appeared (from Watch or initial List)
    Updated  DeltaType = "Updated"   // object was modified
    Deleted  DeltaType = "Deleted"   // object was removed
    Replaced DeltaType = "Replaced"  // full re-list replaced the object (resync)
    Sync     DeltaType = "Sync"      // periodic resync — object unchanged but re-delivered
)

Deduplication: dedupDeltas

dedupDeltas prevents redundant back-to-back Deleted events for the same object:

Before dedup:  [{Updated, obj}, {Deleted, obj}, {Deleted, obj}]
After dedup:   [{Updated, obj}, {Deleted, obj}]

Only the last two deltas are considered for deduplication — older history is always preserved.

4. Consuming Messages: Pop

The consumer side is driven by the Informer's internal Controller, which calls Pop() in its processLoop:

func (f *DeltaFIFO) Pop(process PopProcessFunc) (interface{}, error) {
    f.lock.Lock()
    defer f.lock.Unlock()

    for {
        // ① Block if queue is empty — wait for producer to signal
        for len(f.queue) == 0 {
            if f.closed {
                return nil, ErrFIFOClosed
            }
            f.cond.Wait()   // releases lock and sleeps until cond.Broadcast()
        }

        // ② Dequeue the first key (FIFO order)
        id := f.queue[0]
        f.queue = f.queue[1:]

        if f.initialPopulationCount > 0 {
            f.initialPopulationCount--
        }

        // ③ Retrieve and remove the delta list for this key
        item, ok := f.items[id]
        if !ok {
            continue  // key was already processed (race condition guard)
        }
        delete(f.items, id)

        // ④ Pass the delta list to the callback function (HandleDeltas)
        err := process(item)

        // ⑤ If processing failed, re-enqueue for retry
        if e, ok := err.(ErrRequeue); ok {
            f.addIfNotPresent(id, item)
            err = e.Err
        }

        return item, err
    }
}

Consumption flow:

Controller.processLoop()
     │
     └── DeltaFIFO.Pop(HandleDeltas)
           │
           ├── ① cond.Wait()       → sleep if queue empty
           ├── ② dequeue key       → take first key from FIFO queue
           ├── ③ delete(items,id)  → remove delta list from map
           ├── ④ process(item)     → call HandleDeltas callback
           └── ⑤ on error:
                 addIfNotPresent() → re-enqueue for retry

5. HandleDeltas: Routing Events Downstream

Once Pop() delivers a delta list to the callback, HandleDeltas routes each event to two destinations:

// k8s.io/client-go/tools/cache/shared_informer.go
func (s *sharedIndexInformer) HandleDeltas(obj interface{}) error {
    s.blockDeltas.Lock()
    defer s.blockDeltas.Unlock()

    for _, d := range obj.(Deltas) {
        switch d.Type {

        case Sync, Replaced, Added, Updated:
            if old, exists, err := s.indexer.Get(d.Object); err == nil && exists {
                // Object already in cache → Update
                s.indexer.Update(d.Object)

                isSync := (d.Type == Sync) ||
                    (d.Type == Replaced && sameResourceVersion(d.Object, old))

                // Route to ShareInformer listeners
                s.processor.distribute(updateNotification{oldObj: old, newObj: d.Object}, isSync)
            } else {
                // Object not in cache → Add
                s.indexer.Add(d.Object)
                s.processor.distribute(addNotification{newObj: d.Object}, false)
            }

        case Deleted:
            s.indexer.Delete(d.Object)
            s.processor.distribute(deleteNotification{oldObj: d.Object}, false)
        }
    }
    return nil
}

The Two Destinations

HandleDeltas(Deltas)
     │
     ├── ① Indexer (local cache)
     │     ├── Added/Updated/Replaced/Sync → indexer.Add() or indexer.Update()
     │     └── Deleted                     → indexer.Delete()
     │
     └── ② processor.distribute() (ShareInformer fan-out)
           ├── addNotification    → triggers OnAdd  callbacks
           ├── updateNotification → triggers OnUpdate callbacks
           └── deleteNotification → triggers OnDelete callbacks
                      │
                      ▼
             informer.AddEventHandler(ResourceEventHandlerFuncs{
                 AddFunc:    func(obj) { workqueue.Add(key) },
                 UpdateFunc: func(old, new) { workqueue.Add(key) },
                 DeleteFunc: func(obj) { workqueue.Add(key) },
             })

6. The Complete DeltaFIFO Lifecycle

┌─────────────────────────────────────────────────────────────────┐
│                     DeltaFIFO Lifecycle                         │
│                                                                 │
│  PRODUCER (Reflector)                                           │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  r.syncWith()      → queueActionLocked(Added, obj)       │   │
│  │  r.watchHandler()  → queueActionLocked(Updated/Deleted…) │   │
│  └──────────────────────────────────────────────────────────┘   │
│                          │ cond.Broadcast()                     │
│                          ▼                                      │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │                    DeltaFIFO                             │   │
│  │  queue: [key1, key2, key3, ...]   (FIFO order)           │   │
│  │  items: {key1: [Δ,Δ], key2: [Δ], key3: [Δ,Δ,Δ]}        │   │
│  └──────────────────────────────────────────────────────────┘   │
│                          │ cond.Wait → unblock                  │
│                          ▼                                      │
│  CONSUMER (Controller.processLoop)                              │
│  ┌──────────────────────────────────────────────────────────┐   │
│  │  Pop(HandleDeltas)                                       │   │
│  │    → HandleDeltas                                        │   │
│  │        ├── Indexer.Add/Update/Delete  (local cache)      │   │
│  │        └── processor.distribute()    (event callbacks)   │   │
│  └──────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────┘

7. Summary

Concept	Detail
queue	Ordered slice of object keys — guarantees FIFO processing order
items	Map of key → `[]Delta` — accumulates all pending changes per object
Delta	`{Type, Object}` — carries both the change type and the full object snapshot
queueActionLocked	Producer entry point — appends delta, deduplicates, signals consumers
dedupDeltas	Removes redundant back-to-back Deleted events
Pop	Consumer entry point — blocks when empty, dequeues FIFO, calls HandleDeltas
HandleDeltas	Routes each delta to Indexer (storage) AND processor.distribute (callbacks)
cond.Broadcast/Wait	Efficient producer-consumer synchronization — no busy polling

DeltaFIFO is the reliable handoff point between Reflector (producer) and the Informer's internal Controller (consumer). Its dual structure — an ordered queue for sequencing plus a map for delta accumulation — ensures that no event is lost, no object is processed out of order, and downstream consumers always receive the full change context.

Next in this series: Indexer: Fast In-Memory Resource Storage with Thread-Safe Indexing (Part 4)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

client-go Deep Dive: Reflector — How Kubernetes Syncs Resources from the API Server

James Lee — Tue, 19 May 2026 10:06:09 +0000

In the previous article we saw that Informer's core mechanism is List/Watch. The component responsible for executing that mechanism is Reflector. Every time an Informer starts up, it's Reflector that connects to the API Server, fetches the full resource snapshot, and then keeps watching for changes.

Source: k8s.io/client-go/tools/cache/reflector.go

1. Where Reflector Fits

Informer starts
     │
     ▼
┌──────────────────────────────────────────────────┐
│                   Reflector                      │
│                                                  │
│  Phase 1: LIST                                   │
│  ├── call listerWatcher.List()                   │
│  ├── get ResourceVersion                         │
│  ├── extract object list                         │
│  └── syncWith() → store all objects in DeltaFIFO │
│                                                  │
│  Phase 2: WATCH                                  │
│  ├── call listerWatcher.Watch()                  │
│  ├── receive incremental events (HTTP chunked)   │
│  └── watchHandler() → push events to DeltaFIFO  │
└──────────────────────────────────────────────────┘
     │
     ▼
DeltaFIFO (event queue)

2. The Reflector Struct

// k8s.io/client-go/tools/cache/reflector.go
type Reflector struct {
    name             string
    expectedTypeName string
    expectedType     reflect.Type          // the resource type being watched (e.g. *v1.Pod)
    expectedGVK      *schema.GroupVersionKind

    store            Store                 // DeltaFIFO — where events are written
    listerWatcher    ListerWatcher         // the actual List/Watch implementation

    backoffManager         wait.BackoffManager  // retry backoff for Watch reconnects
    initConnBackoffManager wait.BackoffManager  // backoff for initial connection

    resyncPeriod     time.Duration         // how often to force a full resync (0 = never)
    ShouldResync     func() bool           // optional: custom resync decision function

    clock            clock.Clock
    paginatedResult  bool

    // ResourceVersion tracking — critical for Watch correctness
    lastSyncResourceVersion              string
    isLastSyncResourceVersionUnavailable bool
    lastSyncResourceVersionMutex         sync.RWMutex

    WatchListPageSize  int64
    watchErrorHandler  WatchErrorHandler
}

Key fields explained:

Field	Purpose
`store`	The DeltaFIFO queue — Reflector writes all events here
`listerWatcher`	Interface that provides the actual `List()` and `Watch()` calls
`lastSyncResourceVersion`	Tracks the latest version seen — Watch resumes from this point after reconnect
`resyncPeriod`	If > 0, Reflector periodically forces a full re-List to catch any missed events
`backoffManager`	Exponential backoff when Watch connection drops

3. The ListerWatcher Interface

listerWatcher is an interface — Reflector doesn't know or care which resource type it's watching. The concrete implementation is injected per resource type.

type ListerWatcher interface {
    Lister
    Watcher
}

type Lister interface {
    List(options metav1.ListOptions) (runtime.Object, error)
}

type Watcher interface {
    Watch(options metav1.ListOptions) (watch.Interface, error)
}

For a Pod Informer, the concrete implementation looks like this:

// k8s.io/client-go/informers/core/v1/pods.go
func NewFilteredPodInformer(
    client kubernetes.Interface,
    namespace string,
    resyncPeriod time.Duration,
    indexers cache.Indexers,
    tweakListOptions internalinterfaces.TweakListOptionsFunc,
) cache.SharedIndexInformer {
    return cache.NewSharedIndexInformer(
        &cache.ListWatch{
            // List: calls the real Kubernetes API
            ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.CoreV1().Pods(namespace).List(context.TODO(), options)
            },
            // Watch: opens a long-lived HTTP connection to the API Server
            WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
                if tweakListOptions != nil {
                    tweakListOptions(&options)
                }
                return client.CoreV1().Pods(namespace).Watch(context.TODO(), options)
            },
        },
        &corev1.Pod{},
        resyncPeriod,
        indexers,
    )
}

This is why Clientset must be created before SharedInformerFactory. The ListFunc and WatchFunc are closures over the Clientset — without it, Reflector has no way to talk to the API Server.

func NewKubeController(...) *KubeController {
    kc := &KubeController{clientset: clientset}

    // Clientset is passed in here — it flows down into every ListFunc/WatchFunc
    kc.factory = informers.NewSharedInformerFactory(clientset, defaultResync)

    kc.podInformer      = kc.factory.Core().V1().Pods()
    kc.podsLister       = kc.podInformer.Lister()
    kc.podsSynced       = kc.podInformer.Informer().HasSynced
    kc.deploymentInformer   = kc.factory.Apps().V1().Deployments()
    kc.deploymentsLister    = kc.deploymentInformer.Lister()
    kc.deploymentsSynced    = kc.deploymentInformer.Informer().HasSynced
    return kc
}

4. ListAndWatch: Phase 1 — Full List

The ListAndWatch method is the heart of Reflector. Phase 1 fetches the complete current state:

func (r *Reflector) ListAndWatch(stopCh <-chan struct{}) error {
    var resourceVersion string
    options := metav1.ListOptions{ResourceVersion: r.relistResourceVersion()}

    if err := func() error {
        // ① Fetch all resources (with pagination support)
        go func() {
            pager := pager.New(pager.SimplePageFunc(func(opts metav1.ListOptions) (runtime.Object, error) {
                return r.listerWatcher.List(opts)   // → calls ListFunc → Clientset.CoreV1().Pods().List()
            }))
            list, paginatedResult, err = pager.List(context.Background(), options)
            // handle expired ResourceVersion: retry with fresh version
            if isExpiredError(err) || isTooLargeResourceVersionError(err) {
                list, paginatedResult, err = pager.List(context.Background(),
                    metav1.ListOptions{ResourceVersion: r.relistResourceVersion()})
            }
            close(listCh)
        }()

        // wait for list to complete or stop signal
        select {
        case <-stopCh:   return nil
        case r := <-panicCh: panic(r)
        case <-listCh:
        }

        // ② Extract ResourceVersion from the list response
        listMetaInterface, _ := meta.ListAccessor(list)
        resourceVersion = listMetaInterface.GetResourceVersion()

        // ③ Convert list result into a slice of runtime.Object
        items, _ := meta.ExtractList(list)

        // ④ Store all objects + ResourceVersion into DeltaFIFO
        r.syncWith(items, resourceVersion)

        return nil
    }(); err != nil {
        return err
    }

    // ⑤ Record the latest ResourceVersion — Watch will resume from here
    r.setLastSyncResourceVersion(resourceVersion)
    ...
}

Phase 1 call chain:

ListAndWatch()
  │
  ├── ① r.listerWatcher.List()        → fetch all objects from API Server
  ├── ② listMetaInterface.GetResourceVersion()  → extract version stamp
  ├── ③ meta.ExtractList()            → convert to []runtime.Object
  ├── ④ r.syncWith()                  → write all objects into DeltaFIFO
  └── ⑤ r.setLastSyncResourceVersion() → save version for Watch resume point

5. ListAndWatch: Phase 2 — Continuous Watch

After the full List completes, Reflector enters an infinite loop watching for incremental changes:

    // Periodic resync goroutine (if resyncPeriod > 0)
    go func() {
        for {
            select {
            case <-resyncCh:
                if r.ShouldResync == nil || r.ShouldResync() {
                    r.store.Resync()   // force re-sync of Indexer from DeltaFIFO
                }
            case <-stopCh:
                return
            }
        }
    }()

    // Main Watch loop
    for {
        select {
        case <-stopCh:
            return nil
        default:
        }

        // Add randomized timeout to prevent thundering herd on reconnect
        timeoutSeconds := int64(minWatchTimeout.Seconds() * (rand.Float64() + 1.0))
        options = metav1.ListOptions{
            ResourceVersion:     resourceVersion,  // resume from last known version
            TimeoutSeconds:      &timeoutSeconds,
            AllowWatchBookmarks: true,
        }

        // ① Open a long-lived Watch connection to the API Server
        w, err := r.listerWatcher.Watch(options)
        if err != nil {
            if utilnet.IsConnectionRefused(err) {
                <-r.initConnBackoffManager.Backoff().C()  // backoff and retry
                continue
            }
            return err
        }

        // ② Process incoming events: write to DeltaFIFO + update ResourceVersion
        if err := r.watchHandler(start, w, &resourceVersion, resyncerrc, stopCh); err != nil {
            if !isExpiredError(err) {
                // log unexpected errors
            }
            return nil
        }
    }

Phase 2 call chain:

ListAndWatch() — Watch loop
  │
  ├── ① r.listerWatcher.Watch()    → open HTTP long-poll to API Server
  │                                   (passes ResourceVersion to resume from)
  └── ② r.watchHandler()           → for each incoming event:
                                       write object to DeltaFIFO
                                       update lastSyncResourceVersion

6. ResourceVersion: The Key to Reliable Watch

ResourceVersion is a monotonically increasing version stamp that etcd assigns to every resource object. It's the mechanism that makes Watch reliable:

List response:  ResourceVersion = "1000"
                (all objects as of version 1000)
     │
     ▼
Watch request:  ResourceVersion = "1000"
                (give me all events AFTER version 1000)
     │
     ▼
API Server streams:
  - Pod "nginx" Updated  (RV=1001)
  - Pod "redis" Deleted  (RV=1002)
  - ...

If Watch connection drops:
     │
     ▼
Reflector reconnects with:  ResourceVersion = "1002"
(resumes exactly where it left off — no events missed, no duplicates)

Scenario	ResourceVersion behavior
Fresh start	`""` or `"0"` — get latest snapshot
After List	Set to the version returned by List response
After each Watch event	Updated to the event's ResourceVersion
Watch reconnect	Resumes from `lastSyncResourceVersion`
Expired version (too old)	Triggers a fresh List from scratch

7. Under the Hood: How Watch Works (HTTP Chunked Transfer)

The Watch connection is not a WebSocket or gRPC stream — it's a plain HTTP/1.1 long-lived connection using Chunked Transfer Encoding.

Client (Reflector)                    API Server
     │                                     │
     │  GET /api/v1/pods?watch=true        │
     │  ResourceVersion=1000               │
     │ ──────────────────────────────────► │
     │                                     │
     │  HTTP/1.1 200 OK                    │
     │  Transfer-Encoding: chunked         │
     │ ◄────────────────────────────────── │
     │                                     │
     │  chunk: {"type":"ADDED","object":…} │
     │ ◄────────────────────────────────── │
     │                                     │
     │  chunk: {"type":"MODIFIED","object":…}
     │ ◄────────────────────────────────── │
     │                                     │
     │  (connection stays open…)           │
     │  (new chunks arrive as events occur)│

Why Chunked Transfer Encoding?

In standard HTTP, the Content-Length header tells the client how many bytes to expect. But for a Watch stream, the server doesn't know in advance how many events will occur — the stream is open-ended.

Chunked Transfer Encoding solves this:

The response body is split into variable-size chunks
Each chunk is prefixed with its size in hex
The server sends chunks as events arrive — no pre-declared content length needed
A zero-length chunk signals the end of the stream

HTTP/1.1 200 OK
Transfer-Encoding: chunked

1a\r\n                          ← chunk size (hex)
{"type":"ADDED","object":…}\r\n ← chunk data
\r\n
2f\r\n
{"type":"MODIFIED","object":…}\r\n
\r\n
0\r\n                           ← end of stream
\r\n

This is only available in HTTP/1.1 and later. It's what makes Kubernetes Watch efficient — a single persistent connection replaces thousands of polling requests.

8. Summary

Reflector.Run()
     │
     └── ListAndWatch()
           │
           ├── Phase 1: LIST
           │     ├── listerWatcher.List()          → full snapshot from API Server
           │     ├── GetResourceVersion()           → stamp the snapshot version
           │     ├── meta.ExtractList()             → parse into objects
           │     └── syncWith() → DeltaFIFO         → seed the local cache
           │
           └── Phase 2: WATCH (infinite loop)
                 ├── listerWatcher.Watch(RV)        → HTTP chunked long-poll
                 ├── watchHandler() → DeltaFIFO     → stream events to queue
                 └── on disconnect: backoff + retry
                     on expired RV: re-List from scratch

Concept	Detail
ListerWatcher	Interface injected per resource type; backed by Clientset API calls
ResourceVersion	etcd version stamp; enables Watch to resume without missing events
DeltaFIFO	The downstream queue; Reflector is its sole producer
Chunked Transfer	HTTP/1.1 mechanism that keeps the Watch connection open indefinitely
Backoff	Exponential retry when Watch connection drops or API Server is unavailable

Reflector is the bridge between the Kubernetes API Server and the local Informer cache. Once you understand its List → syncWith → Watch → watchHandler pipeline, the reliability guarantees of the entire Informer framework become clear.

Next in this series: DeltaFIFO: The Event Queue Behind Informer (Part 3)

Follow the series for more deep dives into Kubernetes development.

📦 Source Code: github.com/muzinan123/operator

Forem: James Lee

Full Observability in Istio: Metrics with Prometheus/Grafana + Distributed Tracing with Jaeger

Part 1: Metrics with Prometheus + Grafana

Why Prometheus Over StatsD or InfluxDB?

Prometheus Architecture

Prometheus Data Model

The Four Metric Types

Grafana Dashboard: What to Look At

Services Dashboard

Workload Dashboard

Part 2: Distributed Tracing with Jaeger

The Three Pillars of Observability — Compared

How Distributed Tracing Works (Dapper Model)

A Real Trace Record

Jaeger Architecture

The One Thing Your Code Must Do

What Jaeger Shows You

Summary: Metrics vs Tracing — When to Use Which

Traffic Management in Istio: Circuit Breaking & Rate Limiting with DestinationRule

The Entry Point: TrafficPolicy in DestinationRule

Part 1: Rate Limiting with connectionPool

TCP Settings

HTTP Settings

Full connectionPool Example

Part 2: Circuit Breaking with outlierDetection

How It Works

Configuration Fields

Full outlierDetection Example

Key Design Decisions Worth Understanding

1. Progressive Ejection Time

2. The Safety Net: maxEjectionPercent + minHealthPercent

Putting It All Together

Summary

Istio & Envoy Service Mesh Architecture: How the Control Plane and Data Plane Work Together

Part 1: How Pilot and Envoy Work Together

Part 2: The Four Core Resources in Envoy

1. Listener (LDS)

2. Cluster (CDS + EDS)

3. Route (RDS)

4. Filter (Envoy Filter)

Full Architecture: Putting It All Together

Why This Architecture Matters

Summary

xDS Protocol Deep Dive: The Universal Control Plane API Behind Envoy and Istio

What is xDS?

The Four Core Resources in Envoy

1. LDS — Listener Discovery Service

2. RDS — Route Discovery Service

3. CDS — Cluster Discovery Service

4. EDS — Endpoint Discovery Service

5. SDS — Secret Discovery Service

How xDS Works: gRPC Streaming Subscription

Request Flow for a Typical HTTP Route

Full Subscription vs. Delta (Incremental) Subscription

xDS Protocol Deep Dive

Request Message Example

Response Message Example

ACK and NACK

Resource Update & The Nonce Problem

Summary

client-go Deep Dive: Operator in Practice — Building a Canary Release Controller

1. What Is an Operator?

2. Project Structure

3. Generating the Typed Client Code

4. Controller Struct & Wiring

5. Run → Worker → syncHandler

6. The Three Release Strategies

NormalDeploy — Rolling Update

CanaryDeploy — Batched Release

CanaryRollback — Reverse the Canary

7. Status Reporting: updateCanaryStatus

8. Complete Execution Flow

9. Key Design Principles

Series Complete 🎉

client-go Deep Dive: CRD — Extending the Kubernetes API with Custom Resources

1. The Kubernetes Declarative Model

2. Anatomy of a CRD YAML

Key fields explained

Register the CRD with the cluster

3. A Custom Resource Instance

Part 1: Rate Limiting with `connectionPool`

Part 2: Circuit Breaking with `outlierDetection`

2. The Safety Net: `maxEjectionPercent` + `minHealthPercent`

What happens internally when `index.Add(pod1)` is called

Query flow: `ByIndex("testmodes", "bar")`