painless: register ProcessorsWhitelistExtension to painless engine (#162)

* painless: register available PainlessExtensions to painless engine

Enables the use of the `org.elasticsearch.ingest.common.Processors` in ingest
pipelines, as is found in integrations like `logs-aws.cloudtrail`.

Fixes: #161

* Add an integration test for the painless script which points to Ingest Processors.

* Update CHANGELOG.md

---------

Co-authored-by: Mashhur <[email protected]>

Author: yaauie

CHANGELOG.md

@@ -1,3 +1,6 @@
+## 0.1.14
+  - Fix: register available PainlessExtension-s, resolving an issue where the pipelines for some integrations would fail to compile [#162](https://github.com/elastic/logstash-filter-elastic_integration/pull/162)
+
 ## 0.1.13
   - Update default elasticsearch tree branch to 8.15 [#156](https://github.com/elastic/logstash-filter-elastic_integration/pull/156)
 

VERSION

@@ -1 +1 @@
-0.1.13
+0.1.14

build.gradle

@@ -354,6 +354,16 @@ task importMinimalElasticsearch() {
                 include jarPackageNamed("spi/elasticsearch-scripting-painless-spi")
             }
 
+            from(buildElasticsearchLocalDistro.module("wildcard")) {
+                include jarPackageNamed("x-pack-wildcard")
+            }
+            from(buildElasticsearchLocalDistro.module("constant-keyword")) {
+                include jarPackageNamed("x-pack-constant-keyword")
+            }
+            from(buildElasticsearchLocalDistro.module("spatial")) {
+                include jarPackageNamed("spatial")
+            }
+
             into ext.jars
 
             includeEmptyDirs(false)

spec/integration/elastic_integration_spec.rb

@@ -745,33 +745,65 @@
 
     end
 
-    describe 'with script processor' do
-      let(:pipeline_processor) {
-        '{
-          "script": {
-            "lang": "painless",
-            "source": "ctx[\'_index\'] = ctx[\'lang\'] + \'-\' + params[\'dataset\'];",
-            "params": {
-              "dataset": "catalog"
+    context "#painless script" do
+
+      describe 'with simple script' do
+        let(:pipeline_processor) {
+          '{
+            "script": {
+              "lang": "painless",
+              "source": "ctx[\'_index\'] = ctx[\'lang\'] + \'-\' + params[\'dataset\'];",
+              "params": {
+                "dataset": "catalog"
+              }
             }
-          }
-        }'
-      }
+          }'
+        }
 
-      it 'runs painless script on a given field' do
-        events = [LogStash::Event.new(
-          "message" => "Should extract prod tag from env field",
-          "lang" => "uz",
-          "data_stream" => data_stream)]
+        it 'runs painless script on a given field' do
+          events = [LogStash::Event.new(
+            "message" => "Should extract prod tag from env field",
+            "lang" => "uz",
+            "data_stream" => data_stream)]
 
-        subject.multi_filter(events).each do |event|
-          expect(event.get("[@metadata][_ingest_document][index]")).to eql "uz-catalog"
-          expect(event.get("[@metadata][target_ingest_pipeline]")).to eql '_none'
+          subject.multi_filter(events).each do |event|
+            expect(event.get("[@metadata][_ingest_document][index]")).to eql "uz-catalog"
+            expect(event.get("[@metadata][target_ingest_pipeline]")).to eql '_none'
+          end
         end
+
       end
 
+      # makes sure Ingest Processors can be loaded
+      describe 'with ingest Processors in the script' do
+        let(:pipeline_processor) {
+          '{
+            "script": {
+              "lang": "painless",
+              "source": "long bytes = Processors.bytes(params[\'size\']); ctx.size_in_bytes = bytes;",
+              "params": {
+                "size": "1kb"
+              }
+            }
+          }'
+        }
+
+        it 'calculates the bytes with Processors.bytes()' do
+          events = [LogStash::Event.new(
+            "message" => "Should run painless script which has an ingest Processor#bytes in it",
+            "data_stream" => data_stream)]
+
+          subject.multi_filter(events).each do |event|
+            # make sure AT LEAST we don't get pipeline load error
+            expect(event.get("[@metadata][_ingest_document][index]")).to eql pipeline_setting
+            expect(event.get("[@metadata][target_ingest_pipeline]")).to eql '_none'
+          end
+        end
+      end
     end
 
+    # TODO: add more painless script integration tests
+
     describe 'with set processor' do
       let(:pipeline_processor) {
         '{

src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java

@@ -28,21 +28,27 @@
 import org.elasticsearch.ingest.LogstashInternalBridge;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.common.IngestCommonPlugin;
+import org.elasticsearch.ingest.common.ProcessorsWhitelistExtension;
 import org.elasticsearch.ingest.useragent.IngestUserAgentPlugin;
 import org.elasticsearch.painless.PainlessPlugin;
 import org.elasticsearch.painless.PainlessScriptEngine;
+import org.elasticsearch.painless.spi.PainlessExtension;
 import org.elasticsearch.painless.spi.Whitelist;
+import org.elasticsearch.plugins.ExtensiblePlugin;
 import org.elasticsearch.plugins.IngestPlugin;
 import org.elasticsearch.script.IngestConditionalScript;
 import org.elasticsearch.script.IngestScript;
-import org.elasticsearch.script.ScriptContext;
 import org.elasticsearch.script.ScriptEngine;
 import org.elasticsearch.script.ScriptModule;
 import org.elasticsearch.script.ScriptService;
 import org.elasticsearch.script.mustache.MustacheScriptEngine;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.xpack.constantkeyword.ConstantKeywordPainlessExtension;
+import org.elasticsearch.xpack.spatial.SpatialPainlessExtension;
+import org.elasticsearch.xpack.wildcard.WildcardPainlessExtension;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.HashMap;
@@ -307,37 +313,42 @@ public synchronized EventProcessor build(final PluginContext pluginContext) {
         }
     }
 
-    private static ScriptService initScriptService(final Settings settings, final ThreadPool threadPool) {
-        final List<Whitelist> painlessBaseWhitelist = getPainlessBaseWhiteList();
-        final Map<ScriptContext<?>, List<Whitelist>> scriptContexts = Map.of(
-                IngestScript.CONTEXT, painlessBaseWhitelist,
-                IngestConditionalScript.CONTEXT, painlessBaseWhitelist);
-
+    private static ScriptService initScriptService(final Settings settings, final ThreadPool threadPool) throws IOException {
         Map<String, ScriptEngine> engines = new HashMap<>();
-        engines.put(PainlessScriptEngine.NAME, new PainlessScriptEngine(settings, scriptContexts));
+        engines.put(PainlessScriptEngine.NAME, getPainlessScriptEngine(settings));
         engines.put(MustacheScriptEngine.NAME, new MustacheScriptEngine());
         return new ScriptService(settings, engines, ScriptModule.CORE_CONTEXTS, threadPool::absoluteTimeInMillis);
     }
 
     /**
-     * @implNote handles breaking changes introduced in Elasticsearch 8.14 series; once 8.14 is
-     *           released and all builds of this plugin depend on Elasticsearch 8.14+, this can
-     *           be simplified to call {@code PainlessPlugin.baseWhiteList()} directly.
-     * @return the PainlessPlugin's default base whitelists
+     * @param settings the Elasticsearch settings object
+     * @return a {@link ScriptEngine} for painless scripts for use in {@link IngestScript} and
+     *         {@link IngestConditionalScript} contexts, including all available {@link PainlessExtension}s.
+     * @throws IOException when the underlying script engine cannot be created
      */
-    @SuppressWarnings({"JavaReflectionMemberAccess", "unchecked"})
-    static List<Whitelist> getPainlessBaseWhiteList() {
-        Class<PainlessPlugin> cls = PainlessPlugin.class;
-        try {
-            try {
-                // In 8.14+: PainlessPlugin.baseWhiteList()
-                return (List<Whitelist>) cls.getMethod("baseWhiteList").invoke(null);
-            } catch (NoSuchMethodException e) {
-                // in 8.x->8.13.x: PainlessPlugin.BASE_WHITELISTS
-                return (List<Whitelist>) cls.getField("BASE_WHITELISTS").get(null);
-            }
-        } catch (java.lang.reflect.InvocationTargetException | IllegalAccessException | NoSuchFieldException e) {
-            throw new RuntimeException("Unsupported PainlessPlugin does not provide access to its base whitelists", e);
+    private static ScriptEngine getPainlessScriptEngine(final Settings settings) throws IOException {
+        try (final PainlessPlugin painlessPlugin = new PainlessPlugin()) {
+
+            painlessPlugin.loadExtensions(new ExtensiblePlugin.ExtensionLoader() {
+                @Override
+                @SuppressWarnings("unchecked")
+                public <T> List<T> loadExtensions(Class<T> extensionPointType) {
+                    if (extensionPointType.isAssignableFrom(PainlessExtension.class)) {
+                        final List<PainlessExtension> extensions = new ArrayList<>();
+
+                        extensions.add(new ConstantKeywordPainlessExtension()); // module: constant-keyword
+                        extensions.add(new ProcessorsWhitelistExtension());     // module: ingest-common
+                        extensions.add(new SpatialPainlessExtension());         // module: spatial
+                        extensions.add(new WildcardPainlessExtension());        // module: wildcard
+
+                        return (List<T>) extensions;
+                    } else {
+                        return List.of();
+                    }
+                }
+            });
+
+            return painlessPlugin.getScriptEngine(settings, Set.of(IngestScript.CONTEXT, IngestConditionalScript.CONTEXT));
         }
     }
 }

src/test/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilderTest.java

@@ -237,6 +237,33 @@ public void testSinglePipelineMutatingEvents() {
         }));
     }
 
