DEV Community: Shuhei Kitagawa

Understanding the controller-runtime Cache Seriously

Shuhei Kitagawa — Sat, 11 May 2024 00:37:06 +0000

The controller-runtime package provides a Cache mechanism to users so that they don need to be aware of Informer's presence. By default, when you Get/List Kubernetes Objects, the Cache will automatically start Informers in the background and cache all Objects with the same GVK (GroupVersionKind).

This (somewhat naïve) behavior of the Cache usually doesn't pose an issue in small Kubernetes Clusters. However, as the cluster scales up, this can lead Controllers to unexpectedly consume a significant amount of memory. To address this, the Cache comes equipped with Options to finely control the behavior of Informers.

However, there is a lack of documentation regarding these Options, making it challenging to understand the intricate behaviors. Therefore, we will first delve into the Cache code and then consider the behavior of the Options. When we refer to "Cache" below, it encompasses the Cache interface, as well as its implementations, namely informerCache, multiNamespaceCache, and delegatingByGVKCache.

Get/List and Cache

Before getting into the details of Cache, let's check the general flow of Get/List Kubernetes Objects using Cache. When using controller-runtime, you typically start by obtaining a Client to interact with the Kubernetes API using the GetClient method.

The actual GetClient method is controllerManager's GetClient method, which internally calls the cluster's GetClient method This method simply returns the value of the cluster's client field, so we'll now look at what is set in this client field.

What is set in the client field is clientWriter. The clientWriter is initialized using the options.NewClient function, with the client package's New function set as the default. Here, Cache is also initialized using the options.NewCache function, with the cache package's New function set as the default. The initialized Cache is passed as an argument to options.NewClient¹².

The cache package's New function then calls the newClient function to initialize the Client. It is clear that the Client's cache field is set with the Cache we initialized earlier (link).

All we need to do now is to look at the client's Get and List methods. After calling the shouldBypassCache method, if the Kubernetes Object is allowed to be cached, the Get/List methods of Cache are called.

Although this was rather extensive, we've now determined that calling Get/List using a client obtained from the GetClient method actually calls the Get/List methods of Cache.

Types of Cache

The following types of Cache implement the Cache interface:

informerCache is the most basic implementation of Cache, and it is directly used as Cache without specifying the Cache Options DefaultNamespaces or ByObject. multiNamespaceCache is used when setting the DefaultNamespaces option, and internally has a map of namespace name to Cache (= informerCache), allowing you to specify the behavior of Informers for particular namespaces.

delegatingByGVKCache is used when setting the ByObject option, and internally a map of GVK to Cache. It allows you to specify the behavior of Informers for particular GVKs. If the Namespaces option of ByObject is set³, then multiNamespaceCache is used as Cache for the corresponding GVK; otherwise, informerCache is used. Let's take a closer look at the behavior of each Cache.

informerCache

Using the informerCache's Get method as an example, let's confirm the flow of dynamically created Informers. The Get method internally calls the getInformerForKind method of informerCache, which in calls the Get method of Informers. If the Informer does not exist, Get method of Informers invokes the addInformerToMap method to create an Informer.

The created Informer is packed into a Cache entry (this Cache is not the Cache interface of the cache package, but rather the Cache struct of the internal package), and is then saved to the tracker field of Informers via the informersByType method⁴.

The Get method of informerCache eventually obtains the Object through the Get method of CacheReader set in the Reader field of this Cache entry. The Get method of CacheReader retrieves the Object from the Indexer.

multiNamespaceCache

For multiNamespaceCache as well, let's quickly look at its behavior. The multiNamespaceCache uses the newMultiNamespaceCache function to create an informerCache for each namespace and saves them in the namespaceToCache field. The Get method of multiNamespaceCache extracts the corresponding informerCache from the namespaceToCache field based on the namespace of the given Object, and calls the Get method of that informerCache.

delegatingByGVKCache

The delegatingByGVKCache generates a Cache for each GVK based on the ByObject option and saves them in the caches field of delegatingByGVKCache. As mentioned before, the Cache created here will be either an informerCache or a multiNamespaceCache depending on the option. The Get method of delegatingByGVKCache extracts the Cache corresponding to the given Object's GVK and calls the Get method of that Cache.

Builder's For, Watches, and Owns, and Cache

In addition to calling Client's Get/List, you can also register Informers through Builder's For, Watches, and Owns methods. Let's check the behavior of Cache in these cases as well. If we look at the Builder's doWatch method, whether you use For, Watches, or Owns, you ultimately call the Watch method of the Controller, which then uses the Start method of Kind to register the Informer using the GetInformer method of Cache.

