diff --git a/changelog/unreleased/issue-22085.toml b/changelog/unreleased/issue-22085.toml new file mode 100644 index 0000000000..795dc17cdd --- /dev/null +++ b/changelog/unreleased/issue-22085.toml @@ -0,0 +1,5 @@ +type = "f" +message = "Add additional data fields for the Kinesis/Cloudwatch input." + +issues = ["22085"] +pulls = ["22370"] diff --git a/changelog/unreleased/issue-22137.toml b/changelog/unreleased/issue-22137.toml new file mode 100644 index 0000000000..b235980db1 --- /dev/null +++ b/changelog/unreleased/issue-22137.toml @@ -0,0 +1,5 @@ +type = "f" +message = "Improve error handling for the Kinesis/Cloudwatch input." + +issues = ["22137"] +pulls = ["22370"] diff --git a/changelog/unreleased/issue-22138.toml b/changelog/unreleased/issue-22138.toml new file mode 100644 index 0000000000..7f32e8a523 --- /dev/null +++ b/changelog/unreleased/issue-22138.toml @@ -0,0 +1,5 @@ +type = "f" +message = "Allow the Kinesis/Cloudwatch input to be created for empty Kinesis streams." + +issues = ["22138"] +pulls = ["22370"] diff --git a/changelog/unreleased/issue-22140.toml b/changelog/unreleased/issue-22140.toml new file mode 100644 index 0000000000..4e4626a2a5 --- /dev/null +++ b/changelog/unreleased/issue-22140.toml @@ -0,0 +1,5 @@ +type = "f" +message = "Allow the source field to be overridden for the Kinesis/Cloudwatch input." + +issues = ["22140"] +pulls = ["22370"] diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/AWSMessageType.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/AWSMessageType.java index a403ea546d..83141dc4e6 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/AWSMessageType.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/AWSMessageType.java @@ -38,14 +38,14 @@ public enum AWSMessageType { * A raw string stored in CloudWatch or Kinesis. */ KINESIS_RAW(Source.KINESIS, "Kinesis Raw", KinesisRawLogCodec.NAME, - KinesisRawLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), + KinesisRawLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), KINESIS_CLOUDWATCH_RAW(Source.KINESIS, "Kinesis CloudWatch Raw", KinesisRawLogCodec.NAME, - KinesisRawLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), + KinesisRawLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), // Flow Logs delivered to Kinesis via CloudWatch subscriptions. KINESIS_CLOUDWATCH_FLOW_LOGS(Source.KINESIS, "Kinesis CloudWatch Flow Log", KinesisCloudWatchFlowLogCodec.NAME, - KinesisCloudWatchFlowLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), + KinesisCloudWatchFlowLogCodec.Factory.class, KinesisTransport.NAME, KinesisTransport.Factory.class), UNKNOWN(); @@ -110,6 +110,6 @@ public enum AWSMessageType { */ public static List getMessageTypes() { - return Arrays.stream(values()).filter(m -> !m.equals(UNKNOWN)).collect(Collectors.toList()); + return Arrays.stream(values()).filter(m -> !(m.equals(UNKNOWN))).collect(Collectors.toList()); } -} \ No newline at end of file +} diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/cloudwatch/KinesisLogEntry.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/cloudwatch/KinesisLogEntry.java index c28fb3989b..19dfe26e71 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/cloudwatch/KinesisLogEntry.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/cloudwatch/KinesisLogEntry.java @@ -23,6 +23,9 @@ import com.google.auto.value.AutoValue; import org.graylog.autovalue.WithBeanGetter; import org.joda.time.DateTime; +import javax.annotation.Nullable; +import java.util.List; + @JsonAutoDetect @AutoValue @WithBeanGetter @@ -33,6 +36,9 @@ public abstract class KinesisLogEntry { private static final String LOG_STREAM = "log_stream"; private static final String TIMESTAMP = "timestamp"; private static final String MESSAGE = "message"; + private static final String OWNER = "owner"; + private static final String MESSAGE_TYPE = "message_type"; + private static final String SUBSCRIPTION_FILTERS = "subscription_filters"; @JsonProperty(KINESIS_STREAM) public abstract String kinesisStream(); @@ -54,12 +60,27 @@ public abstract class KinesisLogEntry { @JsonProperty(MESSAGE) public abstract String message(); + @JsonProperty(OWNER) + public abstract String owner(); + + @Nullable + @JsonProperty(MESSAGE_TYPE) + public abstract String messageType(); + + @Nullable + @JsonProperty(SUBSCRIPTION_FILTERS) + public abstract List subscriptionFilters(); + @JsonCreator public static KinesisLogEntry create(@JsonProperty(KINESIS_STREAM) String kinesisStream, @JsonProperty(LOG_GROUP) String logGroup, @JsonProperty(LOG_STREAM) String logStream, @JsonProperty(TIMESTAMP) DateTime timestamp, - @JsonProperty(MESSAGE) String message) { - return new AutoValue_KinesisLogEntry(kinesisStream, logGroup, logStream, timestamp, message); + @JsonProperty(MESSAGE) String message, + @JsonProperty(OWNER) String owner, + @JsonProperty(MESSAGE_TYPE) String messageType, + @JsonProperty(SUBSCRIPTION_FILTERS) List subscriptionFilters) { + return new AutoValue_KinesisLogEntry(kinesisStream, logGroup, logStream, timestamp, message, + owner, messageType, subscriptionFilters); } -} \ No newline at end of file +} diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AWSCodec.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AWSCodec.java index 424bccb8c0..a449f4b338 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AWSCodec.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AWSCodec.java @@ -40,6 +40,9 @@ import java.util.Map; import java.util.Optional; import java.util.stream.Collectors; +import static org.graylog.integrations.aws.inputs.AWSInput.getKinesisStreamARNDefinition; +import static org.graylog.integrations.aws.inputs.AWSInput.getOverrideSourceFieldDefinition; + public class AWSCodec extends AbstractCodec { public static final String NAME = "AWSCodec"; @@ -116,6 +119,9 @@ public class AWSCodec extends AbstractCodec { "Add field with the Flow Log prefix e. g. \"src_addr\" -> \"flow_log_src_addr\"." )); + request.addField(getKinesisStreamARNDefinition()); + request.addField(getOverrideSourceFieldDefinition()); + return request; } diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AbstractKinesisCodec.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AbstractKinesisCodec.java index 97e0be258d..1dec7b72e5 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AbstractKinesisCodec.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AbstractKinesisCodec.java @@ -17,28 +17,30 @@ package org.graylog.integrations.aws.codecs; import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.commons.collections4.CollectionUtils; import org.graylog.integrations.aws.cloudwatch.KinesisLogEntry; +import org.graylog.integrations.aws.transports.KinesisTransport; import org.graylog2.plugin.Message; import org.graylog2.plugin.configuration.Configuration; import org.graylog2.plugin.inputs.codecs.AbstractCodec; import org.graylog2.plugin.inputs.codecs.CodecAggregator; import org.graylog2.plugin.inputs.failure.InputProcessingException; import org.graylog2.plugin.journal.RawMessage; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; import javax.annotation.Nonnull; import javax.annotation.Nullable; import java.util.Optional; public abstract class AbstractKinesisCodec extends AbstractCodec { - - private static final Logger LOG = LoggerFactory.getLogger(AbstractKinesisCodec.class); - + public static final String SOURCE = "aws-kinesis-raw-logs"; static final String SOURCE_GROUP_IDENTIFIER = "aws_source"; static final String FIELD_KINESIS_STREAM = "aws_kinesis_stream"; + static final String FIELD_KINESIS_STREAM_ARN = "aws_kinesis_stream_arn"; static final String FIELD_LOG_GROUP = "aws_log_group"; static final String FIELD_LOG_STREAM = "aws_log_stream"; + public static final String FIELD_SUBSCRIPTION_FILTERS = "aws_subscription_filters"; + public static final String FIELD_MESSAGE_TYPE = "aws_kinesis_message_type"; + private static final String FIELD_OWNER = "aws_owner"; private final ObjectMapper objectMapper; @@ -66,6 +68,18 @@ public abstract class AbstractKinesisCodec extends AbstractCodec { protected abstract Optional decodeLogData(@Nonnull final KinesisLogEntry event); + public void setCommonFields(KinesisLogEntry logEvent, Message result) { + result.addField(FIELD_LOG_GROUP, logEvent.logGroup()); + result.addField(FIELD_LOG_STREAM, logEvent.logStream()); + result.addField(FIELD_KINESIS_STREAM, logEvent.kinesisStream()); + result.addField(FIELD_KINESIS_STREAM_ARN, configuration.getString(KinesisTransport.CK_KINESIS_STREAM_ARN)); + result.addField(FIELD_OWNER, logEvent.owner()); + result.addField(FIELD_MESSAGE_TYPE, logEvent.messageType()); + if (CollectionUtils.isNotEmpty(logEvent.subscriptionFilters())) { + result.addField(FIELD_SUBSCRIPTION_FILTERS, logEvent.subscriptionFilters()); + } + } + @Nonnull @Override public Configuration getConfiguration() { diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisCloudWatchFlowLogCodec.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisCloudWatchFlowLogCodec.java index 28f2be0d78..8bfd2185cb 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisCloudWatchFlowLogCodec.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisCloudWatchFlowLogCodec.java @@ -77,10 +77,8 @@ public class KinesisCloudWatchFlowLogCodec extends AbstractKinesisCodec { buildSummary(flowLogMessage), source, flowLogMessage.getTimestamp()); - result.addFields(buildFields(flowLogMessage)); - result.addField(FIELD_KINESIS_STREAM, logEvent.kinesisStream()); - result.addField(FIELD_LOG_GROUP, logEvent.logGroup()); - result.addField(FIELD_LOG_STREAM, logEvent.logStream()); + result.addFields(addFlowLogFields(flowLogMessage)); + setCommonFields(logEvent, result); result.addField(SOURCE_GROUP_IDENTIFIER, true); return Optional.of(result); @@ -100,8 +98,7 @@ public class KinesisCloudWatchFlowLogCodec extends AbstractKinesisCodec { .toString(); } - private Map buildFields(FlowLogMessage msg) { - + private Map addFlowLogFields(FlowLogMessage msg) { final String prefix = this.noFlowLogPrefix ? "" : FLOW_LOG_PREFIX; final HashMap fields = new HashMap<>(); fields.put(prefix + FIELD_ACCOUNT_ID, msg.getAccountId()); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisRawLogCodec.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisRawLogCodec.java index fe78cf276b..177b3dd393 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisRawLogCodec.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisRawLogCodec.java @@ -33,7 +33,6 @@ import java.util.Optional; public class KinesisRawLogCodec extends AbstractKinesisCodec { public static final String NAME = "CloudWatchRawLog"; - static final String SOURCE = "aws-kinesis-raw-logs"; private final MessageFactory messageFactory; @Inject @@ -51,10 +50,7 @@ public class KinesisRawLogCodec extends AbstractKinesisCodec { source, logEvent.timestamp() ); - result.addField(FIELD_KINESIS_STREAM, logEvent.kinesisStream()); - result.addField(FIELD_LOG_GROUP, logEvent.logGroup()); - result.addField(FIELD_LOG_STREAM, logEvent.logStream()); - + setCommonFields(logEvent, result); return Optional.of(result); } catch (Exception e) { throw new RuntimeException("Could not deserialize AWS FlowLog record.", e); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/inputs/AWSInput.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/inputs/AWSInput.java index 2a7a0d6d59..fe96fe504e 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/inputs/AWSInput.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/inputs/AWSInput.java @@ -20,6 +20,7 @@ import com.codahale.metrics.MetricRegistry; import com.google.inject.assistedinject.Assisted; import jakarta.inject.Inject; import org.graylog.integrations.aws.codecs.AWSCodec; +import org.graylog.integrations.aws.codecs.KinesisRawLogCodec; import org.graylog.integrations.aws.service.AWSService; import org.graylog.integrations.aws.transports.AWSTransport; import org.graylog.integrations.aws.transports.KinesisTransport; @@ -55,6 +56,8 @@ public class AWSInput extends MessageInput { public static final String CK_IAM_ENDPOINT = "iam_endpoint"; public static final String CK_KINESIS_ENDPOINT = "kinesis_endpoint"; + public static final String CK_OVERRIDE_SOURCE = "override_source"; + @Inject public AWSInput(@Assisted Configuration configuration, MetricRegistry metricRegistry, @@ -182,6 +185,7 @@ public class AWSInput extends MessageInput { "The name of the Kinesis stream that receives your messages. See README for instructions on how to connect messages to a Kinesis Stream.", ConfigurationField.Optional.NOT_OPTIONAL)); + request.addField(getKinesisStreamARNDefinition()); request.addField(new NumberField( KinesisTransport.CK_KINESIS_RECORD_BATCH_SIZE, "Kinesis Record batch size.", @@ -190,7 +194,27 @@ public class AWSInput extends MessageInput { ConfigurationField.Optional.OPTIONAL, NumberField.Attribute.ONLY_POSITIVE)); + request.addField(getOverrideSourceFieldDefinition()); + return request; } } + + public static TextField getOverrideSourceFieldDefinition() { + return new TextField( + CK_OVERRIDE_SOURCE, + "Override Source", + "", + "The message source is set to '%s' by default. If desired, you may override it with a custom value.".formatted(KinesisRawLogCodec.SOURCE), + ConfigurationField.Optional.OPTIONAL); + } + + public static TextField getKinesisStreamARNDefinition() { + return new TextField( + KinesisTransport.CK_KINESIS_STREAM_ARN, + "Kinesis Stream ARN", + "", + "The ARN of the Kinesis stream.", + ConfigurationField.Optional.OPTIONAL); + } } diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/AWSResource.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/AWSResource.java index 6eb021f0a0..50a921c372 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/AWSResource.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/AWSResource.java @@ -23,6 +23,7 @@ import io.swagger.annotations.ApiParam; import jakarta.inject.Inject; import jakarta.validation.Valid; import jakarta.validation.constraints.NotNull; +import jakarta.ws.rs.BadRequestException; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.DefaultValue; import jakarta.ws.rs.GET; @@ -38,7 +39,8 @@ import org.graylog.integrations.audit.IntegrationsAuditEventTypes; import org.graylog.integrations.aws.AWSPermissions; 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.KinesisRequest; +import org.graylog.integrations.aws.resources.responses.CreateLogSubscriptionResponse; import org.graylog.integrations.aws.resources.responses.KinesisHealthCheckResponse; import org.graylog.integrations.aws.resources.responses.LogGroupsResponse; import org.graylog.integrations.aws.resources.responses.RegionsResponse; @@ -53,6 +55,7 @@ import org.graylog2.plugin.rest.PluginRestResource; import org.graylog2.rest.resources.system.inputs.AbstractInputsResource; import org.graylog2.shared.inputs.MessageInputFactory; import org.graylog2.shared.security.RestPermissions; +import org.graylog2.shared.utilities.ExceptionUtils; import java.io.IOException; import java.util.concurrent.ExecutionException; @@ -106,6 +109,23 @@ public class AWSResource extends AbstractInputsResource implements PluginRestRes return kinesisService.getKinesisStreamNames(request); } + @POST + @Timed + @Path("/kinesis/stream_arn") + @ApiOperation(value = "Get stream ARN for the specified stream and region.") + @RequiresPermissions(AWSPermissions.AWS_READ) + @NoAuditEvent("This does not change any data") + public Response getStreamArn(@ApiParam(name = "JSON body", required = true) @Valid @NotNull KinesisRequest request) { + String response; + try { + response = kinesisService.getStreamArn(request); + } catch (IllegalArgumentException e) { + throw new BadRequestException(ExceptionUtils.formatMessageCause(e)); + } + final CreateLogSubscriptionResponse createLogSubscriptionResponse = CreateLogSubscriptionResponse.create(response); + return Response.ok().entity(createLogSubscriptionResponse).build(); + } + @POST @Timed @Path("/kinesis/health_check") @@ -115,7 +135,7 @@ public class AWSResource extends AbstractInputsResource implements PluginRestRes ) @RequiresPermissions(AWSPermissions.AWS_READ) @NoAuditEvent("This does not change any data") - public Response kinesisHealthCheck(@ApiParam(name = "JSON body", required = true) @Valid @NotNull KinesisHealthCheckRequest heathCheckRequest) throws ExecutionException, IOException { + public Response kinesisHealthCheck(@ApiParam(name = "JSON body", required = true) @Valid @NotNull KinesisRequest heathCheckRequest) throws ExecutionException, IOException { KinesisHealthCheckResponse response = kinesisService.healthCheck(heathCheckRequest); return Response.accepted().entity(response).build(); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/AWSInputCreateRequest.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/AWSInputCreateRequest.java index 21f97e7126..6eab03d390 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/AWSInputCreateRequest.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/AWSInputCreateRequest.java @@ -34,9 +34,11 @@ public abstract class AWSInputCreateRequest implements AWSRequest { private static final String NAME = "name"; private static final String AWS_MESSAGE_TYPE = "aws_input_type"; private static final String STREAM_NAME = "stream_name"; + private static final String KINESIS_STREAM_ARN = "kinesis_stream_arn"; private static final String BATCH_SIZE = "batch_size"; private static final String THROTTLING_ALLOWED = "enable_throttling"; private static final String ADD_FLOW_LOG_PREFIX = "add_flow_log_prefix"; + private static final String OVERRIDE_SOURCE = "override_source"; @JsonProperty(NAME) public abstract String name(); @@ -47,12 +49,18 @@ public abstract class AWSInputCreateRequest implements AWSRequest { @JsonProperty(STREAM_NAME) public abstract String streamName(); + @JsonProperty(KINESIS_STREAM_ARN) + public abstract String streamArn(); + @JsonProperty(BATCH_SIZE) public abstract int batchSize(); @JsonProperty(THROTTLING_ALLOWED) public abstract boolean throttlingAllowed(); + @JsonProperty(OVERRIDE_SOURCE) + public abstract String overrideSource(); + @JsonProperty(ADD_FLOW_LOG_PREFIX) public abstract boolean addFlowLogPrefix(); @@ -76,12 +84,18 @@ public abstract class AWSInputCreateRequest implements AWSRequest { @JsonProperty(STREAM_NAME) public abstract Builder streamName(String streamName); + @JsonProperty(KINESIS_STREAM_ARN) + public abstract Builder streamArn(String streamArn); + @JsonProperty(BATCH_SIZE) public abstract Builder batchSize(int batchSize); @JsonProperty(THROTTLING_ALLOWED) public abstract Builder throttlingAllowed(boolean throttlingAllowed); + @JsonProperty(OVERRIDE_SOURCE) + public abstract Builder overrideSource(String overrideSource); + @JsonProperty(ADD_FLOW_LOG_PREFIX) public abstract Builder addFlowLogPrefix(boolean addFlowLogPrefix); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisHealthCheckRequest.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisRequest.java similarity index 84% rename from graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisHealthCheckRequest.java rename to graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisRequest.java index 99393e03bc..c4530927b6 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisHealthCheckRequest.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisRequest.java @@ -24,8 +24,8 @@ import org.graylog.autovalue.WithBeanGetter; @AutoValue @WithBeanGetter -@JsonDeserialize(builder = KinesisHealthCheckRequest.Builder.class) -public abstract class KinesisHealthCheckRequest implements AWSRequest { +@JsonDeserialize(builder = KinesisRequest.Builder.class) +public abstract class KinesisRequest implements AWSRequest { private static final String STREAM_NAME = "stream_name"; @@ -40,12 +40,12 @@ public abstract class KinesisHealthCheckRequest implements AWSRequest { public static abstract class Builder implements AWSRequest.Builder { @JsonCreator public static Builder create() { - return new AutoValue_KinesisHealthCheckRequest.Builder(); + return new AutoValue_KinesisRequest.Builder(); } @JsonProperty(STREAM_NAME) public abstract Builder streamName(String streamName); - public abstract KinesisHealthCheckRequest build(); + public abstract KinesisRequest build(); } -} \ No newline at end of file +} diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/service/AWSService.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/service/AWSService.java index 1957d99ae3..d5af0ee8dc 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/service/AWSService.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/service/AWSService.java @@ -131,10 +131,12 @@ public class AWSService { configuration.put(AWSInput.CK_DYNAMODB_ENDPOINT, request.dynamodbEndpoint()); configuration.put(AWSInput.CK_IAM_ENDPOINT, request.iamEndpoint()); configuration.put(AWSInput.CK_KINESIS_ENDPOINT, request.kinesisEndpoint()); + configuration.put(AWSInput.CK_OVERRIDE_SOURCE, request.overrideSource()); AWSMessageType inputType = AWSMessageType.valueOf(request.awsMessageType()); if (inputType.isKinesis()) { configuration.put(KinesisTransport.CK_KINESIS_STREAM_NAME, request.streamName()); + configuration.put(KinesisTransport.CK_KINESIS_STREAM_ARN, request.streamArn()); configuration.put(KinesisTransport.CK_KINESIS_RECORD_BATCH_SIZE, request.batchSize()); } else { throw new Exception("The specified input type is not supported."); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/service/KinesisService.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/service/KinesisService.java index 6799f84fee..4366b4921d 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/service/KinesisService.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/service/KinesisService.java @@ -34,8 +34,8 @@ import org.graylog.integrations.aws.cloudwatch.CloudWatchLogSubscriptionData; import org.graylog.integrations.aws.cloudwatch.KinesisLogEntry; import org.graylog.integrations.aws.resources.requests.AWSRequest; import org.graylog.integrations.aws.resources.requests.CreateRolePermissionRequest; -import org.graylog.integrations.aws.resources.requests.KinesisHealthCheckRequest; import org.graylog.integrations.aws.resources.requests.KinesisNewStreamRequest; +import org.graylog.integrations.aws.resources.requests.KinesisRequest; import org.graylog.integrations.aws.resources.responses.CreateRolePermissionResponse; import org.graylog.integrations.aws.resources.responses.KinesisHealthCheckResponse; import org.graylog.integrations.aws.resources.responses.KinesisNewStreamResponse; @@ -57,6 +57,7 @@ import software.amazon.awssdk.services.kinesis.KinesisClient; import software.amazon.awssdk.services.kinesis.KinesisClientBuilder; import software.amazon.awssdk.services.kinesis.model.CreateStreamRequest; import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest; +import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse; import software.amazon.awssdk.services.kinesis.model.GetRecordsRequest; import software.amazon.awssdk.services.kinesis.model.GetRecordsResponse; import software.amazon.awssdk.services.kinesis.model.GetShardIteratorRequest; @@ -74,6 +75,7 @@ import software.amazon.awssdk.services.kinesis.model.StreamStatus; import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.HashMap; import java.util.List; import java.util.Locale; import java.util.Map; @@ -134,7 +136,7 @@ public class KinesisService { * @return a {@code KinesisHealthCheckResponse}, which indicates the type of detected message and a sample parsed * message. */ - public KinesisHealthCheckResponse healthCheck(KinesisHealthCheckRequest request) throws ExecutionException, IOException { + public KinesisHealthCheckResponse healthCheck(KinesisRequest request) throws ExecutionException, IOException { LOG.debug("Executing healthCheck"); LOG.debug("Requesting a list of streams to find out if the indicated stream exists."); @@ -153,8 +155,14 @@ public class KinesisService { KinesisClient kinesisClient = awsClientBuilderUtil.buildClient(kinesisClientBuilder, request); final List records = retrieveRecords(request.streamName(), kinesisClient); - if (records.size() == 0) { - throw new BadRequestException(String.format(Locale.ROOT, "The Kinesis stream [%s] does not contain any messages.", request.streamName())); + + if (records.isEmpty()) { + LOG.warn("The Kinesis stream [{}] is empty. Proceeding with setup at the user's own risk.", request.streamName()); + return KinesisHealthCheckResponse.create( + AWSMessageType.KINESIS_RAW, + String.format(Locale.ROOT, "The Kinesis stream [%s] does not contain any messages.", request.streamName()), + new HashMap<>() + ); } Record record = selectRandomRecord(records); @@ -165,7 +173,7 @@ public class KinesisService { } DateTime timestamp = new DateTime(record.approximateArrivalTimestamp().toEpochMilli(), DateTimeZone.UTC); - return detectAndParseMessage(new String(payloadBytes, StandardCharsets.UTF_8), timestamp, request.streamName(), "", "", compressed); + return detectAndParseMessage(new String(payloadBytes, StandardCharsets.UTF_8), timestamp, request.streamName(), "test-graylog", "", compressed, request); } public StreamsResponse getKinesisStreamNames(AWSRequest request) throws ExecutionException { @@ -221,7 +229,7 @@ public class KinesisService { * message. * @see */ - private KinesisHealthCheckResponse handleCompressedMessages(KinesisHealthCheckRequest request, byte[] payloadBytes) throws IOException { + private KinesisHealthCheckResponse handleCompressedMessages(KinesisRequest request, byte[] payloadBytes) throws IOException { LOG.debug("The supplied payload is GZip compressed. Proceeding to decompress."); // Assume that the payload is from CloudWatch. @@ -238,7 +246,7 @@ public class KinesisService { CloudWatchLogEvent logEntry = logEntryOptional.get(); DateTime timestamp = new DateTime(logEntry.timestamp(), DateTimeZone.UTC); return detectAndParseMessage(logEntry.message(), timestamp, - request.streamName(), data.logGroup(), data.logStream(), true); + request.streamName(), data.logGroup(), data.logStream(), true, request); } /** @@ -327,7 +335,7 @@ public class KinesisService { * @return A {@code KinesisHealthCheckResponse} with the fully parsed message and type. */ private KinesisHealthCheckResponse detectAndParseMessage(String logMessage, DateTime timestamp, String kinesisStreamName, - String logGroupName, String logStreamName, boolean compressed) { + String logGroupName, String logStreamName, boolean compressed, KinesisRequest request) { LOG.debug("Attempting to detect the type of log message. message [{}] stream [{}] log group [{}].", logMessage, kinesisStreamName, logGroupName); @@ -340,7 +348,7 @@ public class KinesisService { final String responseMessage = String.format(Locale.ROOT, "Success. The message is a %s message.", awsMessageType.getLabel()); final KinesisLogEntry logEvent = KinesisLogEntry.create(kinesisStreamName, logGroupName, logStreamName, - timestamp, logMessage); + timestamp, logMessage, "", awsMessageType.getLabel(), new ArrayList<>()); final Codec.Factory codecFactory = this.availableCodecs.get(awsMessageType.getCodecName()); if (codecFactory == null) { @@ -534,4 +542,19 @@ public class KinesisService { LOG.debug("Acquiring the role ARN associated to the role [{}]", roleName); return iamClient.getRole(r -> r.roleName(roleName)).role().arn(); } + + public String getStreamArn(KinesisRequest arnRequest) { + try (KinesisClient kinesisClient = awsClientBuilderUtil.buildClient(kinesisClientBuilder, arnRequest)) { + DescribeStreamRequest request = DescribeStreamRequest.builder() + .streamName(arnRequest.streamName()) + .build(); + + DescribeStreamResponse response = kinesisClient.describeStream(request); + StreamDescription description = response.streamDescription(); + return description.streamARN(); + } catch (Exception e) { + throw new IllegalArgumentException("Failed to get stream ARN. Please " + + "ensure the IAM role includes the 'kinesis:DescribeStream' permission.", e); + } + } } diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoder.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoder.java index 072fff3fc8..9d6f5d8cd0 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoder.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoder.java @@ -18,6 +18,7 @@ package org.graylog.integrations.aws.transports; import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.inject.Inject; import org.graylog.integrations.aws.AWSMessageType; import org.graylog.integrations.aws.cloudwatch.CloudWatchLogSubscriptionData; import org.graylog.integrations.aws.cloudwatch.KinesisLogEntry; @@ -27,8 +28,6 @@ import org.joda.time.DateTimeZone; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import jakarta.inject.Inject; - import java.io.IOException; import java.nio.charset.StandardCharsets; import java.time.Instant; @@ -74,7 +73,6 @@ public class KinesisPayloadDecoder { * @throws IOException */ List processMessages(final byte[] payloadBytes, Instant approximateArrivalTimestamp) throws IOException { - // This method will be called from a codec, and therefore will not perform any detection. It will rely // exclusively on the AWSMessageType detected in the setup HealthCheck. // If a user needs to change the type of data stored in a stream, they will need to set the integration up again. @@ -86,7 +84,12 @@ public class KinesisPayloadDecoder { return KinesisLogEntry.create(kinesisStream, // Use the log group and stream returned from CloudWatch. logSubscriptionData.logGroup(), - logSubscriptionData.logStream(), timestamp, le.message()); + logSubscriptionData.logStream(), + timestamp, + le.message(), + logSubscriptionData.owner(), + logSubscriptionData.messageType(), + logSubscriptionData.subscriptionFilters()); }) .collect(Collectors.toList()); } else if (awsMessageType == AWSMessageType.KINESIS_RAW) { @@ -94,7 +97,7 @@ public class KinesisPayloadDecoder { final DateTime timestamp = new DateTime(approximateArrivalTimestamp.toEpochMilli(), DateTimeZone.UTC); final KinesisLogEntry kinesisLogEntry = KinesisLogEntry.create(kinesisStream, "", "", - timestamp, new String(payloadBytes, StandardCharsets.UTF_8)); + timestamp, new String(payloadBytes, StandardCharsets.UTF_8), "", awsMessageType.getLabel(), new ArrayList<>()); return Collections.singletonList(kinesisLogEntry); } else { LOG.error("The AWSMessageType [{}] is not supported by the KinesisTransport", awsMessageType); diff --git a/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisTransport.java b/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisTransport.java index 34204d33fc..cad4fff0a4 100644 --- a/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisTransport.java +++ b/graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisTransport.java @@ -21,6 +21,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.google.common.eventbus.EventBus; import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.inject.assistedinject.Assisted; +import jakarta.inject.Inject; import org.apache.commons.lang3.StringUtils; import org.graylog.integrations.aws.AWSClientBuilderUtil; import org.graylog.integrations.aws.AWSMessageType; @@ -52,8 +53,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import software.amazon.awssdk.regions.Region; -import jakarta.inject.Inject; - import java.net.URL; import java.util.Locale; import java.util.Objects; @@ -61,7 +60,10 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.function.Consumer; +import static org.graylog.integrations.aws.inputs.AWSInput.getKinesisStreamARNDefinition; + public class KinesisTransport extends ThrottleableTransport2 { + private static final Logger LOG = LoggerFactory.getLogger(KinesisTransport.class); public static final String NAME = "aws-kinesis-transport"; @@ -69,6 +71,7 @@ public class KinesisTransport extends ThrottleableTransport2 { private static final String CK_ACCESS_KEY = "aws_access_key"; private static final String CK_SECRET_KEY = "aws_secret_key"; public static final String CK_KINESIS_STREAM_NAME = "kinesis_stream_name"; + public static final String CK_KINESIS_STREAM_ARN = "kinesis_stream_arn"; public static final String CK_KINESIS_RECORD_BATCH_SIZE = "kinesis_record_batch_size"; public static final int DEFAULT_BATCH_SIZE = 10000; @@ -242,6 +245,7 @@ public class KinesisTransport extends ThrottleableTransport2 { "The name of the Kinesis stream that receives your messages. See README for instructions on how to connect messages to a Kinesis Stream.", ConfigurationField.Optional.NOT_OPTIONAL )); + r.addField(getKinesisStreamARNDefinition()); r.addField(new NumberField( CK_KINESIS_RECORD_BATCH_SIZE, diff --git a/graylog2-server/src/main/resources/log4j2.xml b/graylog2-server/src/main/resources/log4j2.xml index 9c947ac23a..f59948e739 100644 --- a/graylog2-server/src/main/resources/log4j2.xml +++ b/graylog2-server/src/main/resources/log4j2.xml @@ -52,6 +52,8 @@ + + diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/cloudwatch/FlowLogMessageTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/cloudwatch/FlowLogMessageTest.java index 02ade3fcb7..179369f597 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/cloudwatch/FlowLogMessageTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/cloudwatch/FlowLogMessageTest.java @@ -20,10 +20,14 @@ import org.joda.time.DateTime; import org.joda.time.DateTimeZone; import org.junit.Test; +import java.util.ArrayList; + import static org.junit.Assert.assertEquals; public class FlowLogMessageTest { + public static final String TEST_STREAM_ARN = "arn:aws:kinesis:ap-northeast-1:517803882956:stream/graylog2"; + @Test public void testFromPartsDoesNotFailWithMissingIntegerFields() { final String[] strings = { @@ -44,8 +48,8 @@ public class FlowLogMessageTest { }; final KinesisLogEntry logEvent = KinesisLogEntry.create("kinesisStream", "helloGroup", "helloStream", - DateTime.now(DateTimeZone.UTC), - String.join(" ", strings)); + DateTime.now(DateTimeZone.UTC), + String.join(" ", strings), "123456789", TEST_STREAM_ARN, new ArrayList<>()); final FlowLogMessage m = FlowLogMessage.fromLogEvent(logEvent); assertEquals(m.getDestinationPort(), 0); @@ -74,11 +78,11 @@ public class FlowLogMessageTest { }; final KinesisLogEntry logEvent = KinesisLogEntry.create("kinesisStream", "helloGroup", "helloStream", - DateTime.now(DateTimeZone.UTC), - String.join(" ", strings)); + DateTime.now(DateTimeZone.UTC), + String.join(" ", strings), "123456789", TEST_STREAM_ARN, new ArrayList<>()); final FlowLogMessage m = FlowLogMessage.fromLogEvent(logEvent); assertEquals(m.getBytes(), 0); assertEquals(m.getPackets(), 0); } -} \ No newline at end of file +} diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/AWSCodecTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/AWSCodecTest.java index ba22558008..bf9dae5838 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/AWSCodecTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/AWSCodecTest.java @@ -31,6 +31,7 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import java.util.ArrayList; import java.util.HashMap; public class AWSCodecTest { @@ -52,7 +53,7 @@ public class AWSCodecTest { DateTime timestamp = DateTime.now(DateTimeZone.UTC); KinesisLogEntry kinesisLogEntry = KinesisLogEntry.create("a-stream", "log-group", "log-stream", timestamp, - "2 423432432432 eni-3244234 172.1.1.2 172.1.1.2 80 2264 6 1 52 1559738144 1559738204 ACCEPT OK"); + "2 423432432432 eni-3244234 172.1.1.2 172.1.1.2 80 2264 6 1 52 1559738144 1559738204 ACCEPT OK", "123456789", "", new ArrayList<>()); Message message = codec.decodeSafe(new RawMessage(objectMapper.writeValueAsBytes(kinesisLogEntry))).get(); Assert.assertEquals("log-group", message.getField(AbstractKinesisCodec.FIELD_LOG_GROUP)); @@ -87,13 +88,14 @@ public class AWSCodecTest { final DateTime timestamp = DateTime.now(DateTimeZone.UTC); final KinesisLogEntry kinesisLogEntry = KinesisLogEntry.create("a-stream", "log-group", "log-stream", timestamp, - "This a raw message"); + "This a raw message", "123456789", "", new ArrayList<>()); Message message = codec.decodeSafe(new RawMessage(objectMapper.writeValueAsBytes(kinesisLogEntry))).get(); Assert.assertEquals("log-group", message.getField(AbstractKinesisCodec.FIELD_LOG_GROUP)); Assert.assertEquals("log-stream", message.getField(AbstractKinesisCodec.FIELD_LOG_STREAM)); Assert.assertEquals("a-stream", message.getField(AbstractKinesisCodec.FIELD_KINESIS_STREAM)); - Assert.assertEquals(KinesisRawLogCodec.SOURCE, message.getField("source")); + Assert.assertEquals("123456789", message.getField("aws_owner")); + Assert.assertEquals("aws-kinesis-raw-logs", message.getField("source")); Assert.assertEquals("This a raw message", message.getField("message")); Assert.assertEquals(timestamp, message.getTimestamp()); } diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/CloudWatchFlowLogCodecTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/CloudWatchFlowLogCodecTest.java index 4cd8f5ef3b..3e258a97c1 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/CloudWatchFlowLogCodecTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/CloudWatchFlowLogCodecTest.java @@ -28,6 +28,8 @@ import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import java.util.ArrayList; + public class CloudWatchFlowLogCodecTest { private KinesisCloudWatchFlowLogCodec codec; @@ -49,7 +51,7 @@ public class CloudWatchFlowLogCodecTest { final String flowLogMessage = "2 423432432432 eni-3244234 172.1.1.2 172.1.1.2 80 2264 6 1 52 1559738144 1559738204 ACCEPT OK"; final DateTime timestamp = DateTime.now(DateTimeZone.UTC); final KinesisLogEntry logEvent = KinesisLogEntry.create("a-stream", "log-group", "log-stream", - timestamp, flowLogMessage); + timestamp, flowLogMessage, "123456789", "TEST_STREAM_ARN", new ArrayList<>()); final Message message = codec.decodeLogData(logEvent).get(); Assert.assertEquals("log-group", message.getField(AbstractKinesisCodec.FIELD_LOG_GROUP)); Assert.assertEquals("log-stream", message.getField(AbstractKinesisCodec.FIELD_LOG_STREAM)); diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/service/AWSServiceTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/service/AWSServiceTest.java index 2f7771899a..f57b0180d0 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/service/AWSServiceTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/service/AWSServiceTest.java @@ -89,13 +89,15 @@ public class AWSServiceTest { AWSInputCreateRequest.builder().region(Region.US_EAST_1.id()) .awsAccessKeyId("a-key") .awsSecretAccessKey(encryptedValue) - .name("AWS Input") - .awsMessageType(AWSMessageType.KINESIS_CLOUDWATCH_FLOW_LOGS.toString()) - .streamName("a-stream") - .batchSize(10000) - .addFlowLogPrefix(true) - .throttlingAllowed(true) - .build(); + .name("AWS Input") + .awsMessageType(AWSMessageType.KINESIS_CLOUDWATCH_FLOW_LOGS.toString()) + .streamName("a-stream") + .batchSize(10000) + .addFlowLogPrefix(true) + .throttlingAllowed(true) + .streamArn("test-arn") + .overrideSource("test-source") + .build(); awsService.saveInput(request, user); // Verify that inputService received a valid input to save. diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/service/KinesisServiceTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/service/KinesisServiceTest.java index 079d66609c..11ec65cd99 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/service/KinesisServiceTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/service/KinesisServiceTest.java @@ -22,7 +22,7 @@ import org.graylog.integrations.aws.AWSLogMessage; import org.graylog.integrations.aws.AWSMessageType; import org.graylog.integrations.aws.AWSTestingUtils; import org.graylog.integrations.aws.resources.requests.AWSRequestImpl; -import org.graylog.integrations.aws.resources.requests.KinesisHealthCheckRequest; +import org.graylog.integrations.aws.resources.requests.KinesisRequest; import org.graylog.integrations.aws.resources.requests.KinesisNewStreamRequest; import org.graylog.integrations.aws.resources.responses.KinesisHealthCheckResponse; import org.graylog.integrations.aws.resources.responses.KinesisNewStreamResponse; @@ -85,7 +85,7 @@ public class KinesisServiceTest { @Rule public MockitoRule mockitoRule = MockitoJUnit.rule(); - + @Mock private KinesisClient kinesisClient; @Mock @@ -138,7 +138,7 @@ public class KinesisServiceTest { assertEquals(AWSMessageType.KINESIS_CLOUDWATCH_FLOW_LOGS, response.inputType()); Map fields = response.messageFields(); assertEquals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP, fields.get("timestamp")); - assertEquals(21, fields.size()); + assertEquals(22, fields.size()); assertEquals(6, fields.get("protocol_number")); assertEquals("TCP", fields.get("protocol")); assertEquals(1L, fields.get("packets")); @@ -154,7 +154,7 @@ public class KinesisServiceTest { assertEquals(AWSMessageType.KINESIS_CLOUDWATCH_RAW, response.inputType()); Map fields = response.messageFields(); assertEquals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP, fields.get("timestamp")); - assertEquals(7, fields.size()); + assertEquals(8, fields.size()); } @Test @@ -167,7 +167,7 @@ public class KinesisServiceTest { assertEquals(AWSMessageType.KINESIS_RAW, response.inputType()); Map fields = response.messageFields(); assertEquals(new DateTime("2000-01-01T01:01:01.000Z", DateTimeZone.UTC), fields.get("timestamp")); - assertEquals(5, fields.size()); + assertEquals(7, fields.size()); } private KinesisHealthCheckResponse executeHealthCheckTest(byte[] payloadData, Instant recordArrivalTime) throws IOException, ExecutionException { @@ -193,7 +193,7 @@ public class KinesisServiceTest { .thenReturn(GetRecordsResponse.builder().records(record).millisBehindLatest(10000L).build()) .thenReturn(GetRecordsResponse.builder().records(record).millisBehindLatest(0L).build()); - KinesisHealthCheckRequest request = KinesisHealthCheckRequest.builder() + KinesisRequest request = KinesisRequest.builder() .region(Region.EU_WEST_1.id()) .awsAccessKeyId("a-key") .awsSecretAccessKey(encryptedValue) @@ -201,6 +201,42 @@ public class KinesisServiceTest { return kinesisService.healthCheck(request); } + @Test + public void healthCheckReturnsWarningWhenStreamIsEmpty() throws ExecutionException, IOException { + + when(awsClientBuilderUtil.buildClient(any(KinesisClientBuilder.class), any())).thenReturn(kinesisClient); + + when(kinesisClient.listStreams(isA(ListStreamsRequest.class))) + .thenReturn(ListStreamsResponse.builder() + .streamNames(TEST_STREAM_1) + .hasMoreStreams(false) + .build()); + + Shard shard = Shard.builder().shardId("shardId-000000000000").build(); + when(kinesisClient.listShards(isA(ListShardsRequest.class))) + .thenReturn(ListShardsResponse.builder().shards(shard).build()); + + when(kinesisClient.getShardIterator(isA(GetShardIteratorRequest.class))) + .thenReturn(GetShardIteratorResponse.builder().shardIterator("shardIterator").build()); + + when(kinesisClient.getRecords(isA(GetRecordsRequest.class))) + .thenReturn(GetRecordsResponse.builder().records(new ArrayList<>()).millisBehindLatest(0L).build()); + + KinesisRequest request = KinesisRequest.builder() + .region(TEST_REGION) + .awsAccessKeyId("dummy-access-key") + .awsSecretAccessKey(encryptedValue) + .streamName(TEST_STREAM_1) + .build(); + + KinesisHealthCheckResponse response = kinesisService.healthCheck(request); + + assertEquals(AWSMessageType.KINESIS_RAW, response.inputType()); + assertNotNull(response.messageFields()); + assertEquals(0, response.messageFields().size()); + } + + @Test public void testGetStreams() throws ExecutionException { diff --git a/graylog2-server/src/test/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoderTest.java b/graylog2-server/src/test/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoderTest.java index 435257b419..bc5659729f 100644 --- a/graylog2-server/src/test/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoderTest.java +++ b/graylog2-server/src/test/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoderTest.java @@ -36,6 +36,7 @@ public class KinesisPayloadDecoderTest { private KinesisPayloadDecoder flowLogDecoder; private KinesisPayloadDecoder rawDecoder; + public static final String TEST_REGION = "ap-northeast-1"; @Before public void setUp() { @@ -54,7 +55,7 @@ public class KinesisPayloadDecoderTest { final List logEntries = flowLogDecoder.processMessages(AWSTestingUtils.cloudWatchFlowLogPayload(), - Instant.ofEpochMilli(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP.getMillis())); + Instant.ofEpochMilli(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP.getMillis())); Assert.assertEquals(2, logEntries.size()); @@ -66,8 +67,8 @@ public class KinesisPayloadDecoderTest { // Verify that both messages have to correct timestamp. Assert.assertEquals(2, logEntries.stream() - .filter(logEntry -> logEntry.timestamp().equals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP)) - .count()); + .filter(logEntry -> logEntry.timestamp().equals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP)) + .count()); } @Test @@ -85,8 +86,8 @@ public class KinesisPayloadDecoderTest { // Verify that both messages have to correct timestamp. Assert.assertEquals(2, logEntries.stream() - .filter(logEntry -> logEntry.timestamp().equals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP)) - .count()); + .filter(logEntry -> logEntry.timestamp().equals(AWSTestingUtils.CLOUD_WATCH_TIMESTAMP)) + .count()); } @Test diff --git a/graylog2-web-interface/src/integrations/aws/cloudwatch/FormAdvancedOptions.tsx b/graylog2-web-interface/src/integrations/aws/cloudwatch/FormAdvancedOptions.tsx index 98f15354d2..9e33b2c0ce 100644 --- a/graylog2-web-interface/src/integrations/aws/cloudwatch/FormAdvancedOptions.tsx +++ b/graylog2-web-interface/src/integrations/aws/cloudwatch/FormAdvancedOptions.tsx @@ -30,7 +30,7 @@ const FormAdvancedOptions = ({ onChange }: FormAdvancedOptionsProps) => { const { formData } = useContext(FormDataContext); const { isAdvancedOptionsVisible, setAdvancedOptionsVisibility } = useContext(AdvancedOptionsContext); - const { awsCloudWatchBatchSize, awsCloudWatchThrottleEnabled, awsCloudWatchAddFlowLogPrefix } = formData; + const { awsCloudWatchBatchSize, overrideSource, awsCloudWatchThrottleEnabled, awsCloudWatchAddFlowLogPrefix } = formData; const handleToggle = (visible) => { setAdvancedOptionsVisibility(visible); @@ -54,6 +54,15 @@ const FormAdvancedOptions = ({ onChange }: FormAdvancedOptionsProps) => { help='Add field with the Flow Log prefix e. g. "src_addr" -> "flow_log_src_addr".' /> + + . */ -import React, { useContext, useEffect, useState, useCallback } from 'react'; +import React, {useContext, useEffect, useState, useCallback} from 'react'; import styled from 'styled-components'; -import { Button, Panel, Input } from 'components/bootstrap'; +import {Button, Panel, Input} from 'components/bootstrap'; import FormWrap from 'integrations/aws/common/FormWrap'; import SkipHealthCheck from 'integrations/aws/common/SkipHealthCheck'; import useFetch from 'integrations/aws/common/hooks/useFetch'; -import { ApiRoutes } from 'integrations/aws/common/Routes'; +import {ApiRoutes} from 'integrations/aws/common/Routes'; import Countdown from 'integrations/aws/common/Countdown'; -import { DEFAULT_KINESIS_LOG_TYPE, KINESIS_LOG_TYPES } from 'integrations/aws/common/constants'; -import { ApiContext } from 'integrations/aws/context/Api'; -import { FormDataContext } from 'integrations/aws/context/FormData'; +import {KINESIS_LOG_TYPES} from 'integrations/aws/common/constants'; +import {ApiContext} from 'integrations/aws/context/Api'; +import {FormDataContext} from 'integrations/aws/context/FormData'; import Icon from 'components/common/Icon'; -import useProductName from 'brand-customization/useProductName'; const Notice = styled.span` display: flex; @@ -52,17 +51,38 @@ type StepHealthCheckProps = { onChange: (...args: any[]) => void; }; -const StepHealthCheck = ({ onChange, onSubmit }: StepHealthCheckProps) => { - const { logData, setLogData } = useContext(ApiContext); - const { formData } = useContext(FormDataContext); +const StepHealthCheck = ({onChange, onSubmit}: StepHealthCheckProps) => { + const {logData, setLogData} = useContext(ApiContext); + const {formData} = useContext(FormDataContext); const [pauseCountdown, setPauseCountdown] = useState(false); - const productName = useProductName(); + + const [fetchStreamArnStatus, setStreamArnFetch] = useFetch( + null, + (response) => { + onChange({target: {name: 'awsCloudwatchKinesisStreamArn', value: response.result}}); + }, + 'POST', + { + region: formData.awsCloudWatchAwsRegion.value, + stream_name: formData.awsCloudWatchKinesisStream.value, + }, + ); + + useEffect(() => { + setStreamArnFetch(ApiRoutes.INTEGRATIONS.AWS.KINESIS.STREAM_ARN); + }, [setStreamArnFetch]); + + useEffect(() => { + if (fetchStreamArnStatus.error) { + // Proceed even if fetching the stream ARN fails — failure is non-blocking and intentionally ignored + } + }, [fetchStreamArnStatus]); const [logDataProgress, setLogDataUrl] = useFetch( null, (response) => { setLogData(response); - onChange({ target: { name: 'awsCloudWatchKinesisInputType', value: response.type } }); + onChange({target: {name: 'awsCloudWatchKinesisInputType', value: response.type}}); }, 'POST', { @@ -74,7 +94,7 @@ const StepHealthCheck = ({ onChange, onSubmit }: StepHealthCheckProps) => { const checkForLogs = useCallback(() => { setPauseCountdown(true); setLogDataUrl(ApiRoutes.INTEGRATIONS.AWS.KINESIS.HEALTH_CHECK); - }, [setLogDataUrl]); + }, [setPauseCountdown, setLogDataUrl]); useEffect(() => { if (!logData) { @@ -127,18 +147,11 @@ const StepHealthCheck = ({ onChange, onSubmit }: StepHealthCheckProps) => { ); } - - const knownLog = logData.type === DEFAULT_KINESIS_LOG_TYPE; - const iconName = knownLog ? 'check_circle' : 'warning'; - const acknowledgment = knownLog ? 'Awesome!' : 'Drats!'; - const bsStyle = knownLog ? 'success' : 'warning'; - const logTypeLabel = KINESIS_LOG_TYPES.find((type) => type.value === logData.type).label; - const logType = knownLog ? `a ${logTypeLabel}` : 'an unknown'; - const handleSubmit = () => { onSubmit(); - onChange({ target: { name: 'awsCloudWatchKinesisInputType', value: logData.type } }); + onChange({target: {name: 'awsCloudWatchKinesisInputType', value: logData.type}}); }; + const logTypeLabel = KINESIS_LOG_TYPES.find((type) => type.value === logData.type).label; return ( { disabled={false} title="Create Kinesis Stream" description={ -

