enhancement: azure_blob sink

Cargo.toml

@@ -142,6 +142,11 @@ rusoto_signature = { version = "0.46.0", optional = true }
 rusoto_sqs = { version = "0.46.0", optional = true }
 rusoto_sts = { version = "0.46.0", optional = true }
 
+# Azure
+azure_core = { git = "https://github.com/Azure/azure-sdk-for-rust.git", rev = "16bcf0ab1bb6e380d966a69d314de1e99ede553a", default-features = false, features = ["enable_reqwest"], optional = true }
+azure_storage = { git = "https://github.com/Azure/azure-sdk-for-rust.git", rev = "16bcf0ab1bb6e380d966a69d314de1e99ede553a", default-features = false, features = ["blob"], optional = true }
+reqwest = { version = "0.11", optional = true }
+
 # Tower
 tower = { version = "0.4.8", default-features = false, features = ["buffer", "limit", "retry", "timeout", "util"] }
 tower-layer = { version = "0.3.1", default-features = false }
@@ -208,7 +213,7 @@ getset = { version = "0.1.1", default-features = false }
 glob = { version = "0.3.0", default-features = false }
 grok = { version = "1.1.0", default-features = false, optional = true }
 headers = { version = "0.3.4", default-features = false }
-heim = { version = "0.1.0-rc.1", default-features = false, features = ["cpu", "disk", "host", "memory", "net"], optional = true }
+heim = { git = "https://github.com/heim-rs/heim.git", rev="b292f1535bb27c03800cdb7509fa81a40859fbbb", default-features = false, features = ["cpu", "disk", "host", "memory", "net"], optional = true }
 hostname = { version = "0.3.1", default-features = false }
 http = { version = "0.2.4", default-features = false }
 hyper = { version = "0.14.9", default-features = false, features = ["stream"] }
@@ -229,7 +234,7 @@ async-nats = { version = "0.9.18", default-features = false, optional = true }
 nom = { version = "6.1.2", default-features = false, optional = true }
 notify = { version = "4.0.17", default-features = false }
 num_cpus = { version = "1.13.0", default-features = false }
-once_cell = { version = "1.3", default-features = false }
+once_cell = { version = "1.7", default-features = false }
 openssl = { version = "0.10.35", default-features = false }
 openssl-probe = { version = "0.1.4", default-features = false }
 percent-encoding = { version = "2.1.0", default-features = false }
@@ -307,6 +312,8 @@ tower-test = "0.4.0"
 walkdir = "2.3.2"
 quickcheck = "1.0.3"
 lookup = { path = "lib/lookup", features = ["arbitrary"] }
+azure_core = { git = "https://github.com/Azure/azure-sdk-for-rust.git", rev = "16bcf0ab1bb6e380d966a69d314de1e99ede553a", features = ["azurite_workaround"] }
+azure_storage = { git = "https://github.com/Azure/azure-sdk-for-rust.git", rev = "16bcf0ab1bb6e380d966a69d314de1e99ede553a", features = ["azurite_workaround"] }
 
 
 [patch.crates-io]
