feat(new source): A generic http_scrape source

.github/workflows/integration-test.yml

@@ -72,6 +72,7 @@ jobs:
           - test: 'fluent'
           - test: 'gcp'
           - test: 'humio'
+          - test: 'http-scrape'
           - test: 'influxdb'
           - test: 'kafka'
           - test: 'logstash'

Cargo.toml

@@ -456,6 +456,7 @@ sources-logs = [
   "sources-gcp_pubsub",
   "sources-heroku_logs",
   "sources-http",
+  "sources-http_scrape",
   "sources-internal_logs",
   "sources-journald",
   "sources-kafka",
@@ -503,6 +504,7 @@ sources-gcp_pubsub = ["gcp", "dep:h2", "dep:prost-types", "protobuf-build", "dep
 sources-heroku_logs = ["sources-utils-http", "sources-utils-http-query", "sources-http"]
 sources-host_metrics = ["dep:heim"]
 sources-http = ["sources-utils-http", "sources-utils-http-query"]
+sources-http_scrape = ["sources-utils-http-scrape"]
 sources-internal_logs = []
 sources-internal_metrics = []
 sources-journald = []
@@ -514,7 +516,7 @@ sources-nats = ["dep:nats", "dep:nkeys"]
 sources-nginx_metrics = ["dep:nom"]
 sources-opentelemetry = ["dep:hex", "dep:opentelemetry-proto", "dep:prost-types", "sources-http", "sources-utils-http", "sources-vector"]
 sources-postgresql_metrics = ["dep:postgres-openssl", "dep:tokio-postgres"]
-sources-prometheus = ["dep:prometheus-parser", "sinks-prometheus", "sources-http", "sources-utils-http"]
+sources-prometheus = ["dep:prometheus-parser", "sinks-prometheus", "sources-utils-http-scrape"]
 sources-redis= ["dep:redis"]
 sources-socket = ["listenfd", "tokio-util/net", "sources-utils-udp", "sources-utils-tcp-keepalive", "sources-utils-tcp-socket", "sources-utils-tls", "sources-utils-unix"]
 sources-splunk_hec = ["sources-utils-tls", "dep:roaring"]
@@ -527,6 +529,7 @@ sources-utils-http-encoding = ["dep:snap", "sources-utils-http-error"]
 sources-utils-http-error = []
 sources-utils-http-prelude = ["sources-utils-http", "sources-utils-tls", "sources-utils-http-auth", "sources-utils-http-encoding", "sources-utils-http-error"]
 sources-utils-http-query = []
+sources-utils-http-scrape = ["sources-utils-http", "sources-http"]
 sources-utils-tcp-keepalive = []
 sources-utils-tcp-socket = []
 sources-utils-tls = []
@@ -715,6 +718,7 @@ all-integration-tests = [
   "gcp-cloud-storage-integration-tests",
   "gcp-integration-tests",
   "gcp-pubsub-integration-tests",
+  "http-scrape-integration-tests",
   "humio-integration-tests",
   "influxdb-integration-tests",
   "kafka-integration-tests",
@@ -772,6 +776,7 @@ gcp-cloud-storage-integration-tests = ["sinks-gcp"]
 gcp-integration-tests = ["sinks-gcp"]
 gcp-pubsub-integration-tests = ["sinks-gcp", "sources-gcp_pubsub"]
 humio-integration-tests = ["sinks-humio"]
+http-scrape-integration-tests = ["sources-http_scrape"]
 influxdb-integration-tests = ["sinks-influxdb"]
 kafka-integration-tests = ["sinks-kafka", "sources-kafka"]
 logstash-integration-tests = ["docker", "sources-logstash"]

Makefile

@@ -334,7 +334,7 @@ test-enterprise: ## Runs enterprise related behavioral tests
 test-integration: ## Runs all integration tests
 test-integration: test-integration-apex test-integration-aws test-integration-axiom test-integration-azure test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
 test-integration: test-integration-azure test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
-test-integration: test-integration-eventstoredb test-integration-fluent test-integration-gcp test-integration-humio test-integration-influxdb
+test-integration: test-integration-eventstoredb test-integration-fluent test-integration-gcp test-integration-humio test-integration-http-scrape test-integration-influxdb
 test-integration: test-integration-kafka test-integration-logstash test-integration-loki test-integration-mongodb test-integration-nats
 test-integration: test-integration-nginx test-integration-opentelemetry test-integration-postgres test-integration-prometheus test-integration-pulsar
 test-integration: test-integration-redis test-integration-splunk test-integration-dnstap test-integration-datadog-agent test-integration-datadog-logs

lib/codecs/src/decoding/framing/character_delimited.rs

@@ -34,12 +34,12 @@ impl CharacterDelimitedDecoderConfig {
 pub struct CharacterDelimitedDecoderOptions {
     /// The character that delimits byte sequences.
     #[serde(with = "vector_core::serde::ascii_char")]
-    delimiter: u8,
+    pub delimiter: u8,
     /// The maximum length of the byte buffer.
     ///
     /// This length does *not* include the trailing delimiter.
     #[serde(skip_serializing_if = "vector_core::serde::skip_serializing_if_default")]
-    max_length: Option<usize>,
+    pub max_length: Option<usize>,
 }
 
 impl CharacterDelimitedDecoderOptions {
@@ -56,9 +56,9 @@ impl CharacterDelimitedDecoderOptions {
 #[derive(Copy, Clone, Debug, Eq, PartialEq, Ord, PartialOrd, Hash)]
 pub struct CharacterDelimitedDecoder {
     /// The delimiter used to separate byte sequences.
-    delimiter: u8,
+    pub delimiter: u8,
     /// The maximum length of the byte buffer.
-    max_length: usize,
+    pub max_length: usize,
 }
 
 impl CharacterDelimitedDecoder {

lib/codecs/src/decoding/mod.rs

@@ -324,6 +324,38 @@ impl DeserializerConfig {
             DeserializerConfig::Gelf => GelfDeserializerConfig.schema_definition(log_namespace),
         }
     }
+
+    /// Get the HTTP content type.
+    pub const fn content_type(&self, framer: &FramingConfig) -> &'static str {
+        match (&self, framer) {
+            (
+                DeserializerConfig::Json | DeserializerConfig::NativeJson,
+                FramingConfig::NewlineDelimited { .. },
+            ) => "application/x-ndjson",
+            (
+                DeserializerConfig::Gelf
+                | DeserializerConfig::Json
+                | DeserializerConfig::NativeJson,
+                FramingConfig::CharacterDelimited {
+                    character_delimited:
+                        CharacterDelimitedDecoderOptions {
+                            delimiter: b',',
+                            max_length: Some(usize::MAX),
+                        },
+                },
+            ) => "application/json",
+            (DeserializerConfig::Native, _) => "application/octet-stream",
+            (
+                DeserializerConfig::Json
+                | DeserializerConfig::NativeJson
+                | DeserializerConfig::Bytes
+                | DeserializerConfig::Gelf,
+                _,
+            ) => "text/plain",
+            #[cfg(feature = "syslog")]
+            (DeserializerConfig::Syslog, _) => "text/plain",
+        }
+    }
 }
 
 /// Parse structured events from bytes.

scripts/integration/docker-compose.axiom.yml

@@ -51,10 +51,12 @@ services:
       - axiom-db
     volumes:
       - ${PWD}:/code
+      - target:/code/target
       - cargogit:/usr/local/cargo/git
       - cargoregistry:/usr/local/cargo/registry
 
 volumes:
+  target: {}
   cargogit: {}
   cargoregistry: {}
   postgres_data: {}

scripts/integration/docker-compose.chronicle.yml

@@ -31,11 +31,13 @@ services:
       - CHRONICLE_ADDRESS=http://chronicle-emulator:3000
     volumes:
       - ${PWD}:/code
+      - target:/code/target
       - cargogit:/usr/local/cargo/git
       - cargoregistry:/usr/local/cargo/registry
       - ${PWD}/scripts/integration/chronicleauth.json:/chronicleauth.json
       - ${PWD}/scripts/integration/invalidchronicleauth.json:/invalidchronicleauth.json
 
 volumes:
+  target: {}
   cargogit: {}
   cargoregistry: {}

scripts/integration/docker-compose.http-scrape.yml

@@ -0,0 +1,81 @@
+version: "3"
+
+services:
+  # https://github.com/sigoden/dufs
+  # serves static files at an HTTP endpoint
+  dufs:
+    image: docker.io/sigoden/dufs:latest
+    networks:
+      - backend
+    command:
+      - "/data"
+    volumes:
+      - ${PWD}/tests/data/http-scrape/serve:/data
+  # To validate Basic HTTP authentication option
+  dufs-auth:
+    image: docker.io/sigoden/dufs:latest
+    networks:
+      - backend
+    command:
+      - "-a"
+      - "/@user:pass"
+      - "--auth-method"
+      - "basic"
+      - "/data"
+    volumes:
+      - ${PWD}/tests/data/http-scrape/serve:/data
+  # To validate TLS options
+  dufs-https:
+    image: docker.io/sigoden/dufs:latest
+    networks:
+      - backend
+    command:
+      - "--tls-cert"
+      - "/certs/ca.cert.pem"
+      - "--tls-key"
+      - "/certs/ca.key.pem"
+      - "/data"
+    volumes:
+      - ${PWD}/tests/data/http-scrape/serve:/data
+      - ${PWD}/tests/data/ca/intermediate_server/certs/dufs-https-chain.cert.pem:/certs/ca.cert.pem
+      - ${PWD}/tests/data/ca/intermediate_server/private/dufs-https.key.pem:/certs/ca.key.pem
+  runner:
+    build:
+      context: ${PWD}
+      dockerfile: scripts/integration/Dockerfile
+      args:
+        - RUST_VERSION=${RUST_VERSION}
+    working_dir: /code
+    networks:
+      - backend
+    command:
+      - "cargo"
+      - "nextest"
+      - "run"
+      - "--no-fail-fast"
+      - "--no-default-features"
+      - "--features"
+      - "http-scrape-integration-tests"
+      - "--lib"
+      - "sources::http_scrape::"
+    depends_on:
+      - dufs
+      - dufs-auth
+      - dufs-https
+    environment:
+      - DUFS_ADDRESS=http://dufs:5000
+      - DUFS_AUTH_ADDRESS=http://dufs-auth:5000
+      - DUFS_HTTPS_ADDRESS=https://dufs-https:5000
+    volumes:
+      - ${PWD}:/code
+      - target:/code/target
+      - cargogit:/usr/local/cargo/git
+      - cargoregistry:/usr/local/cargo/registry
+
+networks:
+  backend: {}
+
+volumes:
+  target: {}
+  cargogit: {}
+  cargoregistry: {}

src/codecs/encoding/encoder.rs

@@ -104,7 +104,7 @@ impl Encoder<Framer> {
     }
 
     /// Get the HTTP content type.
-    pub const fn content_type(&self) -> &str {
+    pub const fn content_type(&self) -> &'static str {
         match (&self.serializer, &self.framer) {
             (Serializer::Json(_) | Serializer::NativeJson(_), Framer::NewlineDelimited(_)) => {
                 "application/x-ndjson"

src/internal_events/http_scrape.rs

@@ -0,0 +1,90 @@
+use metrics::counter;
+use vector_core::internal_event::InternalEvent;
+
+use super::prelude::{error_stage, error_type, http_error_code};
+
+#[derive(Debug)]
+pub struct HttpScrapeEventsReceived {
+    pub byte_size: usize,
+    pub count: usize,
+    pub url: String,
+}
+
+impl InternalEvent for HttpScrapeEventsReceived {
+    fn emit(self) {
+        trace!(
+            message = "Events received.",
+            count = %self.count,
+            byte_size = %self.byte_size,
+            url = %self.url,
+        );
+        counter!(
+            "component_received_events_total", self.count as u64,
+            "uri" => self.url.clone(),
+        );
+        counter!(
+            "component_received_event_bytes_total", self.byte_size as u64,
+            "uri" => self.url.clone(),
+        );
+        // deprecated
+        counter!(
+            "events_in_total", self.count as u64,
+            "uri" => self.url,
+        );
+    }
+}
+
+#[derive(Debug)]
+pub struct HttpScrapeHttpResponseError {
+    pub code: hyper::StatusCode,
+    pub url: String,
+}
+
+impl InternalEvent for HttpScrapeHttpResponseError {
+    fn emit(self) {
+        error!(
+            message = "HTTP error response.",
+            url = %self.url,
+            stage = error_stage::RECEIVING,
+            error_type = error_type::REQUEST_FAILED,
+            error_code = %http_error_code(self.code.as_u16()),
+            internal_log_rate_secs = 10,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "url" => self.url,
+            "stage" => error_stage::RECEIVING,
+            "error_type" => error_type::REQUEST_FAILED,
+            "error_code" => http_error_code(self.code.as_u16()),
+        );
+        // deprecated
+        counter!("http_error_response_total", 1);
+    }
+}
+
+#[derive(Debug)]
+pub struct HttpScrapeHttpError {
+    pub error: crate::Error,
+    pub url: String,
+}
+
+impl InternalEvent for HttpScrapeHttpError {
+    fn emit(self) {
+        error!(
+            message = "HTTP request processing error.",
+            url = %self.url,
+            error = ?self.error,
+            error_type = error_type::REQUEST_FAILED,
+            stage = error_stage::RECEIVING,
+            internal_log_rate_secs = 10,
+        );
+        counter!(
+            "component_errors_total", 1,
+            "url" => self.url,
+            "error_type" => error_type::REQUEST_FAILED,
+            "stage" => error_stage::RECEIVING,
+        );
+        // deprecated
+        counter!("http_request_errors_total", 1);
+    }
+}

src/internal_events/mod.rs

@@ -53,6 +53,8 @@ mod geoip;
 mod heartbeat;
 mod http;
 pub mod http_client;
+#[cfg(feature = "sources-utils-http-scrape")]
+mod http_scrape;
 #[cfg(feature = "sources-internal_logs")]
 mod internal_logs;
 #[cfg(all(unix, feature = "sources-journald"))]
@@ -187,6 +189,8 @@ pub(crate) use self::geoip::*;
     feature = "sources-splunk_hec",
 ))]
 pub(crate) use self::http::*;
+#[cfg(feature = "sources-utils-http-scrape")]
+pub(crate) use self::http_scrape::*;
 #[cfg(feature = "sources-internal_logs")]
 pub(crate) use self::internal_logs::*;
 #[cfg(all(unix, feature = "sources-journald"))]

src/internal_events/prelude.rs

@@ -47,11 +47,10 @@ pub mod error_type {
 #[cfg(any(
     feature = "sinks-azure_blob",
     feature = "sinks-elasticsearch",
-    feature = "sinks-prometheus",
     feature = "sources-apache_metrics",
     feature = "sources-aws_ecs_metrics",
     feature = "sources-aws_kinesis_firehose",
-    feature = "sources-prometheus",
+    feature = "sources-http-scrape",
     feature = "sources-utils-http",
 ))]
 pub(crate) fn http_error_code(code: u16) -> String {