Fix failing tests

Author: joshcoughlan

LICENSE-3rdparty.csv

@@ -1,6 +1,5 @@
 Component,Origin,License,Copyright
 Inflector,https://github.com/whatisinternet/inflector,BSD-2-Clause,Josh Teeter<[email protected]>
-RustyXML,https://github.com/Florob/RustyXML,MIT OR Apache-2.0,Florian Zeitz <[email protected]>
 addr2line,https://github.com/gimli-rs/addr2line,Apache-2.0 OR MIT,The addr2line Authors
 adler2,https://github.com/oyvindln/adler2,0BSD OR MIT OR Apache-2.0,"Jonas Schievink <[email protected]>, oyvindln <[email protected]>"
 adler32,https://github.com/remram44/adler32-rs,Zlib,Remi Rampin <[email protected]>
@@ -104,11 +103,8 @@ aws-types,https://github.com/smithy-lang/smithy-rs,Apache-2.0,"AWS Rust SDK Team
 axum,https://github.com/tokio-rs/axum,MIT,The axum Authors
 axum-core,https://github.com/tokio-rs/axum,MIT,The axum-core Authors
 azure_core,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft
-azure_core,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft Corp.
-azure_identity,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft
-azure_storage,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft Corp.
-azure_storage_blobs,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft Corp.
-azure_svc_blobstorage,https://github.com/azure/azure-sdk-for-rust,MIT,The azure_svc_blobstorage Authors
+azure_core_macros,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft
+azure_storage_blob,https://github.com/azure/azure-sdk-for-rust,MIT,Microsoft
 backoff,https://github.com/ihrwein/backoff,MIT OR Apache-2.0,Tibor Benke <[email protected]>
 backon,https://github.com/Xuanwo/backon,Apache-2.0,The backon Authors
 backtrace,https://github.com/rust-lang/backtrace-rs,MIT OR Apache-2.0,The Rust Project Developers
@@ -358,7 +354,6 @@ http-body,https://github.com/hyperium/http-body,MIT,"Carl Lerche <me@carllerche.
 http-body-util,https://github.com/hyperium/http-body,MIT,"Carl Lerche <[email protected]>, Lucio Franco <[email protected]>, Sean McArthur <[email protected]>"
 http-range-header,https://github.com/MarcusGrass/parse-range-headers,MIT,The http-range-header Authors
 http-serde,https://gitlab.com/kornelski/http-serde,Apache-2.0 OR MIT,Kornel <[email protected]>
-http-types,https://github.com/http-rs/http-types,MIT OR Apache-2.0,Yoshua Wuyts <[email protected]>
 httparse,https://github.com/seanmonstar/httparse,MIT OR Apache-2.0,Sean McArthur <[email protected]>
 httpdate,https://github.com/pyfisch/httpdate,MIT OR Apache-2.0,Pyfisch <[email protected]>
 humantime,https://github.com/chronotope/humantime,MIT OR Apache-2.0,The humantime Authors
@@ -389,7 +384,6 @@ idna_adapter,https://github.com/hsivonen/idna_adapter,Apache-2.0 OR MIT,The rust
 indexmap,https://github.com/bluss/indexmap,Apache-2.0 OR MIT,The indexmap Authors
 indexmap,https://github.com/indexmap-rs/indexmap,Apache-2.0 OR MIT,The indexmap Authors
 indoc,https://github.com/dtolnay/indoc,MIT OR Apache-2.0,David Tolnay <[email protected]>
-infer,https://github.com/bojand/infer,MIT,Bojan <[email protected]>
 influxdb-line-protocol,https://github.com/influxdata/influxdb_iox/tree/main/influxdb_line_protocol,MIT OR Apache-2.0,InfluxDB IOx Project Developers
 inotify,https://github.com/hannobraun/inotify,ISC,"Hanno Braun <[email protected]>, Félix Saparelli <[email protected]>, Cristian Kubis <[email protected]>, Frank Denis <[email protected]>"
 inotify-sys,https://github.com/hannobraun/inotify-sys,ISC,Hanno Braun <[email protected]>
