Tolerate unresolvable stream references during content pack install

changelog/unreleased/pr-26581.toml

@@ -0,0 +1,5 @@
+type = "fixed"
+message = "Fix content pack installation failing with \"Missing Stream for widget entity\" when a view references a stream that cannot be resolved. The unresolvable stream reference is now skipped with a warning instead of aborting the whole installation."
+
+issues = ["Graylog2/graylog-plugin-enterprise#12987"]
+pulls = ["26581"]

graylog2-server/src/main/java/org/graylog2/contentpacks/model/entities/QueryEntity.java

@@ -38,6 +38,8 @@
 import org.graylog2.contentpacks.model.entities.references.ValueReference;
 import org.graylog2.plugin.indexer.searches.timeranges.TimeRange;
 import org.graylog2.plugin.streams.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -61,6 +63,8 @@
 @JsonDeserialize(builder = QueryEntity.Builder.class)
 public abstract class QueryEntity implements NativeEntityConverter<Query> {
 
+    private static final Logger LOG = LoggerFactory.getLogger(QueryEntity.class);
+
     @JsonProperty
     public abstract String id();
 
@@ -150,15 +154,26 @@ private Filter shallowMappedFilter(Map<EntityDescriptor, Object> nativeEntities)
                           .map(filter -> {
                               if (filter.type().matches(StreamFilter.NAME)) {
                                   final StreamFilter streamFilter = (StreamFilter) filter;
-                                  final Stream stream = (Stream) resolveStreamEntityObject(streamFilter.streamId(), nativeEntities);
-                                  if (Objects.isNull(stream)) {
-                                      throw new ContentPackException("Could not find matching stream id: " +
-                                              streamFilter.streamId());
+                                  final Object object = resolveStreamEntityObject(streamFilter.streamId(), nativeEntities);
+                                  if (object == null) {
+                                      // Skip a dangling stream reference instead of aborting the whole content pack
+                                      // installation. This mirrors the export side, which also drops unresolvable references.
+                                      LOG.warn("Skipping unresolvable stream reference <{}> in query filter for query <{}> during content pack installation",
+                                              streamFilter.streamId(), id());
+                                      return null;
+                                  } else if (object instanceof final Stream stream) {
+                                      return streamFilter.toBuilder().streamId(stream.getId()).build();
+                                  } else {
+                                      // A non-null, non-Stream value indicates a corrupt entity map rather than a
+                                      // missing reference, so abort the installation like the other view resolvers.
+                                      throw new ContentPackException(
+                                              "Invalid type for stream Stream for query filter: " + object.getClass());
                                   }
-                                  return streamFilter.toBuilder().streamId(stream.getId()).build();
                               }
                               return filter;
-                          }).collect(Collectors.toSet());
+                          })
+                          .filter(Objects::nonNull)
+                          .collect(Collectors.toSet());
                   return optFilter.toGenericBuilder().filters(newFilters).build();
                })
                .orElse(null);

graylog2-server/src/main/java/org/graylog2/contentpacks/model/entities/SearchTypeEntity.java

@@ -29,6 +29,8 @@
 import org.graylog2.contentpacks.exceptions.ContentPackException;
 import org.graylog2.contentpacks.model.entities.references.ValueReference;
 import org.graylog2.plugin.streams.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 import java.util.Collections;