Cache Options

Let's take a closer look at the behaviors of Cache Options.

ReaderFailOnMissingInformer

ReaderFailOnMissingInformer is an option that prevents the dynamic start-up of Informers through Get/List actions. By looking at the getInformerForKind method of informerCache, we can see that it will return an error if there is no corresponding Informer for an Object's GVK. You can preemptively start Informers either by calling Cache's GetInformer method through Builder's For, Watches, and Owns methods, or by directly invoking Cache's GetInformer method.

DefaultNamespaces

DefaultNamespaces is an option for using multiNamespaceCache to specify Informer behavior per namespace. To see what options can be specified with DefaultNamespaces, check out the Config. Let's consider a few edge cases when DefaultNamespaces is specified.

Get/List for Cluster-scoped Object

Upon reviewing the Get method of multiNamespaceCache, it is clear that if an Object is cluster-scoped, then it calls the Get of clusterCache. The clusterCache is an informerCache without namespace restriction⁵. This means that even if DefaultNamespaces is specified, Informers for cluster-scoped Objects can be started using clusterCache, and Get/List operations can be performed.

Get/List for Objects Outside the Specified Namespace

When we revisit the Get method of multiNamespaceCache, the first step is to search namespaceToCache using the namespace of the provided Object. If a Cache is not found, it searches namespaceToCache again using metav1.NamespaceAll (=""). This indicates that for Get/List of Objects outside the specified Namespace, if the empty string namespace is specified in DefaultNamespaces, informerCache is utilized; otherwise, an error is returned.

DefaultTransform

DefaultTransform specifies the default Transform function to pass to Informers. The Transform function allows modifications to be made to an Object before it is processed by the Informer. A common use case includes deleting fields from an Object (such as Metadata) to reduce memory consumption⁶.

DefaultUnsafeDisableDeepCopy

DefaultUnsafeDisableDeepCopy is an option to conserve memory usage when calling Get/List by allowing the CacheReader to return Kubernetes Objects cached by the Informer without performing a DeepCopy. As stated in the comment, enabling this option requires you to perform a DeepCopy when making changes to the Object. This option is unlikely to be enabled unless there are special circumstances, such as listing a large number of Objects in a single reconcile.

ByObject

ByObject is an option to use delegatingByGVKCache to specify the behavior of Informers for each GVK. The available options that can be specified with ByObject can be found in the ByObject struct. This option is similar to DefaultNamespaces, but the behavior is different when an Object with a GVK not specified by ByObject is passed to Get/List.

Unlike DefaultNamespaces, which returns an error for most unspecified Namespace Get/List operations, the delegatingByGVKCache's cacheForGVK method returns dbt.defaultCache when the provided GVK is not in dbt.caches.

The defaultCache will be either a multiNamespaceCache or an informerCache depending on other options. Thus, with ByObject, if an Object with an unspecified GVK is provided, an error typically does not return, and an Informer may be dynamically started.

In addition to the above, there are other Cache Options such as SyncPeriod, DefaultLabelSelector, DefaultFieldSelector, as well as the Config from DefaultNamespaces and the ByObject option. However, we won't delve into these here as their behaviors can be easily understood from the discussion so far.

The initialized Cache is set to the cluster's cache field, from where it is started through the cluster's Start method. The cluster itself is added as a Runnable to the controllerManager's runnables and started. The concrete implementation of the Start method held by the Cache interface is the Informers' Start method for the informerCache, whereas multiNamespaceCache and delegatingByGVKCache each have their own distinct Start methods. ↩
The Cache itself also implements the Start method through the Informers interface, so it can be directly added to the controllerManager as a Runnable. By creating a Client using this Cache, you can utilize a Client with Cache Options that are completely different from those of the controllerManager. ↩
The multiNamespaceCache is also employed when the DefaultNamespaces option is specified. As noted in the comment, to avoid using the DefaultNamespaces option, one must specify an empty list in the Namespaces option of ByObject. ↩
For more details about client-go's Informer and Indexer, refer to other documents like client-go under the hood. ↩
The globalConfig, when the multiNamespaceCache is generated from DefaultNamespaces, is produced using the optionDefaultsToConfig function, but when the multiNamespaceCache is created through the delegatingByGVKCache originating from ByObject, it results in nil. ↩
Another method to limit memory usage is by using PartialObjectMetadata. By utilizing PartialObjectMetadata, only the metadata of an Object is cached, which can help to reduce the amount of memory used. ↩

