Serialize Once, Deliver to a Thousand Watchers: The cachingObject

TL;DR: A watch event that fans out to a thousand watchers would normally be serialized a thousand times, once per connection, because each connection negotiates its own content type and its own framing. The watch cache wraps every event object in a cachingObject that memoizes the encoded bytes per runtime.Identifier (which encodes the content type), so the JSON or protobuf is produced exactly once and then handed out. The deep copy that would normally protect the shared object from mutation is done lazily — only on the first actual mutation, and only if the mutation is not a no-op — because ADDED and MODIFIED events usually need no mutation at all.

Here is the shape of the problem. The watch cache holds one copy of every object. A thousand clients are watching. An object changes. The cacher fans the change out to every watcher, and each watcher writes the object into its own HTTP response stream, encoded in whatever content type that client negotiated.

Naively, that is a thousand JSON encodes of the same object, each producing a fresh byte slice, each of which then has to be copied into that connection’s buffer. On a busy apiserver with a large object and a thousand watchers, the encoding dominates the CPU cost of serving the watch.

The watch cache’s answer is a memo attached to the object itself.

The memo and the object it wraps

The memo is a small struct — a sync.Once, the bytes, and an error:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:45
// serializationResult captures a result of serialization.
type serializationResult struct {
	// once should be used to ensure serialization is computed once.
	once sync.Once

	// raw is serialized object.
	raw []byte
	// err is error from serialization.
	err error
}

// serializationsCache is a type for caching serialization results.
type serializationsCache map[runtime.Identifier]*serializationResult

The key is a runtime.Identifier, and identifiers are not just “json” or “protobuf”. They are a composite of everything that affects the output bytes — the encoder plus the group-versioner, plus, in the versioning codec’s case, the negotiated API version:

go
// staging/src/k8s.io/apimachinery/pkg/runtime/serializer/versioning/versioning.go:96
type codecIdentifier struct {
	EncodeGV string `json:"encodeGV,omitempty"`
	Encoder  string `json:"encoder,omitempty"`
	Name     string `json:"name,omitempty"`
}

