Skip to content

Commit b4535b3

Browse files
authored
🤖 feat: add WATCH support to aggregated API server storage (#72)
## Summary Add WATCH support (`rest.Watcher`) to the aggregated API server's codersdk-backed storage for `coderworkspaces` and `codertemplates` resources, enabling `kubectl get --watch`, informers, and other standard Kubernetes watch clients. ## Background The aggregated API server (`aggregation.coder.com/v1alpha1`) previously served CoderWorkspace and CoderTemplate resources through custom REST storage backed by codersdk, but did not implement `rest.Watcher`. This meant clients could not use `?watch=true` — a fundamental Kubernetes API capability used by `kubectl --watch`, controller informers, and client-go watch helpers. This is a best-effort watch implementation: events are emitted when mutations pass through this API server. Out-of-band Coder changes (e.g., builds completing, changes via the Coder UI) are not reflected until a future enhancement adds background polling or bridges codersdk watch streams. ## Implementation ### New shared helpers (`internal/aggregated/storage/watch.go`) - `filterForListOptions(requestNamespace, opts)` — builds a `watch.FilterFunc` that applies namespace scoping, label selector, and field selector filtering - `validateFieldSelector(sel)` — rejects unsupported fields; allows `metadata.name` and `metadata.namespace` - `watchBroadcasterQueueLen` constant (100 events) ### Storage changes (`workspace.go`, `template.go`) - Added `*watch.Broadcaster` and `sync.Once` to each storage struct - Initialized broadcaster in constructors with `watch.DropIfChannelFull` policy - `Destroy()` now shuts down broadcaster (idempotent via `sync.Once`) - Added `Watch(ctx, opts)` implementing `rest.Watcher`: - Defensive nil assertions - Namespace, label, and field selector filtering - Context cancellation and timeout support - Compile-time `_ rest.Watcher = (*Storage)(nil)` assertions - Emit watch events from successful mutations (best-effort, never fails the API request): - **WorkspaceStorage:** `Added` on Create, `Modified` on Update (non-no-op only), `Modified` on Delete (async deletion) - **TemplateStorage:** `Added` on Create, `Modified` on Update, `Deleted` on Delete (synchronous) ### Tests (`internal/aggregated/storage/watch_test.go`) - `TestTemplateStorageWatch_AddedModifiedDeleted` — full lifecycle - `TestWorkspaceStorageWatch_AddedModified` — create + update toggle - `TestWatchRespectsFieldSelectorMetadataName` — field selector filtering - `TestWatchStopsOnContextCancel` — cleanup on disconnect - `TestValidateFieldSelector` — shared helper unit tests - `TestFilterForListOptions` — filter construction unit tests ## Validation - `make build` ✅ - `make test` ✅ - `make lint` ✅ - `make verify-vendor` ✅ ## Risks - **Low:** Watch semantics are best-effort (no resume/resourceVersion tracking, no out-of-band event bridging). This is documented and intentional for the MVP — clients that depend on complete event streams should be aware of this limitation. - **None for existing behavior:** All changes are additive (new interface implementation + new file). Existing CRUD paths are unchanged except for the best-effort broadcast call after success returns, which is fire-and-forget. --- <details> <summary>📋 Implementation Plan</summary> # Plan: Add WATCH support to aggregated API server storage ## Context / Why The aggregated API server currently serves `aggregation.coder.com/v1alpha1` resources (`coderworkspaces`, `codertemplates`) via custom, codersdk-backed REST storage. Today those storage implementations do **not** implement `k8s.io/apiserver/pkg/registry/rest.Watcher`, so Kubernetes clients cannot do `?watch=true` (e.g. `kubectl get … --watch`, informers). Goal: implement a **best-effort** WATCH endpoint by: - Implementing `rest.Watcher` on both storages. - Broadcasting watch events when this API server successfully performs `CREATE` / `UPDATE` / `DELETE` operations. - Supporting basic label/field selector filtering so normal client-go/kubectl usage works. > Note: because these resources are *not* persisted in etcd (they are proxied to Coder), watch semantics will not be identical to native Kubernetes resources. This plan focuses on correctness + practicality for the common case. ## Evidence (what I verified) - Storage implementations: - `internal/aggregated/storage/workspace.go` and `internal/aggregated/storage/template.go` implement CRUD and list/get, but **do not** implement `rest.Watcher`. - They are **codersdk-backed**, not in-memory. - Aggregated API server installs these storages directly: - `internal/app/apiserverapp/apiserverapp.go` (`VersionedResourcesStorageMap` maps `coderworkspaces`/`codertemplates` to `storage.New*Storage`). - Required interface: - `vendor/k8s.io/apiserver/pkg/registry/rest/rest.go` defines `type Watcher interface { Watch(ctx, *internalversion.ListOptions) (watch.Interface, error) }`. - Watch primitives available in-tree: - `vendor/k8s.io/apimachinery/pkg/watch/mux.go` provides `watch.Broadcaster`. - `vendor/k8s.io/apimachinery/pkg/watch/filter.go` provides `watch.Filter`. ## Implementation details ### 1) Add a watch broadcaster to each storage and make `Destroy()` shut it down **Files:** - `internal/aggregated/storage/workspace.go` - `internal/aggregated/storage/template.go` **Changes:** - Extend each storage struct with: - `broadcaster *watch.Broadcaster` - `broadcasterShutdown sync.Once` (so `Destroy()` can be called multiple times) - Initialize in constructors: - `watch.NewBroadcaster(<queueLen>, watch.DropIfChannelFull)` - Prefer `DropIfChannelFull` so slow/abandoned watchers can’t stall API requests. - Update `Destroy()`: - `broadcasterShutdown.Do(func(){ broadcaster.Shutdown() })` **Also update interface assertions** at the top of each file: ```go var _ rest.Watcher = (*WorkspaceStorage)(nil) // and var _ rest.Watcher = (*TemplateStorage)(nil) ``` ### 2) Implement `Watch(ctx, options)` for both storage types **Files:** - `internal/aggregated/storage/workspace.go` - `internal/aggregated/storage/template.go` - (new) `internal/aggregated/storage/watch.go` (shared helpers) **High-level behavior:** - Create a new watcher from the broadcaster. - Ensure it is stopped when the request context is done. - Apply label/field selector filtering using `watch.Filter`. **Suggested helper shape (shared):** ```go // supportedFieldKeys: metadata.name, metadata.namespace, metadata.uid func validateFieldSelector(sel fields.Selector) error { ... } func filterForListOptions( requestNamespace string, opts *metainternalversion.ListOptions, ) (watch.FilterFunc, error) { ... } ``` **Watch method pseudo-code:** ```go func (s *TemplateStorage) Watch(ctx context.Context, opts *metainternalversion.ListOptions) (watch.Interface, error) { if s == nil { return nil, fmt.Errorf("assertion failed: template storage must not be nil") } if ctx == nil { return nil, fmt.Errorf("assertion failed: context must not be nil") } if s.broadcaster == nil { return nil, fmt.Errorf("assertion failed: template broadcaster must not be nil") } // Apply timeout from opts.TimeoutSeconds if set. if opts != nil && opts.TimeoutSeconds != nil { var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, time.Duration(*opts.TimeoutSeconds)*time.Second) defer cancel() } w, err := s.broadcaster.Watch() if err != nil { return nil, err } // Ensure cleanup on disconnect. go func() { <-ctx.Done() w.Stop() }() ns, err := namespaceFromRequestContext(ctx) if err != nil { w.Stop(); return nil, err } filter, err := filterForListOptions(ns, opts) if err != nil { w.Stop(); return nil, err } if filter != nil { w = watch.Filter(w, filter) } return w, nil } ``` **Selector behavior details:** - `LabelSelector`: - If nil/empty: match all. - Else: evaluate against `ObjectMeta.Labels`. - `FieldSelector`: - Validate requirements up-front and return `BadRequest` for unsupported fields. - Support exact-match selection on: - `metadata.name` - `metadata.namespace` - `metadata.uid` - Evaluate via `fields.Set{...}` + `sel.Matches(...)`. - Namespace scoping: - If request namespace is non-empty (normal namespaced watch), filter out events whose `metadata.namespace` differs. - If request namespace is empty (`--all-namespaces`), do not apply a namespace filter. ### 3) Emit watch events from successful mutations **Files:** - `internal/aggregated/storage/workspace.go` - `internal/aggregated/storage/template.go` **Pattern:** - After the SDK call succeeds and we have the object we’re returning, broadcast a deep-copied event: - `watch.Added` after `Create` - `watch.Modified` after `Update` - `watch.Deleted` after **template** `Delete` **Template delete:** - We already fetch the template before deletion (`TemplateByName`). Convert that to k8s once and reuse it for: - delete validation - broadcasting the `Deleted` event **Workspace delete (special):** - `Delete` is asynchronous in Coder and returns `deleted=false`. - Don’t emit a `Deleted` watch event. - Instead, capture the returned delete-transition build (currently ignored), update the in-memory `workspace` struct like `Update()` does, and emit a `Modified` event: - This gives watchers a signal that deletion was requested without lying about final deletion. **Important:** do **not** fail the API request if broadcasting fails (broadcaster may be shutting down). Treat broadcaster errors as best-effort. ### 4) Add focused unit tests for watch behavior **Files:** - (new) `internal/aggregated/storage/watch_test.go` (preferred) or extend `internal/aggregated/storage/storage_test.go`. **Tests to add:** 1. `TestTemplateStorageWatchEmitsAddedModifiedDeleted` - Start watch, create template, update template, delete template. - Assert watch channel receives the corresponding event types and object names. 2. `TestWorkspaceStorageWatchEmitsAddedModified` - Start watch, create workspace, update workspace (toggle running). - Assert `Added` then `Modified` events. 3. `TestWatchRespectsFieldSelectorMetadataName` - Start a watch with `fields.OneTermEqualSelector("metadata.name", "acme.ops-template")`. - Create a different template and assert no event. - Create the target template and assert event received. 4. `TestWatchStopsOnContextCancel` - Start watch with cancellable context; cancel; ensure watcher stops (ResultChan closes or no goroutine leak). Use existing helpers: - `newMockCoderServer(t)` - `newTestClientProvider(t, server.URL)` - `namespacedContext("control-plane")` ### 5) Validation Run (in Exec mode): - `make test` - (if files touched beyond tests) `make lint` and `make build` <details> <summary>Future enhancements (intentionally out of MVP scope)</summary> - **Reflect out-of-band Coder changes:** - Workspace build status changes, deletions completing, template changes done outside Kubernetes won’t be emitted unless they pass through this API server. - Options: 1. Background poller (List+diff) feeding the broadcaster. 2. For workspaces, bridge `codersdk.Client.WatchWorkspace(ctx, id)` for per-object watches. - **ResourceVersion / resume semantics:** - This MVP treats watch as “from now” best-effort. - Supporting robust resume would require an event cache / RV tracking. - **SendInitialEvents / bookmarks:** - Client-go may use watch-list (`sendInitialEvents=true`) in some configurations. - If needed, implement initial synthetic Added events + Bookmark via `Broadcaster.WatchWithPrefix`. </details> </details> --- _Generated with `mux` • Model: `anthropic:claude-opus-4-6` • Thinking: `xhigh` • Cost: `$2.27`_ <!-- mux-attribution: model=anthropic:claude-opus-4-6 thinking=xhigh costs=2.27 -->
1 parent 57aee35 commit b4535b3

4 files changed

Lines changed: 1260 additions & 10 deletions

File tree

internal/aggregated/storage/template.go

Lines changed: 177 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,13 +8,15 @@ import (
88
"os"
99
"reflect"
1010
"strings"
11+
"sync"
1112
"time"
1213

1314
"github.com/google/uuid"
1415
apierrors "k8s.io/apimachinery/pkg/api/errors"
1516
metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
1617
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
1718
"k8s.io/apimachinery/pkg/runtime"
19+
"k8s.io/apimachinery/pkg/watch"
1820
"k8s.io/apiserver/pkg/registry/rest"
1921

2022
aggregationv1alpha1 "github.com/coder/coder-k8s/api/aggregation/v1alpha1"
@@ -47,6 +49,7 @@ var (
4749
_ rest.Storage = (*TemplateStorage)(nil)
4850
_ rest.Getter = (*TemplateStorage)(nil)
4951
_ rest.Lister = (*TemplateStorage)(nil)
52+
_ rest.Watcher = (*TemplateStorage)(nil)
5053
_ rest.Creater = (*TemplateStorage)(nil) //nolint:misspell // Kubernetes rest interface name is Creater.
5154
_ rest.Updater = (*TemplateStorage)(nil)
5255
_ rest.GracefulDeleter = (*TemplateStorage)(nil)
@@ -60,6 +63,10 @@ var (
6063
type TemplateStorage struct {
6164
provider coder.ClientProvider
6265
tableConvertor rest.TableConvertor
66+
broadcaster *watch.Broadcaster
67+
watchEvents chan watch.Event
68+
watchEventsWG sync.WaitGroup
69+
destroyOnce sync.Once
6370
}
6471

6572
// NewTemplateStorage builds codersdk-backed storage for CoderTemplate resources.
@@ -68,10 +75,16 @@ func NewTemplateStorage(provider coder.ClientProvider) *TemplateStorage {
6875
panic("assertion failed: template client provider must not be nil")
6976
}
7077

71-
return &TemplateStorage{
78+
storage := &TemplateStorage{
7279
provider: provider,
7380
tableConvertor: rest.NewDefaultTableConvertor(aggregationv1alpha1.Resource("codertemplates")),
81+
broadcaster: watch.NewBroadcaster(watchBroadcasterQueueLen, watch.DropIfChannelFull),
82+
watchEvents: make(chan watch.Event, watchBroadcasterQueueLen),
7483
}
84+
storage.watchEventsWG.Add(1)
85+
go storage.dispatchWatchEvents()
86+
87+
return storage
7588
}
7689

7790
// New returns an empty CoderTemplate object.
@@ -80,7 +93,23 @@ func (s *TemplateStorage) New() runtime.Object {
8093
}
8194

8295
// Destroy cleans up storage resources.
83-
func (s *TemplateStorage) Destroy() {}
96+
func (s *TemplateStorage) Destroy() {
97+
if s == nil {
98+
return
99+
}
100+
101+
s.destroyOnce.Do(func() {
102+
if s.watchEvents == nil {
103+
panic("assertion failed: template watch event queue must not be nil")
104+
}
105+
close(s.watchEvents)
106+
s.watchEventsWG.Wait()
107+
108+
if s.broadcaster != nil {
109+
s.broadcaster.Shutdown()
110+
}
111+
})
112+
}
84113

85114
// NamespaceScoped returns true because CoderTemplate is namespaced.
86115
func (s *TemplateStorage) NamespaceScoped() bool {
@@ -189,6 +218,70 @@ func (s *TemplateStorage) List(ctx context.Context, _ *metainternalversion.ListO
189218
return list, nil
190219
}
191220

221+
// Watch watches CoderTemplate objects backed by codersdk.
222+
func (s *TemplateStorage) Watch(ctx context.Context, opts *metainternalversion.ListOptions) (watch.Interface, error) {
223+
if s == nil {
224+
return nil, fmt.Errorf("assertion failed: template storage must not be nil")
225+
}
226+
if ctx == nil {
227+
return nil, fmt.Errorf("assertion failed: context must not be nil")
228+
}
229+
if s.broadcaster == nil {
230+
return nil, fmt.Errorf("assertion failed: template broadcaster must not be nil")
231+
}
232+
233+
requestNamespace, err := namespaceFromRequestContext(ctx)
234+
if err != nil {
235+
return nil, err
236+
}
237+
238+
if err := validateUnsupportedWatchListOptions(opts); err != nil {
239+
return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid watch options: %v", err))
240+
}
241+
242+
filter, err := filterForListOptions(requestNamespace, opts)
243+
if err != nil {
244+
return nil, apierrors.NewBadRequest(fmt.Sprintf("invalid watch options: %v", err))
245+
}
246+
247+
w, err := s.broadcaster.Watch()
248+
if err != nil {
249+
return nil, fmt.Errorf("failed to create template watcher: %w", err)
250+
}
251+
stopAwareWatcher := newStopAwareWatch(w)
252+
253+
var timeoutTimer *time.Timer
254+
if opts != nil && opts.TimeoutSeconds != nil && *opts.TimeoutSeconds > 0 {
255+
timeoutTimer = time.NewTimer(time.Duration(*opts.TimeoutSeconds) * time.Second)
256+
}
257+
258+
go func() {
259+
if timeoutTimer == nil {
260+
select {
261+
case <-ctx.Done():
262+
stopAwareWatcher.Stop()
263+
case <-stopAwareWatcher.Done():
264+
}
265+
return
266+
}
267+
268+
defer timeoutTimer.Stop()
269+
select {
270+
case <-ctx.Done():
271+
case <-timeoutTimer.C:
272+
case <-stopAwareWatcher.Done():
273+
return
274+
}
275+
stopAwareWatcher.Stop()
276+
}()
277+
278+
if filter != nil {
279+
return watch.Filter(stopAwareWatcher, filter), nil
280+
}
281+
282+
return stopAwareWatcher, nil
283+
}
284+
192285
// Create creates a CoderTemplate through codersdk.
193286
func (s *TemplateStorage) Create(
194287
ctx context.Context,
@@ -205,6 +298,9 @@ func (s *TemplateStorage) Create(
205298
if obj == nil {
206299
return nil, fmt.Errorf("assertion failed: object must not be nil")
207300
}
301+
if s.broadcaster == nil {
302+
return nil, fmt.Errorf("assertion failed: template broadcaster must not be nil")
303+
}
208304

209305
templateObj, ok := obj.(*aggregationv1alpha1.CoderTemplate)
210306
if !ok {
@@ -284,7 +380,14 @@ func (s *TemplateStorage) Create(
284380
return nil, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), templateObj.Name)
285381
}
286382

287-
return convert.TemplateToK8s(namespace, createdTemplate), nil
383+
result := convert.TemplateToK8s(namespace, createdTemplate)
384+
if result == nil {
385+
return nil, fmt.Errorf("assertion failed: converted template must not be nil")
386+
}
387+
388+
s.enqueueWatchEvent(watch.Added, result.DeepCopy())
389+
390+
return result, nil
288391
}
289392

290393
request, err := convert.TemplateCreateRequestFromK8s(templateObj, templateName)
@@ -297,7 +400,14 @@ func (s *TemplateStorage) Create(
297400
return nil, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), templateObj.Name)
298401
}
299402

300-
return convert.TemplateToK8s(namespace, createdTemplate), nil
403+
result := convert.TemplateToK8s(namespace, createdTemplate)
404+
if result == nil {
405+
return nil, fmt.Errorf("assertion failed: converted template must not be nil")
406+
}
407+
408+
s.enqueueWatchEvent(watch.Added, result.DeepCopy())
409+
410+
return result, nil
301411
}
302412

303413
// Update applies a template metadata/source reconcile.
@@ -322,6 +432,9 @@ func (s *TemplateStorage) Update(
322432
if objInfo == nil {
323433
return nil, false, fmt.Errorf("assertion failed: updated object info must not be nil")
324434
}
435+
if s.broadcaster == nil {
436+
return nil, false, fmt.Errorf("assertion failed: template broadcaster must not be nil")
437+
}
325438
if forceAllowCreate {
326439
return nil, false, apierrors.NewMethodNotSupported(
327440
aggregationv1alpha1.Resource("codertemplates"),
@@ -531,7 +644,17 @@ func (s *TemplateStorage) Update(
531644
return nil, false, err
532645
}
533646

534-
return refreshedObj, false, nil
647+
result, ok := refreshedObj.(*aggregationv1alpha1.CoderTemplate)
648+
if !ok {
649+
return nil, false, fmt.Errorf("assertion failed: expected *CoderTemplate, got %T", refreshedObj)
650+
}
651+
if result == nil {
652+
return nil, false, fmt.Errorf("assertion failed: refreshed template must not be nil")
653+
}
654+
655+
s.enqueueWatchEvent(watch.Modified, result.DeepCopy())
656+
657+
return result, false, nil
535658
}
536659

537660
// Delete deletes a CoderTemplate through codersdk.
@@ -550,6 +673,9 @@ func (s *TemplateStorage) Delete(
550673
if name == "" {
551674
return nil, false, fmt.Errorf("assertion failed: template name must not be empty")
552675
}
676+
if s.broadcaster == nil {
677+
return nil, false, fmt.Errorf("assertion failed: template broadcaster must not be nil")
678+
}
553679

554680
namespace, badNamespaceErr := requiredNamespaceFromRequestContext(ctx)
555681
if badNamespaceErr != nil {
@@ -586,9 +712,55 @@ func (s *TemplateStorage) Delete(
586712
return nil, false, coder.MapCoderError(err, aggregationv1alpha1.Resource("codertemplates"), name)
587713
}
588714

715+
templateObj := convert.TemplateToK8s(namespace, template)
716+
if templateObj == nil {
717+
return nil, false, fmt.Errorf("assertion failed: converted template must not be nil")
718+
}
719+
720+
// Emit a Deleted event with the last-known template state.
721+
s.enqueueWatchEvent(watch.Deleted, templateObj.DeepCopy())
722+
589723
return &metav1.Status{Status: metav1.StatusSuccess}, true, nil
590724
}
591725

726+
func (s *TemplateStorage) dispatchWatchEvents() {
727+
defer s.watchEventsWG.Done()
728+
729+
for event := range s.watchEvents {
730+
_ = s.broadcaster.Action(event.Type, event.Object)
731+
}
732+
}
733+
734+
func (s *TemplateStorage) enqueueWatchEvent(action watch.EventType, obj runtime.Object) {
735+
if s == nil {
736+
panic("assertion failed: template storage must not be nil")
737+
}
738+
if s.watchEvents == nil {
739+
panic("assertion failed: template watch event queue must not be nil")
740+
}
741+
if obj == nil {
742+
panic("assertion failed: template watch event object must not be nil")
743+
}
744+
745+
event := watch.Event{Type: action, Object: obj}
746+
select {
747+
case s.watchEvents <- event:
748+
return
749+
default:
750+
}
751+
752+
// Keep mutation request handlers non-blocking under watch backpressure by
753+
// dropping the oldest queued event and retaining the most recent state.
754+
select {
755+
case <-s.watchEvents:
756+
default:
757+
}
758+
select {
759+
case s.watchEvents <- event:
760+
default:
761+
}
762+
}
763+
592764
// ConvertToTable converts a template object or list into kubectl table output.
593765
func (s *TemplateStorage) ConvertToTable(ctx context.Context, object, tableOptions runtime.Object) (*metav1.Table, error) {
594766
if s == nil {

0 commit comments

Comments
 (0)