#6935 [V2] Directly get CR from informer cache Issue created by machichima Context Currently,

ActionsClient

in

actions/k8s/client.go

uses a direct

client.WithWatch

for all K8s operations. Every

Get()

and

List()

call hits the API server directly. Since we already run a Watch loop (

doWatch

) that tracks all TaskAction changes, we can use a controller-runtime informer cache to serve reads from local memory instead. This reduces API server load and improves read latency. Additionally, the manual Watch loop becomes redundant — the cache internally runs an informer that watches the resource. We replace `doWatch`/`watchLoop` with informer event handlers that feed the existing subscriber notification system. Originally posted by @machichima in #6902 (comment) Approach Use controller-runtime

cache.Cache

for reads (Get/List) and informer event handlers for watch notifications. Keep the existing

client.WithWatch

(or just

client.Client

) for writes (Create, Delete, Status().Update()). Files to modify 1.

app/context.go

— Add K8sCache field • Add

K8sCache cache.Cache

field to

SetupContext

(may be nil) • Add import for

<http://sigs.k8s.io/controller-runtime/pkg/cache|sigs.k8s.io/controller-runtime/pkg/cache>

2.

app/k8s.go

— Add InitK8sCache factory • Add new function: func InitK8sCache(restConfig *rest.Config, scheme *runtime.Scheme) (cache.Cache, error) • Uses

cache.New(restConfig, cache.Options{Scheme: scheme})

• Does NOT start the cache (caller starts it as a background worker) 3.

actions/k8s/client.go

— Core changes Struct changes: • Replace

k8sClient client.WithWatch

with: •

k8sClient client.Client

— for writes only •

k8sCache cache.Cache

— for reads + informer event handlers • Replace

stopCh chan struct{}

and

watching bool

with

handlerRegistration toolscache.ResourceEventHandlerRegistration

Constructor: • Update

NewActionsClient

to accept

cache.Cache

as second parameter Read operations — redirect to cache: • All 5

c.k8sClient.Get(...)

calls →

c.k8sCache.Get(...)

• Line 123: Get parent TaskAction in

Enqueue()

• Line 165: Get TaskAction in

AbortAction()

• Line 182: Get TaskAction in

GetState()

• Line 199: Get TaskAction in

PutState()

• Line 288: Get TaskAction in

GetTaskAction()

• All 2

c.k8sClient.List(...)

calls →

c.k8sCache.List(...)

• Line 236:

ListRunActions()

• Line 261:

ListChildActions()

Write operations — unchanged, stay on k8sClient: • Line 94:

EnsureNamespaceExists()

(writes) • Line 150:

Create()

• Line 169:

Delete()

• Line 215:

Status().Update()

Watch — replace manual loop with informer event handlers: • Replace

StartWatching()

— use

c.k8sCache.GetInformer(ctx, &executorv1.TaskAction{})

then

AddEventHandler(toolscache.ResourceEventHandlerFuncs{...})

•

AddFunc

→ calls

handleWatchEvent

with

watch.Added

•

UpdateFunc

→ calls

handleWatchEvent

with

watch.Modified

•

DeleteFunc

→ calls

handleWatchEvent

with

watch.Deleted

• Remove

watchLoop()

and

doWatch()

— no longer needed • Simplify

StopWatching()

— remove event handler registration via

RemoveEventHandler

, or make it a no-op (cache lifecycle handles cleanup) •

handleWatchEvent()

stays as-is • Add import

toolscache "<http://k8s.io/client-go/tools/cache|k8s.io/client-go/tools/cache>"

4.

actions/setup.go

— Wire cache creation and startup • Create cache: call

app.InitK8sCache(sc.K8sConfig, executor.Scheme())

• Start cache as worker:

sc.AddWorker("actions-cache", func(ctx) { k8sCache.Start(ctx) })

• Add readiness check: verify

k8sCache.WaitForCacheSync(ctx)

• Pass cache to

NewActionsClient(sc.K8sClient, k8sCache, ...)

• Replace the "actions-watcher" worker (currently just waits for ctx.Done) — no longer needed since cache lifecycle manages the informer 5.

actions/cmd/main.go

— Fix standalone binary • Pass

executor.Scheme()

instead of

nil

to

InitKubernetesClient

(needed for cache to decode TaskAction) • Store and set

sc.K8sConfig = k8sConfig

(currently discarded with

_

) 6.

actions/k8s/client_test.go

— Update tests • Update

ActionsClient

struct literals in tests to include the new

k8sCache

field (nil is fine for existing tests that only test

notifyRunService

and helpers) • Add new tests for informer-based watch behavior Verification 1. Unit tests:

go test ./actions/k8s/... -v

2. Integration: Run the manager (

manager/cmd/main.go

), create a TaskAction, verify: • `GetState`/`GetTaskAction` reads from cache (check logs — no direct API calls) • Watch subscribers still receive updates via informer events • Create/Delete/Status.Update still go to API server 3. Standalone: Run

actions/cmd/main.go

standalone, verify cache starts and syncs flyteorg/flyte