From dcf5203ff4f31184d8e1ea91052ceaad050570e0 Mon Sep 17 00:00:00 2001 From: Rajendra kadam Date: Wed, 17 Jun 2026 14:36:10 +0530 Subject: [PATCH] Extract cron schedule registration out of runServeCmd (#47562) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extracts the cron schedule registration out of `runServeCmd` and into a new `cmd/fleet/cron_registration.go`. Same pattern as the prior extractions on this issue (#44929, #45343, #45583, #46166, #46421, #46517, #46742, #46830, #46893, #47151). This is the largest slice so far — `runServeCmd` drops from ~1300 to ~1000 lines, and `serve.go` from 1776 to 1472. The 33 `StartCronSchedule` registrations move into one `startCronSchedules` entry point backed by a `cronSchedulesDeps` struct (the dependencies the closures previously captured from `runServeCmd`). Registration is grouped by domain: - `registerCleanupAndMaintenanceCrons` — chart data collection, the `cron_stats` cleanup goroutine, software migrations, frequent cleanups, cleanups-then-aggregation, query results cleanup, upcoming activities, usage statistics, batch activities. - `registerVulnerabilityCrons` — the vulnerabilities schedule, or the remote-trigger proxy when processing is disabled on this instance. - `registerWorkerCrons` — automations and worker integrations. - `registerMDMCrons` — Apple MDM worker, DEP profile assigner, service discovery, the Apple/Windows/Android profile managers, the Android device reconciler, the Android policy migrations, and the APNs pusher. - `registerPremiumCrons` — iPhone/iPad refetcher and reviver, maintained apps, VPP app version refresh (and the one-shot VPP country backfill), recovery lock passwords, managed local account rotation, activities streaming, and the calendar schedule. - `registerMiscCrons` — host vitals label membership and the batch activity completion checker. Behavior is preserved — the schedules register in the same order with the same arguments, the same conditionals gate them (premium, audit log, env vars, software store presence), and the `config` is threaded as a pointer so the `&config` and `config.Calendar` mutations inside the calendar closure keep their original semantics. `cmd/fleet/cron.go` (the schedule definitions) is intentionally untouched; only the wiring moved. One unit test added: `TestVulnerabilityProcessingDisabled` covers the vuln enable/disable predicate extracted into `vulnerabilityProcessingDisabled`, including the legacy `current_instance_checks` `"0"` value. The rest of the file is dependency-wiring relocation with no further decision logic to unit-test — those paths construct real schedules, so they stay covered by the existing suite and integration tests. The full `cmd/fleet` suite passes against MySQL + Redis, and a local server boot confirms the same 30 cron schedules start as before (verified against the "started cron schedules" log line). **Related issue:** Refs #33370 # Checklist for submitter - [x] Added/updated automated tests - [x] QA'd all new/changed functionality manually (verified via local server boot — same 30 cron schedules start) - Changes file: not applicable — internal refactor with no user-visible behavior change ## Summary by CodeRabbit * **Refactor** * Centralized background cron schedule startup and standardized job initialization sequencing for maintenance, vulnerability handling, integrations, MDM workflows, and premium tasks. * **New Features / Behavior** * Added config- and license-controlled enablement for vulnerability processing (local vs remote triggering), MDM automation (including APNs delivery and device reconciliation), and premium-only refresh/recovery behaviors. * Made chart data collection and optional activity streaming configurable, with safe fallbacks for scheduling periodicity. * **Tests** * Added coverage for vulnerability-schedule enable/disable decision logic. --- cmd/fleet/cron_registration.go | 357 +++++++++++++++++++++++++++ cmd/fleet/cron_registration_test.go | 46 ++++ cmd/fleet/serve.go | 364 ++-------------------------- server/config/config.go | 7 + 4 files changed, 436 insertions(+), 338 deletions(-) create mode 100644 cmd/fleet/cron_registration.go create mode 100644 cmd/fleet/cron_registration_test.go diff --git a/cmd/fleet/cron_registration.go b/cmd/fleet/cron_registration.go new file mode 100644 index 0000000000..07c66fe475 --- /dev/null +++ b/cmd/fleet/cron_registration.go @@ -0,0 +1,357 @@ +package main + +import ( + "context" + "fmt" + "log/slog" + "os" + "strings" + "time" + + activity_api "github.com/fleetdm/fleet/v4/server/activity/api" + chart_api "github.com/fleetdm/fleet/v4/server/chart/api" + "github.com/fleetdm/fleet/v4/server/config" + "github.com/fleetdm/fleet/v4/server/cron" + "github.com/fleetdm/fleet/v4/server/datastore/mysql" + "github.com/fleetdm/fleet/v4/server/fleet" + acme_api "github.com/fleetdm/fleet/v4/server/mdm/acme/api" + "github.com/fleetdm/fleet/v4/server/mdm/android" + apple_mdm "github.com/fleetdm/fleet/v4/server/mdm/apple" + "github.com/fleetdm/fleet/v4/server/mdm/apple/apple_apps" + "github.com/fleetdm/fleet/v4/server/mdm/apple/vpp" + "github.com/fleetdm/fleet/v4/server/service/redis_key_value" + "github.com/fleetdm/fleet/v4/server/service/schedule" +) + +// cronSchedulesDeps carries the runServeCmd-scoped dependencies that the cron +// schedule registrations close over. Registration is grouped by domain in +// startCronSchedules; the call order is identical to the previous inline +// sequence in runServeCmd. +type cronSchedulesDeps struct { + instanceID string + config *config.FleetConfig + license *fleet.LicenseInfo + logger *slog.Logger + cronSchedules *fleet.CronSchedules + ds fleet.Datastore + svc fleet.Service + carveStore fleet.CarveStore + enrollHostLimiter fleet.EnrollHostLimiter + liveQueryStore fleet.LiveQueryStore + failingPolicySet fleet.FailingPolicySet + redisPool fleet.RedisPool + commander *apple_mdm.MDMAppleCommander + depStorage *mysql.NanoDEPStorage + softwareInstallStore fleet.SoftwareInstallerStore + bootstrapPackageStore fleet.MDMBootstrapPackageStore + softwareTitleIconStore fleet.SoftwareTitleIconStore + androidSvc android.Service + activitySvc activity_api.Service + acmeSvc acme_api.Service + chartSvc chart_api.Service + auditLogger fleet.JSONLogger + distributedLock fleet.Lock + initFatal func(err error, msg string) +} + +// register starts a single cron schedule and routes a registration failure +// through initFatal with the given message. It removes the repetitive +// error-check boilerplate from each registration site. +func (deps cronSchedulesDeps) register(failMsg string, newSchedule func() (fleet.CronSchedule, error)) { + if err := deps.cronSchedules.StartCronSchedule(newSchedule); err != nil { + deps.initFatal(err, failMsg) + } +} + +// startCronSchedules registers every cron schedule the server runs, grouped by +// domain. The registration order is preserved exactly from the previous inline +// sequence in runServeCmd. +func startCronSchedules(ctx context.Context, deps cronSchedulesDeps) { + registerCleanupAndMaintenanceCrons(ctx, deps) + registerVulnerabilityCrons(ctx, deps) + registerWorkerCrons(ctx, deps) + registerMDMCrons(ctx, deps) + registerPremiumCrons(ctx, deps) + registerMiscCrons(ctx, deps) + + deps.logger.InfoContext(ctx, fmt.Sprintf("started cron schedules: %s", strings.Join(deps.cronSchedules.ScheduleNames(), ", "))) +} + +// registerCleanupAndMaintenanceCrons covers chart data collection, cron_stats +// cleanup, software migrations, frequent cleanups, the cleanups-then-aggregation +// schedule, query results cleanup, upcoming activities maintenance, usage +// statistics, and batch activities. +func registerCleanupAndMaintenanceCrons(ctx context.Context, deps cronSchedulesDeps) { + if os.Getenv("FLEET_SKIP_CHART_DATA_COLLECTION") == "" { + deps.register("failed to register chart_data_collection schedule", func() (fleet.CronSchedule, error) { + return newChartDataCollectionSchedule(ctx, deps.instanceID, deps.ds, deps.chartSvc, deps.logger) + }) + } else { + deps.logger.InfoContext(ctx, "skipping chart data collection cron (FLEET_SKIP_CHART_DATA_COLLECTION is set)") + } + + // Perform a cleanup of cron_stats outside of the cronSchedules because the + // schedule package uses cron_stats entries to decide whether a schedule will + // run or not (see https://github.com/fleetdm/fleet/issues/9486). + go func() { + cleanupCronStats := func() { + deps.logger.DebugContext(ctx, "cleaning up cron_stats") + // Datastore.CleanupCronStats should be safe to run by multiple fleet + // instances at the same time and it should not be an expensive operation. + if err := deps.ds.CleanupCronStats(ctx); err != nil { + deps.logger.InfoContext(ctx, "failed to clean up cron_stats", "err", err) + } + } + + cleanupCronStats() + + cleanUpCronStatsTick := time.NewTicker(1 * time.Hour) + defer cleanUpCronStatsTick.Stop() + for { + select { + case <-ctx.Done(): + return + case <-cleanUpCronStatsTick.C: + cleanupCronStats() + } + } + }() + + if deps.softwareInstallStore != nil { + deps.register(fmt.Sprintf("failed to register %s", fleet.CronUninstallSoftwareMigration), func() (fleet.CronSchedule, error) { + return cronUninstallSoftwareMigration(ctx, deps.instanceID, deps.ds, deps.softwareInstallStore, deps.logger) + }) + + deps.register(fmt.Sprintf("failed to register %s", fleet.CronUpgradeCodeSoftwareMigration), func() (fleet.CronSchedule, error) { + return cronUpgradeCodeSoftwareMigration(ctx, deps.instanceID, deps.ds, deps.softwareInstallStore, deps.logger) + }) + } + + if deps.config.Server.FrequentCleanupsEnabled { + deps.register("failed to register frequent_cleanups schedule", func() (fleet.CronSchedule, error) { + return newFrequentCleanupsSchedule(ctx, deps.instanceID, deps.ds, deps.liveQueryStore, deps.logger) + }) + } + + deps.register("failed to register cleanups_then_aggregations schedule", func() (fleet.CronSchedule, error) { + return newCleanupsAndAggregationSchedule( + ctx, deps.instanceID, deps.ds, deps.carveStore, deps.svc, deps.logger, deps.enrollHostLimiter, deps.config, deps.commander, deps.softwareInstallStore, deps.bootstrapPackageStore, deps.softwareTitleIconStore, deps.androidSvc, deps.activitySvc, deps.acmeSvc, deps.chartSvc, + ) + }) + + deps.register("failed to register query_results_cleanup schedule", func() (fleet.CronSchedule, error) { + return newQueryResultsCleanupSchedule(ctx, deps.instanceID, deps.ds, deps.liveQueryStore, deps.logger) + }) + + deps.register("failed to register upcoming_activities_maintenance schedule", func() (fleet.CronSchedule, error) { + return newUpcomingActivitiesSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) + + deps.register("failed to register stats schedule", func() (fleet.CronSchedule, error) { + return newUsageStatisticsSchedule(ctx, deps.instanceID, deps.ds, *deps.config, deps.logger) + }) + + deps.register("failed to register batch activities schedule", func() (fleet.CronSchedule, error) { + return newBatchActivitiesSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) +} + +// vulnerabilityProcessingDisabled reports whether this instance should skip +// running the vulnerabilities processing schedule. Processing is disabled +// either explicitly (disable_schedule) or when this instance opts out of +// current_instance_checks ("no" or the legacy "0"); in that case a remote +// trigger proxy is registered instead. +func vulnerabilityProcessingDisabled(cfg config.VulnerabilitiesConfig) bool { + return cfg.DisableSchedule || cfg.IsDisabledByInstanceCheck() +} + +// registerVulnerabilityCrons registers either the vulnerabilities processing +// schedule or, when processing is disabled on this instance, a remote trigger +// proxy so triggering still works when the schedule runs on a separate server. +func registerVulnerabilityCrons(ctx context.Context, deps cronSchedulesDeps) { + // Log the specific reason(s) processing is disabled, if any. + if deps.config.Vulnerabilities.DisableSchedule { + deps.logger.InfoContext(ctx, "vulnerabilities schedule disabled via vulnerabilities.disable_schedule") + } + if deps.config.Vulnerabilities.IsDisabledByInstanceCheck() { + deps.logger.InfoContext(ctx, "vulnerabilities schedule disabled via vulnerabilities.current_instance_checks") + } + if !vulnerabilityProcessingDisabled(deps.config.Vulnerabilities) { + // vuln processing by default is run by internal cron mechanism + deps.register("failed to register vulnerabilities schedule", func() (fleet.CronSchedule, error) { + return newVulnerabilitiesSchedule(ctx, deps.instanceID, deps.ds, deps.logger, &deps.config.Vulnerabilities) + }) + } else { + // Register a remote trigger proxy so triggering still works + // when the vulnerability schedule runs on a separate server. + deps.register("failed to register remote vulnerability trigger", func() (fleet.CronSchedule, error) { + return schedule.NewRemoteTriggerSchedule(string(fleet.CronVulnerabilities), deps.ds), nil + }) + } +} + +// registerWorkerCrons covers the automations schedule and the worker +// integrations schedule. +func registerWorkerCrons(ctx context.Context, deps cronSchedulesDeps) { + deps.register("failed to register automations schedule", func() (fleet.CronSchedule, error) { + return newAutomationsSchedule(ctx, deps.instanceID, deps.ds, deps.logger, 5*time.Minute, deps.failingPolicySet) + }) + + deps.register("failed to register worker integrations schedule", func() (fleet.CronSchedule, error) { + return newWorkerIntegrationsSchedule(ctx, deps.instanceID, deps.ds, deps.logger, deps.depStorage, deps.commander, deps.androidSvc, deps.chartSvc, deps.config.MDM.AndroidBatchSize) + }) +} + +// registerMDMCrons covers the Apple MDM worker, DEP profile assigner, service +// discovery, the Apple/Windows/Android profile managers, the Android device +// reconciler, the Android default-policy and per-host policy migrations, and +// the APNs pusher. +func registerMDMCrons(ctx context.Context, deps cronSchedulesDeps) { + deps.register("failed to register apple_mdm_worker schedule", func() (fleet.CronSchedule, error) { + vppInstaller := deps.svc.(fleet.AppleMDMVPPInstaller) + return newAppleMDMWorkerSchedule(ctx, deps.instanceID, deps.ds, deps.logger, deps.commander, deps.bootstrapPackageStore, vppInstaller, deps.svc.NewActivity) + }) + + deps.register("failed to register apple_mdm_dep_profile_assigner schedule", func() (fleet.CronSchedule, error) { + return newAppleMDMDEPProfileAssigner(ctx, deps.instanceID, deps.config.MDM.AppleDEPSyncPeriodicity, deps.ds, deps.depStorage, deps.logger) + }) + + deps.register("failed to register mdm_apple_service_discovery schedule", func() (fleet.CronSchedule, error) { + return newMDMAppleServiceDiscoverySchedule(ctx, deps.instanceID, deps.ds, deps.depStorage, deps.logger, deps.config.Server.URLPrefix) + }) + + deps.register("failed to register mdm_apple_profile_manager schedule", func() (fleet.CronSchedule, error) { + return newAppleMDMProfileManagerSchedule( + ctx, + deps.instanceID, + deps.ds, + deps.commander, + redis_key_value.New(deps.redisPool), + deps.logger, + deps.config.MDM.CertificateProfilesLimit, + ) + }) + + deps.register("failed to register mdm_windows_profile_manager schedule", func() (fleet.CronSchedule, error) { + return newWindowsMDMProfileManagerSchedule( + ctx, + deps.instanceID, + deps.ds, + deps.logger, + ) + }) + + deps.register("failed to register mdm_android_profile_manager schedule", func() (fleet.CronSchedule, error) { + return newAndroidMDMProfileManagerSchedule( + ctx, + deps.instanceID, + deps.ds, + deps.logger, + deps.config.License.Key, // NOTE: this requires the license key, not the parsed *LicenseInfo available in the ctx + deps.config.MDM.AndroidAgent, + deps.config.MDM.AndroidBatchSize, + ) + }) + + // Register Android MDM Device Reconciler schedule (same interval as Android profile manager) + deps.register("failed to register mdm_android_device_reconciler schedule", func() (fleet.CronSchedule, error) { + return newAndroidMDMDeviceReconcilerSchedule( + ctx, + deps.instanceID, + deps.ds, + deps.logger, + deps.config.License.Key, + deps.svc.NewActivity, + ) + }) + + deps.register("failed to register enable_android_app_reports_on_default_policy cron", func() (fleet.CronSchedule, error) { + return cronEnableAndroidAppReportsOnDefaultPolicy(ctx, deps.instanceID, deps.ds, deps.logger, deps.androidSvc) + }) + + deps.register("failed to register migrate_to_per_host_policy cron", func() (fleet.CronSchedule, error) { + return cronMigrateToPerHostPolicy(ctx, deps.instanceID, deps.ds, deps.logger, deps.androidSvc) + }) + + deps.register("failed to register APNs pusher schedule", func() (fleet.CronSchedule, error) { + return newMDMAPNsPusher( + ctx, + deps.instanceID, + deps.ds, + deps.commander, + deps.logger, + ) + }) +} + +// registerPremiumCrons covers the Fleet Premium schedules: iPhone/iPad +// refetcher and reviver, maintained apps, VPP app version refresh (and the +// one-shot VPP country backfill), recovery lock passwords, managed local +// account rotation, activities streaming, and the calendar schedule. +func registerPremiumCrons(ctx context.Context, deps cronSchedulesDeps) { + if !deps.license.IsPremium() { + return + } + + deps.register("failed to register apple_mdm_iphone_ipad_refetcher schedule", func() (fleet.CronSchedule, error) { + return newIPhoneIPadRefetcher(ctx, deps.instanceID, 10*time.Minute, deps.ds, deps.commander, deps.logger, deps.svc.NewActivity) + }) + + deps.register("failed to register apple_mdm_iphone_ipad_reviver schedule", func() (fleet.CronSchedule, error) { + return newIPhoneIPadReviver(ctx, deps.instanceID, deps.ds, deps.commander, deps.logger) + }) + + deps.register("failed to register maintained apps schedule", func() (fleet.CronSchedule, error) { + return newMaintainedAppSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) + + deps.register("failed to register refresh vpp app versions schedule", func() (fleet.CronSchedule, error) { + return newRefreshVPPAppVersionsSchedule(ctx, deps.instanceID, deps.ds, deps.logger, apple_apps.Configure(ctx, deps.ds, deps.config.License.Key, deps.config.MDM.AppleConnectJWT)) + }) + + // One-shot backfill for VPP token and app country codes that + // predate the country_code column. Fire-and-forget is safe because + // the work is idempotent and ctx cancels on shutdown. + go vpp.BackfillLegacyCountries(ctx, deps.ds, deps.logger) + + deps.register("failed to register recovery lock password schedule", func() (fleet.CronSchedule, error) { + return newRecoveryLockPasswordSchedule(ctx, deps.instanceID, deps.ds, deps.commander, deps.logger, deps.svc.NewActivity) + }) + + deps.register("failed to register managed local account rotation schedule", func() (fleet.CronSchedule, error) { + return newManagedLocalAccountRotationSchedule(ctx, deps.instanceID, deps.ds, deps.commander, deps.logger, deps.svc.NewActivity) + }) + + deps.register("failed to register cleanup expired ADUE challenges schedule", func() (fleet.CronSchedule, error) { + return newCleanupExpiredADUEChallengesSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) + + if deps.config.Activity.EnableAuditLog { + deps.register("failed to register activities streaming schedule", func() (fleet.CronSchedule, error) { + return newActivitiesStreamingSchedule(ctx, deps.instanceID, deps.activitySvc, deps.ds, deps.logger, deps.auditLogger) + }) + } + + deps.register("failed to register calendar schedule", func() (fleet.CronSchedule, error) { + if deps.config.Calendar.Periodicity > 0 { + deps.config.Calendar.SetAlwaysReloadEvent(true) + } else { + deps.config.Calendar.Periodicity = 5 * time.Minute + } + return cron.NewCalendarSchedule(ctx, deps.instanceID, deps.ds, deps.distributedLock, deps.config.Calendar, deps.logger) + }) +} + +// registerMiscCrons covers the host vitals label membership schedule and the +// batch activity completion checker. +func registerMiscCrons(ctx context.Context, deps cronSchedulesDeps) { + // Start the service that calculates and updates host vitals label membership. + deps.register("failed to register host vitals label membership schedule", func() (fleet.CronSchedule, error) { + return newHostVitalsLabelMembershipSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) + + // Start the service that marks activities as completed. + deps.register("failed to register batch activity completion checker schedule", func() (fleet.CronSchedule, error) { + return newBatchActivityCompletionCheckerSchedule(ctx, deps.instanceID, deps.ds, deps.logger) + }) +} diff --git a/cmd/fleet/cron_registration_test.go b/cmd/fleet/cron_registration_test.go new file mode 100644 index 0000000000..b474cd7c8d --- /dev/null +++ b/cmd/fleet/cron_registration_test.go @@ -0,0 +1,46 @@ +package main + +import ( + "testing" + + "github.com/fleetdm/fleet/v4/server/config" + "github.com/stretchr/testify/assert" +) + +func TestVulnerabilityProcessingDisabled(t *testing.T) { + for _, tc := range []struct { + name string + cfg config.VulnerabilitiesConfig + want bool + }{ + { + name: "enabled by default", + cfg: config.VulnerabilitiesConfig{CurrentInstanceChecks: "auto"}, + want: false, + }, + { + name: "disabled via disable_schedule", + cfg: config.VulnerabilitiesConfig{DisableSchedule: true, CurrentInstanceChecks: "auto"}, + want: true, + }, + { + name: "disabled via current_instance_checks no", + cfg: config.VulnerabilitiesConfig{CurrentInstanceChecks: "no"}, + want: true, + }, + { + name: "disabled via legacy current_instance_checks 0", + cfg: config.VulnerabilitiesConfig{CurrentInstanceChecks: "0"}, + want: true, + }, + { + name: "empty current_instance_checks does not disable", + cfg: config.VulnerabilitiesConfig{CurrentInstanceChecks: ""}, + want: false, + }, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, vulnerabilityProcessingDisabled(tc.cfg)) + }) + } +} diff --git a/cmd/fleet/serve.go b/cmd/fleet/serve.go index ab33bbd51a..aa6c92ee98 100644 --- a/cmd/fleet/serve.go +++ b/cmd/fleet/serve.go @@ -51,7 +51,6 @@ import ( "github.com/fleetdm/fleet/v4/server/contexts/ctxerr" "github.com/fleetdm/fleet/v4/server/contexts/installersize" licensectx "github.com/fleetdm/fleet/v4/server/contexts/license" - "github.com/fleetdm/fleet/v4/server/cron" "github.com/fleetdm/fleet/v4/server/datastore/failing" "github.com/fleetdm/fleet/v4/server/datastore/filesystem" "github.com/fleetdm/fleet/v4/server/datastore/mysql" @@ -69,8 +68,6 @@ import ( acme_bootstrap "github.com/fleetdm/fleet/v4/server/mdm/acme/bootstrap" android_service "github.com/fleetdm/fleet/v4/server/mdm/android/service" apple_mdm "github.com/fleetdm/fleet/v4/server/mdm/apple" - "github.com/fleetdm/fleet/v4/server/mdm/apple/apple_apps" - "github.com/fleetdm/fleet/v4/server/mdm/apple/vpp" "github.com/fleetdm/fleet/v4/server/mdm/cryptoutil" microsoft_mdm "github.com/fleetdm/fleet/v4/server/mdm/microsoft" "github.com/fleetdm/fleet/v4/server/mdm/nanomdm/push" @@ -90,7 +87,6 @@ import ( "github.com/fleetdm/fleet/v4/server/service/redis_key_value" "github.com/fleetdm/fleet/v4/server/service/redis_lock" "github.com/fleetdm/fleet/v4/server/service/redis_policy_set" - "github.com/fleetdm/fleet/v4/server/service/schedule" "github.com/fleetdm/fleet/v4/server/sso" "github.com/fleetdm/fleet/v4/server/version" "github.com/getsentry/sentry-go" @@ -624,45 +620,6 @@ func runServeCmd(cmd *cobra.Command, configManager configpkg.Manager, debug, dev // Bootstrap chart bounded context chartSvc, chartRoutes := createChartBoundedContext(dbConns, svc, logger) - if os.Getenv("FLEET_SKIP_CHART_DATA_COLLECTION") == "" { - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return newChartDataCollectionSchedule(ctx, instanceID, ds, chartSvc, logger) - }, - ); err != nil { - initFatal(err, "failed to register chart_data_collection schedule") - } - } else { - logger.InfoContext(ctx, "skipping chart data collection cron (FLEET_SKIP_CHART_DATA_COLLECTION is set)") - } - - // Perform a cleanup of cron_stats outside of the cronSchedules because the - // schedule package uses cron_stats entries to decide whether a schedule will - // run or not (see https://github.com/fleetdm/fleet/issues/9486). - go func() { - cleanupCronStats := func() { - logger.DebugContext(ctx, "cleaning up cron_stats") - // Datastore.CleanupCronStats should be safe to run by multiple fleet - // instances at the same time and it should not be an expensive operation. - if err := ds.CleanupCronStats(ctx); err != nil { - logger.InfoContext(ctx, "failed to clean up cron_stats", "err", err) - } - } - - cleanupCronStats() - - cleanUpCronStatsTick := time.NewTicker(1 * time.Hour) - defer cleanUpCronStatsTick.Stop() - for { - select { - case <-ctx.Done(): - return - case <-cleanUpCronStatsTick.C: - cleanupCronStats() - } - } - }() - // Trace sampler runtime control. The poller re-reads trace_sampler_settings every 60s and atomically swaps the sampler's // ratios and force_full so support can flip a 100% debug window via PATCH /debug/trace_sampler without restarting any // replicas. No-op when OTEL is disabled. @@ -670,301 +627,32 @@ func runServeCmd(cmd *cobra.Command, configManager configpkg.Manager, debug, dev go tracing.StartSettingsPoller(ctx, traceSampler, ds, logger) } - if softwareInstallStore != nil { - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return cronUninstallSoftwareMigration(ctx, instanceID, ds, softwareInstallStore, logger) - }, - ); err != nil { - initFatal(err, fmt.Sprintf("failed to register %s", fleet.CronUninstallSoftwareMigration)) - } - - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return cronUpgradeCodeSoftwareMigration(ctx, instanceID, ds, softwareInstallStore, logger) - }, - ); err != nil { - initFatal(err, fmt.Sprintf("failed to register %s", fleet.CronUpgradeCodeSoftwareMigration)) - } - } - - if config.Server.FrequentCleanupsEnabled { - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return newFrequentCleanupsSchedule(ctx, instanceID, ds, liveQueryStore, logger) - }, - ); err != nil { - initFatal(err, "failed to register frequent_cleanups schedule") - } - } - - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newCleanupsAndAggregationSchedule( - ctx, instanceID, ds, carveStore, svc, logger, redisWrapperDS, &config, commander, softwareInstallStore, bootstrapPackageStore, softwareTitleIconStore, androidSvc, activitySvc, acmeSvc, chartSvc, - ) - }, - ); err != nil { - initFatal(err, "failed to register cleanups_then_aggregations schedule") - } - - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return newQueryResultsCleanupSchedule(ctx, instanceID, ds, liveQueryStore, logger) - }, - ); err != nil { - initFatal(err, "failed to register query_results_cleanup schedule") - } - - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return newUpcomingActivitiesSchedule(ctx, instanceID, ds, logger) - }, - ); err != nil { - initFatal(err, "failed to register upcoming_activities_maintenance schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newUsageStatisticsSchedule(ctx, instanceID, ds, config, logger) - }); err != nil { - initFatal(err, "failed to register stats schedule") - } - - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - return newBatchActivitiesSchedule(ctx, instanceID, ds, logger) - }); err != nil { - initFatal(err, "failed to register batch activities schedule") - } - - vulnerabilityScheduleDisabled := false - if config.Vulnerabilities.DisableSchedule { - vulnerabilityScheduleDisabled = true - logger.InfoContext(ctx, "vulnerabilities schedule disabled via vulnerabilities.disable_schedule") - } - if config.Vulnerabilities.CurrentInstanceChecks == "no" || config.Vulnerabilities.CurrentInstanceChecks == "0" { - logger.InfoContext(ctx, "vulnerabilities schedule disabled via vulnerabilities.current_instance_checks") - vulnerabilityScheduleDisabled = true - } - if !vulnerabilityScheduleDisabled { - // vuln processing by default is run by internal cron mechanism - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newVulnerabilitiesSchedule(ctx, instanceID, ds, logger, &config.Vulnerabilities) - }); err != nil { - initFatal(err, "failed to register vulnerabilities schedule") - } - } else { - // Register a remote trigger proxy so triggering still works - // when the vulnerability schedule runs on a separate server. - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return schedule.NewRemoteTriggerSchedule(string(fleet.CronVulnerabilities), ds), nil - }); err != nil { - initFatal(err, "failed to register remote vulnerability trigger") - } - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newAutomationsSchedule(ctx, instanceID, ds, logger, 5*time.Minute, failingPolicySet) - }); err != nil { - initFatal(err, "failed to register automations schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newWorkerIntegrationsSchedule(ctx, instanceID, ds, logger, depStorage, commander, androidSvc, chartSvc, config.MDM.AndroidBatchSize) - }); err != nil { - initFatal(err, "failed to register worker integrations schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - vppInstaller := svc.(fleet.AppleMDMVPPInstaller) - return newAppleMDMWorkerSchedule(ctx, instanceID, ds, logger, commander, bootstrapPackageStore, vppInstaller, svc.NewActivity) - }); err != nil { - initFatal(err, "failed to register apple_mdm_worker schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newAppleMDMDEPProfileAssigner(ctx, instanceID, config.MDM.AppleDEPSyncPeriodicity, ds, depStorage, logger) - }); err != nil { - initFatal(err, "failed to register apple_mdm_dep_profile_assigner schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newMDMAppleServiceDiscoverySchedule(ctx, instanceID, ds, depStorage, logger, config.Server.URLPrefix) - }); err != nil { - initFatal(err, "failed to register mdm_apple_service_discovery schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newAppleMDMProfileManagerSchedule( - ctx, - instanceID, - ds, - apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService), - redis_key_value.New(redisPool), - logger, - config.MDM.CertificateProfilesLimit, - ) - }); err != nil { - initFatal(err, "failed to register mdm_apple_profile_manager schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newWindowsMDMProfileManagerSchedule( - ctx, - instanceID, - ds, - logger, - ) - }); err != nil { - initFatal(err, "failed to register mdm_windows_profile_manager schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newAndroidMDMProfileManagerSchedule( - ctx, - instanceID, - ds, - logger, - config.License.Key, // NOTE: this requires the license key, not the parsed *LicenseInfo available in the ctx - config.MDM.AndroidAgent, - config.MDM.AndroidBatchSize, - ) - }); err != nil { - initFatal(err, "failed to register mdm_android_profile_manager schedule") - } - - // Register Android MDM Device Reconciler schedule (same interval as Android profile manager) - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newAndroidMDMDeviceReconcilerSchedule( - ctx, - instanceID, - ds, - logger, - config.License.Key, - svc.NewActivity, - ) - }); err != nil { - initFatal(err, "failed to register mdm_android_device_reconciler schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return cronEnableAndroidAppReportsOnDefaultPolicy(ctx, instanceID, ds, logger, androidSvc) - }); err != nil { - initFatal(err, "failed to register enable_android_app_reports_on_default_policy cron") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return cronMigrateToPerHostPolicy(ctx, instanceID, ds, logger, androidSvc) - }); err != nil { - initFatal(err, "failed to register migrate_to_per_host_policy cron") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newMDMAPNsPusher( - ctx, - instanceID, - ds, - apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService), - logger, - ) - }); err != nil { - initFatal(err, "failed to register APNs pusher schedule") - } - - if license.IsPremium() { - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newIPhoneIPadRefetcher(ctx, instanceID, 10*time.Minute, ds, commander, logger, svc.NewActivity) - }); err != nil { - initFatal(err, "failed to register apple_mdm_iphone_ipad_refetcher schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newIPhoneIPadReviver(ctx, instanceID, ds, commander, logger) - }); err != nil { - initFatal(err, "failed to register apple_mdm_iphone_ipad_reviver schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newMaintainedAppSchedule(ctx, instanceID, ds, logger) - }); err != nil { - initFatal(err, "failed to register maintained apps schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newRefreshVPPAppVersionsSchedule(ctx, instanceID, ds, logger, apple_apps.Configure(ctx, ds, config.License.Key, config.MDM.AppleConnectJWT)) - }); err != nil { - initFatal(err, "failed to register refresh vpp app versions schedule") - } - - // One-shot backfill for VPP token and app country codes that - // predate the country_code column. Fire-and-forget is safe because - // the work is idempotent and ctx cancels on shutdown. - go vpp.BackfillLegacyCountries(ctx, ds, logger) - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newRecoveryLockPasswordSchedule(ctx, instanceID, ds, commander, logger, svc.NewActivity) - }); err != nil { - initFatal(err, "failed to register recovery lock password schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - commander := apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService) - return newManagedLocalAccountRotationSchedule(ctx, instanceID, ds, commander, logger, svc.NewActivity) - }); err != nil { - initFatal(err, "failed to register managed local account rotation schedule") - } - - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newCleanupExpiredADUEChallengesSchedule(ctx, instanceID, ds, logger) - }); err != nil { - initFatal(err, "failed to register cleanup expired ADUE challenges schedule") - } - } - - if license.IsPremium() && config.Activity.EnableAuditLog { - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newActivitiesStreamingSchedule(ctx, instanceID, activitySvc, ds, logger, auditLogger) - }); err != nil { - initFatal(err, "failed to register activities streaming schedule") - } - } - - if license.IsPremium() { - if err := cronSchedules.StartCronSchedule( - func() (fleet.CronSchedule, error) { - if config.Calendar.Periodicity > 0 { - config.Calendar.SetAlwaysReloadEvent(true) - } else { - config.Calendar.Periodicity = 5 * time.Minute - } - return cron.NewCalendarSchedule(ctx, instanceID, ds, distributedLock, config.Calendar, logger) - }, - ); err != nil { - initFatal(err, "failed to register calendar schedule") - } - } - - // Start the service that calculates and updates host vitals label membership. - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newHostVitalsLabelMembershipSchedule(ctx, instanceID, ds, logger) - }); err != nil { - initFatal(err, "failed to register host vitals label membership schedule") - } - - // Start the service that marks activities as completed. - if err := cronSchedules.StartCronSchedule(func() (fleet.CronSchedule, error) { - return newBatchActivityCompletionCheckerSchedule(ctx, instanceID, ds, logger) - }); err != nil { - initFatal(err, "failed to register batch activity completion checker schedule") - } - - logger.InfoContext(ctx, fmt.Sprintf("started cron schedules: %s", strings.Join(cronSchedules.ScheduleNames(), ", "))) + startCronSchedules(ctx, cronSchedulesDeps{ + instanceID: instanceID, + config: &config, + license: license, + logger: logger, + cronSchedules: cronSchedules, + ds: ds, + svc: svc, + carveStore: carveStore, + enrollHostLimiter: redisWrapperDS, + liveQueryStore: liveQueryStore, + failingPolicySet: failingPolicySet, + redisPool: redisPool, + commander: apple_mdm.NewMDMAppleCommander(mdmStorage, mdmPushService), + depStorage: depStorage, + softwareInstallStore: softwareInstallStore, + bootstrapPackageStore: bootstrapPackageStore, + softwareTitleIconStore: softwareTitleIconStore, + androidSvc: androidSvc, + activitySvc: activitySvc, + acmeSvc: acmeSvc, + chartSvc: chartSvc, + auditLogger: auditLogger, + distributedLock: distributedLock, + initFatal: initFatal, + }) // StartCollectors starts a goroutine per collector, using ctx to cancel. task.StartCollectors(ctx, logger.With("cron", "async_task")) diff --git a/server/config/config.go b/server/config/config.go index 8e85f9cba5..b4fd2ffc91 100644 --- a/server/config/config.go +++ b/server/config/config.go @@ -715,6 +715,13 @@ type VulnerabilitiesConfig struct { MaxConcurrency int `json:"max_concurrency" yaml:"max_concurrency"` } +// IsDisabledByInstanceCheck reports whether vulnerability processing is disabled +// on this instance via the current_instance_checks setting ("no", or the legacy +// "0"). It does not consider DisableSchedule, which disables processing globally. +func (v VulnerabilitiesConfig) IsDisabledByInstanceCheck() bool { + return v.CurrentInstanceChecks == "no" || v.CurrentInstanceChecks == "0" +} + // UpgradesConfig defines configs related to fleet server upgrades. type UpgradesConfig struct { AllowMissingMigrations bool `json:"allow_missing_migrations" yaml:"allow_missing_migrations"`