Add support for exporting pipelines and rules

Closes #2507
Closes #4222

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/PipelineService.java

@@ -25,6 +25,8 @@ public interface PipelineService {
 
     PipelineDao load(String id) throws NotFoundException;
 
+    PipelineDao loadByName(String name) throws NotFoundException;
+
     Collection<PipelineDao> loadAll();
 
     void delete(String id);

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/PipelineStreamConnectionsService.java

@@ -28,5 +28,7 @@ public interface PipelineStreamConnectionsService {
 
     Set<PipelineConnections> loadAll();
 
+    Set<PipelineConnections> loadByPipelineId(String pipelineId);
+
     void delete(String streamId);
 }

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/RuleService.java

@@ -25,6 +25,8 @@ public interface RuleService {
 
     RuleDao load(String id) throws NotFoundException;
 
+    RuleDao loadByName(String name) throws NotFoundException;
+
     Collection<RuleDao> loadAll();
 
     void delete(String id);

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/memory/InMemoryPipelineService.java

@@ -64,6 +64,15 @@ public PipelineDao load(String id) throws NotFoundException {
         return pipeline;
     }
 
+    @Override
+    public PipelineDao loadByName(String name) throws NotFoundException {
+        final String id = titleToId.get(name);
+        if (id == null) {
+            throw new NotFoundException("No pipeline with name " + name);
+        }
+        return load(id);
+    }
+
     @Override
     public Collection<PipelineDao> loadAll() {
         return ImmutableSet.copyOf(store.values());

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/memory/InMemoryPipelineStreamConnectionsService.java

@@ -17,21 +17,22 @@
 package org.graylog.plugins.pipelineprocessor.db.memory;
 
 import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.MapMaker;
 import org.graylog.plugins.pipelineprocessor.db.PipelineStreamConnectionsService;
 import org.graylog.plugins.pipelineprocessor.rest.PipelineConnections;
 import org.graylog2.database.NotFoundException;
 
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 public class InMemoryPipelineStreamConnectionsService implements PipelineStreamConnectionsService {
 
     // poor man's id generator
     private AtomicLong idGen = new AtomicLong(0);
 
-    private Map<String, PipelineConnections> store = new MapMaker().makeMap();
+    private Map<String, PipelineConnections> store = new ConcurrentHashMap<>();
 
     @Override
     public PipelineConnections save(PipelineConnections connections) {
@@ -57,6 +58,13 @@ public Set<PipelineConnections> loadAll() {
         return ImmutableSet.copyOf(store.values());
     }
 
+    @Override
+    public Set<PipelineConnections> loadByPipelineId(String pipelineId) {
+        return store.values().stream()
+                .filter(connection -> connection.pipelineIds().contains(pipelineId))
+                .collect(Collectors.toSet());
+    }
+
     @Override
     public void delete(String streamId) {
         try {

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/memory/InMemoryRuleService.java

@@ -67,6 +67,15 @@ public RuleDao load(String id) throws NotFoundException {
         return rule;
     }
 
+    @Override
+    public RuleDao loadByName(String name) throws NotFoundException {
+        final String id = titleToId.get(name);
+        if (id == null) {
+            throw new NotFoundException("No rule with name " + name);
+        }
+        return load(id);
+    }
+
     @Override
     public Collection<RuleDao> loadAll() {
         return ImmutableSet.copyOf(store.values());

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/mongodb/MongoDbPipelineService.java

@@ -25,6 +25,7 @@
 import org.graylog2.database.MongoConnection;
 import org.graylog2.database.NotFoundException;
 import org.mongojack.DBCursor;
+import org.mongojack.DBQuery;
 import org.mongojack.DBSort;
 import org.mongojack.JacksonDBCollection;
 import org.mongojack.WriteResult;
@@ -67,6 +68,16 @@ public PipelineDao load(String id) throws NotFoundException {
         return pipeline;
     }
 
+    @Override
+    public PipelineDao loadByName(String name) throws NotFoundException {
+        final DBQuery.Query query = DBQuery.is("title", name);
+        final PipelineDao pipeline = dbCollection.findOne(query);
+        if (pipeline == null) {
+            throw new NotFoundException("No pipeline with name " + name);
+        }
+        return pipeline;
+    }
+
     @Override
     public Collection<PipelineDao> loadAll() {
         try (DBCursor<PipelineDao> daos = dbCollection.find()) {

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/mongodb/MongoDbPipelineStreamConnectionsService.java

@@ -18,6 +18,7 @@
 
 import com.google.common.collect.ImmutableSet;
 import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
 import com.mongodb.MongoException;
 import org.graylog.plugins.pipelineprocessor.db.PipelineStreamConnectionsService;
 import org.graylog.plugins.pipelineprocessor.rest.PipelineConnections;
@@ -53,7 +54,6 @@ public MongoDbPipelineStreamConnectionsService(MongoConnection mongoConnection,
         dbCollection.createIndex(DBSort.asc("stream_id"), new BasicDBObject("unique", true));
     }
 
-
     @Override
     public PipelineConnections save(PipelineConnections connections) {
         PipelineConnections existingConnections = dbCollection.findOne(DBQuery.is("stream_id", connections.streamId()));
@@ -86,6 +86,19 @@ public Set<PipelineConnections> loadAll() {
         }
     }
 
+    @Override
+    public Set<PipelineConnections> loadByPipelineId(String pipelineId) {
+        // Thanks, MongoJack!
+        // https://github.com/mongojack/mongojack/issues/12
+        final DBObject query = new BasicDBObject("pipeline_ids", new BasicDBObject("$in", Collections.singleton(pipelineId)));
+        try (DBCursor<PipelineConnections> pipelineConnections = dbCollection.find(query)) {
+            return ImmutableSet.copyOf((Iterable<PipelineConnections>) pipelineConnections);
+        } catch (MongoException e) {
+            log.error("Unable to load pipeline connections for pipeline ID " + pipelineId, e);
+            return Collections.emptySet();
+        }
+    }
+
     @Override
     public void delete(String streamId) {
         try {

graylog2-server/src/main/java/org/graylog/plugins/pipelineprocessor/db/mongodb/MongoDbRuleService.java

@@ -71,6 +71,16 @@ public RuleDao load(String id) throws NotFoundException {
         return rule;
     }
 
+    @Override
+    public RuleDao loadByName(String name) throws NotFoundException {
+        final DBQuery.Query query = DBQuery.is("title", name);
+        final RuleDao rule = dbCollection.findOne(query);
+        if (rule == null) {
+            throw new NotFoundException("No rule with name " + name);
+        }
+        return rule;
+    }
+
     @Override
     public Collection<RuleDao> loadAll() {
         try(DBCursor<RuleDao> ruleDaos = dbCollection.find().sort(DBSort.asc("title"))) {

graylog2-server/src/main/java/org/graylog2/contentpacks/ContentPacksModule.java

@@ -24,6 +24,8 @@
 import org.graylog2.contentpacks.catalogs.LookupDataAdapterCatalog;
 import org.graylog2.contentpacks.catalogs.LookupTableCatalog;
 import org.graylog2.contentpacks.catalogs.OutputCatalog;
+import org.graylog2.contentpacks.catalogs.PipelineCatalog;
+import org.graylog2.contentpacks.catalogs.PipelineRuleCatalog;
 import org.graylog2.contentpacks.catalogs.StreamCatalog;
 import org.graylog2.contentpacks.jersey.ModelIdParamConverter;
 import org.graylog2.plugin.PluginModule;
@@ -44,6 +46,8 @@ protected void configure() {
         addEntityCatalog(LookupDataAdapterCatalog.TYPE, LookupDataAdapterCatalog.class);
         addEntityCatalog(LookupTableCatalog.TYPE, LookupTableCatalog.class);
         addEntityCatalog(OutputCatalog.TYPE, OutputCatalog.class);
+        addEntityCatalog(PipelineCatalog.TYPE, PipelineCatalog.class);
+        addEntityCatalog(PipelineRuleCatalog.TYPE, PipelineRuleCatalog.class);
         addEntityCatalog(StreamCatalog.TYPE, StreamCatalog.class);
     }
 }

graylog2-server/src/main/java/org/graylog2/contentpacks/catalogs/PipelineCatalog.java

@@ -0,0 +1,124 @@
+/**
+ * This file is part of Graylog.
+ *
+ * Graylog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Graylog 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
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Graylog.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.graylog2.contentpacks.catalogs;
+
+import com.google.common.graph.Graph;
+import com.google.common.graph.GraphBuilder;
+import com.google.common.graph.ImmutableGraph;
+import com.google.common.graph.MutableGraph;
+import org.graylog.plugins.pipelineprocessor.ast.Pipeline;
+import org.graylog.plugins.pipelineprocessor.ast.Rule;
+import org.graylog.plugins.pipelineprocessor.ast.Stage;
+import org.graylog.plugins.pipelineprocessor.db.PipelineDao;
+import org.graylog.plugins.pipelineprocessor.db.PipelineService;
+import org.graylog.plugins.pipelineprocessor.db.PipelineStreamConnectionsService;
+import org.graylog.plugins.pipelineprocessor.parser.PipelineRuleParser;
+import org.graylog.plugins.pipelineprocessor.rest.PipelineConnections;
+import org.graylog2.contentpacks.codecs.PipelineCodec;
+import org.graylog2.contentpacks.model.ModelId;
+import org.graylog2.contentpacks.model.ModelType;
+import org.graylog2.contentpacks.model.ModelTypes;
+import org.graylog2.contentpacks.model.entities.Entity;
+import org.graylog2.contentpacks.model.entities.EntityDescriptor;
+import org.graylog2.contentpacks.model.entities.EntityExcerpt;
+import org.graylog2.database.NotFoundException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.util.Collection;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class PipelineCatalog implements EntityCatalog {
+    private static final Logger LOG = LoggerFactory.getLogger(PipelineCatalog.class);
+
+    public static final ModelType TYPE = ModelTypes.PIPELINE;
+
+    private final PipelineService pipelineService;
+    private final PipelineStreamConnectionsService streamConnectionsService;
+    private final PipelineRuleParser pipelineRuleParser;
+    private final PipelineCodec codec;
+
+    @Inject
+    public PipelineCatalog(PipelineService pipelineService,
+                           PipelineStreamConnectionsService streamConnectionsService,
+                           PipelineRuleParser pipelineRuleParser,
+                           PipelineCodec codec) {
+        this.pipelineService = pipelineService;
+        this.streamConnectionsService = streamConnectionsService;
+        this.pipelineRuleParser = pipelineRuleParser;
+        this.codec = codec;
+    }
+
+    @Override
+    public Set<EntityExcerpt> listEntityExcerpts() {
+        return pipelineService.loadAll().stream()
+                .map(codec::createExcerpt)
+                .collect(Collectors.toSet());
+    }
+
+    @Override
+    public Optional<Entity> collectEntity(EntityDescriptor entityDescriptor) {
+        final ModelId modelId = entityDescriptor.id();
+        try {
+            final PipelineDao pipelineDao = pipelineService.loadByName(modelId.id());
+            return Optional.of(codec.encode(pipelineDao));
+        } catch (NotFoundException e) {
+            LOG.debug("Couldn't find pipeline {}", entityDescriptor, e);
+            return Optional.empty();
+        }
+    }
+
+    @Override
+    public Graph<EntityDescriptor> resolve(EntityDescriptor entityDescriptor) {
+        final MutableGraph<EntityDescriptor> mutableGraph = GraphBuilder.directed().build();
+        mutableGraph.addNode(entityDescriptor);
+
+        final ModelId modelId = entityDescriptor.id();
+        try {
+            final PipelineDao pipelineDao = pipelineService.loadByName(modelId.id());
+            final String pipelineSource = pipelineDao.source();
+            final Collection<String> referencedRules = referencedRules(pipelineSource);
+            referencedRules.stream()
+                    .map(ModelId::of)
+                    .map(id -> EntityDescriptor.create(id, ModelTypes.PIPELINE_RULE))
+                    .forEach(output -> mutableGraph.putEdge(entityDescriptor, output));
+
+            final Set<PipelineConnections> pipelineConnections = streamConnectionsService.loadByPipelineId(pipelineDao.id());
+            pipelineConnections.stream()
+                    .map(PipelineConnections::streamId)
+                    .map(ModelId::of)
+                    .map(id -> EntityDescriptor.create(id, ModelTypes.STREAM))
+                    .forEach(output -> mutableGraph.putEdge(entityDescriptor, output));
+        } catch (NotFoundException e) {
+            LOG.debug("Couldn't find pipeline {}", entityDescriptor, e);
+        }
+
+        return ImmutableGraph.copyOf(mutableGraph);
+    }
+
+    private Collection<String> referencedRules(String pipelineSource) {
+        final Pipeline pipeline = pipelineRuleParser.parsePipeline("dummy", pipelineSource);
+        return pipeline.stages().stream()
+                .map(Stage::getRules)
+                .flatMap(Collection::stream)
+                .map(Rule::name)
+                .collect(Collectors.toSet());
+    }
+}

graylog2-server/src/main/java/org/graylog2/contentpacks/catalogs/PipelineRuleCatalog.java

@@ -0,0 +1,70 @@
+/**
+ * This file is part of Graylog.
+ *
+ * Graylog is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * Graylog 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
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with Graylog.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.graylog2.contentpacks.catalogs;
+
+import org.graylog.plugins.pipelineprocessor.db.RuleDao;
+import org.graylog.plugins.pipelineprocessor.db.RuleService;
+import org.graylog2.contentpacks.codecs.PipelineRuleCodec;
+import org.graylog2.contentpacks.model.ModelId;
+import org.graylog2.contentpacks.model.ModelType;
+import org.graylog2.contentpacks.model.ModelTypes;
+import org.graylog2.contentpacks.model.entities.Entity;
+import org.graylog2.contentpacks.model.entities.EntityDescriptor;
+import org.graylog2.contentpacks.model.entities.EntityExcerpt;
+import org.graylog2.database.NotFoundException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class PipelineRuleCatalog implements EntityCatalog {
+    private static final Logger LOG = LoggerFactory.getLogger(PipelineRuleCatalog.class);
+
+    public static final ModelType TYPE = ModelTypes.PIPELINE_RULE;
+
+    private final RuleService ruleService;
+    private final PipelineRuleCodec codec;
+
+    @Inject
+    public PipelineRuleCatalog(RuleService ruleService,
+                               PipelineRuleCodec codec) {
+        this.ruleService = ruleService;
+        this.codec = codec;
+    }
+
+    @Override
+    public Set<EntityExcerpt> listEntityExcerpts() {
+        return ruleService.loadAll().stream()
+                .map(codec::createExcerpt)
+                .collect(Collectors.toSet());
+    }
+
+    @Override
+    public Optional<Entity> collectEntity(EntityDescriptor entityDescriptor) {
+        final ModelId modelId = entityDescriptor.id();
+        try {
+            final RuleDao ruleDao = ruleService.loadByName(modelId.id());
+            return Optional.of(codec.encode(ruleDao));
+        } catch (NotFoundException e) {
+            LOG.debug("Couldn't find pipeline rule {}", entityDescriptor, e);
+            return Optional.empty();
+        }
+    }
+}