SQS configure multiple consumers

[lhh528634141]

Consume data from S3 to Kafka. If I configure multiple vectors, will it be consumed repeatedly?

[jszwedko]

Hi @lhh528634141 !

You can configure multiple Vector instances to consume from AWS S3 using the same SQS queue. SQS handles ensuring messages are only delivered to one instance.

[lhh528634141]

When the vector is idle and no message comes in, this error will be reported

2024-12-12T09:39:38.613153Z ERROR source{component_kind="source" component_id=s3 component_type=aws_s3}: vector::internal_events::aws_sqs::s3: Failed to process SQS message. message_id=9dff2cbe-846b-43ca-bc14-bfba3a9ee1a0 error=Could not parse SQS message with id 9dff2cbe-846b-43ca-bc14-bfba3a9ee1a0 as S3 notification: expected ident at line 1 column 2 error_code="failed_processing_sqs_message" error_type="parser_failed" stage="processing" internal_log_rate_limit=true

[jszwedko]

That indicates that the SQS message could not be parsed as as a S3 bucket notification: https://docs.aws.amazon.com/AmazonS3/latest/userguide/EventNotifications.html . It sounds like it might not even be JSON 🤔

Are you trying to process data in S3? Or are you trying to treat SQS messages as events themselves?

[lhh528634141]

this is my config，parse alb logs

[sources.s3]
type = "aws_s3"
compression = "auto"
sqs.queue_url = "https://sqs.ap-northeast-1.amazonaws.com/xxxx/lambda-log"
region = "ap-northeast-1"
auth.region = "ap-northeast-1"
auth.assume_role = "arn:aws:iam::xxxx:role/upex-role-vector-server"



[transforms.elasticloadbalancing_fields_parsed]
type = "remap" # required
inputs = ["s3"]
drop_on_error = false
source = '''
. = parse_aws_alb_log!(string!(.message))
.target_port_list = .target_port_list[0]
.target_status_code_list = .target_status_code_list[0]

'''

[sinks.my_kafka_job]
type = "kafka"
inputs = ["elasticloadbalancing_fields_parsed"]
bootstrap_servers = "xxx:9092"
topic = "jusetest"
encoding.codec = "json"
sasl.enabled = true
sasl.mechanism = "PLAIN"
sasl.username = "xxx"
sasl.password = "xxx"

[jszwedko]

Gotcha, it sounds like you have some invalid SQS messages in the queue. Could you try to fetch them manually via the UI and inspect them to verify if they match the S3 bucket notification format? Feel free to share them here if you aren't sure and want another opinion.

[jszwedko]

That indicates that the logs didn't match the format expected by parse_aws_alb_log. I'd suggest using parse_grok instead and providing a pattern that matches your input.