@@ -404,6 +398,7 @@ ipconfig,https://github.com/liranringel/ipconfig,MIT OR Apache-2.0,Liran Ringel
 ipcrypt-rs,https://github.com/jedisct1/rust-ipcrypt2,ISC,Frank Denis <[email protected]>
 ipnet,https://github.com/krisprice/ipnet,MIT OR Apache-2.0,Kris Price <[email protected]>
 ipnetwork,https://github.com/achanda/ipnetwork,MIT OR Apache-2.0,"Abhishek Chanda <[email protected]>, Linus Färnstrand <[email protected]>"
+iri-string,https://github.com/lo48576/iri-string,MIT OR Apache-2.0,YOSHIOKA Takuma <[email protected]>
 is-terminal,https://github.com/sunfishcode/is-terminal,MIT,"softprops <[email protected]>, Dan Gohman <[email protected]>"
 is_ci,https://github.com/zkat/is_ci,ISC,Kat Marchán <[email protected]>
 itertools,https://github.com/rust-itertools/itertools,MIT OR Apache-2.0,bluss
@@ -621,7 +616,6 @@ rand,https://github.com/rust-random/rand,MIT OR Apache-2.0,"The Rand Project Dev
 rand_chacha,https://github.com/rust-random/rand,MIT OR Apache-2.0,"The Rand Project Developers, The Rust Project Developers, The CryptoCorrosion Contributors"
 rand_core,https://github.com/rust-random/rand,MIT OR Apache-2.0,"The Rand Project Developers, The Rust Project Developers"
 rand_distr,https://github.com/rust-random/rand_distr,MIT OR Apache-2.0,The Rand Project Developers
-rand_hc,https://github.com/rust-random/rand,MIT OR Apache-2.0,The Rand Project Developers
 rand_xorshift,https://github.com/rust-random/rngs,MIT OR Apache-2.0,"The Rand Project Developers, The Rust Project Developers"
 ratatui,https://github.com/ratatui/ratatui,MIT,"Florian Dehau <[email protected]>, The Ratatui Developers"
 raw-cpuid,https://github.com/gz/rust-cpuid,MIT,Gerd Zellweger <[email protected]>
@@ -700,7 +694,6 @@ serde_json,https://github.com/serde-rs/json,MIT OR Apache-2.0,"Erick Tryzelaar <
 serde_nanos,https://github.com/caspervonb/serde_nanos,MIT OR Apache-2.0,Casper Beyer <[email protected]>
 serde_path_to_error,https://github.com/dtolnay/path-to-error,MIT OR Apache-2.0,David Tolnay <[email protected]>
 serde_plain,https://github.com/mitsuhiko/serde-plain,MIT OR Apache-2.0,Armin Ronacher <[email protected]>
-serde_qs,https://github.com/samscott89/serde_qs,MIT OR Apache-2.0,Sam Scott <[email protected]>
 serde_repr,https://github.com/dtolnay/serde-repr,MIT OR Apache-2.0,David Tolnay <[email protected]>
 serde_spanned,https://github.com/toml-rs/toml,MIT OR Apache-2.0,The serde_spanned Authors
 serde_urlencoded,https://github.com/nox/serde_urlencoded,MIT OR Apache-2.0,Anthony Ramine <[email protected]>
@@ -876,6 +869,7 @@ wasm-timer,https://github.com/tomaka/wasm-timer,MIT,Pierre Krieger <pierre.krieg
 web-sys,https://github.com/rustwasm/wasm-bindgen/tree/master/crates/web-sys,MIT OR Apache-2.0,The wasm-bindgen Developers
 web-time,https://github.com/daxpedda/web-time,MIT OR Apache-2.0,The web-time Authors
 webbrowser,https://github.com/amodm/webbrowser-rs,MIT OR Apache-2.0,Amod Malviya @amodm
+webpki-roots,https://github.com/rustls/webpki-roots,CDLA-Permissive-2.0,The webpki-roots Authors
 webpki-roots,https://github.com/rustls/webpki-roots,MPL-2.0,The webpki-roots Authors
 whoami,https://github.com/ardaku/whoami,Apache-2.0 OR BSL-1.0 OR MIT,The whoami Authors
 widestring,https://github.com/starkat99/widestring-rs,MIT OR Apache-2.0,Kathryn Long <[email protected]>

src/sinks/azure_blob/integration_tests.rs

@@ -1,11 +1,7 @@
-use std::{
-    io::{BufRead, BufReader},
-    num::NonZeroU32,
-};
+use std::io::{BufRead, BufReader};
 
 use azure_core::http::StatusCode;
 
