diff --git a/Cargo.lock b/Cargo.lock index 6257c794..152c51e4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1201,7 +1201,7 @@ dependencies = [ "aws-sdk-ssooidc", "aws-sdk-sts", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1221,9 +1221,9 @@ dependencies = [ [[package]] name = "aws-credential-types" -version = "1.2.7" +version = "1.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "799a1290207254984cb7c05245111bc77958b92a3c9bb449598044b36341cce6" +checksum = "3cd362783681b15d136480ad555a099e82ecd8e2d10a841e14dfd0078d67fee3" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", @@ -1255,15 +1255,15 @@ dependencies = [ [[package]] name = "aws-runtime" -version = "1.5.10" +version = "1.5.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c034a1bc1d70e16e7f4e4caf7e9f7693e4c9c24cd91cf17c2a0b21abaebc7c8b" +checksum = "7ce527fb7e53ba9626fc47824f25e256250556c40d8f81d27dd92aa38239d632" dependencies = [ "aws-credential-types", "aws-sigv4", "aws-smithy-async", "aws-smithy-eventstream", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -1288,7 +1288,7 @@ dependencies = [ "aws-runtime", "aws-smithy-async", "aws-smithy-compression", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-query", "aws-smithy-runtime", @@ -1314,7 +1314,7 @@ dependencies = [ "aws-runtime", "aws-smithy-async", "aws-smithy-eventstream", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1336,7 +1336,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1358,7 +1358,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1381,7 +1381,7 @@ dependencies = [ "aws-runtime", "aws-smithy-async", "aws-smithy-eventstream", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1403,7 +1403,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1428,7 +1428,7 @@ dependencies = [ "aws-smithy-async", "aws-smithy-checksums", "aws-smithy-eventstream", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1459,7 +1459,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1481,7 +1481,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-query", "aws-smithy-runtime", @@ -1504,7 +1504,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1526,7 +1526,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1548,7 +1548,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-runtime", "aws-smithy-runtime-api", @@ -1570,7 +1570,7 @@ dependencies = [ "aws-credential-types", "aws-runtime", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-json", "aws-smithy-query", "aws-smithy-runtime", @@ -1586,13 +1586,13 @@ dependencies = [ [[package]] name = "aws-sigv4" -version = "1.3.4" +version = "1.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "084c34162187d39e3740cb635acd73c4e3a551a36146ad6fe8883c929c9f876c" +checksum = "efa49f3c607b92daae0c078d48a4571f599f966dce3caee5f1ea55c4d9073f99" dependencies = [ "aws-credential-types", "aws-smithy-eventstream", - "aws-smithy-http", + "aws-smithy-http 0.63.3", "aws-smithy-runtime-api", "aws-smithy-types", "bytes 1.10.1", @@ -1609,9 +1609,9 @@ dependencies = [ [[package]] name = "aws-smithy-async" -version = "1.2.5" +version = "1.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e190749ea56f8c42bf15dd76c65e14f8f765233e6df9b0506d9d934ebef867c" +checksum = "52eec3db979d18cb807fc1070961cc51d87d069abe9ab57917769687368a8c6c" dependencies = [ "futures-util", "pin-project-lite", @@ -1624,7 +1624,7 @@ version = "0.63.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4dbef71cd3cf607deb5c407df52f7e589e6849b296874ee448977efbb6d0832b" dependencies = [ - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-types", "bytes 1.10.1", "crc-fast", @@ -1657,9 +1657,9 @@ dependencies = [ [[package]] name = "aws-smithy-eventstream" -version = "0.60.10" +version = "0.60.18" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "604c7aec361252b8f1c871a7641d5e0ba3a7f5a586e51b66bc9510a5519594d9" +checksum = "35b9c7354a3b13c66f60fe4616d6d1969c9fd36b1b5333a5dfb3ee716b33c588" dependencies = [ "aws-smithy-types", "bytes 1.10.1", @@ -1668,9 +1668,9 @@ dependencies = [ [[package]] name = "aws-smithy-http" -version = "0.62.3" +version = "0.62.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c4dacf2d38996cf729f55e7a762b30918229917eca115de45dfa8dfb97796c9" +checksum = "826141069295752372f8203c17f28e30c464d22899a43a0c9fd9c458d469c88b" dependencies = [ "aws-smithy-eventstream", "aws-smithy-runtime-api", @@ -1678,6 +1678,7 @@ dependencies = [ "bytes 1.10.1", "bytes-utils", "futures-core", + "futures-util", "http 0.2.12", "http 1.3.1", "http-body 0.4.6", @@ -1687,11 +1688,32 @@ dependencies = [ "tracing 0.1.41", ] +[[package]] +name = "aws-smithy-http" +version = "0.63.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "630e67f2a31094ffa51b210ae030855cb8f3b7ee1329bdd8d085aaf61e8b97fc" +dependencies = [ + "aws-smithy-runtime-api", + "aws-smithy-types", + "bytes 1.10.1", + "bytes-utils", + "futures-core", + "futures-util", + "http 1.3.1", + "http-body 1.0.1", + "http-body-util", + "percent-encoding", + "pin-project-lite", + "pin-utils", + "tracing 0.1.41", +] + [[package]] name = "aws-smithy-http-client" -version = "1.1.2" +version = "1.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "734b4282fbb7372923ac339cc2222530f8180d9d4745e582de19a18cee409fd8" +checksum = "12fb0abf49ff0cab20fd31ac1215ed7ce0ea92286ba09e2854b42ba5cabe7525" dependencies = [ "aws-smithy-async", "aws-smithy-runtime-api", @@ -1728,9 +1750,9 @@ dependencies = [ [[package]] name = "aws-smithy-observability" -version = "0.1.3" +version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9364d5989ac4dd918e5cc4c4bdcc61c9be17dcd2586ea7f69e348fc7c6cab393" +checksum = "c0a46543fbc94621080b3cf553eb4cbbdc41dd9780a30c4756400f0139440a1d" dependencies = [ "aws-smithy-runtime-api", ] @@ -1747,12 +1769,12 @@ dependencies = [ [[package]] name = "aws-smithy-runtime" -version = "1.9.2" +version = "1.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4fa63ad37685ceb7762fa4d73d06f1d5493feb88e3f27259b9ed277f4c01b185" +checksum = "f3df87c14f0127a0d77eb261c3bc45d5b4833e2a1f63583ebfb728e4852134ee" dependencies = [ "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.63.3", "aws-smithy-http-client", "aws-smithy-observability", "aws-smithy-runtime-api", @@ -1763,6 +1785,7 @@ dependencies = [ "http 1.3.1", "http-body 0.4.6", "http-body 1.0.1", + "http-body-util", "pin-project-lite", "pin-utils", "tokio", @@ -1771,9 +1794,9 @@ dependencies = [ [[package]] name = "aws-smithy-runtime-api" -version = "1.9.0" +version = "1.11.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "07f5e0fc8a6b3f2303f331b94504bbf754d85488f402d6f1dd7a6080f99afe56" +checksum = "49952c52f7eebb72ce2a754d3866cc0f87b97d2a46146b79f80f3a93fb2b3716" dependencies = [ "aws-smithy-async", "aws-smithy-types", @@ -1788,9 +1811,9 @@ dependencies = [ [[package]] name = "aws-smithy-types" -version = "1.3.2" +version = "1.4.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d498595448e43de7f4296b7b7a18a8a02c61ec9349128c80a368f7c3b4ab11a8" +checksum = "3b3a26048eeab0ddeba4b4f9d51654c79af8c3b32357dc5f336cee85ab331c33" dependencies = [ "base64-simd", "bytes 1.10.1", @@ -1823,9 +1846,9 @@ dependencies = [ [[package]] name = "aws-types" -version = "1.3.8" +version = "1.3.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b069d19bf01e46298eaedd7c6f283fe565a59263e53eebec945f3e6398f42390" +checksum = "1d980627d2dd7bfc32a3c025685a033eeab8d365cc840c631ef59d1b8f428164" dependencies = [ "aws-credential-types", "aws-smithy-async", @@ -4713,6 +4736,7 @@ dependencies = [ "deltalake-aws", "deltalake-azure", "deltalake-core", + "deltalake-gcp", ] [[package]] @@ -4821,6 +4845,23 @@ dependencies = [ "syn 2.0.106", ] +[[package]] +name = "deltalake-gcp" +version = "0.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df2708856cfa92e8309141fda1ee8d41e174b0868f26e557d0b2ea8d30fb92e" +dependencies = [ + "async-trait", + "bytes 1.10.1", + "deltalake-core", + "futures 0.3.31", + "object_store 0.12.4", + "thiserror 2.0.15", + "tokio", + "tracing 0.1.41", + "url", +] + [[package]] name = "der" version = "0.7.10" @@ -6780,7 +6821,6 @@ dependencies = [ "hyper 0.14.32", "log", "rustls 0.21.12", - "rustls-native-certs 0.6.3", "tokio", "tokio-rustls 0.24.1", ] @@ -13664,7 +13704,7 @@ dependencies = [ "aws-sdk-sts", "aws-sigv4", "aws-smithy-async", - "aws-smithy-http", + "aws-smithy-http 0.62.6", "aws-smithy-runtime", "aws-smithy-runtime-api", "aws-smithy-types", @@ -14031,6 +14071,7 @@ dependencies = [ "async-recursion", "async-trait", "aws-config", + "aws-runtime", "aws-sdk-s3", "aws-smithy-types", "azure_storage_blobs", diff --git a/Cargo.toml b/Cargo.toml index 5972af76..87425f57 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,9 @@ default-run = "vector" [profile.release] codegen-units = 1 lto = "fat" +strip = true +panic = "abort" +opt-level = "z" [[bin]] name = "vector" @@ -19,14 +22,15 @@ async-compression = { git = "https://github.com/nolouch/async-compression", rev async-recursion = "1.1.1" async-trait = { version = "0.1.88", default-features = false } arrow = { version = "56.2.0" } -aws-config = { version = "1.0" } -aws-sdk-s3 = { version = "1.82.0", default-features = false, features = ["behavior-version-latest"] } -aws-smithy-types = { version = "1.3.2", default-features = false } +aws-config = { version = "1.8.6" } +aws-runtime = { version = "1.5.16" } +aws-sdk-s3 = { version = "1.103.0", default-features = false, features = ["behavior-version-latest"] } +aws-smithy-types = { version = "1.3.4", default-features = false } azure_storage_blobs = { version = "0.17.0", default-features = false, features = ["enable_reqwest"] } base64 = { version = "0.22.1", default-features = false } bytes = { version = "1.10.1", default-features = false, features = ["serde"] } chrono = { version = "0.4.41", default-features = false, features = ["clock", "serde"] } -deltalake = { version = "0.29.3", features = ["datafusion", "s3", "azure"] } +deltalake = { version = "0.29.3", features = ["datafusion", "s3", "azure", "gcs"] } datafusion = { version = "48" } duckdb = { version = "1.0", features = ["bundled"] } etcd-client = { version = "0.14", features = ["tls-roots"] } diff --git a/Makefile b/Makefile index 733b2f0b..7aa1ac2f 100644 --- a/Makefile +++ b/Makefile @@ -191,6 +191,20 @@ release-docker: target/aarch64-unknown-linux-gnu/release/vector @scripts/release-docker.sh @echo "Done releasing docker image." +.PHONY: release-docker-alpine +release-docker-alpine: target/x86_64-unknown-linux-musl/release/vector +release-docker-alpine: target/aarch64-unknown-linux-musl/release/vector + @echo "Releasing Alpine docker image..." + @DOCKER_BASE=alpine scripts/release-docker.sh + @echo "Done releasing Alpine docker image." + +.PHONY: release-docker-minimal +release-docker-minimal: target/x86_64-unknown-linux-musl/release/vector +release-docker-minimal: target/aarch64-unknown-linux-musl/release/vector + @echo "Releasing minimal scratch docker image..." + @DOCKER_BASE=minimal scripts/release-docker.sh + @echo "Done releasing minimal scratch docker image." + .PHONY: release-docker-nextgen release-docker-nextgen: target/x86_64-unknown-linux-gnu/release/vector-nextgen release-docker-nextgen: target/aarch64-unknown-linux-gnu/release/vector-nextgen @@ -199,6 +213,13 @@ release-docker-nextgen: target/aarch64-unknown-linux-gnu/release/vector-nextgen @NEXTGEN=true scripts/release-docker.sh @echo "Done releasing docker image (nextgen mode)." +.PHONY: release-docker-nextgen-alpine +release-docker-nextgen-alpine: target/x86_64-unknown-linux-musl/release/vector-nextgen +release-docker-nextgen-alpine: target/aarch64-unknown-linux-musl/release/vector-nextgen + @echo "Releasing Alpine docker image (nextgen mode)..." + @NEXTGEN=true DOCKER_BASE=alpine scripts/release-docker.sh + @echo "Done releasing Alpine docker image (nextgen mode)." + .PHONY: test-integration test-integration: @echo "Running integration tests..." diff --git a/scripts/cross/aarch64-unknown-linux-gnu.dockerfile b/scripts/cross/aarch64-unknown-linux-gnu.dockerfile index a3243b30..67e41c73 100644 --- a/scripts/cross/aarch64-unknown-linux-gnu.dockerfile +++ b/scripts/cross/aarch64-unknown-linux-gnu.dockerfile @@ -4,8 +4,3 @@ COPY bootstrap-ubuntu.sh . COPY install-protoc.sh . RUN ./bootstrap-ubuntu.sh RUN ./install-protoc.sh - -RUN apt-get update && \ - apt-get remove --assume-yes gcc-9 && \ - apt-get --assume-yes install clang && \ - rm -rf /var/lib/apt/lists/* diff --git a/scripts/cross/bootstrap-ubuntu.sh b/scripts/cross/bootstrap-ubuntu.sh index 802e8068..ac03c3e0 100755 --- a/scripts/cross/bootstrap-ubuntu.sh +++ b/scripts/cross/bootstrap-ubuntu.sh @@ -3,28 +3,17 @@ set -o errexit echo 'Acquire::Retries "5";' > /etc/apt/apt.conf.d/80-retries -apt-get update -apt-get upgrade -y - -apt-get install -y \ - apt-transport-https \ - gnupg \ - wget - -# we need LLVM >= 3.9 for onig_sys/bindgen - -cat <<-EOF > /etc/apt/sources.list.d/llvm.list -deb http://apt.llvm.org/xenial/ llvm-toolchain-xenial-9 main -deb-src http://apt.llvm.org/xenial/ llvm-toolchain-xenial-9 main -EOF - -wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key| apt-key add - +export DEBIAN_FRONTEND=noninteractive apt-get update apt-get upgrade -y -# needed by onig_sys +# LLVM >= 3.9 is required by onig_sys/bindgen. Use distro packages from the +# cross-rs base image (Ubuntu 24.04) instead of the legacy xenial LLVM repo. apt-get install -y \ - libclang1-9 \ - llvm-9 \ - unzip + clang \ + curl \ + libclang-dev \ + unzip + +rm -rf /var/lib/apt/lists/* diff --git a/scripts/cross/x86_64-unknown-linux-gnu.dockerfile b/scripts/cross/x86_64-unknown-linux-gnu.dockerfile index a0bebb83..f5adc36a 100644 --- a/scripts/cross/x86_64-unknown-linux-gnu.dockerfile +++ b/scripts/cross/x86_64-unknown-linux-gnu.dockerfile @@ -4,8 +4,3 @@ COPY bootstrap-ubuntu.sh . COPY install-protoc.sh . RUN ./bootstrap-ubuntu.sh RUN ./install-protoc.sh - -RUN apt-get update && \ - apt-get remove --assume-yes gcc-9 && \ - apt-get --assume-yes install clang && \ - rm -rf /var/lib/apt/lists/* diff --git a/scripts/docker/Dockerfile b/scripts/docker/Dockerfile index 076abf0c..55606c79 100644 --- a/scripts/docker/Dockerfile +++ b/scripts/docker/Dockerfile @@ -1,5 +1,25 @@ +# Multi-stage build for smaller final image +FROM docker.io/debian:13-slim as builder + +# Install build dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates \ + tzdata \ + systemd \ + curl \ + binutils \ + && rm -rf /var/lib/apt/lists/* + +# Final stage with minimal runtime dependencies FROM docker.io/debian:13-slim -RUN apt-get update && apt-get install -y --no-install-recommends ca-certificates tzdata systemd curl binutils + +# Install only runtime dependencies +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates \ + tzdata \ + systemd \ + && rm -rf /var/lib/apt/lists/* \ + && apt-get clean ARG TARGETARCH COPY vector-$TARGETARCH /usr/bin/vector diff --git a/scripts/docker/Dockerfile.alpine b/scripts/docker/Dockerfile.alpine new file mode 100644 index 00000000..be2f5467 --- /dev/null +++ b/scripts/docker/Dockerfile.alpine @@ -0,0 +1,31 @@ +# Alpine-based Dockerfile for minimal image size +FROM alpine:3.19 as builder + +# Install build dependencies +RUN apk add --no-cache \ + ca-certificates \ + tzdata \ + curl \ + binutils \ + musl-dev + +# Final stage with minimal runtime dependencies +FROM alpine:3.19 + +# Install only runtime dependencies +RUN apk add --no-cache \ + ca-certificates \ + tzdata \ + && rm -rf /var/cache/apk/* + +ARG TARGETARCH +COPY vector-$TARGETARCH /usr/bin/vector +# COPY vector.toml /etc/vector/vector.toml + +ENV VECTOR_LOG="info" + +RUN ["chmod", "+x", "/usr/bin/vector"] +# Smoke test +RUN ["vector", "--version"] + +ENTRYPOINT ["/usr/bin/vector"] diff --git a/scripts/docker/Dockerfile.minimal b/scripts/docker/Dockerfile.minimal new file mode 100644 index 00000000..d9b2c7db --- /dev/null +++ b/scripts/docker/Dockerfile.minimal @@ -0,0 +1,22 @@ +# Minimal scratch-based Dockerfile for smallest possible image +FROM scratch + +# Copy CA certificates from a build stage +FROM alpine:3.19 as certs +RUN apk add --no-cache ca-certificates + +# Final minimal image +FROM scratch + +# Copy only what's needed +COPY --from=certs /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ +COPY --from=certs /usr/share/ca-certificates /usr/share/ca-certificates + +ARG TARGETARCH +COPY vector-$TARGETARCH /vector + +# Use static binary for scratch compatibility +ENV VECTOR_LOG="info" +ENV SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt + +ENTRYPOINT ["/vector"] diff --git a/scripts/release-docker.sh b/scripts/release-docker.sh index c924e555..0c1ca831 100755 --- a/scripts/release-docker.sh +++ b/scripts/release-docker.sh @@ -40,10 +40,19 @@ cp target/aarch64-unknown-linux-gnu/release/${BINARY_NAME} "$WORK_DIR"/vector-ar VERSION="${VECTOR_VERSION:-"$(scripts/version.sh)"}" REPO="${REPO:-"tidbcloud/vector"}" -BASE=debian +BASE="${DOCKER_BASE:-debian}" TAG="${TAG:-$REPO:$VERSION-$BASE}" -DOCKERFILE="scripts/docker/Dockerfile" +# Default Debian image uses scripts/docker/Dockerfile (no .debian suffix). +# Alpine/minimal use scripts/docker/Dockerfile.. +case "${BASE}" in +debian) + DOCKERFILE="scripts/docker/Dockerfile" + ;; +*) + DOCKERFILE="scripts/docker/Dockerfile.${BASE}" + ;; +esac #PLATFORMS="linux/amd64,linux/arm64,linux/arm/v7" PLATFORMS="linux/amd64,linux/arm64" diff --git a/src/common/deltalake_writer/delta_ops.rs b/src/common/deltalake_writer/delta_ops.rs index 7c8c3681..920d8537 100644 --- a/src/common/deltalake_writer/delta_ops.rs +++ b/src/common/deltalake_writer/delta_ops.rs @@ -9,6 +9,7 @@ use deltalake::DeltaOps; use tracing::{error, info, warn}; use url::Url; +use super::errors::is_stale_delta_log_error; use super::schema::SchemaManager; use super::types::TypeConverter; @@ -46,7 +47,7 @@ impl DeltaOpsManager { *session_token = "***REDACTED***".to_string(); } info!( - "Using storage options for S3 authentication: {:?}", + "Using storage options: {:?}", redacted_options ); Ok(DeltaOps::try_from_uri_with_storage_options( @@ -276,6 +277,24 @@ impl DeltaOpsManager { ); return Err(e.into()); } + } + // Stale log view after external compaction or concurrent writers + else if is_stale_delta_log_error(&error_str) { + if attempt < MAX_RETRIES - 1 { + warn!( + "Stale Delta log detected (attempt {}/{}): {}. Reloading table and retrying...", + attempt + 1, + MAX_RETRIES, + error_str + ); + continue; + } else { + error!( + "Stale Delta log after {} retries: {}", + MAX_RETRIES, error_str + ); + return Err(e.into()); + } } else { // Other error, fail immediately error!("Failed to write to Delta table: {}", e); diff --git a/src/common/deltalake_writer/errors.rs b/src/common/deltalake_writer/errors.rs new file mode 100644 index 00000000..6d2991de --- /dev/null +++ b/src/common/deltalake_writer/errors.rs @@ -0,0 +1,33 @@ +/// Returns true when a Delta Lake write error likely reflects a stale table log view, +/// for example after external compaction removed older `_delta_log` JSON files. +pub fn is_stale_delta_log_error(error_msg: &str) -> bool { + error_msg.contains("log segment") + || error_msg.contains("Invalid table version") + || error_msg.contains("not found") + || error_msg.contains("No such file or directory") + || error_msg.contains("Kernel error") + || error_msg.contains("No table metadata or protocol found in delta log") + || error_msg.contains("Expected ordered contiguous commit files") +} + +#[cfg(test)] +mod tests { + use super::is_stale_delta_log_error; + + #[test] + fn detects_missing_delta_log_file() { + let msg = "Kernel error: File not found: deltalake/org=1/_delta_log/00000000000000020406.json"; + assert!(is_stale_delta_log_error(msg)); + } + + #[test] + fn detects_kernel_error_without_file_not_found() { + let msg = "Kernel error: No table metadata or protocol found in delta log."; + assert!(is_stale_delta_log_error(msg)); + } + + #[test] + fn ignores_unrelated_errors() { + assert!(!is_stale_delta_log_error("permission denied")); + } +} diff --git a/src/common/deltalake_writer/mod.rs b/src/common/deltalake_writer/mod.rs index 9a3d0fc7..b030c2d5 100644 --- a/src/common/deltalake_writer/mod.rs +++ b/src/common/deltalake_writer/mod.rs @@ -9,12 +9,14 @@ use vector_lib::event::Event; // Module declarations pub mod converter; pub mod delta_ops; +pub mod errors; pub mod schema; pub mod types; // Re-export main types pub use converter::EventConverter; pub use delta_ops::DeltaOpsManager; +pub use errors::is_stale_delta_log_error; pub use schema::SchemaManager; pub use types::TypeConverter; diff --git a/src/sinks/deltalake/processor.rs b/src/sinks/deltalake/processor.rs index 31058be2..1b0a6b40 100644 --- a/src/sinks/deltalake/processor.rs +++ b/src/sinks/deltalake/processor.rs @@ -1,13 +1,16 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use futures::{stream::BoxStream, StreamExt}; use tokio::sync::Mutex; use vector_lib::event::Event; use vector_lib::sink::StreamSink; -use crate::common::deltalake_writer::{DeltaLakeWriter, DeltaTableConfig, WriteConfig}; +use crate::common::deltalake_writer::{ + is_stale_delta_log_error, DeltaLakeWriter, DeltaTableConfig, WriteConfig, +}; /// Delta Lake sink processor pub struct DeltaLakeSink { @@ -69,30 +72,41 @@ impl DeltaLakeSink { // Write each table's events for (table_name, table_events) in table_events { if let Err(e) = self.write_table_events(&table_name, table_events).await { - let error_msg = e.to_string(); - if error_msg.contains("log segment") - || error_msg.contains("Invalid table version") - || error_msg.contains("not found") - || error_msg.contains("No such file or directory") - { - panic!( - "Delta Lake corruption detected for table {}: {}", - table_name, error_msg - ); - } else { - error!("Failed to write events to table {}: {}", table_name, e); - } + error!("Failed to write events to table {}: {}", table_name, e); } } Ok(()) } - /// Write events to a specific table + /// Write events to a specific table, evicting and reopening the writer once on stale log errors. async fn write_table_events( &self, table_name: &str, events: Vec, + ) -> Result<(), Box> { + match self + .write_table_events_once(table_name, &events) + .await + { + Ok(()) => Ok(()), + Err(e) if is_stale_delta_log_error(&e.to_string()) => { + warn!( + "Stale Delta log for table {}, evicting cached writer and retrying once: {}", + table_name, e + ); + self.writers.lock().await.remove(table_name); + tokio::time::sleep(Duration::from_millis(200)).await; + self.write_table_events_once(table_name, &events).await + } + Err(e) => Err(e), + } + } + + async fn write_table_events_once( + &self, + table_name: &str, + events: &[Event], ) -> Result<(), Box> { // Get or create writer for this table let mut writers = self.writers.lock().await; @@ -135,7 +149,7 @@ impl DeltaLakeSink { }); // Write events - writer.write_events(events).await?; + writer.write_events(events.to_vec()).await?; Ok(()) } @@ -165,9 +179,36 @@ impl StreamSink for DeltaLakeSink { mod tests { use super::*; use std::collections::BTreeMap; - use vector_lib::event::LogEvent; + use std::fs; + use vector_lib::event::{LogEvent, ObjectMap}; + + fn create_test_event(table_name: &str, index: i64) -> Event { + let mut log = LogEvent::from(BTreeMap::new()); + log.insert("_vector_table", table_name); + log.insert("_vector_source_table", "TEST_SOURCE"); + log.insert("_vector_source_schema", "test_schema"); + log.insert("_vector_instance", "test-instance"); + log.insert("_vector_timestamp", "2024-06-01T00:00:00Z"); + log.insert("id", index); + log.insert("value", format!("row-{index}")); + + let mut schema_meta = ObjectMap::new(); + schema_meta.insert("_partition_by".into(), vector_lib::event::Value::from("date")); + let mut id_meta = ObjectMap::new(); + id_meta.insert("mysql_type".into(), vector_lib::event::Value::from("bigint")); + schema_meta.insert("id".into(), vector_lib::event::Value::Object(id_meta)); + let mut value_meta = ObjectMap::new(); + value_meta.insert( + "mysql_type".into(), + vector_lib::event::Value::from("varchar(64)"), + ); + schema_meta.insert("value".into(), vector_lib::event::Value::Object(value_meta)); + log.insert("_schema_metadata", vector_lib::event::Value::Object(schema_meta)); + + Event::Log(log) + } - fn create_test_event(table_field: &str, table_name: &str) -> Event { + fn create_test_event_legacy(table_field: &str, table_name: &str) -> Event { let mut log = LogEvent::from(BTreeMap::new()); log.insert(table_field, table_name); log.insert("test_field", "test_value"); @@ -176,7 +217,7 @@ mod tests { #[test] fn test_table_name_extraction_from_vector_table() { - let event = create_test_event("_vector_table", "test_table"); + let event = create_test_event_legacy("_vector_table", "test_table"); if let Event::Log(log) = &event { let table_name = log .get("_vector_table") @@ -189,7 +230,7 @@ mod tests { #[test] fn test_table_name_extraction_from_dest_table() { - let event = create_test_event("dest_table", "my_dest_table"); + let event = create_test_event_legacy("dest_table", "my_dest_table"); if let Event::Log(log) = &event { let table_name = log .get("_vector_table") @@ -202,7 +243,7 @@ mod tests { #[test] fn test_table_name_extraction_from_table() { - let event = create_test_event("table", "fallback_table"); + let event = create_test_event_legacy("table", "fallback_table"); if let Event::Log(log) = &event { let table_name = log .get("_vector_table") @@ -235,9 +276,9 @@ mod tests { #[test] fn test_events_grouping_by_table() { let events = vec![ - create_test_event("_vector_table", "table_a"), - create_test_event("_vector_table", "table_b"), - create_test_event("_vector_table", "table_a"), + create_test_event_legacy("_vector_table", "table_a"), + create_test_event_legacy("_vector_table", "table_b"), + create_test_event_legacy("_vector_table", "table_a"), ]; let mut table_events: HashMap> = HashMap::new(); @@ -262,4 +303,64 @@ mod tests { assert_eq!(table_events.get("table_a").unwrap().len(), 2); assert_eq!(table_events.get("table_b").unwrap().len(), 1); } + + fn delta_log_json_files(delta_log_path: &std::path::Path) -> Vec { + let mut files: Vec = fs::read_dir(delta_log_path) + .expect("read _delta_log") + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "json")) + .collect(); + files.sort(); + files + } + + #[tokio::test] + async fn sink_process_events_recovers_after_simulated_compaction() { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let base_path = std::env::temp_dir().join(format!("deltalake_sink_stale_log_{nanos}")); + let table_name = "recover_table"; + let table_path = base_path.join(table_name); + fs::create_dir_all(&table_path).expect("create table dir"); + + let sink = DeltaLakeSink::new( + base_path.clone(), + vec![DeltaTableConfig { + name: table_name.to_string(), + schema_evolution: Some(true), + }], + WriteConfig { + batch_size: 1000, + timeout_secs: 30, + }, + None, + ); + + for batch in 0..5 { + let events: Vec = (0..3) + .map(|i| create_test_event(table_name, i)) + .collect(); + sink.process_events(events) + .await + .expect("seed batch should write"); + let _ = batch; + } + + let delta_log_path = table_path.join("_delta_log"); + let json_files = delta_log_json_files(&delta_log_path); + assert!(json_files.len() >= 3); + let latest = json_files.last().expect("latest delta log json").clone(); + fs::remove_file(latest).expect("simulate compaction"); + + // Must not panic; stale-log recovery should allow the batch to be written. + sink.process_events(vec![create_test_event(table_name, 999)]) + .await + .expect("sink should recover after simulated compaction"); + + assert!(delta_log_json_files(&delta_log_path).len() >= 1); + let _ = fs::remove_dir_all(&base_path); + } } diff --git a/src/sinks/topsql_data_deltalake/processor.rs b/src/sinks/topsql_data_deltalake/processor.rs index b3004bc8..f60e03b0 100644 --- a/src/sinks/topsql_data_deltalake/processor.rs +++ b/src/sinks/topsql_data_deltalake/processor.rs @@ -1,6 +1,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use futures::{stream::BoxStream, StreamExt}; use tokio::sync::mpsc; @@ -8,7 +9,9 @@ use tokio::sync::Mutex; use vector_lib::event::{Event, LogEvent}; use vector_lib::sink::StreamSink; -use crate::common::deltalake_writer::{DeltaLakeWriter, DeltaTableConfig, WriteConfig}; +use crate::common::deltalake_writer::{ + is_stale_delta_log_error, DeltaLakeWriter, DeltaTableConfig, WriteConfig, +}; use crate::common::keyspace_cluster::{ path_contains_keyspace_route_segments, replace_keyspace_route_segments, route_resolution_retry_delay, KeyspaceRoute, PdKeyspaceResolver, @@ -401,24 +404,12 @@ impl TopSQLDeltaLakeSink { for (writer_key, mut events) in table_events { self.add_schema_info(&mut events, &writer_key.table_name); if let Err(e) = self.write_table_events(&writer_key, events).await { - let error_msg = e.to_string(); - if error_msg.contains("log segment") - || error_msg.contains("Invalid table version") - || error_msg.contains("not found") - || error_msg.contains("No such file or directory") - { - panic!( - "Delta Lake corruption detected for table {}: {}", - writer_key.table_name, error_msg - ); - } else { - error!( - "Failed to write events to table {} at {}: {}", - writer_key.table_name, - writer_key.table_path.display(), - e - ); - } + error!( + "Failed to write events to table {} at {}: {}", + writer_key.table_name, + writer_key.table_path.display(), + e + ); } } @@ -567,11 +558,33 @@ impl TopSQLDeltaLakeSink { } } - /// Write events to a specific table + /// Write events to a specific table, evicting and reopening the writer once on stale log errors. async fn write_table_events( &self, writer_key: &WriterKey, events: Vec, + ) -> Result<(), Box> { + match self.write_table_events_once(writer_key, &events).await { + Ok(()) => Ok(()), + Err(e) if is_stale_delta_log_error(&e.to_string()) => { + warn!( + "Stale Delta log for table {} at {}, evicting cached writer and retrying once: {}", + writer_key.table_name, + writer_key.table_path.display(), + e + ); + self.writers.lock().await.remove(writer_key); + tokio::time::sleep(Duration::from_millis(200)).await; + self.write_table_events_once(writer_key, &events).await + } + Err(e) => Err(e), + } + } + + async fn write_table_events_once( + &self, + writer_key: &WriterKey, + events: &[Event], ) -> Result<(), Box> { let mut writers = self.writers.lock().await; let writer = writers.entry(writer_key.clone()).or_insert_with(|| { @@ -594,7 +607,7 @@ impl TopSQLDeltaLakeSink { }); // Write events - writer.write_events(events).await?; + writer.write_events(events.to_vec()).await?; Ok(()) } diff --git a/src/sinks/topsql_meta_deltalake/processor.rs b/src/sinks/topsql_meta_deltalake/processor.rs index dff95265..10af7f1c 100644 --- a/src/sinks/topsql_meta_deltalake/processor.rs +++ b/src/sinks/topsql_meta_deltalake/processor.rs @@ -10,7 +10,9 @@ use tokio::sync::Mutex; use vector_lib::event::{Event, LogEvent}; use vector_lib::sink::StreamSink; -use crate::common::deltalake_writer::{DeltaLakeWriter, DeltaTableConfig, WriteConfig}; +use crate::common::deltalake_writer::{ + is_stale_delta_log_error, DeltaLakeWriter, DeltaTableConfig, WriteConfig, +}; use crate::common::keyspace_cluster::{ path_contains_keyspace_route_segments, replace_keyspace_route_segments, route_resolution_retry_delay, KeyspaceRoute, PdKeyspaceResolver, @@ -380,24 +382,12 @@ impl TopSQLDeltaLakeSink { for (writer_key, mut events) in table_events { self.add_schema_info(&writer_key.table_name, &mut events); if let Err(e) = self.write_table_events(&writer_key, events).await { - let error_msg = e.to_string(); - if error_msg.contains("log segment") - || error_msg.contains("Invalid table version") - || error_msg.contains("not found") - || error_msg.contains("No such file or directory") - { - panic!( - "Delta Lake corruption detected for table {}: {}", - writer_key.table_name, error_msg - ); - } else { - error!( - "Failed to write events to table {} at {}: {}", - writer_key.table_name, - writer_key.table_path.display(), - e - ); - } + error!( + "Failed to write events to table {} at {}: {}", + writer_key.table_name, + writer_key.table_path.display(), + e + ); } else if let Some(keys) = table_dedup_keys.remove(&writer_key) { committed_dedup_keys.extend(keys); } @@ -600,11 +590,33 @@ impl TopSQLDeltaLakeSink { ); } - /// Write events to a specific table + /// Write events to a specific table, evicting and reopening the writer once on stale log errors. async fn write_table_events( &self, writer_key: &WriterKey, events: Vec, + ) -> Result<(), Box> { + match self.write_table_events_once(writer_key, &events).await { + Ok(()) => Ok(()), + Err(e) if is_stale_delta_log_error(&e.to_string()) => { + warn!( + "Stale Delta log for table {} at {}, evicting cached writer and retrying once: {}", + writer_key.table_name, + writer_key.table_path.display(), + e + ); + self.writers.lock().await.remove(writer_key); + tokio::time::sleep(Duration::from_millis(200)).await; + self.write_table_events_once(writer_key, &events).await + } + Err(e) => Err(e), + } + } + + async fn write_table_events_once( + &self, + writer_key: &WriterKey, + events: &[Event], ) -> Result<(), Box> { // Get or create writer for this table let mut writers = self.writers.lock().await; @@ -628,7 +640,7 @@ impl TopSQLDeltaLakeSink { }); // Write events - writer.write_events(events).await?; + writer.write_events(events.to_vec()).await?; Ok(()) } diff --git a/tests/deltalake_stale_log_test.rs b/tests/deltalake_stale_log_test.rs new file mode 100644 index 00000000..e158ee70 --- /dev/null +++ b/tests/deltalake_stale_log_test.rs @@ -0,0 +1,147 @@ +//! Local Delta Lake test: simulate external compaction removing old _delta_log JSON +//! files and verify writes recover instead of failing permanently. + +#![allow(clippy::print_stdout)] + +use std::collections::BTreeMap; +use std::fs; +use std::path::{Path, PathBuf}; + +use vector_lib::event::{Event, LogEvent, ObjectMap}; + +use vector_extensions::sinks::deltalake::{DeltaLakeWriter, DeltaTableConfig, WriteConfig}; + +fn make_event(table_name: &str, index: i64) -> Event { + let mut log = LogEvent::from(BTreeMap::new()); + log.insert("_vector_table", table_name); + log.insert("_vector_source_table", "TEST_SOURCE"); + log.insert("_vector_source_schema", "test_schema"); + log.insert("_vector_instance", "test-instance"); + log.insert("_vector_timestamp", "2024-06-01T00:00:00Z"); + log.insert("id", index); + log.insert("value", format!("row-{index}")); + + let mut schema_meta = ObjectMap::new(); + schema_meta.insert("_partition_by".into(), vector_lib::event::Value::from("date")); + let mut id_meta = ObjectMap::new(); + id_meta.insert("mysql_type".into(), vector_lib::event::Value::from("bigint")); + schema_meta.insert("id".into(), vector_lib::event::Value::Object(id_meta)); + let mut value_meta = ObjectMap::new(); + value_meta.insert( + "mysql_type".into(), + vector_lib::event::Value::from("varchar(64)"), + ); + schema_meta.insert("value".into(), vector_lib::event::Value::Object(value_meta)); + log.insert("_schema_metadata", vector_lib::event::Value::Object(schema_meta)); + + Event::Log(log) +} + +fn delta_log_json_files(delta_log_path: &Path) -> Vec { + let mut files: Vec = fs::read_dir(delta_log_path) + .expect("read _delta_log") + .filter_map(|entry| entry.ok()) + .map(|entry| entry.path()) + .filter(|path| path.extension().is_some_and(|ext| ext == "json")) + .collect(); + files.sort(); + files +} + +/// Remove the latest commit log JSON file while keeping the contiguous history intact. +/// This mimics a concurrent compact/remove race where a reader still chases a log +/// segment that was already deleted, while the table remains valid at an earlier version. +fn simulate_external_compaction(delta_log_path: &Path) { + let json_files = delta_log_json_files(delta_log_path); + assert!( + json_files.len() >= 3, + "need at least 3 json log files before compaction simulation" + ); + + let stale_file = json_files.last().expect("latest delta log json"); + println!("Simulating compaction: removing {:?}", stale_file); + fs::remove_file(stale_file).expect("remove stale delta log json"); +} + +fn unique_table_dir(name: &str) -> PathBuf { + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + std::env::temp_dir().join(format!("{name}_{nanos}")) +} + +async fn write_batch(writer: &mut DeltaLakeWriter, table_name: &str, start: i64, count: i64) { + let events: Vec = (start..start + count) + .map(|i| make_event(table_name, i)) + .collect(); + writer + .write_events(events) + .await + .expect("seed write should succeed"); +} + +#[tokio::test] +async fn recover_write_after_simulated_delta_log_compaction() { + let table_path = unique_table_dir("deltalake_stale_log_recovery"); + fs::create_dir_all(&table_path).expect("create table dir"); + + let table_name = "metrics_table"; + let write_config = WriteConfig { + batch_size: 1000, + timeout_secs: 30, + }; + let table_config = DeltaTableConfig { + name: table_name.to_string(), + schema_evolution: Some(true), + }; + + let mut writer = DeltaLakeWriter::new( + table_path.clone(), + table_config.clone(), + write_config.clone(), + None, + ); + + // Build a multi-version table so _delta_log has several JSON commits. + for batch in 0..5 { + write_batch(&mut writer, table_name, batch * 10, 3).await; + } + + let delta_log_path = table_path.join("_delta_log"); + let json_before = delta_log_json_files(&delta_log_path); + println!("Delta log json files before compaction simulation: {}", json_before.len()); + assert!(json_before.len() >= 3, "expected multiple delta log commits"); + + simulate_external_compaction(&delta_log_path); + + let recovery_events = vec![make_event(table_name, 999)]; + match writer.write_events(recovery_events.clone()).await { + Ok(()) => {} + Err(error) if vector_extensions::common::deltalake_writer::is_stale_delta_log_error(&error.to_string()) => { + writer = DeltaLakeWriter::new( + table_path.clone(), + DeltaTableConfig { + name: table_name.to_string(), + schema_evolution: Some(true), + }, + write_config.clone(), + None, + ); + tokio::time::sleep(std::time::Duration::from_millis(200)).await; + writer + .write_events(recovery_events) + .await + .expect("write should recover after reopening table writer"); + } + Err(error) => panic!("unexpected write failure: {error}"), + } + + let json_after = delta_log_json_files(&delta_log_path); + assert!( + !json_after.is_empty(), + "table should remain writable after recovery" + ); + + let _ = fs::remove_dir_all(&table_path); +}