[Fleet] added output validation when creating package policy (#175985)

## Summary

Closes https://github.com/elastic/kibana/issues/165816

Added validation to prevent adding an integration to an agent policy
that uses an output different than `elasticsearch`.

To verify:
1. Create a logstash output and make it default output by enabling the
toggles.
2. Navigate to Integrations and search Fleet Server integration.
3. Add this fleet server integration to a new or existing policy.
4. There should be an error popup saying that this integration can't be
added to the agent policy.

<img width="1439" alt="image"
src="13f39bfc-8af1-4fa7-95e8-7ff41c6439a4">


### 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/common/services/output_helpers.ts

@@ -19,18 +19,21 @@ import {
   RESERVED_CONFIG_YML_KEYS,
 } from '../constants';
 
+const sameClusterRestrictedPackages = [
+  FLEET_SERVER_PACKAGE,
+  FLEET_SYNTHETICS_PACKAGE,
+  FLEET_APM_PACKAGE,
+];
+
 /**
  * Return allowed output type for a given agent policy,
  * Fleet Server and APM cannot use anything else than same cluster ES
  */
-export function getAllowedOutputTypeForPolicy(agentPolicy: AgentPolicy) {
+export function getAllowedOutputTypeForPolicy(agentPolicy: AgentPolicy): string[] {
   const isRestrictedToSameClusterES =
     agentPolicy.package_policies &&
     agentPolicy.package_policies.some(
-      (p) =>
-        p.package?.name === FLEET_SERVER_PACKAGE ||
-        p.package?.name === FLEET_SYNTHETICS_PACKAGE ||
-        p.package?.name === FLEET_APM_PACKAGE
+      (p) => p.package?.name && sameClusterRestrictedPackages.includes(p.package?.name)
     );
 
   if (isRestrictedToSameClusterES) {
@@ -40,6 +43,16 @@ export function getAllowedOutputTypeForPolicy(agentPolicy: AgentPolicy) {
   return Object.values(outputType);
 }
 
+export function getAllowedOutputTypesForIntegration(packageName: string): string[] {
+  const isRestrictedToSameClusterES = sameClusterRestrictedPackages.includes(packageName);
+
+  if (isRestrictedToSameClusterES) {
+    return [outputType.Elasticsearch];
+  }
+
+  return Object.values(outputType);
+}
+
 export function outputYmlIncludesReservedPerformanceKey(
   configYml: string,
   // Dependency injection for `safeLoad` prevents bundle size issues 🤷‍♀️

x-pack/plugins/fleet/server/services/agent_policies/output_helpers.test.ts

@@ -13,6 +13,7 @@ import { appContextService } from '..';
 import { outputService } from '../output';
 
 import { validateOutputForPolicy } from '.';
+import { validateOutputForNewPackagePolicy } from './outputs_helpers';
 
 jest.mock('../app_context');
 jest.mock('../output');
@@ -252,3 +253,94 @@ describe('validateOutputForPolicy', () => {
     });
   });
 });
+
+describe('validateOutputForNewPackagePolicy', () => {
+  it('should not allow fleet_server integration to be added to a policy using a logstash output', async () => {
+    mockHasLicence(true);
+    mockedOutputService.get.mockResolvedValue({
+      type: 'logstash',
+    } as any);
+    await expect(
+      validateOutputForNewPackagePolicy(
+        savedObjectsClientMock.create(),
+        {
+          name: 'Agent policy',
+          data_output_id: 'test1',
+          monitoring_output_id: 'test1',
+        } as any,
+        'fleet_server'
+      )
+    ).rejects.toThrow(
+      'Integration "fleet_server" cannot be added to agent policy "Agent policy" because it uses output type "logstash".'
+    );
+  });
+
+  it('should not allow apm integration to be added to a policy using a kafka output', async () => {
+    mockHasLicence(true);
+    mockedOutputService.get.mockResolvedValue({
+      type: 'kafka',
+    } as any);
+    await expect(
+      validateOutputForNewPackagePolicy(
+        savedObjectsClientMock.create(),
+        {
+          name: 'Agent policy',
+          data_output_id: 'test1',
+          monitoring_output_id: 'test1',
+        } as any,
+        'apm'
+      )
+    ).rejects.toThrow(
+      'Integration "apm" cannot be added to agent policy "Agent policy" because it uses output type "kafka".'
+    );
+  });
+
+  it('should not allow synthetics integration to be added to a policy using a default logstash output', async () => {
+    mockHasLicence(true);
+    mockedOutputService.get.mockResolvedValue({
+      type: 'logstash',
+    } as any);
+    mockedOutputService.getDefaultDataOutputId.mockResolvedValue('default');
+    await expect(
+      validateOutputForNewPackagePolicy(
+        savedObjectsClientMock.create(),
+        {
+          name: 'Agent policy',
+        } as any,
+        'synthetics'
+      )
+    ).rejects.toThrow(
+      'Integration "synthetics" cannot be added to agent policy "Agent policy" because it uses output type "logstash".'
+    );
+  });
+
+  it('should allow other integration to be added to a policy using logstash output', async () => {
+    mockHasLicence(true);
+    mockedOutputService.get.mockResolvedValue({
+      type: 'logstash',
+    } as any);
+
+    await validateOutputForNewPackagePolicy(
+      savedObjectsClientMock.create(),
+      {
+        name: 'Agent policy',
+      } as any,
+      'nginx'
+    );
+  });
+
+  it('should allow fleet_server integration to be added to a policy using elasticsearch output', async () => {
+    mockHasLicence(true);
+    mockedOutputService.get.mockResolvedValue({
+      type: 'elasticsearch',
+    } as any);
+
+    await validateOutputForNewPackagePolicy(
+      savedObjectsClientMock.create(),
+      {
+        name: 'Agent policy',
+      } as any,
+      'fleet_server'
+    );
+  });
+});

x-pack/plugins/fleet/server/services/agent_policies/outputs_helpers.ts

@@ -7,6 +7,8 @@
 
 import type { SavedObjectsClientContract } from '@kbn/core/server';
 
+import { getAllowedOutputTypesForIntegration } from '../../../common/services/output_helpers';
+
 import type { AgentPolicySOAttributes, AgentPolicy } from '../../types';
 import { LICENCE_FOR_PER_POLICY_OUTPUT, outputType } from '../../../common/constants';
 import { policyHasFleetServer, policyHasSyntheticsIntegration } from '../../../common/services';
@@ -46,7 +48,7 @@ export async function validateOutputForPolicy(
   soClient: SavedObjectsClientContract,
   newData: Partial<AgentPolicySOAttributes>,
   existingData: Partial<AgentPolicySOAttributes> = {},
-  allowedOutputTypeForPolicy = Object.values(outputType)
+  allowedOutputTypeForPolicy: string[] = Object.values(outputType)
 ) {
   if (
     newData.data_output_id === existingData.data_output_id &&
@@ -93,3 +95,23 @@ export async function validateOutputForPolicy(
     );
   }
 }
+
+export async function validateOutputForNewPackagePolicy(
+  soClient: SavedObjectsClientContract,
+  agentPolicy: AgentPolicy,
+  packageName: string
+) {
+  const allowedOutputTypeForPolicy = getAllowedOutputTypesForIntegration(packageName);
+
+  const isOutputTypeRestricted =
+    allowedOutputTypeForPolicy.length !== Object.values(outputType).length;
+
+  if (isOutputTypeRestricted) {
+    const dataOutput = await getDataOutputForAgentPolicy(soClient, agentPolicy);
+    if (!allowedOutputTypeForPolicy.includes(dataOutput.type)) {
+      throw new OutputInvalidError(
+        `Integration "${packageName}" cannot be added to agent policy "${agentPolicy.name}" because it uses output type "${dataOutput.type}".`
+      );
+    }
+  }
+}

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

@@ -46,8 +46,6 @@ import {
 } from '../../common/services';
 import {
   SO_SEARCH_LIMIT,
-  FLEET_APM_PACKAGE,
-  outputType,
   PACKAGES_SAVED_OBJECT_TYPE,
   DATASET_VAR_NAME,
 } from '../../common/constants';
@@ -103,7 +101,6 @@ import { getAuthzFromRequest, doesNotHaveRequiredFleetAuthz } from './security';
 
 import { storedPackagePolicyToAgentInputs } from './agent_policies';
 import { agentPolicyService } from './agent_policy';
-import { getDataOutputForAgentPolicy } from './agent_policies';
 import { getPackageInfo, getInstallation, ensureInstalledPackage } from './epm/packages';
 import { getAssetsDataFromAssetsMap } from './epm/packages/assets';
 import { compileTemplate } from './epm/agent/agent';
@@ -124,6 +121,7 @@ import {
   isSecretStorageEnabled,
 } from './secrets';
 import { getPackageAssetsMap } from './epm/packages/get';
+import { validateOutputForNewPackagePolicy } from './agent_policies/outputs_helpers';
 
 export type InputsOverride = Partial<NewPackagePolicyInput> & {
   vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
@@ -225,11 +223,12 @@ class PackagePolicyClientImpl implements PackagePolicyClient {
       true
     );
 
-    if (agentPolicy && enrichedPackagePolicy.package?.name === FLEET_APM_PACKAGE) {
-      const dataOutput = await getDataOutputForAgentPolicy(soClient, agentPolicy);
-      if (dataOutput.type === outputType.Logstash) {
-        throw new FleetError('You cannot add APM to a policy using a logstash output');
-      }
+    if (agentPolicy && enrichedPackagePolicy.package?.name) {
+      await validateOutputForNewPackagePolicy(
+        soClient,
+        agentPolicy,
+        enrichedPackagePolicy.package?.name
+      );
     }
     await validateIsNotHostedPolicy(soClient, enrichedPackagePolicy.policy_id, options?.force);