[Fleet] only show remote ES output health status if later than last u…

…pdated time (elastic#177685)

## Summary

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

Added a filter when querying remote ES output health status, to only
return results after the last update time of the output (`updated_at`
field of the SO).
This makes the health status reporting more accurate, so old statuses
are not staying on the UI, only latest status after the last update.
If the output query errors out or the `updated_at` field is not present,
the filter is omitted.

To verify:
- create a remote ES output (can be the same as the local ES), use it as
monitoring output of an agent policy
- enroll an agent to this agent policy
- update output to use an invalid host url
- wait until the remote ES output is showing up with error state on UI
- stop the Fleet-server
- update the remote ES output to use a correct host url
- wait until the remote ES output status is cleared on the UI
- start Fleet-server, wait until the agent checks in again (can be a few
minutes)
- verify that the remote ES output status shows up as healthy on the UI

Invalid url:
<img width="581" alt="image"
src="https://github.com/elastic/kibana/assets/90178898/b8a98cb1-4a1b-4d74-b260-b95bf8eaac62">

Fleet-server stopped and updated to valid url:
<img width="1133" alt="image"
src="https://github.com/elastic/kibana/assets/90178898/0e8a047f-48d8-4a3e-90e5-9a2ae1c2f874">

Fleet-server restarted:
<img width="1131" alt="image"
src="https://github.com/elastic/kibana/assets/90178898/0cf642e5-b26f-41d7-ad45-acc2c6c6111f">

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

(cherry picked from commit 2005cef)

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

@@ -65,7 +65,7 @@ const CONFIG_WITHOUT_ES_HOSTS = {
   },
 };
 
-function mockOutputSO(id: string, attributes: any = {}) {
+function mockOutputSO(id: string, attributes: any = {}, updatedAt?: string) {
   return {
     id: outputIdToUuid(id),
     type: 'ingest-outputs',
@@ -74,6 +74,7 @@ function mockOutputSO(id: string, attributes: any = {}) {
       output_id: id,
       ...attributes,
     },
+    updated_at: updatedAt,
   };
 }
 
@@ -146,7 +147,9 @@ function getMockedSoClient(
       }
 
       default:
-        throw new Error('not found: ' + id);
+        return mockOutputSO(id, {
+          type: 'remote_elasticsearch',
+        });
     }
   });
   soClient.update.mockImplementation(async (type, id, data) => {
@@ -1868,6 +1871,11 @@ describe('Output Service', () => {
   });
 
   describe('getLatestOutputHealth', () => {
+    let soClient: any;
+    beforeEach(() => {
+      soClient = getMockedSoClient();
+    });
+
     it('should return unknown state if no hits', async () => {
       esClientMock.search.mockResolvedValue({
         hits: {
@@ -1907,6 +1915,51 @@ describe('Output Service', () => {
         timestamp: '2023-11-30T14:25:31Z',
       });
     });
+
+    it('should apply range filter if updated_at available', async () => {
+      const updatedAt = '2023-11-30T14:25:31Z';
+      soClient.get.mockResolvedValue(
+        mockOutputSO(
+          'id',
+          {
+            type: 'remote_elasticsearch',
+          },
+          updatedAt
+        )
+      );
+
+      await outputService.getLatestOutputHealth(esClientMock, 'id');
+
+      expect((esClientMock.search.mock.lastCall?.[0] as any)?.query.bool.must).toEqual([
+        {
+          range: {
+            '@timestamp': {
+              gte: updatedAt,
+            },
+          },
+        },
+      ]);
+    });
+
+    it('should not apply range filter if updated_at is not available', async () => {
+      soClient.get.mockResolvedValue(
+        mockOutputSO('id', {
+          type: 'remote_elasticsearch',
+        })
+      );
+
+      await outputService.getLatestOutputHealth(esClientMock, 'id');
+
+      expect((esClientMock.search.mock.lastCall?.[0] as any)?.query.bool.must).toEqual([]);
+    });
+
+    it('should not apply range filter if output query returns error', async () => {
+      soClient.get.mockResolvedValue({ error: { message: 'error' } });
+
+      await outputService.getLatestOutputHealth(esClientMock, 'id');
+
+      expect((esClientMock.search.mock.lastCall?.[0] as any)?.query.bool.must).toEqual([]);
+    });
   });
 
   describe('backfillAllOutputPresets', () => {

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

@@ -1085,10 +1085,23 @@ class OutputService {
   }
 
   async getLatestOutputHealth(esClient: ElasticsearchClient, id: string): Promise<OutputHealth> {
+    const lastUpdateTime = await this.getOutputLastUpdateTime(id);
+
+    const mustFilter = [];
+    if (lastUpdateTime) {
+      mustFilter.push({
+        range: {
+          '@timestamp': {
+            gte: lastUpdateTime,
+          },
+        },
+      });
+    }
+
     const response = await esClient.search(
       {
         index: OUTPUT_HEALTH_DATA_STREAM,
-        query: { bool: { filter: { term: { output: id } } } },
+        query: { bool: { filter: { term: { output: id } }, must: mustFilter } },
         sort: { '@timestamp': 'desc' },
         size: 1,
       },
@@ -1109,6 +1122,24 @@ class OutputService {
       timestamp: latestHit['@timestamp'],
     };
   }
+
+  async getOutputLastUpdateTime(id: string): Promise<string | undefined> {
+    const outputSO = await this.encryptedSoClient.get<OutputSOAttributes>(
+      SAVED_OBJECT_TYPE,
+      outputIdToUuid(id)
+    );
+
+    if (outputSO.error) {
+      appContextService
+        .getLogger()
+        .debug(
+          `Error getting output ${id} SO, using updated_at:undefined, cause: ${outputSO.error.message}`
+        );
+      return undefined;
+    }
+
+    return outputSO.updated_at;
+  }
 }
 
 interface OutputHealth {

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

@@ -274,7 +274,7 @@ export default function (providerContext: FtrProviderContext) {
           document: {
             state: 'HEALTHY',
             message: '',
-            '@timestamp': '' + Date.parse('2023-11-29T14:25:31Z'),
+            '@timestamp': new Date(Date.now() - 1).toISOString(),
             output: defaultOutputId,
           },
         });
@@ -285,7 +285,7 @@ export default function (providerContext: FtrProviderContext) {
           document: {
             state: 'DEGRADED',
             message: 'connection error',
-            '@timestamp': '' + Date.parse('2023-11-30T14:25:31Z'),
+            '@timestamp': new Date().toISOString(),
             output: defaultOutputId,
           },
         });
@@ -297,7 +297,7 @@ export default function (providerContext: FtrProviderContext) {
             state: 'HEALTHY',
             message: '',
             '@timestamp': '' + Date.parse('2023-11-31T14:25:31Z'),
-            output: 'remote2',
+            output: ESOutputId,
           },
         });
       });
@@ -310,6 +310,13 @@ export default function (providerContext: FtrProviderContext) {
         expect(outputHealth.message).to.equal('connection error');
         expect(outputHealth.timestamp).not.to.be.empty();
       });
+      it('should not return output health if older than output last updated time', async () => {
+        const { body: outputHealth } = await supertest
+          .get(`/api/fleet/outputs/${ESOutputId}/health`)
+          .expect(200);
+
+        expect(outputHealth.state).to.equal('UNKNOWN');
+      });
     });
 
     describe('PUT /outputs/{outputId}', () => {