ECE Adminconsole logs

.github/CODEOWNERS

@@ -210,6 +210,7 @@
 /packages/elastic_security @elastic/security-service-integrations
 /packages/elasticsearch @elastic/stack-monitoring
 /packages/aws_elb_otel @elastic/obs-infraobs-integrations
+/packages/ece @elastic/stack-monitoring
 /packages/endace @elastic/integration-experience @elastic/sec-linux-platform
 /packages/endace/data_stream/flow @elastic/sec-linux-platform
 /packages/endace/data_stream/log @elastic/integration-experience

packages/ece/_dev/build/build.yml

@@ -0,0 +1,3 @@
+dependencies:
+  ecs:
+    reference: git@v8.17.0

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

@@ -0,0 +1,19 @@
+# Elastic Cloud Enterprise Integration
+
+The Elastic Cloud Enterprise (ECE) Integration allows you to collect the Adminconsole logs which contain all actions performed through the admin UI and as well as through the API. The Elastic Agent collecting this logs needs to be installed on all Control Planes in ECE, as the Control Planes usually host the adminconsole container.
+
+## Overview
+
+### Compability
+
+This has been tested and verified on version ECE 3.8, as well as 4.0
+
+### How it works
+
+Install the Elastic Agent on the hosts that are running the adminconsole, which are usually your control plane hosts. This will spawn a filestream input which reads the adminconsole.log file.
+
+## Adminconsole
+
+{{ event "adminconsole" }}
+
+{{ fields "adminconsole" }}

packages/ece/_dev/deploy/docker/docker-compose.yml

@@ -0,0 +1,8 @@
+version: "2.3"
+services:
+  ece_adminconsole.log-filestream:
+    image: alpine
+    volumes:
+      - ./sample_logs:/sample_logs:ro
+      - ${SERVICE_LOGS_DIR}:/mnt/data/elastic/abc/services/admin-console/logs/
+    command: /bin/sh -c "cp /sample_logs/* /mnt/data/elastic/abc/services/admin-console/logs/"

packages/ece/changelog.yml

@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/13864

packages/ece/data_stream/adminconsole/_dev/test/system/test-filestream-config.yml

@@ -0,0 +1,14 @@
+service: ece_adminconsole.log-filestream
+input: filestream
+data_stream:
+  vars:
+    preserve_original_event: true
+    preserve_duplicate_custom_fields: true
+    paths:
+      - '{{SERVICE_LOGS_DIR}}/*.ndjson'
+numeric_keyword_fields:
+  - log.file.device_id
+  - log.file.inode
+  - log.file.fingerprint
+  - log.offset
+  - input.type

packages/ece/data_stream/adminconsole/agent/stream/filestream.yml.hbs

