From f6f28fd82dd58da0a62595601b2d2ba00e9842a0 Mon Sep 17 00:00:00 2001
From: Kibana Machine <42973632+kibanamachine@users.noreply.github.com>
Date: Fri, 6 Dec 2024 02:52:32 +1100
Subject: [PATCH] =?UTF-8?q?[8.x]=20=F0=9F=8C=8A=20Add=20type=20safety=20to?=
 =?UTF-8?q?=20Painless=20conditions=20(#202603)=20(#203104)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

# Backport

This will backport the following commits from `main` to `8.x`:
- [🌊 Add type safety to Painless conditions
(#202603)](https://github.com/elastic/kibana/pull/202603)

<!--- Backport version: 9.4.3 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Chris
Cowan","email":"chris@elastic.co"},"sourceCommit":{"committedDate":"2024-12-05T13:58:36Z","message":"🌊
Add type safety to Painless conditions (#202603)\n\n## 🍒
Summary\r\n\r\nThis PR closes
https://github.com/elastic/streams-program/issues/18 by\r\nadding some
basic type checking to the painless output for the Stream\r\nconditions.
The new code will check to ensure that none of the fields\r\nused in the
condition are `Map` objects. Then it wraps the if statement\r\nin a
`try/catch`.\r\n\r\n### Condition\r\n```Typescript\r\n{\r\n and: [\r\n {
field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy'
},\r\n {\r\n or: [\r\n { field: 'log.level', operator: 'eq' as const,
value: 'error' },\r\n { field: 'log.level', operator: 'eq' as const,
value: 'ERROR' },\r\n ],\r\n },\r\n ],\r\n}\r\n```\r\n\r\n###
Before\r\n\r\n```\r\n(ctx.log?.logger !== null && ctx.log?.logger ==
\"nginx_proxy\") && ((ctx.log?.level !== null && ctx.log?.level ==
\"error\") || (ctx.log?.level !== null && ctx.log?.level ==
\"ERROR\"))\r\n```\r\n\r\n### After\r\n\r\n```\r\nif (ctx.log?.logger
instanceof Map || ctx.log?.level instanceof Map) {\r\n return
false;\r\n}\r\ntry {\r\n if ((ctx.log?.logger !== null &&
ctx.log?.logger == \"nginx_proxy\") && ((ctx.log?.level !== null &&
ctx.log?.level == \"error\") || (ctx.log?.level !== null &&
ctx.log?.level == \"ERROR\"))) {\r\n return true;\r\n }\r\n return
false;\r\n} catch (Exception e) {\r\n return
false;\r\n}\r\n```","sha":"fe56d6d90af45b971ff82731c1ec7eb4c4c0eff3","branchLabelMapping":{"^v9.0.0$":"main","^v8.18.0$":"8.x","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["release_note:skip","v9.0.0","backport:prev-minor","v8.18.0","Feature:Streams"],"title":"🌊
Add type safety to Painless
conditions","number":202603,"url":"https://github.com/elastic/kibana/pull/202603","mergeCommit":{"message":"🌊
Add type safety to Painless conditions (#202603)\n\n## 🍒
Summary\r\n\r\nThis PR closes
https://github.com/elastic/streams-program/issues/18 by\r\nadding some
basic type checking to the painless output for the Stream\r\nconditions.
The new code will check to ensure that none of the fields\r\nused in the
condition are `Map` objects. Then it wraps the if statement\r\nin a
`try/catch`.\r\n\r\n### Condition\r\n```Typescript\r\n{\r\n and: [\r\n {
field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy'
},\r\n {\r\n or: [\r\n { field: 'log.level', operator: 'eq' as const,
value: 'error' },\r\n { field: 'log.level', operator: 'eq' as const,
value: 'ERROR' },\r\n ],\r\n },\r\n ],\r\n}\r\n```\r\n\r\n###
Before\r\n\r\n```\r\n(ctx.log?.logger !== null && ctx.log?.logger ==
\"nginx_proxy\") && ((ctx.log?.level !== null && ctx.log?.level ==
\"error\") || (ctx.log?.level !== null && ctx.log?.level ==
\"ERROR\"))\r\n```\r\n\r\n### After\r\n\r\n```\r\nif (ctx.log?.logger
instanceof Map || ctx.log?.level instanceof Map) {\r\n return
false;\r\n}\r\ntry {\r\n if ((ctx.log?.logger !== null &&
ctx.log?.logger == \"nginx_proxy\") && ((ctx.log?.level !== null &&
ctx.log?.level == \"error\") || (ctx.log?.level !== null &&
ctx.log?.level == \"ERROR\"))) {\r\n return true;\r\n }\r\n return
false;\r\n} catch (Exception e) {\r\n return
false;\r\n}\r\n```","sha":"fe56d6d90af45b971ff82731c1ec7eb4c4c0eff3"}},"sourceBranch":"main","suggestedTargetBranches":["8.x"],"targetPullRequestStates":[{"branch":"main","label":"v9.0.0","branchLabelMappingKey":"^v9.0.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/202603","number":202603,"mergeCommit":{"message":"🌊
Add type safety to Painless conditions (#202603)\n\n## 🍒
Summary\r\n\r\nThis PR closes
https://github.com/elastic/streams-program/issues/18 by\r\nadding some
basic type checking to the painless output for the Stream\r\nconditions.
The new code will check to ensure that none of the fields\r\nused in the
condition are `Map` objects. Then it wraps the if statement\r\nin a
`try/catch`.\r\n\r\n### Condition\r\n```Typescript\r\n{\r\n and: [\r\n {
field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy'
},\r\n {\r\n or: [\r\n { field: 'log.level', operator: 'eq' as const,
value: 'error' },\r\n { field: 'log.level', operator: 'eq' as const,
value: 'ERROR' },\r\n ],\r\n },\r\n ],\r\n}\r\n```\r\n\r\n###
Before\r\n\r\n```\r\n(ctx.log?.logger !== null && ctx.log?.logger ==
\"nginx_proxy\") && ((ctx.log?.level !== null && ctx.log?.level ==
\"error\") || (ctx.log?.level !== null && ctx.log?.level ==
\"ERROR\"))\r\n```\r\n\r\n### After\r\n\r\n```\r\nif (ctx.log?.logger
instanceof Map || ctx.log?.level instanceof Map) {\r\n return
false;\r\n}\r\ntry {\r\n if ((ctx.log?.logger !== null &&
ctx.log?.logger == \"nginx_proxy\") && ((ctx.log?.level !== null &&
ctx.log?.level == \"error\") || (ctx.log?.level !== null &&
ctx.log?.level == \"ERROR\"))) {\r\n return true;\r\n }\r\n return
false;\r\n} catch (Exception e) {\r\n return
false;\r\n}\r\n```","sha":"fe56d6d90af45b971ff82731c1ec7eb4c4c0eff3"}},{"branch":"8.x","label":"v8.18.0","branchLabelMappingKey":"^v8.18.0$","isSourceBranch":false,"state":"NOT_CREATED"}]}]
BACKPORT-->

Co-authored-by: Chris Cowan <chris@elastic.co>
---
 .../helpers/condition_to_painless.test.ts     | 242 ++++++++++++------
 .../streams/helpers/condition_to_painless.ts  | 102 ++++++--
 .../api_integration/apis/streams/full_flow.ts | 131 ++++++++++
 3 files changed, 376 insertions(+), 99 deletions(-)

diff --git a/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.test.ts b/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.test.ts
index db0a5eaea8efd..6927b85d5aed8 100644
--- a/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.test.ts
+++ b/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.test.ts
@@ -5,44 +5,53 @@
  * 2.0.
  */
 
-import { conditionToPainless } from './condition_to_painless';
+import { conditionToPainless, conditionToStatement } from './condition_to_painless';
 
 const operatorConditionAndResults = [
   {
     condition: { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
-    result: '(ctx.log?.logger !== null && ctx.log?.logger == "nginx_proxy")',
+    result:
+      '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy"))',
   },
   {
     condition: { field: 'log.logger', operator: 'neq' as const, value: 'nginx_proxy' },
-    result: '(ctx.log?.logger !== null && ctx.log?.logger != "nginx_proxy")',
+    result:
+      '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() != "nginx_proxy") || ctx.log?.logger != "nginx_proxy"))',
   },
   {
     condition: { field: 'http.response.status_code', operator: 'lt' as const, value: 500 },
-    result: '(ctx.http?.response?.status_code !== null && ctx.http?.response?.status_code < 500)',
+    result:
+      '(ctx.http?.response?.status_code !== null && ((ctx.http?.response?.status_code instanceof String && Float.parseFloat(ctx.http?.response?.status_code) < 500) || ctx.http?.response?.status_code < 500))',
   },
   {
     condition: { field: 'http.response.status_code', operator: 'lte' as const, value: 500 },
-    result: '(ctx.http?.response?.status_code !== null && ctx.http?.response?.status_code <= 500)',
+    result:
+      '(ctx.http?.response?.status_code !== null && ((ctx.http?.response?.status_code instanceof String && Float.parseFloat(ctx.http?.response?.status_code) <= 500) || ctx.http?.response?.status_code <= 500))',
   },
   {
     condition: { field: 'http.response.status_code', operator: 'gt' as const, value: 500 },
-    result: '(ctx.http?.response?.status_code !== null && ctx.http?.response?.status_code > 500)',
+    result:
+      '(ctx.http?.response?.status_code !== null && ((ctx.http?.response?.status_code instanceof String && Float.parseFloat(ctx.http?.response?.status_code) > 500) || ctx.http?.response?.status_code > 500))',
   },
   {
     condition: { field: 'http.response.status_code', operator: 'gte' as const, value: 500 },
-    result: '(ctx.http?.response?.status_code !== null && ctx.http?.response?.status_code >= 500)',
+    result:
+      '(ctx.http?.response?.status_code !== null && ((ctx.http?.response?.status_code instanceof String && Float.parseFloat(ctx.http?.response?.status_code) >= 500) || ctx.http?.response?.status_code >= 500))',
   },
   {
     condition: { field: 'log.logger', operator: 'startsWith' as const, value: 'nginx' },
-    result: '(ctx.log?.logger !== null && ctx.log?.logger.startsWith("nginx"))',
+    result:
+      '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString().startsWith("nginx")) || ctx.log?.logger.startsWith("nginx")))',
   },
   {
     condition: { field: 'log.logger', operator: 'endsWith' as const, value: 'proxy' },
-    result: '(ctx.log?.logger !== null && ctx.log?.logger.endsWith("proxy"))',
+    result:
+      '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString().endsWith("proxy")) || ctx.log?.logger.endsWith("proxy")))',
   },
   {
     condition: { field: 'log.logger', operator: 'contains' as const, value: 'proxy' },
-    result: '(ctx.log?.logger !== null && ctx.log?.logger.contains("proxy"))',
+    result:
+      '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString().contains("proxy")) || ctx.log?.logger.contains("proxy")))',
   },
   {
     condition: { field: 'log.logger', operator: 'exists' as const },
@@ -55,87 +64,152 @@ const operatorConditionAndResults = [
 ];
 
 describe('conditionToPainless', () => {
-  describe('operators', () => {
-    operatorConditionAndResults.forEach((setup) => {
-      test(`${setup.condition.operator}`, () => {
-        expect(conditionToPainless(setup.condition)).toEqual(setup.result);
+  describe('conditionToStatement', () => {
+    describe('operators', () => {
+      operatorConditionAndResults.forEach((setup) => {
+        test(`${setup.condition.operator}`, () => {
+          expect(conditionToStatement(setup.condition)).toEqual(setup.result);
+        });
       });
-    });
-  });
 
-  describe('and', () => {
-    test('simple', () => {
-      const condition = {
-        and: [
-          { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
-          { field: 'log.level', operator: 'eq' as const, value: 'error' },
-        ],
-      };
-      expect(
-        expect(conditionToPainless(condition)).toEqual(
-          '(ctx.log?.logger !== null && ctx.log?.logger == "nginx_proxy") && (ctx.log?.level !== null && ctx.log?.level == "error")'
-        )
-      );
+      test('ensure number comparasion works with string values', () => {
+        const condition = {
+          field: 'http.response.status_code',
+          operator: 'gt' as const,
+          value: '500',
+        };
+        expect(conditionToStatement(condition)).toEqual(
+          '(ctx.http?.response?.status_code !== null && ((ctx.http?.response?.status_code instanceof String && Float.parseFloat(ctx.http?.response?.status_code) > 500) || ctx.http?.response?.status_code > 500))'
+        );
+      });
+      test('ensure string comparasion works with number values', () => {
+        const condition = {
+          field: 'message',
+          operator: 'contains' as const,
+          value: 500,
+        };
+        expect(conditionToStatement(condition)).toEqual(
+          '(ctx.message !== null && ((ctx.message instanceof Number && ctx.message.toString().contains("500")) || ctx.message.contains("500")))'
+        );
+      });
     });
-  });
 
-  describe('or', () => {
-    test('simple', () => {
-      const condition = {
-        or: [
-          { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
-          { field: 'log.level', operator: 'eq' as const, value: 'error' },
-        ],
-      };
-      expect(
-        expect(conditionToPainless(condition)).toEqual(
-          '(ctx.log?.logger !== null && ctx.log?.logger == "nginx_proxy") || (ctx.log?.level !== null && ctx.log?.level == "error")'
-        )
-      );
+    describe('and', () => {
+      test('simple', () => {
+        const condition = {
+          and: [
+            { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
+            { field: 'log.level', operator: 'eq' as const, value: 'error' },
+          ],
+        };
+        expect(
+          expect(conditionToStatement(condition)).toEqual(
+            '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy")) && (ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "error") || ctx.log?.level == "error"))'
+          )
+        );
+      });
     });
-  });
 
-  describe('nested', () => {
-    test('and with a filter and or with 2 filters', () => {
-      const condition = {
-        and: [
-          { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
-          {
-            or: [
-              { field: 'log.level', operator: 'eq' as const, value: 'error' },
-              { field: 'log.level', operator: 'eq' as const, value: 'ERROR' },
-            ],
-          },
-        ],
-      };
-      expect(
-        expect(conditionToPainless(condition)).toEqual(
-          '(ctx.log?.logger !== null && ctx.log?.logger == "nginx_proxy") && ((ctx.log?.level !== null && ctx.log?.level == "error") || (ctx.log?.level !== null && ctx.log?.level == "ERROR"))'
-        )
-      );
+    describe('or', () => {
+      test('simple', () => {
+        const condition = {
+          or: [
+            { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
+            { field: 'log.level', operator: 'eq' as const, value: 'error' },
+          ],
+        };
+        expect(
+          expect(conditionToStatement(condition)).toEqual(
+            '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy")) || (ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "error") || ctx.log?.level == "error"))'
+          )
+        );
+      });
     });
-    test('and with 2 or with filters', () => {
-      const condition = {
-        and: [
-          {
-            or: [
-              { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
-              { field: 'service.name', operator: 'eq' as const, value: 'nginx' },
-            ],
-          },
-          {
-            or: [
-              { field: 'log.level', operator: 'eq' as const, value: 'error' },
-              { field: 'log.level', operator: 'eq' as const, value: 'ERROR' },
-            ],
-          },
-        ],
-      };
-      expect(
-        expect(conditionToPainless(condition)).toEqual(
-          '((ctx.log?.logger !== null && ctx.log?.logger == "nginx_proxy") || (ctx.service?.name !== null && ctx.service?.name == "nginx")) && ((ctx.log?.level !== null && ctx.log?.level == "error") || (ctx.log?.level !== null && ctx.log?.level == "ERROR"))'
-        )
-      );
+
+    describe('nested', () => {
+      test('and with a filter and or with 2 filters', () => {
+        const condition = {
+          and: [
+            { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
+            {
+              or: [
+                { field: 'log.level', operator: 'eq' as const, value: 'error' },
+                { field: 'log.level', operator: 'eq' as const, value: 'ERROR' },
+              ],
+            },
+          ],
+        };
+        expect(
+          expect(conditionToStatement(condition)).toEqual(
+            '(ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy")) && ((ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "error") || ctx.log?.level == "error")) || (ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "ERROR") || ctx.log?.level == "ERROR")))'
+          )
+        );
+      });
+      test('and with 2 or with filters', () => {
+        const condition = {
+          and: [
+            {
+              or: [
+                { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
+                { field: 'service.name', operator: 'eq' as const, value: 'nginx' },
+              ],
+            },
+            {
+              or: [
+                { field: 'log.level', operator: 'eq' as const, value: 'error' },
+                { field: 'log.level', operator: 'eq' as const, value: 'ERROR' },
+              ],
+            },
+          ],
+        };
+        expect(
+          expect(conditionToStatement(condition)).toEqual(
+            '((ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy")) || (ctx.service?.name !== null && ((ctx.service?.name instanceof Number && ctx.service?.name.toString() == "nginx") || ctx.service?.name == "nginx"))) && ((ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "error") || ctx.log?.level == "error")) || (ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "ERROR") || ctx.log?.level == "ERROR")))'
+          )
+        );
+      });
     });
   });
+
+  test('wrapped with type checks for uinary conditions', () => {
+    const condition = { field: 'log', operator: 'exists' as const };
+    expect(conditionToPainless(condition)).toEqual(`try {
+  if (ctx.log !== null) {
+    return true;
+  }
+  return false;
+} catch (Exception e) {
+  return false;
+}
+`);
+  });
+
+  test('wrapped with typechecks and try/catch', () => {
+    const condition = {
+      and: [
+        { field: 'log.logger', operator: 'eq' as const, value: 'nginx_proxy' },
+        {
+          or: [
+            { field: 'log.level', operator: 'eq' as const, value: 'error' },
+            { field: 'log.level', operator: 'eq' as const, value: 'ERROR' },
+          ],
+        },
+      ],
+    };
+    expect(
+      expect(conditionToPainless(condition))
+        .toEqual(`if (ctx.log?.logger instanceof Map || ctx.log?.level instanceof Map) {
+  return false;
+}
+try {
+  if ((ctx.log?.logger !== null && ((ctx.log?.logger instanceof Number && ctx.log?.logger.toString() == "nginx_proxy") || ctx.log?.logger == "nginx_proxy")) && ((ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "error") || ctx.log?.level == "error")) || (ctx.log?.level !== null && ((ctx.log?.level instanceof Number && ctx.log?.level.toString() == "ERROR") || ctx.log?.level == "ERROR")))) {
+    return true;
+  }
+  return false;
+} catch (Exception e) {
+  return false;
+}
+`)
+    );
+  });
 });
diff --git a/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts b/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts
index dccc15b2ec8fc..3585d009ee090 100644
--- a/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts
+++ b/x-pack/plugins/streams/server/lib/streams/helpers/condition_to_painless.ts
@@ -5,7 +5,7 @@
  * 2.0.
  */
 
-import { isBoolean, isString } from 'lodash';
+import { isBoolean, isString, uniq } from 'lodash';
 import {
   AndCondition,
   BinaryFilterCondition,
@@ -32,8 +32,11 @@ function isOrCondition(subject: any): subject is RerouteOrCondition {
   return result.success && subject.or != null;
 }
 
-function safePainlessField(condition: FilterCondition) {
-  return `ctx.${condition.field.split('.').join('?.')}`;
+function safePainlessField(conditionOrField: FilterCondition | string) {
+  if (isFilterCondition(conditionOrField)) {
+    return `ctx.${conditionOrField.field.split('.').join('?.')}`;
+  }
+  return `ctx.${conditionOrField.split('.').join('?.')}`;
 }
 
 function encodeValue(value: string | number | boolean) {
@@ -49,23 +52,59 @@ function encodeValue(value: string | number | boolean) {
 function binaryToPainless(condition: BinaryFilterCondition) {
   switch (condition.operator) {
     case 'neq':
-      return `${safePainlessField(condition)} != ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(condition)} instanceof Number && ${safePainlessField(
+        condition
+      )}.toString() != ${encodeValue(String(condition.value))}) || ${safePainlessField(
+        condition
+      )} != ${encodeValue(String(condition.value))})`;
     case 'lt':
-      return `${safePainlessField(condition)} < ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(
+        condition
+      )} instanceof String && Float.parseFloat(${safePainlessField(condition)}) < ${encodeValue(
+        Number(condition.value)
+      )}) || ${safePainlessField(condition)} < ${encodeValue(Number(condition.value))})`;
     case 'lte':
-      return `${safePainlessField(condition)} <= ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(
+        condition
+      )} instanceof String && Float.parseFloat(${safePainlessField(condition)}) <= ${encodeValue(
+        Number(condition.value)
+      )}) || ${safePainlessField(condition)} <= ${encodeValue(Number(condition.value))})`;
     case 'gt':
-      return `${safePainlessField(condition)} > ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(
+        condition
+      )} instanceof String && Float.parseFloat(${safePainlessField(condition)}) > ${encodeValue(
+        Number(condition.value)
+      )}) || ${safePainlessField(condition)} > ${encodeValue(Number(condition.value))})`;
     case 'gte':
-      return `${safePainlessField(condition)} >= ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(
+        condition
+      )} instanceof String && Float.parseFloat(${safePainlessField(condition)}) >= ${encodeValue(
+        Number(condition.value)
+      )}) || ${safePainlessField(condition)} >= ${encodeValue(Number(condition.value))})`;
     case 'startsWith':
-      return `${safePainlessField(condition)}.startsWith(${encodeValue(condition.value)})`;
+      return `((${safePainlessField(condition)} instanceof Number && ${safePainlessField(
+        condition
+      )}.toString().startsWith(${encodeValue(String(condition.value))})) || ${safePainlessField(
+        condition
+      )}.startsWith(${encodeValue(String(condition.value))}))`;
     case 'endsWith':
-      return `${safePainlessField(condition)}.endsWith(${encodeValue(condition.value)})`;
+      return `((${safePainlessField(condition)} instanceof Number && ${safePainlessField(
+        condition
+      )}.toString().endsWith(${encodeValue(String(condition.value))})) || ${safePainlessField(
+        condition
+      )}.endsWith(${encodeValue(String(condition.value))}))`;
     case 'contains':
-      return `${safePainlessField(condition)}.contains(${encodeValue(condition.value)})`;
+      return `((${safePainlessField(condition)} instanceof Number && ${safePainlessField(
+        condition
+      )}.toString().contains(${encodeValue(String(condition.value))})) || ${safePainlessField(
+        condition
+      )}.contains(${encodeValue(String(condition.value))}))`;
     default:
-      return `${safePainlessField(condition)} == ${encodeValue(condition.value)}`;
+      return `((${safePainlessField(condition)} instanceof Number && ${safePainlessField(
+        condition
+      )}.toString() == ${encodeValue(String(condition.value))}) || ${safePainlessField(
+        condition
+      )} == ${encodeValue(String(condition.value))})`;
   }
 }
 
@@ -82,7 +121,18 @@ function isUnaryFilterCondition(subject: FilterCondition): subject is UnaryFilte
   return !('value' in subject);
 }
 
-export function conditionToPainless(condition: Condition, nested = false): string {
+function extractAllFields(condition: Condition, fields: string[] = []): string[] {
+  if (isFilterCondition(condition) && !isUnaryFilterCondition(condition)) {
+    return uniq([...fields, condition.field]);
+  } else if (isAndCondition(condition)) {
+    return uniq(condition.and.map((cond) => extractAllFields(cond, fields)).flat());
+  } else if (isOrCondition(condition)) {
+    return uniq(condition.or.map((cond) => extractAllFields(cond, fields)).flat());
+  }
+  return uniq(fields);
+}
+
+export function conditionToStatement(condition: Condition, nested = false): string {
   if (isFilterCondition(condition)) {
     if (isUnaryFilterCondition(condition)) {
       return unaryToPainless(condition);
@@ -90,12 +140,34 @@ export function conditionToPainless(condition: Condition, nested = false): strin
     return `(${safePainlessField(condition)} !== null && ${binaryToPainless(condition)})`;
   }
   if (isAndCondition(condition)) {
-    const and = condition.and.map((filter) => conditionToPainless(filter, true)).join(' && ');
+    const and = condition.and.map((filter) => conditionToStatement(filter, true)).join(' && ');
     return nested ? `(${and})` : and;
   }
   if (isOrCondition(condition)) {
-    const or = condition.or.map((filter) => conditionToPainless(filter, true)).join(' || ');
+    const or = condition.or.map((filter) => conditionToStatement(filter, true)).join(' || ');
     return nested ? `(${or})` : or;
   }
   return 'false';
 }
+
+export function conditionToPainless(condition: Condition): string {
+  const fields = extractAllFields(condition);
+  let fieldCheck = '';
+  if (fields.length !== 0) {
+    fieldCheck = `if (${fields
+      .map((field) => `${safePainlessField(field)} instanceof Map`)
+      .join(' || ')}) {
+  return false;
+}
+`;
+  }
+  return `${fieldCheck}try {
+  if (${conditionToStatement(condition)}) {
+    return true;
+  }
+  return false;
+} catch (Exception e) {
+  return false;
+}
+`;
+}
diff --git a/x-pack/test/api_integration/apis/streams/full_flow.ts b/x-pack/test/api_integration/apis/streams/full_flow.ts
index 03c0cc9e0e219..aad931ab11816 100644
--- a/x-pack/test/api_integration/apis/streams/full_flow.ts
+++ b/x-pack/test/api_integration/apis/streams/full_flow.ts
@@ -135,6 +135,137 @@ export default function ({ getService }: FtrProviderContext) {
           log: { level: 'info', logger: 'nginx' },
         });
       });
+
+      it('Fork logs to logs.nginx.error with invalid condition', async () => {
+        const body = {
+          stream: {
+            id: 'logs.nginx.error',
+            fields: [],
+            processing: [],
+          },
+          condition: { field: 'log', operator: 'eq', value: 'error' },
+        };
+        const response = await forkStream(supertest, 'logs.nginx', body);
+        expect(response).to.have.property('acknowledged', true);
+      });
+
+      it('Index an Nginx error log message, should goto logs.nginx.error but fails', async () => {
+        const doc = {
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: JSON.stringify({
+            'log.level': 'error',
+            'log.logger': 'nginx',
+            message: 'test',
+          }),
+        };
+        const response = await indexDocument(esClient, 'logs', doc);
+        expect(response.result).to.eql('created');
+
+        await waitForDocumentInIndex({
+          esClient,
+          indexName: 'logs.nginx',
+          retryService,
+          logger,
+          docCountTarget: 2,
+        });
+
+        const result = await fetchDocument(esClient, 'logs.nginx', response._id);
+        expect(result._index).to.match(/^\.ds\-logs.nginx-.*/);
+        expect(result._source).to.eql({
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: 'test',
+          log: { level: 'error', logger: 'nginx' },
+        });
+      });
+
+      it('Fork logs to logs.number-test', async () => {
+        const body = {
+          stream: {
+            id: 'logs.number-test',
+            fields: [],
+            processing: [],
+          },
+          condition: { field: 'code', operator: 'gte', value: '500' },
+        };
+        const response = await forkStream(supertest, 'logs', body);
+        expect(response).to.have.property('acknowledged', true);
+      });
+
+      it('Index documents with numbers and strings for logs.number-test condition', async () => {
+        const doc1 = {
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: JSON.stringify({
+            code: '500',
+            message: 'test',
+          }),
+        };
+        const doc2 = {
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: JSON.stringify({
+            code: 500,
+            message: 'test',
+          }),
+        };
+        const response1 = await indexDocument(esClient, 'logs', doc1);
+        expect(response1.result).to.eql('created');
+        const response2 = await indexDocument(esClient, 'logs', doc2);
+        expect(response2.result).to.eql('created');
+
+        await waitForDocumentInIndex({
+          esClient,
+          indexName: 'logs.number-test',
+          retryService,
+          logger,
+          docCountTarget: 2,
+          retries: 20,
+        });
+      });
+
+      it('Fork logs to logs.string-test', async () => {
+        const body = {
+          stream: {
+            id: 'logs.string-test',
+            fields: [],
+            processing: [],
+          },
+          condition: {
+            or: [
+              { field: 'message', operator: 'contains', value: '500' },
+              { field: 'message', operator: 'contains', value: 400 },
+            ],
+          },
+        };
+        const response = await forkStream(supertest, 'logs', body);
+        expect(response).to.have.property('acknowledged', true);
+      });
+
+      it('Index documents with numbers and strings for logs.string-test condition', async () => {
+        const doc1 = {
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: JSON.stringify({
+            message: 'status_code: 500',
+          }),
+        };
+        const doc2 = {
+          '@timestamp': '2024-01-01T00:00:20.000Z',
+          message: JSON.stringify({
+            message: 'status_code: 400',
+          }),
+        };
+        const response1 = await indexDocument(esClient, 'logs', doc1);
+        expect(response1.result).to.eql('created');
+        const response2 = await indexDocument(esClient, 'logs', doc2);
+        expect(response2.result).to.eql('created');
+
+        await waitForDocumentInIndex({
+          esClient,
+          indexName: 'logs.string-test',
+          retryService,
+          logger,
+          docCountTarget: 2,
+          retries: 20,
+        });
+      });
     });
   });
 }