diff --git a/pkg/client/generated/models/pkg_server_query_freights_response.go b/pkg/client/generated/models/pkg_server_query_freights_response.go index 1a929962c2..e0a2d1ad90 100644 --- a/pkg/client/generated/models/pkg_server_query_freights_response.go +++ b/pkg/client/generated/models/pkg_server_query_freights_response.go @@ -19,6 +19,13 @@ type PkgServerQueryFreightsResponse struct { // groups Groups map[string]PkgServerFreightGroupList `json:"groups,omitempty"` + + // ResourceVersion is the Kubernetes list resourceVersion clients use to + // seed a follow-up Freight watch so the API server does not replay every + // existing Freight as an ADDED event. It is empty for the stage-scoped + // query, whose result is assembled from multiple sources rather than a + // single watchable namespace list. + ResourceVersion string `json:"resourceVersion,omitempty"` } // Validate validates this pkg server query freights response diff --git a/pkg/server/kubernetes/client.go b/pkg/server/kubernetes/client.go index 23c00dc3e9..3a9f0262ad 100644 --- a/pkg/server/kubernetes/client.go +++ b/pkg/server/kubernetes/client.go @@ -93,9 +93,6 @@ func setOptionsDefaults(opts ClientOptions) (ClientOptions, error) { return opts, fmt.Errorf("error adding Kargo API to scheme: %w", err) } } - if opts.NewInternalClient == nil { - opts.NewInternalClient = newDefaultInternalClient - } if opts.KargoNamespace == "" { opts.KargoNamespace = "kargo" } @@ -124,11 +121,20 @@ type Client interface { // client. This is useful for cases where the API server needs to bypass // the extra authorization checks performed by this client. InternalClient() libClient.WithWatch + + // APIReader returns an uncached reader that reads straight from the + // Kubernetes API server, bypassing the internal client's cache. It is the + // same reader the cache delegates to (the cluster's GetAPIReader), and is + // useful where a cache-served list ResourceVersion would be too old to seed + // a follow-up watch. Like InternalClient, it does NOT enforce Kargo's RBAC, + // so callers must authorize the user before using it. + APIReader() libClient.Reader } // client implements Client. type client struct { internalClient libClient.WithWatch + apiReader libClient.Reader opts ClientOptions getAuthorizedClientFn func( @@ -162,17 +168,44 @@ func NewClient( if opts, err = setOptionsDefaults(opts); err != nil { return nil, fmt.Errorf("error setting client options defaults: %w", err) } - internalClient, err := opts.NewInternalClient( - ctx, - restCfg, - opts.Scheme, - opts.KargoNamespace, + var ( + internalClient libClient.WithWatch + apiReader libClient.Reader ) - if err != nil { - return nil, fmt.Errorf("error building internal client: %w", err) + if opts.NewInternalClient != nil { + // A custom internal client (typically a fake in tests) has no separate + // cache to bypass, so it doubles as the uncached API reader. + if internalClient, err = opts.NewInternalClient( + ctx, + restCfg, + opts.Scheme, + opts.KargoNamespace, + ); err != nil { + return nil, fmt.Errorf("error building internal client: %w", err) + } + apiReader = internalClient + } else { + // The cluster provides both the cache-backed client and the uncached + // reader the cache delegates to (GetAPIReader) -- the same direct-read + // pattern the controllers and external webhooks server use. + var cluster libCluster.Cluster + if cluster, err = newDefaultCluster( + ctx, + restCfg, + opts.Scheme, + opts.KargoNamespace, + ); err != nil { + return nil, fmt.Errorf("error building internal cluster: %w", err) + } + var ok bool + if internalClient, ok = cluster.GetClient().(libClient.WithWatch); !ok { + return nil, errors.New("internal client does not implement WithWatch") + } + apiReader = cluster.GetAPIReader() } c := &client{ internalClient: internalClient, + apiReader: apiReader, opts: opts, } if opts.SkipAuthorization { @@ -195,12 +228,12 @@ func NewClient( return c, nil } -func newDefaultInternalClient( +func newDefaultCluster( ctx context.Context, restCfg *rest.Config, scheme *runtime.Scheme, kargoNamespace string, -) (libClient.WithWatch, error) { +) (libCluster.Cluster, error) { cluster, err := libCluster.New( restCfg, func(clusterOptions *libCluster.Options) { @@ -306,11 +339,7 @@ func newDefaultInternalClient( return nil, fmt.Errorf("error starting cluster: %w", err) } - cl, ok := cluster.GetClient().(libClient.WithWatch) - if !ok { - return nil, errors.New("internal client does not implement WithWatch") - } - return cl, nil + return cluster, nil } func (c *client) Get( @@ -677,6 +706,10 @@ func (c *client) InternalClient() libClient.WithWatch { return c.internalClient } +func (c *client) APIReader() libClient.Reader { + return c.apiReader +} + func (c *client) Watch( ctx context.Context, list libClient.ObjectList, diff --git a/pkg/server/kubernetes/client_test.go b/pkg/server/kubernetes/client_test.go index efeb1726b4..697756be39 100644 --- a/pkg/server/kubernetes/client_test.go +++ b/pkg/server/kubernetes/client_test.go @@ -22,7 +22,9 @@ import ( func TestSetOptionsDefaults(t *testing.T) { opts, err := setOptionsDefaults(ClientOptions{}) require.NoError(t, err) - require.NotNil(t, opts.NewInternalClient) + // NewInternalClient is intentionally left nil; NewClient builds a real + // cluster on the default path and only calls this hook when a test sets it. + require.Nil(t, opts.NewInternalClient) require.NotNil(t, opts.Scheme) } diff --git a/pkg/server/list_for_watch_seed.go b/pkg/server/list_for_watch_seed.go new file mode 100644 index 0000000000..a78a5499df --- /dev/null +++ b/pkg/server/list_for_watch_seed.go @@ -0,0 +1,57 @@ +package server + +import ( + "context" + "fmt" + + "sigs.k8s.io/controller-runtime/pkg/client" + + kargoapi "github.com/akuity/kargo/api/v1alpha1" +) + +// listForWatchSeed lists Kargo resources straight from the Kubernetes API, +// bypassing the API server's controller-runtime read cache via the client's +// uncached APIReader. +// +// The cached client can return a list ResourceVersion of "0" or one that is +// older than the apiserver's compacted floor, which makes follow-up watches +// either replay the full set or fail with a "too old" error and force the +// client into a refetch loop. listForWatchSeed avoids that by reading through +// the uncached APIReader for list+watch seed endpoints, where the returned +// resourceVersion is used to start a follow-up watch. +// +// The APIReader does not enforce Kargo's RBAC, so we authorize the caller first +// with the same list SubjectAccessReview the cached client would perform. +// resource is the lowercase resource name in the Kargo API group (e.g. +// "stages", "warehouses", "promotions", "freights"). +func (s *server) listForWatchSeed( + ctx context.Context, + resource string, + list client.ObjectList, + opts ...client.ListOption, +) error { + if s.client == nil { + return fmt.Errorf("kubernetes client is not configured") + } + if s.authorizeFn == nil { + return fmt.Errorf("authorize function is not configured") + } + + var listOpts client.ListOptions + listOpts.ApplyOptions(opts) + + // Authorize the user before bypassing the cache. The APIReader runs with the + // API server's own credentials, so without this check a caller could read + // data they would otherwise be denied. + if err := s.authorizeFn( + ctx, + "list", + kargoapi.GroupVersion.WithResource(resource), + "", + client.ObjectKey{Namespace: listOpts.Namespace}, + ); err != nil { + return err + } + + return s.client.APIReader().List(ctx, list, opts...) +} diff --git a/pkg/server/list_for_watch_seed_test.go b/pkg/server/list_for_watch_seed_test.go new file mode 100644 index 0000000000..ae61356d87 --- /dev/null +++ b/pkg/server/list_for_watch_seed_test.go @@ -0,0 +1,174 @@ +package server + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + + kargoapi "github.com/akuity/kargo/api/v1alpha1" + "github.com/akuity/kargo/pkg/server/kubernetes" +) + +// newSeedKubeClient builds a kubernetes.Client whose internal client (and thus +// its uncached APIReader) is the supplied fake. +func newSeedKubeClient(t *testing.T, internalClient client.WithWatch) kubernetes.Client { + t.Helper() + kubeClient, err := kubernetes.NewClient( + t.Context(), + &rest.Config{}, + kubernetes.ClientOptions{ + SkipAuthorization: true, + Scheme: mustNewScheme(), + NewInternalClient: func( + context.Context, + *rest.Config, + *runtime.Scheme, + string, + ) (client.WithWatch, error) { + return internalClient, nil + }, + }, + ) + require.NoError(t, err) + return kubeClient +} + +func TestServer_listForWatchSeed(t *testing.T) { + t.Parallel() + + scheme := mustNewScheme() + + t.Run("authorizes and lists through the API reader", func(t *testing.T) { + t.Parallel() + + var ( + authorized bool + authorizedVerb string + authorizedGVR schema.GroupVersionResource + authorizedSub string + authorizedKey client.ObjectKey + listCalled bool + ) + internalClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func( + ctx context.Context, + cl client.WithWatch, + list client.ObjectList, + opts ...client.ListOption, + ) error { + listCalled = true + if err := cl.List(ctx, list, opts...); err != nil { + return err + } + if promotions, ok := list.(*kargoapi.PromotionList); ok { + promotions.ResourceVersion = "42" + } + return nil + }, + }). + Build() + + s := &server{ + client: newSeedKubeClient(t, internalClient), + authorizeFn: func( + _ context.Context, + verb string, + gvr schema.GroupVersionResource, + subresource string, + key client.ObjectKey, + ) error { + authorized = true + authorizedVerb = verb + authorizedGVR = gvr + authorizedSub = subresource + authorizedKey = key + return nil + }, + } + + promotions := &kargoapi.PromotionList{} + err := s.listForWatchSeed( + t.Context(), + "promotions", + promotions, + client.InNamespace("fake-project"), + ) + require.NoError(t, err) + require.True(t, authorized) + require.True(t, listCalled) + require.Equal(t, "list", authorizedVerb) + require.Empty(t, authorizedSub) + require.Equal(t, kargoapi.GroupVersion.WithResource("promotions"), authorizedGVR) + require.Equal(t, client.ObjectKey{Namespace: "fake-project"}, authorizedKey) + require.Equal(t, "42", promotions.ResourceVersion) + }) + + t.Run("authorization failure short-circuits the list", func(t *testing.T) { + t.Parallel() + + authErr := errors.New("not authorized") + var listCalled bool + internalClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithInterceptorFuncs(interceptor.Funcs{ + List: func( + context.Context, + client.WithWatch, + client.ObjectList, + ...client.ListOption, + ) error { + listCalled = true + return nil + }, + }). + Build() + + s := &server{ + client: newSeedKubeClient(t, internalClient), + authorizeFn: func( + context.Context, + string, + schema.GroupVersionResource, + string, + client.ObjectKey, + ) error { + return authErr + }, + } + + err := s.listForWatchSeed( + t.Context(), + "promotions", + &kargoapi.PromotionList{}, + client.InNamespace("fake-project"), + ) + require.ErrorIs(t, err, authErr) + require.False(t, listCalled) + }) + + t.Run("errors when authorize function is not configured", func(t *testing.T) { + t.Parallel() + + s := &server{ + client: newSeedKubeClient(t, fake.NewClientBuilder().WithScheme(scheme).Build()), + } + + err := s.listForWatchSeed( + t.Context(), + "promotions", + &kargoapi.PromotionList{}, + client.InNamespace("fake-project"), + ) + require.ErrorContains(t, err, "authorize function is not configured") + }) +} diff --git a/pkg/server/list_promotions_v1alpha1.go b/pkg/server/list_promotions_v1alpha1.go index a28a96bddd..519797eab4 100644 --- a/pkg/server/list_promotions_v1alpha1.go +++ b/pkg/server/list_promotions_v1alpha1.go @@ -74,21 +74,20 @@ func (s *server) listPromotions(c *gin.Context) { stage := c.Query("stage") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchPromotions(c, project, stage) + s.watchPromotions(c, project, stage, c.Query("resourceVersion")) return } list := &kargoapi.PromotionList{} - opts := []client.ListOption{ - client.InNamespace(project), - } - if stage != "" { - opts = append(opts, client.MatchingFields{indexer.PromotionsByStageField: stage}) - } - if err := s.client.List(ctx, list, opts...); err != nil { + if err := s.listForWatchSeed(ctx, "promotions", list, client.InNamespace(project)); err != nil { _ = c.Error(err) return } + if stage != "" { + list.Items = filterPromotionsByStage(list.Items, stage) + } + + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) // Sort ascending by name slices.SortFunc(list.Items, func(lhs, rhs kargoapi.Promotion) int { @@ -98,14 +97,39 @@ func (s *server) listPromotions(c *gin.Context) { c.JSON(http.StatusOK, list) } -func (s *server) watchPromotions(c *gin.Context, project, stage string) { +// filterPromotionsByStage returns Promotions that target the specified Stage. +// +// Stage filtering is done in-process rather than via the PromotionsByStage +// field index because the watch-seed list goes through listForWatchSeed's +// uncached reader, which cannot serve controller-runtime field indexes. This +// over-fetches the whole namespace, but it matches the (also unfiltered) +// follow-up watch and keeps the returned ResourceVersion watchable. +func filterPromotionsByStage(promotions []kargoapi.Promotion, stage string) []kargoapi.Promotion { + filtered := make([]kargoapi.Promotion, 0, len(promotions)) + for _, promotion := range promotions { + if promotion.Spec.Stage == stage { + filtered = append(filtered, promotion) + } + } + return filtered +} + +// watchPromotions streams Promotion changes through the REST SSE endpoint. +func (s *server) watchPromotions(c *gin.Context, project, stage, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) // Note: We can't filter by stage using field selector in the watch API. // The indexer is for List operations only. We filter events client-side. - w, err := s.client.Watch(ctx, &kargoapi.PromotionList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.PromotionList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch promotions: %w", err)) return @@ -133,18 +157,26 @@ func (s *server) watchPromotions(c *gin.Context, project, stage string) { logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } promotion, ok := convertWatchEventObject(c, e, (*kargoapi.Promotion)(nil)) if !ok { continue } - // Filter by stage if specified (client-side filtering) - if stage != "" && promotion.Spec.Stage != stage { - continue + eventType := e.Type + if stage != "" { + var send bool + eventType, send = filteredWatchEventType(e.Type, promotion.Spec.Stage == stage) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, promotion) { + if !sendSSEWatchEvent(c, eventType, promotion) { return } } diff --git a/pkg/server/list_stages_v1alpha1.go b/pkg/server/list_stages_v1alpha1.go index 0ee937477b..447e05fa4b 100644 --- a/pkg/server/list_stages_v1alpha1.go +++ b/pkg/server/list_stages_v1alpha1.go @@ -73,30 +73,33 @@ func (s *server) listStages(c *gin.Context) { warehouses := c.QueryArray("freightOrigins") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchStages(c, project, warehouses) + s.watchStages(c, project, warehouses, c.Query("resourceVersion")) return } - items, err := api.ListStagesByWarehouses( - ctx, - s.client, - project, - &api.ListStagesOptions{Warehouses: warehouses}, - ) + list, err := s.listStagesByWarehouses(ctx, project, warehouses) if err != nil { _ = c.Error(err) return } - c.JSON(http.StatusOK, &kargoapi.StageList{Items: items}) + c.JSON(http.StatusOK, list) } -func (s *server) watchStages(c *gin.Context, project string, warehouses []string) { +// watchStages streams Stage changes through the REST SSE endpoint. +func (s *server) watchStages(c *gin.Context, project string, warehouses []string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.StageList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.StageList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch stages: %w", err)) return @@ -124,19 +127,59 @@ func (s *server) watchStages(c *gin.Context, project string, warehouses []string logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } stage, ok := convertWatchEventObject(c, e, (*kargoapi.Stage)(nil)) if !ok { continue } - if len(warehouses) > 0 && !api.StageMatchesAnyWarehouse(stage, warehouses) { - continue + eventType := e.Type + if len(warehouses) > 0 { + var send bool + eventType, send = filteredWatchEventType( + e.Type, + api.StageMatchesAnyWarehouse(stage, warehouses), + ) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, stage) { + if !sendSSEWatchEvent(c, eventType, stage) { return } } } } + +// listStagesByWarehouses lists Stages in the given project, optionally +// filtered to those that request Freight from at least one of the specified +// warehouses (directly or through upstream stages). When warehouses is empty, +// all Stages are returned. The returned StageList carries an effective +// ResourceVersion derived from the list response or listed items. +func (s *server) listStagesByWarehouses( + ctx context.Context, + project string, + warehouses []string, +) (*kargoapi.StageList, error) { + var list kargoapi.StageList + if err := s.listForWatchSeed(ctx, "stages", &list, client.InNamespace(project)); err != nil { + return nil, err + } + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) + if len(warehouses) == 0 { + return &list, nil + } + var stages []kargoapi.Stage + for _, stage := range list.Items { + if api.StageMatchesAnyWarehouse(&stage, warehouses) { + stages = append(stages, stage) + } + } + list.Items = stages + return &list, nil +} diff --git a/pkg/server/list_warehouses_v1alpha1.go b/pkg/server/list_warehouses_v1alpha1.go index bcc0a7a697..fcdf57b260 100644 --- a/pkg/server/list_warehouses_v1alpha1.go +++ b/pkg/server/list_warehouses_v1alpha1.go @@ -69,25 +69,34 @@ func (s *server) listWarehouses(c *gin.Context) { project := c.Param("project") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchWarehouses(c, project) + s.watchWarehouses(c, project, c.Query("resourceVersion")) return } list := &kargoapi.WarehouseList{} - if err := s.client.List(ctx, list, client.InNamespace(project)); err != nil { + if err := s.listForWatchSeed(ctx, "warehouses", list, client.InNamespace(project)); err != nil { _ = c.Error(err) return } + list.ResourceVersion = normalizeListResourceVersion(list.ResourceVersion) c.JSON(http.StatusOK, list) } -func (s *server) watchWarehouses(c *gin.Context, project string) { +// watchWarehouses streams Warehouse changes through the REST SSE endpoint. +func (s *server) watchWarehouses(c *gin.Context, project string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.WarehouseList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.WarehouseList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch warehouses: %w", err)) return @@ -115,6 +124,10 @@ func (s *server) watchWarehouses(c *gin.Context, project string) { logger.Debug("watch channel closed") return } + // convertAndSendWatchEvent surfaces watch.Error events itself, so + // no separate errorFromWatchEvent check is needed here (unlike the + // filtered Stage/Promotion handlers, which inspect the event before + // applying their event-type filter). if !convertAndSendWatchEvent(c, e, (*kargoapi.Warehouse)(nil)) { return } diff --git a/pkg/server/query_freights_v1alpha1.go b/pkg/server/query_freights_v1alpha1.go index d070090acc..d7dce47a3b 100644 --- a/pkg/server/query_freights_v1alpha1.go +++ b/pkg/server/query_freights_v1alpha1.go @@ -31,6 +31,12 @@ type freightGroupList struct { type queryFreightsResponse struct { Groups map[string]freightGroupList `json:"groups"` + // ResourceVersion is the Kubernetes list resourceVersion clients use to + // seed a follow-up Freight watch so the API server does not replay every + // existing Freight as an ADDED event. It is empty for the stage-scoped + // query, whose result is assembled from multiple sources rather than a + // single watchable namespace list. + ResourceVersion string `json:"resourceVersion,omitempty"` } const ( @@ -395,7 +401,7 @@ func (s *server) queryFreight(c *gin.Context) { origins := c.QueryArray("origins") if watchMode := c.Query("watch") == trueStr; watchMode { - s.watchFreight(c, project, origins) + s.watchFreight(c, project, origins, c.Query("resourceVersion")) return } @@ -406,6 +412,7 @@ func (s *server) queryFreight(c *gin.Context) { } var freight []kargoapi.Freight + var resourceVersion string var err error switch { @@ -428,23 +435,32 @@ func (s *server) queryFreight(c *gin.Context) { _ = c.Error(fmt.Errorf("get available freight for stage: %w", err)) return } + // The stage-availability list is assembled from multiple sources rather + // than a single watchable namespace list, so ResourceVersion is left + // empty and clients fall back to an unseeded watch for this view. case len(origins) > 0: - // Get freight from specific warehouses - freight, err = s.getFreightFromWarehousesREST(ctx, project, origins) - if err != nil { - _ = c.Error(fmt.Errorf("get freight from warehouses: %w", err)) + // Get freight from specific warehouses. The seed list goes through the + // uncached reader for a watchable ResourceVersion, then origins are + // filtered in-process (see filterFreightByOrigins) to match the + // unfiltered follow-up watch. + freightList := &kargoapi.FreightList{} + if err = s.listForWatchSeed(ctx, "freights", freightList, client.InNamespace(project)); err != nil { + _ = c.Error(fmt.Errorf("list freight: %w", err)) return } + freight = filterFreightByOrigins(freightList.Items, origins) + resourceVersion = normalizeListResourceVersion(freightList.ResourceVersion) default: // Get all freight in the project freightList := &kargoapi.FreightList{} - if err := s.client.List(ctx, freightList, client.InNamespace(project)); err != nil { + if err = s.listForWatchSeed(ctx, "freights", freightList, client.InNamespace(project)); err != nil { _ = c.Error(fmt.Errorf("list freight: %w", err)) return } freight = freightList.Items + resourceVersion = normalizeListResourceVersion(freightList.ResourceVersion) } // Split the Freight into groups using the generic functions @@ -467,15 +483,28 @@ func (s *server) queryFreight(c *gin.Context) { result[k] = freightGroupList{Items: v} } - c.JSON(http.StatusOK, queryFreightsResponse{Groups: result}) + c.JSON(http.StatusOK, queryFreightsResponse{ + Groups: result, + ResourceVersion: resourceVersion, + }) } -func (s *server) watchFreight(c *gin.Context, project string, origins []string) { +// watchFreight streams Freight changes through the REST SSE endpoint, seeding +// the watch from the provided resourceVersion so the API server does not replay +// every existing Freight as an ADDED event. +func (s *server) watchFreight(c *gin.Context, project string, origins []string, resourceVersion string) { ctx := c.Request.Context() logger := logging.LoggerFromContext(ctx) - w, err := s.client.Watch(ctx, &kargoapi.FreightList{}, client.InNamespace(project)) + w, err := s.client.Watch( + ctx, + &kargoapi.FreightList{}, + buildWatchListOptions(project, resourceVersion)..., + ) if err != nil { + if sendSSEWatchStartError(c, err) { + return + } logger.Error(err, "failed to start watch") _ = c.Error(fmt.Errorf("watch freight: %w", err)) return @@ -503,51 +532,49 @@ func (s *server) watchFreight(c *gin.Context, project string, origins []string) logger.Debug("watch channel closed") return } + if watchErr := errorFromWatchEvent(e); watchErr != nil { + sendSSEWatchError(c, watchErr) + return + } freight, ok := convertWatchEventObject(c, e, (*kargoapi.Freight)(nil)) if !ok { continue } - if len(origins) > 0 && !slices.Contains(origins, freight.Origin.Name) { - continue + eventType := e.Type + if len(origins) > 0 { + var send bool + eventType, send = filteredWatchEventType( + e.Type, + slices.Contains(origins, freight.Origin.Name), + ) + if !send { + continue + } } - if !sendSSEWatchEvent(c, e.Type, freight) { + if !sendSSEWatchEvent(c, eventType, freight) { return } } } } -// getFreightFromWarehousesREST is a helper for the REST endpoint that gets freight from warehouses -func (s *server) getFreightFromWarehousesREST( - ctx context.Context, - project string, - warehouses []string, -) ([]kargoapi.Freight, error) { - var allFreight []kargoapi.Freight - for _, warehouse := range warehouses { - var freight kargoapi.FreightList - if err := s.client.List( - ctx, - &freight, - &client.ListOptions{ - Namespace: project, - FieldSelector: fields.OneTermEqualSelector( - indexer.FreightByWarehouseField, - warehouse, - ), - }, - ); err != nil { - return nil, fmt.Errorf( - "error listing Freight for Warehouse %q in namespace %q: %w", - warehouse, - project, - err, - ) +// filterFreightByOrigins returns Freight whose origin Warehouse is one of the +// provided origins. +// +// Origin filtering is done in-process rather than via the FreightByWarehouse +// field index because the watch-seed list goes through listForWatchSeed's +// uncached reader, which cannot serve controller-runtime field indexes. This +// over-fetches the whole namespace, but it matches the (also unfiltered) +// follow-up watch and keeps the returned ResourceVersion watchable. +func filterFreightByOrigins(freight []kargoapi.Freight, origins []string) []kargoapi.Freight { + filtered := make([]kargoapi.Freight, 0, len(freight)) + for _, f := range freight { + if slices.Contains(origins, f.Origin.Name) { + filtered = append(filtered, f) } - allFreight = append(allFreight, freight.Items...) } - return allFreight, nil + return filtered } diff --git a/pkg/server/resource_version.go b/pkg/server/resource_version.go new file mode 100644 index 0000000000..8c82f22c29 --- /dev/null +++ b/pkg/server/resource_version.go @@ -0,0 +1,34 @@ +package server + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// normalizeListResourceVersion returns a list ResourceVersion safe to seed a +// follow-up Watch with: the value as-is when it is a real version, or "" when +// it is the cached client's "0" sentinel. +// +// The list+watch seed endpoints read through listForWatchSeed's uncached +// reader, which always returns a real list-level resource version. The "0" +// sentinel only appears on the degraded path where no direct reader is wired +// (tests, or no rest.Config) and the cached client is used; there we return "" +// so the client opens an unseeded watch — the pre-change behavior. +func normalizeListResourceVersion(rv string) string { + if rv == "0" { + return "" + } + return rv +} + +// buildWatchListOptions returns namespace-scoped list options for watch calls, +// including ResourceVersion when the caller has one. +func buildWatchListOptions(namespace string, resourceVersion string) []client.ListOption { + watchOpts := []client.ListOption{client.InNamespace(namespace)} + if resourceVersion != "" { + watchOpts = append(watchOpts, &client.ListOptions{ + Raw: &metav1.ListOptions{ResourceVersion: resourceVersion}, + }) + } + return watchOpts +} diff --git a/pkg/server/resource_version_test.go b/pkg/server/resource_version_test.go new file mode 100644 index 0000000000..66871efcbb --- /dev/null +++ b/pkg/server/resource_version_test.go @@ -0,0 +1,26 @@ +package server + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func Test_normalizeListResourceVersion(t *testing.T) { + t.Parallel() + testCases := []struct { + name string + rv string + expected string + }{ + {name: "real rv passthrough", rv: "12345", expected: "12345"}, + {name: "zero sentinel becomes empty", rv: "0", expected: ""}, + {name: "empty stays empty", rv: "", expected: ""}, + } + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tc.expected, normalizeListResourceVersion(tc.rv)) + }) + } +} diff --git a/pkg/server/rest_test.go b/pkg/server/rest_test.go index 4248dc3216..a2664d6e3b 100644 --- a/pkg/server/rest_test.go +++ b/pkg/server/rest_test.go @@ -113,6 +113,7 @@ func testRESTEndpoint( }, ) require.NoError(t, err) + s.authorizeFn = s.client.Authorize s.rolesDB = rbac.NewKubernetesRolesDatabase( s.client, s.client, @@ -226,6 +227,7 @@ func testRESTWatchEndpoint( }, ) require.NoError(t, err) + s.authorizeFn = s.client.Authorize s.rolesDB = rbac.NewKubernetesRolesDatabase( s.client, s.client, diff --git a/pkg/server/watch_helpers.go b/pkg/server/watch_helpers.go index 916581a02b..aeffd3c095 100644 --- a/pkg/server/watch_helpers.go +++ b/pkg/server/watch_helpers.go @@ -2,9 +2,14 @@ package server import ( "encoding/json" + "errors" "fmt" + "net/http" + "connectrpc.com/connect" "github.com/gin-gonic/gin" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/watch" @@ -59,12 +64,105 @@ func convertWatchEventObject[T any](c *gin.Context, e watch.Event, _ T) (T, bool return obj, true } +// errorFromWatchEvent maps Kubernetes watch.Error events into client-visible +// errors that watch handlers can return or stream. +func errorFromWatchEvent(e watch.Event) error { + if e.Type != watch.Error { + return nil + } + + status, ok := e.Object.(*metav1.Status) + if !ok { + return connect.NewError( + connect.CodeUnknown, + fmt.Errorf("watch error: unexpected object type %T", e.Object), + ) + } + + message := status.Message + if message == "" { + message = string(status.Reason) + } + if status.Code == http.StatusGone || status.Reason == metav1.StatusReasonExpired { + return connect.NewError( + connect.CodeOutOfRange, + fmt.Errorf("watch resource version expired: %s", message), + ) + } + return connect.NewError( + connect.CodeUnknown, + fmt.Errorf("watch error: %s", message), + ) +} + +// errorFromWatchStartError maps startup failures from Kubernetes Watch calls +// into the same Connect errors used for watch.Error events. +func errorFromWatchStartError(err error) error { + if err == nil { + return nil + } + if !apierrors.IsResourceExpired(err) { + return err + } + return connect.NewError( + connect.CodeOutOfRange, + fmt.Errorf("watch resource version expired: %s", err.Error()), + ) +} + +// sendSSEWatchError sends a watch error as an SSE error event. Watch endpoints +// may have already sent response headers, so HTTP status codes are no longer a +// reliable way to report expired resource versions after streaming begins. +func sendSSEWatchError(c *gin.Context, err error) { + logger := logging.LoggerFromContext(c.Request.Context()) + + message := err.Error() + var connectErr *connect.Error + if errors.As(err, &connectErr) { + message = connectErr.Message() + } + + event := struct { + Code string `json:"code"` + Message string `json:"message"` + }{ + Code: connect.CodeOf(err).String(), + Message: message, + } + data, marshalErr := json.Marshal(event) + if marshalErr != nil { + logger.Error(marshalErr, "failed to marshal watch error event") + return + } + if _, writeErr := fmt.Fprintf(c.Writer, "event: error\ndata: %s\n\n", data); writeErr != nil { + logger.Debug("failed to write watch error event", "error", writeErr) + return + } + c.Writer.Flush() +} + +// sendSSEWatchStartError sends startup watch errors that are part of the watch +// protocol as SSE error events. It returns true when the error was handled. +func sendSSEWatchStartError(c *gin.Context, err error) bool { + watchErr := errorFromWatchStartError(err) + if connect.CodeOf(watchErr) != connect.CodeOutOfRange { + return false + } + setSSEHeaders(c) + sendSSEWatchError(c, watchErr) + return true +} + // convertAndSendWatchEvent converts a watch event's object to the target type // and sends it as an SSE event. It handles both unstructured objects (from real // clients) and typed objects (from fake clients in tests). Returns true if the // caller should continue processing (including on conversion errors), false if -// the watch should be terminated (write failure). +// the watch should be terminated (write failure or a watch.Error event). func convertAndSendWatchEvent[T any](c *gin.Context, e watch.Event, target T) bool { + if err := errorFromWatchEvent(e); err != nil { + sendSSEWatchError(c, err) + return false + } obj, ok := convertWatchEventObject(c, e, target) if !ok { return true // continue processing, don't terminate watch @@ -72,6 +170,28 @@ func convertAndSendWatchEvent[T any](c *gin.Context, e watch.Event, target T) bo return sendSSEWatchEvent(c, e.Type, obj) } +// filteredWatchEventType returns the event type to send for a client-side +// filtered watch event. Kubernetes server-side selectors send a DELETED event +// when a previously matching object is modified so it no longer matches; this +// helper mirrors that behavior for filters we must evaluate in-process. +// +// Because we hold no per-client matched set, we cannot tell whether a +// non-matching MODIFIED object previously matched, so we emit a synthetic +// DELETED for every non-matching MODIFIED event — including objects the client +// never received an ADDED for. This over-emits DELETEs relative to a real +// server-side selector, but a DELETE for an object the client is not tracking +// is a harmless no-op. Achieving exact fidelity would require tracking sent +// object identities per client, which is not worth the complexity here. +func filteredWatchEventType(eventType watch.EventType, matches bool) (watch.EventType, bool) { + if matches { + return eventType, true + } + if eventType == watch.Modified { + return watch.Deleted, true + } + return "", false +} + // sendSSEWatchEvent sends an object as an SSE watch event. Returns true if // the caller should continue processing, false if the watch should be // terminated (write failure). diff --git a/pkg/server/watch_helpers_test.go b/pkg/server/watch_helpers_test.go new file mode 100644 index 0000000000..8499faae49 --- /dev/null +++ b/pkg/server/watch_helpers_test.go @@ -0,0 +1,136 @@ +package server + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + + "connectrpc.com/connect" + "github.com/gin-gonic/gin" + "github.com/stretchr/testify/require" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" +) + +func Test_errorFromWatchStartError(t *testing.T) { + t.Parallel() + + t.Run("passes through ordinary errors", func(t *testing.T) { + t.Parallel() + err := errors.New("network unavailable") + + require.ErrorIs(t, errorFromWatchStartError(err), err) + }) + + t.Run("maps expired resource versions to out of range", func(t *testing.T) { + t.Parallel() + err := apierrors.NewResourceExpired("too old resource version") + + mapped := errorFromWatchStartError(err) + + require.Error(t, mapped) + require.Equal(t, connect.CodeOutOfRange, connect.CodeOf(mapped)) + require.ErrorContains(t, mapped, "watch resource version expired") + }) + + t.Run("returns nil for nil", func(t *testing.T) { + t.Parallel() + require.NoError(t, errorFromWatchStartError(nil)) + }) +} + +func Test_errorFromWatchStartError_withStatusError(t *testing.T) { + t.Parallel() + + err := apierrors.NewNotFound( + schema.GroupResource{Group: "kargo.akuity.io", Resource: "promotions"}, + "promotion-1", + ) + + mapped := errorFromWatchStartError(err) + + require.ErrorIs(t, mapped, err) + require.Equal(t, connect.CodeUnknown, connect.CodeOf(mapped)) +} + +func Test_sendSSEWatchError(t *testing.T) { + t.Parallel() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/", nil) + + err := connect.NewError(connect.CodeOutOfRange, errors.New("resource version expired")) + + sendSSEWatchError(c, err) + + require.Contains(t, recorder.Body.String(), "event: error") + require.Contains(t, recorder.Body.String(), `"code":"out_of_range"`) + require.Contains(t, recorder.Body.String(), `"message":"resource version expired"`) + require.NotContains(t, recorder.Body.String(), "out_of_range: resource version expired") +} + +func Test_convertAndSendWatchEvent_errorEvent(t *testing.T) { + t.Parallel() + + recorder := httptest.NewRecorder() + c, _ := gin.CreateTestContext(recorder) + c.Request = httptest.NewRequest(http.MethodGet, "/", nil) + + keepGoing := convertAndSendWatchEvent(c, watch.Event{ + Type: watch.Error, + Object: &metav1.Status{ + Status: metav1.StatusFailure, + Message: "too old resource version", + Reason: metav1.StatusReasonExpired, + Code: http.StatusGone, + }, + }, (*metav1.PartialObjectMetadata)(nil)) + + require.False(t, keepGoing) + require.Contains(t, recorder.Body.String(), "event: error") + require.Contains(t, recorder.Body.String(), `"code":"out_of_range"`) +} + +func Test_filteredWatchEventType(t *testing.T) { + t.Parallel() + + testCases := []struct { + name string + eventType watch.EventType + matches bool + wantType watch.EventType + wantSend bool + }{ + { + name: "matching event passes through", + eventType: watch.Modified, + matches: true, + wantType: watch.Modified, + wantSend: true, + }, + { + name: "non-matching modified becomes delete", + eventType: watch.Modified, + wantType: watch.Deleted, + wantSend: true, + }, + { + name: "non-matching added is skipped", + eventType: watch.Added, + }, + } + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + t.Parallel() + + eventType, send := filteredWatchEventType(testCase.eventType, testCase.matches) + + require.Equal(t, testCase.wantSend, send) + require.Equal(t, testCase.wantType, eventType) + }) + } +} diff --git a/swagger.json b/swagger.json index 81d0587bc7..f8d82d2ae9 100644 --- a/swagger.json +++ b/swagger.json @@ -8668,6 +8668,10 @@ "additionalProperties": { "$ref": "#/definitions/pkg_server.freightGroupList" } + }, + "resourceVersion": { + "description": "ResourceVersion is the Kubernetes list resourceVersion clients use to\nseed a follow-up Freight watch so the API server does not replay every\nexisting Freight as an ADDED event. It is empty for the stage-scoped\nquery, whose result is assembled from multiple sources rather than a\nsingle watchable namespace list.", + "type": "string" } } }, diff --git a/ui/src/features/project/pipelines/pipelines.tsx b/ui/src/features/project/pipelines/pipelines.tsx index 45b7855192..1df4dc03fb 100644 --- a/ui/src/features/project/pipelines/pipelines.tsx +++ b/ui/src/features/project/pipelines/pipelines.tsx @@ -147,7 +147,7 @@ export const Pipelines = (props: { creatingStage?: boolean; creatingWarehouse?: usePersistPreferredFilter(projectName || '', preferredFilter); - useWatchFreight(projectName || ''); + useWatchFreight(projectName || '', preferredFilter.warehouses, !getFreightQuery.isLoading); if (loading) { return ; diff --git a/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts b/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts index 7d2647a0b7..d2b121c842 100644 --- a/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts +++ b/ui/src/features/project/pipelines/promotion/use-watch-promotion.ts @@ -20,6 +20,9 @@ export const useWatchPromotion = (project: string, promotion: string) => { (async () => { for await (const event of readSSEStream(url, abort.signal)) { + if ('error' in event) { + continue; + } const p = event.object; if (event.type === 'DELETED') { diff --git a/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts b/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts index 6537733ba2..43e92caf9d 100644 --- a/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts +++ b/ui/src/features/project/pipelines/promotion/use-watch-promotions.ts @@ -4,40 +4,63 @@ import { useEffect } from 'react'; import { getListPromotionsQueryKey, listPromotionsResponse } from '@ui/gen/api/v2/core/core'; import { Promotion } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from '../watch-utils'; +import { runSeededWatch, upsertOrDelete } from '../watch-utils'; -export const useWatchPromotions = (project: string, stage: string) => { +// enabled gates the watch on the initial list having loaded, so it never opens +// before a seed resourceVersion is available — otherwise it would start an +// unseeded watch that replays every Promotion. (Unlike the pipeline page, this +// page has no loading gate above the watch, so the caller must pass it.) +export const useWatchPromotions = (project: string, stage: string, enabled = true) => { const client = useQueryClient(); useEffect(() => { - if (!project || !stage) { + if (!project || !stage || !enabled) { return; } const abort = new AbortController(); - const params = new URLSearchParams({ watch: 'true', stage }); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/promotions?${params}`; const listKey = getListPromotionsQueryKey(project, { stage }); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const promotion = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listPromotionsResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueryData(listKey, (old: listPromotionsResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { - ...old.data, - items: upsertOrDelete(old.data.items ?? [], promotion, event.type) - } - }; - }); + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true', stage }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); } - })(); + return `/v1beta1/projects/${encodeURIComponent(project)}/promotions?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, promotion: Promotion) => { + client.setQueryData(listKey, (old: listPromotionsResponse | undefined) => { + if (!old?.data) { + return old; + } + return { + ...old, + data: { + ...old.data, + items: upsertOrDelete(old.data.items ?? [], promotion, type) + } + }; + }); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => abort.abort(); - }, [project, stage, client]); + }, [project, stage, client, enabled]); }; diff --git a/ui/src/features/project/pipelines/use-watch-freight.ts b/ui/src/features/project/pipelines/use-watch-freight.ts index 8278b2e957..affefd5048 100644 --- a/ui/src/features/project/pipelines/use-watch-freight.ts +++ b/ui/src/features/project/pipelines/use-watch-freight.ts @@ -9,59 +9,83 @@ import { } from '@ui/gen/api/v2/core/core'; import { Freight } from '@ui/gen/api/v2/models'; -import { readSSEStream, upsertOrDelete } from './watch-utils'; +import { runSeededWatch, upsertOrDelete } from './watch-utils'; -export const useWatchFreight = (project: string) => { +// origins must match the params the page's useQueryFreightsRest uses, so the +// seed resourceVersion is read from the same query cache entry. enabled gates +// the watch on the initial list having loaded, so it never opens before a seed +// resourceVersion is available (which would start an unseeded, replaying watch). +export const useWatchFreight = (project: string, origins?: string[], enabled = true) => { const client = useQueryClient(); useEffect(() => { - if (!project) { + if (!project || !enabled) { return; } const abort = new AbortController(); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/freight?watch=true`; - const listKey = getQueryFreightsRestQueryKey(project, undefined); + const listKey = getQueryFreightsRestQueryKey(project, { origins }); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const freight = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as queryFreightsRestResponse | undefined)?.data + ?.resourceVersion; - // Update all queryFreight caches for this project, including - // warehouse-filtered variants, which use a different cache key. - // Using setQueriesData (rather than setQueryData) ensures we only - // touch caches that already have an active query backing them, - // avoiding orphaned entries with no queryFn that would crash on - // refetch. - client.setQueriesData( - { queryKey: listKey, exact: false }, - (old) => { - if (!old?.data?.groups) { - return old; - } - const updatedGroups = Object.fromEntries( - Object.entries(old.data.groups).map(([key, group]) => [ - key, - { ...group, items: upsertOrDelete(group.items ?? [], freight, event.type) } - ]) - ); - return { ...old, data: { ...old.data, groups: updatedGroups } }; - } - ); + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/freight?${params}`; + }; - const freightKey = getGetFreightQueryKey(project, freight.metadata?.name); + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: freightKey }); - } else { - client.setQueryData(freightKey, (old: getFreightResponse | undefined) => ({ - ...old, - data: freight - })); + const onEvent = (type: string, freight: Freight) => { + // Update all queryFreight caches for this project, including + // warehouse-filtered variants, which use a different cache key. + // Using setQueriesData (rather than setQueryData) ensures we only + // touch caches that already have an active query backing them, + // avoiding orphaned entries with no queryFn that would crash on + // refetch. + client.setQueriesData( + { queryKey: getQueryFreightsRestQueryKey(project, undefined), exact: false }, + (old) => { + if (!old?.data?.groups) { + return old; + } + const updatedGroups = Object.fromEntries( + Object.entries(old.data.groups).map(([key, group]) => [ + key, + { ...group, items: upsertOrDelete(group.items ?? [], freight, type) } + ]) + ); + return { ...old, data: { ...old.data, groups: updatedGroups } }; } + ); + + const freightKey = getGetFreightQueryKey(project, freight.metadata?.name); + + if (type === 'DELETED') { + client.removeQueries({ queryKey: freightKey }); + } else { + client.setQueryData(freightKey, (old: getFreightResponse | undefined) => ({ + ...old, + data: freight + })); } - })(); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => abort.abort(); - }, [project]); + }, [project, client, enabled, (origins || []).join(',')]); }; diff --git a/ui/src/features/project/pipelines/use-watch-stages.ts b/ui/src/features/project/pipelines/use-watch-stages.ts index 1de642ed0c..2a2cf528aa 100644 --- a/ui/src/features/project/pipelines/use-watch-stages.ts +++ b/ui/src/features/project/pipelines/use-watch-stages.ts @@ -9,7 +9,7 @@ import { } from '@ui/gen/api/v2/core/core'; import { Stage } from '@ui/gen/api/v2/models'; -import { debounce, readSSEStream, upsertOrDelete } from './watch-utils'; +import { debounce, runSeededWatch, upsertOrDelete } from './watch-utils'; export const useWatchStages = ( project: string, @@ -24,58 +24,70 @@ export const useWatchStages = ( } const abort = new AbortController(); - - const params = new URLSearchParams({ watch: 'true' }); - for (const wh of warehouses || []) { - params.append('freightOrigins', wh); - } - const url = `/v1beta1/projects/${encodeURIComponent(project)}/stages?${params}`; const listKey = getListStagesQueryKey(project, { freightOrigins: warehouses || [] }); // Coalesce bursts of stage events so the graph recompute is triggered once // per burst rather than once per event. const emitStageEvent = debounce((stage: Stage) => onStageEvent?.(stage)); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const stage = event.object; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listStagesResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueriesData( - { exact: false, queryKey: listKey }, - (old: listStagesResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { ...old.data, items: upsertOrDelete(old.data.items ?? [], stage, event.type) } - }; + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + for (const wh of warehouses || []) { + params.append('freightOrigins', wh); + } + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/stages?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, stage: Stage) => { + client.setQueriesData( + { exact: false, queryKey: listKey }, + (old: listStagesResponse | undefined) => { + if (!old?.data) { + return old; } - ); + return { + ...old, + data: { ...old.data, items: upsertOrDelete(old.data.items ?? [], stage, type) } + }; + } + ); - const stageKey = getGetStageQueryKey(project, stage.metadata?.name); - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: stageKey }); - } else { - client.setQueriesData( - { exact: false, queryKey: stageKey }, - (old: getStageResponse | undefined) => - old - ? { - ...old, - data: { - // WATCH ENDPOINT STAGE COMES WITHOUT kind, apiVersion etc.. - // SO WE NEED TO PRESERVE IT FROM INITIAL DATA - ...old?.data, - ...stage - } + const stageKey = getGetStageQueryKey(project, stage.metadata?.name); + if (type === 'DELETED') { + client.removeQueries({ queryKey: stageKey }); + } else { + client.setQueriesData( + { exact: false, queryKey: stageKey }, + (old: getStageResponse | undefined) => + old + ? { + ...old, + data: { + // WATCH ENDPOINT STAGE COMES WITHOUT kind, apiVersion etc.. + // SO WE NEED TO PRESERVE IT FROM INITIAL DATA + ...old?.data, + ...stage } - : old - ); - emitStageEvent.call(stage); - } + } + : old + ); + emitStageEvent.call(stage); } - })(); + }; + + runSeededWatch({ signal: abort.signal, buildUrl, seedResourceVersion, relist, onEvent }); return () => { abort.abort(); diff --git a/ui/src/features/project/pipelines/use-watch-warehouses.ts b/ui/src/features/project/pipelines/use-watch-warehouses.ts index a2fe59900f..f253f5c400 100644 --- a/ui/src/features/project/pipelines/use-watch-warehouses.ts +++ b/ui/src/features/project/pipelines/use-watch-warehouses.ts @@ -9,7 +9,7 @@ import { } from '@ui/gen/api/v2/core/core'; import { Warehouse } from '@ui/gen/api/v2/models'; -import { debounce, readSSEStream, upsertOrDelete } from './watch-utils'; +import { debounce, runSeededWatch, upsertOrDelete } from './watch-utils'; export const useWatchWarehouses = ( project: string, @@ -26,7 +26,6 @@ export const useWatchWarehouses = ( } const abort = new AbortController(); - const url = `/v1beta1/projects/${encodeURIComponent(project)}/warehouses?watch=true`; const listKey = getListWarehousesQueryKey(project); const pendingRefresh: Record = {}; @@ -36,66 +35,88 @@ export const useWatchWarehouses = ( opts?.onWarehouseEvent?.(warehouse) ); - (async () => { - for await (const event of readSSEStream(url, abort.signal)) { - const warehouse = event.object; - const name = warehouse.metadata?.name || ''; + const seedResourceVersion = () => + (client.getQueryData(listKey) as listWarehousesResponse | undefined)?.data?.metadata + ?.resourceVersion; - client.setQueriesData( - { exact: false, queryKey: listKey }, - (old: listWarehousesResponse | undefined) => { - if (!old?.data) { - return old; - } - return { - ...old, - data: { - ...old.data, - items: upsertOrDelete(old.data.items ?? [], warehouse, event.type) - } - }; + const buildUrl = (resourceVersion: string) => { + const params = new URLSearchParams({ watch: 'true' }); + if (resourceVersion) { + params.append('resourceVersion', resourceVersion); + } + return `/v1beta1/projects/${encodeURIComponent(project)}/warehouses?${params}`; + }; + + const relist = async () => { + await client.refetchQueries({ queryKey: listKey, exact: false }); + return seedResourceVersion(); + }; + + const onEvent = (type: string, warehouse: Warehouse) => { + const name = warehouse.metadata?.name || ''; + + client.setQueriesData( + { exact: false, queryKey: listKey }, + (old: listWarehousesResponse | undefined) => { + if (!old?.data) { + return old; } - ); + return { + ...old, + data: { + ...old.data, + items: upsertOrDelete(old.data.items ?? [], warehouse, type) + } + }; + } + ); + + const warehouseKey = getGetWarehouseQueryKey(project, name); - const warehouseKey = getGetWarehouseQueryKey(project, name); - - if (event.type === 'DELETED') { - client.removeQueries({ queryKey: warehouseKey }); - } else { - client.setQueriesData( - { - exact: false, - queryKey: warehouseKey - }, - (old: getWarehouseResponse | undefined) => - old - ? { - ...old, - data: { - // WATCH ENDPOINT WAREHOUSE COMES WITHOUT kind, apiVersion etc.. - // SO WE NEED TO PRESERVE IT FROM INITIAL DATA - ...old?.data, - ...warehouse - } + if (type === 'DELETED') { + client.removeQueries({ queryKey: warehouseKey }); + } else { + client.setQueriesData( + { + exact: false, + queryKey: warehouseKey + }, + (old: getWarehouseResponse | undefined) => + old + ? { + ...old, + data: { + // WATCH ENDPOINT WAREHOUSE COMES WITHOUT kind, apiVersion etc.. + // SO WE NEED TO PRESERVE IT FROM INITIAL DATA + ...old?.data, + ...warehouse } - : old - ); - - const refreshRequest = warehouse.metadata?.annotations?.['kargo.akuity.io/refresh']; - const refreshStatus = warehouse.status?.lastHandledRefresh; - const isRefreshing = refreshRequest !== undefined && refreshRequest !== refreshStatus; - - if (isRefreshing) { - pendingRefresh[name] = true; - } else if (pendingRefresh[name]) { - delete pendingRefresh[name]; - opts?.refreshHook?.(); - } + } + : old + ); - emitWarehouseEvent.call(warehouse); + const refreshRequest = warehouse.metadata?.annotations?.['kargo.akuity.io/refresh']; + const refreshStatus = warehouse.status?.lastHandledRefresh; + const isRefreshing = refreshRequest !== undefined && refreshRequest !== refreshStatus; + + if (isRefreshing) { + pendingRefresh[name] = true; + } else if (pendingRefresh[name]) { + delete pendingRefresh[name]; + opts?.refreshHook?.(); } + + emitWarehouseEvent.call(warehouse); } - })(); + }; + + runSeededWatch({ + signal: abort.signal, + buildUrl, + seedResourceVersion, + relist, + onEvent + }); return () => { abort.abort(); diff --git a/ui/src/features/project/pipelines/watch-utils.ts b/ui/src/features/project/pipelines/watch-utils.ts index 7f0665596f..0f58f7d8af 100644 --- a/ui/src/features/project/pipelines/watch-utils.ts +++ b/ui/src/features/project/pipelines/watch-utils.ts @@ -2,7 +2,14 @@ import { authTokenKey } from '@ui/config/auth'; export const getBaseUrl = () => (import.meta.env.VITE_API_URL as string | undefined) || ''; -export type SSEWatchEvent = { type: string; object: T }; +// Code the server emits (as `event: error`) when the resourceVersion used to +// seed a watch is older than the API server's watch window. Clients respond by +// relisting to obtain a fresh, watchable resourceVersion. +export const WATCH_ERROR_EXPIRED = 'out_of_range'; + +export type SSEWatchError = { code: string; message: string }; + +export type SSEWatchEvent = { type: string; object: T } | { error: SSEWatchError }; export async function* readSSEStream( url: string, @@ -33,12 +40,17 @@ export async function* readSSEStream( buffer = parts.pop() ?? ''; for (const part of parts) { - const dataLine = part.split('\n').find((l) => l.startsWith('data: ')); + const lines = part.split('\n'); + const dataLine = lines.find((l) => l.startsWith('data: ')); if (!dataLine) { continue; } + const isError = lines.some((l) => l.startsWith('event: error')); try { - yield JSON.parse(dataLine.slice(6)) as SSEWatchEvent; + const parsed = JSON.parse(dataLine.slice(6)); + yield isError + ? { error: parsed as SSEWatchError } + : (parsed as { type: string; object: T }); } catch (_) { // skip malformed events } @@ -49,6 +61,67 @@ export async function* readSSEStream( } } +// runSeededWatch opens a Kubernetes-style list-then-watch SSE stream seeded from +// the list's resourceVersion, so the API server does not replay every existing +// object as an ADDED event. The seed resourceVersion is read once per connection +// from `seedResourceVersion()` (kept out of any React effect deps by the caller +// so a cache-advancing event never tears the stream down). If the seed is too +// old, the server reports `WATCH_ERROR_EXPIRED`; we then `relist()` for a fresh +// resourceVersion and reopen. A clean close or generic error stops the stream +// (matching the pre-seeding behavior — no auto-reconnect). +export async function runSeededWatch(params: { + signal: AbortSignal; + buildUrl: (resourceVersion: string) => string; + seedResourceVersion: () => string | undefined; + relist: () => Promise; + onEvent: (type: string, object: T) => void; +}): Promise { + let resourceVersion = params.seedResourceVersion() || ''; + + for (;;) { + let expired = false; + try { + for await (const event of readSSEStream(params.buildUrl(resourceVersion), params.signal)) { + if ('error' in event) { + if (event.error.code === WATCH_ERROR_EXPIRED) { + expired = true; + break; + } + continue; + } + params.onEvent(event.type, event.object); + } + } catch (_) { + // Aborted unmount or a network failure: stop. + return; + } + + if (params.signal.aborted || !expired) { + return; + } + + resourceVersion = (await params.relist()) || ''; + } +} + +// isNewerResourceVersion reports whether incoming is a strictly newer Kubernetes +// resourceVersion than existing. resourceVersions are officially opaque, so we +// only compare numerically (their actual encoding today) and otherwise fail open +// (treat as newer) to avoid suppressing a real update. +function isNewerResourceVersion(incoming?: string, existing?: string): boolean { + if (!incoming) { + return false; + } + if (!existing) { + return true; + } + try { + return BigInt(incoming) > BigInt(existing); + } catch (_) { + return true; + } +} + // Reads an SSE stream of raw text chunks (e.g. AnalysisRun logs). The server // writes each line of a chunk as its own `data:` line, so rejoining the data // lines with `\n` reconstructs the original chunk verbatim. Unlike @@ -123,16 +196,25 @@ export function debounce( }; } -export function upsertOrDelete( - items: T[], - item: T, - eventType: string -): T[] { +export function upsertOrDelete< + T extends { metadata?: { name?: string; resourceVersion?: string } } +>(items: T[], item: T, eventType: string): T[] { const index = items.findIndex((i) => i.metadata?.name === item.metadata?.name); if (eventType === 'DELETED') { return index !== -1 ? [...items.slice(0, index), ...items.slice(index + 1)] : items; } - return index !== -1 - ? [...items.slice(0, index), item, ...items.slice(index + 1)] - : [...items, item]; + if (index === -1) { + return [...items, item]; + } + // Skip a replayed ADDED for an object we already hold at the same or newer + // resourceVersion. Seeding should prevent these server-side; this is a + // belt-and-suspenders guard for an unseeded or just-relisted watch. MODIFIED + // events always carry a newer resourceVersion and are applied unconditionally. + if ( + eventType === 'ADDED' && + !isNewerResourceVersion(item.metadata?.resourceVersion, items[index].metadata?.resourceVersion) + ) { + return items; + } + return [...items.slice(0, index), item, ...items.slice(index + 1)]; } diff --git a/ui/src/features/stage/promotions.tsx b/ui/src/features/stage/promotions.tsx index 32fb482b85..7c89abea0f 100644 --- a/ui/src/features/stage/promotions.tsx +++ b/ui/src/features/stage/promotions.tsx @@ -51,7 +51,7 @@ export const Promotions = ({ argocdShard }: { argocdShard?: ArgoCDShard }) => { // modal kept in the same component for live view const [selectedPromotion, setSelectedPromotion] = useState(); - useWatchPromotions(projectName || '', stageName || ''); + useWatchPromotions(projectName || '', stageName || '', !listPromotionsQuery.isLoading); const promotions = React.useMemo(() => { // Immutable sorting diff --git a/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts b/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts index ccc87d29a1..3fe472cf69 100644 --- a/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts +++ b/ui/src/gen/api/v2/models/pkgServerQueryFreightsResponse.ts @@ -9,4 +9,10 @@ import type { PkgServerQueryFreightsResponseGroups } from './pkgServerQueryFreig export interface PkgServerQueryFreightsResponse { groups?: PkgServerQueryFreightsResponseGroups; + /** ResourceVersion is the Kubernetes list resourceVersion clients use to +seed a follow-up Freight watch so the API server does not replay every +existing Freight as an ADDED event. It is empty for the stage-scoped +query, whose result is assembled from multiple sources rather than a +single watchable namespace list. */ + resourceVersion?: string; }