[Fleet] Add task for automatic agent upgrades (elastic#211019)

jillguyonnet · kibanamachine · elasticmachine · juliaElastic · commit e9908f237d03 · 2025-04-08T10:45:20.000+02:00
Closes elastic/ingest-dev#4722 - [x] Handle fetching agent policies and agents at scale - [x] Only consider active agents for upgrade - [x] Agents already on or upgrading to target version are included in the count but not considered for upgrade - [x] Agents stuck in updating are considered for upgrade - [x] Bulk upgrade actions triggered by the task have an added `isAutomatic:true` flag - [x] Use rollout duration to spread bulk upgrade in time (1h or longer depending on agent count) - This should be tested with real Elastic Agents (that will upgrade and have `upgrade_details`). - Edit the task interval in order to test how the task logic handles agents already upgrading. - Edit the agents batch size in order to test how the task logic handles agents at scale. - We should also check that space awareness is respected if enabled. - [x] [Unit or functional tests](https://www.elastic.co/guide/en/kibana/master/development-tests.html) were updated or added to match the most common scenarios - [ ] [Flaky Test Runner](https://ci-stats.kibana.dev/trigger_flaky_test_runner/1) was used on any tests changed - [x] The PR description includes the appropriate Release Notes section, and the correct `release_note:*` label is applied per the [guidelines](https://www.elastic.co/guide/en/kibana/master/contributing.html#kibana-release-notes-process) Risk of incorrectly triggering agent upgrades. Probability should be very low if the agent policy does not have `required_versions` set. --------- Co-authored-by: kibanamachine <42973632+kibanamachine@users.noreply.github.com> Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
diff --git a/packages/kbn-check-mappings-update-cli/current_fields.json b/packages/kbn-check-mappings-update-cli/current_fields.json
@@ -500,6 +500,7 @@
     "name",
     "namespace",
     "overrides",
+    "required_versions",
     "revision",
     "schema_version",
     "status",
@@ -624,6 +625,7 @@
     "name",
     "namespace",
     "overrides",
+    "required_versions",
     "revision",
     "schema_version",
     "status",
diff --git a/packages/kbn-check-mappings-update-cli/current_mappings.json b/packages/kbn-check-mappings-update-cli/current_mappings.json
@@ -1685,6 +1685,10 @@
         "index": false,
         "type": "flattened"
       },
+      "required_versions": {
+        "index": false,
+        "type": "flattened"
+      },
       "revision": {
         "type": "integer"
       },
@@ -2071,6 +2075,10 @@
         "index": false,
         "type": "flattened"
       },
+      "required_versions": {
+        "index": false,
+        "type": "flattened"
+      },
       "revision": {
         "type": "integer"
       },
diff --git a/src/core/server/integration_tests/ci_checks/saved_objects/check_registered_types.test.ts b/src/core/server/integration_tests/ci_checks/saved_objects/check_registered_types.test.ts
@@ -105,7 +105,7 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "file": "6b65ae5899b60ebe08656fd163ea532e557d3c98",
         "file-upload-usage-collection-telemetry": "06e0a8c04f991e744e09d03ab2bd7f86b2088200",
         "fileShare": "5be52de1747d249a221b5241af2838264e19aaa1",
-        "fleet-agent-policies": "4a5c6477d2a61121e95ea9865ed1403a28c38706",
+        "fleet-agent-policies": "f69f7c5639f4cf9e85077c904e161f3574ac3ca2",
         "fleet-fleet-server-host": "69be15f6b6f2a2875ad3c7050ddea7a87f505417",
         "fleet-message-signing-keys": "93421f43fed2526b59092a4e3c65d64bc2266c0f",
         "fleet-package-policies": "8173220091e28ff4afa8238bb37749599378f9e5",
