[Cloud Security][usage collector] enhance cloud security metering (#163828)

x-pack/plugins/security_solution_serverless/server/cloud_security/cloud_security_metering.ts

@@ -10,9 +10,11 @@ import {
   KSPM_POLICY_TEMPLATE,
   CNVM_POLICY_TEMPLATE,
 } from '@kbn/cloud-security-posture-plugin/common/constants';
+import { ProductLine } from '../../common/product';
 import { getCloudSecurityUsageRecord } from './cloud_security_metering_task';
 import type { PostureType } from './types';
-import type { MeteringCallbackInput, UsageRecord } from '../types';
+import type { MeteringCallbackInput, Tier, UsageRecord } from '../types';
+import type { ServerlessSecurityConfig } from '../config';
 
 export const CLOUD_SECURITY_TASK_TYPE = 'cloud_security';
 export const AGGREGATION_PRECISION_THRESHOLD = 40000;
@@ -23,40 +25,50 @@ export const cloudSecurityMetringCallback = async ({
   logger,
   taskId,
   lastSuccessfulReport,
+  config,
 }: MeteringCallbackInput): Promise<UsageRecord[]> => {
-  const projectId = cloudSetup?.serverless?.projectId || 'missing project id';
+  const projectId = cloudSetup?.serverless?.projectId || 'missing_project_id';
 
   if (!cloudSetup?.serverless?.projectId) {
     logger.error('no project id found');
   }
 
-  try {
-    const cloudSecurityUsageRecords: UsageRecord[] = [];
+  const tier: Tier = getCloudProductTier(config);
 
+  try {
     const postureTypes: PostureType[] = [
       CSPM_POLICY_TEMPLATE,
       KSPM_POLICY_TEMPLATE,
       CNVM_POLICY_TEMPLATE,
     ];
 
-    for (const postureType of postureTypes) {
-      const usageRecord = await getCloudSecurityUsageRecord({
-        esClient,
-        projectId,
-        logger,
-        taskId,
-        lastSuccessfulReport,
-        postureType,
-      });
+    const cloudSecurityUsageRecords = await Promise.all(
+      postureTypes.map((postureType) =>
+        getCloudSecurityUsageRecord({
+          esClient,
+          projectId,
+          logger,
+          taskId,
+          lastSuccessfulReport,
+          postureType,
+          tier,
+        })
+      )
+    );
 
-      if (usageRecord) {
-        cloudSecurityUsageRecords.push(usageRecord);
-      }
-    }
-
-    return cloudSecurityUsageRecords;
+    // remove any potential undefined values from the array,
+    return cloudSecurityUsageRecords.filter(Boolean) as UsageRecord[];
   } catch (err) {
     logger.error(`Failed to fetch Cloud Security metering data ${err}`);
     return [];
   }
 };
+
+export const getCloudProductTier = (config: ServerlessSecurityConfig): Tier => {
+  const cloud = config.productTypes.find(
+    (productType) => productType.product_line === ProductLine.cloud
+  );
+  const tier = cloud ? cloud.product_tier : 'none';
+
+  return tier;
+};

x-pack/plugins/security_solution_serverless/server/cloud_security/cloud_security_metering_task.test.ts

@@ -11,9 +11,12 @@ import {
   KSPM_POLICY_TEMPLATE,
   CNVM_POLICY_TEMPLATE,
 } from '@kbn/cloud-security-posture-plugin/common/constants';
-import { CLOUD_SECURITY_TASK_TYPE } from './cloud_security_metering';
+import { CLOUD_SECURITY_TASK_TYPE, getCloudProductTier } from './cloud_security_metering';
 import { getCloudSecurityUsageRecord } from './cloud_security_metering_task';
+
+import type { ServerlessSecurityConfig } from '../config';
 import type { PostureType } from './types';
+import type { ProductTier } from '../../common/product';
 
 const mockEsClient = elasticsearchServiceMock.createStart().client.asInternalUser;
 const logger: ReturnType<typeof loggingSystemMock.createLogger> = loggingSystemMock.createLogger();
@@ -25,7 +28,7 @@ const postureTypes: PostureType[] = [
   CNVM_POLICY_TEMPLATE,
 ];
 
-describe('getCspmUsageRecord', () => {
+describe('getCloudSecurityUsageRecord', () => {
   beforeEach(() => {
     jest.resetAllMocks();
   });
@@ -38,6 +41,8 @@ describe('getCspmUsageRecord', () => {
     const taskId = chance.guid();
     const postureType = CSPM_POLICY_TEMPLATE;
 
+    const tier = 'essentials' as ProductTier;
+
     const result = await getCloudSecurityUsageRecord({
       esClient: mockEsClient,
       projectId,
@@ -45,6 +50,7 @@ describe('getCspmUsageRecord', () => {
       taskId,
       lastSuccessfulReport: new Date(),
       postureType,
+      tier,
     });
 
     expect(result).toBeUndefined();
@@ -54,9 +60,14 @@ describe('getCspmUsageRecord', () => {
     'should return usageRecords with correct values for cspm and kspm when Elasticsearch response has aggregations',
     async (postureType) => {
       // @ts-ignore
-      mockEsClient.search.mockResolvedValue({
+      mockEsClient.search.mockResolvedValueOnce({
+        hits: { hits: [{ _id: 'someRecord', _index: 'mockIndex' }] }, // mocking for indexHasDataInDateRange
+      });
+
+      // @ts-ignore
+      mockEsClient.search.mockResolvedValueOnce({
         aggregations: {
-          unique_resources: {
+          unique_assets: {
             value: 10,
           },
           min_timestamp: {
@@ -68,6 +79,8 @@ describe('getCspmUsageRecord', () => {
       const projectId = chance.guid();
       const taskId = chance.guid();
 
+      const tier = 'essentials' as ProductTier;
+
       const result = await getCloudSecurityUsageRecord({
         esClient: mockEsClient,
         projectId,
@@ -75,10 +88,11 @@ describe('getCspmUsageRecord', () => {
         taskId,
         lastSuccessfulReport: new Date(),
         postureType,
+        tier,
       });
 
       expect(result).toEqual({
-        id: `${CLOUD_SECURITY_TASK_TYPE}:${postureType}`,
+        id: expect.stringContaining(`${CLOUD_SECURITY_TASK_TYPE}_${postureType}_${projectId}`),
         usage_timestamp: '2023-07-30T15:11:41.738Z',
         creation_timestamp: expect.any(String), // Expect a valid ISO string
         usage: {
@@ -90,6 +104,9 @@ describe('getCspmUsageRecord', () => {
         source: {
           id: taskId,
           instance_group_id: projectId,
+          metadata: {
+            tier: 'essentials',
+          },
         },
       });
     }
@@ -103,6 +120,8 @@ describe('getCspmUsageRecord', () => {
     const taskId = chance.guid();
     const postureType = CSPM_POLICY_TEMPLATE;
 
+    const tier = 'essentials' as ProductTier;
+
     const result = await getCloudSecurityUsageRecord({
       esClient: mockEsClient,
       projectId,
@@ -110,6 +129,7 @@ describe('getCspmUsageRecord', () => {
       taskId,
       lastSuccessfulReport: new Date(),
       postureType,
+      tier,
     });
 
     expect(result).toBeUndefined();
@@ -123,6 +143,8 @@ describe('getCspmUsageRecord', () => {
     const taskId = chance.guid();
     const postureType = CSPM_POLICY_TEMPLATE;
 
+    const tier = 'essentials' as ProductTier;
+
     const result = await getCloudSecurityUsageRecord({
       esClient: mockEsClient,
       projectId,
@@ -130,8 +152,38 @@ describe('getCspmUsageRecord', () => {
       taskId,
       lastSuccessfulReport: new Date(),
       postureType,
+      tier,
     });
 
     expect(result).toBeUndefined();
   });
 });
+
+describe('should return the relevant product tier', () => {
+  it('should return the relevant product tier for cloud product line', async () => {
+    const serverlessSecurityConfig = {
+      enabled: true,
+      developer: {},
+      productTypes: [
+        { product_line: 'endpoint', product_tier: 'essentials' },
+        { product_line: 'cloud', product_tier: 'complete' },
+      ],
+    } as unknown as ServerlessSecurityConfig;
+
+    const tier = getCloudProductTier(serverlessSecurityConfig);
+
+    expect(tier).toBe('complete');
+  });
+
+  it('should return none tier in case cloud product line is missing ', async () => {
+    const serverlessSecurityConfig = {
+      enabled: true,
+      developer: {},
+      productTypes: [{ product_line: 'endpoint', product_tier: 'complete' }],
+    } as unknown as ServerlessSecurityConfig;
+
+    const tier = getCloudProductTier(serverlessSecurityConfig);
+
+    expect(tier).toBe('none');
+  });
+});

x-pack/plugins/security_solution_serverless/server/cloud_security/cloud_security_metering_task.ts

@@ -10,10 +10,9 @@ import {
   CSPM_POLICY_TEMPLATE,
   KSPM_POLICY_TEMPLATE,
   LATEST_FINDINGS_INDEX_PATTERN,
-  LATEST_FINDINGS_RETENTION_POLICY,
   LATEST_VULNERABILITIES_INDEX_PATTERN,
-  LATEST_VULNERABILITIES_RETENTION_POLICY,
 } from '@kbn/cloud-security-posture-plugin/common/constants';
+import type { ElasticsearchClient } from '@kbn/core-elasticsearch-server';
 import type { UsageRecord } from '../types';
 
 import {
@@ -27,18 +26,20 @@ import type {
   ResourceCountAggregation,
 } from './types';
 
+const ASSETS_SAMPLE_GRANULARITY = '24h';
+
 const queryParams = {
   [CSPM_POLICY_TEMPLATE]: {
     index: LATEST_FINDINGS_INDEX_PATTERN,
-    timeRange: LATEST_FINDINGS_RETENTION_POLICY,
+    assets_identifier: 'resource.id',
   },
   [KSPM_POLICY_TEMPLATE]: {
     index: LATEST_FINDINGS_INDEX_PATTERN,
-    timeRange: LATEST_FINDINGS_RETENTION_POLICY,
+    assets_identifier: 'agent.id',
   },
   [CNVM_POLICY_TEMPLATE]: {
     index: LATEST_VULNERABILITIES_INDEX_PATTERN,
-    timeRange: LATEST_VULNERABILITIES_RETENTION_POLICY,
+    assets_identifier: 'cloud.instance.id',
   },
 };
 
@@ -48,6 +49,7 @@ export const getCloudSecurityUsageRecord = async ({
   logger,
   taskId,
   postureType,
+  tier,
 }: CloudSecurityMeteringCallbackInput): Promise<UsageRecord | undefined> => {
   try {
     if (!postureType) {
@@ -55,6 +57,8 @@ export const getCloudSecurityUsageRecord = async ({
       return;
     }
 
+    if (!(await indexHasDataInDateRange(esClient, postureType))) return;
+
     const response = await esClient.search<unknown, ResourceCountAggregation>(
       getAggQueryByPostureType(postureType)
     );
@@ -62,7 +66,7 @@ export const getCloudSecurityUsageRecord = async ({
     if (!response.aggregations) {
       return;
     }
-    const resourceCount = response.aggregations.unique_resources.value;
+    const resourceCount = response.aggregations.unique_assets.value;
     if (resourceCount > AGGREGATION_PRECISION_THRESHOLD) {
       logger.warn(
         `The number of unique resources for {${postureType}} is ${resourceCount}, which is higher than the AGGREGATION_PRECISION_THRESHOLD of ${AGGREGATION_PRECISION_THRESHOLD}.`
@@ -72,10 +76,12 @@ export const getCloudSecurityUsageRecord = async ({
       ? new Date(response.aggregations.min_timestamp.value_as_string).toISOString()
       : new Date().toISOString();
 
+    const creationTimestamp = new Date().toISOString();
+
     const usageRecord = {
-      id: `${CLOUD_SECURITY_TASK_TYPE}:${postureType}`,
+      id: `${CLOUD_SECURITY_TASK_TYPE}_${postureType}_${projectId}_${creationTimestamp}`,
       usage_timestamp: minTimestamp,
-      creation_timestamp: new Date().toISOString(),
+      creation_timestamp: creationTimestamp,
       usage: {
         type: CLOUD_SECURITY_TASK_TYPE,
         sub_type: postureType,
@@ -85,6 +91,7 @@ export const getCloudSecurityUsageRecord = async ({
       source: {
         id: taskId,
         instance_group_id: projectId,
+        metadata: { tier },
       },
     };
 
@@ -96,13 +103,24 @@ export const getCloudSecurityUsageRecord = async ({
   }
 };
 
-export const getAggQueryByPostureType = (postureType: PostureType) => {
+const indexHasDataInDateRange = async (esClient: ElasticsearchClient, postureType: PostureType) => {
+  const response = await esClient.search({
+    index: queryParams[postureType].index,
+    size: 1,
+    _source: false,
+    query: getSearchQueryByPostureType(postureType),
+  });
+
+  return response.hits.hits.length > 0;
+};
+
+export const getSearchQueryByPostureType = (postureType: PostureType) => {
   const mustFilters = [];
 
   mustFilters.push({
     range: {
       '@timestamp': {
-        gte: `now-${queryParams[postureType].timeRange}`,
+        gte: `now-${ASSETS_SAMPLE_GRANULARITY}`,
       },
     },
   });
@@ -115,18 +133,22 @@ export const getAggQueryByPostureType = (postureType: PostureType) => {
     });
   }
 
+  return {
+    bool: {
+      must: mustFilters,
+    },
+  };
+};
+
+export const getAggQueryByPostureType = (postureType: PostureType) => {
   const query = {
     index: queryParams[postureType].index,
-    query: {
-      bool: {
-        must: mustFilters,
-      },
-    },
+    query: getSearchQueryByPostureType(postureType),
     size: 0,
     aggs: {
-      unique_resources: {
+      unique_assets: {
         cardinality: {
-          field: 'resource.id',
+          field: queryParams[postureType].assets_identifier,
           precision_threshold: AGGREGATION_PRECISION_THRESHOLD,
         },
       },

x-pack/plugins/security_solution_serverless/server/cloud_security/types.ts

@@ -10,11 +10,11 @@ import type {
   KSPM_POLICY_TEMPLATE,
   CNVM_POLICY_TEMPLATE,
 } from '@kbn/cloud-security-posture-plugin/common/constants';
-import type { MeteringCallbackInput } from '../types';
+import type { MeteringCallbackInput, Tier } from '../types';
 
 export interface ResourceCountAggregation {
   min_timestamp: MinTimestamp;
-  unique_resources: {
+  unique_assets: {
     value: number;
   };
 }
@@ -33,4 +33,5 @@ export interface CloudSecurityMeteringCallbackInput
   extends Omit<MeteringCallbackInput, 'cloudSetup' | 'abortController' | 'config'> {
   projectId: string;
   postureType: PostureType;
+  tier: Tier;
 }

x-pack/plugins/security_solution_serverless/server/types.ts

@@ -71,9 +71,11 @@ export interface UsageSource {
 }
 
 export interface UsageSourceMetadata {
-  tier?: ProductTier;
+  tier?: Tier;
 }
 
+export type Tier = ProductTier | 'none';
+
 export interface SecurityUsageReportingTaskSetupContract {
   core: CoreSetup;
   logFactory: LoggerFactory;

x-pack/plugins/security_solution_serverless/tsconfig.json

@@ -36,6 +36,7 @@
     "@kbn/task-manager-plugin",
     "@kbn/cloud-plugin",
     "@kbn/cloud-security-posture-plugin",
-    "@kbn/fleet-plugin"
+    "@kbn/fleet-plugin",
+    "@kbn/core-elasticsearch-server"
   ]
 }