Merge branch 'main' of github.com:elastic/logstash-filter-elastic_integration into add_eav_header

kaisecheng · kaisecheng · commit 8093e810e52c · 2023-09-27T19:46:08.000+01:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,4 +1,6 @@
 ## 0.0.3 (UNRELEASED)
+  - Re-syncs with Elasticsearch 8.10 [#78](http://github.com/elastic/logstash-filter-elastic_integration/pull/78)
+    - BREAKING: The `redact` processor was removed from upstream IngestCommon, and therefore no longer available here.
   - Documentation added for required privileges and unsupported processors [#72](http://github.com/elastic/logstash-filter-elastic_integration/pull/72)
   - Added request header `Elastic-Api-Version` for serverless [#84](http://github.com/elastic/logstash-filter-elastic_integration/pull/84)
 
diff --git a/build.gradle b/build.gradle
@@ -235,15 +235,37 @@ task shadeElasticsearchIngestGeoIpModule(type: com.github.jengelman.gradle.plugi
     relocate('com.fasterxml.jackson', "${shadeNamespace}.com.fasterxml.jackson")
     relocate('com.maxmind',           "${shadeNamespace}.com.maxmind")
 
-    exclude 'module-info.class'
-    exclude 'META-INF/versions/9/module-info.class'
+    exclude '**/module-info.class'
+}
+
+task shadeElasticsearchGrokImplementation(type: com.github.jengelman.gradle.plugins.shadow.tasks.ShadowJar) {
+    description "Shades embedded dependencies of the Elasticsearch Grok implementation"
+
+    dependsOn buildElasticsearchLocalDistro
+
+    from(buildElasticsearchLocalDistro.lib.orElse(objects.fileCollection())) {
+        include jarPackageNamed("elasticsearch-grok")
+        include jarPackageNamed("joni")
+        include jarPackageNamed("jcodings")
+    }
+
+    archiveFileName = "elasticsearch-grok-uber.jar"
+    destinationDirectory = file("${buildDir}/shaded")
+
+    mergeServiceFiles()
+    String shadeNamespace = "org.elasticsearch.grok.shaded"
+    relocate('org.joni', "${shadeNamespace}.org.joni")
+    relocate('org.jcodings', "${shadeNamespace}.org.jcodings")
+
+    exclude '**/module-info.class'
 }
 
 task importMinimalElasticsearch(dependsOn: buildElasticsearchLocalDistro) {
     description "Imports minimal portions of Elasticsearch localDistro"
 
     dependsOn buildElasticsearchLocalDistro
     dependsOn shadeElasticsearchIngestGeoIpModule
+    dependsOn shadeElasticsearchGrokImplementation
 
     ext.jars = "${buildDir}/elasticsearch-minimal-jars"
 
@@ -261,10 +283,10 @@ task importMinimalElasticsearch(dependsOn: buildElasticsearchLocalDistro) {
                 include jarPackageNamed("lucene-core")
                 include jarPackageNamed("lucene-analysis-common")
             }
+            from(shadeElasticsearchGrokImplementation)
 
             from(buildElasticsearchLocalDistro.module("ingest-common")) {
                 include jarPackageNamed("ingest-common")
-                include jarPackageNamed("elasticsearch-grok")
                 include jarPackageNamed("elasticsearch-dissect")
             }
 
@@ -306,6 +328,9 @@ tasks.withType(JavaCompile) {
 
 test {
     useJUnitPlatform()
+    testLogging {
+        exceptionFormat "full"
+    }
 }
 
 task generateTestCertificates(type: Exec) {
diff --git a/docs/index.asciidoc b/docs/index.asciidoc
@@ -166,7 +166,6 @@ It has access to the Painless and Mustache scripting engines where applicable:
 | `lowercase` | _none_
 | `networkdirection` | _none_
 | `pipeline` | resolved pipeline _must_ be wholly-composed of supported processors
-| `redact` | _none_
 | `registereddomain` | _none_
 | `remove` | _none_
 | `rename` | _none_
diff --git a/gradle.properties b/gradle.properties
@@ -1,2 +1,2 @@
 LOGSTASH_PATH=../../ls
-ELASTICSEARCH_TREEISH=8.7
+ELASTICSEARCH_TREEISH=8.10
diff --git a/src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java b/src/main/java/co/elastic/logstash/filters/elasticintegration/EventProcessorBuilder.java
@@ -22,6 +22,7 @@
 import org.elasticsearch.core.IOUtils;
 import org.elasticsearch.core.TimeValue;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.ingest.IngestService;
 import org.elasticsearch.ingest.Processor;
 import org.elasticsearch.ingest.common.IngestCommonPlugin;
 import org.elasticsearch.ingest.useragent.IngestUserAgentPlugin;
@@ -46,10 +47,12 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import static co.elastic.logstash.filters.elasticintegration.ingest.SafeSubsetIngestPlugin.safeSubset;
 import static com.google.common.util.concurrent.AbstractScheduledService.Scheduler.newFixedRateSchedule;
 
 @SuppressWarnings("UnusedReturnValue")
@@ -78,7 +81,41 @@ public static EventProcessorBuilder fromElasticsearch(final RestClient elasticse
     }
 
     public EventProcessorBuilder() {
-        this.addProcessorsFromPlugin(IngestCommonPlugin::new);
+        this.addProcessorsFromPlugin(IngestCommonPlugin::new, Set.of(
+                org.elasticsearch.ingest.common.AppendProcessor.TYPE,
+                org.elasticsearch.ingest.common.BytesProcessor.TYPE,
+                org.elasticsearch.ingest.common.CommunityIdProcessor.TYPE,
+                org.elasticsearch.ingest.common.ConvertProcessor.TYPE,
+                org.elasticsearch.ingest.common.CsvProcessor.TYPE,
+                org.elasticsearch.ingest.common.DateIndexNameProcessor.TYPE,
+                org.elasticsearch.ingest.common.DateProcessor.TYPE,
+                org.elasticsearch.ingest.common.DissectProcessor.TYPE,
+                "dot_expander", // note: upstream constant is package-private
+                org.elasticsearch.ingest.DropProcessor.TYPE, // note: not in ingest-common
+                org.elasticsearch.ingest.common.FailProcessor.TYPE,
+                org.elasticsearch.ingest.common.FingerprintProcessor.TYPE,
+                org.elasticsearch.ingest.common.ForEachProcessor.TYPE,
+                org.elasticsearch.ingest.common.GrokProcessor.TYPE,
+                org.elasticsearch.ingest.common.GsubProcessor.TYPE,
+                org.elasticsearch.ingest.common.HtmlStripProcessor.TYPE,
+                org.elasticsearch.ingest.common.JoinProcessor.TYPE,
+                org.elasticsearch.ingest.common.JsonProcessor.TYPE,
+                org.elasticsearch.ingest.common.KeyValueProcessor.TYPE,
+                org.elasticsearch.ingest.common.LowercaseProcessor.TYPE,
+                org.elasticsearch.ingest.common.NetworkDirectionProcessor.TYPE,
+                // note: no `pipeline` processor, as we provide our own
+                org.elasticsearch.ingest.common.RegisteredDomainProcessor.TYPE,
+                org.elasticsearch.ingest.common.RemoveProcessor.TYPE,
+                org.elasticsearch.ingest.common.RenameProcessor.TYPE,
+                org.elasticsearch.ingest.common.RerouteProcessor.TYPE,
+                org.elasticsearch.ingest.common.ScriptProcessor.TYPE,
+                org.elasticsearch.ingest.common.SetProcessor.TYPE,
+                org.elasticsearch.ingest.common.SortProcessor.TYPE,
+                org.elasticsearch.ingest.common.SplitProcessor.TYPE,
+                org.elasticsearch.ingest.common.TrimProcessor.TYPE,
+                org.elasticsearch.ingest.common.URLDecodeProcessor.TYPE,
+                org.elasticsearch.ingest.common.UppercaseProcessor.TYPE,
+                org.elasticsearch.ingest.common.UriPartsProcessor.TYPE));
         this.addProcessorsFromPlugin(IngestUserAgentPlugin::new);
         this.addProcessor(SetSecurityUserProcessor.TYPE, SetSecurityUserProcessor.Factory::new);
     }
@@ -133,6 +170,10 @@ public EventProcessorBuilder addProcessor(final String type, final Supplier<Proc
         return this.addProcessorsFromPlugin(SingleProcessorIngestPlugin.of(type, processorFactorySupplier));
     }
 
+    public EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPlugin> pluginSupplier, Set<String> requiredProcessors) {
+        return this.addProcessorsFromPlugin(safeSubset(pluginSupplier, requiredProcessors));
+    }
+
     public synchronized EventProcessorBuilder addProcessorsFromPlugin(Supplier<IngestPlugin> pluginSupplier) {
         this.ingestPlugins.add(pluginSupplier);
         return this;
@@ -158,16 +199,18 @@ synchronized EventProcessor build(final PluginContext pluginContext) {
             final ScriptService scriptService = initScriptService(settings, threadPool);
             resourcesToClose.add(scriptService);
 
+            final Environment env = new Environment(settings, null);
             final Processor.Parameters processorParameters = new Processor.Parameters(
-                    new Environment(settings, null),
+                    env,
                     scriptService,
                     null,
                     threadPool.getThreadContext(),
                     threadPool::relativeTimeInMillis,
                     (delay, command) -> threadPool.schedule(command, TimeValue.timeValueMillis(delay), ThreadPool.Names.GENERIC),
                     null,
                     null,
-                    threadPool.generic()::execute
+                    threadPool.generic()::execute,
+                    IngestService.createGrokThreadWatchdog(env, threadPool)
             );
 
             IngestPipelineFactory ingestPipelineFactory = new IngestPipelineFactory(scriptService);
diff --git a/src/main/java/co/elastic/logstash/filters/elasticintegration/ingest/SafeSubsetIngestPlugin.java b/src/main/java/co/elastic/logstash/filters/elasticintegration/ingest/SafeSubsetIngestPlugin.java
@@ -0,0 +1,64 @@
+package co.elastic.logstash.filters.elasticintegration.ingest;
+
+import org.elasticsearch.core.IOUtils;
+import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.plugins.IngestPlugin;
+
+import javax.annotation.Nonnull;
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.function.Supplier;
+
+public class SafeSubsetIngestPlugin implements IngestPlugin, Closeable {
+    private final IngestPlugin ingestPlugin;
+    private final Set<String> requiredProcessors;
+
+    public static Supplier<IngestPlugin> safeSubset(final @Nonnull Supplier<IngestPlugin> ingestPluginSupplier,
+                                                    final @Nonnull Set<String> requiredProcessors) {
+        return () -> new SafeSubsetIngestPlugin(ingestPluginSupplier, requiredProcessors);
+    }
+
+    private SafeSubsetIngestPlugin(final @Nonnull Supplier<IngestPlugin> ingestPluginSupplier,
+                                   final @Nonnull Set<String> requiredProcessors) {
+        try {
+            this.ingestPlugin = Objects.requireNonNull(ingestPluginSupplier.get(), "an IngestPlugin must be supplied!");
+            this.requiredProcessors = Set.copyOf(requiredProcessors);
+        } catch (Exception e) {
+            IOUtils.closeWhileHandlingException(this);
+            throw e;
+        }
+    }
+
+    @Override
+    public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
+        final Map<String, Processor.Factory> providedProcessors = this.ingestPlugin.getProcessors(parameters);
+
+        final Map<String, Processor.Factory> acceptedProcessors = new HashMap<>();
+        final Set<String> missingProcessors = new HashSet<>();
+
+        for (String requiredProcessor : this.requiredProcessors) {
+            final Processor.Factory processor = providedProcessors.get(requiredProcessor);
+            if (!Objects.nonNull(processor)) {
+                missingProcessors.add(requiredProcessor);
+            } else {
+                acceptedProcessors.put(requiredProcessor, processor);
+            }
+        }
+        if (!missingProcessors.isEmpty()) {
+            throw new IllegalStateException(String.format("Expected IngestPlugin %s to provide processors %s, but they were not provided", this.ingestPlugin, missingProcessors));
+        }
+        return Map.copyOf(acceptedProcessors);
+    }
+
+    @Override
+    public void close() throws IOException {
+        if (ingestPlugin instanceof Closeable) {
+            ((Closeable) ingestPlugin).close();
+        }
+    }
+}
diff --git a/src/test/resources/co/elastic/logstash/filters/elasticintegration/ssl-test-certs/generate.sh b/src/test/resources/co/elastic/logstash/filters/elasticintegration/ssl-test-certs/generate.sh
@@ -45,9 +45,9 @@ openssl verify -CAfile root.crt client_no_matching_subject.crt
 openssl verify -CAfile client_self_signed.crt client_self_signed.crt
 
 # create encrypted pkcs8 versions of all keys
-openssl pkcs8 -topk8 -inform PEM -outform PEM -passout "pass:12345678" -in client_from_root.key -out client_from_root.key.pkcs8
-openssl pkcs8 -topk8 -inform PEM -outform PEM -passout "pass:12345678" -in server_from_root.key -out server_from_root.key.pkcs8
-openssl pkcs8 -topk8 -inform PEM -outform PEM -passout "pass:12345678" -in client_no_matching_subject.key -out client_no_matching_subject.key.pkcs8
+openssl pkcs8 -topk8 -v1 PBE-SHA1-3DES -inform PEM -outform PEM -passout "pass:12345678" -in client_from_root.key -out client_from_root.key.pkcs8
+openssl pkcs8 -topk8 -v1 PBE-SHA1-3DES -inform PEM -outform PEM -passout "pass:12345678" -in server_from_root.key -out server_from_root.key.pkcs8
+openssl pkcs8 -topk8 -v1 PBE-SHA1-3DES -inform PEM -outform PEM -passout "pass:12345678" -in client_no_matching_subject.key -out client_no_matching_subject.key.pkcs8
 
 # create pkcs12 keystores (pass:12345678)
 openssl pkcs12 -export -in server_from_root.crt -inkey server_from_root.key -out server_from_root.p12 -name "server_from_root" -passout 'pass:12345678'

Original file line number	Diff line number	Diff line change
`@@ -1,2 +1,2 @@`
`1`	`1`	`LOGSTASH_PATH=../../ls`
`2`		`-ELASTICSEARCH_TREEISH=8.7`
	`2`	`+ELASTICSEARCH_TREEISH=8.10`