diff --git a/changes/25682-migrate-pending-existing-activities b/changes/25682-migrate-pending-existing-activities new file mode 100644 index 0000000000..12de643228 --- /dev/null +++ b/changes/25682-migrate-pending-existing-activities @@ -0,0 +1 @@ +* Added a DB migration to migrate existing pending activities to the new unified queue. diff --git a/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities.go b/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities.go new file mode 100644 index 0000000000..adba22e7ba --- /dev/null +++ b/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities.go @@ -0,0 +1,337 @@ +package tables + +import ( + "database/sql" + "fmt" +) + +func init() { + MigrationClient.AddMigration(Up_20250217093329, Down_20250217093329) +} + +func Up_20250217093329(tx *sql.Tx) error { + // this migration inserts pending software installs, software uninstalls, + // VPP app installs and script executions in the upcoming_activities table + // (inserts them already marked as "activated" since they are ready to be + // processed). There is no ordering guarantee for those already-pending + // activities, but any new upcoming activity will follow the unified queue + // order. + if err := migrateSoftwareInstalls(tx); err != nil { + return err + } + if err := migrateSoftwareUninstalls(tx); err != nil { + return err + } + if err := migrateVPPInstalls(tx); err != nil { + return err + } + if err := migrateScriptExecs(tx); err != nil { + return err + } + return nil +} + +func migrateSoftwareInstalls(tx *sql.Tx) error { + _, err := tx.Exec(` +INSERT INTO upcoming_activities + ( + host_id, + priority, + user_id, + fleet_initiated, + activity_type, + execution_id, + payload, + activated_at + ) +SELECT + hsi.host_id, + 0, + hsi.user_id, + hsi.policy_id IS NOT NULL, -- true if fleet-initiated + 'software_install', + hsi.execution_id, + JSON_OBJECT( + 'self_service', hsi.self_service, + 'installer_filename', hsi.installer_filename, + 'version', hsi.version, + 'software_title_name', hsi.software_title_name, + 'user', (SELECT JSON_OBJECT('name', name, 'email', email, 'gravatar_url', gravatar_url) FROM users WHERE id = hsi.user_id) + ), + hsi.created_at +FROM + host_software_installs hsi + LEFT OUTER JOIN upcoming_activities ua + ON hsi.execution_id = ua.execution_id +WHERE + ua.id IS NULL AND + hsi.status = 'pending_install' AND + hsi.host_deleted_at IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending software installs: %w", err) + } + + _, err = tx.Exec(` +INSERT INTO software_install_upcoming_activities + ( + upcoming_activity_id, + software_installer_id, + policy_id, + software_title_id, + created_at + ) +SELECT + ua.id, + hsi.software_installer_id, + hsi.policy_id, + hsi.software_title_id, + hsi.created_at +FROM + upcoming_activities ua + INNER JOIN host_software_installs hsi + ON hsi.execution_id = ua.execution_id + LEFT OUTER JOIN software_install_upcoming_activities sia + ON sia.upcoming_activity_id = ua.id +WHERE + ua.activity_type = 'software_install' AND + hsi.status = 'pending_install' AND + hsi.host_deleted_at IS NULL AND + sia.upcoming_activity_id IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending software installs secondary table: %w", err) + } + return nil +} + +func migrateSoftwareUninstalls(tx *sql.Tx) error { + _, err := tx.Exec(` +INSERT INTO upcoming_activities + ( + host_id, + priority, + user_id, + fleet_initiated, + activity_type, + execution_id, + payload, + activated_at + ) +SELECT + hsi.host_id, + 0, + hsi.user_id, + hsi.policy_id IS NOT NULL, -- true if fleet-initiated + 'software_uninstall', + hsi.execution_id, + JSON_OBJECT( + 'installer_filename', '', + 'version', 'unknown', + 'software_title_name', hsi.software_title_name, + 'user', (SELECT JSON_OBJECT('name', name, 'email', email, 'gravatar_url', gravatar_url) FROM users WHERE id = hsi.user_id) + ), + hsi.created_at +FROM + host_software_installs hsi + LEFT OUTER JOIN upcoming_activities ua + ON hsi.execution_id = ua.execution_id +WHERE + ua.id IS NULL AND + hsi.status = 'pending_uninstall' AND + hsi.host_deleted_at IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending software uninstalls: %w", err) + } + + _, err = tx.Exec(` +INSERT INTO software_install_upcoming_activities + ( + upcoming_activity_id, + software_installer_id, + software_title_id, + created_at + ) +SELECT + ua.id, + hsi.software_installer_id, + hsi.software_title_id, + hsi.created_at +FROM + upcoming_activities ua + INNER JOIN host_software_installs hsi + ON hsi.execution_id = ua.execution_id + LEFT OUTER JOIN software_install_upcoming_activities sia + ON sia.upcoming_activity_id = ua.id +WHERE + ua.activity_type = 'software_uninstall' AND + hsi.status = 'pending_uninstall' AND + hsi.host_deleted_at IS NULL AND + sia.upcoming_activity_id IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending software uninstalls secondary table: %w", err) + } + return nil +} + +func migrateVPPInstalls(tx *sql.Tx) error { + _, err := tx.Exec(` +INSERT INTO upcoming_activities + ( + host_id, + priority, + user_id, + fleet_initiated, + activity_type, + execution_id, + payload, + activated_at + ) +SELECT + hvi.host_id, + 0, + hvi.user_id, + hvi.policy_id IS NOT NULL, -- true if fleet-initiated + 'vpp_app_install', + hvi.command_uuid, + JSON_OBJECT( + 'self_service', hvi.self_service, + 'associated_event_id', hvi.associated_event_id, + 'user', (SELECT JSON_OBJECT('name', name, 'email', email, 'gravatar_url', gravatar_url) FROM users WHERE id = hvi.user_id) + ), + COALESCE(hvi.created_at, NOW(6)) +FROM + host_vpp_software_installs hvi + INNER JOIN + nano_view_queue nvq ON nvq.command_uuid = hvi.command_uuid + LEFT OUTER JOIN upcoming_activities ua + ON hvi.command_uuid = ua.execution_id +WHERE + ua.id IS NULL AND + nvq.status IS NULL AND + hvi.removed = 0 +`) + if err != nil { + return fmt.Errorf("failed to insert pending vpp app installs: %w", err) + } + + _, err = tx.Exec(` +INSERT INTO vpp_app_upcoming_activities + ( + upcoming_activity_id, + adam_id, + platform, + policy_id, + created_at + ) +SELECT + ua.id, + hvi.adam_id, + hvi.platform, + hvi.policy_id, + hvi.created_at +FROM + upcoming_activities ua + INNER JOIN host_vpp_software_installs hvi + ON hvi.command_uuid = ua.execution_id + INNER JOIN + nano_view_queue nvq ON nvq.command_uuid = hvi.command_uuid + LEFT OUTER JOIN vpp_app_upcoming_activities vaua + ON vaua.upcoming_activity_id = ua.id +WHERE + ua.activity_type = 'vpp_app_install' AND + hvi.removed = 0 AND + nvq.status IS NULL AND + vaua.upcoming_activity_id IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending vpp app installs secondary table: %w", err) + } + return nil +} + +func migrateScriptExecs(tx *sql.Tx) error { + // we don't want to migrate software uninstall scripts (those are already + // covered by the software uninstalls), but we don't have anything special to + // do as we will automatically ignore them with the left join on + // upcoming_activities (and the fact that software uninstalls are processed + // before scripts), because the uninstall scripts have the same execution id + // as the corresponding software uninstall. + _, err := tx.Exec(` +INSERT INTO upcoming_activities + ( + host_id, + priority, + user_id, + fleet_initiated, + activity_type, + execution_id, + payload, + activated_at + ) +SELECT + hsr.host_id, + 0, + hsr.user_id, + hsr.policy_id IS NOT NULL, -- true if fleet-initiated + 'script', + hsr.execution_id, + JSON_OBJECT( + 'sync_request', hsr.sync_request, + 'is_internal', hsr.is_internal, + 'user', (SELECT JSON_OBJECT('name', name, 'email', email, 'gravatar_url', gravatar_url) FROM users WHERE id = hsr.user_id) + ), + hsr.created_at +FROM + host_script_results hsr + LEFT OUTER JOIN upcoming_activities ua + ON hsr.execution_id = ua.execution_id +WHERE + ua.id IS NULL AND + hsr.exit_code IS NULL AND -- script is pending execution + hsr.host_deleted_at IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending script executions: %w", err) + } + + _, err = tx.Exec(` +INSERT INTO script_upcoming_activities + ( + upcoming_activity_id, + script_id, + script_content_id, + policy_id, + setup_experience_script_id, + created_at + ) +SELECT + ua.id, + hsr.script_id, + hsr.script_content_id, + hsr.policy_id, + hsr.setup_experience_script_id, + hsr.created_at +FROM + upcoming_activities ua + INNER JOIN host_script_results hsr + ON hsr.execution_id = ua.execution_id + LEFT OUTER JOIN script_upcoming_activities sua + ON sua.upcoming_activity_id = ua.id +WHERE + ua.activity_type = 'script' AND + hsr.exit_code IS NULL AND + hsr.host_deleted_at IS NULL AND + sua.upcoming_activity_id IS NULL +`) + if err != nil { + return fmt.Errorf("failed to insert pending script executions secondary table: %w", err) + } + return nil +} + +func Down_20250217093329(tx *sql.Tx) error { + return nil +} diff --git a/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities_test.go b/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities_test.go new file mode 100644 index 0000000000..23a465502f --- /dev/null +++ b/server/datastore/mysql/migrations/tables/20250217093329_MigratePendingUpcomingActivities_test.go @@ -0,0 +1,332 @@ +package tables + +import ( + "testing" + "time" + + "github.com/google/uuid" + "github.com/stretchr/testify/require" +) + +func TestUp_20250217093329_None(t *testing.T) { + db := applyUpToPrev(t) + // Apply current migration. + applyNext(t, db) + assertRowCount(t, db, "upcoming_activities", 0) +} + +func TestUp_20250217093329_Script(t *testing.T) { + db := applyUpToPrev(t) + hostID := insertHost(t, db, nil) + + // create a script content + contentIDs := insertScriptContents(t, db, 1) + scriptContentID := contentIDs[0] + + // insert a couple pending but one has host_deleted_at set, and a non-pending script + execIDPending, execIDDeleted, execIDDone := uuid.NewString(), uuid.NewString(), uuid.NewString() + execNoErr(t, db, `INSERT INTO host_script_results + (host_id, execution_id, output, script_content_id, host_deleted_at) + VALUES (?, ?, '', ?, ?)`, hostID, execIDPending, scriptContentID, nil) + execNoErr(t, db, `INSERT INTO host_script_results + (host_id, execution_id, output, script_content_id, host_deleted_at) + VALUES (?, ?, '', ?, ?)`, hostID, execIDDeleted, scriptContentID, time.Now()) + execNoErr(t, db, `INSERT INTO host_script_results + (host_id, execution_id, output, script_content_id, exit_code) + VALUES (?, ?, '', ?, 0)`, hostID, execIDDone, scriptContentID) + + // Apply current migration. + applyNext(t, db) + assertRowCount(t, db, "upcoming_activities", 1) + assertRowCount(t, db, "script_upcoming_activities", 1) + assertRowCount(t, db, "software_install_upcoming_activities", 0) + assertRowCount(t, db, "vpp_app_upcoming_activities", 0) + + var execID string + err := db.Get(&execID, `SELECT execution_id FROM upcoming_activities`) + require.NoError(t, err) + require.Equal(t, execIDPending, execID) + + var count int + err = db.Get(&count, `SELECT COUNT(*) FROM upcoming_activities WHERE activated_at IS NULL`) + require.NoError(t, err) + require.EqualValues(t, 0, count) +} + +func TestUp_20250217093329_SoftwareInstall(t *testing.T) { + db := applyUpToPrev(t) + hostID := insertHost(t, db, nil) + + installerIDs, _ := insertSoftwareInstallers(t, db, 1) + installerID := installerIDs[0] + + // insert a few pending but one has host_deleted_at, uninstall or removed set, and a non-pending install + hsiStmt := ` + INSERT INTO host_software_installs ( + host_id, execution_id, software_installer_id, install_script_exit_code, + host_deleted_at, removed, uninstall + ) VALUES (?, ?, ?, ?, ?, ?, ?)` + execIDPending, execIDDeleted, execIDUninstall, execIDRemoved, execIDFailed := + uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString() + execNoErr(t, db, hsiStmt, hostID, execIDPending, installerID, nil, nil, false, false) + execNoErr(t, db, hsiStmt, hostID, execIDDeleted, installerID, nil, time.Now(), false, false) + execNoErr(t, db, hsiStmt, hostID, execIDUninstall, installerID, nil, nil, false, true) + execNoErr(t, db, hsiStmt, hostID, execIDRemoved, installerID, nil, nil, true, false) + execNoErr(t, db, hsiStmt, hostID, execIDFailed, installerID, 1, nil, false, false) + + t.Log("exec IDs: ", execIDPending, execIDDeleted, execIDUninstall, execIDRemoved, execIDFailed) + + applyNext(t, db) + assertRowCount(t, db, "upcoming_activities", 2) + assertRowCount(t, db, "software_install_upcoming_activities", 2) + assertRowCount(t, db, "vpp_app_upcoming_activities", 0) + assertRowCount(t, db, "script_upcoming_activities", 0) + + var execIDs []string + err := db.Select(&execIDs, `SELECT execution_id FROM upcoming_activities`) + require.NoError(t, err) + // will add both the pending install and uninstall to upcoming, but not the + // host deleted entry, the removed and the failed install + require.ElementsMatch(t, []string{execIDPending, execIDUninstall}, execIDs) + + var count int + err = db.Get(&count, `SELECT COUNT(*) FROM upcoming_activities WHERE activated_at IS NULL`) + require.NoError(t, err) + require.EqualValues(t, 0, count) +} + +func TestUp_20250217093329_SoftwareUninstall(t *testing.T) { + db := applyUpToPrev(t) + hostID := insertHost(t, db, nil) + + installerIDs, _ := insertSoftwareInstallers(t, db, 1) + installerID := installerIDs[0] + + // insert a few pending but one has host_deleted_at, is an install or has + // removed set, and a non-pending uninstall + hsiStmt := ` + INSERT INTO host_software_installs ( + host_id, execution_id, software_installer_id, uninstall_script_exit_code, + host_deleted_at, removed, uninstall + ) VALUES (?, ?, ?, ?, ?, ?, ?)` + execIDPending, execIDDeleted, execIDInstall, execIDRemoved, execIDFailed := + uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString(), uuid.NewString() + execNoErr(t, db, hsiStmt, hostID, execIDPending, installerID, nil, nil, false, true) + execNoErr(t, db, hsiStmt, hostID, execIDDeleted, installerID, nil, time.Now(), false, true) + execNoErr(t, db, hsiStmt, hostID, execIDInstall, installerID, nil, nil, false, false) + execNoErr(t, db, hsiStmt, hostID, execIDRemoved, installerID, nil, nil, true, true) + execNoErr(t, db, hsiStmt, hostID, execIDFailed, installerID, 1, nil, false, true) + + t.Log("exec IDs: ", execIDPending, execIDDeleted, execIDInstall, execIDRemoved, execIDFailed) + + applyNext(t, db) + assertRowCount(t, db, "upcoming_activities", 2) + assertRowCount(t, db, "software_install_upcoming_activities", 2) + assertRowCount(t, db, "vpp_app_upcoming_activities", 0) + assertRowCount(t, db, "script_upcoming_activities", 0) + + var execIDs []string + err := db.Select(&execIDs, `SELECT execution_id FROM upcoming_activities`) + require.NoError(t, err) + // will add both the pending install and uninstall to upcoming, but not the + // host deleted entry, the removed and the failed uninstall + require.ElementsMatch(t, []string{execIDPending, execIDInstall}, execIDs) + + var count int + err = db.Get(&count, `SELECT COUNT(*) FROM upcoming_activities WHERE activated_at IS NULL`) + require.NoError(t, err) + require.EqualValues(t, 0, count) +} + +func TestUp_20250217093329_VPPInstall(t *testing.T) { + db := applyUpToPrev(t) + hostID := insertHost(t, db, nil) + hostUUID := "12345678-1234-1234-1234-123456789012" + + adamIDs, _ := insertVPPApps(t, db, 1, "darwin") + adamID := adamIDs[0] + + execNoErr(t, db, `INSERT INTO nano_devices (id, authenticate) + VALUES (?, ?)`, hostUUID, "auth") + execNoErr(t, db, `INSERT INTO nano_enrollments (id, device_id, type, topic, push_magic, token_hex, last_seen_at) + VALUES (?, ?, ?, ?, ?, ?, ?)`, hostUUID, hostUUID, "device", "topic", "magic", "hex", time.Now()) + + // create a few pending but one is removed, and a non-pending install + execIDPending, execIDRemoved, execIDDone := uuid.NewString(), uuid.NewString(), uuid.NewString() + execNoErr(t, db, `INSERT INTO host_vpp_software_installs + (host_id, adam_id, platform, command_uuid, removed) VALUES (?, ?, ?, ?, ?)`, + hostID, adamID, "darwin", execIDPending, false) + execNoErr(t, db, `INSERT INTO nano_commands (command_uuid, request_type, command) VALUES (?, ?, ?)`, + execIDPending, "InstallApplication", "