@@ -121,7 +121,7 @@ describe('checking migration metadata changes on all registered SO types', () =>
         "infra-custom-dashboards": "1a5994f2e05bb8a1609825ddbf5012f77c5c67f3",
         "infrastructure-monitoring-log-view": "5f86709d3c27aed7a8379153b08ee5d3d90d77f5",
         "infrastructure-ui-source": "113182d6895764378dfe7fa9fa027244f3a457c4",
-        "ingest-agent-policies": "57ebfb047cf0b81c6fa0ceed8586fa7199c7c5e2",
+        "ingest-agent-policies": "cfe66f4aeca8f53b26bd4ddb0e956de1637d774e",
         "ingest-download-sources": "279a68147e62e4d8858c09ad1cf03bd5551ce58d",
         "ingest-outputs": "daafff49255ab700e07491376fe89f04fc998b91",
         "ingest-package-policies": "870f8c21fe3602f31075430a1fdfb052c62d4a14",
diff --git a/x-pack/platform/plugins/shared/fleet/common/constants/agent.ts b/x-pack/platform/plugins/shared/fleet/common/constants/agent.ts
@@ -48,6 +48,16 @@ export const AgentStatuses = [
   'degraded',
 ] as const;
 
+export const ActiveAgentStatuses = [
+  'online',
+  'offline',
+  'enrolling',
+  'updating',
+  'degraded',
+  'error',
+  'orphaned',
+]; // excluded: unenrolling, unenrolled, inactive, uninstalled
+
 // Kueries for finding unprivileged and privileged agents
 // Privileged is `not` because the metadata field can be undefined
 export const UNPRIVILEGED_AGENT_KUERY = `${AGENTS_PREFIX}.local_metadata.elastic.agent.unprivileged: true`;
diff --git a/x-pack/platform/plugins/shared/fleet/common/services/agent_status.ts b/x-pack/platform/plugins/shared/fleet/common/services/agent_status.ts
@@ -5,6 +5,7 @@
  * 2.0.
  */
 
+import { ActiveAgentStatuses } from '../constants';
 import type { Agent, AgentStatus, FleetServerAgent } from '../types';
 
 export function getPreviousAgentStatusForOfflineAgents(
@@ -57,6 +58,10 @@ export function buildKueryForInactiveAgents() {
   return 'status:inactive';
 }
 
+export function buildKueryForActiveAgents() {
+  return `(${ActiveAgentStatuses.map((s) => `status:${s}`).join(' or ')})`;
+}
+
 export const AGENT_UPDATING_TIMEOUT_HOURS = 2;
 
 export function isStuckInUpdating(agent: Agent): boolean {
diff --git a/x-pack/platform/plugins/shared/fleet/common/types/models/agent.ts b/x-pack/platform/plugins/shared/fleet/common/types/models/agent.ts
@@ -70,6 +70,7 @@ export interface NewAgentAction {
   rollout_duration_seconds?: number;
   source_uri?: string;
   total?: number;
+  is_automatic?: boolean;
 }
 
 export interface AgentAction extends NewAgentAction {
@@ -444,6 +445,11 @@ export interface FleetServerAgentAction {
     signature: string;
   };
 
+  /**
+   * True if action was generated by an automated task.
+   */
+  is_automatic?: boolean;
+
   [k: string]: unknown;
 }
 
diff --git a/x-pack/platform/plugins/shared/fleet/server/mocks/index.ts b/x-pack/platform/plugins/shared/fleet/server/mocks/index.ts
@@ -143,6 +143,7 @@ export const createAppContextStartContractMock = (
     unenrollInactiveAgentsTask: {} as any,
     deleteUnenrolledAgentsTask: {} as any,
     updateAgentlessDeploymentsTask: {} as any,
+    automaticAgentUpgradeTask: {} as any,
   };
 };
 
diff --git a/x-pack/platform/plugins/shared/fleet/server/plugin.ts b/x-pack/platform/plugins/shared/fleet/server/plugin.ts
@@ -147,6 +147,7 @@ import { registerDeployAgentPoliciesTask } from './services/agent_policies/deplo
 import { DeleteUnenrolledAgentsTask } from './tasks/delete_unenrolled_agents_task';
 import { registerBumpAgentPoliciesTask } from './services/agent_policies/bump_agent_policies_task';
 import { UpgradeAgentlessDeploymentsTask } from './tasks/upgrade_agentless_deployment';
+import { AutomaticAgentUpgradeTask } from './tasks/automatic_agent_upgrade_task';
 
 export interface FleetSetupDeps {
   security: SecurityPluginSetup;
@@ -199,6 +200,7 @@ export interface FleetAppContext {
   unenrollInactiveAgentsTask: UnenrollInactiveAgentsTask;
   deleteUnenrolledAgentsTask: DeleteUnenrolledAgentsTask;
   updateAgentlessDeploymentsTask: UpgradeAgentlessDeploymentsTask;
+  automaticAgentUpgradeTask: AutomaticAgentUpgradeTask;
   taskManagerStart?: TaskManagerStartContract;
 }
 
@@ -299,6 +301,7 @@ export class FleetPlugin
   private unenrollInactiveAgentsTask?: UnenrollInactiveAgentsTask;
   private deleteUnenrolledAgentsTask?: DeleteUnenrolledAgentsTask;
   private updateAgentlessDeploymentsTask?: UpgradeAgentlessDeploymentsTask;
+  private automaticAgentUpgradeTask?: AutomaticAgentUpgradeTask;
 
   private agentService?: AgentService;
   private packageService?: PackageService;
@@ -628,7 +631,7 @@ export class FleetPlugin
     registerRoutes(fleetAuthzRouter, config);
 
     this.telemetryEventsSender.setup(deps.telemetry);
-    // Register task
+    // Register tasks
     registerUpgradeManagedPackagePoliciesTask(deps.taskManager);
     registerDeployAgentPoliciesTask(deps.taskManager);
     registerBumpAgentPoliciesTask(deps.taskManager);
@@ -654,6 +657,11 @@ export class FleetPlugin
       taskManager: deps.taskManager,
       logFactory: this.initializerContext.logger,
     });
+    this.automaticAgentUpgradeTask = new AutomaticAgentUpgradeTask({
+      core,
+      taskManager: deps.taskManager,
+      logFactory: this.initializerContext.logger,
+    });
 
     // Register fields metadata extractors
     registerFieldsMetadataExtractors({ core, fieldsMetadata: deps.fieldsMetadata });
@@ -702,6 +710,7 @@ export class FleetPlugin
       unenrollInactiveAgentsTask: this.unenrollInactiveAgentsTask!,
       deleteUnenrolledAgentsTask: this.deleteUnenrolledAgentsTask!,
       updateAgentlessDeploymentsTask: this.updateAgentlessDeploymentsTask!,
+      automaticAgentUpgradeTask: this.automaticAgentUpgradeTask!,
       taskManagerStart: plugins.taskManager,
     });
     licenseService.start(plugins.licensing.license$);
