[Cloud Security] Metering integration tests

package.json

@@ -946,6 +946,7 @@
     "@mapbox/mapbox-gl-rtl-text": "0.2.3",
     "@mapbox/mapbox-gl-supported": "2.0.1",
     "@mapbox/vector-tile": "1.3.1",
+    "@mswjs/http-middleware": "^0.10.1",
     "@opentelemetry/api": "^1.1.0",
     "@opentelemetry/api-metrics": "^0.31.0",
     "@opentelemetry/exporter-metrics-otlp-grpc": "^0.34.0",
@@ -1758,4 +1759,4 @@
     "zod-to-json-schema": "^3.22.3"
   },
   "packageManager": "[email protected]"
-}
+}

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

@@ -15,72 +15,119 @@ import {
   CSPM,
   KSPM,
   METERING_CONFIGS,
-  THRESHOLD_MINUTES,
   BILLABLE_ASSETS_CONFIG,
 } from './constants';
 import type { Tier, UsageRecord } from '../types';
 import type {
   CloudSecurityMeteringCallbackInput,
   CloudSecuritySolutions,
   AssetCountAggregation,
+  ResourceSubtypeAggregationBucket,
+  ResourceSubtypeCounter,
 } from './types';
 
 export const getUsageRecords = (
-  assetCountAggregations: AssetCountAggregation[],
+  assetCountAggregation: AssetCountAggregation,
   cloudSecuritySolution: CloudSecuritySolutions,
   taskId: string,
   tier: Tier,
   projectId: string,
   periodSeconds: number,
   logger: Logger
-): UsageRecord[] => {
-  const usageRecords = assetCountAggregations.map((assetCountAggregation) => {
-    const assetCount = assetCountAggregation.unique_assets.value;
-
-    if (assetCount > AGGREGATION_PRECISION_THRESHOLD) {
-      logger.warn(
-        `The number of unique resources for {${cloudSecuritySolution}} is ${assetCount}, which is higher than the AGGREGATION_PRECISION_THRESHOLD of ${AGGREGATION_PRECISION_THRESHOLD}.`
-      );
-    }
-
-    const minTimestamp = new Date(
-      assetCountAggregation.min_timestamp.value_as_string
-    ).toISOString();
-
-    const creationTimestamp = new Date();
-    const minutes = creationTimestamp.getMinutes();
-    if (minutes >= 30) {
-      creationTimestamp.setMinutes(30, 0, 0);
-    } else {
-      creationTimestamp.setMinutes(0, 0, 0);
-    }
-    const roundedCreationTimestamp = creationTimestamp.toISOString();
-
-    const usageRecord: UsageRecord = {
-      id: `${CLOUD_SECURITY_TASK_TYPE}_${cloudSecuritySolution}_${projectId}_${roundedCreationTimestamp}`,
-      usage_timestamp: minTimestamp,
-      creation_timestamp: creationTimestamp.toISOString(),
-      usage: {
-        type: CLOUD_SECURITY_TASK_TYPE,
-        sub_type: cloudSecuritySolution,
-        quantity: assetCount,
-        period_seconds: periodSeconds,
-      },
-      source: {
-        id: taskId,
-        instance_group_id: projectId,
-        metadata: { tier },
+): UsageRecord => {
+  let assetCount;
+  let resourceSubtypeCounter;
+
+  if (cloudSecuritySolution === CSPM || cloudSecuritySolution === KSPM) {
+    const resourceSubtypeBuckets: ResourceSubtypeAggregationBucket[] =
+      assetCountAggregation.resource_sub_type.buckets;
+
+    const billableAssets = BILLABLE_ASSETS_CONFIG[cloudSecuritySolution].values;
+    assetCount = resourceSubtypeBuckets
+      .filter((bucket) => billableAssets.includes(bucket.key))
+      .reduce((acc, bucket) => acc + bucket.unique_assets.value, 0);
+
+    resourceSubtypeCounter = assetCountAggregation.resource_sub_type.buckets.reduce(
+      (resourceMap, item) => {
+        resourceMap[item.key] = {
+          doc_count: item.doc_count as number,
+          unique_assets: item.unique_assets.value as number,
+        };
+        return resourceMap;
       },
-    };
+      {} as ResourceSubtypeCounter
+    );
+  } else {
+    assetCount = assetCountAggregation.unique_assets.value;
+  }
+
+  if (assetCount > AGGREGATION_PRECISION_THRESHOLD) {
+    logger.warn(
+      `The number of unique resources for {${cloudSecuritySolution}} is ${assetCount}, which is higher than the AGGREGATION_PRECISION_THRESHOLD of ${AGGREGATION_PRECISION_THRESHOLD}.`
+    );
+  }
+
+  const minTimestamp = new Date(assetCountAggregation.min_timestamp.value_as_string).toISOString();
 
-    return usageRecord;
-  });
-  return usageRecords;
+  const creationTimestamp = new Date();
+  const minutes = creationTimestamp.getMinutes();
+  if (minutes >= 30) {
+    creationTimestamp.setMinutes(30, 0, 0);
+  } else {
+    creationTimestamp.setMinutes(0, 0, 0);
+  }
+  const roundedCreationTimestamp = creationTimestamp.toISOString();
+
+  const metadata = resourceSubtypeCounter
+    ? { tier, resource_sub_type_count: resourceSubtypeCounter }
+    : { tier };
+
+  const usageRecord: UsageRecord = {
+    id: `${CLOUD_SECURITY_TASK_TYPE}_${cloudSecuritySolution}_${projectId}_${roundedCreationTimestamp}`,
+    usage_timestamp: minTimestamp,
+    creation_timestamp: creationTimestamp.toISOString(),
+    usage: {
+      type: CLOUD_SECURITY_TASK_TYPE,
+      sub_type: cloudSecuritySolution,
+      quantity: assetCount,
+      period_seconds: periodSeconds,
+    },
+    source: {
+      id: taskId,
+      instance_group_id: projectId,
+      // metadata: { tier },
+      metadata,
+    },
+  };
+
+  return usageRecord;
 };
 
 export const getAggregationByCloudSecuritySolution = (
   cloudSecuritySolution: CloudSecuritySolutions
 ) => {
+  if (cloudSecuritySolution === CSPM || cloudSecuritySolution === KSPM)
+    return {
+      resource_sub_type: {
+        terms: {
+          field: BILLABLE_ASSETS_CONFIG[cloudSecuritySolution].filter_attribute,
+        },
+        aggs: {
+          unique_assets: {
+            cardinality: {
+              field: METERING_CONFIGS[cloudSecuritySolution].assets_identifier,
+              precision_threshold: AGGREGATION_PRECISION_THRESHOLD,
+            },
+          },
+        },
+      },
+      min_timestamp: {
+        min: {
+          field: '@timestamp',
+        },
+      },
+    };
+
   return {
     unique_assets: {
       cardinality: {
@@ -97,8 +144,7 @@ export const getAggregationByCloudSecuritySolution = (
 };
 
 export const getSearchQueryByCloudSecuritySolution = (
-  cloudSecuritySolution: CloudSecuritySolutions,
-  searchFrom: Date
+  cloudSecuritySolution: CloudSecuritySolutions
 ) => {
   const mustFilters = [];
 
@@ -117,20 +163,11 @@ export const getSearchQueryByCloudSecuritySolution = (
   }
 
   if (cloudSecuritySolution === CSPM || cloudSecuritySolution === KSPM) {
-    const billableAssetsConfig = BILLABLE_ASSETS_CONFIG[cloudSecuritySolution];
-
     mustFilters.push({
       term: {
         'rule.benchmark.posture_type': cloudSecuritySolution,
       },
     });
-
-    // filter in only billable assets
-    mustFilters.push({
-      terms: {
-        [billableAssetsConfig.filter_attribute]: billableAssetsConfig.values,
-      },
-    });
   }
 
   return {
@@ -141,10 +178,9 @@ export const getSearchQueryByCloudSecuritySolution = (
 };
 
 export const getAssetAggQueryByCloudSecuritySolution = (
-  cloudSecuritySolution: CloudSecuritySolutions,
-  searchFrom: Date
+  cloudSecuritySolution: CloudSecuritySolutions
 ) => {
-  const query = getSearchQueryByCloudSecuritySolution(cloudSecuritySolution, searchFrom);
+  const query = getSearchQueryByCloudSecuritySolution(cloudSecuritySolution);
   const aggs = getAggregationByCloudSecuritySolution(cloudSecuritySolution);
 
   return {
@@ -157,51 +193,34 @@ export const getAssetAggQueryByCloudSecuritySolution = (
 
 export const getAssetAggByCloudSecuritySolution = async (
   esClient: ElasticsearchClient,
-  cloudSecuritySolution: CloudSecuritySolutions,
-  searchFrom: Date
-): Promise<AssetCountAggregation[]> => {
-  const assetsAggQuery = getAssetAggQueryByCloudSecuritySolution(cloudSecuritySolution, searchFrom);
+  cloudSecuritySolution: CloudSecuritySolutions
+): Promise<AssetCountAggregation | undefined> => {
+  const assetsAggQuery = getAssetAggQueryByCloudSecuritySolution(cloudSecuritySolution);
 
   const response = await esClient.search<unknown, AssetCountAggregation>(assetsAggQuery);
-  if (!response.aggregations) return [];
 
-  return [response.aggregations];
+  if (!response.aggregations) return;
+
+  return response.aggregations;
 };
 
 const indexHasDataInDateRange = async (
   esClient: ElasticsearchClient,
-  cloudSecuritySolution: CloudSecuritySolutions,
-  searchFrom: Date
+  cloudSecuritySolution: CloudSecuritySolutions
 ) => {
   const response = await esClient.search(
     {
       index: METERING_CONFIGS[cloudSecuritySolution].index,
       size: 1,
       _source: false,
-      query: getSearchQueryByCloudSecuritySolution(cloudSecuritySolution, searchFrom),
+      query: getSearchQueryByCloudSecuritySolution(cloudSecuritySolution),
     },
     { ignore: [404] }
   );
 
   return response.hits?.hits.length > 0;
 };
 
-const getSearchStartDate = (lastSuccessfulReport: Date): Date => {
-  const initialDate = new Date();
-  const thresholdDate = new Date(initialDate.getTime() - THRESHOLD_MINUTES * 60 * 1000);
-
-  if (lastSuccessfulReport) {
-    const lastSuccessfulReportDate = new Date(lastSuccessfulReport);
-
-    const searchFrom =
-      lastSuccessfulReport && lastSuccessfulReportDate > thresholdDate
-        ? lastSuccessfulReportDate
-        : thresholdDate;
-    return searchFrom;
-  }
-  return thresholdDate;
-};
-
 export const getCloudSecurityUsageRecord = async ({
   esClient,
   projectId,
@@ -212,21 +231,20 @@ export const getCloudSecurityUsageRecord = async ({
   logger,
 }: CloudSecurityMeteringCallbackInput): Promise<UsageRecord[] | undefined> => {
   try {
-    const searchFrom = getSearchStartDate(lastSuccessfulReport);
-
-    if (!(await indexHasDataInDateRange(esClient, cloudSecuritySolution, searchFrom))) return;
+    if (!(await indexHasDataInDateRange(esClient, cloudSecuritySolution))) return;
 
     // const periodSeconds = Math.floor((new Date().getTime() - searchFrom.getTime()) / 1000);
     const periodSeconds = 1800; // Workaround to prevent overbilling by charging for a constant time window. The issue should be resolved in https://github.com/elastic/security-team/issues/9424.
 
-    const assetCountAggregations = await getAssetAggByCloudSecuritySolution(
+    const assetCountAggregation = await getAssetAggByCloudSecuritySolution(
       esClient,
-      cloudSecuritySolution,
-      searchFrom
+      cloudSecuritySolution
     );
 
+    if (!assetCountAggregation) return [];
+
     const usageRecords = await getUsageRecords(
-      assetCountAggregations,
+      assetCountAggregation,
       cloudSecuritySolution,
       taskId,
       tier,
@@ -235,7 +253,7 @@ export const getCloudSecurityUsageRecord = async ({
       logger
     );
 
-    return usageRecords;
+    return [usageRecords];
   } catch (err) {
     logger.error(`Failed to fetch ${cloudSecuritySolution} metering data ${err}`);
   }

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

@@ -76,7 +76,8 @@ export const getUsageRecords = async (
             {
               range: {
                 'event.ingested': {
-                  gt: searchFrom.toISOString(),
+                  // gt: searchFrom.toISOString()
+                  gte: `now-30m`,
                 },
               },
             },

x-pack/plugins/security_solution_serverless/server/common/services/usage_reporting_service.ts

@@ -19,11 +19,13 @@ export class UsageReportingService {
     records: UsageRecord[],
     url = USAGE_SERVICE_USAGE_URL
   ): Promise<Response> {
+    const isHttps = url.includes('https');
+
     return fetch(url, {
       method: 'post',
       body: JSON.stringify(records),
       headers: { 'Content-Type': 'application/json' },
-      agent,
+      agent: isHttps ? agent : undefined, // Conditionally add agent if URL is HTTPS for supporting integration tests.
     });
   }
 }

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

@@ -17,11 +17,17 @@ export const configSchema = schema.object({
   enabled: schema.boolean({ defaultValue: false }),
   productTypes,
   /**
-   * Usage Reporting: the interval between runs of the task
+   * Usage Reporting: the interval between runs of the endpoint task
    */
 
   usageReportingTaskInterval: schema.string({ defaultValue: '5m' }),
 
+  /**
+   * Usage Reporting: the interval between runs of the cloud security task
+   */
+
+  cloudSecurityUsageReportingTaskInterval: schema.string({ defaultValue: '30m' }),
+
   /**
    * Usage Reporting: timeout value for how long the task should run.
    */

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

@@ -113,7 +113,7 @@ export class SecuritySolutionServerlessPlugin
     this.cloudSecurityUsageReportingTask
       ?.start({
         taskManager: pluginsSetup.taskManager,
-        interval: cloudSecurityMetringTaskProperties.interval,
+        interval: this.config.cloudSecurityUsageReportingTaskInterval,
       })
       .catch(() => {});
 

x-pack/test_serverless/api_integration/test_suites/security/cloud_security_posture/index.ts

@@ -17,6 +17,7 @@ export default function ({ loadTestFile }: FtrProviderContext) {
     loadTestFile(require.resolve('./benchmark/v2'));
     loadTestFile(require.resolve('./find_csp_benchmark_rule'));
     loadTestFile(require.resolve('./telemetry'));
+    loadTestFile(require.resolve('./serverless_metering/cloud_security_metering'));
 
     // TODO: migrate status_unprivileged tests from stateful, if it feasible in serverless with the new security model
     // loadTestFile(require.resolve('./status/status_unprivileged'));