Shells/mocks for Kinesis Auto setup API endpoints

pom.xml

@@ -6,12 +6,12 @@
     <parent>
         <groupId>org.graylog.plugins</groupId>
         <artifactId>graylog-plugin-web-parent</artifactId>
-        <version>3.1.0-beta.3-SNAPSHOT</version>
+        <version>3.1.0-beta.4-SNAPSHOT</version>
         <relativePath>../graylog2-server/graylog-plugin-parent/graylog-plugin-web-parent</relativePath>
     </parent>
 
     <artifactId>graylog-plugin-integrations</artifactId>
-    <version>3.1.0-beta.3-SNAPSHOT</version>
+    <version>3.1.0-beta.4-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <name>${project.artifactId}</name>
@@ -31,7 +31,7 @@
         <maven.install.skip>true</maven.install.skip>
         <maven.deploy.skip>true</maven.deploy.skip>
         <maven.site.skip>true</maven.site.skip>
-        <graylog.version>3.1.0-beta.3-SNAPSHOT</graylog.version>
+        <graylog.version>3.1.0-beta.4-SNAPSHOT</graylog.version>
         <aws-java-sdk-2.version>2.5.33</aws-java-sdk-2.version>
         <aws-kinesis-client.version>2.2.0</aws-kinesis-client.version>
     </properties>

src/main/java/org/graylog/integrations/IntegrationsModule.java

@@ -22,6 +22,7 @@
 import org.graylog.integrations.aws.codecs.KinesisRawLogCodec;
 import org.graylog.integrations.aws.inputs.AWSInput;
 import org.graylog.integrations.aws.resources.AWSResource;
+import org.graylog.integrations.aws.resources.KinesisSetupResource;
 import org.graylog.integrations.aws.transports.AWSTransport;
 import org.graylog.integrations.aws.transports.KinesisTransport;
 import org.graylog.integrations.inputs.paloalto.PaloAltoCodec;