- We are going to attempt to parse a single log to help you out! If we are unable to, or you would like it - parsed differently, head on over to Pipeline Rules to set up your own parser! -

+

If available, a parsed sample {logTypeLabel} message from the stream will be shown below.

}> - - - - {acknowledgment} looks like {logType} message type. - - - }> - {knownLog - ? 'Take a look at what we have parsed so far and you can create Pipeline Rules to handle even more!' - : `Not to worry, ${productName} can still read in these log messages. We have parsed what we could and you can build Pipeline Rules to do the rest!`} - diff --git a/graylog2-web-interface/src/integrations/aws/cloudwatch/StepReview.tsx b/graylog2-web-interface/src/integrations/aws/cloudwatch/StepReview.tsx index b44a08066a..d1225794a5 100644 --- a/graylog2-web-interface/src/integrations/aws/cloudwatch/StepReview.tsx +++ b/graylog2-web-interface/src/integrations/aws/cloudwatch/StepReview.tsx @@ -28,16 +28,6 @@ import FormWrap from 'integrations/aws/common/FormWrap'; import { ApiRoutes } from 'integrations/aws/common/Routes'; import { DEFAULT_KINESIS_LOG_TYPE, KINESIS_LOG_TYPES } from 'integrations/aws/common/constants'; -type DefaultProps = { - value: string; -}; - -const Default = ({ value }: DefaultProps) => ( - <> - {value} (default) - -); - const Container = styled.div` border: 1px solid #a6afbd; margin: 25px 0; @@ -83,6 +73,24 @@ const EditAnchor = styled.a` } `; +const ArnErrorMessage = styled.span` + color: #856404; + background-color: #fff3cd; + padding: 4px 8px; + border-radius: 4px; + display: inline-block; +`; + +type DefaultProps = { + value: string; +}; + +const Default = ({ value }: DefaultProps) => ( + <> + {value} (default) + +); + type StepReviewProps = { onSubmit: (...args: any[]) => void; onEditClick: (...args: any[]) => (...args: any[]) => void; @@ -108,10 +116,12 @@ const StepReview = ({ onSubmit, onEditClick, externalInputSubmit = false }: Step awsEndpointDynamoDB = { value: undefined }, awsEndpointIAM = { value: undefined }, awsEndpointKinesis = { value: undefined }, + overrideSource = {value: undefined}, } = formData; const throttleEnabled = !!awsCloudWatchThrottleEnabled.value; const addPrefix = !!awsCloudWatchAddFlowLogPrefix.value; + const awsCloudwatchKinesisStreamArn = formData.awsCloudwatchKinesisStreamArn?.value ?? ''; const [fetchSubmitStatus, setSubmitFetch] = useFetch( null, @@ -127,6 +137,8 @@ const StepReview = ({ onSubmit, onEditClick, externalInputSubmit = false }: Step batch_size: Number(awsCloudWatchBatchSize.value || awsCloudWatchBatchSize.defaultValue), enable_throttling: throttleEnabled, add_flow_log_prefix: addPrefix, + kinesis_stream_arn: awsCloudwatchKinesisStreamArn, + override_source: overrideSource?.value ?? '', }, ); @@ -231,9 +243,22 @@ const StepReview = ({ onSubmit, onEditClick, externalInputSubmit = false }: Step
  • - Stream + Kinesis Stream {awsCloudWatchKinesisStream.value}
    • +
  • + Kinesis Stream ARN + + {!awsCloudwatchKinesisStreamArn ? ( + + Error: Failed to get stream ARN. Please ensure the IAM role includes the kinesis:DescribeStream permission. + + ) : ( + awsCloudwatchKinesisStreamArn + )} + +
    • +
  • Global Input @@ -262,6 +287,10 @@ const StepReview = ({ onSubmit, onEditClick, externalInputSubmit = false }: Step
    • + {overrideSource.value && (
  • + Override Source + {overrideSource.value} +
    • )}     Formatting diff --git a/graylog2-web-interface/src/integrations/aws/cloudwatch/_initialFormData.js b/graylog2-web-interface/src/integrations/aws/cloudwatch/_initialFormData.js index fd7dba42a4..54ca80b744 100644 --- a/graylog2-web-interface/src/integrations/aws/cloudwatch/_initialFormData.js +++ b/graylog2-web-interface/src/integrations/aws/cloudwatch/_initialFormData.js @@ -32,6 +32,9 @@ const DEFAULT_SETTINGS = { awsCloudWatchAddFlowLogPrefix: { value: true, // We want to default to true on render, but never compare the default }, + overrideSource: { + value: '', + }, }; export default DEFAULT_SETTINGS; diff --git a/graylog2-web-interface/src/integrations/aws/common/FormWrap.tsx b/graylog2-web-interface/src/integrations/aws/common/FormWrap.tsx index a69cd7e995..d8e09c947d 100644 --- a/graylog2-web-interface/src/integrations/aws/common/FormWrap.tsx +++ b/graylog2-web-interface/src/integrations/aws/common/FormWrap.tsx @@ -15,7 +15,7 @@ * . */ import React, { useEffect, useRef, useState } from 'react'; -import styled, { createGlobalStyle, css } from 'styled-components'; +import styled, { createGlobalStyle, css, useTheme } from 'styled-components'; import { Button, Panel } from 'components/bootstrap'; import Icon from 'components/common/Icon'; @@ -25,6 +25,21 @@ type ErrorMessageProps = { niceMessage?: string | React.ReactNode; }; +type FormWrapProps = { + buttonContent?: string | React.ReactNode; + children: any; + disabled?: boolean; + error?: { + full_message: string; + nice_message?: string | React.ReactNode; + }; + description?: string | React.ReactNode; + loading?: boolean; + onSubmit?: (...args: any[]) => void; + title?: string | React.ReactNode; + className?: string; +}; + const ErrorOutputStyle = createGlobalStyle` /* NOTE: This is to remove Bootstrap styles from the anchor element I can't override in Panel.Header */ form { @@ -46,10 +61,10 @@ const ErrorOutput = styled.span` display: block; `; -const ErrorToggleInfo = styled.button` +const ErrorToggleInfo = styled.button<{ isDarkMode: boolean }>` border: 0; background: none; - color: #1f1f1f; + color:${({ isDarkMode }) => (isDarkMode ? 'white' : 'black')}; font-size: 11px; text-transform: uppercase; margin: 12px 0 0; @@ -65,13 +80,15 @@ const MoreIcon = styled(Icon)<{ expanded: boolean }>( export const ErrorMessage = ({ fullMessage, niceMessage = null }: ErrorMessageProps) => { const [expanded, toggleExpanded] = useState(false); + const theme = useTheme(); + const isDarkMode = theme.mode === 'dark'; const Header = ( <> {niceMessage || fullMessage} {niceMessage && ( - toggleExpanded(!expanded)}> + toggleExpanded(!expanded)}> More Info )} @@ -90,21 +107,6 @@ export const ErrorMessage = ({ fullMessage, niceMessage = null }: ErrorMessagePr ); }; -type FormWrapProps = { - buttonContent?: string | React.ReactNode; - children: any; - disabled?: boolean; - error?: { - full_message: string; - nice_message?: string | React.ReactNode; - }; - description?: string | React.ReactNode; - loading?: boolean; - onSubmit?: (...args: any[]) => void; - title?: string | React.ReactNode; - className?: string; -}; - const FormWrap = ({ buttonContent = 'Submit', children, diff --git a/graylog2-web-interface/src/integrations/aws/common/Routes.ts b/graylog2-web-interface/src/integrations/aws/common/Routes.ts index 828621d5e3..701e4cd32b 100644 --- a/graylog2-web-interface/src/integrations/aws/common/Routes.ts +++ b/graylog2-web-interface/src/integrations/aws/common/Routes.ts @@ -36,6 +36,7 @@ const ApiRoutes = { KINESIS: { HEALTH_CHECK: '/plugins/org.graylog.integrations/aws/kinesis/health_check', STREAMS: '/plugins/org.graylog.integrations/aws/kinesis/streams', + STREAM_ARN: '/plugins/org.graylog.integrations/aws/kinesis/stream_arn', SAVE: '/plugins/org.graylog.integrations/aws/inputs?setup_wizard=true', }, KINESIS_AUTO_SETUP: { diff --git a/graylog2-web-interface/src/integrations/aws/common/constants.js b/graylog2-web-interface/src/integrations/aws/common/constants.js index 818224c26a..3fa94e76b0 100644 --- a/graylog2-web-interface/src/integrations/aws/common/constants.js +++ b/graylog2-web-interface/src/integrations/aws/common/constants.js @@ -18,6 +18,7 @@ const KINESIS_LOG_TYPES = [ { value: 'KINESIS_CLOUDWATCH_FLOW_LOGS', label: 'Kinesis CloudWatch Flow Logs' }, { value: 'KINESIS_CLOUDWATCH_RAW', label: 'Kinesis CloudWatch Raw' }, { value: 'KINESIS_RAW', label: 'Kinesis Raw' }, + { value: 'UNKNOWN', label: 'Unknown' }, ]; const DEFAULT_KINESIS_LOG_TYPE = 'KINESIS_CLOUDWATCH_FLOW_LOGS'; diff --git a/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.js b/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.js index 34b348a541..dc0cb53e3a 100644 --- a/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.js +++ b/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.js @@ -29,6 +29,7 @@ export const toAWSRequest = (formData, options) => { awsEndpointKinesis, key, secret, + overrideSource, } = formData; return { @@ -46,6 +47,7 @@ export const toAWSRequest = (formData, options) => { dynamodb_endpoint: awsEndpointDynamoDB?.value, iam_endpoint: awsEndpointIAM?.value, kinesis_endpoint: awsEndpointKinesis?.value, + override_source: overrideSource?.value, ...options, }; }; @@ -68,6 +70,7 @@ export const toGenericInputCreateRequest = ({ awsCloudWatchAwsSecret, key, secret, + overrideSource, }) => ({ type: 'org.graylog.integrations.aws.inputs.AWSInput', title: awsCloudWatchName.value, @@ -92,5 +95,6 @@ export const toGenericInputCreateRequest = ({ kinesis_endpoint: awsEndpointKinesis?.value, kinesis_stream_name: awsCloudWatchKinesisStream.value, kinesis_record_batch_size: Number(awsCloudWatchBatchSize.value || awsCloudWatchBatchSize.defaultValue), + override_source: overrideSource?.value, }, }); diff --git a/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.test.js b/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.test.js index ce25c69fe9..6fa2ed242f 100644 --- a/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.test.js +++ b/graylog2-web-interface/src/integrations/aws/common/formDataAdapter.test.js @@ -45,6 +45,7 @@ describe('formDataAdapter', () => { kinesis_record_batch_size: 'awsCloudWatchBatchSize', kinesis_stream_name: 'awsCloudWatchKinesisStream', throttling_allowed: 'awsCloudWatchThrottleEnabled', + override_source: 'overrideSource' }; const request = toGenericInputCreateRequest(formData); @@ -83,6 +84,7 @@ describe('formDataAdapter', () => { dynamodb_endpoint: 'awsEndpointDynamoDB', iam_endpoint: 'awsEndpointIAM', kinesis_endpoint: 'awsEndpointKinesis', + override_source: 'overrideSource', }; const request = toAWSRequest(formData, options); @@ -106,7 +108,7 @@ describe('formDataAdapter', () => { }; it('adapts formData into an AWS request with key & secret', () => { - testAWSRequest({ + const request = testAWSRequest({ awsAuthenticationType: { value: AWS_AUTH_TYPES.keysecret }, awsCloudWatchAssumeARN: { value: '' }, awsCloudWatchAwsKey: { value: 'mykey' }, @@ -115,11 +117,14 @@ describe('formDataAdapter', () => { awsEndpointIAM: { value: undefined }, awsEndpointKinesis: { value: undefined }, awsCloudWatchAwsSecret: { value: 'mysecret' }, + overrideSource: { value: '' }, }); + + expect(request).toBeDefined(); }); it('adapts formData into an AWS request with automatic auth', () => { - testAWSRequest({ + const request = testAWSRequest({ awsAuthenticationType: { value: AWS_AUTH_TYPES.automatic }, awsCloudWatchAssumeARN: { value: '' }, key: 'mykey', @@ -127,8 +132,11 @@ describe('formDataAdapter', () => { awsEndpointDynamoDB: { value: undefined }, awsEndpointIAM: { value: undefined }, awsEndpointKinesis: { value: undefined }, + overrideSource: { value: '' }, secret: 'mysecret', }); + + expect(request).toBeDefined(); }); it('adapts formData into an AWS request with additional options', () => { @@ -147,6 +155,7 @@ describe('formDataAdapter', () => { awsEndpointIAM: { value: undefined }, awsEndpointKinesis: { value: undefined }, awsCloudWatchAwsSecret: { value: 'mysecret' }, + overrideSource: { value: '' }, }, options, ); @@ -155,10 +164,14 @@ describe('formDataAdapter', () => { }); it('adapts formData into an InputCreateRequest with key & secret', () => { - testGenericInputCreateRequest(exampleFormDataWithKeySecretAuth); + const request = testGenericInputCreateRequest(exampleFormDataWithKeySecretAuth); + + expect(request).toBeDefined(); }); it('adapts formData into an InputCreateRequest with automatic auth', () => { - testGenericInputCreateRequest(exampleFormDataWithAutomaticAuth); + const request = testGenericInputCreateRequest(exampleFormDataWithAutomaticAuth); + + expect(request).toBeDefined(); }); }); diff --git a/graylog2-web-interface/test/fixtures/aws/FormData.fixtures.js b/graylog2-web-interface/test/fixtures/aws/FormData.fixtures.js index 31ced2ed56..da88df9bb6 100644 --- a/graylog2-web-interface/test/fixtures/aws/FormData.fixtures.js +++ b/graylog2-web-interface/test/fixtures/aws/FormData.fixtures.js @@ -33,6 +33,7 @@ export const exampleFormDataWithKeySecretAuth = { awsEndpointIAM: { value: undefined }, awsEndpointKinesis: { value: undefined }, awsCloudWatchAwsSecret: { value: 'mysecret' }, + overrideSource: { value: '' }, }; export const exampleFormDataWithAutomaticAuth = { @@ -49,6 +50,7 @@ export const exampleFormDataWithAutomaticAuth = { awsEndpointDynamoDB: { value: undefined }, awsEndpointIAM: { value: undefined }, awsEndpointKinesis: { value: undefined }, + overrideSource: { value: '' }, key: 'mykey', secret: 'mysecret', };