[sysdig] Add support for cspm datastream

packages/sysdig/_dev/build/docs/README.md

@@ -2,12 +2,14 @@
 This integration allows for the shipping of [Sysdig](https://sysdig.com/) logs to Elastic for security, observability and organizational awareness. Logs can then be analyzed by using either the dashboard included with the integration or via the creation of custom dashboards within Kibana.
 
 ## Data Streams
-The Sysdig integration collects two type of logs:
+The Sysdig integration collects three types of logs:
 
 **Alerts** The Alerts data stream collected by the Sysdig integration is comprised of Sysdig Alerts. See more details about Sysdig Alerts in [Sysdig's Alerts Documentation](https://docs.sysdig.com/en/docs/sysdig-monitor/alerts/). A complete list of potential fields used by this integration can be found in the [Logs reference](#logs-reference)
 
 **Event** The event data stream collected through the Sysdig integration consists of Sysdig Security Events. See more details about Security Events in [Sysdig's Events Feed Documentation](https://docs.sysdig.com/en/docs/sysdig-secure/threats/activity/events-feed/).
 
+**CSPM** The CSPM data stream collected through the Sysdig integration consists of Sysdig compliance results. See more details about compliance results in [Sysdig's Compliance documentation](https://docs.sysdig.com/en/sysdig-secure/compliance/).
+
 ## Requirements
 
 ### Agent-based installation
@@ -28,7 +30,7 @@ The HTTP input allows the Elastic Agent to receive Sysdig Alerts via HTTP webhoo
 
 **Required:** To configure Sysdig to output JSON, you must set up as webhook notification channel as outlined in the [Sysdig Documentation](https://docs.sysdig.com/en/docs/administration/administration-settings/outbound-integrations/notifications-management/set-up-notification-channels/configure-a-webhook-channel/).
 
-### To collect data from the Sysdig Next Gen API:
+### To collect data from the Sysdig API:
 
 - Retrieve the API Token by following [Sysdig's API Token Guide](https://docs.sysdig.com/en/retrieve-the-sysdig-api-token).
 
@@ -66,3 +68,13 @@ This is the `event` dataset.
 {{event "event"}}
 
 {{fields "event"}}
+
+### CSPM
+
+This is the `CSPM` dataset.
+
+#### Example
+
+{{event "cspm"}}
+
+{{fields "cspm"}}

packages/sysdig/changelog.yml

@@ -1,4 +1,9 @@
 # newer versions go on top
+- version: "2.1.0"
+  changes:
+    - description: Add support for cspm datastream.
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/14907
 - version: "2.0.0"
   changes:
     - description: Fix the conflicting data types for `sysdig.event.category` by changing the alerts data stream from `text` to `keyword`.

packages/sysdig/data_stream/cspm/_dev/test/pipeline/test-common-config.yml

@@ -0,0 +1,3 @@
+fields:
+  tags:
+    - preserve_duplicate_custom_fields

packages/sysdig/data_stream/cspm/_dev/test/system/test-default-config.yml

@@ -0,0 +1,11 @@
+input: cel
+service: sysdig
+vars:
+  url: http://{{Hostname}}:{{Port}}
+  api_token: xxxx
+data_stream:
+  vars:
+    preserve_original_event: true
+    batch_size: 2
+assert:
+  hit_count: 7

packages/sysdig/data_stream/cspm/agent/stream/cel.yml.hbs

@@ -0,0 +1,87 @@
+config_version: 2
+interval: {{interval}}
+resource.tracer:
+  enabled: {{enable_request_tracer}}
+  filename: "../../logs/cel/http-request-trace-*.ndjson"
+  maxbackups: 5
+{{#if proxy_url}}
+resource.proxy_url: {{proxy_url}}
+{{/if}}
+{{#if ssl}}
+resource.ssl: {{ssl}}
+{{/if}}
+{{#if http_client_timeout}}
+resource.timeout: {{http_client_timeout}}
+{{/if}}
+{{#if max_executions}}
+max_executions: {{max_executions}}
+{{/if}}
+resource.url: {{url}}
+state:
+  batch_size: {{batch_size}}
+  api_token: {{api_token}}
+redact:
+  fields:
+    - api_token
+program: |
+  request("GET",
+    state.url.trim_right("/") + "/api/cspm/v1/compliance/requirements?" + {
+      "pageNumber": [string(state.?page_number.orValue(1))],
+      "pageSize": [string(state.batch_size)],
+    }.format_query()
+  ).with({
+    "Header":{
+      "Authorization": ["Bearer " + state.api_token],
+    },
+  }).do_request().as(resp, resp.StatusCode == 200 ?
+    resp.Body.decode_json().as(body, {
+      "events": has(body.data) && size(body.data) > 0 ?
+        // Create an event for each control with its parent's field.
+        body.data.map(result,
+          result.controls.map(con, {
+            "message": result.with({"control": con}).drop("controls").encode_json()
+          })
+        ).flatten()
+      :
+        [],
+      "page_number": int(state.?page_number.orValue(1)) + 1,
+      "batch_size": state.batch_size,
+      "api_token": state.api_token,
+      "want_more": has(body.data) && size(body.data) > 0,
+    })
+  :
+    {
+      "events": {
+        "error": {
+          "code": string(resp.StatusCode),
+          "id": string(resp.Status),
+          "message": "GET " + state.url.trim_right("/") + "/api/cspm/v1/compliance/requirements: " + (
+            size(resp.Body) != 0 ?
+              string(resp.Body)
+            :
+              string(resp.Status) + ' (' + string(resp.StatusCode) + ')'
+          ),
+        },
+      },
+      "batch_size": state.batch_size,
+      "api_token": state.api_token,
+      "want_more": false
+    }
+  )
+tags:
+{{#if preserve_original_event}}
+  - preserve_original_event
+{{/if}}
+{{#if preserve_duplicate_custom_fields}}
+  - preserve_duplicate_custom_fields
+{{/if}}
+{{#each tags as |tag|}}
+  - {{tag}}
+{{/each}}
+{{#contains "forwarded" tags}}
+publisher_pipeline.disable_host: true
+{{/contains}}
+{{#if processors}}
+processors:
+{{processors}}
+{{/if}}