Support env variable in condition

logstash-core/spec/logstash/compiler/compiler_spec.rb

@@ -52,6 +52,7 @@ def rand_meta
   describe "compiling imperative" do
     let(:source_id) { "fake_sourcefile" }
     let(:source_with_metadata) { org.logstash.common.SourceWithMetadata.new(source_protocol, source_id, 0, 0, source) }
+    let(:cve) { org.logstash.plugins.ConfigVariableExpander.without_secret(org.logstash.common.EnvironmentVariableProvider.default_provider()) }
     subject(:compiled) { described_class.compile_imperative(source_with_metadata, settings.get_value("config.support_escapes")) }
 
     context "when config.support_escapes" do
@@ -70,7 +71,7 @@ def rand_meta
       }
 
       let(:compiled_string) do
-        compiled[:input].toGraph.vertices.toArray.first.getPluginDefinition.arguments["bar"]
+        compiled[:input].toGraph(cve).vertices.toArray.first.getPluginDefinition.arguments["bar"]
       end
 
       before do

logstash-core/src/main/java/org/logstash/config/ir/CompiledPipeline.java

@@ -19,12 +19,10 @@
 
 package org.logstash.config.ir;
 
-import co.elastic.logstash.api.Codec;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.jruby.RubyArray;
 import org.jruby.RubyHash;
-import org.jruby.javasupport.JavaUtil;
 import org.jruby.runtime.builtin.IRubyObject;
 import org.logstash.RubyUtil;
 import org.logstash.Rubyfier;
@@ -38,6 +36,7 @@
 import org.logstash.config.ir.compiler.EventCondition;
 import org.logstash.config.ir.compiler.RubyIntegration;
 import org.logstash.config.ir.compiler.SplitDataset;
+import org.logstash.config.ir.expression.*;
 import org.logstash.config.ir.graph.SeparatorVertex;
 import org.logstash.config.ir.graph.IfVertex;
 import org.logstash.config.ir.graph.PluginVertex;
@@ -48,6 +47,8 @@
 import org.logstash.plugins.ConfigVariableExpander;
 import org.logstash.secret.store.SecretStore;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.util.*;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
@@ -207,7 +208,6 @@ private Collection<IRubyObject> setupInputs(ConfigVariableExpander cve) {
         return nodes;
     }
 
