feat: ELB routing and pipeline

Signed-off-by: Kavindu Dodanduwa <[email protected]>

Author: Kavindu-Dodan

packages/aws/data_stream/elb_logs/elasticsearch/ingest_pipeline/default.yml

@@ -9,14 +9,15 @@ processors:
       field: message
       target_field: event.original
       ignore_missing: true
-      if: 'ctx.event?.original == null'
+      if: ctx.event?.original == null
       description: 'Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.'
   - remove:
       field: message
       ignore_missing: true
-      if: 'ctx.event?.original != null'
+      if: ctx.event?.original != null
       description: 'The `message` field is no longer required if the document has an `event.original` field.'
   - grok:
+      if: ctx?.event?.source != 'otel'
       field: event.original
       # Classic ELB patterns documented in https://docs.aws.amazon.com/elasticloadbalancing/latest/classic/access-log-collection.html
       # ELB v2 Application load balancers https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-access-logs.html
@@ -182,6 +183,7 @@ processors:
       separator: ' '
       ignore_missing: true
   - date:
+      if: ctx?.event?.source != 'otel'
       field: _tmp.timestamp
       formats:
         - ISO8601

packages/otel_shim/data_stream/aws_elblogs/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,77 @@
+---
+description: "Pipeline for processing sample logs"
+
+processors:
+  - dot_expander:
+      field: "*"
+      path: attributes
+  - dot_expander:
+      field: "*"
+      path: resource.attributes
+  # Common fields
+  - set:
+      field: aws.elb.type
+      copy_from: attributes.network.protocol.name
+  - set:
+      field: aws.elb.name
+      copy_from: resource.attributes.cloud.resource_id
+  - set:
+      field: source.address
+      copy_from: attributes.client.address
+  - set:
+      field: source.port
+      copy_from: attributes.client.port
+  - set:
+      field: http.request.body.bytes
+      copy_from: attributes.http.request.size
+  - set:
+      field: http.response.body.bytes
+      copy_from: attributes.http.response.size
+  - set:
+      field: source.bytes
+      copy_from: attributes.http.request.size
+  - set:
+      field: destination.bytes
+      copy_from: attributes.http.response.size
+  # Fields that may be missing between ALB, NLB & CLB
+  - set:
+      if: 'ctx?.attributes?.http?.request?.method != null'
+      field: http.request.method
+      copy_from: attributes.http.request.method
+  - set:
+      if: 'ctx?.attributes?.url?.full != null'
+      field: _tmp.uri_orig
+      copy_from: attributes.url.full
+  - set:
+      if: 'ctx?.attributes?.network?.protocol?.version != null'
+      field: http.version
+      copy_from: attributes.network.protocol.version
+  - set:
+      if: 'ctx?.attributes?.tls?.cipher != null'
+      field: ssl_cipher
+      copy_from: attributes.tls.cipher
+  - set:
+      if: 'ctx?.attributes?.tls?.protocol?.version != null'
+      field: ssl_protocol
+      copy_from: attributes.tls.protocol.version
+  - set:
+      if: 'ctx?.attributes?.aws?.elb?.tls?.listener?.resource_id != null'
+      field: aws.elb.listener
+      copy_from: attributes.aws.elb.tls.listener.resource_id
+  - set:
+      if: 'ctx?.attributes?.aws?.elb?.status?.code!= null'
+      field: http.response.status_code
+      copy_from: attributes.aws.elb.status.code
+  # Drop OTel attributes
+  - remove:
+      field:
+        - attributes
+        - resource.attributes
+      ignore_missing: true
+  - set:
+      field: event.source
+      value: otel
+on_failure:
+  - set:
+      field: error.message
+      value: '{{ _ingest.on_failure_message }}'

packages/otel_shim/data_stream/aws_elblogs/fields/base-fields.yml

@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.

packages/otel_shim/data_stream/aws_elblogs/manifest.yml

@@ -0,0 +1,6 @@
+title: "Route OTel AWS ELB logs to ECS"
+type: logs
+dataset: aws.elbaccess.otel
+elasticsearch:
+  dynamic_dataset: true
+  dynamic_namespace: true

packages/otel_shim/data_stream/aws_elblogs/routing_rules.yml

@@ -0,0 +1,8 @@
+- source_dataset: aws.elbaccess.otel
+  rules:
+    # Route to aws.elb_logs dataset if event parsing is successful
+    - target_dataset: aws.elb_logs
+      if: ctx?.event?.source == "otel"
+      namespace:
+        - "{{data_stream.namespace}}"
+        - default