opensource/graylog2-server
1
0
Fork 0
mirror of https://github.com/Graylog2/graylog2-server.git synced 2026-03-13 09:32:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

AWS Kinesis/Cloudwatch input: Fix several issues (#22370)

Browse Source 
* error visibility and proceed with empty messages in stream fix

* reverted change

* added additional message fields and updated test cases

* fix lint errors

* update lint errors fix

* updated lint errors

* Fix build errors

* fix build issue

* updated lint errors

* updated input setup fix

* stream arn update to configuration

* fix lint errors

* fix lint error

* resolve conficts

* review comments commit

* naming consistency

* fixed compile errors

* Use client builder for ARN request, refactoring

* Cleanup

* fix bot error and update test cases

* Remove None log type, default to Kinesis Raw

* Silence verbose SDK messages

* Improve handling of missing stream ARN, cleanup

* Fix tests

* Add missing Override Source field on the review page

* Post string source

---------

Co-authored-by: Dan Torrey <dan.torrey@graylog.com>
This commit is contained in:
Nithin Kasam
2025-06-05 04:58:42 +05:30
committed by GitHub
parent 4857aa9f0a
commit c91fdab281
35 changed files with 403 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
changelog/unreleased/issue-22085.toml Normal file
View File

@@ -0,0 +1,5 @@
type = "f"
message = "Add additional data fields for the Kinesis/Cloudwatch input."
issues = ["22085"]
pulls = ["22370"]

5
changelog/unreleased/issue-22137.toml Normal file
View File

@@ -0,0 +1,5 @@
type = "f"
message = "Improve error handling for the Kinesis/Cloudwatch input."
issues = ["22137"]
pulls = ["22370"]

5
changelog/unreleased/issue-22138.toml Normal file
View File

@@ -0,0 +1,5 @@
type = "f"
message = "Allow the Kinesis/Cloudwatch input to be created for empty Kinesis streams."
issues = ["22138"]
pulls = ["22370"]

5
changelog/unreleased/issue-22140.toml Normal file
View File

@@ -0,0 +1,5 @@
type = "f"
message = "Allow the source field to be overridden for the Kinesis/Cloudwatch input."
issues = ["22140"]
pulls = ["22370"]

10
graylog2-server/src/main/java/org/graylog/integrations/aws/AWSMessageType.java
View File

@@ -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<AWSMessageType> 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());
}
}
}

27
graylog2-server/src/main/java/org/graylog/integrations/aws/cloudwatch/KinesisLogEntry.java
View File

@@ -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<String> 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<String> subscriptionFilters) {
return new AutoValue_KinesisLogEntry(kinesisStream, logGroup, logStream, timestamp, message,
owner, messageType, subscriptionFilters);
}
}
}

6
graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AWSCodec.java
View File

@@ -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;
}

24
graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/AbstractKinesisCodec.java
View File

@@ -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<Message> 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() {

9
graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisCloudWatchFlowLogCodec.java
View File

@@ -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<String, Object> buildFields(FlowLogMessage msg) {
private Map<String, Object> addFlowLogFields(FlowLogMessage msg) {
final String prefix = this.noFlowLogPrefix ? "" : FLOW_LOG_PREFIX;
final HashMap<String, Object> fields = new HashMap<>();
fields.put(prefix + FIELD_ACCOUNT_ID, msg.getAccountId());

6
graylog2-server/src/main/java/org/graylog/integrations/aws/codecs/KinesisRawLogCodec.java
View File

@@ -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);

24
graylog2-server/src/main/java/org/graylog/integrations/aws/inputs/AWSInput.java
View File

@@ -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);
}
}

24
graylog2-server/src/main/java/org/graylog/integrations/aws/resources/AWSResource.java
View File

@@ -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();

14
graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/AWSInputCreateRequest.java
View File

@@ -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);

10
graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisHealthCheckRequest.java → graylog2-server/src/main/java/org/graylog/integrations/aws/resources/requests/KinesisRequest.java
View File

@@ -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<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();
}
}
}

2
graylog2-server/src/main/java/org/graylog/integrations/aws/service/AWSService.java
View File

@@ -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.");

41
graylog2-server/src/main/java/org/graylog/integrations/aws/service/KinesisService.java
View File

@@ -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<Record> 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 <a href="https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html"/>
*/
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<? extends Codec> 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);
}
}
}

13
graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoder.java
View File

@@ -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<KinesisLogEntry> 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);

8
graylog2-server/src/main/java/org/graylog/integrations/aws/transports/KinesisTransport.java
View File

@@ -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,

2
graylog2-server/src/main/resources/log4j2.xml
View File

@@ -52,6 +52,8 @@
<RegexFilter regex=".*An error occurred while aborting the stream.*" onMatch="DENY" onMismatch="ACCEPT"/>
</Logger>
<Logger name="org.apache.iceberg.metrics.LoggingMetricsReporter" level="warn"/>
<!-- Silence AWS Kinesis log chatter -->
<Logger name="software.amazon.kinesis" level="warn"/>
<Root level="info">
<AppenderRef ref="STDOUT"/>
<AppenderRef ref="graylog-internal-logs"/>