@@ -714,7 +723,7 @@ export class FleetPlugin
     this.updateAgentlessDeploymentsTask
       ?.start({ taskManager: plugins.taskManager })
       .catch(() => {});
-
+    this.automaticAgentUpgradeTask?.start({ taskManager: plugins.taskManager }).catch(() => {});
     startFleetUsageLogger(plugins.taskManager).catch(() => {});
     this.fleetMetricsTask
       ?.start(plugins.taskManager, core.elasticsearch.client.asInternalUser)
diff --git a/x-pack/platform/plugins/shared/fleet/server/saved_objects/index.ts b/x-pack/platform/plugins/shared/fleet/server/saved_objects/index.ts
@@ -254,6 +254,7 @@ export const getSavedObjectTypes = (
           monitoring_pprof_enabled: { type: 'boolean', index: false },
           monitoring_http: { type: 'flattened', index: false },
           monitoring_diagnostics: { type: 'flattened', index: false },
+          required_versions: { type: 'flattened', index: false },
         },
       },
       migrations: {
@@ -331,6 +332,16 @@ export const getSavedObjectTypes = (
             },
           ],
         },
+        '7': {
+          changes: [
+            {
+              type: 'mappings_addition',
+              addedMappings: {
+                required_versions: { type: 'flattened', index: false },
+              },
+            },
+          ],
+        },
       },
     },
     [AGENT_POLICY_SAVED_OBJECT_TYPE]: {
@@ -379,6 +390,7 @@ export const getSavedObjectTypes = (
             dynamic: false,
             properties: {},
           },
+          required_versions: { type: 'flattened', index: false },
         },
       },
       modelVersions: {
@@ -390,6 +402,16 @@ export const getSavedObjectTypes = (
             },
           ],
         },
