Refactor collector ingest input management

graylog2-server/src/main/java/org/graylog/collectors/CollectorIngestInputService.java

@@ -0,0 +1,95 @@
+/*
+ * 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.graylog.collectors;
+
+import jakarta.inject.Inject;
+import jakarta.ws.rs.BadRequestException;
+import jakarta.ws.rs.ForbiddenException;
+import jakarta.ws.rs.NotFoundException;
+import org.apache.shiro.subject.Subject;
+import org.graylog.collectors.input.CollectorIngestHttpInput;
+import org.graylog2.Configuration;
+import org.graylog2.inputs.Input;
+import org.graylog2.inputs.InputService;
+import org.graylog2.plugin.configuration.ConfigurationException;
+import org.graylog2.plugin.database.ValidationException;
+import org.graylog2.rest.models.system.inputs.requests.InputCreateRequest;
+import org.graylog2.shared.inputs.MessageInputFactory;
+import org.graylog2.shared.inputs.NoSuchInputTypeException;
+import org.graylog2.shared.security.RestPermissions;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.graylog2.shared.utilities.StringUtils.f;
+
+public class CollectorIngestInputService {
+    private final InputService inputService;
+    private final MessageInputFactory messageInputFactory;
+    private final Configuration configuration;
+
+    @Inject
+    public CollectorIngestInputService(InputService inputService,
+                                       MessageInputFactory messageInputFactory,
+                                       Configuration configuration) {
+        this.inputService = inputService;
+        this.messageInputFactory = messageInputFactory;
+        this.configuration = configuration;
+    }
+
+    public List<String> getInputIds() {
+        return inputService.allByType(CollectorIngestHttpInput.class.getCanonicalName())
+                .stream()
+                .map(Input::getId)
+                .toList();
+    }
+
+    public void createInput(Subject subject, String userName, int port) throws ValidationException {
+        if (configuration.isCloud()) {
+            throw new BadRequestException("Creating collector ingest inputs is not supported in cloud environments");
+        }
+
+        if (!subject.isPermitted(RestPermissions.INPUTS_CREATE)) {
+            throw new ForbiddenException("Not permitted to create inputs");
+        }
+        if (!subject.isPermitted(f("%s:%s", RestPermissions.INPUT_TYPES_CREATE, CollectorIngestHttpInput.class.getCanonicalName()))) {
+            throw new ForbiddenException(f("Not permitted to create input type %s", CollectorIngestHttpInput.class.getCanonicalName()));
+        }
+
+        try {
+            final var inputCreateRequest = InputCreateRequest.create(
+                    CollectorIngestHttpInput.NAME,
+                    CollectorIngestHttpInput.class.getCanonicalName(),
+                    true,
+                    Map.of(
+                            "bind_address", "0.0.0.0",
+                            "port", port
+                    ),
+                    null
+            );
+            final var messageInput = messageInputFactory.create(
+                    inputCreateRequest, userName, inputCreateRequest.node(), false);
+            messageInput.checkConfiguration();
+            final var input = inputService.create(messageInput.asMap());
+            inputService.save(input);
+        } catch (NoSuchInputTypeException e) {
+            throw new NotFoundException("No such input type registered", e);
+        } catch (ConfigurationException e) {
+            throw new BadRequestException("Invalid input configuration", e);
+        }
+    }
+}

graylog2-server/src/main/java/org/graylog/collectors/CollectorsConfig.java

@@ -95,7 +95,7 @@ public static Builder createDefaultBuilder(String hostname) {
         requireNonBlank(hostname, "hostname can't be blank");
 
         return CollectorsConfig.builder()
-                .http(new IngestEndpointConfig(true, hostname, DEFAULT_HTTP_PORT, null));
+                .http(new IngestEndpointConfig(hostname, DEFAULT_HTTP_PORT));
     }
 
     public static CollectorsConfig createDefault(String hostname) {

graylog2-server/src/main/java/org/graylog/collectors/CollectorsConfigService.java

@@ -19,9 +19,11 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import org.graylog.collectors.events.CollectorCaConfigUpdated;
+import org.graylog2.configuration.HttpConfiguration;
 import org.graylog2.events.ClusterEventBus;
 import org.graylog2.plugin.cluster.ClusterConfigService;
 
+import java.net.URI;
 import java.util.Objects;
 import java.util.Optional;
 
@@ -34,11 +36,15 @@ public class CollectorsConfigService {
 
     private final ClusterConfigService clusterConfigService;
     private final ClusterEventBus clusterEventBus;
+    private final URI httpExternalUri;
 
     @Inject
-    public CollectorsConfigService(ClusterConfigService clusterConfigService, ClusterEventBus clusterEventBus) {
+    public CollectorsConfigService(ClusterConfigService clusterConfigService,
+                                   ClusterEventBus clusterEventBus,
+                                   HttpConfiguration httpConfiguration) {
         this.clusterConfigService = clusterConfigService;
         this.clusterEventBus = clusterEventBus;
+        this.httpExternalUri = httpConfiguration.getHttpExternalUri();
     }
 
     /**
@@ -56,7 +62,7 @@ public Optional<CollectorsConfig> get() {
      * @return the current config or a default config
      */
     public CollectorsConfig getOrDefault() {
-        return get().orElse(CollectorsConfig.createDefault("localhost"));
+        return get().orElse(CollectorsConfig.createDefault(httpExternalUri.getHost()));
     }
 
     /**

graylog2-server/src/main/java/org/graylog/collectors/IngestEndpointConfig.java

@@ -17,11 +17,8 @@
 package org.graylog.collectors;
 
 import com.fasterxml.jackson.annotation.JsonProperty;
-import jakarta.annotation.Nullable;
 
 public record IngestEndpointConfig(
-        @JsonProperty("enabled") boolean enabled,
         @JsonProperty("hostname") String hostname,
-        @JsonProperty("port") int port,
-        @JsonProperty("input_id") @Nullable String inputId
+        @JsonProperty("port") int port
 ) {}

graylog2-server/src/main/java/org/graylog/collectors/input/CollectorIngestHttpInput.java

@@ -70,6 +70,17 @@ public static class Descriptor extends MessageInput.Descriptor {
         public Descriptor() {
             super(NAME, false, "");
         }
+
+        @Override
+        public String getDescription() {
+            return "This input receives data from managed collectors over mTLS-secured HTTP. "
+                    + "Managed collectors are configured to send their data to the external address "
+                    + "specified in the Collectors Settings page. The port configured on this input must "
+                    + "either match the external port from the Collectors Settings, or the external port "
+                    + "must be routed to this input's port (e.g. via a load balancer or port mapping). "
+                    + "Changing this input's port without updating the routing will prevent collectors "
+                    + "from delivering data.";
+        }
     }
 
     @ConfigClass