From e8f26ec4ef2e8a872778cb364b03b60a597871fe Mon Sep 17 00:00:00 2001 From: Juan Fernandez Date: Wed, 1 Jul 2026 14:00:59 -0400 Subject: [PATCH] Fix S3 file carve cleanup hang and rework reconciliation MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Relates to #48549 The S3 carve cleanup (server/datastore/s3, run by the cleanups_then_aggregation cron) advanced ListObjectsV2 pagination using the response's ContinuationToken — an echo of the request token — instead of NextContinuationToken. On any bucket with more than one page of objects this looped forever, hanging the entire serial cleanup cron and stalling every cleanup/aggregation job ordered after it. Replace the bucket-listing reconciliation with a direct HeadObject probe per carve, which is exact and independent of listing order or object counts: - Only carves older than 24h with a completed upload are reconciled (mirrors the MySQL carve store's floor; skips in-flight multipart uploads). A carve is expired only on a definitive not-found; transient or other probe errors leave it for a future run, so a carve whose object still exists is never expired. - Probes run with bounded concurrency; expirations are written in one batched, retryable UPDATE (new ExpireCarves datastore method) rather than one per carve. - The number of carves reconciled per run is capped so a large backlog drains across runs without any single run making unbounded S3 requests. Add S3-carve-store-only server settings (the MySQL carve store is unaffected): - s3.carves_cleanup_disabled — skip reconciliation entirely - s3.carves_cleanup_max_per_run — per-run cap (default 1000) - s3.carves_cleanup_concurrency — concurrent probes (default 32) Also log the expired count per run and fix the test bucket cleanup helper to paginate. Adds unit tests (transient-error safety, partial failure, concurrency) and a MySQL integration test for ExpireCarves. --- changes/fix-s3-carve-cleanup-hang | 3 + cmd/fleet/cron.go | 5 +- .../fleet-server-configuration.md | 45 +++ server/config/config.go | 53 ++-- server/datastore/mysql/carves.go | 19 ++ server/datastore/mysql/carves_test.go | 53 ++++ server/datastore/s3/carves.go | 229 +++++++++----- server/datastore/s3/carves_test.go | 297 +++++++++++++++++- server/datastore/s3/s3.go | 43 +-- server/datastore/s3/s3_test.go | 2 +- server/fleet/datastore.go | 3 + server/mock/datastore_mock.go | 12 + 12 files changed, 646 insertions(+), 118 deletions(-) create mode 100644 changes/fix-s3-carve-cleanup-hang diff --git a/changes/fix-s3-carve-cleanup-hang b/changes/fix-s3-carve-cleanup-hang new file mode 100644 index 0000000000..cfed8067d8 --- /dev/null +++ b/changes/fix-s3-carve-cleanup-hang @@ -0,0 +1,3 @@ +- Fixed an issue where cleanup of expired file carves stored in S3 could stall on buckets containing a large number of objects, which prevented other scheduled cleanup and aggregation tasks from running. +- Added the `s3.carves_cleanup_disabled` server setting to skip S3 file carve reconciliation for deployments that rely solely on the bucket's lifecycle policy to remove carve objects. +- Added the `s3.carves_cleanup_max_per_run` and `s3.carves_cleanup_concurrency` server settings to tune how many carves the S3 cleanup reconciles per run and how many concurrent S3 requests it makes. diff --git a/cmd/fleet/cron.go b/cmd/fleet/cron.go index 4261f2e7cd..1ef5eb246d 100644 --- a/cmd/fleet/cron.go +++ b/cmd/fleet/cron.go @@ -1315,7 +1315,10 @@ func newCleanupsAndAggregationSchedule( schedule.WithJob( "carves", func(ctx context.Context) error { - _, err := carveStore.CleanupCarves(ctx, time.Now()) + expired, err := carveStore.CleanupCarves(ctx, time.Now()) + if expired > 0 { + logger.InfoContext(ctx, "expired carves", "count", expired) + } return err }, ), diff --git a/docs/Configuration/fleet-server-configuration.md b/docs/Configuration/fleet-server-configuration.md index 43b074e7ca..ef218b0133 100644 --- a/docs/Configuration/fleet-server-configuration.md +++ b/docs/Configuration/fleet-server-configuration.md @@ -2975,6 +2975,51 @@ On GCE, GKE, or Cloud Run, ADC typically resolves to the runtime workload identi carves_force_s3_path_style: false ``` +### s3_carves_cleanup_disabled + +When `true`, the S3 carve store skips the periodic reconciliation that marks carves whose S3 +object no longer exists as expired. Set this if you rely solely on the bucket lifecycle policy +to remove carve objects and do not need the `expired` flag reconciled. This applies only to the +S3 carve store; it has no effect when carves are stored in MySQL. + +- Default value: false +- Environment variable: `FLEET_S3_CARVES_CLEANUP_DISABLED` +- Config file format: + ```yaml + s3: + carves_cleanup_disabled: true + ``` + +### s3_carves_cleanup_max_per_run + +The maximum number of carves the S3 cleanup reconciles per run, which also bounds +the number of S3 `HeadObject` requests a single run makes. A larger carve backlog +is drained across subsequent runs. Raise this to drain a large backlog faster, at +the cost of more work per run; lower it to reduce each run's impact on the shared +cleanup schedule. + +- Default value: 1000 +- Environment variable: `FLEET_S3_CARVES_CLEANUP_MAX_PER_RUN` +- Config file format: + ```yaml + s3: + carves_cleanup_max_per_run: 1000 + ``` + +### s3_carves_cleanup_concurrency + +The number of concurrent S3 `HeadObject` probes the carve cleanup performs. Kept +modest by default to stay well under S3's per-prefix request rate; lower it if you +observe throttling, or raise it to speed up a backlog's probe phase. + +- Default value: 32 +- Environment variable: `FLEET_S3_CARVES_CLEANUP_CONCURRENCY` +- Config file format: + ```yaml + s3: + carves_cleanup_concurrency: 32 + ``` + ### s3_carves_region > Same region-discovery behavior as [`s3_software_installers_region`](#s3_software_installers_region). Set this explicitly to avoid defaulting to `us-east-1`. diff --git a/server/config/config.go b/server/config/config.go index b63157b74b..7d7a6c6cf4 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -480,17 +480,20 @@ type S3Config struct { DisableSSL bool `yaml:"disable_ssl"` ForceS3PathStyle bool `yaml:"force_s3_path_style"` - CarvesBucket string `yaml:"carves_bucket"` - CarvesPrefix string `yaml:"carves_prefix"` - CarvesRegion string `yaml:"carves_region"` - CarvesEndpointURL string `yaml:"carves_endpoint_url"` - CarvesAccessKeyID string `yaml:"carves_access_key_id"` - CarvesSecretAccessKey string `yaml:"carves_secret_access_key"` - CarvesStsAssumeRoleArn string `yaml:"carves_sts_assume_role_arn"` - CarvesStsExternalID string `yaml:"carves_sts_external_id"` - CarvesDisableSSL bool `yaml:"carves_disable_ssl"` - CarvesForceS3PathStyle bool `yaml:"carves_force_s3_path_style"` - CarvesGCSIAMAuth bool `yaml:"carves_gcs_iam_auth"` + CarvesBucket string `yaml:"carves_bucket"` + CarvesPrefix string `yaml:"carves_prefix"` + CarvesRegion string `yaml:"carves_region"` + CarvesEndpointURL string `yaml:"carves_endpoint_url"` + CarvesAccessKeyID string `yaml:"carves_access_key_id"` + CarvesSecretAccessKey string `yaml:"carves_secret_access_key"` + CarvesStsAssumeRoleArn string `yaml:"carves_sts_assume_role_arn"` + CarvesStsExternalID string `yaml:"carves_sts_external_id"` + CarvesDisableSSL bool `yaml:"carves_disable_ssl"` + CarvesForceS3PathStyle bool `yaml:"carves_force_s3_path_style"` + CarvesGCSIAMAuth bool `yaml:"carves_gcs_iam_auth"` + CarvesCleanupDisabled bool `yaml:"carves_cleanup_disabled"` + CarvesCleanupMaxPerRun int `yaml:"carves_cleanup_max_per_run"` + CarvesCleanupConcurrency int `yaml:"carves_cleanup_concurrency"` SoftwareInstallersBucket string `yaml:"software_installers_bucket"` SoftwareInstallersPrefix string `yaml:"software_installers_prefix"` @@ -1617,6 +1620,9 @@ func (man Manager) addConfigs() { man.addConfigBool("s3.carves_disable_ssl", false, "Disable SSL (typically for local testing)") man.addConfigBool("s3.carves_force_s3_path_style", false, "Set this to true to force path-style addressing, i.e., `http://s3.amazonaws.com/BUCKET/KEY`") man.addConfigBool("s3.carves_gcs_iam_auth", false, "Use Google ADC bearer tokens for GCS endpoint authentication instead of S3 HMAC keys") + man.addConfigBool("s3.carves_cleanup_disabled", false, "Disable the periodic cleanup that marks carves whose S3 object no longer exists as expired") + man.addConfigInt("s3.carves_cleanup_max_per_run", 1000, "Maximum number of carves the S3 cleanup reconciles (and S3 HeadObject requests it makes) per run") + man.addConfigInt("s3.carves_cleanup_concurrency", 32, "Number of concurrent S3 HeadObject probes the carve cleanup performs") // S3 for software installers man.addConfigString("s3.software_installers_bucket", "", "Bucket where to store uploaded software installers") @@ -2152,17 +2158,20 @@ func (man Manager) LoadConfig() FleetConfig { func (man Manager) loadS3Config() S3Config { return S3Config{ - CarvesBucket: man.getConfigString("s3.carves_bucket"), - CarvesPrefix: man.getConfigString("s3.carves_prefix"), - CarvesRegion: man.getConfigString("s3.carves_region"), - CarvesEndpointURL: man.getConfigString("s3.carves_endpoint_url"), - CarvesAccessKeyID: man.getConfigString("s3.carves_access_key_id"), - CarvesSecretAccessKey: man.getConfigString("s3.carves_secret_access_key"), - CarvesStsAssumeRoleArn: man.getConfigString("s3.carves_sts_assume_role_arn"), - CarvesStsExternalID: man.getConfigString("s3.carves_sts_external_id"), - CarvesDisableSSL: man.getConfigBool("s3.carves_disable_ssl"), - CarvesForceS3PathStyle: man.getConfigBool("s3.carves_force_s3_path_style"), - CarvesGCSIAMAuth: man.getConfigBool("s3.carves_gcs_iam_auth"), + CarvesBucket: man.getConfigString("s3.carves_bucket"), + CarvesPrefix: man.getConfigString("s3.carves_prefix"), + CarvesRegion: man.getConfigString("s3.carves_region"), + CarvesEndpointURL: man.getConfigString("s3.carves_endpoint_url"), + CarvesAccessKeyID: man.getConfigString("s3.carves_access_key_id"), + CarvesSecretAccessKey: man.getConfigString("s3.carves_secret_access_key"), + CarvesStsAssumeRoleArn: man.getConfigString("s3.carves_sts_assume_role_arn"), + CarvesStsExternalID: man.getConfigString("s3.carves_sts_external_id"), + CarvesDisableSSL: man.getConfigBool("s3.carves_disable_ssl"), + CarvesForceS3PathStyle: man.getConfigBool("s3.carves_force_s3_path_style"), + CarvesGCSIAMAuth: man.getConfigBool("s3.carves_gcs_iam_auth"), + CarvesCleanupDisabled: man.getConfigBool("s3.carves_cleanup_disabled"), + CarvesCleanupMaxPerRun: man.getConfigInt("s3.carves_cleanup_max_per_run"), + CarvesCleanupConcurrency: man.getConfigInt("s3.carves_cleanup_concurrency"), Bucket: man.getConfigString("s3.bucket"), Prefix: man.getConfigString("s3.prefix"), diff --git a/server/datastore/mysql/carves.go b/server/datastore/mysql/carves.go index 2f512da2d6..30c0b98297 100644 --- a/server/datastore/mysql/carves.go +++ b/server/datastore/mysql/carves.go @@ -88,6 +88,25 @@ func (ds *Datastore) UpdateCarve(ctx context.Context, metadata *fleet.CarveMetad return updateCarveDB(ctx, ds.writer(ctx), metadata) } +// ExpireCarves marks the given carves as expired in batches. +func (ds *Datastore) ExpireCarves(ctx context.Context, ids []int64) error { + const batchSize = 500 + for start := 0; start < len(ids); start += batchSize { + end := min(start+batchSize, len(ids)) + stmt, args, err := sqlx.In(`UPDATE carve_metadata SET expired = 1 WHERE id IN (?)`, ids[start:end]) + if err != nil { + return ctxerr.Wrap(ctx, err, "build sqlx.In for expire carves") + } + if err := ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error { + _, err := tx.ExecContext(ctx, stmt, args...) + return err + }); err != nil { + return ctxerr.Wrap(ctx, err, "expire carves") + } + } + return nil +} + func updateCarveDB(ctx context.Context, exec sqlx.ExecerContext, metadata *fleet.CarveMetadata) error { stmt := ` UPDATE carve_metadata SET diff --git a/server/datastore/mysql/carves_test.go b/server/datastore/mysql/carves_test.go index 55187682d1..36868df028 100644 --- a/server/datastore/mysql/carves_test.go +++ b/server/datastore/mysql/carves_test.go @@ -26,6 +26,7 @@ func TestCarves(t *testing.T) { {"Cleanup", testCarvesCleanup}, {"List", testCarvesList}, {"Update", testCarvesUpdate}, + {"Expire", testCarvesExpire}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { @@ -306,3 +307,55 @@ func testCarvesUpdate(t *testing.T, ds *Datastore) { require.NoError(t, err) assert.Equal(t, carve, dbCarve) } + +func testCarvesExpire(t *testing.T, ds *Datastore) { + ctx := context.Background() + h := test.NewHost(t, ds, "foo.local", "192.168.1.10", "1", "1", time.Now()) + + newCarve := func(session string) *fleet.CarveMetadata { + c, err := ds.NewCarve(ctx, &fleet.CarveMetadata{ + HostId: h.ID, + Name: session, + BlockCount: 1, + BlockSize: 1, + CarveSize: 1, + CarveId: session, + RequestId: session, + SessionId: session, + CreatedAt: mockCreatedAt, + }) + require.NoError(t, err) + return c + } + c1 := newCarve("s1") + c2 := newCarve("s2") + c3 := newCarve("s3") + + // Empty ids is a no-op. + require.NoError(t, ds.ExpireCarves(ctx, nil)) + for _, c := range []*fleet.CarveMetadata{c1, c2, c3} { + got, err := ds.Carve(ctx, c.ID) + require.NoError(t, err) + require.False(t, got.Expired) + } + + // Expire c1 and c3, along with enough (nonexistent) ids to span more than one + // batch, exercising the chunked update loop. + ids := []int64{c1.ID, c3.ID} + for i := range int64(600) { + ids = append(ids, 1_000_000+i) + } + require.NoError(t, ds.ExpireCarves(ctx, ids)) + + got1, err := ds.Carve(ctx, c1.ID) + require.NoError(t, err) + require.True(t, got1.Expired, "carve in the id list must be expired") + + got2, err := ds.Carve(ctx, c2.ID) + require.NoError(t, err) + require.False(t, got2.Expired, "carve not in the id list must stay non-expired") + + got3, err := ds.Carve(ctx, c3.ID) + require.NoError(t, err) + require.True(t, got3.Expired, "carve in the id list must be expired") +} diff --git a/server/datastore/s3/carves.go b/server/datastore/s3/carves.go index 1bbfe1a6b6..4be337563d 100644 --- a/server/datastore/s3/carves.go +++ b/server/datastore/s3/carves.go @@ -7,7 +7,7 @@ import ( "fmt" "io" "strconv" - "strings" + "sync" "time" "github.com/aws/aws-sdk-go-v2/service/s3" @@ -19,19 +19,45 @@ import ( ) const ( - defaultMaxS3Keys = 1000 - cleanupSize = 1000 + // defaultCarvesCleanupMaxPerRun bounds how many carves a single cleanup run + // examines (and therefore the number of S3 HeadObject requests it makes) when + // the s3.carves_cleanup_max_per_run config is unset. A larger backlog drains + // across subsequent runs. + defaultCarvesCleanupMaxPerRun = 1000 + // defaultCarvesCleanupConcurrency bounds how many HeadObject probes run at once + // when the s3.carves_cleanup_concurrency config is unset. Kept modest to stay + // well under S3's per-prefix request rate and avoid throttling. + defaultCarvesCleanupConcurrency = 32 + // carveHeadObjectTimeout bounds each existence probe so a hung request cannot + // stall the whole cleanup run (which would hold up the shared cleanup cron). + carveHeadObjectTimeout = 30 * time.Second // This is Golang's way of formatting timestrings, it's confusing, I know. // If you are used to more conventional timestrings, this is equivalent // to %Y/%m/%d/%H (year/month/day/hour) timePrefixFormat = "2006/01/02/15" ) +// s3HeadObjectAPI is the subset of the S3 client used to probe object existence. +// It is an interface so cleanup can be unit-tested without a live S3 backend. +type s3HeadObjectAPI interface { + HeadObject(ctx context.Context, params *s3.HeadObjectInput, optFns ...func(*s3.Options)) (*s3.HeadObjectOutput, error) +} + // CarveStore is a type implementing the CarveStore interface // relying on AWS S3 storage type CarveStore struct { *s3store metadatadb fleet.CarveStore + // headObjectAPI probes object existence during cleanup; defaults to the S3 + // client and is overridable in tests. + headObjectAPI s3HeadObjectAPI + // cleanupDisabled, when true, makes CleanupCarves a no-op so operators can + // rely solely on the bucket lifecycle policy and skip S3 reconciliation. + cleanupDisabled bool + // maxPerRun and probeConcurrency tune CleanupCarves; when <= 0 the + // defaultCarvesCleanup* constants are used. + maxPerRun int + probeConcurrency int } // NewCarveStore creates a new store with the given config @@ -42,8 +68,12 @@ func NewCarveStore(config config.S3Config, metadatadb fleet.CarveStore) (*CarveS } return &CarveStore{ - s3store: s3store, - metadatadb: metadatadb, + s3store: s3store, + metadatadb: metadatadb, + headObjectAPI: s3store.s3Client, + cleanupDisabled: config.CarvesCleanupDisabled, + maxPerRun: config.CarvesCleanupMaxPerRun, + probeConcurrency: config.CarvesCleanupConcurrency, }, nil } @@ -94,86 +124,139 @@ func (c *CarveStore) UpdateCarve(ctx context.Context, metadata *fleet.CarveMetad return c.metadatadb.UpdateCarve(ctx, metadata) } -// listS3Carves lists all keys up to a given one or if the passed max number -// of keys has been reached; keys are returned in a set-like map -func (c *CarveStore) listS3Carves(ctx context.Context, lastPrefix string, maxKeys int) (map[string]bool, error) { - var err error - var continuationToken string - result := make(map[string]bool) - if maxKeys <= 0 { - maxKeys = defaultMaxS3Keys - } - if !strings.HasPrefix(lastPrefix, c.prefix) { - lastPrefix = c.prefix + lastPrefix - } - for { - carveFilesPage, err := c.s3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ - Bucket: &c.bucket, - Prefix: &c.prefix, - ContinuationToken: &continuationToken, - }) - if err != nil { - return nil, err - } - for _, carveObject := range carveFilesPage.Contents { - result[*carveObject.Key] = true - if strings.HasPrefix(*carveObject.Key, lastPrefix) || len(result) >= maxKeys { - return result, nil - } - } - if !*carveFilesPage.IsTruncated { - break - } - continuationToken = *carveFilesPage.ContinuationToken - } - return result, err +// ExpireCarves marks the given carves as expired via the metadata store. +func (c *CarveStore) ExpireCarves(ctx context.Context, ids []int64) error { + return c.metadatadb.ExpireCarves(ctx, ids) } -// CleanupCarves is a noop on the S3 side since users should rely on the bucket -// lifecycle configurations provided by AWS. This will compare a portion of the -// metadata present in the database and mark as expired the carves no longer -// available in S3 (ignores the `now` argument) +// carveObjectExists reports whether the carve's object is present in S3. A missing +// object (NoSuchKey/NotFound) is reported as (false, nil); any other error — +// including a missing bucket — is returned so the caller does not treat a +// transient or configuration failure as a deleted object. +func (c *CarveStore) carveObjectExists(ctx context.Context, key string) (bool, error) { + _, err := c.headObjectAPI.HeadObject(ctx, &s3.HeadObjectInput{ + Bucket: &c.bucket, + Key: &key, + }) + if err != nil { + // AWS S3 signals a missing object on HeadObject with NotFound; NoSuchKey is + // kept as a defensive fallback for S3-compatible backends (e.g. GCS) that may + // surface it instead. Any other error (including NoSuchBucket, throttling, or + // a network failure) is returned so the caller does not treat a transient or + // configuration failure as a deleted object. + if _, ok := errors.AsType[*types.NotFound](err); ok { + return false, nil + } + if _, ok := errors.AsType[*types.NoSuchKey](err); ok { + return false, nil + } + return false, ctxerr.Wrapf(ctx, err, "checking existence of carve %s in S3", key) + } + return true, nil +} + +// CleanupCarves marks carves whose S3 object no longer exists as expired. +// Deletion of the objects themselves is delegated to the bucket lifecycle policy; +// this only reconciles the DB `expired` flag against S3. Carves created within the +// last 24h are not reconciled, since S3 lifecycle expiration has day granularity +// and cannot have removed them yet. Reconciliation can be disabled entirely via +// the s3.carves_cleanup_disabled config. +// +// Each candidate's object is probed directly with HeadObject (rather than listing +// the bucket), which is exact and independent of listing order or object counts. +// Probes run with bounded concurrency (s3.carves_cleanup_concurrency) and are +// capped per run (s3.carves_cleanup_max_per_run); expirations are then written in +// a single batched statement. func (c *CarveStore) CleanupCarves(ctx context.Context, now time.Time) (int, error) { - // Get the 1000 oldest carves + if c.cleanupDisabled { + return 0, nil + } + maxPerRun := c.maxPerRun + if maxPerRun <= 0 { + maxPerRun = defaultCarvesCleanupMaxPerRun + } + concurrency := c.probeConcurrency + if concurrency <= 0 { + concurrency = defaultCarvesCleanupConcurrency + } + // Oldest-first and capped so a backlog drains deterministically across runs + // without any single run making an unbounded number of S3 requests. nonExpiredCarves, err := c.ListCarves(ctx, fleet.CarveListOptions{ - ListOptions: fleet.ListOptions{PerPage: cleanupSize}, - Expired: false, + ListOptions: fleet.ListOptions{ + PerPage: uint(maxPerRun), //nolint:gosec // bounded small positive config value + OrderKey: "created_at", + OrderDirection: fleet.OrderAscending, + }, + Expired: false, }) if err != nil { return 0, ctxerr.Wrap(ctx, err, "s3 carve cleanup") } - if len(nonExpiredCarves) == 0 { + + cutoff := now.Add(-24 * time.Hour) + var candidates []*fleet.CarveMetadata + for _, carve := range nonExpiredCarves { + // Skip carves too new to have been lifecycle-deleted, and carves whose + // multipart upload hasn't completed: their object isn't in S3 yet, so + // expiring one would make it permanently undownloadable. + if carve.CreatedAt.Before(cutoff) && carve.BlocksComplete() { + candidates = append(candidates, carve) + } + } + if len(candidates) == 0 { return 0, nil } - // List carves in S3 up to a hour+1 prefix - lastCarveNextHour := nonExpiredCarves[len(nonExpiredCarves)-1].CreatedAt.Add(time.Hour) - lastCarvePrefix := c.prefix + lastCarveNextHour.Format(timePrefixFormat) - carveKeys, err := c.listS3Carves(ctx, lastCarvePrefix, 2*cleanupSize) - if err != nil { - return 0, ctxerr.Wrap(ctx, err, "s3 carve cleanup") - } - // Compare carve metadata in DB with S3 listing and update expiration flag - cleanCount := 0 - var retErr error - for _, carve := range nonExpiredCarves { - // A carve whose multipart upload has not completed yet has no listable - // object in S3 (ListObjectsV2 does not return in-progress multipart - // uploads), so skip it to avoid expiring a carve that is still - // uploading. Such a carve would otherwise become permanently - // undownloadable. - if !carve.BlocksComplete() { - continue - } - if _, ok := carveKeys[c.generateS3Key(carve)]; !ok { - carve.Expired = true - if uerr := c.UpdateCarve(ctx, carve); uerr != nil { - retErr = errors.Join(retErr, ctxerr.Wrap(ctx, uerr, fmt.Sprintf("marking carve %d expired", carve.ID))) - continue + + // Probe each candidate's object with HeadObject, with bounded concurrency since + // each is a small, latency-bound request. Only S3 is touched here; the DB write + // below is a single batched statement, so there is no concurrent DB access. + var ( + mu sync.Mutex + toExpire []*fleet.CarveMetadata + probeErr error + ) + sem := make(chan struct{}, concurrency) + var wg sync.WaitGroup + for _, carve := range candidates { + wg.Add(1) + sem <- struct{}{} + go func(carve *fleet.CarveMetadata) { + defer wg.Done() + defer func() { <-sem }() + // Bound each probe so a hung request cannot stall wg.Wait indefinitely. + probeCtx, cancel := context.WithTimeout(ctx, carveHeadObjectTimeout) + defer cancel() + exists, err := c.carveObjectExists(probeCtx, c.generateS3Key(carve)) + mu.Lock() + defer mu.Unlock() + switch { + case err != nil: + // Only expire on a definitive not-found; treat anything else (e.g. a + // throttled, timed-out, or failed request) as transient and retry on a + // later run. + probeErr = errors.Join(probeErr, err) + case !exists: + toExpire = append(toExpire, carve) } - cleanCount++ - } + }(carve) } - return cleanCount, retErr + wg.Wait() + + if len(toExpire) == 0 { + return 0, probeErr + } + ids := make([]int64, len(toExpire)) + for i, carve := range toExpire { + ids[i] = carve.ID + } + if err := c.ExpireCarves(ctx, ids); err != nil { + return 0, errors.Join(probeErr, ctxerr.Wrap(ctx, err, "s3 carve cleanup")) + } + // Reflect expiry on the returned metadata only after the durable write succeeds. + for _, carve := range toExpire { + carve.Expired = true + } + return len(ids), probeErr } // Carve returns carve metadata by ID diff --git a/server/datastore/s3/carves_test.go b/server/datastore/s3/carves_test.go index 12373f5509..fccb509dcf 100644 --- a/server/datastore/s3/carves_test.go +++ b/server/datastore/s3/carves_test.go @@ -2,11 +2,15 @@ package s3 import ( "context" + "errors" + "fmt" + "sort" "strings" "testing" "time" awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/aws/aws-sdk-go-v2/service/s3/types" "github.com/fleetdm/fleet/v4/server/config" "github.com/fleetdm/fleet/v4/server/fleet" "github.com/stretchr/testify/require" @@ -16,12 +20,26 @@ import ( // state. It is used by tests that do not require a real S3 backend. type stubCarveMetadataStore struct { carves []*fleet.CarveMetadata + // lastListOpts records the options of the most recent ListCarves call so + // tests can assert how CleanupCarves queries the metadata store. + lastListOpts fleet.CarveListOptions + // listCarvesCalled records whether ListCarves was invoked. + listCarvesCalled bool + // expiredIDs accumulates the ids passed to ExpireCarves. + expiredIDs []int64 + // updateCarveCalled records whether the per-carve UpdateCarve was invoked. + updateCarveCalled bool } func (s *stubCarveMetadataStore) NewCarve(_ context.Context, metadata *fleet.CarveMetadata) (*fleet.CarveMetadata, error) { return metadata, nil } func (s *stubCarveMetadataStore) UpdateCarve(_ context.Context, _ *fleet.CarveMetadata) error { + s.updateCarveCalled = true + return nil +} +func (s *stubCarveMetadataStore) ExpireCarves(_ context.Context, ids []int64) error { + s.expiredIDs = append(s.expiredIDs, ids...) return nil } func (s *stubCarveMetadataStore) Carve(_ context.Context, _ int64) (*fleet.CarveMetadata, error) { @@ -33,8 +51,22 @@ func (s *stubCarveMetadataStore) CarveBySessionId(_ context.Context, _ string) ( func (s *stubCarveMetadataStore) CarveByName(_ context.Context, _ string) (*fleet.CarveMetadata, error) { return nil, nil } -func (s *stubCarveMetadataStore) ListCarves(_ context.Context, _ fleet.CarveListOptions) ([]*fleet.CarveMetadata, error) { - return s.carves, nil +func (s *stubCarveMetadataStore) ListCarves(_ context.Context, opt fleet.CarveListOptions) ([]*fleet.CarveMetadata, error) { + s.listCarvesCalled = true + s.lastListOpts = opt + // Mirror the real datastore: honor ordering by created_at so the S3 listing + // window CleanupCarves derives from the first/last elements is meaningful. + carves := append([]*fleet.CarveMetadata(nil), s.carves...) + if opt.OrderKey == "created_at" { + sort.SliceStable(carves, func(i, j int) bool { + less := carves[i].CreatedAt.Before(carves[j].CreatedAt) + if opt.OrderDirection == fleet.OrderDescending { + return !less + } + return less + }) + } + return carves, nil } func (s *stubCarveMetadataStore) NewBlock(_ context.Context, _ *fleet.CarveMetadata, _ int64, _ []byte) error { return nil @@ -162,3 +194,264 @@ func TestCleanupCarvesSkipsInFlightCarves(t *testing.T) { require.True(t, completed.Expired, "completed carve absent from S3 must be marked expired") require.False(t, inFlight.Expired, "in-flight carve must not be marked expired") } + +// TestCleanupCarvesQueriesOldestFirst verifies that CleanupCarves lists carves +// ordered by created_at ascending and capped, so a backlog drains oldest-first +// across runs without any single run making an unbounded number of requests. +func TestCleanupCarvesQueriesOldestFirst(t *testing.T) { + // No carves: CleanupCarves returns before touching S3, so no backend needed. + stub := &stubCarveMetadataStore{carves: nil} + store := &CarveStore{metadatadb: stub} + + _, err := store.CleanupCarves(t.Context(), time.Now()) + require.NoError(t, err) + + require.Equal(t, "created_at", stub.lastListOpts.OrderKey, "cleanup must order carves by created_at") + require.Equal(t, fleet.OrderAscending, stub.lastListOpts.OrderDirection, "cleanup must order carves ascending (oldest first)") + require.False(t, stub.lastListOpts.Expired, "cleanup must only consider non-expired carves") +} + +// TestCleanupCarvesExpiresAbsentAndBatchesWrites verifies that CleanupCarves +// probes each carve's object directly: carves whose object is present are kept, +// carves whose object is absent are expired, and the expirations are written in a +// single batched ExpireCarves call rather than one UpdateCarve per carve. +// +// Requires a running S3-compatible endpoint (set S3_STORAGE_TEST env var). +func TestCleanupCarvesExpiresAbsentAndBatchesWrites(t *testing.T) { + checkTestEnv(t) + ctx := t.Context() + + const bucket = "carves-batch-test" + const prefix = "carvetest/" + + base := time.Date(2024, 6, 1, 10, 0, 0, 0, time.UTC) + present := &fleet.CarveMetadata{ID: 1, Name: "present", CreatedAt: base, BlockCount: 1, MaxBlock: 0} + absentA := &fleet.CarveMetadata{ID: 2, Name: "absent-a", CreatedAt: base.Add(time.Minute), BlockCount: 1, MaxBlock: 0} + absentB := &fleet.CarveMetadata{ID: 3, Name: "absent-b", CreatedAt: base.Add(2 * time.Minute), BlockCount: 1, MaxBlock: 0} + stub := &stubCarveMetadataStore{carves: []*fleet.CarveMetadata{present, absentA, absentB}} + + store, err := NewCarveStore(config.S3Config{ + CarvesBucket: bucket, + CarvesPrefix: prefix, + CarvesRegion: "localhost", + CarvesEndpointURL: testEndpoint, + CarvesAccessKeyID: testAccessKeyID, + CarvesSecretAccessKey: testSecretAccessKey, + CarvesForceS3PathStyle: true, + CarvesDisableSSL: true, + }, stub) + require.NoError(t, err) + + require.NoError(t, store.CreateTestBucket(ctx, bucket)) + t.Cleanup(func() { + if err := store.CleanupTestBucket(context.Background()); err != nil { + t.Errorf("cleanup s3 bucket %q: %v", bucket, err) + } + }) + + // Only "present" has an object in S3; the other two are absent. + key := store.generateS3Key(present) + _, err = store.s3Client.PutObject(ctx, &awss3.PutObjectInput{ + Bucket: &store.bucket, + Key: &key, + Body: strings.NewReader("x"), + }) + require.NoError(t, err) + + cleaned, err := store.CleanupCarves(ctx, time.Now()) + require.NoError(t, err) + require.Equal(t, 2, cleaned, "both carves absent from S3 should be expired") + require.False(t, present.Expired, "carve with a present object must not be expired") + require.True(t, absentA.Expired) + require.True(t, absentB.Expired) + + require.ElementsMatch(t, []int64{absentA.ID, absentB.ID}, stub.expiredIDs, "expirations must be batched via ExpireCarves") + require.False(t, stub.updateCarveCalled, "cleanup must not fall back to per-carve UpdateCarve") +} + +// TestCleanupCarvesSkipsCarvesYoungerThan24h verifies that carves created within +// the last 24h are not reconciled against S3 (and so never expired), even when +// their object is absent. S3 lifecycle expiration has day granularity and never +// deletes objects created that recently, so checking them is wasted work and risks +// wrongly expiring a carve whose object simply isn't listable yet. Mirrors the 24h +// floor the MySQL-backed carve store already applies. +// +// Requires a running S3-compatible endpoint (set S3_STORAGE_TEST env var). +func TestCleanupCarvesSkipsCarvesYoungerThan24h(t *testing.T) { + checkTestEnv(t) + ctx := t.Context() + now := time.Now() + + const bucket = "carves-age-floor-test" + const prefix = "carvetest/" + + // Neither carve has an object in S3 (empty bucket). Only the old one is old + // enough to be reconciled; the recent one must be left untouched. + old := &fleet.CarveMetadata{ID: 1, Name: "old-gone", CreatedAt: now.Add(-48 * time.Hour), BlockCount: 1, MaxBlock: 0} + recent := &fleet.CarveMetadata{ID: 2, Name: "recent-gone", CreatedAt: now.Add(-1 * time.Hour), BlockCount: 1, MaxBlock: 0} + stub := &stubCarveMetadataStore{carves: []*fleet.CarveMetadata{old, recent}} + + store, err := NewCarveStore(config.S3Config{ + CarvesBucket: bucket, + CarvesPrefix: prefix, + CarvesRegion: "localhost", + CarvesEndpointURL: testEndpoint, + CarvesAccessKeyID: testAccessKeyID, + CarvesSecretAccessKey: testSecretAccessKey, + CarvesForceS3PathStyle: true, + CarvesDisableSSL: true, + }, stub) + require.NoError(t, err) + + require.NoError(t, store.CreateTestBucket(ctx, bucket)) + t.Cleanup(func() { + if err := store.CleanupTestBucket(context.Background()); err != nil { + t.Errorf("cleanup s3 bucket %q: %v", bucket, err) + } + }) + + cleaned, err := store.CleanupCarves(ctx, now) + require.NoError(t, err) + require.Equal(t, 1, cleaned, "only the >24h carve should be reconciled/expired") + require.True(t, old.Expired, "carve older than 24h with no S3 object must be expired") + require.False(t, recent.Expired, "carve younger than 24h must not be reconciled or expired") +} + +// TestCleanupCarvesDisabled verifies that when the S3 carve store is configured +// with cleanup disabled, CleanupCarves is a no-op: it neither queries the metadata +// store nor expires any carve. This must apply to the S3 store only (the MySQL +// carve store is unaffected because it does not carry this flag). +func TestCleanupCarvesDisabled(t *testing.T) { + // An old carve with no S3 object would normally be expired; with cleanup + // disabled it must be left alone. No S3 backend is needed because the store + // returns before doing any work. + old := &fleet.CarveMetadata{ID: 1, Name: "old-gone", CreatedAt: time.Date(2024, 1, 15, 10, 0, 0, 0, time.UTC), BlockCount: 1, MaxBlock: 0} + stub := &stubCarveMetadataStore{carves: []*fleet.CarveMetadata{old}} + store := &CarveStore{metadatadb: stub, cleanupDisabled: true} + + cleaned, err := store.CleanupCarves(t.Context(), time.Now()) + require.NoError(t, err) + require.Equal(t, 0, cleaned, "cleanup must be a no-op when disabled") + require.False(t, stub.listCarvesCalled, "cleanup must not query the metadata store when disabled") + require.False(t, old.Expired, "no carve may be expired when cleanup is disabled") +} + +// fakeHeadObjectAPI lets CleanupCarves be unit-tested without a live S3 backend. +type fakeHeadObjectAPI struct { + fn func(key string) (*awss3.HeadObjectOutput, error) +} + +func (f fakeHeadObjectAPI) HeadObject(_ context.Context, in *awss3.HeadObjectInput, _ ...func(*awss3.Options)) (*awss3.HeadObjectOutput, error) { + return f.fn(*in.Key) +} + +// oldCompletedCarve returns a carve old enough to pass the 24h floor with a +// completed upload, so cleanup will probe it. +func oldCompletedCarve(id int64, name string) *fleet.CarveMetadata { + return &fleet.CarveMetadata{ + ID: id, + Name: name, + CreatedAt: time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC), + BlockCount: 1, + MaxBlock: 0, + } +} + +// TestCleanupCarvesDoesNotExpireOnTransientProbeError verifies the critical safety +// property: a probe error that is NOT a definitive not-found (e.g. throttling, a +// 5xx, or a network failure) must never expire a carve, since its object may well +// still exist. +func TestCleanupCarvesDoesNotExpireOnTransientProbeError(t *testing.T) { + carve := oldCompletedCarve(1, "maybe-gone") + stub := &stubCarveMetadataStore{carves: []*fleet.CarveMetadata{carve}} + store := &CarveStore{ + s3store: &s3store{prefix: "carvetest/", bucket: "test-bucket"}, + metadatadb: stub, + headObjectAPI: fakeHeadObjectAPI{fn: func(string) (*awss3.HeadObjectOutput, error) { + return nil, errors.New("throttled: SlowDown") + }}, + } + + cleaned, err := store.CleanupCarves(t.Context(), time.Now()) + require.Error(t, err, "a non-not-found probe error must surface") + require.Equal(t, 0, cleaned) + require.False(t, carve.Expired, "a carve must not be expired on a transient probe error") + require.Empty(t, stub.expiredIDs, "no ids should be batched for expiry") +} + +// TestCleanupCarvesPartialFailureExpiresOnlyConfirmedAbsent verifies that within a +// single run, a confirmed-absent carve is still expired even when another carve's +// probe fails transiently, and the run surfaces the error. +func TestCleanupCarvesPartialFailureExpiresOnlyConfirmedAbsent(t *testing.T) { + present := oldCompletedCarve(1, "present") + absent := oldCompletedCarve(2, "absent") + flaky := oldCompletedCarve(3, "flaky") + stub := &stubCarveMetadataStore{carves: []*fleet.CarveMetadata{present, absent, flaky}} + store := &CarveStore{ + s3store: &s3store{prefix: "carvetest/", bucket: "test-bucket"}, + metadatadb: stub, + headObjectAPI: fakeHeadObjectAPI{fn: func(key string) (*awss3.HeadObjectOutput, error) { + switch { + case strings.Contains(key, "present"): + return &awss3.HeadObjectOutput{}, nil + case strings.Contains(key, "absent"): + return nil, &types.NotFound{} + default: // flaky + return nil, errors.New("throttled") + } + }}, + } + + cleaned, err := store.CleanupCarves(t.Context(), time.Now()) + require.Error(t, err, "the transient failure must surface") + require.Equal(t, 1, cleaned, "only the confirmed-absent carve is expired") + require.False(t, present.Expired) + require.True(t, absent.Expired) + require.False(t, flaky.Expired, "a carve with a transient probe error must not be expired") + require.Equal(t, []int64{absent.ID}, stub.expiredIDs, "only the absent carve is batched for expiry") + require.False(t, stub.updateCarveCalled, "cleanup must not fall back to per-carve UpdateCarve") +} + +// TestCleanupCarvesConcurrentProbesAllAbsent exercises the bounded-concurrency +// probe fan-out (more candidates than the concurrency limit) and asserts no lost +// updates when collecting results. Run with -race to catch data races. +func TestCleanupCarvesConcurrentProbesAllAbsent(t *testing.T) { + const n = 100 + carves := make([]*fleet.CarveMetadata, n) + for i := range carves { + carves[i] = oldCompletedCarve(int64(i+1), fmt.Sprintf("gone-%03d", i)) + } + stub := &stubCarveMetadataStore{carves: carves} + store := &CarveStore{ + s3store: &s3store{prefix: "carvetest/", bucket: "test-bucket"}, + metadatadb: stub, + headObjectAPI: fakeHeadObjectAPI{fn: func(string) (*awss3.HeadObjectOutput, error) { + return nil, &types.NotFound{} + }}, + } + + cleaned, err := store.CleanupCarves(t.Context(), time.Now()) + require.NoError(t, err) + require.Equal(t, n, cleaned) + require.Len(t, stub.expiredIDs, n, "every absent carve must be batched with no lost updates under concurrency") + for _, c := range carves { + require.True(t, c.Expired) + } +} + +// TestCleanupCarvesRespectsConfiguredMaxPerRun verifies the per-run cap is taken +// from the store's configured value (surfaced as the ListCarves page size), and +// falls back to the default when unset. +func TestCleanupCarvesRespectsConfiguredMaxPerRun(t *testing.T) { + stub := &stubCarveMetadataStore{carves: nil} + store := &CarveStore{metadatadb: stub, maxPerRun: 7} + _, err := store.CleanupCarves(t.Context(), time.Now()) + require.NoError(t, err) + require.Equal(t, uint(7), stub.lastListOpts.PerPage, "configured max-per-run should set the ListCarves page size") + + stubDefault := &stubCarveMetadataStore{carves: nil} + storeDefault := &CarveStore{metadatadb: stubDefault} + _, err = storeDefault.CleanupCarves(t.Context(), time.Now()) + require.NoError(t, err) + require.Equal(t, uint(defaultCarvesCleanupMaxPerRun), stubDefault.lastListOpts.PerPage, "unset max-per-run should fall back to the default") +} diff --git a/server/datastore/s3/s3.go b/server/datastore/s3/s3.go index a8c6eee9ed..2d0b6e3725 100644 --- a/server/datastore/s3/s3.go +++ b/server/datastore/s3/s3.go @@ -237,31 +237,36 @@ func (s *s3store) CreateTestBucket(ctx context.Context, name string) error { // store. Only recommended for local testing. If the bucket no longer exists, // it returns nil. func (s *s3store) CleanupTestBucket(ctx context.Context) error { - resp, err := s.s3Client.ListObjects(ctx, &s3.ListObjectsInput{ + // Delete every object page-by-page (the SDK paginator handles continuation + // tokens) so buckets with more than one page of objects are fully emptied + // before DeleteBucket. + paginator := s3.NewListObjectsV2Paginator(s.s3Client, &s3.ListObjectsV2Input{ Bucket: &s.bucket, }) - var noSuchBucket *types.NoSuchBucket - if errors.As(err, &noSuchBucket) { - return nil - } - if err != nil { - return err - } - - var objs []types.ObjectIdentifier - for _, o := range resp.Contents { - objs = append(objs, types.ObjectIdentifier{Key: o.Key}) - } - if len(objs) > 0 { - if _, err := s.s3Client.DeleteObjects(ctx, &s3.DeleteObjectsInput{ - Bucket: &s.bucket, - Delete: &types.Delete{Objects: objs}, - }); err != nil { + for paginator.HasMorePages() { + resp, err := paginator.NextPage(ctx) + if _, ok := errors.AsType[*types.NoSuchBucket](err); ok { + return nil + } + if err != nil { return err } + + var objs []types.ObjectIdentifier + for _, o := range resp.Contents { + objs = append(objs, types.ObjectIdentifier{Key: o.Key}) + } + if len(objs) > 0 { + if _, err := s.s3Client.DeleteObjects(ctx, &s3.DeleteObjectsInput{ + Bucket: &s.bucket, + Delete: &types.Delete{Objects: objs}, + }); err != nil { + return err + } + } } - _, err = s.s3Client.DeleteBucket(ctx, &s3.DeleteBucketInput{ + _, err := s.s3Client.DeleteBucket(ctx, &s3.DeleteBucketInput{ Bucket: &s.bucket, }) return err diff --git a/server/datastore/s3/s3_test.go b/server/datastore/s3/s3_test.go index 5913cd1607..92a253dda2 100644 --- a/server/datastore/s3/s3_test.go +++ b/server/datastore/s3/s3_test.go @@ -215,7 +215,7 @@ func TestCarveStoreGCSIAMAuthUsesBearerToken(t *testing.T) { }, nil) require.NoError(t, err) - _, err = store.listS3Carves(context.Background(), "", 10) + _, err = store.carveObjectExists(context.Background(), "carves-prefix/some-key") require.NoError(t, err) select { diff --git a/server/fleet/datastore.go b/server/fleet/datastore.go index 0295833e1d..55d1f35bb2 100644 --- a/server/fleet/datastore.go +++ b/server/fleet/datastore.go @@ -27,6 +27,9 @@ import ( type CarveStore interface { NewCarve(ctx context.Context, metadata *CarveMetadata) (*CarveMetadata, error) UpdateCarve(ctx context.Context, metadata *CarveMetadata) error + // ExpireCarves marks the given carves as expired in a single batched + // operation. It is a no-op when ids is empty. + ExpireCarves(ctx context.Context, ids []int64) error Carve(ctx context.Context, carveId int64) (*CarveMetadata, error) CarveBySessionId(ctx context.Context, sessionId string) (*CarveMetadata, error) CarveByName(ctx context.Context, name string) (*CarveMetadata, error) diff --git a/server/mock/datastore_mock.go b/server/mock/datastore_mock.go index faac330823..c50a2baedb 100644 --- a/server/mock/datastore_mock.go +++ b/server/mock/datastore_mock.go @@ -52,6 +52,8 @@ type NewCarveFunc func(ctx context.Context, metadata *fleet.CarveMetadata) (*fle type UpdateCarveFunc func(ctx context.Context, metadata *fleet.CarveMetadata) error +type ExpireCarvesFunc func(ctx context.Context, ids []int64) error + type CarveFunc func(ctx context.Context, carveId int64) (*fleet.CarveMetadata, error) type CarveBySessionIdFunc func(ctx context.Context, sessionId string) (*fleet.CarveMetadata, error) @@ -2181,6 +2183,9 @@ type DataStore struct { UpdateCarveFunc UpdateCarveFunc UpdateCarveFuncInvoked bool + ExpireCarvesFunc ExpireCarvesFunc + ExpireCarvesFuncInvoked bool + CarveFunc CarveFunc CarveFuncInvoked bool @@ -5411,6 +5416,13 @@ func (s *DataStore) UpdateCarve(ctx context.Context, metadata *fleet.CarveMetada return s.UpdateCarveFunc(ctx, metadata) } +func (s *DataStore) ExpireCarves(ctx context.Context, ids []int64) error { + s.mu.Lock() + s.ExpireCarvesFuncInvoked = true + s.mu.Unlock() + return s.ExpireCarvesFunc(ctx, ids) +} + func (s *DataStore) Carve(ctx context.Context, carveId int64) (*fleet.CarveMetadata, error) { s.mu.Lock() s.CarveFuncInvoked = true