[Fleet] Bulk upgrade api response change (elastic#95236) (elastic#95547)

## Summary
`/agents/bulk_upgrade` should return a response with a result for each agent given; including invalid or missing ids. It currently returns an empty object.

This PR includes commits from open PR elastic#95024. The additions from this PR are https://github.com/jfsiii/kibana/compare/bulk-reassign-response-should-include-all-given-agents..871ebcb

[TS type diff for response](https://github.com/jfsiii/kibana/compare/bulk-reassign-response-should-include-all-given-agents..871ebcb#diff-7006a6c170a608c8c7211fc218c0a6f4bc8ff642c170ea264db4b1b5545fb728)

```diff
- // eslint-disable-next-line @typescript-eslint/no-empty-interface
- export interface PostBulkAgentUpgradeResponse {}

+ export type PostBulkAgentUpgradeResponse = Record<
+   Agent['id'],
+   {
+     success: boolean;
+     error?: string;
+   }
+ >;
```

### Checklist

- [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

Co-authored-by: Kibana Machine <[email protected]>

Co-authored-by: John Schulz <[email protected]>

x-pack/plugins/fleet/common/services/is_agent_upgradeable.ts

@@ -17,13 +17,19 @@ export function isAgentUpgradeable(agent: Agent, kibanaVersion: string) {
   } else {
     return false;
   }
-  if (agent.unenrollment_started_at || agent.unenrolled_at) return false;
-  if (!agent.local_metadata.elastic.agent.upgradeable) return false;
+  if (agent.unenrollment_started_at || agent.unenrolled_at) {
+    return false;
+  }
+  if (!agent.local_metadata.elastic.agent.upgradeable) {
+    return false;
+  }
 
   // make sure versions are only the number before comparison
   const agentVersionNumber = semverCoerce(agentVersion);
   if (!agentVersionNumber) throw new Error('agent version is invalid');
   const kibanaVersionNumber = semverCoerce(kibanaVersion);
   if (!kibanaVersionNumber) throw new Error('kibana version is invalid');
-  return semverLt(agentVersionNumber, kibanaVersionNumber);
+  const isAgentLessThanKibana = semverLt(agentVersionNumber, kibanaVersionNumber);
+
+  return isAgentLessThanKibana;
 }

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

@@ -141,8 +141,14 @@ export interface PostBulkAgentUpgradeRequest {
     version: string;
   };
 }
-// eslint-disable-next-line @typescript-eslint/no-empty-interface
-export interface PostBulkAgentUpgradeResponse {}
+
+export type PostBulkAgentUpgradeResponse = Record<
+  Agent['id'],
+  {
+    success: boolean;
+    error?: string;
+  }
+>;
 
 // eslint-disable-next-line @typescript-eslint/no-empty-interface
 export interface PostAgentUpgradeResponse {}

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

@@ -99,9 +99,15 @@ export const postBulkAgentsUpgradeHandler: RequestHandler<
       version,
       force,
     };
-    await AgentService.sendUpgradeAgentsActions(soClient, esClient, upgradeOptions);
+    const results = await AgentService.sendUpgradeAgentsActions(soClient, esClient, upgradeOptions);
+    const body = results.items.reduce<PostBulkAgentUpgradeResponse>((acc, so) => {
+      acc[so.id] = {
+        success: !so.error,
+        error: so.error ? so.error.message || so.error.toString() : undefined,
+      };
+      return acc;
+    }, {});
 
