feat(aws_s3 source): add S3 download processing duration metric

[sanjams2]

Summary

Add a histogram metric s3_object_download_duration_seconds to the AWS S3 source to track how long S3 object download + decompression + framing + decoding takes. This helps operators identify slow buckets and debug S3 performance issues.

The metric includes two labels:

• bucket: The S3 bucket name, to identify which buckets are slow
• status: Either "success" or "error", to distinguish failed downloads

The timing captures the full download lifecycle from the GetObject API call through complete ByteStream body consumption (including decompression).

Vector configuration

[sources.s3]
type = "aws_s3"
region = "us-east-1"
sqs.queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-queue"

How did you test this PR?

• cargo test
• running locally now

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Histogram metric pattern followed:

  vector/src/internal_events/http_client.rs

  Lines 46 to 73 in bc654a7

  	#[derive(Debug)]
  	pub struct GotHttpResponse<'a, T> {
  	pub response: &'a Response<T>,
  	pub roundtrip: Duration,
  	}
  	impl<T: HttpBody> InternalEvent for GotHttpResponse<'_, T> {
  	fn emit(self) {
  	debug!(
  	message = "HTTP response.",
  	status = %self.response.status(),
  	version = ?self.response.version(),
  	headers = ?remove_sensitive(self.response.headers()),
  	body = %FormatBody(self.response.body()),
  	);
  	counter!(
  	"http_client_responses_total",
  	"status" => self.response.status().as_u16().to_string(),
  	)
  	.increment(1);
  	histogram!("http_client_rtt_seconds").record(self.roundtrip);
  	histogram!(
  	"http_client_response_rtt_seconds",
  	"status" => self.response.status().as_u16().to_string(),
  	)
  	.record(self.roundtrip);
  	}
  	}

• Existing S3/SQS internal events:

  vector/src/internal_events/aws_sqs.rs

  Lines 11 to 45 in bc654a7

  	mod s3 {
  	use aws_sdk_sqs::types::{
  	BatchResultErrorEntry, DeleteMessageBatchRequestEntry, DeleteMessageBatchResultEntry,
  	SendMessageBatchRequestEntry, SendMessageBatchResultEntry,
  	};
  	use super::*;
  	use crate::sources::aws_s3::sqs::ProcessingError;
  	#[derive(Debug)]
  	pub struct SqsMessageProcessingError<'a> {
  	pub message_id: &'a str,
  	pub error: &'a ProcessingError,
  	}
  	impl InternalEvent for SqsMessageProcessingError<'_> {
  	fn emit(self) {
  	error!(
  	message = "Failed to process SQS message.",
  	message_id = %self.message_id,
  	error = %self.error,
  	error_code = "failed_processing_sqs_message",
  	error_type = error_type::PARSER_FAILED,
  	stage = error_stage::PROCESSING,
  	);
  	counter!(
  	"component_errors_total",
  	"error_code" => "failed_processing_sqs_message",
  	"error_type" => error_type::PARSER_FAILED,
  	"stage" => error_stage::PROCESSING,
  	)
  	.increment(1);
  	}
  	}

  - Existing
  SqsMessageProcessingError pattern we followed

[github-actions]

All contributors have signed the CLA ✍️ ✅
_{Posted by the CLA Assistant Lite bot.}

[sanjams2]

I have read the CLA Document and I hereby sign the CLA

[sanjams2]

recheck

[thomasqueirozb]

This will be a nice addition! I just think we need to redesign the metrics a bit. Additionally new metrics need to be documented in website/cue/reference/components/sources/internal_metrics.cue.

[thomasqueirozb]

I think we should emit different events here. One on success and another on failure. I don't think any other metric/histogram in Vector uses status as a way to track success/failure but we instead opt to emit different metrics.

[sanjams2]

Yeah I debated on this a bit, but SGTM. Will fix

[sanjams2]

sorry I kinda screwed up the git history here b/c I was juggling a few PRs for this package. This is what changed if that helps: e49004b