@@ -540,6 +547,7 @@ sinks-logs = [
   "sinks-aws_kinesis_streams",
   "sinks-aws_s3",
   "sinks-aws_sqs",
+  "sinks-azure_blob",
   "sinks-azure_monitor_logs",
   "sinks-blackhole",
   "sinks-clickhouse",
@@ -584,6 +592,7 @@ sinks-aws_kinesis_firehose = ["rusoto", "rusoto_firehose"]
 sinks-aws_kinesis_streams = ["rusoto", "rusoto_kinesis"]
 sinks-aws_s3 = ["base64", "bytesize", "md-5", "rusoto", "rusoto_s3", "uuid"]
 sinks-aws_sqs = ["rusoto", "rusoto_sqs"]
+sinks-azure_blob = ["bytesize", "azure_core", "azure_storage", "reqwest", "uuid"]
 sinks-azure_monitor_logs = ["bytesize"]
 sinks-blackhole = []
 sinks-clickhouse = ["bytesize"]
@@ -650,6 +659,10 @@ aws-integration-tests = [
   "aws-sqs-integration-tests",
 ]
 
+azure-integration-tests = [
+  "azure-blob-integration-tests"
+]
+
 aws-cloudwatch-logs-integration-tests = ["sinks-aws_cloudwatch_logs"]
 aws-cloudwatch-metrics-integration-tests = ["sinks-aws_cloudwatch_metrics"]
 aws-ec2-metadata-integration-tests = ["transforms-aws_ec2_metadata"]
@@ -658,6 +671,7 @@ aws-kinesis-firehose-integration-tests = ["rusoto_es", "sinks-aws_kinesis_fireho
 aws-kinesis-streams-integration-tests = ["sinks-aws_kinesis_streams"]
 aws-s3-integration-tests = ["sinks-aws_s3", "sources-aws_s3"]
 aws-sqs-integration-tests = ["sinks-aws_sqs"]
+azure-blob-integration-tests = ["sinks-azure_blob"]
 clickhouse-integration-tests = ["sinks-clickhouse", "warp"]
 docker-logs-integration-tests = ["sources-docker_logs", "unix"]
 es-integration-tests = ["sinks-elasticsearch"]

Makefile

@@ -301,7 +301,7 @@ test-behavior: ## Runs behaviorial test
 
 .PHONY: test-integration
 test-integration: ## Runs all integration tests
-test-integration: test-integration-aws test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
+test-integration: test-integration-aws test-integration-azure test-integration-clickhouse test-integration-docker-logs test-integration-elasticsearch
 test-integration: test-integration-fluent test-integration-gcp test-integration-humio test-integration-influxdb test-integration-kafka
 test-integration: test-integration-logstash test-integration-loki test-integration-mongodb_metrics test-integration-nats
 test-integration: test-integration-nginx test-integration-postgresql_metrics test-integration-prometheus test-integration-pulsar
@@ -319,6 +319,18 @@ ifeq ($(AUTODESPAWN), true)
 	@scripts/setup_integration_env.sh aws stop
 endif
 
+.PHONY: test-integration-azure
+test-integration-azure: ## Runs Azure integration tests
+ifeq ($(AUTOSPAWN), true)
+	@scripts/setup_integration_env.sh azure stop
+	@scripts/setup_integration_env.sh azure start
+	sleep 5 # Many services are very slow... Give them a sec...
+endif
+	${MAYBE_ENVIRONMENT_EXEC} cargo test --no-fail-fast --no-default-features --features azure-integration-tests --lib ::azure_ -- --nocapture
+ifeq ($(AUTODESPAWN), true)
+	@scripts/setup_integration_env.sh azure stop
+endif
+
 .PHONY: test-integration-clickhouse
 test-integration-clickhouse: ## Runs Clickhouse integration tests
 ifeq ($(AUTOSPAWN), true)

scripts/setup_integration/azure_integration_env.sh

@@ -0,0 +1,48 @@
+#!/usr/bin/env bash
+set -o pipefail
+
+# aws_integration_env.sh
+#
+# SUMMARY
+#
+#   Builds and pulls down the Vector AWS Integration test environment
+
+set -x
+
+if [ $# -ne 1 ]; then
+  echo "Usage: $0 {stop|start}" 1>&2
+  exit 1
+  exit 1
+fi
+ACTION=$1
+
+#
+# Functions
+#
+
+start_podman() {
+  podman pod create --replace --name vector-test-integration-azure -p 10000-10001:10000-10001
+  podman run -d --pod=vector-test-integration-azure --name vector_local_azure_blob \
+    mcr.microsoft.com/azure-storage/azurite:3.11.0 azurite --blobHost 0.0.0.0 --loose
+}
+
+start_docker() {
+  docker network create vector-test-integration-azure
+  docker run -d --network=vector-test-integration-azure -v /var/run:/var/run -p 10000-10001:10000-10001 --name vector_local_azure_blob \
+    mcr.microsoft.com/azure-storage/azurite:3.11.0 azurite --blobHost 0.0.0.0 --loose
+}
+
+stop_podman() {
+  podman rm --force vector_local_azure_blob 2>/dev/null; true
+  podman pod stop vector-test-integration-azure 2>/dev/null; true
+  podman pod rm --force vector-test-integration-azure 2>/dev/null; true
+}
+
+stop_docker() {
+  docker rm --force vector_local_azure_blob 2>/dev/null; true
+  docker network rm vector-test-integration-azure 2>/dev/null; true
+}
+
+echo "Running $ACTION action for AWS integration tests environment"
+
+"${ACTION}"_"${CONTAINER_TOOL}"

src/internal_events/azure_blob.rs

@@ -0,0 +1,52 @@
+use super::InternalEvent;
+use metrics::counter;
+use uuid::Uuid;
+
+#[derive(Debug)]
+pub struct AzureBlobErrorResponse {
+    pub code: hyper::StatusCode,
+}
+
+impl InternalEvent for AzureBlobErrorResponse {
+    fn emit_logs(&self) {
+        error!(message = "HTTP error response.", code = %self.code);
+    }
+
+    fn emit_metrics(&self) {
+        counter!("http_error_response_total", 1);
+    }
+}
+
+#[derive(Debug)]
+pub(crate) struct AzureBlobHttpError {
+    pub error: String,
+}
+
+impl InternalEvent for AzureBlobHttpError {
+    fn emit_logs(&self) {
+        error!(
+            message = "Error processing request.",
+            error = %self.error,
+            internal_log_rate_secs = 10
+        );
+    }
+
+    fn emit_metrics(&self) {
+        counter!("http_request_errors_total", 1);
+    }
+}
+
+pub(crate) struct AzureBlobEventSent {
+    pub request_id: Uuid,
+    pub byte_size: usize,
+}
+
+impl InternalEvent for AzureBlobEventSent {
+    fn emit_logs(&self) {
+        trace!(message = "Event sent.", request_id = %self.request_id);
+    }
+
+    fn emit_metrics(&self) {
+        counter!("processed_bytes_total", self.byte_size as u64);
+    }
+}

src/internal_events/mod.rs

@@ -22,6 +22,8 @@ mod aws_kinesis_streams;
 pub(crate) mod aws_s3;
 #[cfg(feature = "sinks-aws_sqs")]
 mod aws_sqs;
+#[cfg(feature = "sinks-azure_blob")]
+pub(crate) mod azure_blob;
 mod blackhole;
 #[cfg(feature = "transforms-coercer")]
 mod coercer;