[Fleet] Show remote es output error state on UI

x-pack/plugins/fleet/common/constants/output.ts

@@ -119,3 +119,5 @@ export const kafkaSupportedVersions = [
   '2.5.1',
   '2.6.0',
 ];
+
+export const OUTPUT_HEALTH_DATA_STREAM = 'logs-fleet_server.output_health-default';

x-pack/plugins/fleet/common/constants/routes.ts

@@ -94,6 +94,7 @@ export const OUTPUT_API_ROUTES = {
   UPDATE_PATTERN: `${API_ROOT}/outputs/{outputId}`,
   DELETE_PATTERN: `${API_ROOT}/outputs/{outputId}`,
   CREATE_PATTERN: `${API_ROOT}/outputs`,
+  GET_OUTPUT_HEALTH_PATTERN: `${API_ROOT}/outputs/{outputId}/health`,
   LOGSTASH_API_KEY_PATTERN: `${API_ROOT}/logstash_api_keys`,
 };
 

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

@@ -248,6 +248,8 @@ export const outputRoutesService = {
     OUTPUT_API_ROUTES.DELETE_PATTERN.replace('{outputId}', outputId),
   getCreatePath: () => OUTPUT_API_ROUTES.CREATE_PATTERN,
   getCreateLogstashApiKeyPath: () => OUTPUT_API_ROUTES.LOGSTASH_API_KEY_PATTERN,
+  getOutputHealthPath: (outputId: string) =>
+    OUTPUT_API_ROUTES.GET_OUTPUT_HEALTH_PATTERN.replace('{outputId}', outputId),
 };
 
 export const fleetProxiesRoutesService = {

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

@@ -43,3 +43,8 @@ export type GetOutputsResponse = ListResult<Output>;
 export interface PostLogstashApiKeyResponse {
   api_key: string;
 }
+
+export interface GetOutputHealthResponse {
+  state: string;
+  message: string;
+}

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

@@ -55,6 +55,7 @@ import { useOutputForm } from './use_output_form';
 import { EncryptionKeyRequiredCallout } from './encryption_key_required_callout';
 import { AdvancedOptionsSection } from './advanced_options_section';
 import { OutputFormRemoteEsSection } from './output_form_remote_es';
+import { OutputHealth } from './output_health';
 
 export interface EditOutputFlyoutProps {
   output?: Output;
@@ -576,6 +577,9 @@ export const EditOutputFlyout: React.FunctionComponent<EditOutputFlyoutProps> =
           <EuiSpacer size="l" />
           <AdvancedOptionsSection enabled={form.isShipperEnabled} inputs={inputs} />
         </EuiForm>
+        {output?.id && output.type === 'remote_elasticsearch' ? (
+          <OutputHealth output={output} />
+        ) : null}
       </EuiFlyoutBody>
       <EuiFlyoutFooter>
         <EuiFlexGroup justifyContent="spaceBetween">

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

@@ -0,0 +1,70 @@
+/*
+ * 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 { EuiCallOut } from '@elastic/eui';
+
+import { i18n } from '@kbn/i18n';
+import React, { useCallback, useEffect, useState } from 'react';
+
+import type { GetOutputHealthResponse } from '../../../../../../../common/types';
+
+import { sendGetOutputHealth, useStartServices } from '../../../../hooks';
+import type { Output } from '../../../../types';
+
+interface Props {
+  output: Output;
+}
+const REFRESH_INTERVAL_MS = 5000;
+
+export const OutputHealth: React.FunctionComponent<Props> = ({ output }) => {
+  const { notifications } = useStartServices();
+  const [outputHealth, setOutputHealth] = useState<GetOutputHealthResponse | null>();
+  const fetchData = useCallback(async () => {
+    try {
+      const response = await sendGetOutputHealth(output.id);
+      if (response.error) {
+        throw response.error;
+      }
+      setOutputHealth(response.data);
+    } catch (error) {
+      notifications.toasts.addError(error, {
+        title: i18n.translate('xpack.fleet.output.errorFetchingOutputHealth', {
+          defaultMessage: 'Error fetching output state',
+        }),
+      });
+    }
+  }, [output.id, notifications.toasts]);
+
+  // Send request to get output health
+  useEffect(() => {
+    fetchData();
+    const interval = setInterval(() => {
+      fetchData();
+    }, REFRESH_INTERVAL_MS);
+
+    return () => clearInterval(interval);
+  }, [fetchData]);
+
+  return outputHealth?.state === 'DEGRADED' ? (
+    <EuiCallOut title="Error" color="danger" iconType="error">
+      <p>
+        {i18n.translate('xpack.fleet.output.calloutText', {
+          defaultMessage: 'Unable to connect to "{name}" at {host}.',
+          values: {
+            name: output.name,
+            host: output.hosts?.join(',') ?? '',
+          },
+        })}
+      </p>{' '}
+      <p>
+        {i18n.translate('xpack.fleet.output.calloutPromptText', {
+          defaultMessage: 'Please check the details are correct.',
+        })}
+      </p>
+    </EuiCallOut>
+  ) : null;
+};

x-pack/plugins/fleet/public/hooks/use_request/outputs.ts

@@ -5,6 +5,8 @@
  * 2.0.
  */
 
+import type { GetOutputHealthResponse } from '../../../common/types';
+
 import { outputRoutesService } from '../../services';
 import type {
   PutOutputRequest,
@@ -65,3 +67,11 @@ export function sendDeleteOutput(outputId: string) {
     version: API_VERSIONS.public.v1,
   });
 }
+
+export function sendGetOutputHealth(outputId: string) {
+  return sendRequest<GetOutputHealthResponse>({
+    method: 'get',
+    path: outputRoutesService.getOutputHealthPath(outputId),
+    version: API_VERSIONS.public.v1,
+  });
+}

x-pack/plugins/fleet/server/constants/index.ts

@@ -81,6 +81,8 @@ export {
   // secrets
   SECRETS_ENDPOINT_PATH,
   SECRETS_MINIMUM_FLEET_SERVER_VERSION,
+  // outputs
+  OUTPUT_HEALTH_DATA_STREAM,
   type PrivilegeMapObject,
 } from '../../common/constants';
 

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

@@ -16,6 +16,7 @@ import { outputType } from '../../../common/constants';
 
 import type {
   DeleteOutputRequestSchema,
+  GetLatestOutputHealthRequestSchema,
   GetOneOutputRequestSchema,
   PostOutputRequestSchema,
   PutOutputRequestSchema,
@@ -207,3 +208,18 @@ export const postLogstashApiKeyHandler: RequestHandler = async (context, request
     return defaultFleetErrorHandler({ error, response });
   }
 };
+
+export const getLatestOutputHealth: RequestHandler<
+  TypeOf<typeof GetLatestOutputHealthRequestSchema.params>
+> = async (context, request, response) => {
+  const esClient = (await context.core).elasticsearch.client.asCurrentUser;
+  try {
+    const outputHealth = await outputService.getLatestOutputHealth(
+      esClient,
+      request.params.outputId
+    );
+    return response.ok({ body: outputHealth });
+  } catch (error) {
+    return defaultFleetErrorHandler({ error, response });
+  }
+};

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

@@ -12,6 +12,7 @@ import { API_VERSIONS } from '../../../common/constants';
 import { OUTPUT_API_ROUTES } from '../../constants';
 import {
   DeleteOutputRequestSchema,
+  GetLatestOutputHealthRequestSchema,
   GetOneOutputRequestSchema,
   GetOutputsRequestSchema,
   PostOutputRequestSchema,
@@ -25,6 +26,7 @@ import {
   postOutputHandler,
   putOutputHandler,
   postLogstashApiKeyHandler,
+  getLatestOutputHealth,
 } from './handler';
 
 export const registerRoutes = (router: FleetAuthzRouter) => {
@@ -115,4 +117,19 @@ export const registerRoutes = (router: FleetAuthzRouter) => {
       },
       postLogstashApiKeyHandler
     );
+
+  router.versioned
+    .get({
+      path: OUTPUT_API_ROUTES.GET_OUTPUT_HEALTH_PATTERN,
+      fleetAuthz: {
+        fleet: { all: true },
+      },
+    })
+    .addVersion(
+      {
+        version: API_VERSIONS.public.v1,
+        validate: { request: GetLatestOutputHealthRequestSchema },
+      },
+      getLatestOutputHealth
+    );
 };

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

@@ -33,6 +33,7 @@ import {
   DEFAULT_OUTPUT,
   DEFAULT_OUTPUT_ID,
   OUTPUT_SAVED_OBJECT_TYPE,
+  OUTPUT_HEALTH_DATA_STREAM,
 } from '../constants';
 import {
   SO_SEARCH_LIMIT,
@@ -952,6 +953,31 @@ class OutputService {
       }
     }
   }
+
+  async getLatestOutputHealth(esClient: ElasticsearchClient, id: string): Promise<OutputHealth> {
+    const response = await esClient.search({
+      index: OUTPUT_HEALTH_DATA_STREAM,
+      query: { bool: { filter: { term: { output: id } } } },
+      sort: { '@timestamp': 'desc' },
+      size: 1,
+    });
+    if (response.hits.hits.length === 0) {
+      return {
+        state: 'UNKOWN',
+        message: '',
+      };
+    }
+    const latestHit = response.hits.hits[0]._source as any;
+    return {
+      state: latestHit.state,
+      message: latestHit.message ?? '',
+    };
+  }
+}
+
+interface OutputHealth {
+  state: string;
+  message: string;
 }
 
 export const outputService = new OutputService();

x-pack/plugins/fleet/server/types/rest_spec/output.ts

@@ -33,3 +33,9 @@ export const PutOutputRequestSchema = {
   }),
   body: UpdateOutputSchema,
 };
+
+export const GetLatestOutputHealthRequestSchema = {
+  params: schema.object({
+    outputId: schema.string(),
+  }),
+};