Lowercase options before they're passed through to object store

This also introduces the S3_OUTPUT_URL environment variable

Author: rtyler

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "hotdog"
-version = "1.0.1"
+version = "1.0.2"
 authors = ["R. Tyler Croy <rtyler@buoyantdata.com>"]
 edition = "2024"
 

README.adoc

@@ -226,6 +226,29 @@ configured Kafka brokers.
 "default topic" for the <<action-forward, Forward action>>.
 
 
+[[yml-parquet]]
+==== Parquet
+
+The link:https://parquet.apache.org[Apache Parquet] sink allows for directly
+writing to an
+link:https://docs.rs/object_store/latest/object_store/index.html[object_store]
+supported `url`
+
+[source,yaml]
+----
+global:
+  parquet:
+    url: 's3://hotdog/streams/'
+    # Bytes to buffer
+    buffer: 1024000
+    flush_ms: 60000
+----
+
+[TIP]
+====
+The `url` can be omitted from the configuration and specified in the environment via `S3_OUTPUT_URL`
+====
+
 [[yml-metrics]]
 ==== Metrics
 

src/sink/parquet.rs

@@ -14,6 +14,7 @@ use parquet::arrow::async_writer::{AsyncArrowWriter, ParquetObjectWriter};
 use smol::stream::StreamExt;
 use tracing::log::*;
 use tracing::{Level, span};
+use url::Url;
 use uuid::Uuid;
 
 use std::collections::HashMap;
@@ -44,7 +45,11 @@ impl Sink for Parquet {
 
     fn new(config: Self::Config, _stats: InputQueueScope) -> Self {
         let (tx, rx) = bounded(1024);
-        let opts: HashMap<String, String> = HashMap::from_iter(std::env::vars());
+        // [object_store] largely expects environment variables to be all lowercased for
+        // consideration as options
+        let opts: HashMap<String, String> = HashMap::from_iter(
+            std::env::vars().map(|(k, v)| (k.to_ascii_lowercase(), v))
+        );
         let (store, _path) = object_store::parse_url_opts(&config.url, opts)
             .expect("Failed to parse the Parquet sink URL");
         let store = Arc::new(store);
@@ -181,8 +186,9 @@ fn flush_to_parquet(store: ObjectStoreRef, destination: &str, buffer: Vec<String
 /// Configuration for [Parquet] sink
 #[derive(Clone, Debug, Deserialize, PartialEq)]
 pub struct Config {
+    #[serde(default = "parquet_url_default")]
     /// Expected to be an S3 compatible URL
-    pub url: url::Url,
+    pub url: Url,
     /// Minimum number of bytes to buffer into each parquet file
     #[serde(default = "parquet_buffer_default")]
     pub buffer: usize,
@@ -191,6 +197,14 @@ pub struct Config {
     pub flush_ms: usize,
 }
 
+/// Retrieves a URL from the environment for parquet if no [Url] has been specified
+fn parquet_url_default() -> Url {
+    Url::parse(&std::env::var("S3_OUTPUT_URL").expect(
+        "There is no url: defined for the parquet sink and no S3_OUTPUT_URL in the environment!",
+    ))
+    .expect("The S3_OUTPUT_URL could not be parsed as a valid URL")
+}
+
 /// Default number of log lines per parquet file
 fn parquet_buffer_default() -> usize {
     1_024 * 1_024 * 100