diff --git a/api/query/v1/query.go b/api/query/v1/query.go index 10b05a8..27c2f47 100644 --- a/api/query/v1/query.go +++ b/api/query/v1/query.go @@ -192,6 +192,7 @@ type DryRunResult struct { Applied int `json:"applied"` Conflicts []DryRunItem `json:"conflicts"` Errors []DryRunItem `json:"errors"` + Skipped []DryRunItem `json:"skipped,omitempty"` OK bool `json:"ok"` } @@ -213,6 +214,7 @@ type ReplayRun struct { FinishedAt *time.Time `json:"finished_at,omitempty"` Applied int `json:"applied"` Errors []DryRunItem `json:"errors,omitempty"` + Skipped []DryRunItem `json:"skipped,omitempty"` } // Health is the server health response. diff --git a/cmd/krply-server/main.go b/cmd/krply-server/main.go index eb8b17c..18deea5 100644 --- a/cmd/krply-server/main.go +++ b/cmd/krply-server/main.go @@ -19,9 +19,10 @@ import ( "github.com/krply/krply/internal/metrics" "github.com/krply/krply/internal/replay" "github.com/krply/krply/internal/storage" + "github.com/krply/krply/internal/version" ) -const version = "0.1.0" +var buildVersion = version.Version func main() { if err := run(); err != nil { @@ -86,7 +87,7 @@ func run() error { flag.Parse() if *showVer { - fmt.Println(version) + fmt.Println(buildVersion) return nil } @@ -123,17 +124,40 @@ func run() error { m := metrics.New() m.RefreshFromStore(ctx, store) + planner.SetMetrics(m) - srv, err := api.NewServer(store, mat, planner, m, version) + // Store-derived gauges are refreshed on a ticker; nothing else in the + // process would update degraded streams, gap counts, or store size. + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + m.RefreshFromStore(ctx, store) + } + } + }() + + srv, err := api.NewServer(store, mat, planner, m, buildVersion) if err != nil { return fmt.Errorf("build api server: %w", err) } - httpSrv := &http.Server{Addr: listenAddr, Handler: srv.Handler()} + httpSrv := &http.Server{ + Addr: listenAddr, + Handler: srv.Handler(), + ReadHeaderTimeout: 10 * time.Second, + ReadTimeout: 60 * time.Second, + WriteTimeout: 2 * time.Minute, + IdleTimeout: 2 * time.Minute, + } errCh := make(chan error, 1) go func() { - slog.Info("krply-server listening", "addr", listenAddr, "store", *storePath, "version", version) + slog.Info("krply-server listening", "addr", listenAddr, "store", *storePath, "version", buildVersion) errCh <- httpSrv.ListenAndServe() }() diff --git a/cmd/krply/coverage.go b/cmd/krply/coverage.go index b487d07..5b882e7 100644 --- a/cmd/krply/coverage.go +++ b/cmd/krply/coverage.go @@ -115,18 +115,26 @@ func coverageServer(ctx context.Context, streamID string) error { q := url.Values{} q.Set("stream_id", streamID) q.Set("record_type", string(event.TypeGap)) - var page queryv1.EventPage - if err := c.get(withQuery("/v1/events", q), &page); err != nil { - return err + var items []queryv1.Event + for { + var page queryv1.EventPage + if err := c.get(withQuery("/v1/events", q), &page); err != nil { + return err + } + items = append(items, page.Items...) + if !page.HasMore || page.NextCursor == "" { + break + } + q.Set("cursor", page.NextCursor) } - if len(page.Items) == 0 { + if len(items) == 0 { out("no gaps recorded\n") return nil } - out("GAPS (%d):\n", len(page.Items)) + out("GAPS (%d):\n", len(items)) w := tabwriter.NewWriter(os.Stdout, 0, 4, 2, ' ', 0) fmt.Fprintln(w, "FROM-RV\tTO-RV\tREASON\tDETECTED") - for _, it := range page.Items { + for _, it := range items { from, to, reason := gapFromEvent(it) fmt.Fprintf(w, "%s\t%s\t%s\t%s\n", from, to, reason, it.ObservedAt.Format(time.RFC3339)) } diff --git a/cmd/krply/diff.go b/cmd/krply/diff.go index 11e2197..bbac41a 100644 --- a/cmd/krply/diff.go +++ b/cmd/krply/diff.go @@ -51,6 +51,9 @@ func runDiff(cmd *cobra.Command, args []string) error { if since.IsZero() || until.IsZero() { return errors.New("diff requires both --since and --until") } + if since.After(until) { + return errors.New("diff: --since must not be after --until") + } if serverURL != "" { return diffServer(cmd.Context(), since, until) diff --git a/cmd/krply/export.go b/cmd/krply/export.go index dcd763e..22d6633 100644 --- a/cmd/krply/export.go +++ b/cmd/krply/export.go @@ -57,7 +57,9 @@ func runExport(cmd *cobra.Command, args []string) error { var w io.Writer = os.Stdout if exportOut != "" { - fh, err := os.Create(exportOut) + // Journal exports can contain recorded ConfigMaps/Secrets, so the file + // is created with owner-only permissions. + fh, err := os.OpenFile(exportOut, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o600) if err != nil { return err } diff --git a/cmd/krply/record.go b/cmd/krply/record.go index f286c4f..2b2ca1c 100644 --- a/cmd/krply/record.go +++ b/cmd/krply/record.go @@ -120,7 +120,6 @@ var clusterScopedKinds = map[string]bool{ "ValidatingWebhookConfiguration": true, "PriorityClass": true, "Node": true, - "PersistentVolumeClaim": true, } // applyRecordNamespace scopes each namespaced resource to the requested diff --git a/cmd/krply/replay.go b/cmd/krply/replay.go index de3858e..e37dc9a 100644 --- a/cmd/krply/replay.go +++ b/cmd/krply/replay.go @@ -41,7 +41,7 @@ var replayPlanCmd = &cobra.Command{ var replayApplyCmd = &cobra.Command{ Use: "apply", - Short: "dry-run and apply a replay plan (requires --server)", + Short: "dry-run and apply a replay plan (local, or via --server)", Args: cobra.NoArgs, RunE: runReplayApply, } @@ -59,8 +59,12 @@ func init() { f.BoolVar(&replayAllowGaps, "allow-gaps", false, "allow a plan with incomplete coverage") g := replayApplyCmd.Flags() - g.StringVar(&replayPlanID, "plan-id", "", "plan id to apply") + g.StringVar(&replayPlanID, "plan-id", "", "plan id to apply (server mode)") + g.StringVar(&replaySnapshotID, "snapshot", "", "snapshot id to re-plan and apply (local mode)") + g.StringVar(&replaySourceNS, "source-namespace", "", "source namespace to replay (empty = all)") + g.StringVar(&replayTargetNS, "target-namespace", "", "target namespace to apply into") g.StringVar(&replayTargetContext, "target-context", "", "target kubeconfig context") + g.BoolVar(&replayAllowGaps, "allow-gaps", false, "allow a plan with incomplete coverage") g.BoolVar(&replayConfirm, "confirm", false, "confirm the apply") } @@ -101,16 +105,73 @@ func runReplayPlan(cmd *cobra.Command, args []string) error { } func runReplayApply(cmd *cobra.Command, args []string) error { - if replayPlanID == "" { - return errors.New("replay apply requires --plan-id") - } if !replayConfirm { return errors.New("refusing apply without --confirm") } - if serverURL == "" { - return errors.New("replay apply requires --server pointing at krply-server") + if serverURL != "" { + if replayPlanID == "" { + return errors.New("replay apply --server requires --plan-id") + } + return replayApplyServer(cmd.Context()) + } + return replayApplyLocal(cmd) +} + +// replayApplyLocal replans the snapshot from the local journal, dry-runs it, +// and applies it directly. This makes the natural plan-then-apply flow work +// without a server. +func replayApplyLocal(cmd *cobra.Command) error { + if replaySnapshotID == "" { + return errors.New("replay apply (local) requires --snapshot") + } + store, err := openStore(storePath) + if err != nil { + return err + } + defer closeStore(store) + + clusters, err := store.ListClusters(cmd.Context()) + if err != nil { + return err + } + clusterID, err := firstClusterID(clusters) + if err != nil { + return err } + mat := materialize.NewMaterializer(store) + policy := replay.DefaultPolicy() + policy.AllowGaps = replayAllowGaps + planner := replay.NewPlanner(store, mat, policy) + plan, err := planner.Plan(cmd.Context(), clusterID, replaySnapshotID, replaySourceNS, replayTargetNS) + if err != nil { + return err + } + printPlan(plan) + + dry, err := plan.DryRun(cmd.Context(), kubeconfig, replayTargetContext) + if err != nil { + return err + } + printDryRunResult(dry) + if !dry.OK { + return errors.New("refusing apply: dry run did not pass") + } + res, err := plan.Apply(cmd.Context(), kubeconfig, replayTargetContext, true) + if err != nil { + return err + } + out("applied %d objects\n", res.Applied) + for _, e := range res.Errors { + warn("apply error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } + for _, e := range res.Skipped { + warn("skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } + return nil +} + +func replayApplyServer(ctx context.Context) error { c := newAPIClient(serverURL) var dry queryv1.DryRunResult @@ -120,13 +181,19 @@ func runReplayApply(cmd *cobra.Command, args []string) error { }, &dry); err != nil { return err } - out("dry run: applied=%d conflicts=%d errors=%d ok=%v\n", dry.Applied, len(dry.Conflicts), len(dry.Errors), dry.OK) + out("dry run: applied=%d conflicts=%d errors=%d skipped=%d ok=%v\n", dry.Applied, len(dry.Conflicts), len(dry.Errors), len(dry.Skipped), dry.OK) for _, ci := range dry.Conflicts { warn("conflict: %s/%s (%s): %s\n", ci.Namespace, ci.Name, ci.Kind, ci.Message) } for _, e := range dry.Errors { warn("dry-run error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) } + for _, e := range dry.Skipped { + warn("dry-run skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } + if !dry.OK { + return errors.New("refusing apply: dry run did not pass") + } var run queryv1.ReplayRun if err := c.post("/v1/replay-runs", map[string]any{ @@ -140,12 +207,28 @@ func runReplayApply(cmd *cobra.Command, args []string) error { for _, e := range run.Errors { warn("apply error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) } + for _, e := range run.Skipped { + warn("skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } if run.Status != "" { out("status: %s\n", run.Status) } return nil } +func printDryRunResult(d *replay.DryRunResult) { + out("dry run: applied=%d conflicts=%d errors=%d skipped=%d ok=%v\n", d.Applied, len(d.Conflicts), len(d.Errors), len(d.Skipped), d.OK) + for _, ci := range d.Conflicts { + warn("conflict: %s/%s (%s): %s\n", ci.Namespace, ci.Name, ci.Kind, ci.Message) + } + for _, e := range d.Errors { + warn("dry-run error: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } + for _, e := range d.Skipped { + warn("dry-run skipped: %s/%s (%s): %s\n", e.Namespace, e.Name, e.Kind, e.Message) + } +} + func replayPlanServer(ctx context.Context) error { c := newAPIClient(serverURL) var plan queryv1.ReplayPlan diff --git a/cmd/krply/timeline.go b/cmd/krply/timeline.go index a36fb59..460ec0a 100644 --- a/cmd/krply/timeline.go +++ b/cmd/krply/timeline.go @@ -102,10 +102,32 @@ func timelineServer(ctx context.Context, name string, since time.Time) error { if err := c.get("/v1/streams", &streams); err != nil { return err } + + // Auto-detect the namespace from the journal when it was not provided, so + // namespaced objects work without --namespace. + namespace := timelineNamespace + if namespace == "" { + q := url.Values{} + q.Set("name", name) + if timelineKind != "" { + q.Set("kind", timelineKind) + } + var page queryv1.EventPage + if err := c.get(withQuery("/v1/events", q), &page); err != nil { + return err + } + for _, it := range page.Items { + if it.Namespace != "" { + namespace = it.Namespace + break + } + } + } + var match *queryv1.Stream for i := range streams { s := &streams[i] - if timelineNamespace != "" && s.Namespace != "" && s.Namespace != timelineNamespace { + if namespace != "" && s.Namespace != "" && s.Namespace != namespace { continue } if timelineKind != "" && !strings.EqualFold(s.Kind, timelineKind) { @@ -115,13 +137,13 @@ func timelineServer(ctx context.Context, name string, since time.Time) error { break } if match == nil { - return fmt.Errorf("no stream found for object %s/%s", nsLabel(timelineNamespace), name) + return fmt.Errorf("no stream found for object %s/%s", nsLabel(namespace), name) } ref := api.EncodeObjectRef(storage.ObjectRef{ ClusterID: match.ClusterID, StreamID: match.ID, - Namespace: timelineNamespace, + Namespace: namespace, Name: name, }) path := "/v1/objects/" + url.PathEscape(ref) + "/history" @@ -191,7 +213,8 @@ func streamsForObject(ctx context.Context, store storage.Store, namespace, kind } // resolveObjectNamespace finds the namespace of an object when it was not -// provided, by searching the journal for the object's name. +// provided, by searching the journal for the object's name. Cluster-scoped +// objects (empty namespace everywhere) resolve to "". func resolveObjectNamespace(ctx context.Context, store storage.Store, name, namespace, kind string) (string, error) { if namespace != "" { return namespace, nil @@ -205,7 +228,11 @@ func resolveObjectNamespace(ctx context.Context, store storage.Store, name, name return recs[i].Resource.Namespace, nil } } - return "", fmt.Errorf("could not determine namespace for object %q (pass --namespace)", name) + if len(recs) == 0 { + return "", fmt.Errorf("no records found for object %q", name) + } + // Every matching record is cluster-scoped. + return "", nil } type tlRow struct { diff --git a/internal/api/server.go b/internal/api/server.go index 7b65a18..d92c9b8 100644 --- a/internal/api/server.go +++ b/internal/api/server.go @@ -266,22 +266,13 @@ func (s *Server) handleEvents(w http.ResponseWriter, r *http.Request) { } f.Limit = limit + 1 + f.SinceSeq = cursorSeq recs, err := s.store.Events(r.Context(), f) if err != nil { s.internalError(w, "events", err) return } - if cursorSeq > 0 { - filtered := recs[:0] - for _, rec := range recs { - if rec.IngestSeq > cursorSeq { - filtered = append(filtered, rec) - } - } - recs = filtered - } - hasMore := len(recs) > limit if hasMore { recs = recs[:limit] @@ -350,6 +341,10 @@ func (s *Server) handleDiff(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() clusterID := q.Get("cluster_id") namespace := q.Get("namespace") + if clusterID == "" { + writeError(w, http.StatusBadRequest, "cluster_id is required") + return + } before, err := parseTimeParam(q.Get("since")) if err != nil { writeError(w, http.StatusBadRequest, "invalid since: "+err.Error()) @@ -437,6 +432,7 @@ type createPlanRequest struct { SnapshotID string `json:"snapshot_id"` SourceNamespace string `json:"source_namespace"` TargetNamespace string `json:"target_namespace"` + TargetContext string `json:"target_context"` AllowGaps bool `json:"allow_gaps"` } @@ -468,6 +464,7 @@ func (s *Server) handleCreatePlan(w http.ResponseWriter, r *http.Request) { } return } + plan.TargetContext = req.TargetContext now := time.Now().UTC() s.planMu.Lock() s.plans[plan.ID] = plan @@ -487,7 +484,6 @@ func (s *Server) handleGetPlan(w http.ResponseWriter, r *http.Request) { } type dryRunRequest struct { - Kubeconfig string `json:"kubeconfig"` TargetContext string `json:"target_context"` } @@ -503,7 +499,9 @@ func (s *Server) handleDryRun(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "invalid body: "+err.Error()) return } - res, err := p.DryRun(r.Context(), req.Kubeconfig, req.TargetContext) + // The server resolves the target cluster through its own kubeconfig and + // context; it never accepts a kubeconfig from a network client. + res, err := p.DryRun(r.Context(), "", req.TargetContext) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return @@ -515,7 +513,6 @@ func (s *Server) handleDryRun(w http.ResponseWriter, r *http.Request) { type replayRunRequest struct { PlanID string `json:"plan_id"` - Kubeconfig string `json:"kubeconfig"` TargetContext string `json:"target_context"` Confirm bool `json:"confirm"` } @@ -539,7 +536,9 @@ func (s *Server) handleReplayRun(w http.ResponseWriter, r *http.Request) { writeError(w, http.StatusBadRequest, "refusing apply without confirm=true") return } - res, err := p.Apply(r.Context(), req.Kubeconfig, req.TargetContext, true) + // The target cluster is resolved through the server's own kubeconfig and + // context, never through a client-supplied kubeconfig. + res, err := p.Apply(r.Context(), "", req.TargetContext, true) if err != nil { writeError(w, http.StatusBadRequest, err.Error()) return @@ -553,6 +552,7 @@ func (s *Server) handleReplayRun(w http.ResponseWriter, r *http.Request) { FinishedAt: &now, Applied: res.Applied, Errors: mapDryRunItems(res.Errors), + Skipped: mapDryRunItems(res.Skipped), }) } @@ -744,6 +744,8 @@ func mapDiffResult(clusterID, namespace string, d *materialize.DiffResult) query } func (s *Server) mapPlan(p *replay.Plan, createdAt time.Time) queryv1.ReplayPlan { + p.Mu.RLock() + defer p.Mu.RUnlock() objects := make([]queryv1.PlanObject, 0, len(p.Objects)) for _, o := range p.Objects { objects = append(objects, queryv1.PlanObject{ @@ -764,6 +766,8 @@ func (s *Server) mapPlan(p *replay.Plan, createdAt time.Time) queryv1.ReplayPlan Reason: e.Reason, }) } + warnings := make([]string, len(p.Warnings)) + copy(warnings, p.Warnings) return queryv1.ReplayPlan{ ID: p.ID, ClusterID: p.ClusterID, @@ -774,7 +778,7 @@ func (s *Server) mapPlan(p *replay.Plan, createdAt time.Time) queryv1.ReplayPlan CreatedAt: createdAt, FieldManager: p.FieldManager, Objects: objects, - Warnings: p.Warnings, + Warnings: warnings, Excluded: excluded, CoverageComplete: p.CoverageComplete, Status: p.Status, @@ -786,6 +790,7 @@ func mapDryRunResult(d *replay.DryRunResult) *queryv1.DryRunResult { Applied: d.Applied, Conflicts: mapDryRunItems(d.Conflicts), Errors: mapDryRunItems(d.Errors), + Skipped: mapDryRunItems(d.Skipped), OK: d.OK, } } @@ -832,7 +837,13 @@ func DecodeObjectRef(token string) (storage.ObjectRef, error) { // mapEvent normalizes a journal record for API consumption. func mapEvent(r event.Record) queryv1.Event { var object any - if len(r.Object) > 0 { + if r.Type == event.TypeGap && r.Gap != nil { + object = map[string]any{ + "from_resource_version": r.Gap.FromResourceVersion, + "to_resource_version": r.Gap.ToResourceVersion, + "reason": r.Gap.Reason, + } + } else if len(r.Object) > 0 { var v any if err := json.Unmarshal(r.Object, &v); err == nil { object = v diff --git a/internal/discovery/discover.go b/internal/discovery/discover.go index 04bedba..1f18cf7 100644 --- a/internal/discovery/discover.go +++ b/internal/discovery/discover.go @@ -76,19 +76,21 @@ func resolveOne(ctx context.Context, client kubernetes.Interface, spec ResourceS return resolveViaDiscovery(ctx, client, spec) } -// mergeAlias fills empty fields on spec from the alias, keeping any explicit -// Namespace and replacing the alias resource name with its canonical form. +// mergeAlias fills empty fields on spec from the alias. Explicit fields on the +// user's spec are never overwritten so a fully- or partially-qualified spec +// whose plural name happens to match an alias (for example a CRD named "pods") +// is not silently rewritten to the built-in resource. func mergeAlias(spec, alias ResourceSpec) ResourceSpec { - if alias.APIGroup != "" { + if spec.APIGroup == "" { spec.APIGroup = alias.APIGroup } - if alias.Version != "" { + if spec.Version == "" { spec.Version = alias.Version } - if alias.Kind != "" { + if spec.Kind == "" { spec.Kind = alias.Kind } - if alias.Resource != "" { + if alias.Resource != "" && !strings.EqualFold(spec.Resource, alias.Resource) { spec.Resource = alias.Resource } return spec @@ -150,28 +152,33 @@ func resolveInGroupVersion(ctx context.Context, client kubernetes.Interface, spe // aliases maps common kubectl-style names to canonical specs. var aliases = map[string]ResourceSpec{ - "deploy": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, - "deployment": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, - "deployments": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, - "sts": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, - "statefulset": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, - "statefulsets": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, - "ds": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, - "daemonset": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, - "daemonsets": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, - "cm": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, - "configmap": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, - "configmaps": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, - "po": {Version: "v1", Resource: "pods", Kind: "Pod"}, - "pod": {Version: "v1", Resource: "pods", Kind: "Pod"}, - "pods": {Version: "v1", Resource: "pods", Kind: "Pod"}, - "svc": {Version: "v1", Resource: "services", Kind: "Service"}, - "service": {Version: "v1", Resource: "services", Kind: "Service"}, - "services": {Version: "v1", Resource: "services", Kind: "Service"}, - "ns": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, - "namespace": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, - "namespaces": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, - "rc": {APIGroup: "node.k8s.io", Version: "v1", Resource: "runtimeclasses", Kind: "RuntimeClass"}, - "runtimeclass": {APIGroup: "node.k8s.io", Version: "v1", Resource: "runtimeclasses", Kind: "RuntimeClass"}, - "runtimeclasses": {APIGroup: "node.k8s.io", Version: "v1", Resource: "runtimeclasses", Kind: "RuntimeClass"}, + "deploy": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, + "deployment": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, + "deployments": {APIGroup: "apps", Version: "v1", Resource: "deployments", Kind: "Deployment"}, + "sts": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, + "statefulset": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, + "statefulsets": {APIGroup: "apps", Version: "v1", Resource: "statefulsets", Kind: "StatefulSet"}, + "ds": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, + "daemonset": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, + "daemonsets": {APIGroup: "apps", Version: "v1", Resource: "daemonsets", Kind: "DaemonSet"}, + "cm": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, + "configmap": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, + "configmaps": {Version: "v1", Resource: "configmaps", Kind: "ConfigMap"}, + "po": {Version: "v1", Resource: "pods", Kind: "Pod"}, + "pod": {Version: "v1", Resource: "pods", Kind: "Pod"}, + "pods": {Version: "v1", Resource: "pods", Kind: "Pod"}, + "svc": {Version: "v1", Resource: "services", Kind: "Service"}, + "service": {Version: "v1", Resource: "services", Kind: "Service"}, + "services": {Version: "v1", Resource: "services", Kind: "Service"}, + "ns": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, + "namespace": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, + "namespaces": {Version: "v1", Resource: "namespaces", Kind: "Namespace"}, + "rc": {Version: "v1", Resource: "replicationcontrollers", Kind: "ReplicationController"}, + "replicationcontroller": {Version: "v1", Resource: "replicationcontrollers", Kind: "ReplicationController"}, + "replicationcontrollers": {Version: "v1", Resource: "replicationcontrollers", Kind: "ReplicationController"}, + "rs": {APIGroup: "apps", Version: "v1", Resource: "replicasets", Kind: "ReplicaSet"}, + "replicaset": {APIGroup: "apps", Version: "v1", Resource: "replicasets", Kind: "ReplicaSet"}, + "replicasets": {APIGroup: "apps", Version: "v1", Resource: "replicasets", Kind: "ReplicaSet"}, + "runtimeclass": {APIGroup: "node.k8s.io", Version: "v1", Resource: "runtimeclasses", Kind: "RuntimeClass"}, + "runtimeclasses": {APIGroup: "node.k8s.io", Version: "v1", Resource: "runtimeclasses", Kind: "RuntimeClass"}, } diff --git a/internal/discovery/discover_test.go b/internal/discovery/discover_test.go index d6ac498..ab61ea5 100644 --- a/internal/discovery/discover_test.go +++ b/internal/discovery/discover_test.go @@ -89,6 +89,7 @@ func TestResolveAliases(t *testing.T) { {Resource: "svc"}, {Resource: "ns"}, {Resource: "rc"}, + {Resource: "rs"}, } resolved, err := Resolve(context.Background(), cli, inputs) if err != nil { @@ -102,7 +103,8 @@ func TestResolveAliases(t *testing.T) { {"pods", "", "Pod"}, {"services", "", "Service"}, {"namespaces", "", "Namespace"}, - {"runtimeclasses", "node.k8s.io", "RuntimeClass"}, + {"replicationcontrollers", "", "ReplicationController"}, + {"replicasets", "apps", "ReplicaSet"}, } for i, w := range want { got := resolved[i] @@ -131,3 +133,17 @@ func TestResolvePartialWithKnownVersion(t *testing.T) { t.Errorf("deployments should be namespaced with empty namespace filter, got %q", s.Namespace) } } + +func TestResolveKeepsExplicitGroup(t *testing.T) { + cli := newFakeClient() + resolved, err := Resolve(context.Background(), cli, []ResourceSpec{ + {APIGroup: "acme.io", Version: "v1", Resource: "pods", Kind: "AcmePod"}, + }) + if err != nil { + t.Fatalf("Resolve: %v", err) + } + s := resolved[0] + if s.APIGroup != "acme.io" || s.Kind != "AcmePod" || s.Resource != "pods" { + t.Errorf("alias clobbered explicit spec: %+v", s) + } +} diff --git a/internal/event/stream.go b/internal/event/stream.go index 4fcd221..72bfe98 100644 --- a/internal/event/stream.go +++ b/internal/event/stream.go @@ -4,6 +4,7 @@ import ( "crypto/sha256" "encoding/hex" "fmt" + "net/url" "strings" ) @@ -20,6 +21,8 @@ type Stream struct { // ID returns the stable stream identifier used across the journal. // Format: cluster/group/version/resource/namespace/selector +// The selector is URL-escaped because label selectors legitimately contain '/' +// (prefix/name label keys) which would otherwise make the ID ambiguous. func (s Stream) ID() string { var b strings.Builder b.WriteString(s.ClusterID) @@ -32,7 +35,7 @@ func (s Stream) ID() string { b.WriteByte('/') b.WriteString(s.Namespace) b.WriteByte('/') - b.WriteString(s.Selector) + b.WriteString(url.PathEscape(s.Selector)) return b.String() } @@ -42,13 +45,17 @@ func StreamID(id string) (Stream, error) { if len(parts) != 6 { return Stream{}, fmt.Errorf("stream id %q: want 6 slash-separated parts", id) } + selector, err := url.PathUnescape(parts[5]) + if err != nil { + return Stream{}, fmt.Errorf("stream id %q: invalid selector: %w", id, err) + } return Stream{ ClusterID: parts[0], Group: parts[1], Version: parts[2], Resource: parts[3], Namespace: parts[4], - Selector: parts[5], + Selector: selector, }, nil } @@ -59,13 +66,17 @@ func (s Stream) GVR() (string, string, string) { // EventID derives a deterministic deduplication key for a watch event. // The same underlying API event must always produce the same key, so a -// duplicate delivery after a reconnect is idempotent. +// duplicate delivery after a reconnect is idempotent. For live events +// observedAt is zero and excluded. Collector-generated synthetic baselines +// pass a non-zero observedAt so each relist produces a distinct key: an +// unchanged object re-listed after a gap is a new observation, not a +// duplicate delivery, and must survive deduplication. func EventID(stream Stream, resource ResourceRef, watchType WatchType, observedAt int64) string { h := sha256.New() fmt.Fprintf(h, "%s\x00%s\x00%s\x00%s\x00%s\x00%s", stream.ID(), resource.Namespace, resource.Name, resource.UID, resource.ResourceVersion, string(watchType)) - // observedAt is deliberately excluded: resource version ordering is the - // dedup domain, not wall-clock observation time. - _ = observedAt + if observedAt != 0 { + fmt.Fprintf(h, "\x00%d", observedAt) + } return hex.EncodeToString(h.Sum(nil)) } diff --git a/internal/materialize/diff.go b/internal/materialize/diff.go index 8cd48c7..0d0d39e 100644 --- a/internal/materialize/diff.go +++ b/internal/materialize/diff.go @@ -36,8 +36,9 @@ type FieldChange struct { Removed bool } -// ignoredKeys are server-owned or hash-only noise fields excluded from diffs. -var ignoredKeys = map[string]bool{ +// metadataIgnoreKeys are server-owned metadata fields, ignored only when they +// appear directly under metadata. +var metadataIgnoreKeys = map[string]bool{ "uid": true, "resourceVersion": true, "creationTimestamp": true, @@ -45,13 +46,30 @@ var ignoredKeys = map[string]bool{ "managedFields": true, "deletionTimestamp": true, "ownerReferences": true, - "status": true, - "kubectl.kubernetes.io/last-applied-configuration": true, +} + +// ignoreKey reports whether a map key at parentPath is server-owned noise. +// Ignoring is path-scoped so a user field named "status", "uid" or +// "resourceVersion" inside data or spec is never silently dropped. +func ignoreKey(parentPath, key string) bool { + if key == "status" && parentPath == "" { + return true + } + if parentPath == "metadata" && metadataIgnoreKeys[key] { + return true + } + if key == "kubectl.kubernetes.io/last-applied-configuration" && parentPath == "metadata.annotations" { + return true + } + return false } // Diff reconstructs cluster state before and after, intersects object keys, // and reports semantic field changes, ignoring server-owned metadata. func (m *Materializer) Diff(ctx context.Context, clusterID, namespace string, before, after time.Time) (*DiffResult, error) { + if !before.IsZero() && !after.IsZero() && before.After(after) { + return nil, fmt.Errorf("materialize: before (%s) must not be after after (%s)", before.Format(time.RFC3339), after.Format(time.RFC3339)) + } streams, err := m.store.Streams(ctx) if err != nil { return nil, err @@ -66,23 +84,28 @@ func (m *Materializer) Diff(ctx context.Context, clusterID, namespace string, be if s.ClusterID != clusterID { continue } - b, err := m.StreamState(ctx, s.StreamID, before) - if err != nil { - return nil, err + if !before.IsZero() { + b, err := m.StreamState(ctx, s.StreamID, before) + if err != nil { + return nil, err + } + if b.HasGaps { + gapped = append(gapped, s.StreamID) + } + for _, st := range b.Objects { + if namespace != "" && st.Namespace != namespace { + continue + } + beforeStates[st.StreamID+"|"+st.Namespace+"/"+st.Name] = st + } } a, err := m.StreamState(ctx, s.StreamID, after) if err != nil { return nil, err } - if b.HasGaps || a.HasGaps { + if a.HasGaps { gapped = append(gapped, s.StreamID) } - for _, st := range b.Objects { - if namespace != "" && st.Namespace != namespace { - continue - } - beforeStates[st.StreamID+"|"+st.Namespace+"/"+st.Name] = st - } for _, st := range a.Objects { if namespace != "" && st.Namespace != namespace { continue @@ -183,7 +206,8 @@ func parseObject(raw json.RawMessage) any { // diffValue recurses into before/after and appends field changes at out. // Arrays are compared by index; a length change is reported as one change at -// the array's path. Map objects recurse by key. +// the array's path. Map objects recurse by key; keys present on only one side +// are reported as a single Added/Removed change. func diffValue(path string, before, after any, out *[]FieldChange) { if reflect.DeepEqual(before, after) { return @@ -204,10 +228,20 @@ func diffValue(path string, before, after any, out *[]FieldChange) { } sort.Strings(sorted) for _, k := range sorted { - if ignoredKeys[k] { + if ignoreKey(path, k) { continue } - diffValue(joinPath(path, k), bm[k], am[k], out) + childPath := joinPath(path, k) + _, beforeHas := bm[k] + _, afterHas := am[k] + switch { + case beforeHas && !afterHas: + *out = append(*out, FieldChange{Path: childPath, Before: bm[k], Removed: true}) + case !beforeHas && afterHas: + *out = append(*out, FieldChange{Path: childPath, After: am[k], Added: true}) + default: + diffValue(childPath, bm[k], am[k], out) + } } return } @@ -226,9 +260,20 @@ func diffValue(path string, before, after any, out *[]FieldChange) { *out = append(*out, FieldChange{Path: path, Before: before, After: after}) } +// joinPath builds a dotted path. Segments that could collide with the path +// syntax ('.', '[', ']', '\') are backslash-escaped. func joinPath(parent, key string) string { + key = escapePathSegment(key) if parent == "" { return key } return parent + "." + key } + +func escapePathSegment(k string) string { + k = strings.ReplaceAll(k, `\`, `\\`) + k = strings.ReplaceAll(k, `.`, `\.`) + k = strings.ReplaceAll(k, `[`, `\[`) + k = strings.ReplaceAll(k, `]`, `\]`) + return k +} diff --git a/internal/materialize/materialize.go b/internal/materialize/materialize.go index 5e0d4c9..74aa805 100644 --- a/internal/materialize/materialize.go +++ b/internal/materialize/materialize.go @@ -84,10 +84,12 @@ func (m *Materializer) StateAt(ctx context.Context, clusterID string, at time.Ti // StreamStateResult is the reduced state of one stream. type StreamStateResult struct { - Objects []ObjectState - HasGaps bool - GapCount int - HasBaseline bool + Objects []ObjectState + HasGaps bool + GapCount int + HasBaseline bool + LastObservedAt time.Time + LastResourceVersion string } // StreamState reduces every event of a stream observed up to at into object @@ -103,26 +105,45 @@ func (m *Materializer) StreamState(ctx context.Context, streamID string, at time objs = append(objs, st) } sortObjects(objs) - return StreamStateResult{Objects: objs, HasGaps: red.hasGaps, GapCount: red.gapCount, HasBaseline: red.hasBaseline}, nil + return StreamStateResult{ + Objects: objs, + HasGaps: red.hasGaps, + GapCount: red.gapCount, + HasBaseline: red.hasBaseline, + LastObservedAt: red.lastObservedAt, + LastResourceVersion: red.lastRV, + }, nil } // reduction is the per-stream reduce result. type reduction struct { - states map[string]ObjectState - hasBaseline bool - hasGaps bool - gapCount int + states map[string]ObjectState + hasBaseline bool + hasGaps bool + gapCount int + lastObservedAt time.Time + lastRV string } // reduce folds records into per-object state. DELETED removes the object, -// BASELINE marks full coverage from that point, and GAP marks lost coverage. +// GAP marks lost coverage, and BASELINE resets the view: a relist after a gap +// re-establishes every object that still exists as a synthetic ADDED, so +// objects deleted during the gap are dropped and any prior open gap is healed. func reduce(streamID string, recs []event.Record) reduction { red := reduction{states: map[string]ObjectState{}} for i := range recs { rec := &recs[i] + if !rec.ObservedAt.IsZero() { + red.lastObservedAt = rec.ObservedAt + } switch rec.Type { case event.TypeBaseline: red.hasBaseline = true + red.hasGaps = false + clear(red.states) + if rec.Resource.ResourceVersion != "" { + red.lastRV = rec.Resource.ResourceVersion + } case event.TypeGap: red.hasGaps = true red.gapCount++ @@ -145,6 +166,13 @@ func reduce(streamID string, recs []event.Record) reduction { Object: rec.Object, At: rec.ObservedAt, } + if rec.Resource.ResourceVersion != "" { + red.lastRV = rec.Resource.ResourceVersion + } + } + case event.TypeCheckpoint: + if rec.Checkpoint != nil && rec.Checkpoint.ResourceVersion != "" { + red.lastRV = rec.Checkpoint.ResourceVersion } } } diff --git a/internal/materialize/snapshot.go b/internal/materialize/snapshot.go index 3e470cc..6b0ae6c 100644 --- a/internal/materialize/snapshot.go +++ b/internal/materialize/snapshot.go @@ -7,25 +7,34 @@ import ( "time" "github.com/google/uuid" + "github.com/krply/krply/internal/event" "github.com/krply/krply/internal/storage" ) // Snapshot is a materialized view of a cluster at a point in time. type Snapshot struct { - ID string - ClusterID string - Name string - At time.Time - Objects []ObjectState - Streams []string - Complete bool - Missing []string - Warning string + ID string + ClusterID string + Name string + At time.Time + Objects []ObjectState + Streams []string + Watermarks []StreamWatermark + Complete bool + Missing []string + Warning string +} + +// StreamWatermark is the durable boundary of one stream at snapshot time. +type StreamWatermark struct { + StreamID string + LastObservedAt time.Time + LastResourceVersion string } // Snapshot materializes every watched object of the cluster at at, records -// per-stream coverage, and persists the snapshot metadata. The materialized -// objects are reconstructable on demand; the snapshot reference is durable. +// per-stream coverage and watermarks, persists a TypeSnapshot journal entry, +// and stores the snapshot metadata. func (m *Materializer) Snapshot(ctx context.Context, clusterID string, at time.Time, name string) (*Snapshot, error) { streams, err := m.store.Streams(ctx) if err != nil { @@ -51,6 +60,11 @@ func (m *Materializer) Snapshot(ctx context.Context, clusterID string, at time.T } snap.Objects = append(snap.Objects, ss.Objects...) snap.Streams = append(snap.Streams, s.StreamID) + snap.Watermarks = append(snap.Watermarks, StreamWatermark{ + StreamID: s.StreamID, + LastObservedAt: ss.LastObservedAt, + LastResourceVersion: ss.LastResourceVersion, + }) if !ss.HasBaseline || ss.HasGaps { missing = append(missing, s.StreamID) @@ -59,6 +73,7 @@ func (m *Materializer) Snapshot(ctx context.Context, clusterID string, at time.T sort.Strings(snap.Streams) sortObjects(snap.Objects) + sort.Slice(snap.Watermarks, func(i, j int) bool { return snap.Watermarks[i].StreamID < snap.Watermarks[j].StreamID }) sort.Strings(missing) if len(missing) > 0 { @@ -76,5 +91,16 @@ func (m *Materializer) Snapshot(ctx context.Context, clusterID string, at time.T return nil, err } + // Persist a TypeSnapshot journal record so the snapshot is reconstructable + // and provable from the event stream, not only from the metadata table. + if _, err := m.store.Append(ctx, &event.Record{ + ClusterID: clusterID, + ObservedAt: at, + Type: event.TypeSnapshot, + Snapshot: &event.SnapshotInfo{Name: name}, + }); err != nil { + return nil, err + } + return snap, nil } diff --git a/internal/replay/apply.go b/internal/replay/apply.go index df5803a..85e043c 100644 --- a/internal/replay/apply.go +++ b/internal/replay/apply.go @@ -20,6 +20,7 @@ type DryRunResult struct { Applied int Conflicts []DryRunItem Errors []DryRunItem + Skipped []DryRunItem OK bool } @@ -37,18 +38,23 @@ type ApplyResult struct { PlanID string Applied int Errors []DryRunItem + Skipped []DryRunItem } // DryRun runs a server-side apply dry run against the target cluster. It never // sends Force, so ownership conflicts are reported instead of overwritten. OK -// is true only when there are no conflicts and no errors. +// is true only when there are no conflicts, no errors, and nothing skipped. func (pl *Plan) DryRun(ctx context.Context, kubeconfig, targetContext string) (*DryRunResult, error) { + pl.Mu.Lock() + defer pl.Mu.Unlock() dyn, err := pl.dynamicClient(kubeconfig, targetContext) if err != nil { return nil, err } res := &DryRunResult{} - for _, it := range pl.applyItems() { + items, skipped := pl.applyItems() + res.Skipped = skipped + for _, it := range items { if err := applyOne(ctx, dyn, it, pl.FieldManager, true); err != nil { if apierrors.IsConflict(err) { res.Conflicts = append(res.Conflicts, DryRunItem{ @@ -58,6 +64,9 @@ func (pl *Plan) DryRun(ctx context.Context, kubeconfig, targetContext string) (* Manager: conflictManager(err), Message: err.Error(), }) + } else if apierrors.IsNotFound(err) && it.namespace != "" { + // The target namespace does not exist yet; Apply creates it + // first, so this is not a real failure for the dry run. } else { res.Errors = append(res.Errors, DryRunItem{ Namespace: it.namespace, @@ -70,7 +79,7 @@ func (pl *Plan) DryRun(ctx context.Context, kubeconfig, targetContext string) (* } res.Applied++ } - res.OK = len(res.Conflicts) == 0 && len(res.Errors) == 0 + res.OK = len(res.Conflicts) == 0 && len(res.Errors) == 0 && len(res.Skipped) == 0 if res.OK { pl.Status = "dry-run-ok" } else { @@ -80,20 +89,28 @@ func (pl *Plan) DryRun(ctx context.Context, kubeconfig, targetContext string) (* } // Apply applies the plan to the target cluster with server-side apply using -// the synthetic field manager. It refuses to run without explicit -// confirmation. Failures are collected per object and never stop the loop. +// the synthetic field manager. It refuses to run without explicit confirmation +// and without a previously successful dry run (the plan status must be +// "dry-run-ok"). Failures are collected per object and never stop the loop. func (pl *Plan) Apply(ctx context.Context, kubeconfig, targetContext string, confirm bool) (*ApplyResult, error) { if !confirm { return nil, errors.New("refusing apply without --confirm") } + pl.Mu.Lock() + defer pl.Mu.Unlock() + if pl.Status != "dry-run-ok" { + return nil, fmt.Errorf("refusing apply: plan %s has not passed a dry run (status %q)", pl.ID, pl.Status) + } dyn, err := pl.dynamicClient(kubeconfig, targetContext) if err != nil { return nil, err } result := &ApplyResult{PlanID: pl.ID} + items, skipped := pl.applyItems() + result.Skipped = skipped ensured := map[string]bool{} - for _, it := range pl.applyItems() { + for _, it := range items { if it.namespace != "" && it.gvr.Resource != "namespaces" && !ensured[it.namespace] { ensured[it.namespace] = true if err := ensureNamespace(ctx, dyn, it.namespace); err != nil { @@ -173,13 +190,22 @@ type applyItem struct { } // applyItems converts plan objects into apply items, skipping objects whose -// apiVersion or kind has no supported GVR mapping. -func (pl *Plan) applyItems() []applyItem { +// apiVersion or kind has no supported GVR mapping. Skipped objects are +// returned explicitly so callers can report them instead of silently dropping +// objects while claiming success. +func (pl *Plan) applyItems() ([]applyItem, []DryRunItem) { items := make([]applyItem, 0, len(pl.Objects)) + var skipped []DryRunItem for _, po := range pl.Objects { gvr, err := gvrFor(po.Object, po.Kind) if err != nil { pl.Warnings = append(pl.Warnings, fmt.Sprintf("replay: skipping %s/%s: %v", po.Kind, po.Name, err)) + skipped = append(skipped, DryRunItem{ + Namespace: po.Namespace, + Name: po.Name, + Kind: po.Kind, + Message: err.Error(), + }) continue } items = append(items, applyItem{ @@ -190,7 +216,45 @@ func (pl *Plan) applyItems() []applyItem { object: po.Object, }) } - return items + return items, skipped +} + +// kindToResource maps object kinds to their plural resource names for the +// dynamic client. Group and version come from the object's apiVersion. +var kindToResource = map[string]string{ + "Deployment": "deployments", + "StatefulSet": "statefulsets", + "DaemonSet": "daemonsets", + "ReplicaSet": "replicasets", + "Service": "services", + "ConfigMap": "configmaps", + "Namespace": "namespaces", + "RuntimeClass": "runtimeclasses", + "Secret": "secrets", + "ServiceAccount": "serviceaccounts", + "Role": "roles", + "ClusterRole": "clusterroles", + "RoleBinding": "rolebindings", + "ClusterRoleBinding": "clusterrolebindings", + "Job": "jobs", + "CronJob": "cronjobs", + "Pod": "pods", + "PersistentVolume": "persistentvolumes", + "PersistentVolumeClaim": "persistentvolumeclaims", + "StorageClass": "storageclasses", + "VolumeAttachment": "volumeattachments", + "MutatingWebhookConfiguration": "mutatingwebhookconfigurations", + "ValidatingWebhookConfiguration": "validatingwebhookconfigurations", + "CustomResourceDefinition": "customresourcedefinitions", + "Ingress": "ingresses", + "NetworkPolicy": "networkpolicies", + "LimitRange": "limitranges", + "ResourceQuota": "resourcequotas", + "HorizontalPodAutoscaler": "horizontalpodautoscalers", + "PodDisruptionBudget": "poddisruptionbudgets", + "Endpoints": "endpoints", + "PriorityClass": "priorityclasses", + "Lease": "leases", } func gvrFor(obj map[string]any, kind string) (schema.GroupVersionResource, error) { @@ -202,23 +266,8 @@ func gvrFor(obj map[string]any, kind string) (schema.GroupVersionResource, error if err != nil { return schema.GroupVersionResource{}, err } - var resource string - switch kind { - case "Deployment": - resource = "deployments" - case "StatefulSet": - resource = "statefulsets" - case "DaemonSet": - resource = "daemonsets" - case "Service": - resource = "services" - case "ConfigMap": - resource = "configmaps" - case "Namespace": - resource = "namespaces" - case "RuntimeClass": - resource = "runtimeclasses" - default: + resource, ok := kindToResource[kind] + if !ok { return schema.GroupVersionResource{}, fmt.Errorf("replay: unsupported kind %q", kind) } return schema.GroupVersionResource{Group: gv.Group, Version: gv.Version, Resource: resource}, nil @@ -262,7 +311,7 @@ func ensureNamespace(ctx context.Context, dyn dynamic.Interface, ns string) erro // conflict status, falling back to an empty string when it is not available. func conflictManager(err error) string { var statusErr *apierrors.StatusError - if errors.As(err, &statusErr) { + if errors.As(err, &statusErr) && statusErr.ErrStatus.Details != nil { for _, c := range statusErr.ErrStatus.Details.Causes { if c.Field == "manager" || strings.Contains(c.Message, "conflict") { return c.Message diff --git a/internal/replay/plan.go b/internal/replay/plan.go index 21181c5..6c9d5d9 100644 --- a/internal/replay/plan.go +++ b/internal/replay/plan.go @@ -6,9 +6,11 @@ import ( "errors" "fmt" "sort" + "sync" "github.com/google/uuid" "github.com/krply/krply/internal/materialize" + "github.com/krply/krply/internal/metrics" "github.com/krply/krply/internal/storage" ) @@ -32,6 +34,9 @@ type Excluded struct { // Plan is a sanitized, reviewable replay of a snapshot into a target cluster. type Plan struct { + // Mu guards Status, Warnings, and Objects so concurrent dry-runs/applies + // and reads from the API do not race. + Mu sync.RWMutex ID string ClusterID string SnapshotID string @@ -48,9 +53,10 @@ type Plan struct { // Planner builds replay plans from a store and a materializer. type Planner struct { - store storage.Store - mat *materialize.Materializer - policy Policy + store storage.Store + mat *materialize.Materializer + policy Policy + metrics *metrics.Metrics } // NewPlanner returns a Planner backed by store and mat using policy. @@ -58,37 +64,52 @@ func NewPlanner(store storage.Store, mat *materialize.Materializer, policy Polic return &Planner{store: store, mat: mat, policy: policy} } +// SetMetrics attaches a Metrics instance so plan failures are counted. +func (p *Planner) SetMetrics(m *metrics.Metrics) { + p.metrics = m +} + +func (p *Planner) countFailure() { + if p.metrics != nil { + p.metrics.ReplayPlanFailures.Inc() + } +} + // Plan reconstructs the objects of the given snapshot, sanitizes them, and // returns an ordered plan. It refuses to plan when stream coverage was // incomplete unless the policy allows gaps. func (p *Planner) Plan(ctx context.Context, clusterID, snapshotID, sourceNS, targetNS string) (*Plan, error) { snap, err := p.lookupSnapshot(ctx, snapshotID) if err != nil { + p.countFailure() return nil, err } states, complete, err := p.mat.StateAt(ctx, snap.ClusterID, snap.At) if err != nil { + p.countFailure() return nil, err } if !complete && !p.policy.AllowGaps { + p.countFailure() return nil, errors.New("refusing plan: incomplete coverage (stream gaps) — pass allow-gaps to override") } - id := "plan-" + uuid.NewString()[:8] + id := "plan-" + uuid.NewString()[:12] plan := &Plan{ ID: id, ClusterID: snap.ClusterID, SnapshotID: snapshotID, SourceNamespace: sourceNS, TargetNamespace: targetNS, - FieldManager: "krply-plan-" + id, + FieldManager: "krply-" + id, Status: "planned", CoverageComplete: complete, } mapped := false + seenTarget := map[string]bool{} for _, st := range states { var obj map[string]any if err := json.Unmarshal(st.Object, &obj); err != nil { @@ -116,6 +137,19 @@ func (p *Planner) Plan(ctx context.Context, clusterID, snapshotID, sourceNS, tar clean, warnings := sanitizeObject(obj, kind, p.policy) effectiveNS := mapNamespace(clean, namespace, sourceNS, targetNS, p.policy, &mapped, &plan.Warnings) + // Namespace remapping can collapse distinct source namespaces into one + // target, which would silently overwrite same-named objects. Detect the + // collision and exclude the later object instead. + if effectiveNS != "" && sourceNS == "" && targetNS != "" { + key := effectiveNS + "/" + kind + "/" + name + if seenTarget[key] { + plan.Excluded = append(plan.Excluded, Excluded{Namespace: effectiveNS, Name: name, Kind: kind, Reason: "name collision after namespace mapping"}) + plan.Warnings = append(plan.Warnings, fmt.Sprintf("excluded %s %s/%s: name collision after mapping into %s", kind, effectiveNS, name, targetNS)) + continue + } + seenTarget[key] = true + } + plan.Objects = append(plan.Objects, PlanObject{ Namespace: effectiveNS, Name: name, @@ -154,8 +188,11 @@ func (p *Planner) lookupSnapshot(ctx context.Context, snapshotID string) (*stora } // mapNamespace applies namespace remapping and returns the effective namespace -// the object will be applied to. When no remap is requested, the namespace is -// dropped from the payload (the caller sets it at apply time). +// the object will be applied to. An explicit targetNS is always honored (even +// when MapNamespaces is false): objects are filtered to sourceNS first, so +// mapping sourceNS -> targetNS never touches other namespaces. When no remap is +// requested, the namespace is dropped from the payload (the caller sets it at +// apply time). func mapNamespace(obj map[string]any, namespace, sourceNS, targetNS string, pol Policy, mapped *bool, warnings *[]string) string { if namespace == "" { return "" @@ -163,18 +200,15 @@ func mapNamespace(obj map[string]any, namespace, sourceNS, targetNS string, pol m := meta(obj) switch { - case pol.MapNamespaces && sourceNS != "" && targetNS != "": + case targetNS != "": m["namespace"] = targetNS remapSpecNamespace(obj, targetNS) if !*mapped { - *warnings = append(*warnings, fmt.Sprintf("namespace mapping: %s -> %s", sourceNS, targetNS)) - *mapped = true - } - return targetNS - case sourceNS == "" && targetNS != "": - m["namespace"] = targetNS - if !*mapped { - *warnings = append(*warnings, fmt.Sprintf("namespace mapping: * -> %s", targetNS)) + if sourceNS == "" { + *warnings = append(*warnings, fmt.Sprintf("namespace mapping: * -> %s", targetNS)) + } else { + *warnings = append(*warnings, fmt.Sprintf("namespace mapping: %s -> %s", sourceNS, targetNS)) + } *mapped = true } return targetNS diff --git a/internal/replay/replay_test.go b/internal/replay/replay_test.go index 89b30d2..481191d 100644 --- a/internal/replay/replay_test.go +++ b/internal/replay/replay_test.go @@ -173,8 +173,8 @@ func TestPlanSanitizesAndExcludes(t *testing.T) { if !plan.CoverageComplete { t.Fatalf("CoverageComplete = false, want true") } - if plan.ID == "" || plan.FieldManager != "krply-plan-"+plan.ID { - t.Fatalf("ID = %q, FieldManager = %q, want krply-plan-", plan.ID, plan.FieldManager) + if plan.ID == "" || plan.FieldManager != "krply-"+plan.ID { + t.Fatalf("ID = %q, FieldManager = %q, want krply-", plan.ID, plan.FieldManager) } if len(plan.Objects) != 1 { t.Fatalf("objects = %d, want 1: %+v", len(plan.Objects), plan.Objects) diff --git a/internal/replay/sanitize.go b/internal/replay/sanitize.go index c5a302f..b5b8d07 100644 --- a/internal/replay/sanitize.go +++ b/internal/replay/sanitize.go @@ -101,8 +101,10 @@ func sanitizeFinalizers(obj map[string]any, pol Policy) []string { return warnings } -// sanitizeService strips dynamic cluster IP fields from ClusterIP services. -// LoadBalancer services are rejected by the caller via excludeReason. +// sanitizeService strips dynamic cluster IP fields from ClusterIP and NodePort +// services. Headless services (clusterIP: "None") keep their explicit value, +// which is user-desired state. LoadBalancer services are rejected by the +// caller via excludeReason. func sanitizeService(obj map[string]any) []string { spec, _ := obj["spec"].(map[string]any) if spec == nil { @@ -110,9 +112,25 @@ func sanitizeService(obj map[string]any) []string { } switch serviceType(obj) { case "", "ClusterIP": + if spec["clusterIP"] == "None" { + return nil + } delete(spec, "clusterIP") delete(spec, "clusterIPs") return []string{"service cluster IP removed"} + case "NodePort": + // NodePort services also receive a server-assigned cluster IP, and a + // fixed nodePort from the source cluster would collide in the target. + delete(spec, "clusterIP") + delete(spec, "clusterIPs") + if ports, ok := spec["ports"].([]any); ok { + for _, p := range ports { + if pm, ok := p.(map[string]any); ok { + delete(pm, "nodePort") + } + } + } + return []string{"service cluster IP and nodePort removed for reassignment"} } return nil } diff --git a/internal/storage/memory.go b/internal/storage/memory.go index bd0c995..1f39452 100644 --- a/internal/storage/memory.go +++ b/internal/storage/memory.go @@ -11,13 +11,15 @@ import ( // NewInMemory returns a Store backed by RAM. It is for tests and for running // the query API without a persistent journal. It satisfies the same atomic -// append+checkpoint contract as the SQLite store. +// append+checkpoint contract as the SQLite store, including deduplication by +// event_id for non-empty keys. func NewInMemory() Store { m := &memoryStore{ records: map[string][]*event.Record{}, seq: 0, streams: map[string]*StreamMeta{}, snapshots: map[string]*SnapshotRef{}, + seqByKey: map[string]int64{}, } return m } @@ -28,6 +30,7 @@ type memoryStore struct { seq int64 streams map[string]*StreamMeta snapshots map[string]*SnapshotRef + seqByKey map[string]int64 // streamID\x00eventID -> ingest seq } func (m *memoryStore) Append(ctx context.Context, rec *event.Record) (int64, error) { @@ -51,13 +54,24 @@ func (m *memoryStore) Appends(ctx context.Context, recs []*event.Record) ([]int6 } func (m *memoryStore) appendLocked(rec *event.Record) (int64, error) { + if rec.EventID != "" { + if seq, ok := m.seqByKey[rec.StreamID+"\x00"+rec.EventID]; ok { + rec.IngestSeq = seq + return seq, nil + } + } m.seq++ rec.IngestSeq = m.seq if rec.ObservedAt.IsZero() { rec.ObservedAt = time.Now().UTC() } m.records[rec.StreamID] = append(m.records[rec.StreamID], rec) - m.updateMeta(rec) + if rec.EventID != "" { + m.seqByKey[rec.StreamID+"\x00"+rec.EventID] = m.seq + } + if rec.StreamID != "" { + m.updateMeta(rec) + } return m.seq, nil } @@ -87,7 +101,9 @@ func (m *memoryStore) updateMeta(rec *event.Record) { case event.TypeCoverageChange: if rec.Coverage != nil { meta.Available = rec.Coverage.Current.Available - if !meta.Available { + if meta.Available { + meta.Degraded = false + } else { meta.Degraded = true } } @@ -149,10 +165,22 @@ func (m *memoryStore) Events(ctx context.Context, f EventFilter) ([]event.Record if !matches(r, f) { continue } + if f.SinceSeq > 0 && r.IngestSeq <= f.SinceSeq { + continue + } out = append(out, *r) } } sort.Slice(out, func(i, j int) bool { return out[i].IngestSeq < out[j].IngestSeq }) + if f.Offset > 0 { + if int64(f.Offset) >= int64(len(out)) { + return nil, nil + } + out = out[f.Offset:] + } + if f.Limit > 0 && len(out) > f.Limit { + out = out[:f.Limit] + } return out, nil } @@ -186,7 +214,7 @@ func (m *memoryStore) ObjectAt(ctx context.Context, ref ObjectRef, ts time.Time) var last *event.Record for i := range recs { if recs[i].ObservedAt.After(ts) { - break + continue } if recs[i].WatchType == event.WatchDeleted { last = nil @@ -242,7 +270,7 @@ func (m *memoryStore) SaveSnapshot(ctx context.Context, snap *SnapshotRef) error defer m.mu.Unlock() cp := *snap if cp.ID == "" { - cp.ID = "snap-" + time.Now().UTC().Format("20060102T150405") + cp.ID = "snap-" + time.Now().UTC().Format("20060102T150405.000000000") } m.snapshots[cp.ID] = &cp return nil diff --git a/internal/storage/sqlite.go b/internal/storage/sqlite.go index 8a62b95..f0b9422 100644 --- a/internal/storage/sqlite.go +++ b/internal/storage/sqlite.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "net/url" "sync" "time" @@ -30,7 +31,7 @@ type sqliteStore struct { func NewSQLiteStore(path string) (Store, error) { dsn := ":memory:" if path != ":memory:" { - dsn = "file:" + path + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)" + dsn = "file:" + (&url.URL{Path: path}).EscapedPath() + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(5000)&_pragma=synchronous(NORMAL)" } db, err := sql.Open("sqlite", dsn) if err != nil { @@ -78,11 +79,6 @@ const ( dedupIndexDDL = `CREATE UNIQUE INDEX IF NOT EXISTS idx_records_dedup ON records (cluster_id, stream_id, event_id) WHERE event_id <> ''` - // dedupIndexDrop removes a pre-migration index that was not partial. A - // unique index over empty event_id would collapse all special records on a - // stream (baselines, gaps, checkpoints) into one row. - dedupIndexDrop = `DROP INDEX IF EXISTS idx_records_dedup` - lookupIndexDDL = `CREATE INDEX IF NOT EXISTS idx_records_lookup ON records (cluster_id, stream_id, observed_at, namespace, name)` @@ -112,11 +108,32 @@ const ( ) func (s *sqliteStore) init(ctx context.Context) error { - for _, ddl := range []string{recordsDDL, dedupIndexDrop, dedupIndexDDL, lookupIndexDDL, streamsDDL, snapshotsDDL} { + for _, ddl := range []string{recordsDDL, dedupIndexDDL, lookupIndexDDL, streamsDDL, snapshotsDDL} { if _, err := s.db.ExecContext(ctx, ddl); err != nil { return err } } + // migrateTimeColumns normalizes timestamps written before the fixed-width + // format to the same shape so lexicographic comparisons stay chronological. + var userVersion int + if err := s.db.QueryRowContext(ctx, `PRAGMA user_version`).Scan(&userVersion); err != nil { + return err + } + if userVersion < 1 { + for _, q := range []string{ + `UPDATE records SET observed_at = strftime('%Y-%m-%dT%H:%M:%fZ', observed_at) WHERE observed_at <> ''`, + `UPDATE streams SET first_observed_at = strftime('%Y-%m-%dT%H:%M:%fZ', first_observed_at) WHERE first_observed_at <> ''`, + `UPDATE streams SET last_observed_at = strftime('%Y-%m-%dT%H:%M:%fZ', last_observed_at) WHERE last_observed_at <> ''`, + `UPDATE snapshots SET at = strftime('%Y-%m-%dT%H:%M:%fZ', at) WHERE at <> ''`, + } { + if _, err := s.db.ExecContext(ctx, q); err != nil { + return err + } + } + if _, err := s.db.ExecContext(ctx, `PRAGMA user_version = 1`); err != nil { + return err + } + } return nil } @@ -142,6 +159,7 @@ func (s *sqliteStore) Append(ctx context.Context, rec *event.Record) (int64, err return 0, err } if err := tx.Commit(); err != nil { + tx.Rollback() return 0, err } return seq, nil @@ -164,6 +182,7 @@ func (s *sqliteStore) Appends(ctx context.Context, recs []*event.Record) ([]int6 seqs = append(seqs, seq) } if err := tx.Commit(); err != nil { + tx.Rollback() return nil, err } return seqs, nil @@ -228,8 +247,10 @@ func (s *sqliteStore) appendRecord(ctx context.Context, tx *sql.Tx, rec *event.R if err := tx.QueryRowContext(ctx, `SELECT last_insert_rowid()`).Scan(&seq); err != nil { return 0, err } - if err := s.upsertMeta(ctx, tx, rec, observedAt); err != nil { - return 0, err + if rec.StreamID != "" { + if err := s.upsertMeta(ctx, tx, rec, observedAt); err != nil { + return 0, err + } } } else { // Duplicate event_id delivery: re-return the existing ingest sequence. @@ -280,6 +301,7 @@ func (s *sqliteStore) upsertMeta(ctx context.Context, tx *sql.Tx, rec *event.Rec if rec.Coverage != nil { if rec.Coverage.Current.Available { available = 1 + degraded = 0 } else { available = 0 degraded = 1 @@ -410,6 +432,10 @@ func (s *sqliteStore) Events(ctx context.Context, f EventFilter) ([]event.Record q += " AND observed_at<=?" args = append(args, formatTime(f.Until)) } + if f.SinceSeq > 0 { + q += " AND ingest_seq>?" + args = append(args, f.SinceSeq) + } q += " ORDER BY ingest_seq ASC" if f.Limit > 0 { q += " LIMIT ?" @@ -545,7 +571,7 @@ func (s *sqliteStore) SaveSnapshot(ctx context.Context, snap *SnapshotRef) error defer s.mu.Unlock() cp := *snap if cp.ID == "" { - cp.ID = "snap-" + time.Now().UTC().Format("20060102T150405") + cp.ID = "snap-" + time.Now().UTC().Format("20060102T150405.000000000") } _, err := s.db.ExecContext(ctx, `INSERT INTO snapshots (id, cluster_id, name, at) VALUES (?,?,?,?) @@ -577,7 +603,11 @@ func (s *sqliteStore) Snapshots(ctx context.Context) ([]SnapshotRef, error) { return out, rows.Err() } -func (s *sqliteStore) Close() error { return s.db.Close() } +func (s *sqliteStore) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + return s.db.Close() +} // scanFunc matches both *sql.Rows.Scan and *sql.Row.Scan. type scanFunc func(dest ...any) error @@ -652,11 +682,15 @@ func scanStreamMeta(scan scanFunc) (StreamMeta, error) { return m, nil } +// formatTime renders t as a fixed-width UTC timestamp. Variable-width RFC3339 +// output breaks lexicographic string comparisons at sub-second boundaries, so +// the fraction is always padded to milliseconds and the zone is always "Z". +// All comparisons against observed_at use this exact shape. func formatTime(t time.Time) string { if t.IsZero() { return "" } - return t.UTC().Format(time.RFC3339Nano) + return t.UTC().Format("2006-01-02T15:04:05.000") + "Z" } func parseTime(s string) (time.Time, error) { diff --git a/internal/storage/store.go b/internal/storage/store.go index 44758d7..f62f675 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -40,6 +40,7 @@ type EventFilter struct { RecordType event.RecordType Since time.Time Until time.Time + SinceSeq int64 // return only records with ingest_seq > SinceSeq (cursor) Limit int Offset int64 } diff --git a/internal/watch/collector.go b/internal/watch/collector.go index 4ef048d..7937fd8 100644 --- a/internal/watch/collector.go +++ b/internal/watch/collector.go @@ -15,6 +15,7 @@ import ( "k8s.io/client-go/dynamic" "github.com/krply/krply/internal/discovery" + "github.com/krply/krply/internal/metrics" "github.com/krply/krply/internal/storage" "github.com/krply/krply/internal/version" ) @@ -49,6 +50,14 @@ type Config struct { MinBackoff time.Duration MaxBackoff time.Duration + // WatchIdleTimeout forces a reconnect when no watch event or bookmark has + // arrived within the window. Zero uses a 10 minute default. It prevents a + // silently dead connection from stalling the stream forever. + WatchIdleTimeout time.Duration + + // Metrics, when non-nil, receives ingest counters for this collector. + Metrics *metrics.Metrics + // DynamicClient, when non-nil, is used instead of building a client from // KubeConfig. It exists to make the collector testable with fake clients. DynamicClient dynamic.Interface @@ -98,6 +107,9 @@ func applyConfigDefaults(cfg *Config) { if cfg.MinBackoff > cfg.MaxBackoff { cfg.MinBackoff = cfg.MaxBackoff } + if cfg.WatchIdleTimeout <= 0 { + cfg.WatchIdleTimeout = 10 * time.Minute + } if cfg.ClusterID == "" { cfg.ClusterID = "cluster-unknown" } @@ -115,8 +127,13 @@ func userAgent(cfg Config) string { // Run starts one stream goroutine per configured resource and waits for all of // them to stop. It returns nil when the context is cancelled and the first -// unrecoverable error (for example a closed store) otherwise. +// unrecoverable error (for example a closed store) otherwise. When one stream +// fails, the remaining streams are cancelled so no goroutine keeps writing to +// the store after Run has returned. func (c *Collector) Run(ctx context.Context) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + var wg sync.WaitGroup errCh := make(chan error, len(c.cfg.Resources)) @@ -125,6 +142,14 @@ func (c *Collector) Run(ctx context.Context) error { wg.Add(1) go func() { defer wg.Done() + defer func() { + if r := recover(); r != nil { + select { + case errCh <- fmt.Errorf("watch stream panic: %v", r): + case <-ctx.Done(): + } + } + }() if err := c.runStream(ctx, spec); err != nil { select { case errCh <- err: @@ -145,6 +170,8 @@ func (c *Collector) Run(ctx context.Context) error { <-done return nil case err := <-errCh: + cancel() + <-done return err case <-done: return nil diff --git a/internal/watch/stream.go b/internal/watch/stream.go index f395ba4..fd8aa68 100644 --- a/internal/watch/stream.go +++ b/internal/watch/stream.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "math/rand" + "strconv" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -65,12 +66,22 @@ func (c *Collector) runStream(ctx context.Context, spec discovery.ResourceSpec) } continue } + if rv == "" { + // A relist without a resource version would restart the watch + // from an arbitrary point and loop forever. + log.Warn("list returned an empty resourceVersion; backing off and relisting") + if err := c.sleepBackoff(ctx, &backoff); err != nil { + return nil + } + continue + } lastRV = rv backoff = c.cfg.MinBackoff } w, err := ri.Watch(ctx, metav1.ListOptions{ ResourceVersion: lastRV, + LabelSelector: c.cfg.Selector, AllowWatchBookmarks: c.cfg.Bookmarks, }) if err != nil { @@ -83,6 +94,9 @@ func (c *Collector) runStream(ctx context.Context, spec discovery.ResourceSpec) } continue } + if m := c.cfg.Metrics; m != nil { + m.Reconnects.WithLabelValues(sid).Inc() + } err = c.drainWatch(ctx, stream, sid, spec, w, &lastRV) w.Stop() @@ -93,9 +107,14 @@ func (c *Collector) runStream(ctx context.Context, spec discovery.ResourceSpec) return err } if errors.Is(err, errGone) { - log.Warn("410 Gone received; relisting") + // Persistent 410s (aggressive etcd compaction, flaky aggregated + // servers) would otherwise hammer the apiserver with an unthrottled + // list/watch/gap loop. + log.Warn("410 Gone received; relisting after backoff") lastRV = "" - backoff = c.cfg.MinBackoff + if err := c.sleepBackoff(ctx, &backoff); err != nil { + return nil + } continue } log.Warn("watch stream ended; reconnecting from last durable RV", "error", err) @@ -109,7 +128,7 @@ func (c *Collector) runStream(ctx context.Context, spec discovery.ResourceSpec) // synthetic ADDED record per item, and returns the collection resource version // to watch from. func (c *Collector) listAndBaseline(ctx context.Context, stream event.Stream, sid string, spec discovery.ResourceSpec, ri dynamic.ResourceInterface) (string, error) { - list, err := ri.List(ctx, metav1.ListOptions{ResourceVersion: "0"}) + list, err := ri.List(ctx, metav1.ListOptions{ResourceVersion: "0", LabelSelector: c.cfg.Selector}) if err != nil { return "", err } @@ -141,7 +160,7 @@ func (c *Collector) listAndBaseline(ctx context.Context, stream event.Stream, si ClusterID: c.cfg.ClusterID, StreamID: sid, Type: event.TypeEvent, - EventID: event.EventID(stream, ref, event.WatchAdded, 0), + EventID: event.EventID(stream, ref, event.WatchAdded, now.UnixNano()), ObservedAt: now, WatchType: event.WatchAdded, Synthetic: true, @@ -158,17 +177,38 @@ func (c *Collector) listAndBaseline(ctx context.Context, stream event.Stream, si // drainWatch consumes a watch stream, persisting every event. It returns nil // when the context is cancelled, errGone on 410 (gap already recorded), -// errReconnect on other errors (gap already recorded), errStore on a store -// failure, or a plain error when the channel closes. +// errReconnect on other errors, errStore on a store failure, or a plain error +// when the channel closes. Only a 410 loses data; every other teardown resumes +// from the last durable resource version, so no gap is recorded for them. func (c *Collector) drainWatch(ctx context.Context, stream event.Stream, sid string, spec discovery.ResourceSpec, w watch.Interface, lastRV *string) error { + var idle *time.Timer + var idleC <-chan time.Time + if c.cfg.WatchIdleTimeout > 0 { + idle = time.NewTimer(c.cfg.WatchIdleTimeout) + defer idle.Stop() + idleC = idle.C + } for { select { case <-ctx.Done(): return ctx.Err() + case <-idleC: + // No event (and no bookmark) for a full idle window: the connection + // is likely dead without a close. Force a reconnect so recording + // does not stall silently. + return errReconnect case ev, ok := <-w.ResultChan(): if !ok { return errors.New("watch channel closed") } + if idle != nil { + if !idle.Reset(c.cfg.WatchIdleTimeout) { + select { + case <-idle.C: + default: + } + } + } switch ev.Type { case watch.Added, watch.Modified, watch.Deleted: obj, ok := ev.Object.(*unstructured.Unstructured) @@ -182,6 +222,10 @@ func (c *Collector) drainWatch(ctx context.Context, stream event.Stream, sid str if _, err := c.cfg.Store.Append(ctx, rec); err != nil { return storeErrf("append watch event: %v", err) } + if m := c.cfg.Metrics; m != nil { + m.EventsIngested.WithLabelValues(sid).Inc() + m.IngestLag.Set(time.Since(rec.ObservedAt).Seconds()) + } if rec.Resource.ResourceVersion != "" { *lastRV = rec.Resource.ResourceVersion } @@ -191,6 +235,9 @@ func (c *Collector) drainWatch(ctx context.Context, stream event.Stream, sid str if rv == "" { continue } + if *lastRV != "" && resourceVersionBefore(rv, *lastRV) { + continue + } now := time.Now().UTC() rec := &event.Record{ ClusterID: c.cfg.ClusterID, @@ -219,21 +266,33 @@ func (c *Collector) drainWatch(ctx context.Context, stream event.Stream, sid str if err := c.writeGap(ctx, sid, spec, *lastRV, "", "410 Gone"); err != nil { return err } + if m := c.cfg.Metrics; m != nil { + m.Gone410.WithLabelValues(sid).Inc() + } return errGone } reason := "watch error" if st := statusOf(ev.Object); st != nil && st.Message != "" { reason = st.Message } - if err := c.writeGap(ctx, sid, spec, *lastRV, "", reason); err != nil { - return err - } + c.cfg.Log.Warn("watch error (not a gap); reconnecting from last durable RV", "reason", reason) return errReconnect } } } } +// resourceVersionBefore reports whether a < b when both parse as integers. +// Non-numeric resource versions are opaque and treated as incomparable. +func resourceVersionBefore(a, b string) bool { + ai, aerr := strconv.ParseInt(a, 10, 64) + bi, berr := strconv.ParseInt(b, 10, 64) + if aerr != nil || berr != nil { + return false + } + return ai < bi +} + func (c *Collector) watchEventRecord(stream event.Stream, sid string, spec discovery.ResourceSpec, wt event.WatchType, obj *unstructured.Unstructured, synthetic bool) (*event.Record, error) { raw, err := json.Marshal(obj.Object) if err != nil { @@ -277,6 +336,9 @@ func (c *Collector) writeGap(ctx context.Context, sid string, spec discovery.Res if _, err := c.cfg.Store.Append(ctx, rec); err != nil { return storeErrf("append gap: %v", err) } + if m := c.cfg.Metrics; m != nil { + m.Gaps.WithLabelValues(sid).Inc() + } return nil }