[8.6] [Stack Monitoring] Filter CCR list by currently viewed cluster (#148018) (#148363)

# Backport This will backport the following commits from `main` to `8.6`: - [[Stack Monitoring] Filter CCR list by currently viewed cluster (#148018)](https://github.com/elastic/kibana/pull/148018)  ### Questions ? Please refer to the [Backport tool documentation](https://github.com/sqren/backport)
2025-04-23 17:28:26 -04:00 · 2023-01-04 10:17:53 +01:00 · 2023-01-04 10:17:53 +01:00 · a02f42b9ac
commit a02f42b9ac
parent 01ae8ea533
6 changed files with 260 additions and 152 deletions
--- a/x-pack/plugins/monitoring/common/http_api/elasticsearch/post_elasticsearch_ccr.ts
+++ b/x-pack/plugins/monitoring/common/http_api/elasticsearch/post_elasticsearch_ccr.ts
@ -6,6 +6,7 @@
 */

 import * as rt from 'io-ts';
+import { ElasticsearchLegacySource, ElasticsearchMetricbeatSource } from '../../types/es';
 import { clusterUuidRT, ccsRT, timeRangeRT } from '../shared';

 export const postElasticsearchCcrRequestParamsRT = rt.type({
@ -25,6 +26,67 @@ export type PostElasticsearchCcrRequestPayload = rt.TypeOf<
  typeof postElasticsearchCcrRequestPayloadRT
 >;

-export const postElasticsearchCcrResponsePayloadRT = rt.type({
-  // TODO: add payload entries
+const errorRt = rt.partial({ error: rt.union([rt.string, rt.undefined]) });
+
+export const CcrShardRT = rt.type({
+  shardId: rt.number,
+  error: errorRt,
+  opsSynced: rt.number,
+  syncLagTime: rt.number,
+  syncLagOps: rt.number,
+  syncLagOpsLeader: rt.number,
+  syncLagOpsFollower: rt.number,
 });
+
+export type CcrShard = rt.TypeOf<typeof CcrShardRT>;
+
+export const postElasticsearchCcrResponsePayloadRT = rt.array(
+  rt.type({
+    id: rt.string,
+    index: rt.string,
+    follows: rt.string,
+    shards: rt.array(CcrShardRT),
+    error: errorRt,
+    opsSynced: rt.number,
+    syncLagTime: rt.number,
+    syncLagOps: rt.number,
+  })
+);
+
+export type PostElasticsearchCcrResponsePayload = rt.TypeOf<
+  typeof postElasticsearchCcrResponsePayloadRT
+>;
+
+interface ValueObj<T> {
+  value: T;
+}
+export interface CcrShardBucket {
+  key: number;
+  ops_synced: ValueObj<string>;
+  lag_ops: ValueObj<number>;
+  leader_lag_ops: ValueObj<number>;
+  follower_lag_ops: ValueObj<number>;
+}
+
+export interface CcrBucket {
+  key: string;
+  by_shard_id: {
+    buckets: CcrShardBucket[];
+  };
+  leader_index: {
+    buckets: Array<{
+      remote_cluster: {
+        buckets: Array<{
+          key: string;
+        }>;
+      };
+    }>;
+  };
+}
+
+export interface CcrFullStats {
+  [key: string]: Array<
+    | NonNullable<ElasticsearchLegacySource['ccr_stats']>
+    | NonNullable<ElasticsearchMetricbeatSource['elasticsearch']>['ccr']
+  >;
+}
--- a/x-pack/plugins/monitoring/public/application/pages/elasticsearch/ccr_page.tsx
+++ b/x-pack/plugins/monitoring/public/application/pages/elasticsearch/ccr_page.tsx
@ -97,7 +97,7 @@ export const ElasticsearchCcrPage: React.FC<ComponentProps> = ({ clusters }) =>
        render={({ flyoutComponent, bottomBarComponent }: SetupModeProps) => (
          <SetupModeContext.Provider value={{ setupModeSupported: true }}>
            {flyoutComponent}
-            <Ccr data={data.data} alerts={alerts} />
+            <Ccr data={data} alerts={alerts} />
            {bottomBarComponent}
          </SetupModeContext.Provider>
        )}
--- a/x-pack/plugins/monitoring/server/routes/api/v1/elasticsearch/ccr.ts
+++ b/x-pack/plugins/monitoring/server/routes/api/v1/elasticsearch/ccr.ts
@ -13,7 +13,12 @@ import {
 import {
  postElasticsearchCcrRequestParamsRT,
  postElasticsearchCcrRequestPayloadRT,
+  PostElasticsearchCcrResponsePayload,
  postElasticsearchCcrResponsePayloadRT,
+  CcrBucket,
+  CcrFullStats,
+  CcrShard,
+  CcrShardBucket,
 } from '../../../../../common/http_api/elasticsearch';
 import { TimeRange } from '../../../../../common/http_api/shared';
 import {
@ -24,7 +29,7 @@ import {
 import { MonitoringConfig } from '../../../../config';
 import { createValidationFunction } from '../../../../lib/create_route_validation_function';
 import { handleError } from '../../../../lib/errors/handle_error';
-import { LegacyRequest, MonitoringCore } from '../../../../types';
+import { MonitoringCore } from '../../../../types';

 function getBucketScript(max: string, min: string) {
  return {
@ -38,12 +43,15 @@ function getBucketScript(max: string, min: string) {
  };
 }

-function buildRequest(
-  req: LegacyRequest<unknown, unknown, { timeRange: TimeRange }>,
-  config: MonitoringConfig,
-  esIndexPattern: string
-) {
-  const { min, max } = req.payload.timeRange;
+interface BuildRequestParams {
+  clusterUuid: string;
+  config: MonitoringConfig;
+  esIndexPattern: string;
+  timeRange: TimeRange;
+}
+
+function buildRequest({ clusterUuid, config, esIndexPattern, timeRange }: BuildRequestParams) {
+  const { min, max } = timeRange;
  const maxBucketSize = config.ui.max_bucket_size;
  const aggs = {
    ops_synced_max: {
@ -125,6 +133,13 @@ function buildRequest(
      query: {
        bool: {
          must: [
+            {
+              term: {
+                cluster_uuid: {
+                  value: clusterUuid,
+                },
+              },
+            },
            {
              bool: {
                should: [
@ -210,6 +225,41 @@ function buildRequest(
  };
 }

+function buildShardStats({
+  fullStats,
+  bucket,
+  shardBucket,
+}: {
+  bucket: CcrBucket;
+  fullStats: CcrFullStats;
+  shardBucket: CcrShardBucket;
+}) {
+  const fullStat: any = fullStats[`${bucket.key}:${shardBucket.key}`][0];
+  const fullLegacyStat: ElasticsearchLegacySource = fullStat._source?.ccr_stats
+    ? fullStat._source
+    : null;
+  const fullMbStat: ElasticsearchMetricbeatSource = fullStat._source?.elasticsearch?.ccr
+    ? fullStat._source
+    : null;
+  const readExceptions =
+    fullLegacyStat?.ccr_stats?.read_exceptions ??
+    fullMbStat?.elasticsearch?.ccr?.read_exceptions ??
+    [];
+  const shardStat = {
+    shardId: shardBucket.key,
+    error: readExceptions.length ? readExceptions[0].exception?.type : null,
+    opsSynced: get(shardBucket, 'ops_synced.value'),
+    syncLagTime:
+      fullLegacyStat?.ccr_stats?.time_since_last_read_millis ??
+      fullMbStat?.elasticsearch?.ccr?.follower?.time_since_last_read?.ms,
+    syncLagOps: get(shardBucket, 'lag_ops.value'),
+    syncLagOpsLeader: get(shardBucket, 'leader_lag_ops.value'),
+    syncLagOpsFollower: get(shardBucket, 'follower_lag_ops.value'),
+  };
+
+  return shardStat;
+}
+
 export function ccrRoute(server: MonitoringCore) {
  const validateParams = createValidationFunction(postElasticsearchCcrRequestParamsRT);
  const validateBody = createValidationFunction(postElasticsearchCcrRequestPayloadRT);
@ -224,6 +274,7 @@ export function ccrRoute(server: MonitoringCore) {
    async handler(req) {
      const config = server.config;
      const ccs = req.payload.ccs;
+      const { clusterUuid } = req.params;
      const dataset = 'ccr';
      const moduleType = 'elasticsearch';
      const esIndexPattern = getIndexPatterns({
@ -235,19 +286,19 @@ export function ccrRoute(server: MonitoringCore) {

      try {
        const { callWithRequest } = req.server.plugins.elasticsearch.getCluster('monitoring');
-        const params = buildRequest(req, config, esIndexPattern);
+        const params = buildRequest({
+          clusterUuid,
+          config,
+          esIndexPattern,
+          timeRange: req.payload.timeRange,
+        });
        const response: ElasticsearchResponse = await callWithRequest(req, 'search', params);

        if (!response || Object.keys(response).length === 0) {
-          return { data: [] };
+          return [];
        }

-        const fullStats: {
-          [key: string]: Array<
-            | NonNullable<ElasticsearchLegacySource['ccr_stats']>
-            | NonNullable<ElasticsearchMetricbeatSource['elasticsearch']>['ccr']
-          >;
-        } =
+        const fullStats: CcrFullStats =
          response.hits?.hits.reduce((accum, hit) => {
            const innerHits = hit.inner_hits?.by_shard.hits?.hits ?? [];
            const grouped = groupBy(innerHits, (innerHit) => {
@ -265,79 +316,39 @@ export function ccrRoute(server: MonitoringCore) {
          }, {}) ?? {};

        const buckets = response.aggregations?.by_follower_index.buckets ?? [];
-        const data = buckets.reduce((accum: any, bucket: any) => {
+
+        const data: PostElasticsearchCcrResponsePayload = buckets.map((bucket: CcrBucket) => {
          const leaderIndex = get(bucket, 'leader_index.buckets[0].key');
          const remoteCluster = get(
            bucket,
            'leader_index.buckets[0].remote_cluster.buckets[0].key'
          );
          const follows = remoteCluster ? `${leaderIndex} on ${remoteCluster}` : leaderIndex;
-          const stat: {
-            [key: string]: any;
-            shards: Array<{
-              error?: string;
-              opsSynced: number;
-              syncLagTime: number;
-              syncLagOps: number;
-            }>;
-          } = {
+
+          const shards: CcrShard[] = get(bucket, 'by_shard_id.buckets').map(
+            (shardBucket: CcrShardBucket) => buildShardStats({ bucket, fullStats, shardBucket })
+          );
+
+          const error = (shards.find((shard) => shard.error) || {}).error;
+          const opsSynced = shards.reduce((sum, curr) => sum + curr.opsSynced, 0);
+          const syncLagTime = shards.reduce((max, curr) => Math.max(max, curr.syncLagTime), 0);
+          const syncLagOps = shards.reduce((max, curr) => Math.max(max, curr.syncLagOps), 0);
+
+          const stat = {
            id: bucket.key,
            index: bucket.key,
            follows,
-            shards: [],
-            error: undefined,
-            opsSynced: undefined,
-            syncLagTime: undefined,
-            syncLagOps: undefined,
+            shards,
+            error,
+            opsSynced,
+            syncLagTime,
+            syncLagOps,
          };

-          stat.shards = get(bucket, 'by_shard_id.buckets').reduce(
-            (accum2: any, shardBucket: any) => {
-              const fullStat: any = fullStats[`${bucket.key}:${shardBucket.key}`][0];
-              const fullLegacyStat: ElasticsearchLegacySource = fullStat._source?.ccr_stats
-                ? fullStat._source
-                : null;
-              const fullMbStat: ElasticsearchMetricbeatSource = fullStat._source?.elasticsearch?.ccr
-                ? fullStat._source
-                : null;
-              const readExceptions =
-                fullLegacyStat?.ccr_stats?.read_exceptions ??
-                fullMbStat?.elasticsearch?.ccr?.read_exceptions ??
-                [];
-              const shardStat = {
-                shardId: shardBucket.key,
-                error: readExceptions.length ? readExceptions[0].exception?.type : null,
-                opsSynced: get(shardBucket, 'ops_synced.value'),
-                syncLagTime:
-                  // @ts-ignore
-                  fullLegacyStat?.ccr_stats?.time_since_last_read_millis ??
-                  fullMbStat?.elasticsearch?.ccr?.follower?.time_since_last_read?.ms,
-                syncLagOps: get(shardBucket, 'lag_ops.value'),
-                syncLagOpsLeader: get(shardBucket, 'leader_lag_ops.value'),
-                syncLagOpsFollower: get(shardBucket, 'follower_lag_ops.value'),
-              };
-              accum2.push(shardStat);
-              return accum2;
-            },
-            []
-          );
+          return stat;
+        });

-          stat.error = (stat.shards.find((shard) => shard.error) || {}).error;
-          stat.opsSynced = stat.shards.reduce((sum, { opsSynced }) => sum + opsSynced, 0);
-          stat.syncLagTime = stat.shards.reduce(
-            (max, { syncLagTime }) => Math.max(max, syncLagTime),
-            0
-          );
-          stat.syncLagOps = stat.shards.reduce(
-            (max, { syncLagOps }) => Math.max(max, syncLagOps),
-            0
-          );
-
-          accum.push(stat);
-          return accum;
-        }, []);
-
-        return postElasticsearchCcrResponsePayloadRT.encode({ data });
+        return postElasticsearchCcrResponsePayloadRT.encode(data);
      } catch (err) {
        return handleError(err, req);
      }
--- a/x-pack/test/api_integration/apis/monitoring/elasticsearch/ccr.js
+++ b/x-pack/test/api_integration/apis/monitoring/elasticsearch/ccr.js
@ -29,7 +29,7 @@ export default function ({ getService }) {

    it('should return all followers and a grouping of stats by follower index', async () => {
      const { body } = await supertest
-        .post('/api/monitoring/v1/clusters/YCxj-RAgSZCP6GuOQ8M1EQ/elasticsearch/ccr')
+        .post('/api/monitoring/v1/clusters/vX4lH4C6QmyrJeYrvKr0-A/elasticsearch/ccr')
        .set('kbn-xsrf', 'xxx')
        .send({
          timeRange,
@ -38,5 +38,17 @@ export default function ({ getService }) {

      expect(body).to.eql(ccrFixture);
    });
+
+    it('should return an empty list of followers if the cluster_uuid does not have any match', async () => {
+      const { body } = await supertest
+        .post('/api/monitoring/v1/clusters/random_uuid/elasticsearch/ccr')
+        .set('kbn-xsrf', 'xxx')
+        .send({
+          timeRange,
+        })
+        .expect(200);
+
+      expect(body).to.eql([]);
+    });
  });
 }
--- a/x-pack/test/api_integration/apis/monitoring/elasticsearch/ccr_mb.js
+++ b/x-pack/test/api_integration/apis/monitoring/elasticsearch/ccr_mb.js
@ -32,7 +32,7 @@ export default function ({ getService }) {

        it('should return all followers and a grouping of stats by follower index', async () => {
          const { body } = await supertest
-            .post('/api/monitoring/v1/clusters/YCxj-RAgSZCP6GuOQ8M1EQ/elasticsearch/ccr')
+            .post('/api/monitoring/v1/clusters/vX4lH4C6QmyrJeYrvKr0-A/elasticsearch/ccr')
            .set('kbn-xsrf', 'xxx')
            .send({
              timeRange,
@ -41,6 +41,18 @@ export default function ({ getService }) {

          expect(body).to.eql(ccrFixture);
        });
+
+        it('should return an empty list of followers if the cluster_uuid does not have any match', async () => {
+          const { body } = await supertest
+            .post('/api/monitoring/v1/clusters/random_uuid/elasticsearch/ccr')
+            .set('kbn-xsrf', 'xxx')
+            .send({
+              timeRange,
+            })
+            .expect(200);
+
+          expect(body).to.eql([]);
+        });
      });
    });
  });
--- a/x-pack/test/api_integration/apis/monitoring/elasticsearch/fixtures/ccr.json
+++ b/x-pack/test/api_integration/apis/monitoring/elasticsearch/fixtures/ccr.json
@ -1,83 +1,94 @@
-{
-  "data": [{
+[
+  {
    "id": "follower2",
    "index": "follower2",
    "follows": "leader2",
-    "shards": [{
-      "shardId": 0,
-      "error": null,
-      "opsSynced": 52,
-      "syncLagTime": 4900,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 1,
-      "error": null,
-      "opsSynced": 47,
-      "syncLagTime": 9919,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 2,
-      "error": null,
-      "opsSynced": 51,
-      "syncLagTime": 14929,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 3,
-      "error": null,
-      "opsSynced": 50,
-      "syncLagTime": 39933,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 4,
-      "error": null,
-      "opsSynced": 55,
-      "syncLagTime": 49923,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }],
+    "shards": [
+      {
+        "shardId": 0,
+        "error": null,
+        "opsSynced": 52,
+        "syncLagTime": 4900,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 1,
+        "error": null,
+        "opsSynced": 47,
+        "syncLagTime": 9919,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 2,
+        "error": null,
+        "opsSynced": 51,
+        "syncLagTime": 14929,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 3,
+        "error": null,
+        "opsSynced": 50,
+        "syncLagTime": 39933,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 4,
+        "error": null,
+        "opsSynced": 55,
+        "syncLagTime": 49923,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      }
+    ],
    "opsSynced": 255,
    "syncLagTime": 49923,
    "syncLagOps": 0
-  }, {
+  },
+  {
    "id": "follower",
    "index": "follower",
    "follows": "leader",
-    "shards": [{
-      "shardId": 0,
-      "error": null,
-      "opsSynced": 85,
-      "syncLagTime": 19886,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 1,
-      "error": null,
-      "opsSynced": 94,
-      "syncLagTime": 4901,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }, {
-      "shardId": 2,
-      "error": null,
-      "opsSynced": 76,
-      "syncLagTime": 14899,
-      "syncLagOps": 0,
-      "syncLagOpsLeader": 0,
-      "syncLagOpsFollower": 0
-    }],
+    "shards": [
+      {
+        "shardId": 0,
+        "error": null,
+        "opsSynced": 85,
+        "syncLagTime": 19886,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 1,
+        "error": null,
+        "opsSynced": 94,
+        "syncLagTime": 4901,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      },
+      {
+        "shardId": 2,
+        "error": null,
+        "opsSynced": 76,
+        "syncLagTime": 14899,
+        "syncLagOps": 0,
+        "syncLagOpsLeader": 0,
+        "syncLagOpsFollower": 0
+      }
+    ],
    "opsSynced": 255,
    "syncLagTime": 19886,
    "syncLagOps": 0
-  }]
-}
+  }
+]