[Cloud Posture] CSPM telemetry accounts stats

x-pack/plugins/cloud_security_posture/server/lib/telemetry/collectors/accounts_stats_collector.ts

@@ -0,0 +1,241 @@
+/*
+ * 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 type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import type { Logger } from '@kbn/core/server';
+import type { SearchRequest } from '@elastic/elasticsearch/lib/api/types';
+import { calculatePostureScore } from '../../../routes/compliance_dashboard/get_stats';
+import type { CspmAccountsStats } from './types';
+import { LATEST_FINDINGS_INDEX_DEFAULT_NS } from '../../../../common/constants';
+
+interface Value {
+  value: number;
+}
+
+interface DocCount {
+  doc_count: number;
+}
+
+interface BenchmarkName {
+  metrics: { 'rule.benchmark.name': string };
+}
+
+interface BenchmarkId {
+  metrics: { 'rule.benchmark.id': string };
+}
+
+interface BenchmarkVersion {
+  metrics: { 'rule.benchmark.version': string };
+}
+
+interface AccountsStats {
+  accounts: {
+    buckets: AccountEntity[];
+  };
+}
+interface AccountEntity {
+  key: string; // account_id
+  doc_count: number; // latest findings doc count
+  passed_findings_count: DocCount;
+  failed_findings_count: DocCount;
+  benchmark_name: { top: BenchmarkName[] };
+  benchmark_id: { top: BenchmarkId[] };
+  benchmark_version: { top: BenchmarkVersion[] };
+  agents_count: Value;
+  nodes_count: Value;
+  pods_count: Value;
+  resources: {
+    pods_count: Value;
+  };
+}
+
+const getAccountsStatsQuery = (index: string): SearchRequest => ({
+  index,
+  query: {
+    match_all: {},
+  },
+  aggs: {
+    accounts: {
+      terms: {
+        field: 'cluster_id',
+        order: {
+          _count: 'desc',
+        },
+        size: 100,
+      },
+      aggs: {
+        nodes_count: {
+          cardinality: {
+            field: 'host.name',
+          },
+        },
+        agents_count: {
+          cardinality: {
+            field: 'agent.id',
+          },
+        },
+        benchmark_id: {
+          top_metrics: {
+            metrics: {
+              field: 'rule.benchmark.id',
+            },
+            size: 1,
+            sort: {
+              '@timestamp': 'desc',
+            },
+          },
+        },
+        benchmark_version: {
+          top_metrics: {
+            metrics: {
+              field: 'rule.benchmark.version',
+            },
+            size: 1,
+            sort: {
+              '@timestamp': 'desc',
+            },
+          },
+        },
+        benchmark_name: {
+          top_metrics: {
+            metrics: {
+              field: 'rule.benchmark.name',
+            },
+            size: 1,
+            sort: {
+              '@timestamp': 'desc',
+            },
+          },
+        },
+        passed_findings_count: {
+          filter: {
+            bool: {
+              filter: [
+                {
+                  bool: {
+                    should: [
+                      {
+                        term: {
+                          'result.evaluation': 'passed',
+                        },
+                      },
+                    ],
+                    minimum_should_match: 1,
+                  },
+                },
+              ],
+            },
+          },
+        },
+        failed_findings_count: {
+          filter: {
+            bool: {
+              filter: [
+                {
+                  bool: {
+                    should: [
+                      {
+                        term: {
+                          'result.evaluation': 'failed',
+                        },
+                      },
+                    ],
+                    minimum_should_match: 1,
+                  },
+                },
+              ],
+            },
+          },
+        },
+        resources: {
+          filter: {
+            bool: {
+              filter: [
+                {
+                  bool: {
+                    should: [
+                      {
+                        term: {
+                          'resource.sub_type': 'Pod',
+                        },
+                      },
+                    ],
+                    minimum_should_match: 1,
+                  },
+                },
+              ],
+            },
+          },
+          aggs: {
+            pods_count: {
+              cardinality: {
+                field: 'resource.id',
+              },
+            },
+          },
+        },
+      },
+    },
+  },
+
+  size: 0,
+  _source: false,
+});
+
+const getCspmAccountsStats = (
+  aggregatedResourcesStats: AccountsStats,
+  logger: Logger
+): CspmAccountsStats[] => {
+  const accounts = aggregatedResourcesStats.accounts.buckets;
+
+  const cspmAccountsStats = accounts.map((account) => ({
+    account_id: account.key,
+    latest_findings_doc_count: account.doc_count,
+    posture_score: calculatePostureScore(
+      account.passed_findings_count.doc_count,
+      account.failed_findings_count.doc_count
+    ),
+    passed_findings_count: account.passed_findings_count.doc_count,
+    failed_findings_count: account.failed_findings_count.doc_count,
+    benchmark_name: account.benchmark_name.top[0].metrics['rule.benchmark.name'],
+    benchmark_id: account.benchmark_id.top[0].metrics['rule.benchmark.id'],
+    benchmark_version: account.benchmark_version.top[0].metrics['rule.benchmark.version'],
+    agents_count: account.agents_count.value,
+    nodes_count: account.nodes_count.value,
+    pods_count: account.resources.pods_count.value,
+  }));
+  logger.info('CSPM telemetry: accounts stats was sent');
+
+  return cspmAccountsStats;
+};
+
+export const getAccountsStats = async (
+  esClient: ElasticsearchClient,
+  logger: Logger
+): Promise<CspmAccountsStats[]> => {
+  try {
+    const isIndexExists = await esClient.indices.exists({
+      index: LATEST_FINDINGS_INDEX_DEFAULT_NS,
+    });
+
+    if (isIndexExists) {
+      const accountsStatsResponse = await esClient.search<unknown, AccountsStats>(
+        getAccountsStatsQuery(LATEST_FINDINGS_INDEX_DEFAULT_NS)
+      );
+
+      const cspmAccountsStats = accountsStatsResponse.aggregations
+        ? getCspmAccountsStats(accountsStatsResponse.aggregations, logger)
+        : [];
+
+      return cspmAccountsStats;
+    }
+
+    return [];
+  } catch (e) {
+    logger.error(`Failed to get resources stats ${e}`);
+    return [];
+  }
+};

x-pack/plugins/cloud_security_posture/server/lib/telemetry/collectors/indices_stats_collector.ts

@@ -4,29 +4,36 @@
  * 2.0; you may not use this file except in compliance with the Elastic License
  * 2.0.
  */
