[Fleet] validate all agent policies before updating them on output de…

…lete (#174921)

## Summary

Closes #165817

Moved the logic of validating agent policies before updating any agent
policies on output delete. This fixes the scenario where the output
delete fails on an agent policy with fleet-server integration, but some
agent policies were already updated by then.

To verify:
- create a logstash output and make it default for data and monitoring
output
- create a fleet server policy (with fleet server integration) with data
and monitoring output set to elasticsearch output
- create another agent policy with data and monitoring output set to
elasticsearch output
- attempt to delete the elasticsearch output, expect a UI error
- check that the agent policies are still referencing the elasticsearch
output, not the logstash one


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

x-pack/plugins/fleet/server/services/agent_policy.test.ts

@@ -479,6 +479,26 @@ describe('agent policy', () => {
           },
         ],
       } as any);
+      soClient.get.mockImplementation((type, id, options): any => {
+        if (id === 'test1') {
+          return Promise.resolve({
+            id,
+            attributes: {
+              data_output_id: 'output-id-123',
+              monitoring_output_id: 'output-id-another-output',
+            },
+          });
+        }
+        if (id === 'test2') {
+          return Promise.resolve({
+            id,
+            attributes: {
+              data_output_id: 'output-id-another-output',
+              monitoring_output_id: 'output-id-123',
+            },
+          });
+        }
+      });
 
       await agentPolicyService.removeOutputFromAll(soClient, esClient, 'output-id-123');
 
@@ -487,13 +507,15 @@ describe('agent policy', () => {
         expect.anything(),
         expect.anything(),
         'test1',
-        { data_output_id: null, monitoring_output_id: 'output-id-another-output' }
+        { data_output_id: null, monitoring_output_id: 'output-id-another-output' },
+        { skipValidation: true }
       );
       expect(mockedAgentPolicyServiceUpdate).toHaveBeenCalledWith(
         expect.anything(),
         expect.anything(),
         'test2',
-        { data_output_id: 'output-id-another-output', monitoring_output_id: null }
+        { data_output_id: 'output-id-another-output', monitoring_output_id: null },
+        { skipValidation: true }
       );
     });
   });

x-pack/plugins/fleet/server/services/agent_policy.ts