-use azure_storage_blob::BlobContainerClient;
 use bytes::{Buf, BytesMut};
 use flate2::read::GzDecoder;
 use futures::{Stream, StreamExt, stream};
@@ -83,7 +79,7 @@ async fn azure_blob_insert_lines_into_blob() {
     let blobs = config.list_blobs(blob_prefix).await;
     assert_eq!(blobs.len(), 1);
     assert!(blobs[0].clone().ends_with(".log"));
-    let (content_type, content_encoding, blob_lines) = config.get_blob(blobs[0].clone()).await;
+    let (content_type, _content_encoding, blob_lines) = config.get_blob(blobs[0].clone()).await;
     assert_eq!(content_type, Some(String::from("text/plain")));
     assert_eq!(lines, blob_lines);
 }
@@ -138,9 +134,9 @@ async fn azure_blob_insert_lines_into_blob_gzip() {
     let blobs = config.list_blobs(blob_prefix).await;
     assert_eq!(blobs.len(), 1);
     assert!(blobs[0].clone().ends_with(".log.gz"));
-    let (blob, blob_lines) = config.get_blob(blobs[0].clone()).await;
-    assert_eq!(blob.properties.content_encoding, Some(String::from("gzip")));
-    assert_eq!(blob.properties.content_type, String::from("text/plain"));
+    let (content_type, content_encoding, blob_lines) = config.get_blob(blobs[0].clone()).await;
+    assert_eq!(content_encoding, Some(String::from("gzip")));
+    assert_eq!(content_type, Some(String::from("text/plain")));
     assert_eq!(lines, blob_lines);
 }
 