-import { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
 import type { Logger } from '@kbn/core/server';
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
+import type { CspmIndicesStats, IndexStats } from './types';
 import {
   BENCHMARK_SCORE_INDEX_DEFAULT_NS,
   FINDINGS_INDEX_DEFAULT_NS,
   LATEST_FINDINGS_INDEX_DEFAULT_NS,
 } from '../../../../common/constants';
-import type { CspmIndicesStats, IndexStats } from './types';
 
-export const getIndicesStats = async (
+const getIndexDocCount = (esClient: ElasticsearchClient, index: string) =>
+  esClient.indices.stats({ index });
+
+const getLatestDocTimestamp = async (
   esClient: ElasticsearchClient,
-  logger: Logger
-): Promise<CspmIndicesStats> => {
-  const [findings, latestFindings, score] = await Promise.all([
-    getIndexStats(esClient, FINDINGS_INDEX_DEFAULT_NS, logger),
-    getIndexStats(esClient, LATEST_FINDINGS_INDEX_DEFAULT_NS, logger),
-    getIndexStats(esClient, BENCHMARK_SCORE_INDEX_DEFAULT_NS, logger),
-  ]);
-  return {
-    findings,
-    latest_findings: latestFindings,
-    score,
-  };
+  index: string
+): Promise<string | null> => {
+  const latestTimestamp = await esClient.search({
+    index,
+    query: {
+      match_all: {},
+    },
+    sort: '@timestamp:desc',
+    size: 1,
+    fields: ['@timestamp'],
+    _source: false,
+  });
+
+  const latestEventTimestamp = latestTimestamp.hits?.hits[0]?.fields;
+
+  return latestEventTimestamp ? latestEventTimestamp['@timestamp'][0] : null;
 };
 
 const getIndexStats = async (
@@ -60,25 +67,18 @@ const getIndexStats = async (
   }
 };
 
-const getIndexDocCount = (esClient: ElasticsearchClient, index: string) =>
-  esClient.indices.stats({ index });
-
-const getLatestDocTimestamp = async (
+export const getIndicesStats = async (
   esClient: ElasticsearchClient,
-  index: string
-): Promise<string | null> => {
-  const latestTimestamp = await esClient.search({
-    index,
-    query: {
-      match_all: {},
-    },
-    sort: '@timestamp:desc',
-    size: 1,
-    fields: ['@timestamp'],
-    _source: false,
-  });
-
-  const latestEventTimestamp = latestTimestamp.hits?.hits[0]?.fields;
-
-  return latestEventTimestamp ? latestEventTimestamp['@timestamp'][0] : null;
+  logger: Logger
+): Promise<CspmIndicesStats> => {
+  const [findings, latestFindings, score] = await Promise.all([
+    getIndexStats(esClient, FINDINGS_INDEX_DEFAULT_NS, logger),
+    getIndexStats(esClient, LATEST_FINDINGS_INDEX_DEFAULT_NS, logger),
+    getIndexStats(esClient, BENCHMARK_SCORE_INDEX_DEFAULT_NS, logger),
+  ]);
+  return {
+    findings,
+    latest_findings: latestFindings,
+    score,
+  };
 };

x-pack/plugins/cloud_security_posture/server/lib/telemetry/collectors/register.ts

@@ -11,6 +11,7 @@ import { getIndicesStats } from './indices_stats_collector';
 import { getResourcesStats } from './resources_stats_collector';
 import { cspmUsageSchema } from './schema';
 import { CspmUsage } from './types';
+import { getAccountsStats } from './accounts_stats_collector';
 
 export function registerCspmUsageCollector(
   logger: Logger,
@@ -26,13 +27,15 @@ export function registerCspmUsageCollector(
     type: 'cloud_security_posture',
     isReady: () => true,
     fetch: async (collectorFetchContext: CollectorFetchContext) => {
-      const [indicesStats, resourcesStats] = await Promise.all([
+      const [indicesStats, accountsStats, resourcesStats] = await Promise.all([
         getIndicesStats(collectorFetchContext.esClient, logger),
-        await getResourcesStats(collectorFetchContext.esClient, logger),
+        getAccountsStats(collectorFetchContext.esClient, logger),
+        getResourcesStats(collectorFetchContext.esClient, logger),
       ]);
 
       return {
         indices: indicesStats,
+        accounts_stats: accountsStats,
         resources_stats: resourcesStats,
       };
     },

x-pack/plugins/cloud_security_posture/server/lib/telemetry/collectors/resources_stats_collector.ts

@@ -110,7 +110,10 @@ const getEvaluationStats = (resourceSubType: ResourceSubType) => {
   return { passed_findings_count: passed, failed_findings_count: failed };
 };
 
-const getCspmResourcesStats = (aggregatedResourcesStats: ResourcesStats): CspmResourcesStats[] => {
+const getCspmResourcesStats = (
+  aggregatedResourcesStats: ResourcesStats,
+  logger: Logger
+): CspmResourcesStats[] => {
   const accounts = aggregatedResourcesStats.accounts.buckets;
 
   const resourcesStats = accounts.map((account) => {
@@ -129,6 +132,8 @@ const getCspmResourcesStats = (aggregatedResourcesStats: ResourcesStats): CspmRe
       });
     });
   });
+  logger.info('CSPM telemetry: resources stats was sent');
+
   return resourcesStats.flat(2);
 };
 
@@ -147,7 +152,7 @@ export const getResourcesStats = async (
       );
 
       const cspmResourcesStats = resourcesStatsResponse.aggregations
-        ? getCspmResourcesStats(resourcesStatsResponse.aggregations)
+        ? getCspmResourcesStats(resourcesStatsResponse.aggregations, logger)
         : [];
 
       return cspmResourcesStats;