Pick backend according to query type

graylog-storage-opensearch2/src/main/java/org/graylog/storage/opensearch2/ViewsOSBackendModule.java

@@ -25,8 +25,6 @@
 import com.google.inject.multibindings.OptionalBinder;
 import org.graylog.plugins.views.ViewsModule;
 import org.graylog.plugins.views.search.SearchType;
-import org.graylog.plugins.views.search.engine.GeneratedQueryContext;
-import org.graylog.plugins.views.search.engine.QueryBackend;
 import org.graylog.plugins.views.search.export.ExportBackend;
 import org.graylog.plugins.views.search.searchtypes.MessageList;
 import org.graylog.plugins.views.search.searchtypes.events.EventList;
@@ -89,8 +87,9 @@ public ViewsOSBackendModule(SearchVersion supportedSearchVersion) {
     protected void configure() {
         install(new FactoryModuleBuilder().build(OSGeneratedQueryContext.Factory.class));
 
-        bindForVersion(supportedSearchVersion, new TypeLiteral<QueryBackend<? extends GeneratedQueryContext>>() {})
-                .to(OpenSearchBackend.class);
+        registerVersionedQueryBackend(supportedSearchVersion, "elasticsearch", OpenSearchBackend.class);
+//        bindForVersion(supportedSearchVersion, new TypeLiteral<QueryBackend<? extends GeneratedQueryContext>>() {})
+//                .to(OpenSearchBackend.class);
 
         registerOSSearchTypeHandler(MessageList.NAME, OSMessageList.class);
         registerOSSearchTypeHandler(EventList.NAME, OSEventListDelegate.class);

graylog2-server/pom.xml

@@ -349,6 +349,10 @@
             <groupId>joda-time</groupId>
             <artifactId>joda-time</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.threeten</groupId>
+            <artifactId>threeten-extra</artifactId>
+        </dependency>
 
         <dependency>
             <groupId>org.graylog2.repackaged</groupId>

graylog2-server/src/main/java/org/graylog/plugins/views/ViewsModule.java

@@ -85,15 +85,16 @@ protected void registerPivotAggregationFunction(String name, String description,
         seriesSpecBinder().addBinding(name).toInstance(SeriesDescription.create(name, description));
     }
 
-    protected MapBinder<String, QueryBackend<? extends GeneratedQueryContext>> queryBackendBinder(SearchVersion version) {
+    protected MapBinder<String, QueryBackend<? extends GeneratedQueryContext>> queryBackendBinder() {
         return MapBinder.newMapBinder(binder(),
                 TypeLiteral.get(String.class),
-                new TypeLiteral<QueryBackend<? extends GeneratedQueryContext>>() {});
+                new TypeLiteral<>() {});
 
     }
 
-    protected ScopedBindingBuilder registerQueryBackend(SearchVersion version, String name, Class<? extends QueryBackend<? extends GeneratedQueryContext>> implementation) {
-        return queryBackendBinder(version).addBinding(name).to(implementation);
+    protected ScopedBindingBuilder registerVersionedQueryBackend(SearchVersion version, String name, Class<? extends QueryBackend<? extends GeneratedQueryContext>> implementation) {
+        bindForVersion(version, new TypeLiteral<QueryBackend<? extends GeneratedQueryContext>>() {}).to(implementation);
+        return queryBackendBinder().addBinding(name).to(implementation);
     }
 
     protected void registerESQueryDecorator(Class<? extends QueryStringDecorator> esQueryDecorator) {

graylog2-server/src/main/java/org/graylog/plugins/views/search/QueryResult.java

@@ -63,7 +63,11 @@ public static Builder builder() {
     public abstract Builder toBuilder();
 
     public static QueryResult emptyResult() {
-        return builder().searchTypes(Collections.emptyMap()).query(Query.emptyRoot()).build();
+        return emptyResultForQuery(Query.emptyRoot());
+    }
+
+    public static QueryResult emptyResultForQuery(Query query) {
+        return builder().searchTypes(Collections.emptyMap()).query(query).build();
     }
 
     public static QueryResult failedQueryWithError(Query query, SearchError error) {

graylog2-server/src/main/java/org/graylog/plugins/views/search/engine/QueryBackend.java

@@ -89,7 +89,6 @@ default QueryResult run(SearchJob job, Query query, GeneratedQueryContext genera
         try {
             final Stopwatch stopwatch = Stopwatch.createStarted();
             final QueryExecutionStats.Builder statsBuilder = QueryExecutionStats.builderWithCurrentTime();
-            // https://www.ibm.com/developerworks/java/library/j-jtp04298/index.html#3.0
             //noinspection unchecked
             final QueryResult result = doRun(job, query, (T) generatedQueryContext);
             stopwatch.stop();
@@ -116,7 +115,7 @@ default QueryResult run(SearchJob job, Query query, GeneratedQueryContext genera
      *
      * @param job          currently executing job
      * @param query        the individual query to run from the current job
-     * @param queryContext the generated query by {@link #generate(Query, Set)}
+     * @param queryContext the generated query by {@link #generate(Query, Set, DateTimeZone)}
      * @return the result for the query
      * @throws RuntimeException if the query could not be executed for some reason
      */
@@ -132,7 +131,7 @@ default ExplainResults.QueryExplainResult explain(SearchJob job, Query query, Ge
      *
      * @param job          currently executing job
      * @param query        the individual query to explain from the current job
-     * @param queryContext the generated query by {@link #generate(Query, Set)}
+     * @param queryContext the generated query by {@link #generate(Query, Set, DateTimeZone)}
      * @return the explain result for the query
      */
     default ExplainResults.QueryExplainResult doExplain(SearchJob job, Query query, T queryContext) {

graylog2-server/src/main/java/org/graylog/plugins/views/search/engine/QueryEngine.java

@@ -52,13 +52,13 @@ public class QueryEngine {
 
     // TODO proper thread pool with tunable settings
     private final Executor queryPool = Executors.newFixedThreadPool(4, new ThreadFactoryBuilder().setNameFormat("query-engine-%d").build());
-    private final QueryBackend<? extends GeneratedQueryContext> backend;
+    private final Map<String, QueryBackend<? extends GeneratedQueryContext>> backends;
 
     @Inject
-    public QueryEngine(QueryBackend<? extends GeneratedQueryContext> backend,
+    public QueryEngine(Map<String, QueryBackend<? extends GeneratedQueryContext>> backends,
                        Set<QueryMetadataDecorator> queryMetadataDecorators,
                        QueryParser queryParser) {
-        this.backend = backend;
+        this.backends = backends;
         this.queryMetadataDecorators = queryMetadataDecorators;
         this.queryParser = queryParser;
     }
@@ -75,6 +75,7 @@ public QueryMetadata parse(Search search, Query query) {
     public ExplainResults explain(SearchJob searchJob, Set<SearchError> validationErrors, DateTimeZone timezone) {
         final Map<String, ExplainResults.QueryExplainResult> queries = searchJob.getSearch().queries().stream()
                 .collect(Collectors.toMap(Query::id, q -> {
+                    final QueryBackend<? extends GeneratedQueryContext> backend = backends.get(q.query().type());
                     final GeneratedQueryContext generatedQueryContext = backend.generate(q, Set.of(), timezone);
 
                     return backend.explain(searchJob, q, generatedQueryContext);
@@ -116,6 +117,7 @@ public SearchJob execute(SearchJob searchJob, Set<SearchError> validationErrors,
     }
 
     private QueryResult prepareAndRun(SearchJob searchJob, Query query, Set<SearchError> validationErrors, DateTimeZone timezone) {
+        final QueryBackend<? extends GeneratedQueryContext> backend = backends.get(query.query().type());
         LOG.debug("[{}] Using {} to generate query", query.id(), backend);
         // with all the results done, we can execute the current query and eventually complete our own result
         // if any of this throws an exception, the handle in #execute will convert it to an error and return a "failed" result instead

graylog2-server/src/main/java/org/graylog/plugins/views/search/engine/normalization/DecorateQueryStringsNormalizer.java

@@ -38,6 +38,10 @@ public DecorateQueryStringsNormalizer(QueryStringDecorators queryStringDecorator
 
     @Override
     public Query normalizeQuery(final Query query, final ParameterProvider parameterProvider) {
+        // TODO this only makes sense for elasticsearch backend queries, find a better way to make this clear in search normalizers
+        if (!(query.query() instanceof ElasticsearchQueryString)) {
+            return query;
+        }
         return query.toBuilder()
                 .query(ElasticsearchQueryString.of(this.queryStringDecorators.decorate(query.query().queryString(), parameterProvider, query)))
                 .filter(normalizeFilter(query.filter(), query, parameterProvider))

graylog2-server/src/main/java/org/graylog2/plugin/indexer/searches/timeranges/TimeRange.java

@@ -22,6 +22,9 @@
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
 import org.graylog.plugins.views.search.timeranges.OffsetRange;
 import org.joda.time.DateTime;
+import org.threeten.extra.Interval;
+
+import java.time.Instant;
 
 @JsonTypeInfo(use = JsonTypeInfo.Id.NAME, include = JsonTypeInfo.As.EXISTING_PROPERTY, property = "type")
 @JsonSubTypes({
@@ -42,4 +45,8 @@ public abstract class TimeRange {
     public abstract DateTime getTo();
 
     public abstract TimeRange withReferenceDate(DateTime now);
+
+    public Interval asInterval() {
+        return Interval.of(Instant.ofEpochMilli(getFrom().toInstant().getMillis()), Instant.ofEpochMilli(getTo().toInstant().getMillis()));
+    }
 }

graylog2-server/src/main/java/org/graylog2/utilities/BoundedPriorityQueue.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.utilities;
+
+import java.util.Comparator;
+import java.util.PriorityQueue;
+
+/**
+ * A priority queue that has a fixed upper size.
+ *
+ * @param <E> entry type
+ */
+public class BoundedPriorityQueue<E> extends PriorityQueue<E> {
+
+    private final int maxSize;
+
+    public BoundedPriorityQueue(int maxSize, Comparator<E> comparator) {
+        super(comparator);
+        this.maxSize = maxSize;
+    }
+
+    public boolean offer(final E e) {
+        if (e == null) {
+            throw new NullPointerException();
+        }
+        if (maxSize <= size()) {
+            final E firstElm = peek();
+            if (comparator().compare(e, firstElm) < 1) {
+                return false;
+            } else {
+                poll();
+            }
+        }
+        return super.offer(e);
+    }
+}

graylog2-server/src/test/java/org/graylog2/utilities/BoundedPriorityQueueTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright (C) 2020 Graylog, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the Server Side Public License, version 1,
+ * as published by MongoDB, Inc.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * Server Side Public License for more details.
+ *
+ * You should have received a copy of the Server Side Public License
+ * along with this program. If not, see
+ * <http://www.mongodb.com/licensing/server-side-public-license>.
+ */
+package org.graylog2.utilities;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Comparator;
+
+class BoundedPriorityQueueTest {
+
+    @Test
+    void add() {
+        // the queue behaves like a standard PriorityQueue, but should never grow past the bound we specify, so this is what we test here
+        // not the rest of the behavior
+        final BoundedPriorityQueue<Integer> queue = new BoundedPriorityQueue<Integer>(10, Comparator.naturalOrder());
+
+        for (int i = 0; i < 42; i++) {
+            queue.add(i);
+        }
+        Assertions.assertEquals(10, queue.size());
+
+    }
+}

pom.xml

@@ -143,6 +143,7 @@
         <jersey.version>3.1.9</jersey.version>
         <jmte.version>7.0.3</jmte.version>
         <joda-time.version>2.13.0</joda-time.version>
+        <threeten.version>1.8.0</threeten.version>
         <jool.version>0.9.15</jool.version>
         <json-org.version>20240303</json-org.version>
         <json-path.version>2.9.0</json-path.version>
@@ -333,6 +334,11 @@
                 <artifactId>joda-time</artifactId>
                 <version>${joda-time.version}</version>
             </dependency>
+            <dependency>
+                <groupId>org.threeten</groupId>
+                <artifactId>threeten-extra</artifactId>
+                <version>${threeten.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.fasterxml.jackson</groupId>
                 <artifactId>jackson-bom</artifactId>