@@ -0,0 +1,22 @@
+paths:
+{{#each paths as |path|}}
+  - {{path}}
+{{/each}}
+exclude_files: [".gz$"]
+tags:
+{{#if preserve_original_event}}
+  - preserve_original_event
+{{/if}}
+{{#each tags as |tag|}}
+  - {{tag}}
+{{/each}}
+{{#if processors}}
+processors:
+{{processors}}
+{{/if}}
+{{#if harvester_limit}}
+harvester_limit: {{harvester_limit}}
+{{/if}}
+{{#if close.on_state_change.inactive}}
+close.on_state_change.inactive: {{close.on_state_change.inactive}}
+{{/if}}

packages/ece/data_stream/adminconsole/elasticsearch/ingest_pipeline/api.yml

@@ -0,0 +1,253 @@
+---
+description: Pipeline for processing the API logs
+processors:
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.process.thread.name"
+      target_field: "process.thread.name"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.request_payload"
+      target_field: "http.request.body.content"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.message"
+      target_field: "message"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.log.logger"
+      target_field: "log.logger"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.log.level"
+      target_field: "log.level"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.trace.id"
+      target_field: "trace.id"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.proxy_ip"
+      target_field: "network.forwarded_ip"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.query_parameters"
+      target_field: "url.query"
+  - set:
+      description: For the URI parts processor we need proper http://... therefore we add it. http if it's not default port.
+      field: url.origin
+      value: http://{{tmp.ece.log.request_url}}
+      if: "!ctx.tmp?.ece?.log?.request_url?.contains('12443')"
+  - set:
+      description: For the URI parts processor we need proper https://... therefore we add it.
+      field: url.origin
+      value: https://{{tmp.ece.log.request_url}}
+      if: ctx.tmp?.ece?.log?.request_url?.contains('12443')
+  - uri_parts:
+      field: url.origin
+      ignore_missing: true
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.request_method"
+      target_field: "http.request.method"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.request_length"
+      target_field: "http.request.body.bytes"
+  - convert:
+      field: http.request.body.bytes
+      type: long
+      ignore_missing: true
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.status_code"
+      target_field: "http.response.status_code"
+  - convert:
+      field: "http.response.status_code"
+      type: long
+      ignore_missing: true
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.auth_user"
+      target_field: "user.name"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.response_length"
+      target_field: "http.response.body.bytes"
+  - convert:
+      field: http.response.body.bytes
+      type: long
+      ignore_missing: true
+  - script:
+      description: "Calculate nanoseconds for duration"
+      if: "ctx.tmp?.ece?.log?.response_time != null"
+      ignore_failure: true
+      tag: "calculating event duration"
+      source: |
+        ctx.event.duration = Long.parseLong(ctx.tmp.ece.log.response_time) * 1000000
+  - user_agent:
+      ignore_missing: true
+      field: "tmp.ece.log.user_agent"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.request_id"
+      target_field: "http.request.id"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.organization_id"
+      target_field: "organization.id"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.transaction.id"
+      target_field: "transaction.id"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.client_ip"
+      target_field: "source.address"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.region"
+      target_field: "cloud.region"
+  - rename:
+      ignore_missing: true
+      field: "tmp.ece.log.control_plane_use_case"
+      target_field: "event.module"
+  - append:
+      if: "ctx.http?.request?.method != null && ctx.url?.original != null"
+      field: "event.category"
+      value:
+        - "api"
+      allow_duplicates: false
+  - append:
+      if: "ctx.http?.response?.status_code != null"
+      field: "event.type"
+      value:
+        - "info"
+      allow_duplicates: false
+  - append:
+      if: "ctx.network?.forwarded_ip != null"
+      field: "related.ip"
+      value: "{{{network.forwarded_ip}}}"
+      allow_duplicates: false
+  - convert:
+      ignore_missing: true
+      field: "source.address"
+      target_field: "source.ip"
+      type: ip
+      on_failure:
+        - set:
+            field: tmp.extraction_failures.source_ip
+            value: true
+  - append:
+      if: "ctx.source?.ip != null"
+      field: "related.ip"
+      value: "{{{source.ip}}}"
+      allow_duplicates: false
+  - remove:
+      field: source.address
+      ignore_missing: true
+      if: "ctx.source?.ip != null"
+  - geoip:
+      ignore_missing: true
+      tag: "geoip_source_ip"
+      field: "source.ip"
+      target_field: "source.geo"
+  - geoip:
+      ignore_missing: true
+      tag: "geoip_source_asn"
+      database_file: "GeoLite2-ASN.mmdb"
+      field: "source.ip"
+      target_field: "source.as"
+      properties:
+        - "asn"
+        - "organization_name"
+  - rename:
+      ignore_missing: true
+      tag: "rename_source_as_asn"
+      field: "source.as.asn"
+      target_field: "source.as.number"
+  - rename:
+      ignore_missing: true
+      tag: "rename_source_as_organization_name"
+      field: "source.as.organization_name"
+      target_field: "source.as.organization.name"
+  - append:
+      if: "ctx.user?.name != null"
+      field: "related.user"
+      value: "{{{user.name}}}"
+      allow_duplicates: false
+  - append:
+      if: "ctx.url?.domain != null"
+      field: "related.hosts"
+      value: "{{{url.domain}}}"
+      allow_duplicates: false
+  - rename:
+      ignore_missing: true
+      tag: "rename_destination_as_organization_name"
+      field: "destination.as.organization_name"
+      target_field: "destination.as.organization.name"
+  - grok:
+      description: Extract the deployment id from the url
+      field: "url.path"
+      if: "ctx.url?.path instanceof String && ctx.url.path.contains('/deployments/') && !['_search','traffic-filter/rulesets', 'templates'].contains(ctx.url.path.splitOnToken('/deployments/')[-1])"
+      patterns:
+        - "\\/deployments\\/(?<ece_adminconsole.log.deployment.id>[^\\/]+).*"
+      tag: "extract-deployment.id-from-url.path"
+  - dissect:
+      description: Extracts the API call against Elasticsearch through the API console
+      field: "url.original"
+      if: ctx.url?.original?.contains('/elasticsearch/main-elasticsearch/proxy/')
+      pattern: "%{}/elasticsearch/main-elasticsearch/proxy/%{ece_adminconsole.log.deployment.es_api}"
+  - script:
+      description: Set event.action based on API documentation
+      if: "ctx.url?.original != null && ctx.http?.request?.method != null && ctx.url.original.contains('/api/v1/')"
+      source: |
+        def temp = params.get(ctx.http.request.method.toLowerCase());
+        // elastic.co/api/v1/deployments
+        // and we only want the part after the /api/v1/
+        String url_parts = ctx.url.original.splitOnToken("/api/v1/")[1];
+        if (url_parts.contains('elasticsearch/elasticsearch/proxy/')){
+          // we handle the special cases here
+          url_parts = url_parts.splitOnToken('elasticsearch/elasticsearch/proxy/')[1];
+          // that should give us `_search`, `_cat/indices`
+          url_parts = url_parts.splitOnToken('?')[0];
+          // we do not care about params ? this will just overload the event.action
+          url_parts = url_parts.splitOnToken('/')[0];
+          // we do not care about '/' like in `_cat/indices`, just getting `_cat` back is good enough
+          ctx.putIfAbsent("event", [:]);
+          ctx.event.action = "elasticsearch_api_through_ece" + "-" + url_parts.toLowerCase();
+        }
+        else if (temp != null){
+            if (temp.get(url_parts) != null){
+                ctx.putIfAbsent("event", [:]);
+                ctx.event.action = temp.get(url_parts);
+            }
+        }
+        if (ctx.event?.action == null){
+            ctx.putIfAbsent("event", [:]);
+            ctx.event.action = ctx.http.request.method.toLowerCase() + "_" + url_parts.splitOnToken("/")[0];
+        }
+      params:
+        get:
+          deployments: "list_deployments"
+        post:
+          deployments: "create_deployment"
+  - json:
+      field: "http.request.body.content"
+      target_field: "tmp.request"
+      on_failure:
+        - set:
+            field: tmp.json_payload_error
+            value: true
+  - rename:
+      field: "tmp.request.name"
+      target_field: "ece_adminconsole.log.deployment.name"
+      ignore_missing: true
+      if: "ctx.tmp?.json_payload_error == null"
+on_failure:
+  - append:
+      field: "error.message"
+      value: "Processor {{{_ingest.on_failure_processor_type}}} with tag {{{_ingest.on_failure_processor_tag}}} in pipeline {{{_ingest.on_failure_pipeline}}} failed with message: {{{_ingest.on_failure_message}}}"
+  - set:
+      field: "event.kind"
+      value: "pipeline_error"