[Fleet] Allow to specify an upgrade window for rolling upgrade

x-pack/plugins/fleet/common/openapi/bundled.json

@@ -3795,6 +3795,9 @@
           "source_uri": {
             "type": "string"
           },
+          "rollout_duration_seconds": {
+            "type": "number"
+          },
           "agents": {
             "oneOf": [
               {

x-pack/plugins/fleet/common/openapi/bundled.yaml

@@ -2391,6 +2391,8 @@ components:
           type: string
         source_uri:
           type: string
+        rollout_duration_seconds:
+          type: number
         agents:
           oneOf:
             - type: array

x-pack/plugins/fleet/common/openapi/components/schemas/bulk_upgrade_agents.yaml

@@ -5,6 +5,8 @@ properties:
     type: string
   source_uri:
     type: string
+  rollout_duration_seconds:
+    type: number
   agents:
     oneOf:
       - type: array

x-pack/plugins/fleet/common/types/models/agent.ts

@@ -44,6 +44,9 @@ export interface NewAgentAction {
   agents: string[];
   created_at?: string;
   id?: string;
+  expiration?: string;
+  start_time?: string;
+  minimum_execution_duration?: number;
 }
 
 export interface AgentAction extends NewAgentAction {

x-pack/plugins/fleet/server/routes/agent/upgrade_handler.ts

@@ -81,7 +81,13 @@ export const postBulkAgentsUpgradeHandler: RequestHandler<
   const coreContext = await context.core;
   const soClient = coreContext.savedObjects.client;
   const esClient = coreContext.elasticsearch.client.asInternalUser;
-  const { version, source_uri: sourceUri, agents, force } = request.body;
+  const {
+    version,
+    source_uri: sourceUri,
+    agents,
+    force,
+    rollout_duration_seconds: upgradeDurationSeconds,
+  } = request.body;
   const kibanaVersion = appContextService.getKibanaVersion();
   try {
     checkVersionIsSame(version, kibanaVersion);
@@ -102,6 +108,7 @@ export const postBulkAgentsUpgradeHandler: RequestHandler<
       sourceUri,
       version,
       force,
+      upgradeDurationSeconds,
     };
     const results = await AgentService.sendUpgradeAgentsActions(soClient, esClient, upgradeOptions);
     const body = results.items.reduce<PostBulkAgentUpgradeResponse>((acc, so) => {

x-pack/plugins/fleet/server/services/agents/actions.ts

@@ -26,11 +26,13 @@ export async function createAgentAction(
   const timestamp = new Date().toISOString();
   const body: FleetServerAgentAction = {
     '@timestamp': timestamp,
-    expiration: new Date(Date.now() + ONE_MONTH_IN_MS).toISOString(),
+    expiration: newAgentAction.expiration ?? new Date(Date.now() + ONE_MONTH_IN_MS).toISOString(),
     agents: newAgentAction.agents,
     action_id: id,
     data: newAgentAction.data,
     type: newAgentAction.type,
+    start_time: newAgentAction.start_time,
+    minimum_execution_duration: newAgentAction.minimum_execution_duration,
   };
 
   await esClient.create({
@@ -49,26 +51,28 @@ export async function createAgentAction(
 
 export async function bulkCreateAgentActions(
   esClient: ElasticsearchClient,
-  newAgentActions: Array<Omit<AgentAction, 'id'>>
+  newAgentActions: NewAgentAction[]
 ): Promise<AgentAction[]> {
   const actions = newAgentActions.map((newAgentAction) => {
-    const id = uuid.v4();
+    const id = newAgentAction.id ?? uuid.v4();
     return {
       id,
       ...newAgentAction,
-    };
+    } as AgentAction;
   });
 
   if (actions.length === 0) {
-    return actions;
+    return [];
   }
 
   await esClient.bulk({
     index: AGENT_ACTIONS_INDEX,
     body: actions.flatMap((action) => {
       const body: FleetServerAgentAction = {
         '@timestamp': new Date().toISOString(),
-        expiration: new Date(Date.now() + ONE_MONTH_IN_MS).toISOString(),
+        expiration: action.expiration ?? new Date(Date.now() + ONE_MONTH_IN_MS).toISOString(),
+        start_time: action.start_time,
+        minimum_execution_duration: action.minimum_execution_duration,
         agents: action.agents,
         action_id: action.id,
         data: action.data,

x-pack/plugins/fleet/server/services/agents/upgrade.ts

@@ -6,6 +6,7 @@
  */
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';
+import moment from 'moment';
 
 import type { Agent, BulkActionResult } from '../../types';
 import { agentPolicyService } from '..';
@@ -28,6 +29,8 @@ import {
 } from './crud';
 import { searchHitToAgent } from './helpers';
 
+const MINIMUM_EXECUTION_DURATION_SECONDS = 1800; // 30m
+
 function isMgetDoc(doc?: estypes.MgetResponseItem<unknown>): doc is estypes.GetGetResult {
   return Boolean(doc && 'found' in doc);
 }
@@ -78,6 +81,7 @@ export async function sendUpgradeAgentsActions(
     version: string;
     sourceUri?: string | undefined;
     force?: boolean;
+    upgradeDurationSeconds?: number;
   }
 ) {
   // Full set of agents
@@ -158,12 +162,21 @@ export async function sendUpgradeAgentsActions(
     source_uri: options.sourceUri,
   };
 
+  const rollingUpgradeOptions = options?.upgradeDurationSeconds
+    ? {
+        start_time: now,
+        minimum_execution_duration: MINIMUM_EXECUTION_DURATION_SECONDS,
+        expiration: moment().add(options?.upgradeDurationSeconds, 'seconds').toISOString(),
+      }
+    : {};
+
   await createAgentAction(esClient, {
     created_at: now,
     data,
     ack_data: data,
     type: 'UPGRADE',
     agents: agentsToUpdate.map((agent) => agent.id),
+    ...rollingUpgradeOptions,
   });
 
   await bulkUpdateAgents(

x-pack/plugins/fleet/server/types/rest_spec/agent.ts

@@ -71,6 +71,7 @@ export const PostBulkAgentUpgradeRequestSchema = {
     source_uri: schema.maybe(schema.string()),
     version: schema.string(),
     force: schema.maybe(schema.boolean()),
+    rollout_duration_seconds: schema.maybe(schema.number({ min: 600 })),
   }),
 };
 

x-pack/test/fleet_api_integration/apis/agents/upgrade.ts

@@ -325,6 +325,57 @@ export default function (providerContext: FtrProviderContext) {
         expect(typeof agent2data.body.item.upgrade_started_at).to.be('undefined');
       });
 
+      it('should create a .fleet-actions document with the agents, version, and upgrade window', async () => {
+        const kibanaVersion = await kibanaServer.version.get();
+        await es.update({
+          id: 'agent1',
+          refresh: 'wait_for',
+          index: AGENTS_INDEX,
+          body: {
+            doc: {
+              local_metadata: { elastic: { agent: { upgradeable: true, version: '0.0.0' } } },
+            },
+          },
+        });
+        await es.update({
+          id: 'agent2',
+          refresh: 'wait_for',
+          index: AGENTS_INDEX,
+          body: {
+            doc: {
+              local_metadata: { elastic: { agent: { upgradeable: true, version: '0.0.0' } } },
+            },
+          },
+        });
+        await supertest
+          .post(`/api/fleet/agents/bulk_upgrade`)
+          .set('kbn-xsrf', 'xxx')
+          .send({
+            version: kibanaVersion,
+            agents: ['agent1', 'agent2'],
+            rollout_duration_seconds: 6000,
+          })
+          .expect(200);
+
+        const actionsRes = await es.search({
+          index: '.fleet-actions',
+          body: {
+            sort: [{ '@timestamp': { order: 'desc' } }],
+          },
+        });
+
+        const action: any = actionsRes.hits.hits[0]._source;
+
+        expect(action).to.have.keys(
+          'agents',
+          'expiration',
+          'start_time',
+          'minimum_execution_duration'
+        );
+        expect(action.agents).contain('agent1');
+        expect(action.agents).contain('agent2');
+      });
+
       it('should allow to upgrade multiple upgradeable agents by kuery', async () => {
         const kibanaVersion = await kibanaServer.version.get();
         await es.update({