// identifier computes Identifier of Encoder based on codec parameters.
func identifier(encodeGV runtime.GroupVersioner, encoder runtime.Encoder) runtime.Identifier {
	result := codecIdentifier{
		Name: "versioning",
	}

	if encodeGV != nil {
		result.EncodeGV = encodeGV.Identifier()
	}
	if encoder != nil {
		result.Encoder = string(encoder.Identifier())
	}

Two clients watching the same object, one asking for application/json and the other for application/vnd.kubernetes.protobuf, get two different identifiers and therefore two different entries. Two clients asking for the same thing share one entry. That is exactly the granularity you want.

The wrapper itself implements metav1.Object and runtime.CacheableObject. It holds the wrapped object, a flag recording whether a deep copy has happened yet, and the memo map itself:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:64
type cachingObject struct {
	lock sync.RWMutex

	// deepCopied defines whether the object below has already been
	// deep copied. The operation is performed lazily on the first
	// setXxx operation.
	//
	// The lazy deep-copy make is useful, as effectively the only
	// case when we are setting some fields are ResourceVersion for
	// DELETE events, so in all other cases we can effectively avoid
	// performing any deep copies.
	deepCopied bool

	// Object for which serializations are cached.
	object metaRuntimeInterface

	// serializations is a cache containing object`s serializations.
	// The value stored in atomic.Value is of type serializationsCache.
	// The atomic.Value type is used to allow fast-path.
	serializations atomic.Value
}

That comment on deepCopied is the key insight of the whole design, and it is worth reading twice. The only field that ever gets mutated in practice is resourceVersion, and only on DELETE events. Everything else is read-only. So the deep copy — which for a big pod spec is a large allocation and a large memcpy — can be deferred until somebody actually tries to write, and skipped entirely if that write turns out to be a no-op.

The read path needs no lock

The memo map lives inside an atomic.Value, and it is replaced copy-on-write. The read path is a plain map lookup with no lock at all:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:102
func (o *cachingObject) getSerializationResult(id runtime.Identifier) *serializationResult {
	// Fast-path for getting from cache.
	serializations := o.serializations.Load().(serializationsCache)
	if result, exists := serializations[id]; exists {
		return result
	}

	// Slow-path (that may require insert).
	o.lock.Lock()
	defer o.lock.Unlock()

	serializations = o.serializations.Load().(serializationsCache)
	// Check if in the meantime it wasn't inserted.
	if result, exists := serializations[id]; exists {
		return result
	}

	// Insert an entry for <id>. This requires copy of existing map.
	newSerializations := make(serializationsCache)
	for k, v := range serializations {
		newSerializations[k] = v
	}
	result := &serializationResult{}
	newSerializations[id] = result
	o.serializations.Store(newSerializations)
	return result
}

The write path copies the whole map and stores the new one. That is O(size of map), but the map holds one entry per distinct content type — one, two, maybe three — so the copy is trivial. And the read path is a single atomic load plus a map lookup, which is what thousands of concurrent watchers actually hammer.

Encode exactly once, splice zero-copy

The memo is populated by CacheEncode, which is the hook the serializers call instead of writing bytes directly:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:136
func (o *cachingObject) CacheEncode(id runtime.Identifier, encode func(runtime.Object, io.Writer) error, w io.Writer) error {
	result := o.getSerializationResult(id)
	result.once.Do(func() {
		buffer := bytes.NewBuffer(nil)
		// TODO(wojtek-t): This is currently making a copy to avoid races
		//   in cases where encoding is making subtle object modifications,
		//   e.g. #82497
		//   Figure out if we can somehow avoid this under some conditions.
		result.err = encode(o.GetObject(), buffer)
		result.raw = buffer.Bytes()
	})
	// Once invoked, fields of serialization will not change.
	if result.err != nil {
		return result.err
	}
	if b, support := w.(runtime.Splice); support {
		b.Splice(result.raw)
		return nil
	}
	_, err := w.Write(result.raw)
	return err
}

Three things to notice.

First, result.once.Do is the guarantee. The encode closure runs exactly once per identifier, no matter how many goroutines call CacheEncode concurrently. The sync.Once also provides the happens-before edge that makes result.raw safe to read afterwards.

Second, o.GetObject() returns a deep copy — so the encoder cannot mutate the shared object. That copy happens inside the once.Do, so it happens once too, not once per watcher. The TODO about issue #82497 is about that copy: it is defensive, and the author would like to remove it.

Third, the runtime.Splice fast path. If the writer supports splicing, the cached bytes are moved into the writer rather than copied.

Splice is a tiny interface in apimachinery, and the contract is explicit about ownership:

go
// staging/src/k8s.io/apimachinery/pkg/runtime/splice.go:24
// Splice is the interface that wraps the Splice method.
//
// Splice moves data from given slice without copying the underlying data for
// efficiency purpose. Therefore, the caller should make sure the underlying
// data is not changed later.
type Splice interface {
	Splice([]byte)
	io.Writer
	Reset()
	Bytes() []byte
}

“Splice moves data from given slice without copying the underlying data.” The watch buffer implements it, so the first watcher to encode an object gets the bytes handed over by reference, and every subsequent watcher gets the same slice handed over again. Instead of a thousand copies of a 200 KiB pod, there is one allocation and a thousand pointer moves.

Lazy deep copy, and the no-op check

Every setter on cachingObject routes through one helper, which does the no-op check twice — once optimistically under a read lock, once properly under the write lock:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:227
func (o *cachingObject) conditionalSet(isNoop func() bool, set func()) {
	if fastPath := func() bool {
		o.lock.RLock()
		defer o.lock.RUnlock()
		return isNoop()
	}(); fastPath {
		return
	}
	o.lock.Lock()
	defer o.lock.Unlock()
	if isNoop() {
		return
	}
	if !o.deepCopied {
		o.object = o.object.DeepCopyObject().(metaRuntimeInterface)
		o.deepCopied = true
	}
	o.invalidateCacheLocked()
	set()
}

Three stages. First, if the write would not change anything, return immediately — no deep copy, no cache invalidation. The no-op predicate is supplied by each setter; for the resource version it is a string compare, for the labels and annotations it is a reflect.DeepEqual:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:363
func (o *cachingObject) SetLabels(labels map[string]string) {
	o.conditionalSet(
		func() bool { return reflect.DeepEqual(o.object.GetLabels(), labels) },
		func() { o.object.SetLabels(labels) },
	)
}

Second, if the write is a real change, deep copy the wrapped object — but only once, ever, guarded by o.deepCopied. Third, invalidate the memo, because the cached bytes no longer describe the object.

The invalidation is treated as a bug, because it should never happen:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:208
func (o *cachingObject) invalidateCacheLocked() {
	if cache, ok := o.serializations.Load().(serializationsCache); ok && len(cache) == 0 {
		return
	}
	// We don't expect cache invalidation to happen - so we want
	// to log the stacktrace to allow debugging if that will happen.
	// OTOH, we don't want to spam logs with it.
	// So we try to log it at most once per second.
	if shouldLogCacheInvalidation(time.Now()) {
		klog.Warningf("Unexpected cache invalidation for %#v\n%s", o.object, string(debug.Stack()))
	}
	o.serializations.Store(make(serializationsCache))
}

A stack trace at warning level, rate-limited to one per second:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/caching_object.go:191
var (
	invalidationCacheTimestampLock sync.Mutex
	invalidationCacheTimestamp     time.Time
)

// shouldLogCacheInvalidation allows for logging cache-invalidation
// at most once per second (to avoid spamming logs in case of issues).
func shouldLogCacheInvalidation(now time.Time) bool {
	invalidationCacheTimestampLock.Lock()
	defer invalidationCacheTimestampLock.Unlock()
	if invalidationCacheTimestamp.Add(time.Second).Before(now) {
		invalidationCacheTimestamp = now
		return true
	}
	return false
}

If you ever see Unexpected cache invalidation in a kubelet or apiserver log, something mutated a shared object behind the cache’s back. That is a real bug, and the stack trace is there to help you find it.

Where the wrapping happens

The wrapping is not done for every event. It is decided per event type, at dispatch time, and it is deliberately asymmetric:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go:987
func setCachingObjects(event *watchCacheEvent, versioner storage.Versioner) {
	switch event.Type {
	case watch.Added, watch.Modified:
		if object, err := newCachingObject(event.Object); err == nil {
			event.Object = object
		} else {
			klog.Errorf("couldn't create cachingObject from: %#v", event.Object)
		}
		// Don't wrap PrevObject for update event (for create events it is nil).
		// We only encode those to deliver DELETE watch events, so if
		// event.Object is not nil it can be used only for watchers for which
		// selector was satisfied for its previous version and is no longer
		// satisfied for the current version.
		// This is rare enough that it doesn't justify making deep-copy of the
		// object (done by newCachingObject) every time.
	case watch.Deleted:
		// Don't wrap Object for delete events - these are not to deliver any
		// events. Only wrap PrevObject.
		if object, err := newCachingObject(event.PrevObject); err == nil {
			// Update resource version of the object.
			// event.PrevObject is used to deliver DELETE watch events and
			// for them, we set resourceVersion to <current> instead of
			// the resourceVersion of the last modification of the object.
			updateResourceVersion(object, versioner, event.ResourceVersion)
			event.PrevObject = object
		} else {
			klog.Errorf("couldn't create cachingObject from: %#v", event.Object)
		}
	}
}

For ADDED and MODIFIED, only the new object is wrapped — the previous version is left alone, because it is only needed for watchers whose selector matched before and no longer matches, which is rare enough that deep-copying it every time would not pay for itself.

For DELETED, the reverse: the current object is irrelevant (a delete delivers the previous state) and only PrevObject is wrapped. And because the delete event has to carry the current resource version — the one at which the deletion happened, not the one at which the object was last modified — the wrapper’s resource version is rewritten here, which is the one mutation that actually triggers the lazy deep copy.

The wrapping is per-dispatch, not per-watch

The wrapping happens in dispatchEvent, once per event, on a shallow copy of the event struct:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go:1018
func (c *Cacher) dispatchEvent(event *watchCacheEvent) {
	c.startDispatching(event)
	defer c.finishDispatching()
	// Watchers stopped after startDispatching will be delayed to finishDispatching,

	// Since add() can block, we explicitly add when cacher is unlocked.
	// Dispatching event in nonblocking way first, which make faster watchers
	// not be blocked by slower ones.
	if event.Type == watch.Bookmark {
		for _, watcher := range c.watchersBuffer {
			watcher.nonblockingAdd(event)
		}
	} else {
		// Set up caching of object serializations only for dispatching this event.
		//
		// Storing serializations in memory would result in increased memory usage,
		// but it would help for caching encodings for watches started from old
		// versions. However, we still don't have a convincing data that the gain
		// from it justifies increased memory usage, so for now we drop the cached
		// serializations after dispatching this event.
		//
		// Given that CachingObject is just wrapping the object and not perfoming
		// deep-copying (until some field is explicitly being modified), we create
		// it unconditionally to ensure safety and reduce deep-copying.
		//
		// Make a shallow copy to allow overwriting Object and PrevObject.
		wcEvent := *event
		setCachingObjects(&wcEvent, c.versioner)
		wcEvent.timeline.MarkAt(metrics.PointWatcherEnqueued, c.clock.Now())
		event = &wcEvent

Note the comment: the cached serializations are dropped after the dispatch finishes. The memo lives only as long as the fan-out. It is not a long-lived cache of encoded objects — it is a per-event, per-dispatch memo that dies with the event.

And then the fan-out itself, which is where the thousand watchers meet the one event:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go:1049
		c.blockedWatchers = c.blockedWatchers[:0]
		for _, watcher := range c.watchersBuffer {
			if !watcher.nonblockingAdd(event) {
				c.blockedWatchers = append(c.blockedWatchers, watcher)
			}
		}

Every watcher gets the same pointer to the same cachingObject. They share it. The memo inside it is what makes sharing safe, and the lazy deep copy is what makes the one mutation that does happen safe.

The watcher’s side

The one place a watcher mutates the shared object is turning a “the object changed” event into an “ADDED”/“MODIFIED”/“DELETED” event, which depends on whether the label selector matched before and after:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher.go:373
func (c *cacheWatcher) convertToWatchEvent(event *watchCacheEvent) *watch.Event {
	if event.Type == watch.Bookmark {
		e := &watch.Event{Type: watch.Bookmark, Object: event.Object.DeepCopyObject()}
		if !c.wasBookmarkAfterRvSent() {
			if err := storage.AnnotateInitialEventsEndBookmark(e.Object); err != nil {
				utilruntime.HandleError(fmt.Errorf("error while accessing object's metadata gr: %v, identifier: %v, obj: %#v, err: %v", c.groupResource, c.identifier, e.Object, err))
				return nil
			}
		}
		return e
	}

	curObjPasses := event.Type != watch.Deleted && c.filter.filterWithAttrsAndPrefix(event.Key, event.ObjLabels, event.ObjFields, event.Object)
	oldObjPasses := false
	if event.PrevObject != nil {
		oldObjPasses = c.filter.filterWithAttrsAndPrefix(event.Key, event.PrevObjLabels, event.PrevObjFields, event.PrevObject)
	}
	if !curObjPasses && !oldObjPasses {
		// Watcher is not interested in that object.
		return nil
	}

	switch {
	case curObjPasses && !oldObjPasses:
		return &watch.Event{Type: watch.Added, Object: getMutableObject(event.Object)}
	case curObjPasses && oldObjPasses:
		return &watch.Event{Type: watch.Modified, Object: getMutableObject(event.Object)}
	case !curObjPasses && oldObjPasses:
		// return a delete event with the previous object content, but with the event's resource version
		oldObj := getMutableObject(event.PrevObject)
		// We know that if oldObj is cachingObject (which can only be set via
		// setCachingObjects), its resourceVersion is already set correctly and
		// we don't need to update it. However, since cachingObject efficiently
		// handles noop updates, we avoid this microoptimization here.
		updateResourceVersion(oldObj, c.versioner, event.ResourceVersion)
		return &watch.Event{Type: watch.Deleted, Object: oldObj}
	}

	return nil
}

The helper that decides whether to hand out the shared object or a copy is the whole design in four lines:

go
// staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher.go:357
func getMutableObject(object runtime.Object) runtime.Object {
	if _, ok := object.(*cachingObject); ok {
		// It is safe to return without deep-copy, because the underlying
		// object will lazily perform deep-copy on the first try to change
		// any of its fields.
		return object
	}
	return object.DeepCopyObject()
}

A cachingObject is handed out as-is, because it knows how to protect itself. Anything else is deep-copied, because it does not.

The memo map is copied on every new content type for an event. With one or two content types that is one or two tiny map allocations per event, in exchange for skipping N encodes. The lazy deep copy means the expensive allocation happens only for DELETE events, where the resource version actually has to be rewritten.

The one genuinely awkward part is that the memo is dropped after each dispatch, so a watch that starts from an old resource version and replays history re-encodes everything it replays. The comment says so explicitly: the tradeoff was measured and the memory cost of keeping the memos was judged not worth it.

Wrapping up

The one thing to remember: the watch cache does not serialize per watcher. It serializes per event, memoized by content type inside a cachingObject, and then hands the same bytes to every watcher through a spliceBuffer that moves rather than copies.

The part people get wrong: the deep copy is not skipped — it is deferred. cachingObject.conditionalSet performs it on the first mutation that is not a no-op, and invalidateCacheLocked throws the memo away when it happens. Any code path that mutates a watched object without going through the metav1.Object setters will corrupt every watcher sharing that event; the Unexpected cache invalidation warning with a stack trace is the fingerprint of exactly that mistake.

What to go do about it: if you are debugging watch latency, look at apiserver_longrunning_requests and the watch-specific histograms before suspecting the cacher. If you see Unexpected cache invalidation in the apiserver log, treat it as a bug in whatever wrote to the object — grep for direct field assignment on objects obtained from a watch. And when profiling, remember that the encode happens once per event per content type, so a profile that shows JSON marshalling dominating is telling you the number of events is too high, not that the encoding path is inefficient.