14
graylog2-server/src/test/java/org/graylog/integrations/aws/cloudwatch/FlowLogMessageTest.java
View File

@@ -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);
}
}
}

8
graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/AWSCodecTest.java
View File

@@ -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());
}

4
graylog2-server/src/test/java/org/graylog/integrations/aws/codecs/CloudWatchFlowLogCodecTest.java
View File

@@ -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));

16
graylog2-server/src/test/java/org/graylog/integrations/aws/service/AWSServiceTest.java
View File

@@ -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.

48
graylog2-server/src/test/java/org/graylog/integrations/aws/service/KinesisServiceTest.java
View File

@@ -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<String, Object> 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<String, Object> 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<String, Object> 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 {

11
graylog2-server/src/test/java/org/graylog/integrations/aws/transports/KinesisPayloadDecoderTest.java
View File

@@ -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<KinesisLogEntry> 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

11
graylog2-web-interface/src/integrations/aws/cloudwatch/FormAdvancedOptions.tsx
View File

@@ -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".'
/>
<Input
id="overrideSource"
type="text"
value={overrideSource?.value}
onChange={onChange}
label="Override Source (optional)"
help="The message source is set to aws-kinesis-raw-logs by default. If desired, you may override it with a custom value."
/>
<Input
id="awsCloudWatchBatchSize"
type="number"

78
graylog2-web-interface/src/integrations/aws/cloudwatch/StepHealthCheck.tsx
View File

@@ -14,20 +14,19 @@
* along with this program. If not, see
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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) => {
</Panel>
);
}
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 (
<FormWrap
@@ -147,31 +160,14 @@ const StepHealthCheck = ({ onChange, onSubmit }: StepHealthCheckProps) => {
disabled={false}
title="Create Kinesis Stream"
description={
<p>
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 <a href="/system/pipelines">Pipeline Rules</a> to set up your own parser!
</p>
<p>If available, a parsed sample {logTypeLabel} message from the stream will be shown below.</p>
}>
<Panel
bsStyle={bsStyle}
header={
<Notice>
<Icon name={iconName} size="2x" />
<span>
{acknowledgment} looks like <em>{logType}</em> message type.
</span>
</Notice>
}>
{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!`}
</Panel>
<Input
id="awsCloudWatchLog"
type="textarea"
label="Formatted Log Message"
value={logData.message}
value={logData.message || 'No messages found in stream.'}
rows={10}
disabled
/>

51
graylog2-web-interface/src/integrations/aws/cloudwatch/StepReview.tsx
View File

@@ -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} <small>(default)</small>
</>
);
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} <small>(default)</small>
</>
);
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
</Subheader>
<ReviewItems>
<li>
<strong>Stream</strong>
<strong>Kinesis Stream</strong>
<span>{awsCloudWatchKinesisStream.value}</span>
</li>
<li>
<strong>Kinesis Stream ARN</strong>
<span>
{!awsCloudwatchKinesisStreamArn ? (
<ArnErrorMessage>
Error: Failed to get stream ARN. Please ensure the IAM role includes the kinesis:DescribeStream permission.
</ArnErrorMessage>
) : (
awsCloudwatchKinesisStreamArn
)}
</span>
</li>
<li>
<strong>Global Input</strong>
<span>
@@ -262,6 +287,10 @@ const StepReview = ({ onSubmit, onEditClick, externalInputSubmit = false }: Step
<Icon name={addPrefix ? 'check_circle' : 'cancel'} />
</span>
</li>
{overrideSource.value && (<li>
<strong>Override Source</strong>
<span>{overrideSource.value}</span>
</li>)}
</ReviewItems>
<Subheader>Formatting</Subheader>

3
graylog2-web-interface/src/integrations/aws/cloudwatch/_initialFormData.js
View File

@@ -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;

40
graylog2-web-interface/src/integrations/aws/common/FormWrap.tsx
View File

@@ -15,7 +15,7 @@
* <http://www.mongodb.com/licensing/server-side-public-license>.
*/
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 = (
<>
<ErrorOutputStyle />
<ErrorOutput>{niceMessage || fullMessage}</ErrorOutput>
{niceMessage && (
<ErrorToggleInfo onClick={() => toggleExpanded(!expanded)}>
<ErrorToggleInfo isDarkMode={isDarkMode} onClick={() => toggleExpanded(!expanded)}>
More Info <MoreIcon name="chevron_right" expanded={expanded} />
</ErrorToggleInfo>
)}
@@ -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,

1
graylog2-web-interface/src/integrations/aws/common/Routes.ts
View File

@@ -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: {

1
graylog2-web-interface/src/integrations/aws/common/constants.js
View File

@@ -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';

4
graylog2-web-interface/src/integrations/aws/common/formDataAdapter.js
View File

@@ -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,
},
});

21
graylog2-web-interface/src/integrations/aws/common/formDataAdapter.test.js
View File

@@ -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();
});
});

2
graylog2-web-interface/test/fixtures/aws/FormData.fixtures.js vendored
View File

@@ -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',
};