@@ -49,6 +51,8 @@
         defaultImpl = SearchTypeEntity.Fallback.class)
 @JsonAutoDetect
 public interface SearchTypeEntity extends NativeEntityConverter<SearchType> {
+    Logger LOG = LoggerFactory.getLogger(SearchTypeEntity.class);
+
     String TYPE_FIELD = "type";
     String FIELD_SEARCH_FILTERS = "filters";
 
@@ -207,17 +211,29 @@ public SearchType toNativeEntity(Map<String, ValueReference> parameters, Map<Ent
 
     default Set<String> mappedStreams(Map<EntityDescriptor, Object> nativeEntities) {
         return streams().stream()
-                .map(id -> resolveStreamEntityObject(id, nativeEntities))
-                .map(object -> {
-                    if (object == null) {
-                        throw new ContentPackException("Missing Stream for event definition");
-                    } else if (object instanceof Stream) {
-                        Stream stream = (Stream) object;
-                        return stream.getId();
-                    } else {
-                        throw new ContentPackException(
-                                "Invalid type for stream Stream for event definition: " + object.getClass());
-                    }
-                }).collect(Collectors.toSet());
+                .map(streamId -> resolveStreamReference(streamId, nativeEntities))
+                .filter(Optional::isPresent)
+                .map(Optional::get)
+                .collect(Collectors.toSet());
+    }
+
+    /**
+     * Resolves a single stream reference from the search type's {@code streams} set to the native stream id.
+     * <p>
+     * A reference that cannot be resolved (e.g. a dangling reference left over in an exported content pack) is
+     * skipped with a warning rather than aborting the whole content pack installation. This mirrors the export
+     * side, which also silently drops stream references it cannot map.
+     */
+    default Optional<String> resolveStreamReference(String streamId, Map<EntityDescriptor, Object> nativeEntities) {
+        final Object object = resolveStreamEntityObject(streamId, nativeEntities);
+        if (object == null) {
+            LOG.warn("Skipping unresolvable stream reference <{}> for search type <{}> during content pack installation", streamId, id());
+            return Optional.empty();
+        } else if (object instanceof Stream) {
+            return Optional.of(((Stream) object).getId());
+        } else {
+            throw new ContentPackException(
+                    "Invalid type for stream Stream for search type: " + object.getClass());
+        }
     }
 }

graylog2-server/src/main/java/org/graylog2/contentpacks/model/entities/WidgetEntity.java

@@ -51,6 +51,8 @@
 import org.graylog2.contentpacks.model.entities.references.ValueReference;
 import org.graylog2.plugin.indexer.searches.timeranges.TimeRange;
 import org.graylog2.plugin.streams.Stream;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
 import java.util.Collections;
@@ -68,6 +70,8 @@
 @AutoValue
 @JsonDeserialize(builder = WidgetEntity.Builder.class)
 public abstract class WidgetEntity implements NativeEntityConverter<WidgetDTO> {
+    private static final Logger LOG = LoggerFactory.getLogger(WidgetEntity.class);
+
     public static final String FIELD_ID = "id";
     public static final String FIELD_TYPE = "type";
     public static final String FIELD_FILTER = "filter";
@@ -181,17 +185,10 @@ public WidgetDTO toNativeEntity(Map<String, ValueReference> parameters, Map<Enti
                 .filters(filters().stream().map(filter -> filter.toNativeEntity(parameters, nativeEntities)).toList())
                 .id(this.id())
                 .streams(this.streams().stream()
-                        .map(id -> resolveStreamEntityObject(id, nativeEntities))
-                        .map(object -> {
-                            if (object == null) {
-                                throw new ContentPackException("Missing Stream for widget entity");
-                            } else if (object instanceof final Stream stream) {
-                                return stream.getId();
-                            } else {
-                                throw new ContentPackException(
-                                        "Invalid type for stream Stream for event definition: " + object.getClass());
-                            }
-                        }).collect(Collectors.toSet()))
+                        .map(streamId -> resolveStreamReference(streamId, nativeEntities))
+                        .filter(Optional::isPresent)
+                        .map(Optional::get)
+                        .collect(Collectors.toSet()))
                 .streamCategories(this.streamCategories())
                 .type(this.type());
         if (this.query().isPresent()) {
@@ -203,6 +200,27 @@ public WidgetDTO toNativeEntity(Map<String, ValueReference> parameters, Map<Enti
         return widgetBuilder.build();
     }
 
+    /**
+     * Resolves a single stream reference from the widget's {@code streams} set to the native stream id.
+     * <p>
+     * A reference that cannot be resolved (e.g. a dangling reference left over in an exported content pack) is
+     * skipped with a warning rather than aborting the whole content pack installation. This mirrors the export
+     * side ({@link org.graylog.plugins.views.search.views.WidgetDTO#toContentPackEntity}), which also silently
+     * drops stream references it cannot map.
+     */
+    private Optional<String> resolveStreamReference(String streamId, Map<EntityDescriptor, Object> nativeEntities) {
+        final Object object = resolveStreamEntityObject(streamId, nativeEntities);
+        if (object == null) {
+            LOG.warn("Skipping unresolvable stream reference <{}> for widget <{}> during content pack installation", streamId, id());
+            return Optional.empty();
+        } else if (object instanceof final Stream stream) {
+            return Optional.of(stream.getId());
+        } else {
+            throw new ContentPackException(
+                    "Invalid type for stream Stream for widget entity: " + object.getClass());
+        }
+    }
+
     @Override
     public void resolveForInstallation(EntityV1 entity,
                                        Map<String, ValueReference> parameters,

graylog2-server/src/test/java/org/graylog2/contentpacks/model/entities/QueryEntityTest.java

@@ -17,19 +17,28 @@
 package org.graylog2.contentpacks.model.entities;
 
 import com.google.common.collect.ImmutableList;
+import org.graylog.plugins.views.search.Filter;
 import org.graylog.plugins.views.search.elasticsearch.ElasticsearchQueryString;
+import org.graylog.plugins.views.search.filter.OrFilter;
+import org.graylog.plugins.views.search.filter.StreamFilter;
 import org.graylog.plugins.views.search.searchfilters.model.DBSearchFilter;
 import org.graylog.plugins.views.search.searchfilters.model.InlineQueryStringSearchFilter;
 import org.graylog.plugins.views.search.searchfilters.model.ReferencedQueryStringSearchFilter;
 import org.graylog.plugins.views.search.searchfilters.model.UsedSearchFilter;
+import org.graylog2.contentpacks.exceptions.ContentPackException;
 import org.graylog2.contentpacks.model.ModelTypes;
 import org.graylog2.plugin.indexer.searches.timeranges.RelativeRange;
+import org.graylog2.plugin.streams.Stream;
 import org.junit.jupiter.api.Test;
 
 import java.util.Collections;
 import java.util.Map;
+import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 class QueryEntityTest {
 
@@ -76,6 +85,61 @@ public void testLoadsSearchFiltersCollectionFromContentPack() {
                 .isEqualTo(expectedSearchFilters);
     }
 
+    @Test
+    public void dropsUnresolvableStreamFilterInsteadOfFailing() {
+        final QueryEntity query = QueryEntity.Builder
+                .createWithDefaults()
+                .id("nvmd")
+                .timerange(RelativeRange.allTime())
+                .query(ElasticsearchQueryString.empty())
+                .filter(OrFilter.or(StreamFilter.ofId("missing-stream-id")))
+                .build();
+
+        assertThat(query.toNativeEntity(Collections.emptyMap(), Collections.emptyMap()).filter().filters())
+                .noneMatch(f -> f instanceof StreamFilter);
+    }
+
+    @Test
+    public void keepsResolvableStreamFilterAndDropsUnresolvable() {
+        final Stream stream = mock(Stream.class);
+        when(stream.getId()).thenReturn("native-stream-id");
+        final Map<EntityDescriptor, Object> nativeEntities =
+                Map.of(EntityDescriptor.create("cp-stream-id", ModelTypes.STREAM_REF_V1), stream);
+
+        final QueryEntity query = QueryEntity.Builder
+                .createWithDefaults()
+                .id("nvmd")
+                .timerange(RelativeRange.allTime())
+                .query(ElasticsearchQueryString.empty())
+                .filter(OrFilter.or(StreamFilter.ofId("cp-stream-id"), StreamFilter.ofId("missing-stream-id")))
+                .build();
+
+        final Set<Filter> mappedFilters =
+                query.toNativeEntity(Collections.emptyMap(), nativeEntities).filter().filters();
+
+        assertThat(mappedFilters).hasSize(1);
+        assertThat(((StreamFilter) mappedFilters.iterator().next()).streamId()).isEqualTo("native-stream-id");
+    }
+
+    @Test
+    public void failsOnWrongTypeStreamReference() {
+        // A non-null value that is not a Stream indicates a corrupt entity map, which must still abort the install.
+        final Map<EntityDescriptor, Object> nativeEntities =
+                Map.of(EntityDescriptor.create("cp-stream-id", ModelTypes.STREAM_REF_V1), "not-a-stream");
+
+        final QueryEntity query = QueryEntity.Builder
+                .createWithDefaults()
+                .id("nvmd")
+                .timerange(RelativeRange.allTime())
+                .query(ElasticsearchQueryString.empty())
+                .filter(OrFilter.or(StreamFilter.ofId("cp-stream-id")))
+                .build();
+
+        assertThatThrownBy(() -> query.toNativeEntity(Collections.emptyMap(), nativeEntities))
+                .isInstanceOf(ContentPackException.class)
+                .hasMessageContaining("Invalid type for stream");
+    }
+
     private class TestDBSearchFilter implements DBSearchFilter {
         String id;
 

graylog2-server/src/test/java/org/graylog2/contentpacks/model/entities/SearchTypeEntityTest.java

@@ -0,0 +1,97 @@
+/*
+ * 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.contentpacks.model.entities;
+
+import com.google.common.collect.ImmutableSet;
+import org.graylog.plugins.views.search.SearchType;
+import org.graylog2.contentpacks.exceptions.ContentPackException;
+import org.graylog2.contentpacks.model.ModelTypes;
+import org.graylog2.plugin.streams.Stream;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class SearchTypeEntityTest {
+
+    @Test
+    public void dropsUnresolvableStreamReferenceInsteadOfFailing() {
+        final MessageListEntity searchType = MessageListEntity.builder()
+                .id("search-type-id")
+                .streams(ImmutableSet.of("missing-stream-id"))
+                .build();
+
+        final SearchType nativeEntity = searchType.toNativeEntity(Collections.emptyMap(), Collections.emptyMap());
+
+        assertThat(nativeEntity.streams()).isEmpty();
+    }
+
+    @Test
+    public void keepsResolvableStreamReference() {
+        final Stream stream = mock(Stream.class);
+        when(stream.getId()).thenReturn("native-stream-id");
+        final Map<EntityDescriptor, Object> nativeEntities =
+                Map.of(EntityDescriptor.create("cp-stream-id", ModelTypes.STREAM_REF_V1), stream);
+
+        final MessageListEntity searchType = MessageListEntity.builder()
+                .id("search-type-id")
+                .streams(ImmutableSet.of("cp-stream-id"))
+                .build();
+
+        final SearchType nativeEntity = searchType.toNativeEntity(Collections.emptyMap(), nativeEntities);
+
+        assertThat(nativeEntity.streams()).containsExactly("native-stream-id");
+    }
+
+    @Test
+    public void dropsOnlyUnresolvableStreamReferences() {
+        final Stream stream = mock(Stream.class);
+        when(stream.getId()).thenReturn("native-stream-id");
+        final Map<EntityDescriptor, Object> nativeEntities =
+                Map.of(EntityDescriptor.create("cp-stream-id", ModelTypes.STREAM_REF_V1), stream);
+
+        final MessageListEntity searchType = MessageListEntity.builder()
+                .id("search-type-id")
+                .streams(ImmutableSet.of("cp-stream-id", "missing-stream-id"))
+                .build();
+
+        final SearchType nativeEntity = searchType.toNativeEntity(Collections.emptyMap(), nativeEntities);
+
+        assertThat(nativeEntity.streams()).containsExactly("native-stream-id");
+    }
+
+    @Test
+    public void failsOnWrongTypeStreamReference() {
+        // A non-null value that is not a Stream indicates a corrupt entity map, which must still abort the install.
+        final Map<EntityDescriptor, Object> nativeEntities =
+                Map.of(EntityDescriptor.create("cp-stream-id", ModelTypes.STREAM_REF_V1), "not-a-stream");
+
+        final MessageListEntity searchType = MessageListEntity.builder()
+                .id("search-type-id")
+                .streams(ImmutableSet.of("cp-stream-id"))
+                .build();
+
+        assertThatThrownBy(() -> searchType.toNativeEntity(Collections.emptyMap(), nativeEntities))
+                .isInstanceOf(ContentPackException.class)
+                .hasMessageContaining("Invalid type for stream");
+    }
+}