Kubernetes Event Aggregation and Spam Filtering in client-go

Shuhei Kitagawa — Mon, 17 Apr 2023 23:09:56 +0000

TL;DR

clint-go has features for event aggregation, which groups similar events into one, and spam filtering, which applies rate limits to events.
Event aggregation uses an aggregation key generated by the EventAggregatorByReasonFunc as a key, and if the same event is published 10 or more times within 10 minutes, instead of posting a new event, update the existing to save etcd capacity.
Spam filtering performs using the value generated by the getSpamKey function as a key. The spam filter uses a rate limiter based on the token bucket algorithm, with an initial 25 tokens and a refill rate of 1 token per 5 minutes. If an event is published beyond the limit, the event is discarded.

Background

When I wanted to create a CustomController that detects the completion of a child Job of a CronJob and performs an action, I noticed that CronJobController (v2) published an Event called SawCompletedJob when the child Job completed. I thought, "Oh, I can simply use the Event as a cue."

However, when I created a CronJob that ran once a minute in a local cluster (kind), I discovered that a SawCompletedJob event became unobservable from the middle of the run (about the 10th run). Specifically, the Events looked like the ones below.

$ kubectl alpha events --for cronjob/hello
...
8m29s (x2 over 8m29s)   Normal   SawCompletedJob    CronJob/hello   Saw completed job: hello-28023907, status: Complete
8m29s                   Normal   SuccessfulDelete   CronJob/hello   Deleted job hello-28023904
7m35s                   Normal   SuccessfulCreate   CronJob/hello   Created job hello-28023908
7m28s (x2 over 7m28s)   Normal   SawCompletedJob    CronJob/hello   Saw completed job: hello-28023908, status: Complete
7m28s                   Normal   SuccessfulDelete   CronJob/hello   Deleted job hello-28023905
6m35s                   Normal   SuccessfulCreate   CronJob/hello   Created job hello-28023909
6m28s                   Normal   SawCompletedJob    CronJob/hello   Saw completed job: hello-28023909, status: Complete
2m35s (x3 over 4m35s)   Normal   SuccessfulCreate   CronJob/hello   (combined from similar events): Created job hello-28023913

SawCompletedJob events were published until 6m28s, but they became unobservable after that. SuccessfulCreate events were published in an aggregated form, but we cannot see all of them. By the way, all events from child Jobs and child Pods are observable.

$ kubectl alpha events

4m18s                 Normal    Scheduled                 Pod/hello-28023914-frb94   Successfully assigned default/hello-28023914-frb94 to kind-control-plane
4m11s                 Normal    Completed                 Job/hello-28023914         Job completed
3m18s                 Normal    Started                   Pod/hello-28023915-5fsh5   Started container hello
3m18s                 Normal    SuccessfulCreate          Job/hello-28023915         Created pod: hello-28023915-5fsh5
3m18s                 Normal    Created                   Pod/hello-28023915-5fsh5   Created container hello
3m18s                 Normal    Pulled                    Pod/hello-28023915-5fsh5   Container image "busybox:1.28" already present on machine
3m18s                 Normal    Scheduled                 Pod/hello-28023915-5fsh5   Successfully assigned default/hello-28023915-5fsh5 to kind-control-plane
3m11s                 Normal    Completed                 Job/hello-28023915         Job completed
2m18s                 Normal    Started                   Pod/hello-28023916-qbqqk   Started container hello
2m18s                 Normal    Pulled                    Pod/hello-28023916-qbqqk   Container image "busybox:1.28" already present on machine
2m18s                 Normal    Created                   Pod/hello-28023916-qbqqk   Created container hello
2m18s                 Normal    SuccessfulCreate          Job/hello-28023916         Created pod: hello-28023916-qbqqk
2m18s                 Normal    Scheduled                 Pod/hello-28023916-qbqqk   Successfully assigned default/hello-28023916-qbqqk to kind-control-plane
2m11s                 Normal    Completed                 Job/hello-28023916         Job completed
78s                   Normal    SuccessfulCreate          Job/hello-28023917         Created pod: hello-28023917-kpxvn
78s                   Normal    Created                   Pod/hello-28023917-kpxvn   Created container hello
78s                   Normal    Pulled                    Pod/hello-28023917-kpxvn   Container image "busybox:1.28" already present on machine
78s                   Normal    Started                   Pod/hello-28023917-kpxvn   Started container hello
78s                   Normal    Scheduled                 Pod/hello-28023917-kpxvn   Successfully assigned default/hello-28023917-kpxvn to kind-control-plane
71s                   Normal    Completed                 Job/hello-28023917         Job completed
18s (x8 over 7m18s)   Normal    SuccessfulCreate          CronJob/hello              (combined from similar events): Created job hello-28023918
18s                   Normal    Started                   Pod/hello-28023918-grvbz   Started container hello
18s                   Normal    Created                   Pod/hello-28023918-grvbz   Created container hello
18s                   Normal    Pulled                    Pod/hello-28023918-grvbz   Container image "busybox:1.28" already present on machine
18s                   Normal    SuccessfulCreate          Job/hello-28023918         Created pod: hello-28023918-grvbz
18s                   Normal    Scheduled                 Pod/hello-28023918-grvbz   Successfully assigned default/hello-28023918-grvbz to kind-control-plane
11s                   Normal    Completed                 Job/hello-28023918         Job completed

