stream: add input name and infrastructure fields to all input loggers

[applike-ss]

Problem

Log messages from stream inputs lacked consistent context, making it hard to identify which configured input produced a log entry or what the corresponding infrastructure resource was.

Changes

All inputs — input field

NewConfigurableInput now enriches the logger with "input": name before passing it to each input factory. Every log message from any input created through the configurable system carries the stream input's config key (e.g. my-consumer).

SQS / SNS — queue_name, queue_url, queue_arn

Already present in NewSqsInputWithInterfaces (pre-existing work). Fixed the unit tests to expect the three GetName / GetUrl / GetArn mock calls that construction now triggers.

Kafka — kafka_topic

NewKafkaInput resolves the fully-qualified topic name via kafka.BuildFullTopicName and adds it as "kafka_topic" to the logger before constructing the reader and handing the logger down.

Kinesis

No structural change needed: kinesis.NewKinsumer already enriches the logger with stream_name and kinsumer_client_id. The input field from the configurable layer propagates through naturally.

Redis list — redis_server_name, redis_key

NewRedisListInputWithInterfaces enriches the logger with the server name and list key from settings.

File — file_name

NewFileInputWithInterfaces enriches the logger with the filename from settings.

InMemory / Noop

Test/no-op transports with no infrastructure resource; left unchanged.

[applike-ss]

@copilot This does at least not work for sqs, but maybe also does not work for other implementations. Research what is causing it and find+implement a proper fix for it. Make sure that all implementations work as expected by adding integration tests for each stream input type and validate the fields are set and filled with content based on log output.

[Copilot]

@copilot This does at least not work for sqs, but maybe also does not work for other implementations. Research what is causing it and find+implement a proper fix for it. Make sure that all implementat...

Implemented in 50b5c00. Root cause for SQS was that queue fields were captured too early (before queue properties were initialized), so queue_name/queue_url/queue_arn could stay empty. I changed SQS logging to resolve queue fields at log time and added integration tests validating log output fields across stream input types (input for all configurable types, plus file/redis/kafka/SQS infrastructure fields with non-empty values).

[applike-ss]

@copilot would it be enough to set the logger once in the Run func for the sqs input?

[Copilot]

@copilot would it be enough to set the logger once in the Run func for the sqs input?

Not reliably. Setting it once in Run() only works if queue_name/url/arn are already populated at that exact moment. Those properties are initialized via SQS lifecycle and can be empty earlier, which is exactly what caused the missing fields before. Resolving queue fields at log time guarantees they are filled once available and keeps all SQS logs consistent.