agent: k8s: coordinate kubelet pod list fetching#7085
Open
c4rlo wants to merge 2 commits into
Open
Conversation
c4rlo
marked this pull request as ready for review
June 21, 2026 09:58
c4rlo
requested review from
MarcosDY,
amartinezfayo,
rturner3 and
sorindumitru
as code owners
June 21, 2026 09:58
There was a problem hiding this comment.
Pull request overview
This PR refactors the Kubernetes workload attestor’s kubelet pod-list polling to use a long-lived “pod list fetcher” that coordinates concurrent callers, versions responses, caches successful results, and enforces a minimum interval between kubelet request start times (replacing the prior singleflight+TTL approach). It also extracts kubelet HTTP client construction into a dedicated component and updates related error handling, logging, and tests.
Changes:
- Introduces
podListFetcherto coordinate pod-list fetching (shared inflight requests, versioned cached results, backoff timing, client reload handling). - Extracts kubelet HTTP client creation/verification into
kubeletclient.goand wires the plugin to configure/validate via the fetcher. - Updates attestation retry/error semantics and extends/adjusts unit tests accordingly.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| pkg/agent/plugin/workloadattestor/k8s/podlistfetcher.go | New long-lived fetcher coordinating pod-list fetch/caching/backoff/versioning and kubelet client reloads |
| pkg/agent/plugin/workloadattestor/k8s/podlistfetcher_test.go | New unit tests covering fetcher caching, sharing, timing, cancellation, reload, and parsing behavior |
| pkg/agent/plugin/workloadattestor/k8s/kubeletclient.go | New kubelet client + transport-config comparison to preserve connection pools when possible |
| pkg/agent/plugin/workloadattestor/k8s/k8s.go | Plugin updated to use the fetcher/client, revised retry flow and error codes, validate/configure integration |
| pkg/agent/plugin/workloadattestor/k8s/k8s_test.go | Test updates/additions to match new retry/backoff semantics and validation behavior |
| pkg/agent/plugin/workloadattestor/k8s/k8s_posix.go | Minor logging/spelling cleanup and small simplifications |
| } | ||
|
|
||
| func (f *podListFetcher) handleConfigResult(result podListConfigResult) { | ||
| f.config = new(result.config) |
Comment on lines
+322
to
329
| if podListErr != nil { | ||
| if ctx.Err() != nil { | ||
| return nil, ctx.Err() | ||
| } | ||
| // Otherwise, we'll log podListErr below, and we may retry. | ||
| } else { | ||
| podListVersion = podListResult.version | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Should be unlikely to happen, but I've addressed it now.
c4rlo
force-pushed
the
kubelet-requests
branch
2 times, most recently
from
June 21, 2026 22:12
28118c6 to
5d15fc3
Compare
Contributor
|
Please hold this PR. The Broker API PR is changing many things in the k8s plugin, this PR should come on top of it. |
3 tasks
Contributor
|
Hey @c4rlo, the Broker API is merged, pls rebase 🙏 |
Move kubelet pod-list fetching, caching and backoff timing into a long-lived fetcher that versions pod-list responses, instead of using the singleflight package. This now enables us to cache successful kubelet responses until another request is eligible to start, which makes more sense. Previously, they were cached for _half_ the polling interval, which could lead to the following situation (illustrated with the default polling interval of 500 ms): - T = 0 ms: Attestation request 1 makes pod-list request, finds no result, sleeps for 500 ms. - T = 300 ms: Attestation request 2 finds cache expired, makes pod-list request that contains an entry that request 1 was looking for. - T = 500 ms: Only now does request 1 wake up and find the desired entry in the cached pod list. Fixing this with the singleflight approach would not have been easily possible, as it could cause retries to re-use the same cached pod list response as in the previous iteration. The new approach also enforces a minimum interval between kubelet request _start times_ across concurrent attestations. Unlike the previous fixed delay after each request, this allows the next request to start immediately when the previous one took longer than the configured interval. In addition, several other improvements (that could also have been implemented with the old singleflight mechanism): - kubelet requests: do not cancel an inflight request just because its _original_ caller is no longer interested, as other callers may be waiting for it. (With singleflight, this could have been fixed via the Group.DoChan method.) - kubelet requests: retry failed fetches via the same backoff mechanism, rather than immediately returning a failure response. Previously, it was only used when the pod list did not contain the workload. - kubelet requests: when reloading the client (whether through explicit Configure call or periodically via reload_interval), only create a fresh http.Transport object if necessary based on config/certificate changes, ensuring we don't trash its connection pool unnecessarily. - kubelet requests: treat invalid "items" JSON type in reponse as error instead of empty. - kubelet requests: replace "attempt" log value rather than appending a new one with each attempt. - kubelet requests: report kubelet and malformed-response failures as Unavailable instead of Internal. (Failure to find the workload in the pod list after attempts exhausted continues to be DeadlineExceeded.) - kubelet requests: use net.JoinHostPort for endpoints (enables IPv6 support). - kube API server requests: correctly reflect runtime changes to api_server.cache.enabled setting via Configure(). - Validate(): more thorough implementation that checks it can read and parse kubelet certificates, etc. - gRPC return codes: preserve whether caller termination was cancellation or deadline expiry. - gRPC return codes: report failed Sigstore policy verification as PermissionDenied instead of Internal. - tests: fix a flakey (order-dependent) Sigstore selector test failure caused by mutating shared expected selectors. - misc: small code cleanups. - docs: add missing descriptions of max_poll_attempts, poll_retry_interval, reload_interval. - docs: assorted fixes. Signed-off-by: Carlo Teubner <cteubner1@bloomberg.net>
Contributor
Author
|
Rebased, and included a few more small improvements; commit message & PR description updated accordingly. |
Comment on lines
+184
to
+193
| func (f *podListFetcher) registerFetchRequest(afterVersion uint64, resultCh chan<- podListFetchResult) { | ||
| if f.cachedPodList.pods != nil && afterVersion < f.cachedPodList.version && | ||
| f.clock.Now().Before(f.cachedFetchStart.Add(f.config.pollRetryInterval)) { | ||
| resultCh <- podListFetchResult{versionedPodList: f.cachedPodList} | ||
| return | ||
| } | ||
|
|
||
| f.fetchWaiters[resultCh] = struct{}{} | ||
| f.scheduleFetch() | ||
| } |
Contributor
Author
There was a problem hiding this comment.
Shouldn't happen since Configure() gets called before any Attest() calls afaik, but still good to defend against this, hence addressed in cbf4385.
Comment on lines
+373
to
+380
| for _, podValue := range items { | ||
| uid := string(podValue.Get("metadata", "uid").GetStringBytes()) | ||
| if uid == "" { | ||
| f.log.Warn("Pod has no UID", "pod", podValue) | ||
| continue | ||
| } | ||
| result[uid] = podValue | ||
| } |
Contributor
Author
There was a problem hiding this comment.
This can't happen because fastjson.Value.Get() and fastjson.Value.GetStringBytes() both have a check for whether they're being called on a nil value:
(Also, this isn't new code, it's just been moved around.)
Signed-off-by: Carlo Teubner <cteubner1@bloomberg.net>
Contributor
|
Hey @c4rlo #7094 merged and introduced some conflicts, sorry about that, can you pls rebase again? 🙏 @sorindumitru We should try to review this one soon so @c4rlo doesn't have to keep solving conflicts 🙏 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Move kubelet pod-list fetching, caching and backoff timing into a long-lived fetcher that versions pod-list responses, instead of using the
singleflightpackage.This now enables us to cache successful kubelet responses until another request is eligible to start, which makes more sense. Previously, they were cached for half the polling interval, which could lead to the following situation (illustrated with the default polling interval of 500 ms):
Fixing this with the
singleflightapproach would not have been easily possible, as it could cause retries to re-use the same cached pod list response as in the previous iteration.The new approach also enforces a minimum interval between kubelet request start times across concurrent attestations. Unlike the previous fixed delay after each request, this allows the next request to start immediately when the previous one took longer than the configured interval.
In addition, several other improvements (that could also have been implemented with the old
singleflightmechanism):singleflight, this could have been fixed via theGroup.DoChanmethod.)Configure()call or periodically viareload_interval, only create a freshhttp.Transportobject if necessary based on config or certificate changes, ensuring we do not trash its connection pool unnecessarily.itemsJSON type in the response as an error instead of empty.attemptlog value rather than appending a new one with each attempt.Unavailableinstead ofInternal. (Failure to find the workload in the pod list after attempts are exhausted continues to beDeadlineExceeded.)net.JoinHostPortfor endpoints, enabling IPv6 support.api_server.cache.enabledsetting viaConfigure().Validate(): perform more thorough validation, including checking that kubelet certificates can be read and parsed.PermissionDeniedinstead ofInternal.max_poll_attempts,poll_retry_interval, andreload_interval.Open questions:
singleflightapproach? I think so, as it is a more principled approach with more logical semantics, but it is probably debatable.poll_retry_intervalconfig setting? Currently it is 500 ms. But its meaning changes with this PR: it used to be the sleep time between retry attempts, with the cache TTL being set to half that value (250 ms); now it is the minimum interval between fetch start times for both new requests and retries, with the cache TTL also being the same. I would be tempted to reduce the default value now, perhaps to 250 ms, but I do not have a strong opinion either.