As I couldn't find any specific description of this behavior in the Kubernetes official documentation, I investigated it by reading the source code to figure out what was happening.

Kubernetes Event Publication Flow

client-go sends Kubernetes Events to kube-apiserver through and etcd stores them. client-go is responsible for event aggregation and spam filtering, but the flow of client-go sending Events to kube-apiserver is quite complex, so we will explain it first. Note that we will use the SawCompleteJob Event of CronJobController as an example, but please note that the details may vary on each controller.

CronJobController publishes an Event via the Recorder's Eventf method. The method internally calls the Broadcaster's ActionOrDrop method and sends the Event to the incoming channel.
The Event in the incoming channel is retrieved by the loop goroutine and forwarded to each Watcher's result channel.
The Event Watcher goroutine calls the eventHandler for the Event received from the result channel. The eventHandler calls recordToSink in the eventBroadcasterImpl, where EventCorrelator performs event aggregation and spam filtering and then calls recordEvent to post an Event (or update the Event if it has been aggregated).

Note: Starting loop goroutine

The loop goroutine starts through the NewLongQueueBroadcaster function, which is called through the NewBroadcaster function. The NewBroadcaster function is called in the NewControllerV2 function.

Note: Starting Event Watcher

The CronJobController calls the StartRecordingToSink method of eventBroadcasterImpl, which starts the Event Watcher from the StartEventWatcher method. The StartEventWatcher method initializes and registers the Watcher through the Watch method of the Broadcaster. What I found interesting is that the registration process of the Watcher itself is sent to the incoming channel, and the loop goroutine executes it, making the events published before the start of the Watcher invisible to it (the comment calls it as a "terrible hack" though).

EventCorrelator

EventCorrelator implements the core logic for Kubernetes Event aggregation and spam filtering. EventCorrelator is initialized through the NewEventCorrelatorWithOptions function called in the StartRecordingToSink method of eventBroadcasterImpl. Note that e.options is empty, so the Controller uses the default values. eventBroadcasterImpl's recordToSink method calls EventCorrelate method, which aggregates Events and applies tge spam filter.

Aggregation

The EventCorrelator's EventAggregate method and the eventObserve method of eventLogger are used for Event aggregation. The source code has detailed comments, so it's recommended to refer to it directly for more information, but here's a brief overview of the process:

Calculate aggregationKey and localKey using EventAggregatorByReasonFunc. aggregationKey consists of event.Source, event.InvolvedObject, event.Type, event.Reason, event.ReportingController, and event.ReportingInstance, while localKey is event.Message.
Search the cache of EventAggregator using the the aggregationKey. The cache value is aggregateRecord. If the number of the localKeys within the maxIntervalInSeconds (default 600 seconds) is greater than or equal to the maxEvents (default 10), return aggregationKey as the key, otherwise return the eventKey as the key. Note that the eventKey should be unique to each Event.
Call the eventObserve method of eventLogger with the key returned from the EventCorrelate method, and search the cache of eventLogger. The cache value is eventLog. If it hits the cache (i.e., the key is aggregationKey), compute the patch to update the Event.

Spam Filtering

For spam filtering, the filterFunc of EventCorrelator is called to apply it. The actual implementation of filterFunc is the Filter method of EventSourceObjectSpamFilter. Again, it's recommended to refer to the source code for details, but here's a brief overview of the process:

