diff --git a/config/observability-dashboards.yaml b/config/observability-dashboards.yaml index 4c05a2d..ab6a1b8 100644 --- a/config/observability-dashboards.yaml +++ b/config/observability-dashboards.yaml @@ -64,6 +64,37 @@ dashboards: query: max(gha_fleet_queue_uncovered_running) unit: count description: GitHub running jobs absent from the durable queue model. + - id: diagnostic_storage + title: Diagnostic storage retention + refresh_seconds: 60 + default_range: 24h + owner: fleet-operations + runbook: https://github.com/NDDev-OpenNetwork/github-actions/blob/main/docs/runbooks/fleet-alerts.md + panels: + - id: expiration_eligible + title: Objects past lifecycle eligibility + kind: timeseries + query: max(gha_diagnostic_storage_expiration_eligible_objects) + unit: count + description: Objects retained after their S3 UTC-midnight expiration boundary. + - id: next_expiration + title: Time to next lifecycle boundary + kind: timeseries + query: max(gha_diagnostic_storage_next_expiration_seconds) + unit: seconds + description: Seconds until the earliest retained object becomes expiration eligible. + - id: objects + title: Retained diagnostic objects + kind: timeseries + query: max(gha_diagnostic_storage_objects) + unit: count + description: Current object population under the managed diagnostic prefix. + - id: oldest_object_age + title: Oldest retained object age + kind: timeseries + query: max(gha_diagnostic_storage_oldest_object_age_seconds) + unit: seconds + description: Age of the oldest retained diagnostic object. - id: lifecycle_latency title: Job lifecycle latency refresh_seconds: 30 diff --git a/config/observability-rules.yaml b/config/observability-rules.yaml index e7f9f01..598aa47 100644 --- a/config/observability-rules.yaml +++ b/config/observability-rules.yaml @@ -34,6 +34,22 @@ rules: summary: Durable diagnostic export is failing. action: Preserve the spool and restore RustFS routing or credentials without pruning evidence. recovery: Consecutive failures and pending backlog return to zero after an exact successful export. + - id: diagnostic_retention_overdue + severity: ticket + query_language: promql + stream_name: gha_diagnostic_storage_expiration_eligible_objects + expression: max(gha_diagnostic_storage_expiration_eligible_objects) + operator: ">" + threshold: 0 + evaluation_seconds: 300 + hold_seconds: 3600 + destination_ref: fleet_oncall + enabled: false + owner: fleet-operations + runbook: https://github.com/NDDev-OpenNetwork/github-actions/blob/main/docs/runbooks/fleet-alerts.md + summary: Diagnostic objects remain after their S3 lifecycle eligibility boundary. + action: Preserve object-age counts and lifecycle read-back, then repair RustFS expiry processing without deleting evidence manually. + recovery: No expiration-eligible diagnostic object remains after the one-hour implementation grace. - id: fleet_platform_unhealthy severity: page query_language: promql diff --git a/internal/diagnosticstore/reconcile.go b/internal/diagnosticstore/reconcile.go index 2faf32e..93336f9 100644 --- a/internal/diagnosticstore/reconcile.go +++ b/internal/diagnosticstore/reconcile.go @@ -6,6 +6,7 @@ import ( "encoding/xml" "fmt" "net/http" + "net/url" "os" "strings" "time" @@ -28,9 +29,16 @@ type Result struct { HeadroomState string `json:"headroom_state"` UsageObservedAt time.Time `json:"usage_observed_at"` Actions []string `json:"actions"` + ObjectCount int `json:"object_count"` + OldestObjectModified time.Time `json:"oldest_object_modified_at,omitempty"` + ExpirationEligible int `json:"expiration_eligible_objects"` + NextExpirationAt time.Time `json:"next_expiration_at,omitempty"` } -type Runner struct{ Requester rustfscache.Requester } +type Runner struct { + Requester rustfscache.Requester + Now func() time.Time +} type quotaInfo struct { Quota int64 `json:"quota"` @@ -59,6 +67,21 @@ type lifecycleRule struct { } `xml:"Expiration"` } +type listBucketResult struct { + IsTruncated bool `xml:"IsTruncated"` + NextContinuationToken string `xml:"NextContinuationToken"` + Contents []struct { + LastModified string `xml:"LastModified"` + } `xml:"Contents"` +} + +type objectInventory struct { + Count int + OldestModified time.Time + ExpirationEligible int + NextExpiration time.Time +} + func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, error) { if err := config.Validate(); err != nil { return Result{}, err @@ -71,6 +94,10 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err return Result{}, err } defer clear(root.SecretKey) + now := time.Now().UTC() + if r.Now != nil { + now = r.Now().UTC() + } state, usage, err := r.inspect(ctx, root, config) if err != nil { return Result{}, err @@ -82,7 +109,12 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err result.CurrentUsageBytes = usage.CurrentUsage result.RemainingQuotaBytes = usage.RemainingQuota result.UsagePercentage = usage.UsagePercentage - result.UsageObservedAt = time.Now().UTC() + result.UsageObservedAt = now + inventory, inventoryErr := r.listObjects(ctx, root, config, now) + if inventoryErr != nil { + return Result{}, inventoryErr + } + applyInventory(&result, inventory) result.HeadroomState = "sufficient" if usage.RemainingQuota < config.MinimumHeadroom { result.HeadroomState = "below-minimum" @@ -111,7 +143,12 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err result.CurrentUsageBytes = usageAfter.CurrentUsage result.RemainingQuotaBytes = usageAfter.RemainingQuota result.UsagePercentage = usageAfter.UsagePercentage - result.UsageObservedAt = time.Now().UTC() + result.UsageObservedAt = now + inventory, err := r.listObjects(ctx, root, config, now) + if err != nil { + return Result{}, err + } + applyInventory(&result, inventory) if usageAfter.RemainingQuota < config.MinimumHeadroom { result.HeadroomState = "below-minimum" } else { @@ -120,6 +157,71 @@ func (r Runner) Run(ctx context.Context, config Config, apply bool) (Result, err return result, nil } +func (r Runner) listObjects(ctx context.Context, root rustfscache.Credential, config Config, now time.Time) (objectInventory, error) { + const maximumPages = 1024 + inventory := objectInventory{} + token := "" + for page := 0; page < maximumPages; page++ { + query := url.Values{"list-type": {"2"}, "prefix": {config.Prefix + "/"}} + if token != "" { + query.Set("continuation-token", token) + } + response, err := r.Requester.Do(ctx, root, http.MethodGet, "/"+config.Bucket+"?"+query.Encode(), "", nil) + if err != nil { + return objectInventory{}, fmt.Errorf("list diagnostic objects: %w", err) + } + if response.StatusCode != http.StatusOK { + return objectInventory{}, responseError("list diagnostic objects", response) + } + var listed listBucketResult + if err := xml.Unmarshal(response.Body, &listed); err != nil { + return objectInventory{}, fmt.Errorf("decode diagnostic object listing: %w", err) + } + for _, object := range listed.Contents { + modified, err := time.Parse(time.RFC3339Nano, object.LastModified) + if err != nil { + return objectInventory{}, fmt.Errorf("decode diagnostic object timestamp: %w", err) + } + modified = modified.UTC() + expires := lifecycleExpirationAt(modified, config.RetentionDays) + inventory.Count++ + if inventory.OldestModified.IsZero() || modified.Before(inventory.OldestModified) { + inventory.OldestModified = modified + } + if !now.Before(expires) { + inventory.ExpirationEligible++ + } + if inventory.NextExpiration.IsZero() || expires.Before(inventory.NextExpiration) { + inventory.NextExpiration = expires + } + } + if !listed.IsTruncated { + return inventory, nil + } + if listed.NextContinuationToken == "" { + return objectInventory{}, fmt.Errorf("truncated diagnostic object listing omitted continuation token") + } + token = listed.NextContinuationToken + } + return objectInventory{}, fmt.Errorf("diagnostic object listing exceeded %d pages", maximumPages) +} + +func lifecycleExpirationAt(modified time.Time, retentionDays int) time.Time { + candidate := modified.UTC().Add(time.Duration(retentionDays) * 24 * time.Hour) + midnight := candidate.Truncate(24 * time.Hour) + if candidate.Equal(midnight) { + return midnight + } + return midnight.Add(24 * time.Hour) +} + +func applyInventory(result *Result, inventory objectInventory) { + result.ObjectCount = inventory.Count + result.OldestObjectModified = inventory.OldestModified + result.ExpirationEligible = inventory.ExpirationEligible + result.NextExpirationAt = inventory.NextExpiration +} + func (r Runner) inspect(ctx context.Context, root rustfscache.Credential, config Config) (string, quotaStats, error) { bucket, err := r.Requester.Do(ctx, root, http.MethodHead, "/"+config.Bucket, "", nil) if err != nil { diff --git a/internal/diagnosticstore/reconcile_test.go b/internal/diagnosticstore/reconcile_test.go index 614c028..a8aff83 100644 --- a/internal/diagnosticstore/reconcile_test.go +++ b/internal/diagnosticstore/reconcile_test.go @@ -6,7 +6,9 @@ import ( "net/http" "os" "path/filepath" + "strings" "testing" + "time" "github.com/NDDev-OpenNetwork/github-actions/internal/rustfscache" ) @@ -16,6 +18,7 @@ type fakeRequester struct { quota int64 usage int64 lifecycle []byte + objects []time.Time } func (f *fakeRequester) Do(_ context.Context, _ rustfscache.Credential, method, path, _ string, body []byte) (rustfscache.Response, error) { @@ -49,6 +52,14 @@ func (f *fakeRequester) Do(_ context.Context, _ rustfscache.Credential, method, case method == http.MethodPut && path == "/example-diagnostics?lifecycle": f.lifecycle = append([]byte(nil), body...) return rustfscache.Response{StatusCode: http.StatusOK}, nil + case method == http.MethodGet && strings.HasPrefix(path, "/example-diagnostics?list-type=2&prefix=diagnostics%2Fv1%2F"): + var payload strings.Builder + payload.WriteString("false") + for _, modified := range f.objects { + payload.WriteString("" + modified.UTC().Format(time.RFC3339Nano) + "") + } + payload.WriteString("") + return rustfscache.Response{StatusCode: http.StatusOK, Body: []byte(payload.String())}, nil default: return rustfscache.Response{StatusCode: http.StatusBadRequest}, nil } @@ -113,6 +124,23 @@ func TestReportsLowRemoteHeadroom(t *testing.T) { } } +func TestObjectInventoryUsesLifecycleMidnightEligibility(t *testing.T) { + config, remote := diagnosticStoreFixture(t) + remote.bucket, remote.quota = true, config.QuotaBytes + remote.lifecycle = lifecycleDocument(config) + modified := time.Date(2026, 8, 19, 0, 36, 58, 0, time.UTC) + remote.objects = []time.Time{modified} + now := time.Date(2026, 8, 26, 3, 37, 39, 0, time.UTC) + result, err := (Runner{Requester: remote, Now: func() time.Time { return now }}).Run(context.Background(), config, false) + if err != nil { + t.Fatal(err) + } + if result.ObjectCount != 1 || result.OldestObjectModified != modified || result.ExpirationEligible != 0 || + result.NextExpirationAt != time.Date(2026, 8, 27, 0, 0, 0, 0, time.UTC) { + t.Fatalf("unexpected lifecycle inventory: %+v", result) + } +} + func TestRejectsQuotaWithoutHeadroom(t *testing.T) { config := Config{SchemaVersion: 1, Endpoint: "https://192.0.2.1:9002", Region: "us-east-1", CAFile: "/tmp/ca", RootAccessKeyFile: "/tmp/access", RootSecretKeyFile: "/tmp/secret", Bucket: "example-diagnostics", Prefix: "diagnostics/v1", diff --git a/internal/diagnosticstoreobserve/observe.go b/internal/diagnosticstoreobserve/observe.go index 285ce98..e8fa70c 100644 --- a/internal/diagnosticstoreobserve/observe.go +++ b/internal/diagnosticstoreobserve/observe.go @@ -119,6 +119,18 @@ func Render(snapshot Snapshot, now time.Time) string { gauge("gha_diagnostic_storage_usage_percent", "Percentage of the diagnostic bucket hard quota currently used.", snapshot.Result.UsagePercentage) gauge("gha_diagnostic_storage_growth_bytes_per_second", "Non-negative growth rate between the two latest successful signed snapshots.", snapshot.GrowthBytesPerSecond) gauge("gha_diagnostic_storage_forecast_exhaustion_seconds", "Seconds to hard-quota exhaustion at the latest positive growth rate, or -1 when not forecastable.", snapshot.ForecastExhaustionSeconds) + gauge("gha_diagnostic_storage_objects", "Objects retained under the diagnostic prefix.", float64(snapshot.Result.ObjectCount)) + oldestObjectAge := float64(-1) + if !snapshot.Result.OldestObjectModified.IsZero() && !snapshot.Result.OldestObjectModified.After(now) { + oldestObjectAge = now.Sub(snapshot.Result.OldestObjectModified).Seconds() + } + gauge("gha_diagnostic_storage_oldest_object_age_seconds", "Age of the oldest retained diagnostic object, or -1 when empty.", oldestObjectAge) + gauge("gha_diagnostic_storage_expiration_eligible_objects", "Objects retained after their S3 lifecycle UTC-midnight eligibility boundary.", float64(snapshot.Result.ExpirationEligible)) + nextExpiration := float64(-1) + if !snapshot.Result.NextExpirationAt.IsZero() { + nextExpiration = max(0, snapshot.Result.NextExpirationAt.Sub(now).Seconds()) + } + gauge("gha_diagnostic_storage_next_expiration_seconds", "Seconds until the earliest retained object reaches its lifecycle eligibility boundary, or -1 when empty.", nextExpiration) return output.String() } diff --git a/internal/diagnosticstoreobserve/observe_test.go b/internal/diagnosticstoreobserve/observe_test.go index 855bf18..c3efcfe 100644 --- a/internal/diagnosticstoreobserve/observe_test.go +++ b/internal/diagnosticstoreobserve/observe_test.go @@ -34,6 +34,26 @@ func TestSampleRendersHeadroomAndForecast(t *testing.T) { } } +func TestRetentionMetricsExposeEligibleObjectsAndNextBoundary(t *testing.T) { + now := time.Date(2026, 8, 26, 3, 0, 0, 0, time.UTC) + snapshot := Snapshot{CapturedAt: now, Result: diagnosticstore.Result{ + StateAfter: "managed", HeadroomState: "sufficient", ObjectCount: 356, + OldestObjectModified: time.Date(2026, 8, 19, 0, 36, 58, 0, time.UTC), + ExpirationEligible: 0, NextExpirationAt: time.Date(2026, 8, 27, 0, 0, 0, 0, time.UTC), + }} + metrics := Render(snapshot, now) + for _, wanted := range []string{ + "gha_diagnostic_storage_objects 356\n", + "gha_diagnostic_storage_oldest_object_age_seconds 613382\n", + "gha_diagnostic_storage_expiration_eligible_objects 0\n", + "gha_diagnostic_storage_next_expiration_seconds 75600\n", + } { + if !strings.Contains(metrics, wanted) { + t.Fatalf("metrics missing %q\n%s", wanted, metrics) + } + } +} + func TestHealthFailsClosedForLowHeadroomAndStaleness(t *testing.T) { now := time.Date(2026, 8, 21, 8, 0, 0, 0, time.UTC) state := &State{snapshot: Snapshot{CapturedAt: now, Result: diagnosticstore.Result{StateAfter: "managed", HeadroomState: "below-minimum"}}} diff --git a/internal/observabilitydashboards/dashboards.go b/internal/observabilitydashboards/dashboards.go index 1d09e56..4690d70 100644 --- a/internal/observabilitydashboards/dashboards.go +++ b/internal/observabilitydashboards/dashboards.go @@ -130,7 +130,7 @@ func (p Panel) Validate() error { if _, ok := map[string]struct{}{"bytes": {}, "count": {}, "percent": {}, "seconds": {}, "state": {}}[p.Unit]; !ok { return fmt.Errorf("unit is invalid") } - if !strings.Contains(p.Query, "gha_fleet_") && !strings.Contains(p.Query, "otelcol_exporter_") { + if !strings.Contains(p.Query, "gha_fleet_") && !strings.Contains(p.Query, "gha_diagnostic_storage_") && !strings.Contains(p.Query, "otelcol_exporter_") { return fmt.Errorf("query does not use an owned fleet or Collector metric") } return nil diff --git a/internal/observabilitydashboards/dashboards_test.go b/internal/observabilitydashboards/dashboards_test.go index 5d0c135..43bd75a 100644 --- a/internal/observabilitydashboards/dashboards_test.go +++ b/internal/observabilitydashboards/dashboards_test.go @@ -21,7 +21,7 @@ func TestPublishedDashboardBundleIsValidAndRenderable(t *testing.T) { if err != nil { t.Fatal(err) } - if len(bundle.Dashboards) != 6 { + if len(bundle.Dashboards) != 7 { t.Fatalf("dashboards=%d", len(bundle.Dashboards)) } rendered, err := Render(bundle) diff --git a/internal/observabilitydashboards/openobserve.go b/internal/observabilitydashboards/openobserve.go index 09b9741..177d420 100644 --- a/internal/observabilitydashboards/openobserve.go +++ b/internal/observabilitydashboards/openobserve.go @@ -7,7 +7,7 @@ import ( const managedDescriptionPrefix = "managed-by:gds;dashboard-contract:v1;" -var metricPattern = regexp.MustCompile(`(?:gha_fleet_|otelcol_exporter_)[a-zA-Z0-9_:]*`) +var metricPattern = regexp.MustCompile(`(?:gha_fleet_|gha_diagnostic_storage_|otelcol_exporter_)[a-zA-Z0-9_:]*`) type OpenObserveDashboard struct { Version int `json:"version"` diff --git a/internal/observabilitydashboards/openobserve_test.go b/internal/observabilitydashboards/openobserve_test.go index 61ca32b..acb9d0e 100644 --- a/internal/observabilitydashboards/openobserve_test.go +++ b/internal/observabilitydashboards/openobserve_test.go @@ -14,7 +14,7 @@ func TestRenderOpenObserveV8IsDeterministicAndManaged(t *testing.T) { if err != nil { t.Fatal(err) } - if len(dashboards) != 6 { + if len(dashboards) != 7 { t.Fatalf("dashboards=%d", len(dashboards)) } for _, dashboard := range dashboards { @@ -29,7 +29,7 @@ func TestRenderOpenObserveV8IsDeterministicAndManaged(t *testing.T) { t.Fatalf("invalid panel %#v", panel) } stream := panel.Queries[0].Fields.Stream - if !strings.HasPrefix(stream, "gha_fleet_") && !strings.HasPrefix(stream, "otelcol_exporter_") { + if !strings.HasPrefix(stream, "gha_fleet_") && !strings.HasPrefix(stream, "gha_diagnostic_storage_") && !strings.HasPrefix(stream, "otelcol_exporter_") { t.Fatalf("panel %q stream is not an owned metric: %q", panel.ID, stream) } if panel.Layout.I != index+1 || panel.Layout.W != 96 || panel.Layout.H != 9 { diff --git a/internal/observabilityrules/rules_test.go b/internal/observabilityrules/rules_test.go index aac9a5d..c8a2296 100644 --- a/internal/observabilityrules/rules_test.go +++ b/internal/observabilityrules/rules_test.go @@ -10,8 +10,8 @@ func TestRepositoryBundleIsValid(t *testing.T) { if err != nil { t.Fatal(err) } - if len(bundle.Rules) != 14 { - t.Fatalf("rules = %d, want 14", len(bundle.Rules)) + if len(bundle.Rules) != 15 { + t.Fatalf("rules = %d, want 15", len(bundle.Rules)) } }