-    const body: PostBulkAgentUpgradeResponse = {};
     return response.ok({ body });
   } catch (error) {
     return defaultIngestErrorHandler({ error, response });

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

@@ -7,16 +7,23 @@
 
 import type { ElasticsearchClient, SavedObjectsClientContract } from 'src/core/server';
 
-import type { AgentAction, AgentActionSOAttributes } from '../../types';
+import type { Agent, AgentAction, AgentActionSOAttributes, BulkActionResult } from '../../types';
 import { AGENT_ACTION_SAVED_OBJECT_TYPE } from '../../constants';
 import { agentPolicyService } from '../../services';
-import { IngestManagerError } from '../../errors';
+import { AgentReassignmentError, IngestManagerError } from '../../errors';
 import { isAgentUpgradeable } from '../../../common/services';
 import { appContextService } from '../app_context';
 
 import { bulkCreateAgentActions, createAgentAction } from './actions';
 import type { GetAgentsOptions } from './crud';
-import { getAgents, updateAgent, bulkUpdateAgents, getAgentPolicyForAgent } from './crud';
+import {
+  getAgentDocuments,
+  getAgents,
+  updateAgent,
+  bulkUpdateAgents,
+  getAgentPolicyForAgent,
+} from './crud';
+import { searchHitToAgent } from './helpers';
 
 export async function sendUpgradeAgentAction({
   soClient,
@@ -77,39 +84,75 @@ export async function ackAgentUpgraded(
 export async function sendUpgradeAgentsActions(
   soClient: SavedObjectsClientContract,
   esClient: ElasticsearchClient,
-  options: GetAgentsOptions & {
+  options: ({ agents: Agent[] } | GetAgentsOptions) & {
     sourceUri: string | undefined;
     version: string;
     force?: boolean;
   }
 ) {
   // Full set of agents
-  const agentsGiven = await getAgents(esClient, options);
+  const outgoingErrors: Record<Agent['id'], Error> = {};
+  let givenAgents: Agent[] = [];
+  if ('agents' in options) {
+    givenAgents = options.agents;
+  } else if ('agentIds' in options) {
+    const givenAgentsResults = await getAgentDocuments(esClient, options.agentIds);
+    for (const agentResult of givenAgentsResults) {
+      if (agentResult.found === false) {
+        outgoingErrors[agentResult._id] = new AgentReassignmentError(
+          `Cannot find agent ${agentResult._id}`
+        );
+      } else {
+        givenAgents.push(searchHitToAgent(agentResult));
+      }
+    }
+  } else if ('kuery' in options) {
+    givenAgents = await getAgents(esClient, options);
+  }
+  const givenOrder =
+    'agentIds' in options ? options.agentIds : givenAgents.map((agent) => agent.id);
+
+  // get any policy ids from upgradable agents
+  const policyIdsToGet = new Set(
+    givenAgents.filter((agent) => agent.policy_id).map((agent) => agent.policy_id!)
+  );
+
+  // get the agent policies for those ids
+  const agentPolicies = await agentPolicyService.getByIDs(soClient, Array.from(policyIdsToGet), {
+    fields: ['is_managed'],
+  });
+  const managedPolicies = agentPolicies.reduce<Record<string, boolean>>((acc, policy) => {
+    acc[policy.id] = policy.is_managed;
+    return acc;
+  }, {});
 
   // Filter out agents currently unenrolling, unenrolled, or not upgradeable b/c of version check
   const kibanaVersion = appContextService.getKibanaVersion();
-  const upgradeableAgents = options.force
-    ? agentsGiven
-    : agentsGiven.filter((agent) => isAgentUpgradeable(agent, kibanaVersion));
-
-  if (!options.force) {
-    // get any policy ids from upgradable agents
-    const policyIdsToGet = new Set(
-      upgradeableAgents.filter((agent) => agent.policy_id).map((agent) => agent.policy_id!)
-    );
-
-    // get the agent policies for those ids
-    const agentPolicies = await agentPolicyService.getByIDs(soClient, Array.from(policyIdsToGet), {
-      fields: ['is_managed'],
-    });
+  const agentResults = await Promise.allSettled(
+    givenAgents.map(async (agent) => {
+      const isAllowed = options.force || isAgentUpgradeable(agent, kibanaVersion);
+      if (!isAllowed) {
+        throw new IngestManagerError(`${agent.id} is not upgradeable`);
+      }
 
-    // throw if any of those agent policies are managed
-    for (const policy of agentPolicies) {
-      if (policy.is_managed) {
-        throw new IngestManagerError(`Cannot upgrade agent in managed policy ${policy.id}`);
+      if (!options.force && agent.policy_id && managedPolicies[agent.policy_id]) {
+        throw new IngestManagerError(`Cannot upgrade agent in managed policy ${agent.policy_id}`);
       }
+      return agent;
+    })
+  );
+
+  // Filter to agents that do not already use the new agent policy ID
+  const agentsToUpdate = agentResults.reduce<Agent[]>((agents, result, index) => {
+    if (result.status === 'fulfilled') {
+      agents.push(result.value);
+    } else {
+      const id = givenAgents[index].id;
+      outgoingErrors[id] = result.reason;
     }
-  }
+    return agents;
+  }, []);
+
   // Create upgrade action for each agent
   const now = new Date().toISOString();
   const data = {
@@ -120,7 +163,7 @@ export async function sendUpgradeAgentsActions(
   await bulkCreateAgentActions(
     soClient,
     esClient,
-    upgradeableAgents.map((agent) => ({
+    agentsToUpdate.map((agent) => ({
       agent_id: agent.id,
       created_at: now,
       data,
@@ -129,14 +172,27 @@ export async function sendUpgradeAgentsActions(
     }))
   );
 
-  return await bulkUpdateAgents(
+  await bulkUpdateAgents(
     esClient,
-    upgradeableAgents.map((agent) => ({
+    agentsToUpdate.map((agent) => ({
       agentId: agent.id,
       data: {
         upgraded_at: null,
         upgrade_started_at: now,
       },
     }))
   );
+  const orderedOut = givenOrder.map((agentId) => {
+    const hasError = agentId in outgoingErrors;
+    const result: BulkActionResult = {
+      id: agentId,
+      success: !hasError,
+    };
+    if (hasError) {
+      result.error = outgoingErrors[agentId];
+    }
+    return result;
+  });
+
+  return { items: orderedOut };
 }

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

@@ -540,7 +540,11 @@ export default function (providerContext: FtrProviderContext) {
           .expect(400);
       });
 
-      it('enrolled in a managed policy bulk upgrade should respond with 400 and not update the agent SOs', async () => {
+      it('enrolled in a managed policy bulk upgrade should respond with 200 and object of results. Should not update the managed agent SOs', async () => {
+        // move agent2 to policy2 to keep it unmanaged
+        await supertest.put(`/api/fleet/agents/agent2/reassign`).set('kbn-xsrf', 'xxx').send({
+          policy_id: 'policy2',
+        });
         // update enrolled policy to managed
         await supertest.put(`/api/fleet/agent_policies/policy1`).set('kbn-xsrf', 'xxxx').send({
           name: 'Test policy',
@@ -567,7 +571,7 @@ export default function (providerContext: FtrProviderContext) {
             doc: {
               local_metadata: {
                 elastic: {
-                  agent: { upgradeable: true, version: semver.inc(kibanaVersion, 'patch') },
+                  agent: { upgradeable: true, version: '0.0.0' },
                 },
               },
             },
@@ -581,16 +585,20 @@ export default function (providerContext: FtrProviderContext) {
             version: kibanaVersion,
             agents: ['agent1', 'agent2'],
           })
-          .expect(400);
-        expect(body.message).to.contain('Cannot upgrade agent in managed policy policy1');
+          .expect(200);
+
+        expect(body).to.eql({
+          agent1: { success: false, error: 'Cannot upgrade agent in managed policy policy1' },
+          agent2: { success: true },
+        });
 
         const [agent1data, agent2data] = await Promise.all([
           supertest.get(`/api/fleet/agents/agent1`),
           supertest.get(`/api/fleet/agents/agent2`),
         ]);
 
         expect(typeof agent1data.body.item.upgrade_started_at).to.be('undefined');
-        expect(typeof agent2data.body.item.upgrade_started_at).to.be('undefined');
+        expect(typeof agent2data.body.item.upgrade_started_at).to.be('string');
       });
 
       it('enrolled in a managed policy bulk upgrade with force flag should respond with 200 and update the agent SOs', async () => {
@@ -635,7 +643,11 @@ export default function (providerContext: FtrProviderContext) {
             agents: ['agent1', 'agent2'],
             force: true,
           });
-        expect(body).to.eql({});
+
+        expect(body).to.eql({
+          agent1: { success: true },
+          agent2: { success: true },
+        });
 
         const [agent1data, agent2data] = await Promise.all([
           supertest.get(`/api/fleet/agents/agent1`),