Calculate the eventKey from the Event using the getSpamKey function.
Search the cache of EventSourceObjectSpamFilter using the eventKey. The cache value is spamRecord, which contains the rate limiter. The default values for qps and burst of the rate limiter are 1/300 and 25, respectively. According to the comment, the rate limiter uses the token bucket algorithm, so there are initially 25 tokens, and then one token is refilled every 5 minutes.
Call the TryAccept method of tokenBucketPassiveRateLimiter to check the rate limit. If it exceeds it, discard the Event.

Why did SawCompletedJob Event become unobservable?

Taking the above into consideration, let's think about why the SawCompleteJob Event became unobservable. In short, it is likely due to be caused by the spam filter.

CronJobController issues three Events, SuccessfulCreate, SawCompletedJob, and SuccessfulDelete, per child Job every minute (Strictly speaking, it publishes SuccessfulDelete only when it reaches the HistoryLimit).
The Controller uses a spam filter whose the key is solely based on the Source and InvolvedObject (See getSpamKey function). Therefore, these thee types of Events are identified as the same Event.
The Controller consumed the first 25 tokens at a rate of three tokens per minute. One token is refilled every five minutes, but around nine minutes, the tokens started running out. After that, A toke was refilled every five minutes, but it was consumed by a (aggregated) SuccessfulCreate Event, so SawCompletedJob and SuccessfulDelete were never published thereafter.

Note: Event Types

I'll list SuccessfulCreate、 SawCompletedJob、SuccessfulDelete events' involvedObject and source below.

SuccessfulCreate Event

apiVersion: v1
kind: Event
involvedObject:
  apiVersion: batch/v1
  kind: CronJob
  name: hello
  namespace: default
  resourceVersion: "289520"
  uid: 5f3cfeca-8a83-452a-beb9-7a5f9c1eff63
source:
  component: cronjob-controller
...
reason: SuccessfulCreate
message: Created job hello-28025408

SawCompletedJob Event

apiVersion: v1
kind: Event
involvedObject:
  apiVersion: batch/v1
  kind: CronJob
  name: hello
  namespace: default
  resourceVersion: "289020"
  uid: 5f3cfeca-8a83-452a-beb9-7a5f9c1eff63
source:
  component: cronjob-controller
...
reason: SawCompletedJob
message: 'Saw completed job: hello-28025408, status: Complete'

SuccessfulDelete Event

apiVersion: v1
kind: Event
involvedObject:
  apiVersion: batch/v1
  kind: CronJob
  name: hello
  namespace: default
  resourceVersion: "289520"
  uid: 5f3cfeca-8a83-452a-beb9-7a5f9c1eff63
source:
  component: cronjob-controller
...
reason: SuccessfulDelete
message: Deleted job hello-28025408

Released new Travis CI API Go client, shuheiktgw/go-travis

Shuhei Kitagawa — Mon, 05 Nov 2018 01:35:46 +0000

I just released new Travis CI API client in Go, which is called shuheiktgw/go-travis, so let me introduce why I built it and how to use it.

Why I started developing shuheiktgw/go-travis

Originally, I was just looking for a Travis CI API's client library in Go to kick a build from a CI environment, and found Ableton/go-travis. Unfortunately, there are a few problems with the library.

It does not seem to be maintained anymore.
It uses API v2, whose latest version is v3.
It only supports limited number of endpoints.

In short, I found a vacuum in the field of Travis CI API client in Go, and I was simply thrilled that I might be the first one to fill it.

Also, I used Travis CI for my private projects a lot, so I thought building a Golang client library for them may be a good way to return a favor.

Thant's how I forked Ableton/go-travis and started developing shuheiktgw/go-travis.

How it works

The way shuheiktgw/go-travis works is deadly simple. Its implementation is heavily inspired by google/go-github (as Ableton/go-travis does), so if you have used go-github, you can instantly understand how it works.

Install

$ go get github.com/shuheiktgw/go-travis

Usage

client := travis.NewClient(travis.ApiComUrl, "TravisApiToken")

// List all the builds which belongs to the current user
_, err := client.Builds.List(context.Background(), nil)

travis.NewClient takes two arguments, baseUrl and travisToken. For baseUrl, you need to choose either travis.ApiComUrl or travis.ApiOrgUrl, so please read URL section.

travisToken can be empty, but then you probably need to authenticate via GitHub token. So please check out Authentication section too for more detailed information on authentication.

Features

Supported features

Supports all the endpoints of Travis CI API v3! 🎉

Unsupported features

Does not support Eager loading so far..

That's all, and thank you for reading the post! If you have chance to use shuheiktgw/go-travis, I'd love to hear your feedbacks. Also, contributions to the library are, of course, always welcomed!