[Fleet] hide remote es output in serverless (#171378)

## Summary

Relates https://github.com/elastic/kibana/issues/104986

Hide Remote Elasticsearch output in serverless from Create/Edit output
flyout.

Should we also add validation to prevent creating it in API?


Verified locally by starting kibana in serverless mode:
<img width="751" alt="image"
src="061514f3-25fe-4e52-ad85-194cc612bea7">

### 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/public/applications/fleet/sections/settings/components/edit_output_flyout/index.test.tsx

@@ -11,6 +11,7 @@ import type { Output } from '../../../../types';
 import { createFleetTestRendererMock } from '../../../../../../mock';
 import { useFleetStatus } from '../../../../../../hooks/use_fleet_status';
 import { ExperimentalFeaturesService } from '../../../../../../services';
+import { useStartServices } from '../../../../hooks';
 
 import { EditOutputFlyout } from '.';
 
@@ -25,6 +26,16 @@ jest.mock('../../../../../../hooks/use_fleet_status', () => ({
   useFleetStatus: jest.fn().mockReturnValue({}),
 }));
 
+jest.mock('../../../../hooks', () => {
+  return {
+    ...jest.requireActual('../../../../hooks'),
+    useBreadcrumbs: jest.fn(),
+    useStartServices: jest.fn(),
+  };
+});
+
+const mockUseStartServices = useStartServices as jest.Mock;
+
 const mockedUsedFleetStatus = useFleetStatus as jest.MockedFunction<typeof useFleetStatus>;
 
 function renderFlyout(output?: Output) {
@@ -67,6 +78,22 @@ const kafkaSectionsLabels = [
 const remoteEsOutputLabels = ['Hosts', 'Service Token'];
 
 describe('EditOutputFlyout', () => {
+  const mockStartServices = (isServerlessEnabled?: boolean) => {
+    mockUseStartServices.mockReturnValue({
+      notifications: { toasts: {} },
+      docLinks: {
+        links: { fleet: {}, logstash: {}, kibana: {} },
+      },
+      cloud: {
+        isServerlessEnabled,
+      },
+    });
+  };
+
+  beforeEach(() => {
+    mockStartServices(false);
+  });
+
   it('should render the flyout if there is not output provided', async () => {
     renderFlyout();
   });
@@ -177,5 +204,26 @@ describe('EditOutputFlyout', () => {
       expect(utils.queryByLabelText(label)).not.toBeNull();
     });
     expect(utils.queryByTestId('serviceTokenCallout')).not.toBeNull();
+
+    expect(utils.queryByTestId('settingsOutputsFlyout.typeInput')?.textContent).toContain(
+      'Remote Elasticsearch'
+    );
+  });
+
+  it('should not display remote ES output in type lists if serverless', async () => {
+    jest.spyOn(ExperimentalFeaturesService, 'get').mockReturnValue({ remoteESOutput: true });
+    mockUseStartServices.mockReset();
+    mockStartServices(true);
+    const { utils } = renderFlyout({
+      type: 'elasticsearch',
+      name: 'dummy',
+      id: 'output',
+      is_default: false,
+      is_default_monitoring: false,
+    });
+
+    expect(utils.queryByTestId('settingsOutputsFlyout.typeInput')?.textContent).not.toContain(
+      'Remote Elasticsearch'
+    );
   });
 });

x-pack/plugins/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/index.tsx

@@ -70,7 +70,7 @@ export const EditOutputFlyout: React.FunctionComponent<EditOutputFlyoutProps> =
   useBreadcrumbs('settings');
   const form = useOutputForm(onClose, output);
   const inputs = form.inputs;
-  const { docLinks } = useStartServices();
+  const { docLinks, cloud } = useStartServices();
   const { euiTheme } = useEuiTheme();
   const { outputSecretsStorage: isOutputSecretsStorageEnabled } = ExperimentalFeaturesService.get();
   const [useSecretsStorage, setUseSecretsStorage] = React.useState(isOutputSecretsStorageEnabled);
@@ -87,10 +87,12 @@ export const EditOutputFlyout: React.FunctionComponent<EditOutputFlyoutProps> =
   const { kafkaOutput: isKafkaOutputEnabled, remoteESOutput: isRemoteESOutputEnabled } =
     ExperimentalFeaturesService.get();
   const isRemoteESOutput = inputs.typeInput.value === outputType.RemoteElasticsearch;
+  // Remote ES output not yet supported in serverless
+  const isStateful = !cloud?.isServerlessEnabled;
 
   const OUTPUT_TYPE_OPTIONS = [
     { value: outputType.Elasticsearch, text: 'Elasticsearch' },
-    ...(isRemoteESOutputEnabled
+    ...(isRemoteESOutputEnabled && isStateful
       ? [{ value: outputType.RemoteElasticsearch, text: 'Remote Elasticsearch' }]
       : []),
     { value: outputType.Logstash, text: 'Logstash' },

x-pack/plugins/fleet/server/routes/output/handler.test.ts

@@ -0,0 +1,91 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import { agentPolicyService, appContextService, outputService } from '../../services';
+
+import { postOutputHandler, putOutputHandler } from './handler';
+
+describe('output handler', () => {
+  const mockContext = {
+    core: Promise.resolve({
+      savedObjects: {},
+      elasticsearch: {
+        client: {},
+      },
+    }),
+  } as any;
+  const mockResponse = {
+    customError: jest.fn().mockImplementation((options) => options),
+    ok: jest.fn().mockImplementation((options) => options),
+  };
+
+  beforeEach(() => {
+    jest.spyOn(appContextService, 'getLogger').mockReturnValue({ error: jest.fn() } as any);
+    jest.spyOn(outputService, 'create').mockResolvedValue({ id: 'output1' } as any);
+    jest.spyOn(outputService, 'update').mockResolvedValue({ id: 'output1' } as any);
+    jest.spyOn(outputService, 'get').mockResolvedValue({ id: 'output1' } as any);
+    jest.spyOn(agentPolicyService, 'bumpAllAgentPoliciesForOutput').mockResolvedValue({} as any);
+  });
+
+  it('should return error on post output using remote_elasticsearch in serverless', async () => {
+    jest.spyOn(appContextService, 'getCloud').mockReturnValue({ isServerlessEnabled: true } as any);
+
+    const res = await postOutputHandler(
+      mockContext,
+      { body: { id: 'output1', type: 'remote_elasticsearch' } } as any,
+      mockResponse as any
+    );
+
+    expect(res).toEqual({
+      body: { message: 'Output type remote_elasticsearch not supported in serverless' },
+      statusCode: 400,
+    });
+  });
+
+  it('should return ok on post output using remote_elasticsearch in stateful', async () => {
+    jest
+      .spyOn(appContextService, 'getCloud')
+      .mockReturnValue({ isServerlessEnabled: false } as any);
+
+    const res = await postOutputHandler(
+      mockContext,
+      { body: { type: 'remote_elasticsearch' } } as any,
+      mockResponse as any
+    );
+
+    expect(res).toEqual({ body: { item: { id: 'output1' } } });
+  });
+
+  it('should return error on put output using remote_elasticsearch in serverless', async () => {
+    jest.spyOn(appContextService, 'getCloud').mockReturnValue({ isServerlessEnabled: true } as any);
+
+    const res = await putOutputHandler(
+      mockContext,
+      { body: { id: 'output1', type: 'remote_elasticsearch' } } as any,
+      mockResponse as any
+    );
+
+    expect(res).toEqual({
+      body: { message: 'Output type remote_elasticsearch not supported in serverless' },
+      statusCode: 400,
+    });
+  });
+
+  it('should return ok on put output using remote_elasticsearch in stateful', async () => {
+    jest
+      .spyOn(appContextService, 'getCloud')
+      .mockReturnValue({ isServerlessEnabled: false } as any);
+
+    const res = await putOutputHandler(
+      mockContext,
+      { body: { type: 'remote_elasticsearch' }, params: { outputId: 'output1' } } as any,
+      mockResponse as any
+    );
+
+    expect(res).toEqual({ body: { item: { id: 'output1' } } });
+  });
+});

x-pack/plugins/fleet/server/routes/output/handler.ts

@@ -10,6 +10,8 @@ import type { TypeOf } from '@kbn/config-schema';
 
 import Boom from '@hapi/boom';
 
+import type { ValueOf } from '@elastic/eui';
+
 import { outputType } from '../../../common/constants';
 
 import type {
@@ -23,11 +25,12 @@ import type {
   GetOneOutputResponse,
   GetOutputsResponse,
   Output,
+  OutputType,
   PostLogstashApiKeyResponse,
 } from '../../../common/types';
 import { outputService } from '../../services/output';
 import { defaultFleetErrorHandler, FleetUnauthorizedError } from '../../errors';
-import { agentPolicyService } from '../../services';
+import { agentPolicyService, appContextService } from '../../services';
 import { generateLogstashApiKey, canCreateLogstashApiKey } from '../../services/api_keys';
 
 function ensureNoDuplicateSecrets(output: Partial<Output>) {
@@ -89,8 +92,9 @@ export const putOutputHandler: RequestHandler<
   const soClient = coreContext.savedObjects.client;
   const esClient = coreContext.elasticsearch.client.asInternalUser;
   const outputUpdate = request.body;
-  ensureNoDuplicateSecrets(outputUpdate);
   try {
+    validateOutputServerless(outputUpdate.type);
+    ensureNoDuplicateSecrets(outputUpdate);
     await outputService.update(soClient, esClient, request.params.outputId, outputUpdate);
     const output = await outputService.get(soClient, request.params.outputId);
     if (output.is_default || output.is_default_monitoring) {
@@ -125,6 +129,7 @@ export const postOutputHandler: RequestHandler<
   const esClient = coreContext.elasticsearch.client.asInternalUser;
   try {
     const { id, ...newOutput } = request.body;
+    validateOutputServerless(newOutput.type);
     ensureNoDuplicateSecrets(newOutput);
     const output = await outputService.create(soClient, esClient, newOutput, { id });
     if (output.is_default || output.is_default_monitoring) {
@@ -141,6 +146,13 @@ export const postOutputHandler: RequestHandler<
   }
 };
 
+function validateOutputServerless(type?: ValueOf<OutputType>): void {
+  const cloudSetup = appContextService.getCloud();
+  if (cloudSetup?.isServerlessEnabled && type === outputType.RemoteElasticsearch) {
+    throw Boom.badRequest('Output type remote_elasticsearch not supported in serverless');
+  }
+}
+
 export const deleteOutputHandler: RequestHandler<
   TypeOf<typeof DeleteOutputRequestSchema.params>
 > = async (context, request, response) => {