diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index ed2f91634327..9f34fb54e47b 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -869,6 +869,8 @@ class BeamModulePlugin implements Plugin { opentelemetry_sdk : "io.opentelemetry:opentelemetry-sdk", // opentelemetry-bom sets version opentelemetry_exporter_otlp : "io.opentelemetry:opentelemetry-exporter-otlp", // opentelemetry-bom sets version opentelemetry_extension_autoconfigure : "io.opentelemetry:opentelemetry-sdk-extension-autoconfigure", // opentelemetry-bom sets version + opentelemetry_proto : "io.opentelemetry.proto:opentelemetry-proto:$opentelemetry_version-alpha", + opentelemetry_sdk_testing : "io.opentelemetry:opentelemetry-sdk-testing:$opentelemetry_version", postgres : "org.postgresql:postgresql:$postgres_version", protobuf_java : "com.google.protobuf:protobuf-java:$protobuf_version", protobuf_java_util : "com.google.protobuf:protobuf-java-util:$protobuf_version", diff --git a/examples/java/src/main/java/org/apache/beam/examples/BatchElementsExample.java b/examples/java/src/main/java/org/apache/beam/examples/BatchElementsExample.java index 0103999d9256..79130d7ac138 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/BatchElementsExample.java +++ b/examples/java/src/main/java/org/apache/beam/examples/BatchElementsExample.java @@ -29,6 +29,19 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +// beam-playground: +// name: BatchElements +// description: Demonstration of BatchElements transform usage. +// multifile: false +// default_example: false +// context_line: 47 +// categories: +// - Core Transforms +// complexity: BASIC +// tags: +// - transforms +// - batch + public class BatchElementsExample { public static void main(String[] args) { PipelineOptions options = PipelineOptionsFactory.create(); diff --git a/sdks/go.mod b/sdks/go.mod index 011b2e0fe8c6..7cbe318bb8f9 100644 --- a/sdks/go.mod +++ b/sdks/go.mod @@ -31,13 +31,13 @@ require ( cloud.google.com/go/profiler v0.6.0 cloud.google.com/go/pubsub v1.50.2 cloud.google.com/go/spanner v1.91.0 - cloud.google.com/go/storage v1.62.2 - github.com/aws/aws-sdk-go-v2 v1.41.11 - github.com/aws/aws-sdk-go-v2/config v1.32.22 - github.com/aws/aws-sdk-go-v2/credentials v1.19.21 - github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.24 - github.com/aws/aws-sdk-go-v2/service/s3 v1.103.1 - github.com/aws/smithy-go v1.27.0 + cloud.google.com/go/storage v1.62.3 + github.com/aws/aws-sdk-go-v2 v1.41.12 + github.com/aws/aws-sdk-go-v2/config v1.32.23 + github.com/aws/aws-sdk-go-v2/credentials v1.19.22 + github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.25 + github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2 + github.com/aws/smithy-go v1.27.1 github.com/docker/go-connections v0.7.0 // indirect github.com/dustin/go-humanize v1.0.1 github.com/go-sql-driver/mysql v1.10.0 @@ -91,7 +91,7 @@ require ( github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.56.0 // indirect github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/apache/arrow/go/v15 v15.0.2 // indirect - github.com/aws/aws-sdk-go-v2/service/signin v1.1.3 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.1.4 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect github.com/containerd/log v0.1.0 // indirect @@ -148,18 +148,18 @@ require ( github.com/apache/arrow/go/arrow v0.0.0-20211112161151-bc219186db40 // indirect github.com/apache/thrift v0.23.0 // indirect github.com/aws/aws-sdk-go v1.55.5 // indirect - github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.12 // indirect - github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.27 // indirect - github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.27 // indirect - github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.27 // indirect - github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.28 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.11 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.20 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.27 // indirect - github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.27 // indirect - github.com/aws/aws-sdk-go-v2/service/sso v1.31.1 // indirect - github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.4 // indirect - github.com/aws/aws-sdk-go-v2/service/sts v1.43.1 // indirect + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.31.2 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.43.2 // indirect github.com/cenkalti/backoff/v4 v4.3.0 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 // indirect diff --git a/sdks/go.sum b/sdks/go.sum index 74dc7d7358ea..0b5cc9db4bac 100644 --- a/sdks/go.sum +++ b/sdks/go.sum @@ -99,8 +99,8 @@ cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RX cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0= cloud.google.com/go/storage v1.12.0/go.mod h1:fFLk2dp2oAhDz8QFKwqrjdJvxSp/W2g7nillojlL5Ho= cloud.google.com/go/storage v1.21.0/go.mod h1:XmRlxkgPjlBONznT2dDUU/5XlpU2OjMnKuqnZI01LAA= -cloud.google.com/go/storage v1.62.2 h1:WgR4U9n7bIzXkkVnwPKKE8bkaKUNsHG+0MAAlh9DGU4= -cloud.google.com/go/storage v1.62.2/go.mod h1:cpYz/kRVZ+UQAF1uHeea10/9ewcRbxGoGNKsS9daSXA= +cloud.google.com/go/storage v1.62.3 h1:SZq1t23NCI+e96dH77Dg3PEfsNNEjqO8zE5AnD8gVD0= +cloud.google.com/go/storage v1.62.3/go.mod h1:cpYz/kRVZ+UQAF1uHeea10/9ewcRbxGoGNKsS9daSXA= cloud.google.com/go/trace v1.0.0/go.mod h1:4iErSByzxkyHWzzlAj63/Gmjz0NH1ASqhJguHpGcr6A= cloud.google.com/go/trace v1.2.0/go.mod h1:Wc8y/uYyOhPy12KEnXG9XGrvfMz5F5SrYecQlbW1rwM= cloud.google.com/go/trace v1.16.0 h1:GmQovzFc5F0CNfl0VLgL64aoTtu7xsM0YajW2GlG9+E= @@ -199,83 +199,83 @@ github.com/aws/aws-sdk-go v1.55.5 h1:KKUZBfBoyqy5d3swXyiC7Q76ic40rYcbqH7qjh59kzU github.com/aws/aws-sdk-go v1.55.5/go.mod h1:eRwEWoyTWFMVYVQzKMNHWP5/RV4xIUGMQfXQHfHkpNU= github.com/aws/aws-sdk-go-v2 v1.16.2/go.mod h1:ytwTPBG6fXTZLxxeeCCWj2/EMYp/xDUgX+OET6TLNNU= github.com/aws/aws-sdk-go-v2 v1.23.0/go.mod h1:i1XDttT4rnf6vxc9AuskLc6s7XBee8rlLilKlc03uAA= -github.com/aws/aws-sdk-go-v2 v1.41.11 h1:9PRf7jyTMEUM6fuNRAJa2mO/skJfrF50rENJwf2LXqw= -github.com/aws/aws-sdk-go-v2 v1.41.11/go.mod h1:iiUX27gOXRuYaoeUVXhUpPwjJHzISfPAjjcuhUbLSVs= +github.com/aws/aws-sdk-go-v2 v1.41.12 h1:DIKX2c31ekm9RA2D9FBj1EWXx++9AdAqRw+e78Tq2Ck= +github.com/aws/aws-sdk-go-v2 v1.41.12/go.mod h1:27+ACypSLljLAEKsCYOmrjKh83vuTRkuAe9Uv/3A4bg= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.4.1/go.mod h1:n8Bs1ElDD2wJ9kCRTczA83gYbBmjSwZp3umc6zF4EeM= github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.5.1/go.mod h1:t8PYl/6LzdAqsU4/9tz28V/kU+asFePvpOMkdul0gEQ= -github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.12 h1:oRtsqWgxbpeXrOlxOoQStx2M9WNbIkPq4C4Xn1or6bc= -github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.12/go.mod h1:Zg0Oe9qT+9wcezlm1a64wGJp2qZdRElVxo/seJf7jYU= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13 h1:p1BBrg/Hhp6uK7zpejeI8QFXHJeC/mynzi04Sl03k9g= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.13/go.mod h1:8cIfkE9MDhkRZGpQ22aV6/lkYeYSozpz16Smrs5x4Ls= github.com/aws/aws-sdk-go-v2/config v1.15.3/go.mod h1:9YL3v07Xc/ohTsxFXzan9ZpFpdTOFl4X65BAKYaz8jg= github.com/aws/aws-sdk-go-v2/config v1.25.3/go.mod h1:tAByZy03nH5jcq0vZmkcVoo6tRzRHEwSFx3QW4NmDw8= -github.com/aws/aws-sdk-go-v2/config v1.32.22 h1:Vfvp7+fYKsVCADcWOEllqEV47aIBXhNchvyDFu1B5fY= -github.com/aws/aws-sdk-go-v2/config v1.32.22/go.mod h1:0+H+0nPKbvWltf5vSIGkApv+hGbaQ4FfwTjGIYQREcw= +github.com/aws/aws-sdk-go-v2/config v1.32.23 h1:PYDobtcsJXK6bQe9I8RQk6s19Bz3xa3xRU08Hy1Em3Y= +github.com/aws/aws-sdk-go-v2/config v1.32.23/go.mod h1:QID4dqUQVgEOYPKsPWd1sNWCCR2c5g7o3jeEtIXPOZU= github.com/aws/aws-sdk-go-v2/credentials v1.11.2/go.mod h1:j8YsY9TXTm31k4eFhspiQicfXPLZ0gYXA50i4gxPE8g= github.com/aws/aws-sdk-go-v2/credentials v1.16.2/go.mod h1:sDdvGhXrSVT5yzBDR7qXz+rhbpiMpUYfF3vJ01QSdrc= -github.com/aws/aws-sdk-go-v2/credentials v1.19.21 h1:0+HscFXtNa4+3buV4IlG6v5lnOdzi5TrpicFGjKHgh4= -github.com/aws/aws-sdk-go-v2/credentials v1.19.21/go.mod h1:UE8+9t5zudFwu5k5ShC1PKArVEdOkQQdCXIHQAVNUcU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.22 h1:SHfH6wyPsEgG7fVsi5rQxWEt7tuIcN2PGhb1mTFv6tE= +github.com/aws/aws-sdk-go-v2/credentials v1.19.22/go.mod h1:54nO8lKD4aQPOntM/VTWjnR+DYzTwx0YkSMZMhAgewQ= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.12.3/go.mod h1:uk1vhHHERfSVCUnqSqz8O48LBYDSC+k6brng09jcMOk= github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.14.4/go.mod h1:t4i+yGHMCcUNIX1x7YVYa6bH/Do7civ5I6cG/6PMfyA= -github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.27 h1:BEfN1sjtiKEdikRDxYkjZNE4tyvw/YbGWCbl3xDZgRw= -github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.27/go.mod h1:ISGSFNbOHRS+JV/17yStzRTPBUHHqF92kCpRLLyH3Nk= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28 h1:b+kcDejJrXc30zU/w8Tc9klISwaO5wh+6T0sMBdDoHM= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.28/go.mod h1:LnI62O9GnSv6GcuLXxOYqlq0C8EmxMcgnF6m7LdYuOY= github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.11.3/go.mod h1:0dHuD2HZZSiwfJSy1FO5bX1hQ1TxVV1QXXjpn3XUE44= github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.14.0/go.mod h1:UcgIwJ9KHquYxs6Q5skC9qXjhYMK+JASDYcXQ4X7JZE= -github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.24 h1:eRIrOCSlSjyb/tT6vAPLwUZQO71XWaLkqwtM8DQuQ2s= -github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.24/go.mod h1:ox5MjGuEKA/oXGNbprU9PW/nc+YALeHjDkIiHfzCUgM= +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.25 h1:xJ3WVH3J0xESIqkavgbNfvQdMzB98bSkGwFXCyM2Tdw= +github.com/aws/aws-sdk-go-v2/feature/s3/manager v1.22.25/go.mod h1:0EOIx2v10rBBcaoTpOsRRNKDxhtRecRbN42YMrJaKZE= github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.9/go.mod h1:AnVH5pvai0pAF4lXRq0bmhbes1u9R8wTE+g+183bZNM= github.com/aws/aws-sdk-go-v2/internal/configsources v1.2.3/go.mod h1:7sGSz1JCKHWWBHq98m6sMtWQikmYPpxjqOydDemiVoM= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.27 h1:8sPbKi1/KRHwl5oR3qN9mUXestCeHuaRutxylnr/eVY= -github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.27/go.mod h1:QV9IVIopJ1dpQUno0f9VYDUwOEjj8u0iEJ4JiZVre3Y= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28 h1:Xf2j7NdVcUKomlZ4iihOP4AZ3Fzlr8h4yKpXeP+OFPg= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.28/go.mod h1:O8cDo1dW63jU7ki//kRe1z+tLGcpnD1jrouitsQddDw= github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.3/go.mod h1:ssOhaLpRlh88H3UmEcsBoVKq309quMvm3Ds8e9d4eJM= github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.5.3/go.mod h1:ify42Rb7nKeDDPkFjKn7q1bPscVPu/+gmHH8d2c+anU= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.27 h1:9d8AoASQY9UwrOSmiJ7uSM0MGUPFhnenwSvpaFfat2c= -github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.27/go.mod h1:x0rldpsnUQaQIs4Rh+Vwm9Z/0vI6BxadGtsgJfZFb8s= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28 h1:KqIfN9kpkKkcBqBbNpNGTIrXO6ExTUvFKvXkC+YAzVo= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.28/go.mod h1:uxtQiKvLtNS4iXVsH2McVD/ls8FKN/uUhe1hGxPjrw0= github.com/aws/aws-sdk-go-v2/internal/ini v1.3.10/go.mod h1:8DcYQcz0+ZJaSxANlHIsbbi6S+zMwjwdDqwW3r9AzaE= github.com/aws/aws-sdk-go-v2/internal/ini v1.7.1/go.mod h1:6fQQgfuGmw8Al/3M2IgIllycxV7ZW7WCdVSqfBeUiCY= github.com/aws/aws-sdk-go-v2/internal/v4a v1.2.3/go.mod h1:5yzAuE9i2RkVAttBl8yxZgQr5OCq4D5yDnG7j9x2L0U= -github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.28 h1:eaS9vwQ5ym4Y9S6+G/K3d3lgZhxs9Sldcn/YS7cmdKY= -github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.28/go.mod h1:oTdbDr+BMs7gAYrNpD0LDTyqQfv6yOYgTDv46+xbwFY= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29 h1:VkE9FuzTQVjBBrnj4+oCdxCLFIz7aqLYKUCjtvxVcOs= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.29/go.mod h1:H32Z2Qth9b+9LqjyBsCnozMQ8H2N7YBUDVXwbs0iggg= github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.9.1/go.mod h1:GeUru+8VzrTXV/83XyMJ80KpH8xO89VPoUileyNQ+tc= github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.10.1/go.mod h1:l9ymW25HOqymeU2m1gbUQ3rUIsTwKs8gYHXkqDQUhiI= -github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.11 h1:rFSsqDfCMPAmG70JOsYqFZCHXkyatoGa1K4YEt/BggQ= -github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.11/go.mod h1:XG68qW+YLLFH0vnSDCou43Cgj5TeAG83O5NRSJgt04Y= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12 h1:ZD2+BSw9vFsNlKYIasSNt3uDbjqqXIBcM13UJv/Lx2k= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.12/go.mod h1:Ms4zlcVBbXbiP7EVLhl+lgjvA/a7YphqQ3Ih3174EmI= github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.1.3/go.mod h1:Seb8KNmD6kVTjwRjVEgOT5hPin6sq+v4C2ycJQDwuH8= github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.2.3/go.mod h1:R+/S1O4TYpcktbVwddeOYg+uwUfLhADP2S/x4QwsCTM= -github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.20 h1:yt2fjgev3Hqm33zPw0ZWtki3sZ0SLcr+PkuvXDAAf/8= -github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.20/go.mod h1:wnPjCjPJ6x5GBhrER8f0QakaQ2LokfhCVYxmAZBpPjY= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21 h1:FsZxbPiVgEHYofziwfylouMki8b1Z7mI4CMU/7bhwBA= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.21/go.mod h1:Mmm30OV+JLXYQUcbSd84THnv3P5JtjhVDujLwMqRG0U= github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.3/go.mod h1:wlY6SVjuwvh3TVRpTqdy4I1JpBFLX4UGeKZdWntaocw= github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.10.3/go.mod h1:Owv1I59vaghv1Ax8zz8ELY8DN7/Y0rGS+WWAmjgi950= -github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.27 h1:2/pUo42hhVmQcM21ttZoBOLHQymyUH8qEnZGTIuGBT8= -github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.27/go.mod h1:p7hwgbwompjCRNTdB3ytlldddNt1rDBgVVMqWEVG1II= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28 h1:axj4mEDletwKmTm/9jR+DkIMmCfcn5vE4jBMAAN+3Vg= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.28/go.mod h1:3Aaz69M0jqfSHLKqxgolgUBFT4hpwSNc7DzC95orEi8= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.13.3/go.mod h1:Bm/v2IaN6rZ+Op7zX+bOUMdL4fsrYZiD0dsjLhNKwZc= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.16.3/go.mod h1:KZgs2ny8HsxRIRbDwgvJcHHBZPOzQr/+NtGwnP+w2ec= -github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.27 h1:JEXSW4wztrl1MoL5EMvJMO7lc/TRZloztrJKNl96SW8= -github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.27/go.mod h1:8eL+YgEqy6IYqjwW6PG0Ubn59a2xtCzbz7Pi18JBu04= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28 h1:li8rTZAAb22g4UsxbjwMdaNVWbgVcDzPqI7nDTI+mF4= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.28/go.mod h1:/brXioSGIMEdcBFoubpSdmighSVp6poP+mma/wB7iHA= github.com/aws/aws-sdk-go-v2/service/kms v1.16.3/go.mod h1:QuiHPBqlOFCi4LqdSskYYAWpQlx3PKmohy+rE2F+o5g= github.com/aws/aws-sdk-go-v2/service/s3 v1.26.3/go.mod h1:g1qvDuRsJY+XghsV6zg00Z4KJ7DtFFCx8fJD2a491Ak= github.com/aws/aws-sdk-go-v2/service/s3 v1.43.0/go.mod h1:NXRKkiRF+erX2hnybnVU660cYT5/KChRD4iUgJ97cI8= -github.com/aws/aws-sdk-go-v2/service/s3 v1.103.1 h1:WkX5IXwcxgO/WPTvhEqoSW2L1GB1OyIxk0vuzzdTftc= -github.com/aws/aws-sdk-go-v2/service/s3 v1.103.1/go.mod h1:9Q9ZHyiTItraw8BXpO48pk398Mou0YCSI+xvFcaGgxU= +github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2 h1:b4ikkRk22T4xYkEgaWc3Voe+3xbt5YbbFhNehOWyUiY= +github.com/aws/aws-sdk-go-v2/service/s3 v1.103.2/go.mod h1:Gp7eHZ0NZ8ZK5RXpoIUp/C8OeAmJqpCgdwEK1D/QOek= github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.15.4/go.mod h1:PJc8s+lxyU8rrre0/4a0pn2wgwiDvOEzoOjcJUBr67o= -github.com/aws/aws-sdk-go-v2/service/signin v1.1.3 h1:t6U7sowMfOjTeZXtDOtgEJXsoJyX4MDag+sfWGwUM9M= -github.com/aws/aws-sdk-go-v2/service/signin v1.1.3/go.mod h1:WhO1EH3phjFWValQDsExaxncgEWJsHeoTvuyQAj3jwU= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.4 h1:YcpVyIPLCbiypN6KSphijN5fC7DDjX114SqA7prnnxg= +github.com/aws/aws-sdk-go-v2/service/signin v1.1.4/go.mod h1:5ZICS++oFTRPfa1GsBqFDWX/8WamZ/QQOcCzIuU/zLw= github.com/aws/aws-sdk-go-v2/service/sns v1.17.4/go.mod h1:kElt+uCcXxcqFyc+bQqZPFD9DME/eC6oHBXvFzQ9Bcw= github.com/aws/aws-sdk-go-v2/service/sqs v1.18.3/go.mod h1:skmQo0UPvsjsuYYSYMVmrPc1HWCbHUJyrCEp+ZaLzqM= github.com/aws/aws-sdk-go-v2/service/ssm v1.24.1/go.mod h1:NR/xoKjdbRJ+qx0pMR4mI+N/H1I1ynHwXnO6FowXJc0= github.com/aws/aws-sdk-go-v2/service/sso v1.11.3/go.mod h1:7UQ/e69kU7LDPtY40OyoHYgRmgfGM4mgsLYtcObdveU= github.com/aws/aws-sdk-go-v2/service/sso v1.17.2/go.mod h1:/pE21vno3q1h4bbhUOEi+6Zu/aT26UK2WKkDXd+TssQ= -github.com/aws/aws-sdk-go-v2/service/sso v1.31.1 h1:TUV8oytPCX1PfVgZn0N8/sPZx7T0YasaMCBHox1erlw= -github.com/aws/aws-sdk-go-v2/service/sso v1.31.1/go.mod h1:tEL1hqCrkgwrDVL04HuLxz1SLUXdh+4kKhWv1pXKeiY= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.2 h1:ySNWu7TPmj5fKFIa1GYvX+Ddxd5ccruqC20aMNuyWDM= +github.com/aws/aws-sdk-go-v2/service/sso v1.31.2/go.mod h1:A+U9luAOwFeB1kseyWCITVg7/NntoPebCFR9pQ4ch9A= github.com/aws/aws-sdk-go-v2/service/ssooidc v1.20.0/go.mod h1:dWqm5G767qwKPuayKfzm4rjzFmVjiBFbOJrpSPnAMDs= -github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.4 h1:p9+Fizo2sUB6wI5Yb3K5lmykQAGs5JrKLBV/me6613Y= -github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.4/go.mod h1:0x10Wy0dVS4Gn552xhHY5th2QdYpfJf44EsfyYGV194= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5 h1:KSzGGqfk39O+WU3OEyYbx6F7sLDQCqxlOJ+2IksfK6U= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.36.5/go.mod h1:ATs88lXDeQB6CZOgQ5BIl9JbYS+EsCWUSDyff6L/oVo= github.com/aws/aws-sdk-go-v2/service/sts v1.16.3/go.mod h1:bfBj0iVmsUyUg4weDB4NxktD9rDGeKSVWnjTnwbx9b8= github.com/aws/aws-sdk-go-v2/service/sts v1.25.3/go.mod h1:4EqRHDCKP78hq3zOnmFXu5k0j4bXbRFfCh/zQ6KnEfQ= -github.com/aws/aws-sdk-go-v2/service/sts v1.43.1 h1:r/vUkpLilfCA3sxbRnkHbJejaoVHEdj4FEhv+Zva4DU= -github.com/aws/aws-sdk-go-v2/service/sts v1.43.1/go.mod h1:t01JURC8Fe5M+7R1K0vzIZ2NT04HqvZR+FjlHrHDT2A= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.2 h1:RTO7mmGyedgnNmcPh3yQizNfc6GKoV5iqfdJavuf9vw= +github.com/aws/aws-sdk-go-v2/service/sts v1.43.2/go.mod h1:fBhUZXDin9YYqhcpOMjIcpdik25rVwWyxLdPH1RZd9s= github.com/aws/smithy-go v1.11.2/go.mod h1:3xHYmszWVx2c0kIwQeEVf9uSm4fYZt67FBJnwub1bgM= github.com/aws/smithy-go v1.17.0/go.mod h1:NukqUGpCZIILqqiV0NIjeFh24kd/FAa4beRb6nbIUPE= -github.com/aws/smithy-go v1.27.0 h1:ZoFioDKJxkSIW2otF9T0aPtNlUwhdVCcuZh/rzH9Hus= -github.com/aws/smithy-go v1.27.0/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/aws/smithy-go v1.27.1 h1:4T340VFndXtADGF52gYa1POyL7s9E4Z1OeZ1hCscIw8= +github.com/aws/smithy-go v1.27.1/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/bobg/gcsobj v0.1.2/go.mod h1:vS49EQ1A1Ib8FgrL58C8xXYZyOCR2TgzAdopy6/ipa8= github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8= diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle new file mode 100644 index 000000000000..1b08b867b291 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/build.gradle @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * License); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an AS IS BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +plugins { id 'org.apache.beam.module' } +applyJavaNature( + automaticModuleName: 'org.apache.beam.sdk.extensions.opentelemetry.gcp.auth', +) + +description = "Apache Beam :: SDKs :: Java :: Extensions :: OpenTelemetry GCP Auth" +ext.summary = "OpenTelemetry extension that provides GCP authentication support for OTLP exporters." + +dependencies { + implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom) + implementation platform(library.java.opentelemetry_bom) + implementation library.java.google_auth_library_oauth2_http + implementation library.java.vendored_guava_32_1_2_jre + compileOnly library.java.opentelemetry_api + compileOnly library.java.opentelemetry_extension_autoconfigure + compileOnly library.java.opentelemetry_exporter_otlp + implementation project(path: ":sdks:java:core", configuration: "shadow") + + testImplementation library.java.junit + testImplementation library.java.mockito_core + testImplementation library.java.mockito_inline + testImplementation library.java.jupiter_api + testImplementation library.java.jupiter_params + testRuntimeOnly library.java.jupiter_engine + testImplementation library.java.truth + testImplementation library.java.opentelemetry_api + testImplementation library.java.opentelemetry_sdk + testImplementation library.java.opentelemetry_exporter_otlp + testImplementation library.java.opentelemetry_sdk_testing + testImplementation library.java.opentelemetry_extension_autoconfigure +} +test { useJUnitPlatform() } diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java new file mode 100644 index 000000000000..e18e5693e3a1 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/ConfigurableOption.java @@ -0,0 +1,163 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static java.util.Locale.ROOT; + +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import java.util.Optional; +import java.util.function.Supplier; + +/** + * An enum representing configurable options for a GCP Authentication Extension. Each option has a + * user-readable name and can be configured using environment variables or system properties. + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/ConfigurableOption.java + */ +enum ConfigurableOption { + /** + * Represents the Google Cloud Project ID option. Can be configured using the environment variable + * `GOOGLE_CLOUD_PROJECT` or the system property `google.cloud.project`. + */ + GOOGLE_CLOUD_PROJECT("Google Cloud Project ID"), + + /** + * Represents the Google Cloud Quota Project ID option. Can be configured using the environment + * variable `GOOGLE_CLOUD_QUOTA_PROJECT` or the system property `google.cloud.quota.project`. The + * quota project is the project that is used for quota management and billing for the API usage. + * + *

The environment variable name is selected to be consistent with the official GCP client + * libraries. + */ + GOOGLE_CLOUD_QUOTA_PROJECT("Google Cloud Quota Project ID"), + + /** + * Specifies a comma-separated list of OpenTelemetry signals for which this authentication + * extension should be active. The authentication mechanisms provided by this extension will only + * be applied to the listed signals. If not set, {@code all} is assumed to be set which means + * authentication is enabled for all supported signals. + * + *

Valid signal values are: + * + *

+ * + *

The values are case-sensitive. Whitespace around commas and values is ignored. Can be + * configured using the environment variable `GOOGLE_OTEL_AUTH_TARGET_SIGNALS` or the system + * property `google.otel.auth.target.signals`. + */ + GOOGLE_OTEL_AUTH_TARGET_SIGNALS("Target Signals for Google Authentication Extension"); + + private final String userReadableName; + private final String environmentVariableName; + private final String systemPropertyName; + + ConfigurableOption(String userReadableName) { + this.userReadableName = userReadableName; + this.environmentVariableName = this.name(); + this.systemPropertyName = this.environmentVariableName.toLowerCase(ROOT).replace('_', '.'); + } + + /** + * Returns the environment variable name associated with this option. + * + * @return the environment variable name (e.g., GOOGLE_CLOUD_PROJECT) + */ + String getEnvironmentVariable() { + return this.environmentVariableName; + } + + /** + * Returns the system property name associated with this option. + * + * @return the system property name (e.g., google.cloud.project) + */ + String getSystemProperty() { + return this.systemPropertyName; + } + + /** + * Returns the user readable name associated with this option. + * + * @return the user readable name (e.g., "Google Cloud Quota Project ID") + */ + String getUserReadableName() { + return this.userReadableName; + } + + /** + * Retrieves the configured value for this option. This method checks the environment variable + * first and then the system property. + * + * @return The configured value as a string, or throws an exception if not configured. + * @throws ConfigurationException if neither the environment variable nor the system property is + * set. + */ + String getConfiguredValue(ConfigProperties configProperties) { + String configuredValue = configProperties.getString(this.getSystemProperty()); + if (configuredValue != null && !configuredValue.isEmpty()) { + return configuredValue; + } else { + throw new ConfigurationException( + String.format( + "GCP Authentication Extension not configured properly: %s not configured. Configure it by exporting environment variable %s or system property %s", + this.userReadableName, this.getEnvironmentVariable(), this.getSystemProperty())); + } + } + + /** + * Retrieves the value for this option, prioritizing environment variables and system properties. + * If neither an environment variable nor a system property is set for this option, the provided + * fallback function is used to determine the value. + * + * @param fallback A {@link Supplier} that provides the default value for the option when it is + * not explicitly configured via an environment variable or system property. + * @return The configured value for the option, obtained from the environment variable, system + * property, or the fallback function, in that order of precedence. + */ + String getConfiguredValueWithFallback( + ConfigProperties configProperties, Supplier fallback) { + try { + return this.getConfiguredValue(configProperties); + } catch (ConfigurationException e) { + return fallback.get(); + } + } + + /** + * Retrieves the value for this option, prioritizing environment variables before system + * properties. If neither an environment variable nor a system property is set for this option, + * then an empty {@link Optional} is returned. + * + * @return The configured value for the option, if set, obtained from the environment variable, + * system property, or empty {@link Optional}, in that order of precedence. + */ + Optional getConfiguredValueAsOptional(ConfigProperties configProperties) { + try { + return Optional.of(this.getConfiguredValue(configProperties)); + } catch (ConfigurationException e) { + return Optional.empty(); + } + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java new file mode 100644 index 000000000000..ce2b142591bd --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java @@ -0,0 +1,282 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static io.opentelemetry.api.common.AttributeKey.stringKey; +import static java.util.Arrays.stream; +import static java.util.stream.Collectors.joining; +import static java.util.stream.Collectors.toMap; + +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auto.service.AutoService; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizer; +import io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import io.opentelemetry.sdk.metrics.export.MetricExporter; +import io.opentelemetry.sdk.resources.Resource; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.apache.beam.sdk.annotations.Internal; +import org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GoogleAuthException.Reason; + +/** + * An AutoConfigurationCustomizerProvider for Google Cloud Platform (GCP) OpenTelemetry (OTLP) + * integration. + * + *

This class is registered as a service provider using {@link AutoService} and is responsible + * for customizing the OpenTelemetry configuration for GCP specific behavior. It retrieves Google + * Application Default Credentials (ADC) and adds them as authorization headers to the configured + * {@link SpanExporter}. It also sets default properties and resource attributes for GCP + * integration. + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/GcpAuthAutoConfigurationCustomizerProvider.java + * + * @see AutoConfigurationCustomizerProvider + * @see GoogleCredentials + */ +@AutoService(AutoConfigurationCustomizerProvider.class) +@Internal +public class GcpAuthAutoConfigurationCustomizerProvider + implements AutoConfigurationCustomizerProvider { + + static final String QUOTA_USER_PROJECT_HEADER = "x-goog-user-project"; + static final String GCP_USER_PROJECT_ID_KEY = "gcp.project_id"; + + static final String SIGNAL_TYPE_TRACES = "traces"; + static final String SIGNAL_TYPE_METRICS = "metrics"; + static final String SIGNAL_TYPE_ALL = "all"; + + private @Nullable GoogleCredentials credentials; + + /** + * Customizes the provided {@link AutoConfigurationCustomizer} such that authenticated exports to + * GCP Telemetry API are possible from the configured OTLP exporter. + * + *

This method performs the following: + * + *

+ * + * The 'customization' performed includes customizing the exporters by adding required headers to + * the export calls made and customizing the resource by adding required resource attributes to + * enable GCP integration. + * + * @param autoConfiguration the AutoConfigurationCustomizer to customize. + */ + @Override + public void customize(@Nonnull AutoConfigurationCustomizer autoConfiguration) { + autoConfiguration + .addSpanExporterCustomizer(this::customizeSpanExporter) + .addMetricExporterCustomizer(this::customizeMetricExporter) + .addResourceCustomizer(this::customizeResource); + } + + @Override + public int order() { + return Integer.MAX_VALUE - 1; + } + + private synchronized GoogleCredentials getCredentials() { + if (credentials == null) { + try { + credentials = GoogleCredentials.getApplicationDefault(); + } catch (IOException e) { + throw new GoogleAuthException(Reason.FAILED_ADC_RETRIEVAL, e); + } + } + return credentials; + } + + private SpanExporter customizeSpanExporter( + SpanExporter exporter, ConfigProperties configProperties) { + if (isSignalTargeted(SIGNAL_TYPE_TRACES, configProperties)) { + return addAuthorizationHeaders(exporter, configProperties); + } + return exporter; + } + + private MetricExporter customizeMetricExporter( + MetricExporter exporter, ConfigProperties configProperties) { + if (isSignalTargeted(SIGNAL_TYPE_METRICS, configProperties)) { + return addAuthorizationHeaders(exporter, configProperties); + } + return exporter; + } + + // Checks if the auth extension is configured to target the passed signal for authentication. + private static boolean isSignalTargeted(String checkSignal, ConfigProperties configProperties) { + String endpoint = configProperties.getString("otel.exporter.otlp." + checkSignal + ".endpoint"); + if (endpoint == null) { + endpoint = configProperties.getString("otel.exporter.otlp.endpoint"); + } + if (endpoint == null) { + return false; + } + + try { + java.net.URI uri = new java.net.URI(endpoint); + String host = uri.getHost(); + String scheme = uri.getScheme(); + if (host == null + || scheme == null + || !scheme.equalsIgnoreCase("https") + || (!host.equalsIgnoreCase("telemetry.googleapis.com") + && !host.equalsIgnoreCase("telemetry.mtls.googleapis.com"))) { + return false; + } + } catch (java.net.URISyntaxException e) { + return false; + } + + String userSpecifiedTargetedSignals = + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getConfiguredValueWithFallback( + configProperties, () -> SIGNAL_TYPE_ALL); + return stream(userSpecifiedTargetedSignals.split(",")) + .map(String::trim) + .anyMatch( + targetedSignal -> + targetedSignal.equals(checkSignal) || targetedSignal.equals(SIGNAL_TYPE_ALL)); + } + + private boolean isAnySignalTargeted(ConfigProperties configProperties) { + return isSignalTargeted(SIGNAL_TYPE_TRACES, configProperties) + || isSignalTargeted(SIGNAL_TYPE_METRICS, configProperties); + } + + // Adds authorization headers to the calls made by the OtlpGrpcSpanExporter and + // OtlpHttpSpanExporter. + private SpanExporter addAuthorizationHeaders( + SpanExporter exporter, ConfigProperties configProperties) { + if (exporter instanceof OtlpHttpSpanExporter) { + SpanExporter result = + ((OtlpHttpSpanExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; + } else if (exporter instanceof OtlpGrpcSpanExporter) { + SpanExporter result = + ((OtlpGrpcSpanExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; + } + return exporter; + } + + // Adds authorization headers to the calls made by the OtlpGrpcMetricExporter and + // OtlpHttpMetricExporter. + private MetricExporter addAuthorizationHeaders( + MetricExporter exporter, ConfigProperties configProperties) { + if (exporter instanceof OtlpHttpMetricExporter) { + MetricExporter result = + ((OtlpHttpMetricExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; + } else if (exporter instanceof OtlpGrpcMetricExporter) { + MetricExporter result = + ((OtlpGrpcMetricExporter) exporter) + .toBuilder() + .setHeaders(() -> getRequiredHeaderMap(configProperties)) + .build(); + exporter.shutdown(); + return result; + } + return exporter; + } + + private Map getRequiredHeaderMap(ConfigProperties configProperties) { + GoogleCredentials creds = getCredentials(); + Map> gcpHeaders; + try { + // this also refreshes the credentials, if required + gcpHeaders = creds.getRequestMetadata(); + } catch (IOException e) { + throw new GoogleAuthException(Reason.FAILED_ADC_REFRESH, e); + } + if (gcpHeaders == null) { + return Map.of(); + } + Map flattenedHeaders = + gcpHeaders.entrySet().stream() + .filter(entry -> entry.getKey() != null && entry.getValue() != null) + .collect( + toMap( + Map.Entry::getKey, + entry -> + entry.getValue().stream() + .filter(Objects::nonNull) // Filter nulls + .filter(s -> !s.isEmpty()) // Filter empty strings + .collect(joining(",")))); + // Add quota user project header if not detected by the auth library and user provided it via + // system properties. + if (!flattenedHeaders.containsKey(QUOTA_USER_PROJECT_HEADER)) { + Optional maybeConfiguredQuotaProjectId = + ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getConfiguredValueAsOptional( + configProperties); + maybeConfiguredQuotaProjectId.ifPresent( + configuredQuotaProjectId -> + flattenedHeaders.put(QUOTA_USER_PROJECT_HEADER, configuredQuotaProjectId)); + } + return flattenedHeaders; + } + + // Updates the current resource with the attributes required for ingesting OTLP data on GCP. + private Resource customizeResource(Resource resource, ConfigProperties configProperties) { + if (!isAnySignalTargeted(configProperties)) { + return resource; + } + + String gcpProjectId; + try { + gcpProjectId = ConfigurableOption.GOOGLE_CLOUD_PROJECT.getConfiguredValue(configProperties); + } catch (ConfigurationException e) { + gcpProjectId = getCredentials().getProjectId(); + if (gcpProjectId == null || gcpProjectId.isEmpty()) { + throw e; + } + } + Resource res = Resource.create(Attributes.of(stringKey(GCP_USER_PROJECT_ID_KEY), gcpProjectId)); + return resource.merge(res); + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java new file mode 100644 index 000000000000..978b48934067 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GoogleAuthException.java @@ -0,0 +1,69 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; + +/** + * An unchecked exception indicating a failure during Google authentication. This exception is + * thrown when there are issues with retrieving or refreshing Google Application Default Credentials + * (ADC). + * + *

Copied from + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/main/java/io/opentelemetry/contrib/gcp/auth/GoogleAuthException.java + */ +public class GoogleAuthException extends RuntimeException { + + private static final long serialVersionUID = 149908685226796448L; + + /** + * Constructs a new {@code GoogleAuthException} with the specified reason and cause. + * + * @param reason the reason for the authentication failure. + * @param cause the underlying cause of the exception (e.g., an IOException). + */ + GoogleAuthException(Reason reason, Throwable cause) { + super(reason.message, cause); + } + + /** Enumerates the possible reasons for a Google authentication failure. */ + enum Reason { + /** Indicates a failure to retrieve Google Application Default Credentials. */ + FAILED_ADC_RETRIEVAL("Unable to retrieve Google Application Default Credentials."), + /** Indicates a failure to retrieve Google Application Default Credentials. */ + FAILED_ADC_REFRESH("Unable to refresh Google Application Default Credentials."); + + private final String message; + + /** + * Constructs a new {@code Reason} with the specified message. + * + * @param message the message describing the reason. + */ + Reason(String message) { + this.message = message; + } + + /** + * Returns the message associated with this reason. + * + * @return the message describing the reason. + */ + public String getMessage() { + return message; + } + } +} diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java new file mode 100644 index 000000000000..e389a1a7dc9e --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/package-info.java @@ -0,0 +1,20 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +/** Google Cloud Platform (GCP) OpenTelemetry (OTLP) authentication extension. */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider new file mode 100644 index 000000000000..bf1ba2cad985 --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/main/resources/META-INF/services/io.opentelemetry.sdk.autoconfigure.spi.AutoConfigurationCustomizerProvider @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider diff --git a/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java new file mode 100644 index 000000000000..e1db55c3635b --- /dev/null +++ b/sdks/java/extensions/opentelemetry-gcp-auth-extension/src/test/java/org/apache/beam/sdk/extensions/opentelemetry/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java @@ -0,0 +1,1234 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.extensions.opentelemetry.gcp.auth; + +import static com.google.common.truth.Truth.assertThat; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.GCP_USER_PROJECT_ID_KEY; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.QUOTA_USER_PROJECT_HEADER; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_ALL; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_METRICS; +import static org.apache.beam.sdk.extensions.opentelemetry.gcp.auth.GcpAuthAutoConfigurationCustomizerProvider.SIGNAL_TYPE_TRACES; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.GoogleCredentials; +import com.google.auto.value.AutoValue; +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.common.Attributes; +import io.opentelemetry.api.metrics.LongCounter; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.common.ComponentLoader; +import io.opentelemetry.context.Scope; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporter; +import io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter; +import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk; +import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder; +import io.opentelemetry.sdk.autoconfigure.internal.SpiHelper; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties; +import io.opentelemetry.sdk.autoconfigure.spi.ConfigurationException; +import io.opentelemetry.sdk.autoconfigure.spi.metrics.ConfigurableMetricExporterProvider; +import io.opentelemetry.sdk.autoconfigure.spi.traces.ConfigurableSpanExporterProvider; +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.common.export.MemoryMode; +import io.opentelemetry.sdk.metrics.Aggregation; +import io.opentelemetry.sdk.metrics.InstrumentType; +import io.opentelemetry.sdk.metrics.data.AggregationTemporality; +import io.opentelemetry.sdk.metrics.data.MetricData; +import io.opentelemetry.sdk.metrics.export.MetricExporter; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import java.io.IOException; +import java.time.Duration; +import java.time.Instant; +import java.util.AbstractMap.SimpleEntry; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.List; +import java.util.Map; +import java.util.Map.Entry; +import java.util.Random; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.function.Supplier; +import java.util.stream.Stream; +import javax.annotation.Nonnull; +import javax.annotation.Nullable; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.ArgumentCaptor; +import org.mockito.Captor; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; +import org.mockito.stubbing.Answer; + +/** + * Copied from open-telemetry. Link: + * https://github.com/open-telemetry/opentelemetry-java-contrib/blob/main/gcp-auth-extension/src/test/java/io/opentelemetry/contrib/gcp/auth/GcpAuthAutoConfigurationCustomizerProviderTest.java + */ +class GcpAuthAutoConfigurationCustomizerProviderTest { + + private static final String DUMMY_GCP_RESOURCE_PROJECT_ID = "my-gcp-resource-project-id"; + private static final String DUMMY_GCP_QUOTA_PROJECT_ID = "my-gcp-quota-project-id"; + private static final Random TEST_RANDOM = new Random(); + + @Mock private GoogleCredentials mockedGoogleCredentials; + + @Captor private ArgumentCaptor>> traceHeaderSupplierCaptor; + @Captor private ArgumentCaptor>> metricHeaderSupplierCaptor; + + private static final ImmutableMap DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER = + ImmutableMap.builder() + .put("otel.exporter.otlp.traces.endpoint", "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "none") + .put("otel.logs.exporter", "none") + .put("otel.resource.attributes", "foo=bar") + .build(); + + private static final ImmutableMap DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER = + ImmutableMap.builder() + .put("otel.exporter.otlp.metrics.endpoint", "https://telemetry.googleapis.com/v1/metrics") + .put("otel.traces.exporter", "none") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .put("otel.resource.attributes", "foo=bar") + .build(); + + @BeforeEach + public void setup() { + MockitoAnnotations.openMocks(this); + } + + @AfterEach + public void teardown() { + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); + } + + // TODO: Use parameterized test for testing traces customizer for http & grpc. + @Test + void testTraceCustomizerOtlpHttp() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpHttpSpanExporter mockOtlpHttpSpanExporter = mock(OtlpHttpSpanExporter.class); + OtlpHttpSpanExporterBuilder otlpSpanExporterBuilder = OtlpHttpSpanExporter.builder(); + OtlpHttpSpanExporterBuilder spyOtlpHttpSpanExporterBuilder = + Mockito.spy(otlpSpanExporterBuilder); + when(spyOtlpHttpSpanExporterBuilder.build()).thenReturn(mockOtlpHttpSpanExporter); + + when(mockOtlpHttpSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess()); + List exportedSpans = new ArrayList<>(); + when(mockOtlpHttpSpanExporter.export(any())) + .thenAnswer( + invocationOnMock -> { + exportedSpans.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + Mockito.when(mockOtlpHttpSpanExporter.toBuilder()).thenReturn(spyOtlpHttpSpanExporterBuilder); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpHttpSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpHttpSpanExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpHttpSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpHttpSpanExporter, Mockito.atLeast(1)).export(Mockito.anyCollection()); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(spanData.getAttributes().asMap()).containsKey(AttributeKey.longKey("work_loop")); + } + } + } + + @Test + void testTraceCustomizerOtlpGrpc() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpGrpcSpanExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpGrpcSpanExporter, Mockito.atLeast(1)).export(Mockito.anyCollection()); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(spanData.getAttributes().asMap()).containsKey(AttributeKey.longKey("work_loop")); + } + } + } + + @Test + void testMetricCustomizerOtlpHttp() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + SIGNAL_TYPE_METRICS); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpHttpMetricExporter mockOtlpHttpMetricExporter = Mockito.mock(OtlpHttpMetricExporter.class); + OtlpHttpMetricExporterBuilder otlpMetricExporterBuilder = OtlpHttpMetricExporter.builder(); + OtlpHttpMetricExporterBuilder spyOtlpHttpMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureHttpMockMetricExporter( + mockOtlpHttpMetricExporter, spyOtlpHttpMetricExporterBuilder, exportedMetrics); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpHttpMetricExporter); + generateTestMetric(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpHttpMetricExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpHttpMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpHttpMetricExporter, Mockito.atLeast(1)) + .export(Mockito.anyCollection()); + + assertThat(exportedMetrics).isNotEmpty(); + for (MetricData metricData : exportedMetrics) { + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(metricData.getLongSumData().getPoints()).isNotEmpty(); + for (io.opentelemetry.sdk.metrics.data.PointData pointData : + metricData.getLongSumData().getPoints()) { + assertThat(pointData.getAttributes().asMap()) + .containsKey(AttributeKey.longKey("work_loop")); + } + } + } + } + + @Test + void testMetricCustomizerOtlpGrpc() { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + SIGNAL_TYPE_METRICS); + // Prepare mocks + prepareMockBehaviorForGoogleCredentials(); + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter = Mockito.mock(OtlpGrpcMetricExporter.class); + OtlpGrpcMetricExporterBuilder otlpMetricExporterBuilder = OtlpGrpcMetricExporter.builder(); + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureGrpcMockMetricExporter( + mockOtlpGrpcMetricExporter, spyOtlpGrpcMetricExporterBuilder, exportedMetrics); + + // begin assertions + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcMetricExporter); + generateTestMetric(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + Mockito.verify(mockOtlpGrpcMetricExporter, Mockito.times(1)).toBuilder(); + Mockito.verify(spyOtlpGrpcMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + + Mockito.verify(mockOtlpGrpcMetricExporter, Mockito.atLeast(1)) + .export(Mockito.anyCollection()); + + assertThat(exportedMetrics).isNotEmpty(); + for (MetricData metricData : exportedMetrics) { + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), DUMMY_GCP_RESOURCE_PROJECT_ID); + assertThat(metricData.getResource().getAttributes().asMap()) + .containsEntry(AttributeKey.stringKey("foo"), "bar"); + assertThat(metricData.getLongSumData().getPoints()).isNotEmpty(); + for (io.opentelemetry.sdk.metrics.data.PointData pointData : + metricData.getLongSumData().getPoints()) { + assertThat(pointData.getAttributes().asMap()) + .containsKey(AttributeKey.longKey("work_loop")); + } + } + } + } + + @Test + void testCustomizerFailWithMissingResourceProject() { + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_ALL); + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + assertThrows( + ConfigurationException.class, + () -> buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter)); + } + } + + @ParameterizedTest + @MethodSource("provideQuotaBehaviorTestCases") + @SuppressWarnings("CannotMockMethod") + void testQuotaProjectBehavior(QuotaProjectIdTestBehavior testCase) throws IOException { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_ALL); + + // Prepare request metadata + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + ImmutableMap> mockedRequestMetadata; + if (testCase.getIsQuotaProjectPresentInMetadata()) { + mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue()), + QUOTA_USER_PROJECT_HEADER, + Collections.singletonList(DUMMY_GCP_QUOTA_PROJECT_ID)); + } else { + mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue())); + } + // mock credentials to return the prepared request metadata + Mockito.when(mockedGoogleCredentials.getRequestMetadata()).thenReturn(mockedRequestMetadata); + + // configure environment according to test case + String quotaProjectId = testCase.getUserSpecifiedQuotaProjectId(); // maybe empty string + if (quotaProjectId != null) { + // user specified a quota project id + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_QUOTA_PROJECT.getSystemProperty(), quotaProjectId); + } + + // prepare mock exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + // Export telemetry to capture headers in the export calls + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + + // assert that the Authorization bearer token header is present + Map exportHeaders = traceHeaderSupplierCaptor.getValue().get(); + assertThat(exportHeaders).containsEntry("Authorization", "Bearer fake"); + + if (testCase.getExpectedQuotaProjectInHeader() == null) { + // there should be no user quota project header + assertThat(exportHeaders).doesNotContainKey(QUOTA_USER_PROJECT_HEADER); + } else { + // there should be user quota project header with expected value + assertThat(exportHeaders) + .containsEntry(QUOTA_USER_PROJECT_HEADER, testCase.getExpectedQuotaProjectInHeader()); + } + } + } + + @ParameterizedTest + @MethodSource("provideProjectIdBehaviorTestCases") + @SuppressWarnings("CannotMockMethod") + void testProjectIdBehavior(ProjectIdTestBehavior testCase) throws IOException { + System.clearProperty(ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty()); + System.clearProperty(ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty()); + + // configure environment according to test case + String userSpecifiedProjectId = testCase.getUserSpecifiedProjectId(); + if (userSpecifiedProjectId != null) { + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), userSpecifiedProjectId); + } + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), SIGNAL_TYPE_TRACES); + + // prepare request metadata (may or may not be called depending on test scenario) + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + ImmutableMap> mockedRequestMetadata = + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue())); + Mockito.lenient() + .when(mockedGoogleCredentials.getRequestMetadata()) + .thenReturn(mockedRequestMetadata); + + // only mock getProjectId() if it will be called (i.e., user didn't specify project ID) + boolean shouldFallbackToCredentials = + userSpecifiedProjectId == null || userSpecifiedProjectId.isEmpty(); + if (shouldFallbackToCredentials) { + Mockito.when(mockedGoogleCredentials.getProjectId()) + .thenReturn(testCase.getCredentialsProjectId()); + } + + // prepare mock exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + if (testCase.getExpectedToThrow()) { + // expect exception to be thrown when project ID is not available + assertThrows( + ConfigurationException.class, + () -> buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter)); + // verify getProjectId() was called to attempt fallback + Mockito.verify(mockedGoogleCredentials, Mockito.times(1)).getProjectId(); + } else { + // export telemetry and verify resource attributes contain expected project ID + OpenTelemetrySdk sdk = buildOpenTelemetrySdkWithExporter(mockOtlpGrpcSpanExporter); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + assertThat(exportedSpans).isNotEmpty(); + for (SpanData spanData : exportedSpans) { + assertThat(spanData.getResource().getAttributes().asMap()) + .containsEntry( + AttributeKey.stringKey(GCP_USER_PROJECT_ID_KEY), + testCase.getExpectedProjectIdInResource()); + } + + // verify whether getProjectId() was called based on whether fallback was needed + if (shouldFallbackToCredentials) { + Mockito.verify(mockedGoogleCredentials, Mockito.times(1)).getProjectId(); + } else { + Mockito.verify(mockedGoogleCredentials, Mockito.never()).getProjectId(); + } + } + } + } + + @ParameterizedTest + @MethodSource("provideTargetSignalBehaviorTestCases") + void testTargetSignalsBehavior(TargetSignalBehavior testCase) { + // Set resource project system property + System.setProperty( + ConfigurableOption.GOOGLE_CLOUD_PROJECT.getSystemProperty(), DUMMY_GCP_RESOURCE_PROJECT_ID); + // Prepare mocks + // Prepare mocked credential + prepareMockBehaviorForGoogleCredentials(); + + // Prepare mocked span exporter + OtlpGrpcSpanExporter mockOtlpGrpcSpanExporter = Mockito.mock(OtlpGrpcSpanExporter.class); + OtlpGrpcSpanExporterBuilder spyOtlpGrpcSpanExporterBuilder = + Mockito.spy(OtlpGrpcSpanExporter.builder()); + List exportedSpans = new ArrayList<>(); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + configureGrpcMockSpanExporter( + mockOtlpGrpcSpanExporter, spyOtlpGrpcSpanExporterBuilder, exportedSpans); + + // Prepare mocked metrics exporter + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter = Mockito.mock(OtlpGrpcMetricExporter.class); + OtlpGrpcMetricExporterBuilder otlpMetricExporterBuilder = OtlpGrpcMetricExporter.builder(); + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder = + Mockito.spy(otlpMetricExporterBuilder); + List exportedMetrics = new ArrayList<>(); + configureGrpcMockMetricExporter( + mockOtlpGrpcMetricExporter, spyOtlpGrpcMetricExporterBuilder, exportedMetrics); + + // configure environment according to test case + System.setProperty( + ConfigurableOption.GOOGLE_OTEL_AUTH_TARGET_SIGNALS.getSystemProperty(), + testCase.getConfiguredTargetSignals()); + + // Build Autoconfigured OpenTelemetry SDK using the mocks and send signals + try (MockedStatic googleCredentialsMockedStatic = + Mockito.mockStatic(GoogleCredentials.class)) { + googleCredentialsMockedStatic + .when(GoogleCredentials::getApplicationDefault) + .thenReturn(mockedGoogleCredentials); + + OpenTelemetrySdk sdk = + buildOpenTelemetrySdkWithExporter( + mockOtlpGrpcSpanExporter, + mockOtlpGrpcMetricExporter, + testCase.getUserSpecifiedOtelProperties()); + generateTestMetric(sdk); + generateTestSpan(sdk); + CompletableResultCode code = sdk.shutdown(); + CompletableResultCode joinResult = code.join(10, TimeUnit.SECONDS); + assertThat(joinResult.isSuccess()).isTrue(); + + // Check Traces modification conditions + if (testCase.getExpectedIsTraceSignalModified()) { + // If traces signal is expected to be modified, auth headers must be present + Mockito.verify(spyOtlpGrpcSpanExporterBuilder, Mockito.times(1)) + .setHeaders(traceHeaderSupplierCaptor.capture()); + assertThat(traceHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(traceHeaderSupplierCaptor.getValue().get())) + .isTrue(); + } else { + // If traces signals is not expected to be modified then no interaction with the builder + // should be made + Mockito.verifyNoInteractions(spyOtlpGrpcSpanExporterBuilder); + } + + // Check Metric modification conditions + if (testCase.getExpectedIsMetricsSignalModified()) { + // If metrics signal is expected to be modified, auth headers must be present + Mockito.verify(spyOtlpGrpcMetricExporterBuilder, Mockito.times(1)) + .setHeaders(metricHeaderSupplierCaptor.capture()); + assertThat(metricHeaderSupplierCaptor.getValue().get().size()).isEqualTo(2); + assertThat(authHeadersQuotaProjectIsPresent(metricHeaderSupplierCaptor.getValue().get())) + .isTrue(); + } else { + // If metrics signals is not expected to be modified then no interaction with the builder + // should be made + Mockito.verifyNoInteractions(spyOtlpGrpcMetricExporterBuilder); + } + } + } + + /** Test cases specifying expected behavior for GOOGLE_OTEL_AUTH_TARGET_SIGNALS. */ + private static Stream provideTargetSignalBehaviorTestCases() { + return Stream.of( + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("traces") + .setUserSpecifiedOtelProperties(DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics") + .setUserSpecifiedOtelProperties(DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("all") + .setUserSpecifiedOtelProperties( + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics, traces") + .setUserSpecifiedOtelProperties( + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("") + .setUserSpecifiedOtelProperties( + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(true) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("all") + .setUserSpecifiedOtelProperties( + ImmutableMap.of( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics", + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces", + "otel.traces.exporter", + "none", + "otel.metrics.exporter", + "none", + "otel.logs.exporter", + "none")) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metric, trace") + .setUserSpecifiedOtelProperties( + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) + .setExpectedIsMetricsSignalModified(false) + .setExpectedIsTraceSignalModified(false) + .build()), + Arguments.of( + TargetSignalBehavior.builder() + .setConfiguredTargetSignals("metrics, trace") + .setUserSpecifiedOtelProperties( + ImmutableMap.builder() + .put( + "otel.exporter.otlp.metrics.endpoint", + "https://telemetry.googleapis.com/v1/metrics") + .put( + "otel.exporter.otlp.traces.endpoint", + "https://telemetry.googleapis.com/v1/traces") + .put("otel.traces.exporter", "otlp") + .put("otel.metrics.exporter", "otlp") + .put("otel.logs.exporter", "none") + .build()) + .setExpectedIsMetricsSignalModified(true) + .setExpectedIsTraceSignalModified(false) + .build())); + } + + /** + * Test cases specifying expected value for the project ID in the resource given the user input + * and the current credentials state. + * + *

{@code null} for {@link ProjectIdTestBehavior#getUserSpecifiedProjectId()} indicates the + * case of user not specifying the project ID. + * + *

{@code null} value for {@link ProjectIdTestBehavior#getCredentialsProjectId()} indicates + * that the mocked credentials are not providing a project ID. + * + *

{@code true} for {@link ProjectIdTestBehavior#getExpectedToThrow()} indicates the + * expectation that an exception should be thrown. + */ + private static Stream provideProjectIdBehaviorTestCases() { + return Stream.of( + // User specified project ID takes precedence + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(DUMMY_GCP_RESOURCE_PROJECT_ID) + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource(DUMMY_GCP_RESOURCE_PROJECT_ID) + .setExpectedToThrow(false) + .build()), + // If user specified project ID is empty, fallback to credentials.getProjectId() + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource("credentials-project-id") + .setExpectedToThrow(false) + .build()), + // If user doesn't specify project ID, fallback to credentials.getProjectId() + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(null) + .setCredentialsProjectId("credentials-project-id") + .setExpectedProjectIdInResource("credentials-project-id") + .setExpectedToThrow(false) + .build()), + // If user doesn't specify and credentials.getProjectId() returns null, throw exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId(null) + .setCredentialsProjectId(null) + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build()), + // If user specified project ID is empty and credentials.getProjectId() returns null, throw + // exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId(null) + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build()), + // If user specifies empty and credentials returns empty (edge case), throw exception + Arguments.of( + ProjectIdTestBehavior.builder() + .setUserSpecifiedProjectId("") + .setCredentialsProjectId("") + .setExpectedProjectIdInResource(null) + .setExpectedToThrow(true) + .build())); + } + + /** + * Test cases specifying expected value for the user quota project header given the user input and + * the current credentials state. + * + *

{@code null} for {@link QuotaProjectIdTestBehavior#getUserSpecifiedQuotaProjectId()} + * indicates the case of user not specifying the quota project ID. + * + *

{@code null} value for {@link QuotaProjectIdTestBehavior#getExpectedQuotaProjectInHeader()} + * indicates the expectation that the QUOTA_USER_PROJECT_HEADER should not be present in the + * export headers. + * + *

{@code true} for {@link QuotaProjectIdTestBehavior#getIsQuotaProjectPresentInMetadata()} + * indicates that the mocked credentials are configured to provide DUMMY_GCP_QUOTA_PROJECT_ID as + * the quota project ID. + */ + private static Stream provideQuotaBehaviorTestCases() { + return Stream.of( + // If quota project present in metadata, it will be used + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(DUMMY_GCP_QUOTA_PROJECT_ID) + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("my-custom-quota-project-id") + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + // If quota project not present in request metadata, then user specified project is used + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(DUMMY_GCP_QUOTA_PROJECT_ID) + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("my-custom-quota-project-id") + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader("my-custom-quota-project-id") + .build()), + // Testing for special edge case inputs + // user-specified quota project is empty + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("") // user explicitly specifies empty + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId("") + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(null) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(null) // user omits specifying quota project + .setIsQuotaProjectPresentInMetadata(true) + .setExpectedQuotaProjectInHeader(DUMMY_GCP_QUOTA_PROJECT_ID) + .build()), + Arguments.of( + QuotaProjectIdTestBehavior.builder() + .setUserSpecifiedQuotaProjectId(null) + .setIsQuotaProjectPresentInMetadata(false) + .setExpectedQuotaProjectInHeader(null) + .build())); + } + + // Configure necessary behavior on the gRPC mock span exporters to work. + // Mockito.lenient is used here because this method is used with parameterized tests where based + // on certain inputs, certain stubbings may not be required. + private static void configureGrpcMockSpanExporter( + OtlpGrpcSpanExporter mockGrpcExporter, + OtlpGrpcSpanExporterBuilder spyGrpcExporterBuilder, + List exportedSpanContainer) { + Mockito.lenient().when(spyGrpcExporterBuilder.build()).thenReturn(mockGrpcExporter); + Mockito.lenient() + .when(mockGrpcExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.lenient().when(mockGrpcExporter.toBuilder()).thenReturn(spyGrpcExporterBuilder); + Mockito.lenient() + .when(mockGrpcExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedSpanContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + } + + // Configure necessary behavior on the http mock metric exporters to work. + private static void configureHttpMockMetricExporter( + OtlpHttpMetricExporter mockOtlpHttpMetricExporter, + OtlpHttpMetricExporterBuilder spyOtlpHttpMetricExporterBuilder, + List exportedMetricContainer) { + Mockito.when(spyOtlpHttpMetricExporterBuilder.build()).thenReturn(mockOtlpHttpMetricExporter); + Mockito.when(mockOtlpHttpMetricExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.when(mockOtlpHttpMetricExporter.toBuilder()) + .thenReturn(spyOtlpHttpMetricExporterBuilder); + Mockito.when(mockOtlpHttpMetricExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedMetricContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + // mock the get default aggregation and aggregation temporality - they're required for valid + // metric collection. + Mockito.when(mockOtlpHttpMetricExporter.getDefaultAggregation(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpHttpMetricExporter.getDefault().getDefaultAggregation(instrumentType); + }); + Mockito.when(mockOtlpHttpMetricExporter.getAggregationTemporality(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpHttpMetricExporter.getDefault() + .getAggregationTemporality(instrumentType); + }); + } + + // Configure necessary behavior on the gRPC mock metrics exporters to work. + // Mockito.lenient is used here because this method is used with parameterized tests where based + // on certain inputs, certain stubbings may not be required. + private static void configureGrpcMockMetricExporter( + OtlpGrpcMetricExporter mockOtlpGrpcMetricExporter, + OtlpGrpcMetricExporterBuilder spyOtlpGrpcMetricExporterBuilder, + List exportedMetricContainer) { + Mockito.lenient() + .when(spyOtlpGrpcMetricExporterBuilder.build()) + .thenReturn(mockOtlpGrpcMetricExporter); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.shutdown()) + .thenReturn(CompletableResultCode.ofSuccess()); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.toBuilder()) + .thenReturn(spyOtlpGrpcMetricExporterBuilder); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.export(Mockito.anyCollection())) + .thenAnswer( + invocationOnMock -> { + exportedMetricContainer.addAll(invocationOnMock.getArgument(0)); + return CompletableResultCode.ofSuccess(); + }); + // mock the get default aggregation and aggregation temporality - they're required for valid + // metric collection. + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getDefaultAggregation(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpGrpcMetricExporter.getDefault().getDefaultAggregation(instrumentType); + }); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getAggregationTemporality(Mockito.any())) + .thenAnswer( + (Answer) + invocationOnMock -> { + InstrumentType instrumentType = invocationOnMock.getArgument(0); + return OtlpGrpcMetricExporter.getDefault() + .getAggregationTemporality(instrumentType); + }); + Mockito.lenient() + .when(mockOtlpGrpcMetricExporter.getMemoryMode()) + .thenReturn(MemoryMode.IMMUTABLE_DATA); + } + + @AutoValue + abstract static class ProjectIdTestBehavior { + // A null user specified project ID represents the use case where user omits specifying it + @Nullable + abstract String getUserSpecifiedProjectId(); + + // The project ID that credentials.getProjectId() returns (can be null) + @Nullable + abstract String getCredentialsProjectId(); + + // The expected project ID in the resource attributes (null if exception expected) + @Nullable + abstract String getExpectedProjectIdInResource(); + + // Whether an exception is expected to be thrown + abstract boolean getExpectedToThrow(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_ProjectIdTestBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setUserSpecifiedProjectId(String projectId); + + abstract Builder setCredentialsProjectId(String projectId); + + abstract Builder setExpectedProjectIdInResource(String projectId); + + abstract Builder setExpectedToThrow(boolean expectedToThrow); + + abstract ProjectIdTestBehavior build(); + } + } + + @AutoValue + abstract static class QuotaProjectIdTestBehavior { + // A null user specified quota represents the use case where user omits specifying quota + @Nullable + abstract String getUserSpecifiedQuotaProjectId(); + + abstract boolean getIsQuotaProjectPresentInMetadata(); + + // If expected quota project in header is null, the header entry should not be present in export + @Nullable + abstract String getExpectedQuotaProjectInHeader(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_QuotaProjectIdTestBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setUserSpecifiedQuotaProjectId(String quotaProjectId); + + abstract Builder setIsQuotaProjectPresentInMetadata(boolean quotaProjectPresentInMetadata); + + /** + * Sets the expected quota project header value for the test case. A null value is allowed, + * and it indicates that the header should not be present in the export request. + * + * @param expectedQuotaProjectInHeader the expected header value to match in the export + * headers. + */ + abstract Builder setExpectedQuotaProjectInHeader(String expectedQuotaProjectInHeader); + + abstract QuotaProjectIdTestBehavior build(); + } + } + + @AutoValue + abstract static class TargetSignalBehavior { + @Nonnull + abstract String getConfiguredTargetSignals(); + + @Nonnull + abstract ImmutableMap getUserSpecifiedOtelProperties(); + + abstract boolean getExpectedIsTraceSignalModified(); + + abstract boolean getExpectedIsMetricsSignalModified(); + + static Builder builder() { + return new AutoValue_GcpAuthAutoConfigurationCustomizerProviderTest_TargetSignalBehavior + .Builder(); + } + + @AutoValue.Builder + abstract static class Builder { + abstract Builder setConfiguredTargetSignals(String targetSignals); + + abstract Builder setUserSpecifiedOtelProperties(Map oTelProperties); + + // Set whether the combination of specified OTel properties and configured target signals + // should lead to modification of the OTLP trace exporters. + abstract Builder setExpectedIsTraceSignalModified(boolean expectedModified); + + // Set whether the combination of specified OTel properties and configured target signals + // should lead to modification of the OTLP metrics exporters. + abstract Builder setExpectedIsMetricsSignalModified(boolean expectedModified); + + abstract TargetSignalBehavior build(); + } + } + + // Mockito.lenient is used here because this method is used with parameterized tests where based + @SuppressWarnings("CannotMockMethod") + private void prepareMockBehaviorForGoogleCredentials() { + AccessToken fakeAccessToken = new AccessToken("fake", Date.from(Instant.now())); + try { + Mockito.lenient() + .when(mockedGoogleCredentials.getRequestMetadata()) + .thenReturn( + ImmutableMap.of( + "Authorization", + Collections.singletonList("Bearer " + fakeAccessToken.getTokenValue()), + QUOTA_USER_PROJECT_HEADER, + Collections.singletonList(DUMMY_GCP_QUOTA_PROJECT_ID))); + } catch (IOException e) { + throw new RuntimeException(e); + } + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(SpanExporter spanExporter) { + return buildOpenTelemetrySdkWithExporter( + spanExporter, OtlpHttpMetricExporter.getDefault(), DEFAULT_OTEL_PROPERTIES_SPAN_EXPORTER); + } + + @SuppressWarnings("UnusedMethod") + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + SpanExporter spanExporter, ImmutableMap customOTelProperties) { + return buildOpenTelemetrySdkWithExporter( + spanExporter, OtlpHttpMetricExporter.getDefault(), customOTelProperties); + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter(MetricExporter metricExporter) { + return buildOpenTelemetrySdkWithExporter( + OtlpHttpSpanExporter.getDefault(), metricExporter, DEFAULT_OTEL_PROPERTIES_METRIC_EXPORTER); + } + + @SuppressWarnings("UnusedMethod") + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + MetricExporter metricExporter, ImmutableMap customOtelProperties) { + return buildOpenTelemetrySdkWithExporter( + OtlpHttpSpanExporter.getDefault(), metricExporter, customOtelProperties); + } + + private OpenTelemetrySdk buildOpenTelemetrySdkWithExporter( + SpanExporter spanExporter, + MetricExporter metricExporter, + ImmutableMap customOtelProperties) { + SpiHelper spiHelper = + SpiHelper.create(GcpAuthAutoConfigurationCustomizerProviderTest.class.getClassLoader()); + AutoConfiguredOpenTelemetrySdkBuilder builder = + AutoConfiguredOpenTelemetrySdk.builder() + .addPropertiesSupplier(() -> customOtelProperties) + .setComponentLoader( + new ComponentLoader() { + @Override + public List load(Class spiClass) { + if (spiClass == ConfigurableSpanExporterProvider.class) { + return Collections.singletonList( + spiClass.cast( + new ConfigurableSpanExporterProvider() { + @Override + public SpanExporter createExporter( + ConfigProperties configProperties) { + return spanExporter; + } + + @Override + public String getName() { + return "otlp"; + } + })); + } + if (spiClass == ConfigurableMetricExporterProvider.class) { + return Collections.singletonList( + spiClass.cast( + new ConfigurableMetricExporterProvider() { + @Override + public MetricExporter createExporter( + ConfigProperties configProperties) { + return metricExporter; + } + + @Override + public String getName() { + return "otlp"; + } + })); + } + return spiHelper.load(spiClass); + } + }); + + return builder.build().getOpenTelemetrySdk(); + } + + private static boolean authHeadersQuotaProjectIsPresent(Map headers) { + Set> headerEntrySet = headers.entrySet(); + return headerEntrySet.contains( + new SimpleEntry<>( + QUOTA_USER_PROJECT_HEADER, + GcpAuthAutoConfigurationCustomizerProviderTest.DUMMY_GCP_QUOTA_PROJECT_ID)) + && headerEntrySet.contains(new SimpleEntry<>("Authorization", "Bearer fake")); + } + + private static void generateTestSpan(OpenTelemetrySdk openTelemetrySdk) { + Span span = openTelemetrySdk.getTracer("test").spanBuilder("sample").startSpan(); + try (Scope ignored = span.makeCurrent()) { + long workOutput = busyloop(); + span.setAttribute("work_loop", workOutput); + } finally { + span.end(); + } + } + + private static void generateTestMetric(OpenTelemetrySdk openTelemetrySdk) { + LongCounter longCounter = + openTelemetrySdk + .getMeter("test") + .counterBuilder("sample") + .setDescription("sample counter") + .setUnit("1") + .build(); + long workOutput = busyloop(); + long randomValue = TEST_RANDOM.nextInt(1000); + longCounter.add(randomValue, Attributes.of(AttributeKey.longKey("work_loop"), workOutput)); + } + + // loop to simulate work done + private static long busyloop() { + Instant start = Instant.now(); + Instant end; + long counter = 0; + do { + counter++; + end = Instant.now(); + } while (Duration.between(start, end).toMillis() < 1000); + return counter; + } +} diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java index e3d947235015..1d618ba685ed 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProvider.java @@ -179,11 +179,11 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { PCollection inputRows = input.getSinglePCollection(); BigQueryIO.Write write = createStorageWriteApiTransform(inputRows.getSchema()); + int numStreams = configuration.getNumStreams() == null ? 0 : configuration.getNumStreams(); if (inputRows.isBounded() == IsBounded.UNBOUNDED) { Long triggeringFrequency = configuration.getTriggeringFrequencySeconds(); Boolean autoSharding = configuration.getAutoSharding(); - int numStreams = configuration.getNumStreams() == null ? 0 : configuration.getNumStreams(); boolean useAtLeastOnceSemantics = configuration.getUseAtLeastOnceSemantics() != null @@ -197,12 +197,13 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) { : Duration.standardSeconds(triggeringFrequency)); } // set num streams if specified, otherwise default to autoSharding - if (numStreams > 0) { - write = write.withNumStorageWriteApiStreams(numStreams); - } else if (autoSharding == null || autoSharding) { + if (numStreams == 0 && (autoSharding == null || autoSharding)) { write = write.withAutoSharding(); } } + if (numStreams > 0) { + write = write.withNumStorageWriteApiStreams(numStreams); + } Schema inputSchema = inputRows.getSchema(); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryWriteConfiguration.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryWriteConfiguration.java index 55d7f7c8d72a..7f09feb245cd 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryWriteConfiguration.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryWriteConfiguration.java @@ -101,6 +101,12 @@ public void validate() { Boolean autoSharding = getAutoSharding(); Integer numStreams = getNumStreams(); + if (numStreams != null) { + checkArgument( + numStreams >= 0, + invalidConfigMessage + "numStreams must be non-negative, but was: %s", + numStreams); + } if (autoSharding != null && autoSharding && numStreams != null) { checkArgument( numStreams == 0, @@ -152,8 +158,7 @@ public static Builder builder() { public abstract Boolean getAutoSharding(); @SchemaFieldDescription( - "Specifies the number of write streams that the Storage API sink will use. " - + "This parameter is only applicable when writing unbounded data.") + "Specifies the number of write streams that the Storage API sink will use.") @Nullable public abstract Integer getNumStreams(); diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProviderTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProviderTest.java index 584309778286..81789f784255 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProviderTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/providers/BigQueryStorageWriteApiSchemaTransformProviderTest.java @@ -122,7 +122,10 @@ public void testInvalidConfig() { Arrays.asList( BigQueryWriteConfiguration.builder() .setTable("project:dataset.table") - .setCreateDisposition("INVALID_DISPOSITION")); + .setCreateDisposition("INVALID_DISPOSITION"), + BigQueryWriteConfiguration.builder() + .setTable("project:dataset.table") + .setNumStreams(-1)); for (BigQueryWriteConfiguration.Builder config : invalidConfigs) { assertThrows( diff --git a/sdks/python/apache_beam/io/gcp/bigquery.py b/sdks/python/apache_beam/io/gcp/bigquery.py index d751d60c905f..a2d17f12569e 100644 --- a/sdks/python/apache_beam/io/gcp/bigquery.py +++ b/sdks/python/apache_beam/io/gcp/bigquery.py @@ -2155,8 +2155,7 @@ def __init__( all of FILE_LOADS, STREAMING_INSERTS, and STORAGE_WRITE_API. Only applicable to unbounded input. num_storage_api_streams: Specifies the number of write streams that the - Storage API sink will use. This parameter is only applicable when - writing unbounded data. + Storage API sink will use. ignore_unknown_columns: Accept rows that contain values that do not match the schema. The unknown values are ignored. Default is False, which treats unknown values as errors. This option is only valid for diff --git a/settings.gradle.kts b/settings.gradle.kts index 603832045f3e..443d9c567752 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -192,6 +192,7 @@ include(":sdks:java:extensions:avro") include(":sdks:java:extensions:euphoria") include(":sdks:java:extensions:kryo") include(":sdks:java:extensions:google-cloud-platform-core") +include(":sdks:java:extensions:opentelemetry-gcp-auth-extension") include(":sdks:java:extensions:jackson") include(":sdks:java:extensions:join-library") include(":sdks:java:extensions:kafka-factories") diff --git a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md index 9c205f092663..00cd1ffbcdef 100644 --- a/website/www/site/content/en/documentation/io/built-in/google-bigquery.md +++ b/website/www/site/content/en/documentation/io/built-in/google-bigquery.md @@ -855,7 +855,7 @@ pipeline uses. You can set it explicitly on the transform via [`withNumStorageWriteApiStreams`](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.Write.html#withNumStorageWriteApiStreams-int-) or provide the `numStorageWriteApiStreams` option to the pipeline as defined in [`BigQueryOptions`](https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/gcp/bigquery/BigQueryOptions.html). -Please note this is only supported for streaming pipelines. +Fixed stream counts can be used with both batch and streaming pipelines. Triggering frequency determines how soon the data is visible for querying in BigQuery. You can explicitly set it via