diff --git a/changes/48719-49805-software-ingestion-perf b/changes/48719-49805-software-ingestion-perf new file mode 100644 index 0000000000..f1f729a669 --- /dev/null +++ b/changes/48719-49805-software-ingestion-perf @@ -0,0 +1 @@ +- Improved software ingestion performance at scale: batched `host_software_installed_paths` deletes (previously unbounded single statements) and eliminated `software_titles` INSERT lock convoys during concurrent ingestion by moving title inserts outside the main transaction with singleflight deduplication. diff --git a/server/datastore/mysql/mysql.go b/server/datastore/mysql/mysql.go index 6e20b2dba0..45516c5400 100644 --- a/server/datastore/mysql/mysql.go +++ b/server/datastore/mysql/mysql.go @@ -37,6 +37,7 @@ import ( "github.com/jmoiron/sqlx" "go.opentelemetry.io/otel/attribute" semconv "go.opentelemetry.io/otel/semconv/v1.40.0" + "golang.org/x/sync/singleflight" ) const ( @@ -98,8 +99,31 @@ type Datastore struct { // This key is used to encrypt sensitive data stored in the Fleet DB, for example MDM // certificates and keys. serverPrivateKey string + + // knownSoftwareTitleKeys caches title keys that are known to exist in software_titles. + // This eliminates redundant INSERT IGNORE statements during concurrent software ingestion, + // preventing lock convoys on the unique index when many hosts report the same software catalog. + // The cache evicts an arbitrary half of entries once it reaches a fixed size cap to avoid + // unbounded growth on long-lived servers without forcing a full cold start. + knownSoftwareTitleKeys map[string]struct{} + // knownSoftwareTitleKeysMu serializes cache writes and clears; reads use RLock. + knownSoftwareTitleKeysMu sync.RWMutex + + // titleInsertSF deduplicates concurrent INSERT IGNORE INTO software_titles calls for the + // same title key. Only one goroutine per title actually executes the INSERT; others wait + // and share the result. This prevents lock convoys on cold-start (#48719). + titleInsertSF singleflight.Group } +// maxKnownSoftwareTitleKeys caps the in-process software title cache at roughly 100k entries so +// long-lived servers do not retain every title they have ever seen. +const maxKnownSoftwareTitleKeys = 100_000 + +// evictKnownSoftwareTitleKeys removes half the cache when the cap is hit. Keeping the other half +// preserves most steady-state hits while avoiding a full cold start that would reintroduce a burst +// of INSERT IGNORE statements. +const evictKnownSoftwareTitleKeys = maxKnownSoftwareTitleKeys / 2 + // WithPusher sets an APNs pusher for the datastore, used when activating // next activities that require MDM commands. func (ds *Datastore) WithPusher(p nano_push.Pusher) { @@ -284,17 +308,18 @@ func NewDBConnections(cfg config.MysqlConfig, opts ...DBOption) (*common_mysql.D // Use this when you need to share database connections with other bounded context datastores. func NewDatastore(conns *common_mysql.DBConnections, cfg config.MysqlConfig, c clock.Clock) (*Datastore, error) { ds := &Datastore{ - primary: conns.Primary, - replica: conns.Replica, - logger: conns.Options.Logger, - clock: c, - config: cfg, - readReplicaConfig: conns.Options.ReplicaConfig, - writeCh: make(chan itemToWrite), - stmtCache: make(map[string]*sqlx.Stmt), - minLastOpenedAtDiff: conns.Options.MinLastOpenedAtDiff, - serverPrivateKey: conns.Options.PrivateKey, - Datastore: NewAndroidDatastore(conns.Options.Logger, conns.Primary, conns.Replica), + primary: conns.Primary, + replica: conns.Replica, + logger: conns.Options.Logger, + clock: c, + config: cfg, + readReplicaConfig: conns.Options.ReplicaConfig, + writeCh: make(chan itemToWrite), + stmtCache: make(map[string]*sqlx.Stmt), + minLastOpenedAtDiff: conns.Options.MinLastOpenedAtDiff, + serverPrivateKey: conns.Options.PrivateKey, + knownSoftwareTitleKeys: make(map[string]struct{}), + Datastore: NewAndroidDatastore(conns.Options.Logger, conns.Primary, conns.Replica), } go ds.writeChanLoop() diff --git a/server/datastore/mysql/software.go b/server/datastore/mysql/software.go index 8eb6a50d4d..04ae25f20d 100644 --- a/server/datastore/mysql/software.go +++ b/server/datastore/mysql/software.go @@ -72,6 +72,18 @@ var cleanupBatchSize = 1000 // Any remaining orphans will be processed on the next hourly cron cycle. var cleanupMaxIterations = 100 +// softwareTitleCacheKey builds a string key for the in-process cache of known software titles. +// It mirrors the titleKey struct used inside preInsertSoftwareInventory. +func softwareTitleCacheKey(name, source, extensionFor, bundleID string, isKernel bool) string { + return strings.Join([]string{ + strings.ToLower(normalizeForCollation(name)), + source, + extensionFor, + strings.ToLower(bundleID), + strconv.FormatBool(isKernel), + }, fleet.SoftwareFieldSeparator) +} + func softwareSliceToMap(softwareItems []fleet.Software) map[string]fleet.Software { result := make(map[string]fleet.Software, len(softwareItems)) for _, s := range softwareItems { @@ -80,6 +92,52 @@ func softwareSliceToMap(softwareItems []fleet.Software) map[string]fleet.Softwar return result } +func (ds *Datastore) cacheKnownSoftwareTitleKey(key string) { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + if _, loaded := ds.knownSoftwareTitleKeys[key]; loaded { + return + } + if len(ds.knownSoftwareTitleKeys) >= maxKnownSoftwareTitleKeys { + ds.evictKnownSoftwareTitleKeysLocked() + } + // Store after potential eviction so the caller's key survives. + ds.knownSoftwareTitleKeys[key] = struct{}{} +} + +func (ds *Datastore) clearKnownSoftwareTitleKeys() { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + ds.knownSoftwareTitleKeys = make(map[string]struct{}) +} + +func (ds *Datastore) deleteKnownSoftwareTitleKey(key string) { + ds.knownSoftwareTitleKeysMu.Lock() + defer ds.knownSoftwareTitleKeysMu.Unlock() + delete(ds.knownSoftwareTitleKeys, key) +} + +func (ds *Datastore) hasKnownSoftwareTitleKey(key string) bool { + ds.knownSoftwareTitleKeysMu.RLock() + defer ds.knownSoftwareTitleKeysMu.RUnlock() + _, ok := ds.knownSoftwareTitleKeys[key] + return ok +} + +func (ds *Datastore) evictKnownSoftwareTitleKeysLocked() { + evicted := 0 + // Go map iteration order is randomized, so this evicts an arbitrary half of the cache. + // That is sufficient here because any retained title key still avoids an INSERT IGNORE, and + // arbitrary bulk eviction is much cheaper than maintaining a strict LRU in this hot path. + for key := range ds.knownSoftwareTitleKeys { + delete(ds.knownSoftwareTitleKeys, key) + evicted++ + if evicted >= evictKnownSoftwareTitleKeys { + return + } + } +} + func (ds *Datastore) UpdateHostSoftware(ctx context.Context, hostID uint, software []fleet.Software) (*fleet.UpdateHostSoftwareDBResult, error) { // OTEL instrumentation. It has no-op behavior when OTEL is not enabled. ctx, span := tracer.Start(ctx, "mysql.UpdateHostSoftware", @@ -293,13 +351,19 @@ func deleteHostSoftwareInstalledPaths( return nil } - stmt := `DELETE FROM host_software_installed_paths WHERE id IN (?)` - stmt, args, err := sqlx.In(stmt, toDelete) - if err != nil { - return ctxerr.Wrap(ctx, err, "building delete statement for delete host_software_installed_paths") - } - if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { - return ctxerr.Wrap(ctx, err, "executing delete statement for delete host_software_installed_paths") + const batchSize = 500 + for i := 0; i < len(toDelete); i += batchSize { + end := min(i+batchSize, len(toDelete)) + batch := toDelete[i:end] + + stmt := `DELETE FROM host_software_installed_paths WHERE id IN (?)` + stmt, args, err := sqlx.In(stmt, batch) + if err != nil { + return ctxerr.Wrap(ctx, err, "building delete statement for delete host_software_installed_paths") + } + if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { + return ctxerr.Wrap(ctx, err, "executing delete statement for delete host_software_installed_paths") + } } return nil @@ -1018,51 +1082,111 @@ func (ds *Datastore) preInsertSoftwareInventory( batchSoftware[key] = needsInsert[key] } - // Each batch in its own transaction - return ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error { - // First insert any needed software titles - newTitlesNeeded := make(map[string]fleet.SoftwareTitle) - for checksum, sw := range batchSoftware { - if _, ok := incomingChecksumsToExistingTitleSummaries[checksum]; !ok { - // there is not an existing software title corresponding to this incoming software version - newTitleName := sw.Name - if sw.BundleIdentifier != "" { - // First check if there's an FMA with this bundle identifier - use its canonical name - if fmaName, ok := fmaNames[sw.BundleIdentifier]; ok { - newTitleName = fmaName - } else { - // Fall back to computed best name from osquery reports - key := titleKey{ - bundleID: sw.BundleIdentifier, - source: sw.Source, - extensionFor: sw.ExtensionFor, - } - if computedName, exists := bestTitleNames[key]; exists { - newTitleName = computedName - } + // Compute which software titles need to be created. + // This is done outside the transaction because the computation is pure (no DB access). + newTitlesNeeded := make(map[string]fleet.SoftwareTitle) + for checksum, sw := range batchSoftware { + if _, ok := incomingChecksumsToExistingTitleSummaries[checksum]; !ok { + // there is not an existing software title corresponding to this incoming software version + newTitleName := sw.Name + if sw.BundleIdentifier != "" { + // First check if there's an FMA with this bundle identifier - use its canonical name + if fmaName, ok := fmaNames[sw.BundleIdentifier]; ok { + newTitleName = fmaName + } else { + // Fall back to computed best name from osquery reports + key := titleKey{ + bundleID: sw.BundleIdentifier, + source: sw.Source, + extensionFor: sw.ExtensionFor, + } + if computedName, exists := bestTitleNames[key]; exists { + newTitleName = computedName } } + } - newTitle := fleet.SoftwareTitle{ - Name: newTitleName, - Source: sw.Source, - ExtensionFor: sw.ExtensionFor, - IsKernel: sw.IsKernel, - } - if sw.BundleIdentifier != "" { - newTitle.BundleIdentifier = ptr.String(sw.BundleIdentifier) - } - if sw.ApplicationID != nil && *sw.ApplicationID != "" { - newTitle.ApplicationID = sw.ApplicationID - } - if sw.UpgradeCode != nil { - // intentionally write both empty and non-empty strings as upgrade codes - newTitle.UpgradeCode = sw.UpgradeCode - } - newTitlesNeeded[checksum] = newTitle + newTitle := fleet.SoftwareTitle{ + Name: newTitleName, + Source: sw.Source, + ExtensionFor: sw.ExtensionFor, + IsKernel: sw.IsKernel, + } + if sw.BundleIdentifier != "" { + newTitle.BundleIdentifier = new(sw.BundleIdentifier) + } + if sw.ApplicationID != nil && *sw.ApplicationID != "" { + newTitle.ApplicationID = sw.ApplicationID + } + if sw.UpgradeCode != nil { + // intentionally write both empty and non-empty strings as upgrade codes + newTitle.UpgradeCode = sw.UpgradeCode + } + newTitlesNeeded[checksum] = newTitle + } + } + + // INSERT IGNORE new software titles OUTSIDE the main transaction (#48719). + // Each INSERT IGNORE is auto-committed independently, so it holds row/gap locks + // for only microseconds instead of the entire transaction duration. This eliminates + // lock convoys when many hosts concurrently report the same software catalog. + if len(newTitlesNeeded) > 0 { + // Build the full set of unique titles (for ID resolution later). + uniqueTitles := make(map[titleKey]fleet.SoftwareTitle) + for _, title := range newTitlesNeeded { + bundleID := "" + if title.BundleIdentifier != nil { + bundleID = *title.BundleIdentifier + } + key := titleKey{ + name: strings.ToLower(normalizeForCollation(title.Name)), + source: title.Source, + extensionFor: title.ExtensionFor, + bundleID: bundleID, + isKernel: title.IsKernel, + } + if _, exists := uniqueTitles[key]; !exists { + uniqueTitles[key] = title } } + // INSERT IGNORE each title individually using auto-commit (outside any transaction). + // singleflight ensures that for each title key, only one goroutine actually + // executes the INSERT; concurrent goroutines wait and share the result. + // The in-process cache prevents future DB hits entirely. + const insertTitleStmt = `INSERT IGNORE INTO software_titles (name, source, extension_for, bundle_identifier, is_kernel, application_id, upgrade_code) VALUES (?,?,?,?,?,?,?)` + for key, title := range uniqueTitles { + cacheKey := softwareTitleCacheKey(title.Name, title.Source, title.ExtensionFor, key.bundleID, title.IsKernel) + if ds.hasKnownSoftwareTitleKey(cacheKey) { + continue + } + // Capture loop variables for the closure. + titleCopy := title + _, sfErr, _ := ds.titleInsertSF.Do(cacheKey, func() (any, error) { + // Double-check cache after winning the singleflight race. + if ds.hasKnownSoftwareTitleKey(cacheKey) { + return nil, nil + } + // Use context.WithoutCancel so the INSERT completes even if the + // leader goroutine's request is canceled mid-flight (#48719). + insertCtx := context.WithoutCancel(ctx) + if _, err := ds.writer(insertCtx).ExecContext(insertCtx, insertTitleStmt, + titleCopy.Name, titleCopy.Source, titleCopy.ExtensionFor, titleCopy.BundleIdentifier, + titleCopy.IsKernel, titleCopy.ApplicationID, titleCopy.UpgradeCode, + ); err != nil { + return nil, ctxerr.Wrap(ctx, err, "pre-insert software_titles") + } + ds.cacheKnownSoftwareTitleKey(cacheKey) + return nil, nil + }) + if sfErr != nil { + return sfErr + } + } + } + + // Each batch in its own transaction (for SELECT title IDs + INSERT software). + return ds.withRetryTxx(ctx, func(tx sqlx.ExtContext) error { // Map to store title IDs for all titles (both existing and new) titleIDsByChecksum := make(map[string]uint, len(incomingChecksumsToExistingTitleSummaries)) @@ -1071,47 +1195,33 @@ func (ds *Datastore) preInsertSoftwareInventory( titleIDsByChecksum[checksum] = titleSummary.ID } if len(newTitlesNeeded) > 0 { - uniqueTitlesToInsert := make(map[titleKey]fleet.SoftwareTitle) + // Build the set of unique titles for the SELECT query. + uniqueTitles := make(map[titleKey]fleet.SoftwareTitle) for _, title := range newTitlesNeeded { bundleID := "" if title.BundleIdentifier != nil { bundleID = *title.BundleIdentifier } key := titleKey{ - // adjust for matching MySQL collation name: strings.ToLower(normalizeForCollation(title.Name)), source: title.Source, extensionFor: title.ExtensionFor, bundleID: bundleID, isKernel: title.IsKernel, } - - if _, exists := uniqueTitlesToInsert[key]; !exists { - uniqueTitlesToInsert[key] = title + if _, exists := uniqueTitles[key]; !exists { + uniqueTitles[key] = title } } - // Insert software titles - const numberOfArgsPerSoftwareTitles = 7 - titlesValues := strings.TrimSuffix(strings.Repeat("(?,?,?,?,?,?,?),", len(uniqueTitlesToInsert)), ",") - titlesStmt := fmt.Sprintf("INSERT IGNORE INTO software_titles (name, source, extension_for, bundle_identifier, is_kernel, application_id, upgrade_code) VALUES %s", titlesValues) - titlesArgs := make([]any, 0, len(uniqueTitlesToInsert)*numberOfArgsPerSoftwareTitles) - - for _, title := range uniqueTitlesToInsert { - titlesArgs = append(titlesArgs, title.Name, title.Source, title.ExtensionFor, title.BundleIdentifier, title.IsKernel, title.ApplicationID, title.UpgradeCode) - } - - if _, err := tx.ExecContext(ctx, titlesStmt, titlesArgs...); err != nil { - return ctxerr.Wrap(ctx, err, "pre-insert software_titles") - } - - // Retrieve the IDs for the titles we just inserted (or that already existed) + // Retrieve the IDs for the titles we just inserted (or that already existed). + // Use uniqueTitles (all unique titles) so we resolve IDs for cached titles too. var retrievedTitleSummaries []fleet.SoftwareTitleSummary - titlePlaceholders := strings.TrimSuffix(strings.Repeat("(?,?,?,?),", len(uniqueTitlesToInsert)), ",") - queryArgs := make([]interface{}, 0, len(uniqueTitlesToInsert)*4) + titlePlaceholders := strings.TrimSuffix(strings.Repeat("(?,?,?,?),", len(uniqueTitles)), ",") + queryArgs := make([]any, 0, len(uniqueTitles)*4) var upgradeCodes []string - for tk := range uniqueTitlesToInsert { - title := uniqueTitlesToInsert[tk] + for tk := range uniqueTitles { + title := uniqueTitles[tk] bundleID := "" if title.BundleIdentifier != nil { bundleID = *title.BundleIdentifier @@ -1239,6 +1349,7 @@ func (ds *Datastore) preInsertSoftwareInventory( } } } + } // Insert software entries @@ -1266,7 +1377,7 @@ func (ds *Datastore) preInsertSoftwareInventory( ) args := make([]any, 0, len(batchKeys)*numberOfArgsPerSoftware) - var missingSoftwareTitles []string + var missingChecksums []string for _, checksum := range batchKeys { sw := batchSoftware[checksum] var titleID *uint @@ -1274,9 +1385,9 @@ func (ds *Datastore) preInsertSoftwareInventory( if id, ok := titleIDsByChecksum[checksum]; ok { titleID = &id } else { - // Track software missing title IDs for debugging - missingSoftwareTitles = append(missingSoftwareTitles, - fmt.Sprintf("%s %s %s", sw.Name, sw.Version, sw.Source)) + // Track software missing title IDs; titles inserted outside the + // transaction may have been deleted by a concurrent CleanupSoftwareTitles. + missingChecksums = append(missingChecksums, checksum) } // Use FMA canonical name if available, otherwise use osquery-reported name. @@ -1310,17 +1421,38 @@ func (ds *Datastore) preInsertSoftwareInventory( ) } - // Log an error if we have software without title IDs - // This shouldn't happen in normal operation. And this code is here to catch bugs. - if len(missingSoftwareTitles) > 0 && ds.logger != nil { - exampleCount := 3 - if len(missingSoftwareTitles) < exampleCount { - exampleCount = len(missingSoftwareTitles) + // When title IDs are missing, a concurrent CleanupSoftwareTitles likely + // deleted the titles we just inserted (they were orphaned briefly outside the + // transaction). Clear those cache entries so they are re-inserted on the next + // agent check-in. The software row proceeds with NULL title_id; the next + // ingestion cycle will re-create the title and link it. + if len(missingChecksums) > 0 { + var examples []string + for _, checksum := range missingChecksums { + sw := batchSoftware[checksum] + if len(examples) < 3 { + examples = append(examples, fmt.Sprintf("%s %s %s", sw.Name, sw.Version, sw.Source)) + } + // Evict from the in-process cache so the next ingestion cycle re-inserts the title. + if title, ok := newTitlesNeeded[checksum]; ok { + bundleID := "" + if title.BundleIdentifier != nil { + bundleID = *title.BundleIdentifier + } + cacheKey := softwareTitleCacheKey(title.Name, title.Source, title.ExtensionFor, bundleID, title.IsKernel) + ds.deleteKnownSoftwareTitleKey(cacheKey) + } + } + // Log rather than return a hard error: the title INSERT is outside the + // transaction, so withRetryTxx cannot re-insert the title on retry. + // The software row proceeds with NULL title_id and the evicted cache + // entry ensures the title is re-created on the next ingestion cycle. + if ds.logger != nil { + ds.logger.ErrorContext(ctx, "inserting software without title_id", + "count", len(missingChecksums), + "examples", strings.Join(examples, "; "), + ) } - ds.logger.ErrorContext(ctx, "inserting software without title_id", - "count", len(missingSoftwareTitles), - "examples", strings.Join(missingSoftwareTitles[:exampleCount], "; "), - ) } if _, err := tx.ExecContext(ctx, stmt, args...); err != nil { @@ -3040,7 +3172,7 @@ func (ds *Datastore) CleanupSoftwareTitles(ctx context.Context) error { return ctxerr.Wrap(ctx, err, "find orphaned software titles for cleanup") } if len(ids) == 0 { - return nil + break } lastID = ids[len(ids)-1] @@ -3055,6 +3187,13 @@ func (ds *Datastore) CleanupSoftwareTitles(ctx context.Context) error { ra, _ := res.RowsAffected() n += ra } + + // If any titles were deleted, clear the in-process title cache so that future + // software ingestions re-insert titles instead of skipping them. + if n > 0 { + ds.clearKnownSoftwareTitleKeys() + } + return nil } diff --git a/server/datastore/mysql/software_lock_convoy_test.go b/server/datastore/mysql/software_lock_convoy_test.go new file mode 100644 index 0000000000..3bf35d3d22 --- /dev/null +++ b/server/datastore/mysql/software_lock_convoy_test.go @@ -0,0 +1,291 @@ +package mysql + +import ( + "context" + "fmt" + "slices" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/fleetdm/fleet/v4/server/fleet" + "github.com/fleetdm/fleet/v4/server/test" + "github.com/jmoiron/sqlx" + "github.com/stretchr/testify/require" + "golang.org/x/sync/errgroup" +) + +// TestSoftwareTitlesInsertIgnoreLockConvoy reproduces the lock convoy described in #48719. +// +// Setup: N hosts all report the SAME software catalog (homogeneous fleet, like imaged +// corporate Windows machines). The software_titles table starts empty (cold start). +// All hosts race to INSERT IGNORE the same titles concurrently. +// +// Expected: With the current code, concurrent INSERT IGNORE statements on the same +// unique-index rows serialize and cause high contention. This test measures timing +// to confirm the convoy is observable even at modest concurrency. +func TestSoftwareTitlesInsertIgnoreLockConvoy(t *testing.T) { + ds := CreateMySQLDS(t) + ctx := context.Background() + + const ( + hostCount = 50 // concurrent hosts + softwareCount = 100 // software items per host (all identical across hosts) + ) + + // Create hosts + hosts := make([]*fleet.Host, hostCount) + for i := range hostCount { + h, err := ds.NewHost(ctx, &fleet.Host{ + OsqueryHostID: new(fmt.Sprintf("convoy-host-%d", i)), + NodeKey: new(fmt.Sprintf("convoy-key-%d", i)), + Platform: "windows", + Hostname: fmt.Sprintf("convoy-host-%d", i), + DetailUpdatedAt: time.Now(), + LabelUpdatedAt: time.Now(), + PolicyUpdatedAt: time.Now(), + SeenTime: time.Now(), + }) + require.NoError(t, err) + hosts[i] = h + } + + // Build a SINGLE software catalog shared by ALL hosts (homogeneous fleet). + // This is the key condition for the lock convoy: every host tries to INSERT IGNORE + // the same software_titles rows. + sharedSoftware := make([]fleet.Software, softwareCount) + for i := range softwareCount { + sharedSoftware[i] = fleet.Software{ + Name: fmt.Sprintf("ConvoyApp %d", i), + Version: "1.0.0", + Source: "programs", + } + } + + // --- Cold-start convoy: all hosts ingest simultaneously with empty software_titles --- + t.Log("Starting cold-start convoy test...") + t.Logf(" Hosts: %d, Software items per host: %d", hostCount, softwareCount) + + var ( + g errgroup.Group + maxElapsed atomic.Int64 + totalMs atomic.Int64 + ready = make(chan struct{}) // barrier to synchronize start + ) + + for i := range hostCount { + hostID := hosts[i].ID + // Copy the slice to avoid data races (UpdateHostSoftware may mutate it in-place). + sw := slices.Clone(sharedSoftware) + g.Go(func() error { + <-ready // wait for all goroutines to be ready + start := time.Now() + _, err := ds.UpdateHostSoftware(ctx, hostID, sw) + elapsed := time.Since(start) + ms := elapsed.Milliseconds() + totalMs.Add(ms) + for { + old := maxElapsed.Load() + if ms <= old || maxElapsed.CompareAndSwap(old, ms) { + break + } + } + if err != nil { + return fmt.Errorf("host %d: %w", hostID, err) + } + return nil + }) + } + + start := time.Now() + close(ready) // release all goroutines at once + err := g.Wait() + wallTime := time.Since(start) + + require.NoError(t, err) + + t.Logf(" Cold-start results:") + t.Logf(" Wall time: %s", wallTime) + t.Logf(" Max single-host ingestion: %dms", maxElapsed.Load()) + t.Logf(" Avg per-host ingestion: %dms", totalMs.Load()/int64(hostCount)) + + // Verify all titles were created + var titleCount int + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.GetContext(ctx, q, &titleCount, `SELECT COUNT(*) FROM software_titles WHERE source = 'programs'`) + }) + t.Logf(" Software titles created: %d (expected %d)", titleCount, softwareCount) + require.Equal(t, softwareCount, titleCount) + + // --- Steady-state: re-ingest same software (should be fast, read-only path) --- + t.Log("Starting steady-state re-ingestion test...") + maxElapsed.Store(0) + totalMs.Store(0) + ready2 := make(chan struct{}) + + var g2 errgroup.Group + for i := range hostCount { + hostID := hosts[i].ID + sw := slices.Clone(sharedSoftware) + g2.Go(func() error { + <-ready2 + start := time.Now() + _, err := ds.UpdateHostSoftware(ctx, hostID, sw) + elapsed := time.Since(start) + ms := elapsed.Milliseconds() + totalMs.Add(ms) + for { + old := maxElapsed.Load() + if ms <= old || maxElapsed.CompareAndSwap(old, ms) { + break + } + } + if err != nil { + return fmt.Errorf("host %d: %w", hostID, err) + } + return nil + }) + } + + start2 := time.Now() + close(ready2) + err = g2.Wait() + wallTime2 := time.Since(start2) + + require.NoError(t, err) + + t.Logf(" Steady-state results:") + t.Logf(" Wall time: %s", wallTime2) + t.Logf(" Max single-host ingestion: %dms", maxElapsed.Load()) + t.Logf(" Avg per-host ingestion: %dms", totalMs.Load()/int64(hostCount)) + + // The cold-start should be significantly slower than steady-state due to lock contention + if wallTime2 > 0 { + t.Logf("\n Convoy factor (cold wall / steady wall): %.1fx", float64(wallTime)/float64(wallTime2)) + } +} + +// TestHostSoftwareInstalledPathsDeleteExplosion reproduces #49805. +// +// A host with many installed paths gets re-enrolled or its software changes significantly, +// triggering a DELETE FROM host_software_installed_paths WHERE id IN (thousands of IDs) +// in a single unbatched statement. +func TestHostSoftwareInstalledPathsDeleteExplosion(t *testing.T) { + ds := CreateMySQLDS(t) + ctx := context.Background() + + // Create a host + host := test.NewHost(t, ds, "delete-explosion-host", "", "de-key", "de-uuid", time.Now()) + + // Insert a large number of software items, each with an installed path + const softwareCount = 500 // a more modest number than 30k for local testing + software := make([]fleet.Software, softwareCount) + for i := range softwareCount { + software[i] = fleet.Software{ + Name: fmt.Sprintf("DeleteTestApp %d", i), + Version: "1.0.0", + Source: "apps", + } + } + + // First ingestion: establish software + _, err := ds.UpdateHostSoftware(ctx, host.ID, software) + require.NoError(t, err) + + // Get the software IDs that were created + var swIDs []struct { + ID uint `db:"id"` + Name string `db:"name"` + } + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.SelectContext(ctx, q, &swIDs, + `SELECT id, name FROM software WHERE name LIKE 'DeleteTestApp%' AND source = 'apps'`) + }) + t.Logf("Created %d software entries", len(swIDs)) + + // Directly insert installed paths to build up the table + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + for _, sw := range swIDs { + _, err := q.ExecContext(ctx, + `INSERT INTO host_software_installed_paths (host_id, software_id, installed_path) VALUES (?, ?, ?)`, + host.ID, sw.ID, fmt.Sprintf("/Applications/%s.app", sw.Name)) + if err != nil { + return err + } + } + return nil + }) + + // Verify the paths are there + var pathCount int + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + return sqlx.GetContext(ctx, q, &pathCount, + `SELECT COUNT(*) FROM host_software_installed_paths WHERE host_id = ?`, host.ID) + }) + t.Logf("Installed paths for host: %d", pathCount) + require.Equal(t, len(swIDs), pathCount) + + // Now simulate a "full replacement" by reporting all-new software with no overlap. + // This causes ALL existing paths to be deleted in one shot. + newSoftware := make([]fleet.Software, softwareCount) + for i := range softwareCount { + newSoftware[i] = fleet.Software{ + Name: fmt.Sprintf("ReplacementApp %d", i), + Version: "2.0.0", + Source: "apps", + } + } + + // This should trigger a massive DELETE of all old installed paths + startDel := time.Now() + _, err = ds.UpdateHostSoftware(ctx, host.ID, newSoftware) + elapsed := time.Since(startDel) + require.NoError(t, err) + t.Logf("Full software replacement took: %s", elapsed) + + // Now test with concurrent hosts doing the same thing + t.Log("Testing concurrent large deletes...") + const concurrentHosts = 10 + var wg sync.WaitGroup + wg.Add(concurrentHosts) + + // Create hosts and insert paths synchronously to avoid require.*/ExecAdhocSQL panics from goroutines. + concurrentTestHosts := make([]*fleet.Host, concurrentHosts) + for i := range concurrentHosts { + concurrentTestHosts[i] = test.NewHost(t, ds, fmt.Sprintf("concurrent-del-%d", i), "", fmt.Sprintf("cd-key-%d", i), fmt.Sprintf("cd-uuid-%d", i), time.Now()) + + swCopy := slices.Clone(software) + _, err := ds.UpdateHostSoftware(ctx, concurrentTestHosts[i].ID, swCopy) + require.NoError(t, err) + + ExecAdhocSQL(t, ds, func(q sqlx.ExtContext) error { + for _, sw := range swIDs { + _, err := q.ExecContext(ctx, + `INSERT IGNORE INTO host_software_installed_paths (host_id, software_id, installed_path) VALUES (?, ?, ?)`, + concurrentTestHosts[i].ID, sw.ID, fmt.Sprintf("/Applications/%s.app", sw.Name)) + if err != nil { + return err + } + } + return nil + }) + } + + for i := range concurrentHosts { + go func(idx int) { + defer wg.Done() + h := concurrentTestHosts[idx] + newSwCopy := slices.Clone(newSoftware) + + startReplace := time.Now() + _, err := ds.UpdateHostSoftware(ctx, h.ID, newSwCopy) + elapsedReplace := time.Since(startReplace) + t.Logf(" Host %d replacement took: %s", idx, elapsedReplace) + if err != nil { + t.Errorf(" Host %d replacement error: %v", idx, err) + } + }(i) + } + wg.Wait() +} diff --git a/server/datastore/mysql/testing_utils_test.go b/server/datastore/mysql/testing_utils_test.go index 087592c810..0aa677a6d1 100644 --- a/server/datastore/mysql/testing_utils_test.go +++ b/server/datastore/mysql/testing_utils_test.go @@ -464,6 +464,9 @@ func TruncateTables(t testing.TB, ds *Datastore, tables ...string) { "DELETE FROM software_categories WHERE team_id != 0") require.NoError(t, err) testing_utils.TruncateTables(t, ds.writer(context.Background()), ds.logger, nonEmptyTables, tables...) + // Clear the in-process software title cache so it doesn't retain entries + // for titles that were just truncated from the database. + ds.clearKnownSoftwareTitleKeys() } // this is meant to be used for debugging/testing that statement uses an efficient