+        '2': {
+          changes: [
+            {
+              type: 'mappings_addition',
+              addedMappings: {
+                required_versions: { type: 'flattened', index: false },
+              },
+            },
+          ],
+        },
       },
     },
     [OUTPUT_SAVED_OBJECT_TYPE]: {
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agents/actions.ts b/x-pack/platform/plugins/shared/fleet/server/services/agents/actions.ts
@@ -64,6 +64,7 @@ export async function createAgentAction(
     rollout_duration_seconds: newAgentAction.rollout_duration_seconds,
     total: newAgentAction.total,
     traceparent: apm.currentTraceparent,
+    is_automatic: newAgentAction.is_automatic,
   };
 
   const messageSigningService = appContextService.getMessageSigningService();
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agents/crud.test.ts b/x-pack/platform/plugins/shared/fleet/server/services/agents/crud.test.ts
@@ -29,6 +29,7 @@ import {
   updateAgent,
   _joinFilters,
   getByIds,
+  fetchAllAgentsByKuery,
 } from './crud';
 
 jest.mock('../audit_logging');
@@ -570,4 +571,42 @@ describe('Agents CRUD test', () => {
       );
     });
   });
+
+  describe('fetchAllAgentsByKuery', () => {
+    const createEsSearchResultMock = (ids: string[]) => {
+      const mock = getEsResponse(ids, ids.length, 'online');
+      return {
+        ...mock,
+        hits: {
+          ...mock.hits,
+          hits: mock.hits.hits.map((item) => ({ ...item, sort: ['enrolled_at'] })),
+        },
+      };
+    };
+
+    it('should return an iterator', async () => {
+      expect(await fetchAllAgentsByKuery(esClientMock, soClientMock, {})).toEqual({
+        [Symbol.asyncIterator]: expect.any(Function),
+      });
+    });
+
+    it('should provide agents on every iteration', async () => {
+      const agentIds = [
+        ['1', '2', '3'],
+        ['4', '5', '6'],
+      ];
+      searchMock
+        .mockResolvedValueOnce(createEsSearchResultMock(agentIds[0]))
+        .mockResolvedValueOnce(createEsSearchResultMock(agentIds[1]))
+        .mockResolvedValueOnce(createEsSearchResultMock([]));
+
+      let testCounter = 0;
+      for await (const agents of await fetchAllAgentsByKuery(esClientMock, soClientMock, {})) {
+        expect(agents.map((agent) => agent.id)).toEqual(agentIds[testCounter]);
+        testCounter++;
+      }
+
+      expect(searchMock).toHaveBeenCalledTimes(3);
+    });
+  });
 });
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agents/crud.ts b/x-pack/platform/plugins/shared/fleet/server/services/agents/crud.ts
@@ -32,6 +32,7 @@ import { getCurrentNamespace } from '../spaces/get_current_namespace';
 import { isSpaceAwarenessEnabled } from '../spaces/helpers';
 import { isAgentInNamespace } from '../spaces/agent_namespaces';
 import { addNamespaceFilteringToQuery } from '../spaces/query_namespaces_filtering';
+import { createEsSearchIterable } from '../utils/create_es_search_iterable';
 
 import { searchHitToAgent, agentSOAttributesToFleetServerAgentDoc } from './helpers';
 import { buildAgentStatusRuntimeField } from './build_status_runtime_field';
@@ -229,16 +230,7 @@ export async function getAgentsByKuery(
     aggregations,
     spaceId,
   } = options;
