[Fleet] Add new API to get current upgrades

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

@@ -93,6 +93,13 @@ export interface AgentSOAttributes extends AgentBase {
   packages?: string[];
 }
 
+export interface CurrentUpgrade {
+  actionId: string;
+  complete: boolean;
+  nbAgents: number;
+  nbAgentsAck: number;
+}
+
 // Generated from FleetServer schema.json
 
 /**

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

@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import type { Agent, AgentAction, NewAgentAction } from '../models';
+import type { Agent, AgentAction, CurrentUpgrade, NewAgentAction } from '../models';
 
 import type { ListResult, ListWithKuery } from './common';
 
@@ -174,3 +174,7 @@ export interface IncomingDataList {
 export interface GetAgentIncomingDataResponse {
   items: IncomingDataList[];
 }
+
+export interface GetCurrentUpgradesResponse {
+  items: CurrentUpgrade[];
+}

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

@@ -41,7 +41,11 @@ import {
   postCancelActionHandlerBuilder,
 } from './actions_handlers';
 import { postAgentUnenrollHandler, postBulkAgentsUnenrollHandler } from './unenroll_handler';
-import { postAgentUpgradeHandler, postBulkAgentsUpgradeHandler } from './upgrade_handler';
+import {
+  getCurrentUpgradesHandler,
+  postAgentUpgradeHandler,
+  postBulkAgentsUpgradeHandler,
+} from './upgrade_handler';
 
 export const registerAPIRoutes = (router: FleetAuthzRouter, config: FleetConfigType) => {
   // Get one
@@ -197,6 +201,18 @@ export const registerAPIRoutes = (router: FleetAuthzRouter, config: FleetConfigT
     },
     postBulkAgentsUpgradeHandler
   );
+  // Current upgrades
+  router.get(
+    {
+      path: AGENT_API_ROUTES.CURRENT_UPGRADES_PATTERN,
+      validate: false,
+      fleetAuthz: {
+        fleet: { all: true },
+      },
+    },
+    getCurrentUpgradesHandler
+  );
+
   // Bulk reassign
   router.post(
     {

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

@@ -12,7 +12,11 @@ import type { SavedObjectsClientContract, ElasticsearchClient } from '@kbn/core/
 import semverCoerce from 'semver/functions/coerce';
 import semverGt from 'semver/functions/gt';
 
-import type { PostAgentUpgradeResponse, PostBulkAgentUpgradeResponse } from '../../../common/types';
+import type {
+  PostAgentUpgradeResponse,
+  PostBulkAgentUpgradeResponse,
+  GetCurrentUpgradesResponse,
+} from '../../../common/types';
 import type { PostAgentUpgradeRequestSchema, PostBulkAgentUpgradeRequestSchema } from '../../types';
 import * as AgentService from '../../services/agents';
 import { appContextService } from '../../services';
@@ -135,6 +139,19 @@ export const postBulkAgentsUpgradeHandler: RequestHandler<
   }
 };
 
+export const getCurrentUpgradesHandler: RequestHandler = async (context, request, response) => {
+  const coreContext = await context.core;
+  const esClient = coreContext.elasticsearch.client.asInternalUser;
+
+  try {
+    const upgrades = await AgentService.getCurrentBulkUpgrades(esClient);
+    const body: GetCurrentUpgradesResponse = { items: upgrades };
+    return response.ok({ body });
+  } catch (error) {
+    return defaultIngestErrorHandler({ error, response });
+  }
+};
+
 export const checkKibanaVersion = (version: string, kibanaVersion: string) => {
   // get version number only in case "-SNAPSHOT" is in it
   const kibanaVersionNumber = semverCoerce(kibanaVersion)?.version;

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

@@ -7,8 +7,9 @@
 import type * as estypes from '@elastic/elasticsearch/lib/api/typesWithBodyKey';
 import type { ElasticsearchClient, SavedObjectsClientContract } from '@kbn/core/server';
 import moment from 'moment';
+import pMap from 'p-map';
 
-import type { Agent, BulkActionResult } from '../../types';
+import type { Agent, BulkActionResult, FleetServerAgentAction, CurrentUpgrade } from '../../types';
 import { agentPolicyService } from '..';
 import {
   AgentReassignmentError,
@@ -17,6 +18,7 @@ import {
 } from '../../errors';
 import { isAgentUpgradeable } from '../../../common/services';
 import { appContextService } from '../app_context';
+import { AGENT_ACTIONS_INDEX, AGENT_ACTIONS_RESULTS_INDEX } from '../../../common';
 
 import { createAgentAction } from './actions';
 import type { GetAgentsOptions } from './crud';
@@ -207,3 +209,134 @@ export async function sendUpgradeAgentsActions(
 
   return { items: orderedOut };
 }
+
+/**
+ * Return current bulk upgrades (non completed or cancelled)
+ */
+export async function getCurrentBulkUpgrades(
+  esClient: ElasticsearchClient,
+  now = new Date().toISOString()
+): Promise<CurrentUpgrade[]> {
+  // Fetch all non expired actions
+  const [_upgradeActions, cancelledActionIds] = await Promise.all([
+    _getUpgradeActions(esClient, now),
+    _getCancelledActionId(esClient, now),
+  ]);
+
+  let upgradeActions = _upgradeActions.filter(
+    (action) => cancelledActionIds.indexOf(action.actionId) < 0
+  );
+
+  // Fetch acknowledged result for every upgrade action
+  upgradeActions = await pMap(
+    upgradeActions,
+    async (upgradeAction) => {
+      const { count } = await esClient.count({
+        index: AGENT_ACTIONS_RESULTS_INDEX,
+        ignore_unavailable: true,
+        query: {
+          bool: {
+            must: [
+              {
+                term: {
+                  action_id: upgradeAction.actionId,
+                },
+              },
+            ],
+          },
+        },
+      });
+
+      return {
+        ...upgradeAction,
+        nbAgentsAck: count,
+        complete: upgradeAction.nbAgents <= count,
+      };
+    },
+    { concurrency: 20 }
+  );
+
+  upgradeActions = upgradeActions.filter((action) => !action.complete);
+
+  return upgradeActions;
+}
+
+async function _getCancelledActionId(
+  esClient: ElasticsearchClient,
+  now = new Date().toISOString()
+) {
+  const res = await esClient.search<FleetServerAgentAction>({
+    index: AGENT_ACTIONS_INDEX,
+    query: {
+      bool: {
+        must: [
+          {
+            term: {
+              type: 'CANCEL',
+            },
+          },
+          {
+            exists: {
+              field: 'agents',
+            },
+          },
+          {
+            range: {
+              expiration: { gte: now },
+            },
+          },
+        ],
+      },
+    },
+  });
+
+  return res.hits.hits.map((hit) => hit._source?.data?.target_id as string);
+}
+
+async function _getUpgradeActions(esClient: ElasticsearchClient, now = new Date().toISOString()) {
+  const res = await esClient.search<FleetServerAgentAction>({
+    index: AGENT_ACTIONS_INDEX,
+    query: {
+      bool: {
+        must: [
+          {
+            term: {
+              type: 'UPGRADE',
+            },
+          },
+          {
+            exists: {
+              field: 'agents',
+            },
+          },
+          {
+            range: {
+              expiration: { gte: now },
+            },
+          },
+        ],
+      },
+    },
+  });
+
+  return Object.values(
+    res.hits.hits.reduce((acc, hit) => {
+      if (!hit._source || !hit._source.action_id) {
+        return acc;
+      }
+
+      if (!acc[hit._source.action_id]) {
+        acc[hit._source.action_id] = {
+          actionId: hit._source.action_id,
+          nbAgents: 0,
+          complete: false,
+          nbAgentsAck: 0,
+        };
+      }
+
+      acc[hit._source.action_id].nbAgents += hit._source.agents?.length ?? 0;
+
+      return acc;
+    }, {} as { [k: string]: CurrentUpgrade })
+  );
+}

x-pack/plugins/fleet/server/types/index.tsx

@@ -12,6 +12,7 @@ export type {
   AgentStatus,
   AgentType,
   AgentAction,
+  CurrentUpgrade,
   PackagePolicy,
   PackagePolicyInput,
   PackagePolicyInputStream,