Skip to content

fix(kubernetes): keep investigations and workloads stable during sync #396

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hoalongnatsu merged 1 commit into from
Oct 31, 2026
Merged

fix(kubernetes): keep investigations and workloads stable during sync #396

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 7 additions & 9 deletions pkg/controllers/kubernetes_admin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,13 +199,6 @@ func connectorResolutionError(err error) error {
return fmt.Errorf("%w: %w", kubernetes.ErrConnectorCredentialUnresolved, err)
}

func (controller *KubernetesAdminController) authorizeInfrastructureView(ctx *fiber.Ctx) error {
if !callerCanViewInfrastructure(ctx) {
return refuseCallerInfrastructureView(ctx)
}
return ctx.Next()
}

// callerCanViewInfrastructure answers the permission itself and records the
// caller's authorization, but does not write a response or advance the chain. A
// gate that still has to look at the deployment asks this first, so the answer it
Expand Down Expand Up @@ -484,7 +477,7 @@ func deltaMatchesNamespace(delta kubernetesindex.Delta, namespace string) bool {
}
func (controller *KubernetesAdminController) issues(ctx *fiber.Ctx) error {
args := ctx.Context().QueryArgs()
for _, key := range []string{"broken", "include", "root_uid", "context"} {
for _, key := range []string{"broken", "include", "root_uid", "context", "occurrence_id", "center"} {
if len(args.PeekMulti(key)) > 1 {
return writeKubernetes(ctx, nil, kubernetes.ErrInvalidArguments)
}
Expand All @@ -506,6 +499,11 @@ func (controller *KubernetesAdminController) issues(ctx *fiber.Ctx) error {
if args.Has("context") && contextMode != "persisted" {
return writeKubernetes(ctx, nil, kubernetes.ErrInvalidArguments)
}
occurrenceID := strings.Clone(ctx.Query("occurrence_id"))
center, centerErr := parseKubernetesTime(ctx.Query("center"))
if centerErr != nil || (args.Has("occurrence_id") || args.Has("center")) && contextMode != "persisted" || args.Has("occurrence_id") && occurrenceID == "" || args.Has("center") && center.IsZero() {
return writeKubernetes(ctx, nil, kubernetes.ErrInvalidArguments)
}
if args.Has("root_uid") && (root == "" || include != "changes" && contextMode != "persisted") {
return writeKubernetes(ctx, nil, kubernetes.ErrInvalidArguments)
}
Expand All @@ -519,7 +517,7 @@ func (controller *KubernetesAdminController) issues(ctx *fiber.Ctx) error {
}
value, err := requestKubernetesService(ctx).Issues(ctx.UserContext(), kubernetes.IssueOptions{
Since: since, Until: until, Namespace: ctx.Query("namespace"), Severity: ctx.Query("severity"), Kind: ctx.Query("kind"), Limit: queryInt(ctx, "limit"), Cursor: ctx.Query("cursor"),
Broken: broken, IncludeChanges: include == "changes", RootUID: root, Context: contextMode,
Broken: broken, IncludeChanges: include == "changes", RootUID: root, Context: contextMode, OccurrenceID: occurrenceID, Center: center,
})
return writeKubernetes(ctx, value, err)
}
Expand Down
20 changes: 20 additions & 0 deletions pkg/controllers/kubernetes_admin_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -359,6 +359,26 @@ func TestKubernetesIssuesQueryValidationAndInvestigate(t *testing.T) {
return ctx.Next()
})
NewKubernetesAdminController(service).Register(app.Group("/api"))
for _, query := range []string{
"?occurrence_id=selected",
"?center=2026-10-01T00:00:00Z",
"?context=persisted&root_uid=deployment-checkout&occurrence_id=",
"?context=persisted&root_uid=deployment-checkout&occurrence_id=bad%2Fid",
"?context=persisted&root_uid=deployment-checkout&occurrence_id=a&occurrence_id=b",
"?context=persisted&root_uid=deployment-checkout&center=",
"?context=persisted&root_uid=deployment-checkout&center=not-a-time",
"?context=persisted&root_uid=deployment-checkout&center=2999-01-01T00:00:00Z",
"?context=persisted&root_uid=deployment-checkout&center=2026-10-01T00:00:00Z&center=2026-10-01T00:00:00Z",
} {
response, err := app.Test(httptest.NewRequest("GET", "/api/admin/kubernetes/issues"+query, nil), -1)
if err != nil {
t.Fatal(err)
}
response.Body.Close()
if response.StatusCode != http.StatusBadRequest {
t.Fatalf("invalid selector accepted: status=%d", response.StatusCode)
}
}
for _, test := range []struct {
query string
want int
Expand Down
55 changes: 50 additions & 5 deletions pkg/kubernetes/cached_authority_regression_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@ package kubernetes

import (
"bytes"
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strconv"
"strings"
"sync/atomic"
"testing"
Expand All @@ -15,6 +17,40 @@ import (
"github.com/VersusControl/versus-incident/tests/fakekube"
)

// Observe the actual durable head and journal writes, without changing their
// timing or guards. The LIST attempt provenance identifies which cycle wrote a
// checkpoint even if its journal commit completes after permission revocation.
type authorityObservedStore struct {
storage.Provider
storage.BlobCAS
storage.EvidenceStore
t *testing.T
mode *atomic.Int32
}

func (p *authorityObservedStore) CompareAndSwapBlob(key string, old, next []byte) (bool, error) {
won, err := p.BlobCAS.CompareAndSwapBlob(key, old, next)
if won && err == nil {
var a, b inventoryLeaseRecord
if json.Unmarshal(old, &a) == nil && json.Unmarshal(next, &b) == nil && b.Revision != "" && b.Revision != a.Revision {
raw, readErr := p.ReadBlob(key + "/data/" + b.DataOwner + "/" + strconv.Itoa(b.Slot))
var generation inventoryGeneration
if readErr == nil && json.Unmarshal(raw, &generation) == nil {
p.t.Logf("durable head mode=%d revision=%s pod=%+v deployment=%+v", p.mode.Load(), b.Revision, generation.View.Kinds["Pod"], generation.View.Kinds["Deployment"])
}
}
}
return won, err
}

func (p *authorityObservedStore) CommitEvidence(ctx context.Context, scope storage.EvidenceScope, batch storage.EvidenceBatch) error {
err := p.EvidenceStore.CommitEvidence(ctx, scope, batch)
if err == nil {
p.t.Logf("durable journal mode=%d id=%s expected=%s at=%s", p.mode.Load(), batch.ID, batch.Expected, batch.At.Format(time.RFC3339Nano))
}
return err
}

func TestAutonomousCachedCatalogClassifiesCurrentMaterialAuthority(t *testing.T) {
for _, backend := range []string{"memory", "file"} {
for _, scenario := range []string{"all", "multiple-kinds", "secret-only"} {
Expand Down Expand Up @@ -84,6 +120,7 @@ func TestAutonomousCachedCatalogClassifiesCurrentMaterialAuthority(t *testing.T)
t.Fatal(err)
}
}
provider = &authorityObservedStore{Provider: provider, BlobCAS: provider.(storage.BlobCAS), EvidenceStore: provider.(storage.EvidenceStore), t: t, mode: &mode}
service.SetChangeStorage(provider)
registry := NewServiceRegistry(service)
defer registry.Close()
Expand Down Expand Up @@ -123,18 +160,24 @@ func TestAutonomousCachedCatalogClassifiesCurrentMaterialAuthority(t *testing.T)
t.Fatal(err)
}
beforeRaw, _ := json.Marshal(before)
checkpoint, err := store.EvidenceCheckpoint(t.Context(), evidenceScope)
if err != nil {
t.Fatal(err)
}
// Ready data is visible before its journal commit finishes. Pin the
// checkpoint belonging to this exact head before revoking permission;
// an authorized baseline commit is not a denied-cycle publication.
var checkpoint storage.EvidenceCheckpoint
waitUntil(t, 8*time.Second, "baseline data-head journal committed", func() bool {
checkpoint, err = store.EvidenceCheckpoint(t.Context(), evidenceScope)
return err == nil && before.EvidenceBatch != nil && checkpoint.ID == before.EvidenceBatch.ID
})
catalogs := catalogReads.Load()
t.Logf("baseline checkpoint=%s at=%s revision=%s revoke_at=%s", checkpoint.ID, checkpoint.At.Format(time.RFC3339Nano), record.Revision, time.Now().UTC().Format(time.RFC3339Nano))
if scenario == "all" {
mode.Store(1)
} else {
mode.Store(3)
}
select {
case <-materialSeen:
t.Logf("first material denial at=%s", time.Now().UTC().Format(time.RFC3339Nano))
case <-time.After(40 * time.Second):
t.Fatal("normal material cadence never observed the changed permissions")
}
Expand Down Expand Up @@ -177,7 +220,9 @@ func TestAutonomousCachedCatalogClassifiesCurrentMaterialAuthority(t *testing.T)
report, _ := health()
workloads, workErr := service.Workloads(t.Context(), WorkloadListOptions{})
issues, issueErr := service.Issues(t.Context(), IssueOptions{})
t.Fatalf("cached catalog bypassed current proof: denied=%d sync=%s failing=%d unreadable=%d Workloads=%d/%v Issues=%d/%v", materialDenials.Load(), report.State, report.FailingKinds, report.UnreadableKinds, len(workloads.Items), workErr, len(issues.Items), issueErr)
current, _ := store.EvidenceCheckpoint(t.Context(), evidenceScope)
latest, head, _ := readInventoryGeneration(provider, scope, "")
t.Fatalf("cached catalog bypassed current proof: denied=%d sync=%s failing=%d unreadable=%d Workloads=%d/%v Issues=%d/%v checkpoint=%s/%s revision=%s pod=%+v", materialDenials.Load(), report.State, report.FailingKinds, report.UnreadableKinds, len(workloads.Items), workErr, len(issues.Items), issueErr, current.ID, current.At.Format(time.RFC3339Nano), head.Revision, latest.View.Kinds["Pod"])
}
pending, _ := health()
if pending.State == "ready" {
Expand Down
40 changes: 35 additions & 5 deletions pkg/kubernetes/evidence.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -721,7 +721,11 @@ func (service *Service) historicalIssues(ctx context.Context, options IssueOptio
}

func (service *Service) persistedRootContext(ctx context.Context, options IssueOptions) (IssuePage, error) {
page := IssuePage{Items: []Issue{}, Totals: map[string]int{}, Broken: BrokenTotals{BySeverity: map[string]int{}}, Sync: SyncStatus{State: "ready"}}
page := IssuePage{Items: []Issue{}, Totals: map[string]int{}, Broken: BrokenTotals{BySeverity: map[string]int{}}, Sync: SyncStatus{State: "ready"}, SelectionStatus: "not_requested"}
selecting := options.OccurrenceID != "" || !options.Center.IsZero()
if selecting {
page.SelectionStatus = "not_found"
}
// Search retained occurrences, not the selected evidence interval. A missing
// live object, recovered issue, or empty selected interval cannot erase this
// identity. Each candidate still passes the normal current authority gate.
Expand All @@ -733,19 +737,45 @@ func (service *Service) persistedRootContext(ctx context.Context, options IssueO
}
page.Sync.Partial = page.Sync.Partial || history.Sync.Partial
page.Partial = append(page.Partial, history.Partial...)
if len(history.Items) > 0 {
issue := history.Items[0]
for _, issue := range history.Items {
if issue.Root.UID != options.RootUID || options.Namespace != "" && issue.Root.Namespace != options.Namespace {
return page, ErrIncompleteInventory
}
page.RootContext = &PersistedRootContext{Root: issue.Root, Occurrence: issue, Retained: true}
return page, nil
if page.RootContext == nil {
page.RootContext = &PersistedRootContext{Root: issue.Root, Occurrence: issue, Retained: true}
}
if !selecting {
return page, nil
}
if options.OccurrenceID != "" {
// ID is authoritative: center never permits a different occurrence.
if issue.OccurrenceID == options.OccurrenceID {
page.SelectedOccurrence, page.SelectionStatus = &issue, "resolved"
return page, nil
}
} else if issue.BrokeAt != nil && issue.BrokeAt.Equal(options.Center) {
if page.SelectionStatus == "resolved" {
page.SelectedOccurrence, page.SelectionStatus = nil, "ambiguous"
} else if page.SelectionStatus != "ambiguous" {
page.SelectedOccurrence, page.SelectionStatus = &issue, "resolved"
}
}
}
if history.Next == "" {
if selecting && page.Sync.Partial && page.SelectionStatus != "ambiguous" {
page.SelectedOccurrence, page.SelectionStatus = nil, "partial"
}
return page, nil
}
options.Cursor = history.Next
}
if selecting {
// A bounded scan cannot prove absence or uniqueness beyond its horizon.
page.SelectedOccurrence, page.SelectionStatus = nil, "partial"
page.Sync.Partial = true
page.Partial = append(page.Partial, PartialFailure{Class: "selection_scan_limit"})
return page, nil
}
return page, storage.ErrEvidenceLimit
}

Expand Down
119 changes: 119 additions & 0 deletions pkg/kubernetes/evidence_review_regression_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -366,6 +366,117 @@ func TestPersistedRootContextIgnoresEvidenceWindowAndLiveExistence(t *testing.T)
}
}

func TestPersistedRootSelectionNeverSubstitutesAnotherOccurrence(t *testing.T) {
for _, backend := range []string{"memory", "file"} {
t.Run(backend, func(t *testing.T) {
service, store, scope := reviewHistoryService(t)
if backend == "file" {
provider, err := storage.NewFile(storage.FileOptions{DataDir: t.TempDir()})
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { provider.Close() })
service.SetChangeStorage(provider)
store = provider.(storage.EvidenceStore)
}
at := time.Now().UTC().Add(-2 * time.Hour).Truncate(time.Second).Add(123456789 * time.Nanosecond)
root := ObjectRef{Kind: "Deployment", Namespace: "shop", Name: "retained", UID: "retained-root"}
batch := storage.EvidenceBatch{ID: "capture:selected", At: at, Epoch: "synthetic"}
for i := 0; i < 3; i++ {
broke := at
if i == 2 {
broke = at.Add(time.Minute)
}
rule := fmt.Sprintf("synthetic.rule-%d", i)
raw, _ := json.Marshal(Issue{Root: root, Rule: rule, Broken: true, BrokeAt: &broke, Severity: "warning"})
batch.Issues = append(batch.Issues, storage.EvidenceIssue{Key: rule, Namespace: root.Namespace, Kind: root.Kind, UID: root.UID, Rule: rule, LastSeen: at, ObservationOnly: true, Payload: raw})
}
if err := store.CommitEvidence(t.Context(), scope, batch); err != nil {
t.Fatal(err)
}
history, err := service.historicalIssues(t.Context(), IssueOptions{RootUID: root.UID, Namespace: root.Namespace, Since: at.Add(-time.Minute), Until: time.Now()})
if err != nil || len(history.Items) != 3 {
t.Fatal("synthetic retained occurrences unavailable")
}
selected := history.Items[2]
for _, test := range []struct {
name, id, namespace, uid, status string
center time.Time
}{
{name: "exact", id: selected.OccurrenceID, status: "resolved"},
{name: "mismatch", id: "missing-id", center: *selected.BrokeAt, status: "not_found"},
{name: "unique-center", center: at.Add(time.Minute), status: "resolved"},
{name: "millisecond-center-does-not-substitute", center: at.Add(time.Minute).Truncate(time.Millisecond), status: "not_found"},
{name: "nearby-center-does-not-substitute", center: at.Add(time.Minute + time.Nanosecond), status: "not_found"},
{name: "ambiguous-center", center: at, status: "ambiguous"},
{name: "no-selector", status: "not_requested"},
{name: "wrong-namespace", id: selected.OccurrenceID, namespace: "other", status: "not_found"},
{name: "same-name-new-uid", id: selected.OccurrenceID, uid: "replacement-root", status: "not_found"},
} {
t.Run(test.name, func(t *testing.T) {
options := IssueOptions{Context: "persisted", RootUID: root.UID, Namespace: root.Namespace}
if test.namespace != "" {
options.Namespace = test.namespace
}
if test.uid != "" {
options.RootUID = test.uid
}
options.OccurrenceID, options.Center = test.id, test.center
page, err := service.Issues(t.Context(), options)
if err != nil {
t.Fatal(err)
}
raw, _ := json.Marshal(page)
var response struct {
Status string `json:"selection_status"`
Selected *Issue `json:"selected_occurrence"`
}
if err := json.Unmarshal(raw, &response); err != nil {
t.Fatal(err)
}
if response.Status != test.status {
t.Fatalf("selection status=%q want=%q", response.Status, test.status)
}
if test.status != "resolved" && response.Selected != nil {
t.Fatal("unresolved selector substituted another recurrence")
}
if test.name == "exact" && (response.Selected == nil || response.Selected.OccurrenceID != selected.OccurrenceID) {
t.Fatal("explicit occurrence did not survive date-free root lookup")
}
if test.name == "unique-center" && (response.Selected == nil || response.Selected.BrokeAt == nil || !response.Selected.BrokeAt.Equal(test.center)) {
t.Fatal("nanosecond center did not preserve the exact retained occurrence")
}
if test.name == "no-selector" && (page.RootContext == nil || response.Selected != nil) {
t.Fatal("legacy root link lost retained context or selected an arbitrary occurrence")
}
if len(page.Items) != 0 || len(page.Totals) != 0 || page.Next != "" {
t.Fatal("selected context polluted historical paging")
}
})
}
})
}
}

func TestPersistedRootSelectionReportsBoundedScanWithoutFalseUniqueness(t *testing.T) {
service, store, scope := reviewHistoryService(t)
at := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
root := ObjectRef{Kind: "Deployment", Namespace: "shop", Name: "retained", UID: "retained-root"}
batch := storage.EvidenceBatch{ID: "capture:many", At: at, Epoch: "synthetic"}
for i := 0; i < 1001; i++ {
rule := fmt.Sprintf("synthetic.rule-%04d", i)
raw, _ := json.Marshal(Issue{Root: root, Rule: rule, BrokeAt: &at, Broken: true, Severity: "warning"})
batch.Issues = append(batch.Issues, storage.EvidenceIssue{Key: rule, Namespace: root.Namespace, Kind: root.Kind, UID: root.UID, Rule: rule, LastSeen: at, ObservationOnly: true, Payload: raw})
}
if err := store.CommitEvidence(t.Context(), scope, batch); err != nil {
t.Fatal(err)
}
page, err := service.Issues(t.Context(), IssueOptions{Context: "persisted", RootUID: root.UID, Namespace: root.Namespace, OccurrenceID: "not-in-scanned-prefix"})
if err != nil || page.SelectionStatus != "partial" || page.SelectedOccurrence != nil || page.RootContext == nil || !page.Sync.Partial {
t.Fatal("bounded scan claimed absence or erased retained root")
}
}

func TestScopedEvidenceProjectsUnavailableRecordsBeforeCountsAndPagination(t *testing.T) {
for _, backend := range []string{"memory", "file", "postgres"} {
t.Run(backend, func(t *testing.T) {
Expand Down Expand Up @@ -502,6 +613,11 @@ func TestScopedEvidenceProjectsUnavailableRecordsBeforeCountsAndPagination(t *te
if err != nil || contextPage.RootContext == nil || contextPage.RootContext.Root.UID != "retained-root" || len(contextPage.Items) != 0 || !contextPage.Sync.Partial {
t.Fatal("persisted context required a live object, hid a malformed observation, or leaked into history")
}
contextOptions.OccurrenceID = contextPage.RootContext.Occurrence.OccurrenceID
selectedPage, err := service.Issues(t.Context(), contextOptions)
if err != nil || selectedPage.SelectedOccurrence == nil || selectedPage.SelectionStatus != "resolved" || selectedPage.SelectedOccurrence.BrokeAt != nil {
t.Fatal("exact missing-time occurrence was substituted or fabricated a timestamp")
}
occurrences, err := service.queryEvidence(t.Context(), storage.EvidenceQuery{Since: at.Add(-time.Minute), Until: at.Add(time.Minute), Namespace: "sample", RootUID: "retained-root", Type: "occurrence", Limit: 1})
if err != nil || occurrences.Total != 2 || occurrences.Counts["issue_opened"] != 1 || occurrences.Counts["coverage_gap"] != 1 || len(occurrences.Items) != 1 || occurrences.Next == "" {
t.Fatal("occurrence projection did not preserve accurate counts and bounded opaque pagination")
Expand Down Expand Up @@ -580,6 +696,9 @@ func TestScopedEvidenceProjectsUnavailableRecordsBeforeCountsAndPagination(t *te
}
denyPod.Store(false)
wrongCluster.Store(true)
if selected, err := service.Issues(t.Context(), contextOptions); err != nil || selected.SelectedOccurrence != nil || selected.RootContext != nil || selected.SelectionStatus != "not_found" {
t.Fatal("selected occurrence bypassed physical cluster identity")
}
page, err := service.queryEvidence(t.Context(), global)
if err != nil || len(page.Items) != 0 {
t.Fatal("physical cluster replacement exposed retained evidence")
Expand Down
Loading
Loading