@@ -116,9 +116,10 @@ class AgentPolicyService {
     id: string,
     agentPolicy: Partial<AgentPolicySOAttributes>,
     user?: AuthenticatedUser,
-    options: { bumpRevision: boolean; removeProtection: boolean } = {
+    options: { bumpRevision: boolean; removeProtection: boolean; skipValidation: boolean } = {
       bumpRevision: true,
       removeProtection: false,
+      skipValidation: false,
     }
   ): Promise<AgentPolicy> {
     auditLoggingService.writeCustomSoAuditLog({
@@ -148,12 +149,14 @@ class AgentPolicyService {
       logger.warn(`Setting tamper protection for Agent Policy ${id} to false`);
     }
 
-    await validateOutputForPolicy(
-      soClient,
-      agentPolicy,
-      existingAgentPolicy,
-      getAllowedOutputTypeForPolicy(existingAgentPolicy)
-    );
+    if (!options.skipValidation) {
+      await validateOutputForPolicy(
+        soClient,
+        agentPolicy,
+        existingAgentPolicy,
+        getAllowedOutputTypeForPolicy(existingAgentPolicy)
+      );
+    }
     await soClient.update<AgentPolicySOAttributes>(SAVED_OBJECT_TYPE, id, {
       ...agentPolicy,
       ...(options.bumpRevision ? { revision: existingAgentPolicy.revision + 1 } : {}),
@@ -498,6 +501,7 @@ class AgentPolicyService {
       force?: boolean;
       spaceId?: string;
       authorizationHeader?: HTTPAuthorizationHeader | null;
+      skipValidation?: boolean;
     }
   ): Promise<AgentPolicy> {
     const logger = appContextService.getLogger();
@@ -551,7 +555,11 @@ class AgentPolicyService {
       });
     }
 
-    return this._update(soClient, esClient, id, agentPolicy, options?.user);
+    return this._update(soClient, esClient, id, agentPolicy, options?.user, {
+      bumpRevision: true,
+      removeProtection: false,
+      skipValidation: options?.skipValidation ?? false,
+    });
   }
 
   public async copy(
@@ -630,6 +638,7 @@ class AgentPolicyService {
         {
           bumpRevision: false,
           removeProtection: false,
+          skipValidation: false,
         }
       );
     }
@@ -654,6 +663,7 @@ class AgentPolicyService {
     const res = await this._update(soClient, esClient, id, {}, options?.user, {
       bumpRevision: true,
       removeProtection: options?.removeProtection ?? false,
+      skipValidation: false,
     });
 
     return res;
@@ -684,16 +694,37 @@ class AgentPolicyService {
     }));
 
     if (agentPolicies.length > 0) {
+      const getAgentPolicy = (agentPolicy: AgentPolicy) => ({
+        data_output_id: agentPolicy.data_output_id === outputId ? null : agentPolicy.data_output_id,
+        monitoring_output_id:
+          agentPolicy.monitoring_output_id === outputId ? null : agentPolicy.monitoring_output_id,
+      });
+      // Validate that the output is not used by any agent policy before updating any policy
+      await pMap(
+        agentPolicies,
+        async (agentPolicy) => {
+          const existingAgentPolicy = await this.get(soClient, agentPolicy.id, true);
+
+          if (!existingAgentPolicy) {
+            throw new AgentPolicyNotFoundError('Agent policy not found');
+          }
+
+          await validateOutputForPolicy(
+            soClient,
+            getAgentPolicy(agentPolicy),
+            existingAgentPolicy,
+            getAllowedOutputTypeForPolicy(existingAgentPolicy)
+          );
+        },
+        {
+          concurrency: 50,
+        }
+      );
       await pMap(
         agentPolicies,
         (agentPolicy) =>
-          this.update(soClient, esClient, agentPolicy.id, {
-            data_output_id:
-              agentPolicy.data_output_id === outputId ? null : agentPolicy.data_output_id,
-            monitoring_output_id:
-              agentPolicy.monitoring_output_id === outputId
-                ? null
-                : agentPolicy.monitoring_output_id,
+          this.update(soClient, esClient, agentPolicy.id, getAgentPolicy(agentPolicy), {
+            skipValidation: true,
           }),
         {
           concurrency: 50,

x-pack/test/fleet_api_integration/apis/outputs/crud.ts

@@ -1536,6 +1536,77 @@ export default function (providerContext: FtrProviderContext) {
 
           expect(deleteResponse.id).to.eql(outputId);
         });
+
+        it('should not modify agent policies when cannot delete an output due to default logstash', async function () {
+          let { body: apiResponse } = await supertest
+            .post(`/api/fleet/outputs`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              name: 'Elastic output',
+              type: 'elasticsearch',
+              hosts: ['http://localhost'],
+            })
+            .expect(200);
+          const esOutputId = apiResponse.item.id;
+
+          const agentPolicyId = '0000-agent-policy';
+          ({ body: apiResponse } = await supertest
+            .post(`/api/fleet/agent_policies`)
+            .set('kbn-xsrf', 'kibana')
+            .send({
+              id: agentPolicyId,
+              name: 'Agent policy 2',
+              namespace: 'default',
+              data_output_id: `${esOutputId}`,
+              monitoring_output_id: `${esOutputId}`,
+            })
+            .expect(200));
+
+          const fleetPolicyId = '1111-fleet-policy';
+          ({ body: apiResponse } = await supertest
+            .post(`/api/fleet/agent_policies`)
+            .set('kbn-xsrf', 'kibana')
+            .send({
+              id: fleetPolicyId,
+              name: 'Fleet Server policy 2',
+              namespace: 'default',
+              has_fleet_server: true,
+              data_output_id: `${esOutputId}`,
+            })
+            .expect(200));
+
+          await supertest
+            .post(`/api/fleet/outputs`)
+            .set('kbn-xsrf', 'xxxx')
+            .send({
+              name: 'Default logstash',
+              type: 'logstash',
+              hosts: ['logstash'],
+              ssl: { certificate: 'CERTIFICATE', key: 'KEY', certificate_authorities: [] },
+              is_default: true,
+              is_default_monitoring: true,
+            })
+            .expect(200);
+
+          const { body: errorResponse } = await supertest
+            .delete(`/api/fleet/outputs/${esOutputId}`)
+            .set('kbn-xsrf', 'xxxx')
+            .expect(400);
+          expect(errorResponse.message).to.eql(
+            'Output of type "logstash" is not usable with policy "Fleet Server policy 2".'
+          );
+
+          const { body: getAgentPolicyResponse } = await supertest.get(
+            `/api/fleet/agent_policies/${agentPolicyId}`
+          );
+          expect(getAgentPolicyResponse.item.data_output_id).to.eql(esOutputId);
+          expect(getAgentPolicyResponse.item.monitoring_output_id).to.eql(esOutputId);
+
+          const { body: getFleetServerAgentPolicyResponse } = await supertest.get(
+            `/api/fleet/agent_policies/${fleetPolicyId}`
+          );
+          expect(getFleetServerAgentPolicyResponse.item.data_output_id).to.eql(esOutputId);
+        });
       });
 
       describe('Kafka output', () => {