+    @Test void testPainlessAccessToIngestCommonProcessors() {
+        final List<Event> matchedEvents = new ArrayList<>();
+        final EventProcessorBuilder eventProcessorBuilder = EventProcessor.builder()
+                .setEventPipelineNameResolver((event, exceptionConsumer) -> Optional.of("pipeline"))
+                .setEventIndexNameResolver((event, handler) -> Optional.empty()) // no index name
+                .setIndexNamePipelineNameResolver(((indexName, handler) -> Optional.empty())) // no default pipeline
+                .setPipelineConfigurationResolver(new LocalDirectoryPipelineConfigurationResolver(getPreparedPipelinesResourcePath("script-processor-pipelines")))
+                .setFilterMatchListener(matchedEvents::add);
+
+        final List<Event> inputEvents = List.of(
+                newEvent(Map.of("id", "baseline", "lower", "lower","mixed", "MiXeD"), Map.of())
+        );
+
+        withEventProcessor(eventProcessorBuilder, (eventProcessor) -> {
+            final Collection<Event> outputEvents = eventProcessor.processEvents(inputEvents);
+            assertThat("event count is unchanged", outputEvents, hasSize(inputEvents.size()));
+
+            validateEvent(outputEvents, eventWithId("baseline"), (event) -> {
+                assertAll(String.format("EVENT(data: %s; meta: %s)", event.getData(), event.getMetadata()), () -> {
+                    assertThat(event, not(isTagged("_ingest_pipeline_failure")));
+                    assertThat(event, includesField("[lower]").withValue(equalTo("mixed")));
+                    assertThat(event, includesField("[mixed]").withValue(equalTo("MiXeD")));
+                });
+            });
+        });
+    }
+
     @Test void testReroutePipelinesMutatingEvents() {
         final List<Event> matchedEvents = new ArrayList<>();
 

src/test/java/co/elastic/logstash/filters/elasticintegration/SmokeTest.java

@@ -0,0 +1,11 @@
+{
+  "description": "An pipeline with a script processor referencing ingest common utils",
+  "processors": [
+    {
+      "script": {
+        "lang": "painless",
+        "source": "ctx.lower = Processors.lowercase(ctx.mixed)"
+      }
+    }
+  ]
+}