mysql: Add script processor to fix grok behaviour of ES 8.7.x+ (#6531)

elastic · Jun 28, 2023 · 588f75c · 588f75c
1 parent 6152bc5
commit 588f75c
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 1 deletion.
diff --git a/packages/mysql/data_stream/slowlog/_dev/test/pipeline/test-mysql-ubuntu-8-0-15.log b/packages/mysql/data_stream/slowlog/_dev/test/pipeline/test-mysql-ubuntu-8-0-15.log
@@ -16,4 +16,4 @@ SELECT last_name, MAX(salary) AS salary FROM employees
     INNER JOIN salaries ON employees.emp_no = salaries.emp_no
     GROUP BY last_name
     ORDER BY salary DESC
-    LIMIT 10;
+    LIMIT 10;
diff --git a/packages/mysql/data_stream/slowlog/elasticsearch/ingest_pipeline/default.json b/packages/mysql/data_stream/slowlog/elasticsearch/ingest_pipeline/default.json
@@ -32,6 +32,30 @@
         "ignore_missing": true
       }
     },
+    {
+      "script": {
+        "lang": "painless",
+        "if": "ctx?.mysql != null",
+        "source": "for (field in params.fields) { if (ctx.mysql[field] instanceof List) { def vals = ctx.mysql[field]; vals = vals.stream().distinct().collect(Collectors.toList()); if (vals.size() == 1) { ctx.mysql[field] = vals[0] } else { ctx.mysql[field] = vals } } }",
+        "params": {
+          "fields": [
+            "thread_id"
+          ]
+        }
+      }
+    },
+    {
+      "script": {
+        "lang": "painless",
+        "if": "ctx?.mysql?.slowlog != null",
+        "source": "for (field in params.fields) { if (ctx.mysql.slowlog[field] instanceof List) { def vals = ctx.mysql.slowlog[field]; vals = vals.stream().distinct().collect(Collectors.toList()); if (vals.size() == 1) { ctx.mysql.slowlog[field] = vals[0] } else { ctx.mysql.slowlog[field] = vals } } }",
+        "params": {
+          "fields": [
+            "schema"
+          ]
+        }
+      }
+    },
     {
       "remove": {
         "field": "message"