@@ -168,12 +164,9 @@ async fn azure_blob_insert_json_into_blob_gzip() {
     let blobs = config.list_blobs(blob_prefix).await;
     assert_eq!(blobs.len(), 1);
     assert!(blobs[0].clone().ends_with(".log.gz"));
-    let (blob, blob_lines) = config.get_blob(blobs[0].clone()).await;
-    assert_eq!(blob.properties.content_encoding, Some(String::from("gzip")));
-    assert_eq!(
-        blob.properties.content_type,
-        String::from("application/x-ndjson")
-    );
+    let (content_type, content_encoding, blob_lines) = config.get_blob(blobs[0].clone()).await;
+    assert_eq!(content_encoding, Some(String::from("gzip")));
+    assert_eq!(content_type, Some(String::from("application/x-ndjson")));
     let expected = events
         .iter()
         .map(|event| serde_json::to_string(&event.as_log().all_event_fields().unwrap()).unwrap())
@@ -204,7 +197,7 @@ async fn azure_blob_rotate_files_after_the_buffer_size_is_reached() {
     assert_eq!(blobs.len(), 3);
     let response = stream::iter(blobs)
         .fold(Vec::new(), |mut acc, blob| async {
-            let (_, lines) = config.get_blob(blob).await;
+            let (_, _, lines) = config.get_blob(blob).await;
             acc.push(lines);
             acc
         })
@@ -260,32 +253,24 @@ impl AzureBlobSinkConfig {
         )
         .unwrap();
 
-        // Use new SDK pager to fetch first page and collect blob names.
+        // Iterate pager results and collect blob names. Filter by prefix server-side.
         let mut pager = client
             .list_blobs(None)
             .expect("Failed to start list blobs pager");
-        let page = pager
-            .next()
-            .await
-            .expect("Failed to fetch blobs")
-            .into_body();
-
-        // Best-effort extraction of names from the page body.
-        // Depending on SDK struct names, this may need tweaking:
-        // ListBlobsFlatSegmentResponse { segment: { blob_items: [{ name, .. }, ..] }, .. }
-        let names = page
-            .segment
-            .blob_items
-            .into_iter()
-            .map(|b| b.name)
-            .filter(|name| name.starts_with(&prefix))
-            .collect::<Vec<_>>();
+        let mut names = Vec::new();
+        while let Some(result) = pager.next().await {
+            let item = result.expect("Failed to fetch blobs");
+            if let Some(name) = item.name.and_then(|bn| bn.content)
+                && name.starts_with(&prefix)
+            {
+                names.push(name);
+            }
+        }
 
         names
     }
 
     pub async fn get_blob(&self, blob: String) -> (Option<String>, Option<String>, Vec<String>) {
-        use azure_storage_blob::clients::BlobClient as _;
         let client = azure_common::config::build_client(
             self.connection_string.clone().into(),
             self.container_name.clone(),
@@ -295,27 +280,39 @@ impl AzureBlobSinkConfig {
         let blob_client = client.blob_client(&blob);
 
         // Fetch properties to obtain content-type and content-encoding
-        let props = blob_client
+        let props_resp = blob_client
             .get_properties(None)
             .await
-            .expect("Failed to get blob properties")
-            .into_body();
-
-        let content_type = props.content_type.clone();
-        let content_encoding = props.content_encoding.clone();
+            .expect("Failed to get blob properties");
+        let headers = props_resp.headers();
+        let content_type = headers.iter().find_map(|(name, value)| {
+            let key = name.as_str();
+            if key.eq_ignore_ascii_case("content-type") {
+                Some(value.as_str().to_string())
+            } else {
+                None
+            }
+        });
+        let content_encoding = headers.iter().find_map(|(name, value)| {
+            let key = name.as_str();
+            if key.eq_ignore_ascii_case("content-encoding") {
+                Some(value.as_str().to_string())
+            } else {
+                None
+            }
+        });
 
         // Download blob content (full or first MB as needed)
         let downloaded = blob_client
             .download(None)
             .await
             .expect("Failed to download blob");
-        let data = downloaded
+        let body_bytes = downloaded
             .into_body()
-            .data
             .collect()
             .await
-            .expect("Failed to read blob body")
-            .to_vec();
+            .expect("Failed to read blob body");
+        let data = body_bytes.to_vec();
 
         (content_type, content_encoding, self.get_blob_content(data))
     }
@@ -344,7 +341,7 @@ impl AzureBlobSinkConfig {
         let response = match result {
             Ok(_) => Ok(()),
             Err(error) => match error.http_status() {
-                Some(status) if status.as_u16() == StatusCode::Conflict => Ok(()),
+                Some(StatusCode::Conflict) => Ok(()),
                 _ => Err(error),
             },
         };

src/sinks/azure_common/connection_string.rs

@@ -33,17 +33,17 @@ SAS handling
 Examples:
 - Access key connection string:
   "DefaultEndpointsProtocol=https;AccountName=myacct;AccountKey=base64key==;EndpointSuffix=core.windows.net"
-  Container URL: https://myacct.blob.core.windows.net/logs
-  Blob URL: https://myacct.blob.core.windows.net/logs/file.txt
+  Container URL: <https://myacct.blob.core.windows.net/logs>
+  Blob URL: <https://myacct.blob.core.windows.net/logs/file.txt>
 
 - SAS connection string:
-  "BlobEndpoint=https://myacct.blob.core.windows.net/;SharedAccessSignature=sv=2022-11-02&ss=b&..."
-  Container URL (with SAS): https://myacct.blob.core.windows.net/logs?sv=2022-11-02&ss=b&...
-  Blob URL (with SAS): https://myacct.blob.core.windows.net/logs/file.txt?sv=2022-11-02&ss=b&...
+  "BlobEndpoint=<https://myacct.blob.core.windows.net/>;SharedAccessSignature=sv=2022-11-02&ss=b&..."
+  Container URL (with SAS): <https://myacct.blob.core.windows.net/logs?sv=2022-11-02&ss=b&...>
+  Blob URL (with SAS): <https://myacct.blob.core.windows.net/logs/file.txt?sv=2022-11-02&ss=b&...>
 
 - Azurite/dev storage:
   "UseDevelopmentStorage=true;DefaultEndpointsProtocol=http;AccountName=devstoreaccount1"
-  Container URL: http://127.0.0.1:10000/devstoreaccount1/logs
+  Container URL: <http://127.0.0.1:10000/devstoreaccount1/logs>
 */
 
 use std::collections::HashMap;
@@ -87,7 +87,7 @@ pub enum Auth {
 }
 
 /// A parsed Azure Storage connection string and helpers to compose URLs for containers/blobs.
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Default)]
 pub struct ParsedConnectionString {
     pub account_name: Option<String>,
     pub account_key: Option<String>,
@@ -99,21 +99,6 @@ pub struct ParsedConnectionString {
     pub development_storage_proxy_uri: Option<String>,
 }
 
-impl Default for ParsedConnectionString {
-    fn default() -> Self {
-        Self {
-            account_name: None,
-            account_key: None,
-            shared_access_signature: None,
-            default_endpoints_protocol: None,
-            endpoint_suffix: None,
-            blob_endpoint: None,
-            use_development_storage: false,
-            development_storage_proxy_uri: None,
-        }
-    }
-}
-
 impl ParsedConnectionString {
     /// Parse a connection string into a `ParsedConnectionString`.
     ///
@@ -135,28 +120,23 @@ impl ParsedConnectionString {
         }
 
         // Build the structure from the parsed map.
-        let mut parsed = ParsedConnectionString::default();
-
-        parsed.account_name = map.get("accountname").cloned();
-        parsed.account_key = map.get("accountkey").cloned();
-        parsed.shared_access_signature = map
-            .get("sharedaccesssignature")
-            .map(|s| normalize_sas(s.as_str()));
-
-        parsed.default_endpoints_protocol = map
-            .get("defaultendpointsprotocol")
-            .map(|s| s.to_ascii_lowercase());
-
-        parsed.endpoint_suffix = map.get("endpointsuffix").cloned();
-
-        parsed.blob_endpoint = map.get("blobendpoint").cloned();
-
-        parsed.use_development_storage = map
-            .get("usedevelopmentstorage")
-            .map(|v| v.eq_ignore_ascii_case("true"))
-            .unwrap_or(false);
-
-        parsed.development_storage_proxy_uri = map.get("developmentstorageproxyuri").cloned();
+        let parsed = ParsedConnectionString {
+            account_name: map.get("accountname").cloned(),
+            account_key: map.get("accountkey").cloned(),
+            shared_access_signature: map
+                .get("sharedaccesssignature")
+                .map(|s| normalize_sas(s.as_str())),
+            default_endpoints_protocol: map
+                .get("defaultendpointsprotocol")
+                .map(|s| s.to_ascii_lowercase()),
+            endpoint_suffix: map.get("endpointsuffix").cloned(),
+            blob_endpoint: map.get("blobendpoint").cloned(),
+            use_development_storage: map
+                .get("usedevelopmentstorage")
+                .map(|v| v.eq_ignore_ascii_case("true"))
+                .unwrap_or(false),
+            development_storage_proxy_uri: map.get("developmentstorageproxyuri").cloned(),
+        };
 
         Ok(parsed)
     }
@@ -256,15 +236,18 @@ impl ParsedConnectionString {
 
     /// Build a blob URL, optionally appending SAS if present.
     pub fn blob_url(&self, container: &str, blob: &str) -> Result<String, ConnectionStringError> {
-        let container_url = self.container_url(container)?;
+        // Build the base container URL without SAS, then append the blob path,
+        // and finally append the SAS so it appears after the full path.
+        let base = self.blob_account_endpoint()?;
+        let container_no_sas = format!("{}/{}", trim_trailing_slash(&base), container);
+        let blob_full = format!(
+            "{}/{}",
+            trim_trailing_slash(&container_no_sas),
+            encode_path_segment(blob)
+        );
         Ok(append_query_segment(
-            &format!(
-                "{}/{}",
-                trim_trailing_slash(&container_url),
-                encode_path_segment(blob)
-            ),
-            // container_url already handled SAS; if it already had query args, append with '&'
-            None, // SAS already appended at container level if present
+            &blob_full,
+            self.shared_access_signature.as_deref(),
         ))
     }
 }
@@ -278,7 +261,7 @@ fn normalize_sas(s: &str) -> String {
 fn append_query_segment(base_url: &str, sas: Option<&str>) -> String {
     match sas {
         None => base_url.to_string(),
-        Some(q) if q.is_empty() => base_url.to_string(),
+        Some("") => base_url.to_string(),
         Some(q) => {
             let sep = if base_url.contains('?') { '&' } else { '?' };
             format!("{base_url}{sep}{q}")
@@ -288,8 +271,8 @@ fn append_query_segment(base_url: &str, sas: Option<&str>) -> String {
 
 /// Trim exactly one trailing slash from a string, if present.
 fn trim_trailing_slash(s: &str) -> String {
-    if s.ends_with('/') {
-        s[..s.len() - 1].to_string()
+    if let Some(stripped) = s.strip_suffix('/') {
+        stripped.to_string()
     } else {
         s.to_string()
     }