-
     final RubyHash convertArgs(final Map<String, Object> input) {
         final RubyHash converted = RubyHash.newHash(RubyUtil.RUBY);
         for (final Map.Entry<String, Object> entry : input.entrySet()) {
@@ -243,7 +243,7 @@ private Map<String, Object> expandArguments(final PluginDefinition pluginDefinit
     }
 
     @SuppressWarnings({"rawtypes", "unchecked"})
-    private Map<String, Object> expandConfigVariables(ConfigVariableExpander cve, Map<String, Object> configArgs) {
+    public static Map<String, Object> expandConfigVariables(ConfigVariableExpander cve, Map<String, Object> configArgs) {
         Map<String, Object> expandedConfig = new HashMap<>();
         for (Map.Entry<String, Object> e : configArgs.entrySet()) {
             if (e.getValue() instanceof List) {

logstash-core/src/main/java/org/logstash/config/ir/ConfigCompiler.java

@@ -20,14 +20,14 @@
 
 package org.logstash.config.ir;
 
-import org.jruby.RubyArray;
 import org.jruby.RubyHash;
 import org.jruby.javasupport.JavaUtil;
 import org.jruby.runtime.builtin.IRubyObject;
 import org.logstash.RubyUtil;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.graph.Graph;
 import org.logstash.config.ir.imperative.Statement;
+import org.logstash.plugins.ConfigVariableExpander;
 
 import java.util.List;
 import java.util.Map;
@@ -49,18 +49,19 @@ private ConfigCompiler() {
     /**
      * @param sourcesWithMetadata Logstash Config partitioned
      * @param supportEscapes The value of the setting {@code config.support_escapes}
+     * @param cve Config variable expander. Substitute variable with value in secret store, env, default config value
      * @return Compiled {@link PipelineIR}
      * @throws InvalidIRException if the the configuration contains errors
      */
     @SuppressWarnings("unchecked")
     public static PipelineIR configToPipelineIR(final List<SourceWithMetadata> sourcesWithMetadata,
-                                                final boolean supportEscapes) throws InvalidIRException {
-        return compileSources(sourcesWithMetadata, supportEscapes);
+                                                final boolean supportEscapes, ConfigVariableExpander cve) throws InvalidIRException {
+        return compileSources(sourcesWithMetadata, supportEscapes, cve);
     }
 
-    public static PipelineIR compileSources(List<SourceWithMetadata> sourcesWithMetadata, boolean supportEscapes) throws InvalidIRException {
+    public static PipelineIR compileSources(List<SourceWithMetadata> sourcesWithMetadata, boolean supportEscapes, ConfigVariableExpander cve) throws InvalidIRException {
         Map<PluginDefinition.Type, List<Graph>> groupedPipelineSections = sourcesWithMetadata.stream()
-                .map(swm -> compileGraph(swm, supportEscapes))
+                .map(swm -> compileGraph(swm, supportEscapes, cve))
                 .flatMap(m -> m.entrySet().stream())
                 .filter(e -> e.getValue() != null)
                 .collect(groupingBy(Map.Entry::getKey,
@@ -108,15 +109,15 @@ private static Statement readStatementFromRubyHash(RubyHash hash, String key) {
         return inputValue.toJava(Statement.class);
     }
 
-    private static Map<PluginDefinition.Type, Graph> compileGraph(SourceWithMetadata swm, boolean supportEscapes) {
+    private static Map<PluginDefinition.Type, Graph> compileGraph(SourceWithMetadata swm, boolean supportEscapes, ConfigVariableExpander cve) {
         Map<PluginDefinition.Type, Statement> pluginStatements = compileImperative(swm, supportEscapes);
         return pluginStatements.entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, e -> toGraphWithUntypedException(e.getValue())));
+                .collect(Collectors.toMap(Map.Entry::getKey, e -> toGraphWithUntypedException(e.getValue(), cve)));
     }
 
-    private static Graph toGraphWithUntypedException(Statement s) {
+    private static Graph toGraphWithUntypedException(Statement s, ConfigVariableExpander cve) {
         try {
-            return s.toGraph();
+            return s.toGraph(cve);
         } catch (InvalidIRException iirex) {
             throw new IllegalArgumentException(iirex);
         }

logstash-core/src/main/java/org/logstash/config/ir/PipelineIR.java

@@ -24,11 +24,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.logstash.common.Util;
-import org.logstash.config.ir.graph.Graph;
-import org.logstash.config.ir.graph.PluginVertex;
-import org.logstash.config.ir.graph.QueueVertex;
-import org.logstash.config.ir.graph.Vertex;
-import org.logstash.config.ir.graph.SeparatorVertex;
+import org.logstash.config.ir.graph.*;
 
 public final class PipelineIR implements Hashable {
 

logstash-core/src/main/java/org/logstash/config/ir/expression/ExpressionSubstitution.java

@@ -0,0 +1,42 @@
+package org.logstash.config.ir.expression;
+
+import org.logstash.common.SourceWithMetadata;
+import org.logstash.config.ir.CompiledPipeline;
+import org.logstash.config.ir.InvalidIRException;
+import org.logstash.plugins.ConfigVariableExpander;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+import java.util.Map;
+
+public class ExpressionSubstitution {
+    public static Expression substituteBoolExpression(ConfigVariableExpander cve, Expression expression) {
+        try {
+            if (expression instanceof BinaryBooleanExpression) {
+                BinaryBooleanExpression binaryBoolExp = (BinaryBooleanExpression) expression;
+                Expression substitutedLeftExp = substituteBoolExpression(cve, binaryBoolExp.getLeft());
+                Expression substitutedRightExp = substituteBoolExpression(cve, binaryBoolExp.getRight());
+                if (substitutedLeftExp != binaryBoolExp.getLeft() || substitutedRightExp != binaryBoolExp.getRight()) {
+                    Constructor<? extends BinaryBooleanExpression> constructor = binaryBoolExp.getClass().getConstructor(SourceWithMetadata.class, Expression.class, Expression.class);
+                    return constructor.newInstance(binaryBoolExp.getSourceWithMetadata(), substitutedLeftExp, substitutedRightExp);
+                }
+            } else if (expression instanceof UnaryBooleanExpression) {
+                UnaryBooleanExpression unaryBoolExp = (UnaryBooleanExpression) expression;
+                Expression substitutedExp = substituteBoolExpression(cve, unaryBoolExp.getExpression());
+                if (substitutedExp != unaryBoolExp.getExpression()) {
+                    Constructor<? extends UnaryBooleanExpression> constructor = unaryBoolExp.getClass().getConstructor(SourceWithMetadata.class, Expression.class);
+                    return constructor.newInstance(unaryBoolExp.getSourceWithMetadata(), substitutedExp);
+                }
+            } else if (expression instanceof ValueExpression && !(expression instanceof RegexValueExpression) && (((ValueExpression) expression).get() != null)) {
+                String key = "placeholder";
+                Map<String, Object> args = Map.of(key, ((ValueExpression) expression).get());
+                Map<String, Object> substitutedArgs = CompiledPipeline.expandConfigVariables(cve, args);
+                return new ValueExpression(expression.getSourceWithMetadata(), substitutedArgs.get(key));
+            }
+
+            return expression;
+        } catch (NoSuchMethodException | InstantiationException | IllegalAccessException | InvocationTargetException | InvalidIRException e) {
+            throw new IllegalStateException("Unable to instantiate substituted condition expression", e);
+        }
+    }
+}

logstash-core/src/main/java/org/logstash/config/ir/imperative/ComposedParallelStatement.java

@@ -23,6 +23,7 @@
 import org.logstash.config.ir.InvalidIRException;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.graph.Graph;
+import org.logstash.plugins.ConfigVariableExpander;
 
 import java.util.List;
 
@@ -37,11 +38,11 @@ protected String composeTypeString() {
     }
 
     @Override
-    public Graph toGraph() throws InvalidIRException {
+    public Graph toGraph(ConfigVariableExpander cve) throws InvalidIRException {
         Graph g = Graph.empty();
 
         for (Statement s : getStatements()) {
-            g = Graph.combine(g, s.toGraph()).graph;
+            g = Graph.combine(g, s.toGraph(cve)).graph;
         }
 
         return g;

logstash-core/src/main/java/org/logstash/config/ir/imperative/ComposedSequenceStatement.java

@@ -23,6 +23,7 @@
 import org.logstash.config.ir.InvalidIRException;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.graph.Graph;
+import org.logstash.plugins.ConfigVariableExpander;
 
 import java.util.List;
 
@@ -37,11 +38,11 @@ protected String composeTypeString() {
     }
 
     @Override
-    public Graph toGraph() throws InvalidIRException {
+    public Graph toGraph(ConfigVariableExpander cve) throws InvalidIRException {
         Graph g = Graph.empty();
 
         for (Statement statement : getStatements()) {
-            Graph sg = statement.toGraph();
+            Graph sg = statement.toGraph(cve);
             g = g.chain(sg);
         }
 

logstash-core/src/main/java/org/logstash/config/ir/imperative/IfStatement.java

@@ -23,11 +23,12 @@
 import org.logstash.config.ir.SourceComponent;
 import org.logstash.config.ir.InvalidIRException;
 import org.logstash.common.SourceWithMetadata;
-import org.logstash.config.ir.expression.BooleanExpression;
+import org.logstash.config.ir.expression.*;
 import org.logstash.config.ir.graph.BooleanEdge;
 import org.logstash.config.ir.graph.Graph;
 import org.logstash.config.ir.graph.IfVertex;
 import org.logstash.config.ir.graph.Vertex;
+import org.logstash.plugins.ConfigVariableExpander;
 
 import java.util.Collection;
 import java.util.stream.Collectors;
@@ -99,9 +100,9 @@ public String toString(int indent) {
 
 
     @Override
-    public Graph toGraph() throws InvalidIRException {
-        Graph trueGraph = getTrueStatement().toGraph();
-        Graph falseGraph = getFalseStatement().toGraph();
+    public Graph toGraph(ConfigVariableExpander cve) throws InvalidIRException {
+        Graph trueGraph = getTrueStatement().toGraph(cve);
+        Graph falseGraph = getFalseStatement().toGraph(cve);
 
         // If there is nothing in the true or false sections of this if statement,
         // we can omit the if statement altogether!
@@ -114,7 +115,8 @@ public Graph toGraph() throws InvalidIRException {
         Collection<Vertex> trueRoots = trueGraph.roots().map(combination.oldToNewVertices::get).collect(Collectors.toList());
         Collection<Vertex> falseRoots = falseGraph.roots().map(combination.oldToNewVertices::get).collect(Collectors.toList());
 
-        IfVertex ifVertex = new IfVertex(this.getSourceWithMetadata(), this.booleanExpression);
+        IfVertex ifVertex = new IfVertex(this.getSourceWithMetadata(),
+                (BooleanExpression) ExpressionSubstitution.substituteBoolExpression(cve, this.booleanExpression));
         newGraph.addVertex(ifVertex);
 
         for (Vertex v : trueRoots) {

logstash-core/src/main/java/org/logstash/config/ir/imperative/NoopStatement.java

@@ -23,6 +23,7 @@
 import org.logstash.config.ir.SourceComponent;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.graph.Graph;
+import org.logstash.plugins.ConfigVariableExpander;
 
 public class NoopStatement extends Statement {
 
@@ -43,7 +44,7 @@ public String toString(int indent) {
     }
 
     @Override
-    public Graph toGraph() {
+    public Graph toGraph(ConfigVariableExpander cve) {
         return Graph.empty();
     }
 

logstash-core/src/main/java/org/logstash/config/ir/imperative/PluginStatement.java

@@ -27,6 +27,7 @@
 import org.logstash.config.ir.graph.Graph;
 import org.logstash.config.ir.graph.PluginVertex;
 import org.logstash.config.ir.graph.Vertex;
+import org.logstash.plugins.ConfigVariableExpander;
 
 public class PluginStatement extends Statement {
     private final PluginDefinition pluginDefinition;
@@ -53,7 +54,7 @@ public String toString(int indent) {
     }
 
     @Override
-    public Graph toGraph() throws InvalidIRException {
+    public Graph toGraph(ConfigVariableExpander cve) throws InvalidIRException {
         Vertex pluginVertex = new PluginVertex(getSourceWithMetadata(), pluginDefinition);
         Graph g = Graph.empty();
         g.addVertex(pluginVertex);

logstash-core/src/main/java/org/logstash/config/ir/imperative/Statement.java

@@ -24,13 +24,14 @@
 import org.logstash.config.ir.BaseSourceComponent;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.graph.Graph;
+import org.logstash.plugins.ConfigVariableExpander;
 
 public abstract class Statement extends BaseSourceComponent {
     public Statement(SourceWithMetadata meta) {
         super(meta);
     }
 
-    public abstract Graph toGraph() throws InvalidIRException;
+    public abstract Graph toGraph(ConfigVariableExpander cve) throws InvalidIRException;
 
     public String toString() {
         return toString(2);

logstash-core/src/main/java/org/logstash/execution/AbstractPipelineExt.java

@@ -51,6 +51,7 @@
 import org.logstash.ackedqueue.ext.JRubyAckedQueueExt;
 import org.logstash.ackedqueue.ext.JRubyWrappedAckedQueueExt;
 import org.logstash.common.DeadLetterQueueFactory;
+import org.logstash.common.EnvironmentVariableProvider;
 import org.logstash.common.SourceWithMetadata;
 import org.logstash.config.ir.ConfigCompiler;
 import org.logstash.config.ir.InvalidIRException;
@@ -62,6 +63,7 @@
 import org.logstash.instrument.metrics.AbstractNamespacedMetricExt;
 import org.logstash.instrument.metrics.MetricKeys;
 import org.logstash.instrument.metrics.NullMetricExt;
+import org.logstash.plugins.ConfigVariableExpander;
 import org.logstash.secret.store.SecretStore;
 import org.logstash.secret.store.SecretStoreExt;
 
@@ -183,8 +185,8 @@ public final AbstractPipelineExt initialize(final ThreadContext context,
             }
         }
         boolean supportEscapes = getSetting(context, "config.support_escapes").isTrue();
-        try {
-            lir = ConfigCompiler.configToPipelineIR(configParts, supportEscapes);
+        try (ConfigVariableExpander cve = new ConfigVariableExpander(getSecretStore(context), EnvironmentVariableProvider.defaultProvider())) {
+            lir = ConfigCompiler.configToPipelineIR(configParts, supportEscapes, cve);
         } catch (InvalidIRException iirex) {
             throw new IllegalArgumentException(iirex);
         }