From abe0a0c3c729118d031671277b3254c8d4680cfc Mon Sep 17 00:00:00 2001
From: Raj Nishtala <rnishtala@sumologic.com>
Date: Tue, 16 May 2023 12:37:14 -0400
Subject: [PATCH] feat(EKS Fargate): Add multiline support to EKS Fargate

---
 .changelog/3059.added.txt                     |  1 +
 .../logs/collector/otelcloudwatch/config.yaml | 59 +++++++++++++++----
 docs/fargate.md                               |  7 +--
 3 files changed, 52 insertions(+), 15 deletions(-)
 create mode 100644 .changelog/3059.added.txt

diff --git a/.changelog/3059.added.txt b/.changelog/3059.added.txt
new file mode 100644
index 0000000000..0df0e2bb04
--- /dev/null
+++ b/.changelog/3059.added.txt
@@ -0,0 +1 @@
+feat(EKS Fargate): Add multiline support to EKS Fargate
\ No newline at end of file
diff --git a/deploy/helm/sumologic/conf/logs/collector/otelcloudwatch/config.yaml b/deploy/helm/sumologic/conf/logs/collector/otelcloudwatch/config.yaml
index 20dd87f2dc..dad9280c7b 100644
--- a/deploy/helm/sumologic/conf/logs/collector/otelcloudwatch/config.yaml
+++ b/deploy/helm/sumologic/conf/logs/collector/otelcloudwatch/config.yaml
@@ -26,24 +26,59 @@ processors:
     send_batch_max_size: 2000
     send_batch_size: 1000
     timeout: 1s
-  transform/cloudwatch_logs:
+  transform/set_source_identifier:
     error_mode: ignore
     log_statements:
       - context: log
         statements:
-          - set(attributes["k8s.container.name"], resource.attributes["cloudwatch.log.stream"])
-          - set(attributes["k8s.pod.name"], resource.attributes["cloudwatch.log.stream"])
-          - set(attributes["k8s.namespace.name"], resource.attributes["cloudwatch.log.stream"])
-          - replace_pattern(attributes["k8s.pod.name"], "^.*kube\\.var\\.log\\.containers\\.([0-9a-zA-Z\\-]+)\\_.*", "$$1")
-          - replace_pattern(attributes["k8s.container.name"], "^.*kube\\.var\\.log\\.containers\\.[0-9a-zA-Z\\-]+\\_[a-zA-Z\\-]*\\_([a-zA-Z]*).*", "$$1")
-          - replace_pattern(attributes["k8s.namespace.name"], "^.*kube\\.var\\.log\\.containers\\.[0-9a-zA-Z\\-]+\\_([a-zA-Z\\-]*)_.*", "$$1")
-  transform/parse:
+          - set(attributes["cloudwatch.log.stream"], resource.attributes["cloudwatch.log.stream"])
+  groupbyattrs/stream:
+    keys:
+      - cloudwatch.log.stream
+  ## need to reset the source identifier after grouping
+  transform/reset_source_identifier:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(attributes["cloudwatch.log.stream"], resource.attributes["cloudwatch.log.stream"])
+  transform/parsejson:
+    error_mode: ignore
     log_statements:
       - context: log
         statements:
           - set(body, ParseJSON(body)) where IsMatch(body, "^{") == true
           - merge_maps(attributes, body, "insert")
           - set(body, "") where IsMatch(body, "^{") == true
+  transform/metadata:
+    error_mode: ignore
+    log_statements:
+      - context: log
+        statements:
+          - set(attributes["k8s.container.name"], resource.attributes["cloudwatch.log.stream"])
+          - set(attributes["k8s.pod.name"], resource.attributes["cloudwatch.log.stream"])
+          - set(attributes["k8s.namespace.name"], resource.attributes["cloudwatch.log.stream"])
+          - replace_pattern(attributes["k8s.pod.name"], "^.*kube\\.var\\.log\\.containers\\.([0-9a-zA-Z\\-]+)\\_.*", "$$1")
+          - replace_pattern(attributes["k8s.container.name"], "^.*kube\\.var\\.log\\.containers\\.[0-9a-zA-Z\\-]+\\_[a-zA-Z\\-]*\\_([a-zA-Z]*).*", "$$1")
+          - replace_pattern(attributes["k8s.namespace.name"], "^.*kube\\.var\\.log\\.containers\\.[0-9a-zA-Z\\-]+\\_([a-zA-Z\\-]*)_.*", "$$1")
+  logstransform/cloudwatch:
+    operators:
+      - id: merge-cri-lines
+        combine_field: attributes.log
+        combine_with: ""
+        is_last_entry: attributes.logtag == "F"
+        output: "merge-multiline-logs"
+        overwrite_with: newest
+        source_identifier: attributes["cloudwatch.log.stream"]
+        type: recombine
+      - id: merge-multiline-logs
+        combine_field: attributes.log
+        combine_with: "\n"
+        is_first_entry: attributes.log matches {{ .Values.sumologic.logs.multiline.first_line_regex | quote }}
+        source_identifier: attributes["cloudwatch.log.stream"]
+        type: recombine
+      - field: attributes["cloudwatch.log.stream"]
+        type: remove
 receivers:
   awscloudwatch:
     region: {{ .Values.sumologic.logs.collector.otelcloudwatch.region }}
@@ -63,8 +98,12 @@ service:
       receivers:
         - awscloudwatch
       processors:
-        - transform/parse
-        - transform/cloudwatch_logs
+        - transform/set_source_identifier
+        - groupbyattrs/stream
+        - transform/reset_source_identifier
+        - transform/parsejson
+        - logstransform/cloudwatch
+        - transform/metadata
         - batch
       exporters:
         - otlphttp
diff --git a/docs/fargate.md b/docs/fargate.md
index d39e4c6511..fcc8785420 100644
--- a/docs/fargate.md
+++ b/docs/fargate.md
@@ -1,11 +1,8 @@
 # Fargate
 
-**NOTE: This is the alpha release of EKS Fargate.**
+**Release Note: Generally Available (GA) Release of EKS Fargate.**
 
-The following are some limitations of deploying this helm chart on EKS fargate
-
-- Only supports EKS version 1.24 and above
-- Does not support multiline logs
+Supports EKS version 1.24 and above
 
 The following documentation assumes that you are using eksctl to manage Fargate cluster. Code snippets are using environment variables in
 order to make them as generic and reusable.