-  const filters = [];
-
-  const useSpaceAwareness = await isSpaceAwarenessEnabled();
-  if (useSpaceAwareness && spaceId) {
-    if (spaceId === DEFAULT_SPACE_ID) {
-      filters.push(`namespaces:"${DEFAULT_SPACE_ID}" or not namespaces:*`);
-    } else {
-      filters.push(`namespaces:"${spaceId}"`);
-    }
-  }
+  const filters = await _getSpaceAwarenessFilter(spaceId);
 
   if (kuery && kuery !== '') {
     filters.push(kuery);
@@ -394,6 +386,59 @@ export async function getAllAgentsByKuery(
   };
 }
 
+/**
+ * Fetch all agents by kuery in batches.
+ * @param esClient
+ * @param soClient
+ * @param options
+ */
+export async function fetchAllAgentsByKuery(
+  esClient: ElasticsearchClient,
+  soClient: SavedObjectsClientContract,
+  options: ListWithKuery & { spaceId?: string }
+): Promise<AsyncIterable<Agent[]>> {
+  const {
+    kuery = '',
+    perPage = SO_SEARCH_LIMIT,
+    sortField = 'enrolled_at',
+    sortOrder = 'desc',
+    spaceId,
+  } = options;
+
+  const filters = await _getSpaceAwarenessFilter(spaceId);
+  if (kuery && kuery !== '') {
+    filters.push(kuery);
+  }
+  const kueryNode = _joinFilters(filters);
+  const query = kueryNode ? { query: toElasticsearchQuery(kueryNode) } : {};
+  const runtimeFields = await buildAgentStatusRuntimeField(soClient);
+  const sort = getSortConfig(sortField, sortOrder);
+
+  try {
+    return createEsSearchIterable<FleetServerAgent>({
+      esClient,
+      searchRequest: {
+        index: AGENTS_INDEX,
+        size: perPage,
+        rest_total_hits_as_int: true,
+        track_total_hits: true,
+        runtime_mappings: runtimeFields,
+        fields: Object.keys(runtimeFields),
+        sort,
+        ...query,
+      },
+      resultsMapper: (data): Agent[] => {
+        return data.hits.hits.map(searchHitToAgent);
+      },
+    });
+  } catch (err) {
+    appContextService
+      .getLogger()
+      .error(`Error fetching all agents by kuery: ${JSON.stringify(err)}`);
+    throw err;
+  }
+}
+
 export async function getAgentById(
   esClient: ElasticsearchClient,
   soClient: SavedObjectsClientContract,
@@ -718,3 +763,15 @@ export async function getAgentPolicyForAgent(
     return agentPolicy;
   }
 }
+
+async function _getSpaceAwarenessFilter(spaceId: string | undefined) {
+  const useSpaceAwareness = await isSpaceAwarenessEnabled();
+  if (!useSpaceAwareness || !spaceId) {
+    return [];
+  }
+  if (spaceId === DEFAULT_SPACE_ID) {
+    return [`namespaces:"${DEFAULT_SPACE_ID}" or not namespaces:*`];
+  } else {
+    return [`namespaces:"${spaceId}"`];
+  }
+}
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agents/upgrade.ts b/x-pack/platform/plugins/shared/fleet/server/services/agents/upgrade.ts
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/agents/upgrade_action_runner.ts b/x-pack/platform/plugins/shared/fleet/server/services/agents/upgrade_action_runner.ts
diff --git a/x-pack/platform/plugins/shared/fleet/server/tasks/automatic_agent_upgrade_task.test.ts b/x-pack/platform/plugins/shared/fleet/server/tasks/automatic_agent_upgrade_task.test.ts
diff --git a/x-pack/platform/plugins/shared/fleet/server/tasks/automatic_agent_upgrade_task.ts b/x-pack/platform/plugins/shared/fleet/server/tasks/automatic_agent_upgrade_task.ts
diff --git a/x-pack/platform/plugins/shared/fleet/server/tasks/unenroll_inactive_agents_task.test.ts b/x-pack/platform/plugins/shared/fleet/server/tasks/unenroll_inactive_agents_task.test.ts
diff --git a/x-pack/test/plugin_api_integration/test_suites/task_manager/check_registered_task_types.ts b/x-pack/test/plugin_api_integration/test_suites/task_manager/check_registered_task_types.ts