@@ -88,7 +89,8 @@ protected void configure() {
         addMessageInput(AWSInput.class);
         addPermissions(AWSPermissions.class);
         addRestResource(AWSResource.class);
-        addTransport(AWSTransport.NAME, AWSTransport .class);
+        addRestResource(KinesisSetupResource.class);
+        addTransport(AWSTransport.NAME, AWSTransport.class);
         addTransport(KinesisTransport.NAME, KinesisTransport.class);
         bind(CloudWatchLogsClientBuilder.class).toProvider(CloudWatchLogsClient::builder);
         bind(KinesisClientBuilder.class).toProvider(KinesisClient::builder);

src/main/java/org/graylog/integrations/aws/resources/AWSResource.java

@@ -12,10 +12,8 @@
 import org.graylog.integrations.aws.resources.requests.AWSInputCreateRequest;
 import org.graylog.integrations.aws.resources.requests.AWSRequestImpl;
 import org.graylog.integrations.aws.resources.requests.KinesisHealthCheckRequest;
-import org.graylog.integrations.aws.resources.requests.KinesisNewStreamRequest;
 import org.graylog.integrations.aws.resources.responses.AvailableServiceResponse;
 import org.graylog.integrations.aws.resources.responses.KinesisHealthCheckResponse;
-import org.graylog.integrations.aws.resources.responses.KinesisNewStreamResponse;
 import org.graylog.integrations.aws.resources.responses.LogGroupsResponse;
 import org.graylog.integrations.aws.resources.responses.RegionsResponse;
 import org.graylog.integrations.aws.resources.responses.StreamsResponse;
@@ -128,17 +126,6 @@ public Response kinesisHealthCheck(@ApiParam(name = "JSON body", required = true
         return Response.accepted().entity(response).build();
     }
 
-    @POST
-    @Timed
-    @Path("/kinesis/create_stream")
-    @ApiOperation(
-            value = "Attempt to create a new kinesis stream."
-    )
-    @RequiresPermissions(AWSPermissions.AWS_READ)
-    public KinesisNewStreamResponse createNewKinesisStream(@ApiParam(name = "JSON body", required = true) @Valid @NotNull KinesisNewStreamRequest kinesisNewStreamRequest) {
-        return kinesisService.createNewKinesisStream(kinesisNewStreamRequest);
-    }
-
     /**
      * Create a new AWS input.
      */

src/main/java/org/graylog/integrations/aws/resources/KinesisSetupResource.java

@@ -0,0 +1,150 @@
+package org.graylog.integrations.aws.resources;
+
+import com.codahale.metrics.annotation.Timed;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import io.swagger.annotations.ApiParam;
+import org.apache.shiro.authz.annotation.RequiresAuthentication;
+import org.apache.shiro.authz.annotation.RequiresPermissions;
+import org.graylog.integrations.aws.AWSPermissions;
+import org.graylog.integrations.aws.resources.requests.CreateLogSubscriptionPolicyRequest;
+import org.graylog.integrations.aws.resources.requests.CreateLogSubscriptionRequest;
+import org.graylog.integrations.aws.resources.requests.KinesisFullSetupRequest;
+import org.graylog.integrations.aws.resources.requests.KinesisNewStreamRequest;
+import org.graylog.integrations.aws.resources.responses.CreateLogSubscriptionPolicyResponse;
+import org.graylog.integrations.aws.resources.responses.CreateLogSubscriptionResponse;
+import org.graylog.integrations.aws.resources.responses.KinesisFullSetupResponse;
+import org.graylog.integrations.aws.resources.responses.KinesisFullSetupResponseStep;
+import org.graylog.integrations.aws.resources.responses.KinesisNewStreamResponse;
+import org.graylog.integrations.aws.service.KinesisService;
+import org.graylog2.plugin.rest.PluginRestResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.POST;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import java.util.ArrayList;
+
+/**
+ * Web endpoints for the Kinesis auto-setup.
+ */
+@Api(value = "AWSKinesisAuto", description = "AWS Kinesis auto-setup")
+@Path("/aws/kinesis/auto_setup")
+@RequiresAuthentication
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+public class KinesisSetupResource implements PluginRestResource {
+
+    private static final Logger LOG = LoggerFactory.getLogger(KinesisSetupResource.class);
+
+    private KinesisService kinesisService;
+
+    @Inject
+    public KinesisSetupResource(KinesisService kinesisService) {
+        this.kinesisService = kinesisService;
+    }
+
+    /**
+     * 1.  Create a new Kinesis stream and check that its active
+     * INPUT: credentials and streamName acquire from the user
+     * OUTPUT: String streamArn
+     * BACKEND DETAILS:
+     * void createKinesisStream(KinesisClient kinesisClient, String stream)
+     * StreamDescription streamDescription = checkKinesisStreamStatus(kinesisClient, stream);
+     * String streamArn = streamDescription.streamARN();
+     */
+    @POST
+    @Timed
+    @Path("/create_stream")
+    @ApiOperation(value = "Step 1: Attempt to create a new kinesis stream and wait for it to be ready.")
+    @RequiresPermissions(AWSPermissions.AWS_READ)
+    public KinesisNewStreamResponse createNewKinesisStream(@ApiParam(name = "JSON body", required = true) @Valid @NotNull
+                                                                   KinesisNewStreamRequest request) {
+
+        LOG.info("Request: [{}]", request);
+        // Real method call is already implemented. Commented out for now to allow UI to be mocked out easier.
+        // kinesisService.createNewKinesisStream(kinesisNewStreamRequest)
+
+        // Mock response
+        return KinesisNewStreamResponse.create(request.streamName(), "a-fake-arn", "The stream is good-to-go");
+    }
+
+    /**
+     * 2.  Add role with required permissions and acquire roleArn
+     * INPUT: roleName acquired from the user
+     * OUTPUT: streamARN acquired from step 3
+     * BACKEND DETAILS:
+     * void setRolePermissions(iam, roleName, streamArn, region.toString());
+     * String roleArn = getNewRolePermissions(iam, roleName);
+     */
+    @POST
+    @Timed
+    @Path("/create_subscription_policy")
+    @ApiOperation(value = "Step 2: Create AWS IAM policy needed for CloudWatch to write logs to Kinesis")
+    @RequiresPermissions(AWSPermissions.AWS_READ)
+    public CreateLogSubscriptionPolicyResponse createPolicies(@ApiParam(name = "JSON body", required = true) @Valid @NotNull
+                                                                      CreateLogSubscriptionPolicyRequest request) {
+        LOG.info("Request: [{}]", request);
+
+        // Mock response
+        return CreateLogSubscriptionPolicyResponse.create("fake-policy-name", "fake-policy-arn");
+    }
+
+    /**
+     * Creates a {@see <a href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html">Cloud Watch Subscription Filter</a>},
+     * which subscribes a kinesis stream to a CloudWatch log group. This will cause batches of CloudWatch loge messages
+     * to be put in the Kinesis stream in a GZipped JSON byte array payload.
+     *
+     * 3.  Add the subscription to the Cloudwatch log group for kinesis
+     * INPUT: credentials, logGroup, filterName, filterPattern acquired from user
+     * OUTPUT: streamArn acquired from step 3
+     * BACKEND DETAILS:
+     * roleArn acquired from step 4
+     * void addSubscriptionFilter(String logGroup, CloudWatchLogsClient cloudWatch, String streamArn, String roleArn, String filterName, String filterPattern)
+     */
+    @POST
+    @Timed
+    @Path("/create_subscription")
+    @ApiOperation(value = "Step 3: Subscribe a Kinesis stream to a CloudWatch log group")
+    @RequiresPermissions(AWSPermissions.AWS_READ)
+    public CreateLogSubscriptionResponse createSubscription(@ApiParam(name = "JSON body", required = true) @Valid @NotNull
+                                                                    CreateLogSubscriptionRequest request) {
+        LOG.info("Request: [{}]", request);
+
+        // TODO: We'll need to give some thought to how to effectively build the UI for the filterPattern and filterName
+        //   Perhaps we can provide default initialized values (eg. " " for filterPattern [matches all], and some generic pattern name).
+
+        // Mock response
+        return CreateLogSubscriptionResponse.create("Subscription created successfully");
+    }
+
+    /**
+     * Full Kinesis setup.
+     *
+     * @param request
+     * @return
+     */
+    @POST
+    @Timed
+    @Path("/full_setup")
+    @ApiOperation(value = "Full setup: Get all available AWS CloudWatch log groups names for the specified region")
+    @RequiresPermissions(AWSPermissions.AWS_READ)
+    public KinesisFullSetupResponse addSubscription(@ApiParam(name = "JSON body", required = true) @Valid @NotNull
+                                                            KinesisFullSetupRequest request) {
+
+        LOG.info("Request: [{}]", request);
+
+        // Mock response.
+        final ArrayList<KinesisFullSetupResponseStep> setupSteps = new ArrayList<>();
+        setupSteps.add(KinesisFullSetupResponseStep.create(true, "Create Stream", "The stream [this-stream-rocks] was successfully created."));
+        setupSteps.add(KinesisFullSetupResponseStep.create(true, "Create Policy", "The policy [this-policy-rocks] was successfully created."));
+        setupSteps.add(KinesisFullSetupResponseStep.create(false, "Subscribe stream to group", "Failed to create the subscription [Some specific AWS error]"));
+        return KinesisFullSetupResponse.create(false, "Auto-setup was not fully successful!", setupSteps);
+    }
+}

src/main/java/org/graylog/integrations/aws/resources/requests/CreateLogSubscriptionPolicyRequest.java

@@ -0,0 +1,45 @@
+package org.graylog.integrations.aws.resources.requests;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.auto.value.AutoValue;
+import org.graylog.autovalue.WithBeanGetter;
+
+@JsonAutoDetect
+@AutoValue
+@WithBeanGetter
+public abstract class CreateLogSubscriptionPolicyRequest implements AWSRequest {
+
+    private static final String ROLE_NAME = "role_name";
+    private static final String STREAM_NAME = "stream_name";
+    private static final String STREAM_ARN = "stream_arn";
+
+    @JsonProperty(REGION)
+    public abstract String region();
+
+    @JsonProperty(AWS_ACCESS_KEY_ID)
+    public abstract String awsAccessKeyId();
+
+    @JsonProperty(AWS_SECRET_ACCESS_KEY)
+    public abstract String awsSecretAccessKey();
+
+    @JsonProperty(ROLE_NAME)
+    public abstract String roleName();
+
+    @JsonProperty(STREAM_NAME)
+    public abstract String streamName();
+
+    @JsonProperty(STREAM_ARN)
+    public abstract String streamArn();
+
+    @JsonCreator
+    public static CreateLogSubscriptionPolicyRequest create(@JsonProperty(REGION) String region,
+                                                            @JsonProperty(AWS_ACCESS_KEY_ID) String awsAccessKeyId,
+                                                            @JsonProperty(AWS_SECRET_ACCESS_KEY) String awsSecretAccessKey,
+                                                            @JsonProperty(ROLE_NAME) String roleName,
+                                                            @JsonProperty(STREAM_NAME) String streamName,
+                                                            @JsonProperty(STREAM_ARN) String streamArn) {
+        return new AutoValue_CreateLogSubscriptionPolicyRequest(region, awsAccessKeyId, awsSecretAccessKey, roleName, streamName, streamArn);
+    }
+}

src/main/java/org/graylog/integrations/aws/resources/requests/CreateLogSubscriptionRequest.java

@@ -0,0 +1,60 @@
+package org.graylog.integrations.aws.resources.requests;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.auto.value.AutoValue;
+import org.graylog.autovalue.WithBeanGetter;
+
+@JsonAutoDetect
+@AutoValue
+@WithBeanGetter
+public abstract class CreateLogSubscriptionRequest implements AWSRequest {
+
+    private static final String LOG_GROUP_NAME = "log_group_name";
+    private static final String FILTER_NAME = "filter_name";
+    private static final String FILTER_PATTERN = "filter_pattern";
+    private static final String DESTINATION_STREAM_ARN = "destination_stream_arn";
+    private static final String ROLE_ARN = "role_arn";
+
+    @JsonProperty(REGION)
+    public abstract String region();
+
+    @JsonProperty(AWS_ACCESS_KEY_ID)
+    public abstract String awsAccessKeyId();
+
+    @JsonProperty(AWS_SECRET_ACCESS_KEY)
+    public abstract String awsSecretAccessKey();
+
+    /**
+     * {@see <a href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html">Cloud Watch Subscription Filter</a>},
+     *
+     * @return
+     */
+    @JsonProperty(LOG_GROUP_NAME)
+    public abstract String getLogGroupName();
+
+    @JsonProperty(FILTER_NAME)
+    public abstract String filterName();
+
+    @JsonProperty(FILTER_PATTERN)
+    public abstract String filterPattern();
+
+    @JsonProperty(DESTINATION_STREAM_ARN)
+    public abstract String destinationStreamArn();
+
+    @JsonProperty(ROLE_ARN)
+    public abstract String getRoleArn();
+
+    @JsonCreator
+    public static CreateLogSubscriptionRequest create(@JsonProperty(REGION) String region,
+                                                      @JsonProperty(AWS_ACCESS_KEY_ID) String awsAccessKeyId,
+                                                      @JsonProperty(AWS_SECRET_ACCESS_KEY) String awsSecretAccessKey,
+                                                      @JsonProperty(LOG_GROUP_NAME) String getLogGroupName,
+                                                      @JsonProperty(FILTER_NAME) String filterName,
+                                                      @JsonProperty(FILTER_PATTERN) String filterPattern,
+                                                      @JsonProperty(DESTINATION_STREAM_ARN) String destinationStreamArn,
+                                                      @JsonProperty(ROLE_ARN) String getRoleArn) {
+        return new AutoValue_CreateLogSubscriptionRequest(region, awsAccessKeyId, awsSecretAccessKey, getLogGroupName, filterName, filterPattern, destinationStreamArn, getRoleArn);
+    }
+}

src/main/java/org/graylog/integrations/aws/resources/requests/KinesisFullSetupRequest.java

@@ -0,0 +1,45 @@
+package org.graylog.integrations.aws.resources.requests;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.auto.value.AutoValue;
+import org.graylog.autovalue.WithBeanGetter;
+
+@JsonAutoDetect
+@AutoValue
+@WithBeanGetter
+public abstract class KinesisFullSetupRequest implements AWSRequest {
+
+    private static final String ROLE_NAME = "role_name";
+    private static final String LOG_GROUP_NAME = "log_group_name";
+    private static final String STREAM_NAME = "stream_name";
+
+    @JsonProperty(REGION)
+    public abstract String region();
+
+    @JsonProperty(AWS_ACCESS_KEY_ID)
+    public abstract String awsAccessKeyId();
+
+    @JsonProperty(AWS_SECRET_ACCESS_KEY)
+    public abstract String awsSecretAccessKey();
+
+    @JsonProperty(ROLE_NAME)
+    public abstract String roleName();
+
+    @JsonProperty(LOG_GROUP_NAME)
+    public abstract String getLogGroupName();
+
+    @JsonProperty(STREAM_NAME)
+    public abstract String streamName();
+
+    @JsonCreator
+    public static KinesisFullSetupRequest create(@JsonProperty(REGION) String region,
+                                                 @JsonProperty(AWS_ACCESS_KEY_ID) String awsAccessKeyId,
+                                                 @JsonProperty(AWS_SECRET_ACCESS_KEY) String awsSecretAccessKey,
+                                                 @JsonProperty(ROLE_NAME) String roleName,
+                                                 @JsonProperty(LOG_GROUP_NAME) String getLogGroupName,
+                                                 @JsonProperty(STREAM_NAME) String streamName) {
+        return new AutoValue_KinesisFullSetupRequest(region, awsAccessKeyId, awsSecretAccessKey, roleName, getLogGroupName, streamName);
+    }
+}