diff --git a/servicetalk-examples/http/opentelemetry-jaeger/build.gradle b/servicetalk-examples/http/opentelemetry-jaeger/build.gradle new file mode 100644 index 0000000000..0fca0a9043 --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/build.gradle @@ -0,0 +1,52 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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. + */ + +apply plugin: "java" +apply plugin: "application" +apply from: "../../gradle/idea.gradle" + +application { + mainClass = "io.servicetalk.examples.http.opentelemetry.jaeger.JaegerOtlpExample" +} + +run { + systemProperty "io.opentelemetry.sdk.common.export.GrpcSenderProvider", "io.servicetalk.opentelemetry.client.ServiceTalkGrpcSenderProvider" +} + +dependencies { + implementation platform("io.opentelemetry:opentelemetry-bom:$opentelemetryVersion") + + implementation project(":servicetalk-annotations") + implementation project(":servicetalk-http-netty") + implementation project(":servicetalk-opentelemetry-http") // OpenTelemetry client/server filters + implementation project(":servicetalk-test-resources") // test certificates + + // ServiceTalk OpenTelemetry transport provider (SPI discovery) + runtimeOnly project(":servicetalk-opentelemetry-client") + runtimeOnly project(":servicetalk-opentelemetry-asynccontext") // OpenTelemetry async context propagation + + // OpenTelemetry Java SDK + implementation "io.opentelemetry:opentelemetry-api" + implementation "io.opentelemetry:opentelemetry-sdk" + // OTLP exporter (uses SPI to find transport) + // Exclude default OkHttp sender to use ServiceTalk's implementation exclusively + implementation("io.opentelemetry:opentelemetry-exporter-otlp") { +// exclude group: 'io.opentelemetry', module: 'opentelemetry-exporter-sender-okhttp' + } + + implementation "org.slf4j:slf4j-api:$slf4jVersion" + runtimeOnly "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" +} diff --git a/servicetalk-examples/http/opentelemetry-jaeger/docker-compose.yml b/servicetalk-examples/http/opentelemetry-jaeger/docker-compose.yml new file mode 100644 index 0000000000..43f6852f9b --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/docker-compose.yml @@ -0,0 +1,40 @@ +# Docker Compose configuration for Jaeger with mTLS-enabled OTLP Collector +# +# This setup includes: +# - Jaeger all-in-one for trace visualization +# - OpenTelemetry Collector with mTLS for secure trace ingestion + +version: '3.8' + +services: + # Jaeger backend for trace storage and visualization + jaeger: + image: jaegertracing/all-in-one:latest + container_name: jaeger + environment: + - COLLECTOR_OTLP_ENABLED=true + ports: + - "16686:16686" # Jaeger UI + - "4317:4317" # OTLP gRPC (internal, for collector) + - "4318:4318" # OTLP HTTP (internal, for collector) + networks: + - otel-network + +# OpenTelemetry Collector with mTLS + # otel-collector: + # image: otel/opentelemetry-collector:latest + # container_name: otel-collector + # command: ["--config=/etc/otel-collector-config.yaml"] + # volumes: + # - ./otel-collector-config.yaml:/etc/otel-collector-config.yaml:ro + # - ./certs:/certs:ro + # ports: + # - "4318:4318" # OTLP HTTP with mTLS + # depends_on: + # - jaeger + # networks: + # - otel-network + +networks: + otel-network: + driver: bridge diff --git a/servicetalk-examples/http/opentelemetry-jaeger/generate-certs.sh b/servicetalk-examples/http/opentelemetry-jaeger/generate-certs.sh new file mode 100755 index 0000000000..76062cdfb5 --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/generate-certs.sh @@ -0,0 +1,85 @@ +#!/bin/bash +# +# Copyright © 2026 Apple Inc. and the ServiceTalk project authors +# +# Licensed 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. +# + +set -e + +CERTS_DIR="./certs" + +echo "Generating mTLS certificates for OTLP collector..." +echo "" + +# Create certs directory +mkdir -p "$CERTS_DIR" +cd "$CERTS_DIR" + +# Generate CA private key and certificate +echo "1. Generating CA certificate..." +openssl genrsa -out ca-key.pem 4096 +openssl req -new -x509 -days 365 -key ca-key.pem -out ca-cert.pem \ + -subj "/C=US/ST=California/L=Cupertino/O=ServiceTalk/CN=ServiceTalk CA" + +# Generate server private key and certificate signing request +echo "2. Generating server certificate..." +openssl genrsa -out server-key.pem 4096 +openssl req -new -key server-key.pem -out server-csr.pem \ + -subj "/C=US/ST=California/L=Cupertino/O=ServiceTalk/CN=localhost" + +# Create server certificate extensions file +cat > server-ext.cnf < client-ext.cnf < + * This example shows: + * + *

+ * Prerequisites: + *

    + *
  1. Generate TLS certificates (run from example directory): + * 
    + *   ./generate-certs.sh
+ *
    + *
    2. + *
  2. Start Jaeger and OpenTelemetry Collector with mTLS: + * 
    + *   docker-compose up -d
+ *
    + *
    3. + *
  3. Run this example
    4. + *
  4. View traces in Jaeger UI: http://localhost:16686
    5. + *
  5. Stop services: + * 
    + *   docker-compose down
+ *
    + *
    6. + *
+ *

+ * Note: Set {@code USE_MTLS = false} to run without mTLS (requires plain HTTP collector). + */ +public final class JaegerOtlpExample { + + private static final Logger LOGGER = LoggerFactory.getLogger(JaegerOtlpExample.class); + private static final char[] PASSWORD = "changeit".toCharArray(); + private static final String SERVICE_NAME = "servicetalk-jaeger-example"; + private static final boolean USE_GRPC = true; + private static final boolean USE_MTLS = false; // Set to false for plain HTTP + private static final String JAEGER_OTLP_HTTP_ENDPOINT = (USE_MTLS ? "https" : "http") + "://localhost:4318/v1/traces"; + + private JaegerOtlpExample() { + // no instances. + } + + public static void main(String[] args) throws Exception { + System.setProperty("io.opentelemetry.sdk.common.export.GrpcSenderProvider", + "io.servicetalk.opentelemetry.client.ServiceTalkGrpcSenderProvider"); + // Configure OpenTelemetry SDK with OTLP exporter + // The ServiceTalk HTTP transport will be automatically discovered via SPI + LOGGER.info("Configuring OpenTelemetry with OTLP exporter..."); + LOGGER.info("ServiceTalk HTTP transport will be automatically discovered via SPI"); + + GlobalOpenTelemetry.set(configureOpenTelemetry()); + + // Start a simple HTTP server + LOGGER.info("Starting HTTP server on port 8080..."); + HttpServerContext serverContext = startServer(); + + try { + LOGGER.info("Server started successfully"); + LOGGER.info("======================================================================"); + LOGGER.info("Making HTTP requests to generate trace data..."); + LOGGER.info("======================================================================"); + // Create HTTP client and make requests + makeRequestsWithTracing(); + + // Give time for spans to be exported + LOGGER.info("Waiting for spans to be exported to Jaeger..."); + Thread.sleep(3000); + + LOGGER.info("======================================================================"); + LOGGER.info("Example completed successfully!"); + LOGGER.info("View traces in Jaeger UI: http://localhost:16686"); + LOGGER.info("Search for service: {}", SERVICE_NAME); + LOGGER.info("======================================================================"); + } finally { + LOGGER.debug("Closing server context"); + serverContext.close(); + // Shutdown OpenTelemetry to flush remaining spans + OpenTelemetry openTelemetry = GlobalOpenTelemetry.get(); + if (openTelemetry instanceof OpenTelemetrySdk) { + LOGGER.debug("Shutting down OpenTelemetry SDK"); + ((OpenTelemetrySdk) openTelemetry).close(); + } + } + } + + private static OpenTelemetry configureOpenTelemetry() throws Exception { + LOGGER.debug("Creating OpenTelemetry resource with service name: {}", SERVICE_NAME); + // Create resource with service name + Resource resource = Resource.getDefault() + .merge(Resource.create(Attributes.of( + AttributeKey.stringKey("service.name"), SERVICE_NAME, + AttributeKey.stringKey("service.version"), "1.0.0" + ))); + + SpanExporter spanExporter = buildSpanExporter(); + + // Create tracer provider with batch span processor + LOGGER.debug("Creating tracer provider with batch span processor"); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .setResource(resource) + .addSpanProcessor(BatchSpanProcessor.builder(spanExporter) + .setScheduleDelay(Duration.ofMillis(500)) + .build()) + .build(); + + // Build and return OpenTelemetry SDK + LOGGER.debug("Building OpenTelemetry SDK"); + return OpenTelemetrySdk.builder() + .setTracerProvider(tracerProvider) + .build(); + } + + private static SpanExporter buildSpanExporter() throws Exception { + return USE_GRPC ? buildGrpcExporter() : buildHttpExporter(); + } + + private static SpanExporter buildGrpcExporter() throws Exception { + OtlpGrpcSpanExporterBuilder exporterBuilder = OtlpGrpcSpanExporter.builder() + .setTimeout(Duration.ofSeconds(10)); + + // Configure mTLS if enabled + if (USE_MTLS) { + LOGGER.info("Configuring mTLS for OTLP exporter..."); + SSLContext sslContext = createMtlsSslContext(); + exporterBuilder.setSslContext(sslContext, createTrustManager()); + LOGGER.info("mTLS configuration complete"); + } + + return exporterBuilder.build(); + } + + private static SpanExporter buildHttpExporter() throws Exception { + OtlpHttpSpanExporterBuilder exporterBuilder = OtlpHttpSpanExporter.builder() + .setEndpoint(JAEGER_OTLP_HTTP_ENDPOINT) + .setTimeout(Duration.ofSeconds(10)); + + // Configure mTLS if enabled + if (USE_MTLS) { + LOGGER.info("Configuring mTLS for OTLP exporter..."); + SSLContext sslContext = createMtlsSslContext(); + exporterBuilder.setSslContext(sslContext, createTrustManager()); + LOGGER.info("mTLS configuration complete"); + } + + return exporterBuilder.build(); + } + + private static SSLContext createMtlsSslContext() throws Exception { + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(DefaultTestCerts.loadTruststoreP12(), PASSWORD); + + TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance( + TrustManagerFactory.getDefaultAlgorithm()); + trustManagerFactory.init(trustStore); + LOGGER.debug("Trust store configured with CA certificate"); + + // Create key store with client certificate and private key + final KeyStore keyStore = KeyStore.getInstance("pkcs12"); + keyStore.load(DefaultTestCerts.loadClientP12(), PASSWORD); + + KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance( + KeyManagerFactory.getDefaultAlgorithm()); + keyManagerFactory.init(keyStore, PASSWORD); + LOGGER.debug("Key store configured with client certificate and private key"); + + // Create SSL context + SSLContext sslContext = SSLContext.getInstance("TLS"); + sslContext.init(keyManagerFactory.getKeyManagers(), + trustManagerFactory.getTrustManagers(), null); + LOGGER.debug("SSL context created with mTLS configuration"); + return sslContext; + } + + /** + * Creates a TrustManager for the CA certificate. + */ + private static X509TrustManager createTrustManager() throws Exception { + TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance( + TrustManagerFactory.getDefaultAlgorithm()); + KeyStore trustStore = KeyStore.getInstance(KeyStore.getDefaultType()); + trustStore.load(DefaultTestCerts.loadTruststoreP12(), PASSWORD); + trustManagerFactory.init(trustStore); + return (X509TrustManager) trustManagerFactory.getTrustManagers()[0]; + } + + private static HttpServerContext startServer() throws Exception { + LOGGER.debug("Creating OpenTelemetry HTTP service filter"); + OpenTelemetryHttpServiceFilter serverFilter = new OpenTelemetryHttpServiceFilter.Builder() + .build(); + + LOGGER.debug("Building HTTP server for port 8080"); + return HttpServers.forPort(8080) + .appendServiceFilter(serverFilter) + .listenBlockingAndAwait((ctx, request, responseFactory) -> { + String path = request.path(); + LOGGER.debug("Server received request: {}", path); + + // Add some processing delay to make spans more visible + Thread.sleep(50); + + if ("/hello".equals(path)) { + return responseFactory.ok() + .payloadBody("Hello from ServiceTalk!", textSerializer()); + } else if ("/world".equals(path)) { + return responseFactory.ok() + .payloadBody("World response from ServiceTalk!", textSerializer()); + } else { + return responseFactory.ok() + .payloadBody("Default response", textSerializer()); + } + }); + } + + private static void makeRequestsWithTracing() throws Exception { + LOGGER.debug("Getting tracer for service: {}", SERVICE_NAME); + Tracer tracer = GlobalOpenTelemetry.get().getTracer(SERVICE_NAME); + + LOGGER.debug("Creating OpenTelemetry HTTP requester filter"); + OpenTelemetryHttpRequesterFilter clientFilter = new OpenTelemetryHttpRequesterFilter.Builder() + .componentName(SERVICE_NAME + "-client") + .build(); + + LOGGER.debug("Building HTTP client"); + try (BlockingHttpClient client = HttpClients.forSingleAddress("localhost", 8080) + .appendClientFilter(clientFilter) + .buildBlocking()) { + + // Request 1: Simple GET to /hello + LOGGER.info("1. Making GET request to /hello..."); + Span span1 = tracer.spanBuilder("request-1-workflow") + .setAttribute("workflow.step", "initial") + .startSpan(); + + try { + HttpResponse response1 = client.request(client.get("/hello")); + String body1 = response1.payloadBody(textDeserializer()); + LOGGER.info(" Response: {} - {}", response1.status(), body1); + span1.addEvent("received-response"); + } finally { + span1.end(); + } + + Thread.sleep(100); + + // Request 2: GET to /world with custom span + LOGGER.info("2. Making GET request to /world..."); + Span span2 = tracer.spanBuilder("request-2-workflow") + .setAttribute("workflow.step", "secondary") + .startSpan(); + + try { + HttpResponse response2 = client.request(client.get("/world")); + String body2 = response2.payloadBody(textDeserializer()); + LOGGER.info(" Response: {} - {}", response2.status(), body2); + span2.addEvent("received-response"); + } finally { + span2.end(); + } + + Thread.sleep(100); + + // Request 3: Demonstrate nested spans + LOGGER.info("3. Making request with nested spans..."); + Span parentSpan = tracer.spanBuilder("complex-workflow") + .setAttribute("workflow.type", "multi-step") + .startSpan(); + + try { + // Child span for preparation + LOGGER.debug("Creating preparation span"); + Span prepSpan = tracer.spanBuilder("prepare-request") + .setParent(io.opentelemetry.context.Context.current().with(parentSpan)) + .startSpan(); + try { + Thread.sleep(50); + prepSpan.addEvent("preparation-complete"); + } finally { + prepSpan.end(); + } + + // Make the actual request + HttpResponse response3 = client.request(client.get("/hello")); + String body3 = response3.payloadBody(textDeserializer()); + LOGGER.info(" Response: {} - {}", response3.status(), body3); + + // Child span for post-processing + LOGGER.debug("Creating post-processing span"); + Span postSpan = tracer.spanBuilder("process-response") + .setParent(io.opentelemetry.context.Context.current().with(parentSpan)) + .startSpan(); + try { + Thread.sleep(50); + postSpan.setAttribute("response.length", body3.length()); + postSpan.addEvent("processing-complete"); + } finally { + postSpan.end(); + } + } finally { + parentSpan.end(); + } + + LOGGER.info("✓ All requests completed successfully"); + LOGGER.info(" Generated spans:"); + LOGGER.info(" - 3 workflow spans (custom instrumentation)"); + LOGGER.info(" - 3 HTTP client spans (automatic)"); + LOGGER.info(" - 3 HTTP server spans (automatic)"); + LOGGER.info(" - 2 nested preparation/processing spans"); + } + } +} diff --git a/servicetalk-examples/http/opentelemetry-jaeger/src/main/java/io/servicetalk/examples/http/opentelemetry/jaeger/package-info.java b/servicetalk-examples/http/opentelemetry-jaeger/src/main/java/io/servicetalk/examples/http/opentelemetry/jaeger/package-info.java new file mode 100644 index 0000000000..b3527b6382 --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/src/main/java/io/servicetalk/examples/http/opentelemetry/jaeger/package-info.java @@ -0,0 +1,27 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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. + */ + +/** + * End-to-end example demonstrating ServiceTalk's OpenTelemetry integration with Jaeger. + *

+ * This package shows how ServiceTalk's HTTP transport is automatically discovered and used + * by OpenTelemetry SDK via Java's Service Provider Interface (SPI) mechanism. No explicit + * transport configuration is required - simply having servicetalk-opentelemetry-client on + * the classpath is sufficient. + * + * @see io.servicetalk.examples.http.opentelemetry.jaeger.JaegerOtlpExample + */ +package io.servicetalk.examples.http.opentelemetry.jaeger; diff --git a/servicetalk-examples/http/opentelemetry-jaeger/src/main/resources/log4j2.xml b/servicetalk-examples/http/opentelemetry-jaeger/src/main/resources/log4j2.xml new file mode 100644 index 0000000000..bc0e38aa3c --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/src/main/resources/log4j2.xml @@ -0,0 +1,38 @@ + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/servicetalk-examples/http/opentelemetry-jaeger/start_jaeger.sh b/servicetalk-examples/http/opentelemetry-jaeger/start_jaeger.sh new file mode 100755 index 0000000000..370a496515 --- /dev/null +++ b/servicetalk-examples/http/opentelemetry-jaeger/start_jaeger.sh @@ -0,0 +1 @@ +docker-compose -f docker-compose.yml up diff --git a/servicetalk-opentelemetry-client/build.gradle b/servicetalk-opentelemetry-client/build.gradle new file mode 100755 index 0000000000..28a2236296 --- /dev/null +++ b/servicetalk-opentelemetry-client/build.gradle @@ -0,0 +1,56 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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. + */ + +apply plugin: "io.servicetalk.servicetalk-gradle-plugin-internal-library" + +dependencies { + implementation project(":servicetalk-annotations") + implementation project(":servicetalk-concurrent-api") + implementation project(":servicetalk-http-api") + implementation project(":servicetalk-http-netty") + implementation project(":servicetalk-http-utils") + implementation project(":servicetalk-grpc-netty") + implementation project(":servicetalk-grpc-api") + implementation project(":servicetalk-buffer-api") + implementation "org.slf4j:slf4j-api:$slf4jVersion" + + implementation platform("io.opentelemetry:opentelemetry-bom:$opentelemetryVersion") + implementation "io.opentelemetry:opentelemetry-sdk" + implementation "io.opentelemetry:opentelemetry-sdk-common" + implementation "io.opentelemetry:opentelemetry-exporter-common" + implementation "io.opentelemetry:opentelemetry-exporter-otlp-common" + + testImplementation enforcedPlatform("org.junit:junit-bom:$junit5Version") + testImplementation "org.junit.jupiter:junit-jupiter-api" + testImplementation "org.mockito:mockito-core:$mockitoCoreVersion" + testImplementation "io.opentelemetry:opentelemetry-sdk-testing" + testImplementation("io.opentelemetry:opentelemetry-exporter-otlp") { + // Exclude OkHttp sender to force use of ServiceTalk sender via SPI + exclude group: 'io.opentelemetry', module: 'opentelemetry-exporter-sender-okhttp' + } + // OpenTelemetry protobuf definitions for decoding and verifying OTLP messages in tests + // This is critical for verifying that gRPC framing is correct + testImplementation "io.opentelemetry.proto:opentelemetry-proto:1.4.0-alpha" + testImplementation project(":servicetalk-test-resources") + testImplementation project(":servicetalk-transport-netty-internal") + testImplementation project(":servicetalk-buffer-netty") + // ServiceTalk's gRPC protobuf serialization infrastructure for validating gRPC framing + testImplementation project(":servicetalk-grpc-protobuf") + + // Log4j2 for test logging + testRuntimeOnly "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" + testRuntimeOnly "org.apache.logging.log4j:log4j-core:$log4jVersion" +} diff --git a/servicetalk-opentelemetry-client/gradle.lockfile b/servicetalk-opentelemetry-client/gradle.lockfile new file mode 100644 index 0000000000..f65d13cbc8 --- /dev/null +++ b/servicetalk-opentelemetry-client/gradle.lockfile @@ -0,0 +1,42 @@ +# This is a Gradle generated file for dependency locking. +# Manual edits can break the build and are not advised. +# This file is expected to be part of source control. +com.google.api.grpc:proto-google-common-protos:2.51.0=compileClasspath,runtimeClasspath +com.google.code.findbugs:jsr305:3.0.2=compileClasspath,runtimeClasspath +com.google.protobuf:protobuf-bom:3.25.8=runtimeClasspath +com.google.protobuf:protobuf-java:3.25.8=runtimeClasspath +io.netty:netty-bom:4.1.133.Final=runtimeClasspath +io.netty:netty-buffer:4.1.133.Final=runtimeClasspath +io.netty:netty-codec-dns:4.1.133.Final=runtimeClasspath +io.netty:netty-codec-http2:4.1.133.Final=runtimeClasspath +io.netty:netty-codec-http:4.1.133.Final=runtimeClasspath +io.netty:netty-codec:4.1.133.Final=runtimeClasspath +io.netty:netty-common:4.1.133.Final=runtimeClasspath +io.netty:netty-handler:4.1.133.Final=runtimeClasspath +io.netty:netty-resolver-dns-classes-macos:4.1.133.Final=runtimeClasspath +io.netty:netty-resolver-dns-native-macos:4.1.133.Final=runtimeClasspath +io.netty:netty-resolver-dns:4.1.133.Final=runtimeClasspath +io.netty:netty-resolver:4.1.133.Final=runtimeClasspath +io.netty:netty-tcnative-boringssl-static:2.0.77.Final=runtimeClasspath +io.netty:netty-tcnative-classes:2.0.77.Final=runtimeClasspath +io.netty:netty-transport-classes-epoll:4.1.133.Final=runtimeClasspath +io.netty:netty-transport-classes-kqueue:4.1.133.Final=runtimeClasspath +io.netty:netty-transport-native-epoll:4.1.133.Final=runtimeClasspath +io.netty:netty-transport-native-kqueue:4.1.133.Final=runtimeClasspath +io.netty:netty-transport-native-unix-common:4.1.133.Final=runtimeClasspath +io.netty:netty-transport:4.1.133.Final=runtimeClasspath +io.opentelemetry:opentelemetry-api:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-bom:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-common:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-context:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-exporter-common:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-exporter-otlp-common:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk-common:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk-extension-autoconfigure-spi:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk-logs:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk-metrics:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk-trace:1.59.0=compileClasspath,runtimeClasspath +io.opentelemetry:opentelemetry-sdk:1.59.0=compileClasspath,runtimeClasspath +org.jctools:jctools-core:4.0.5=runtimeClasspath +org.slf4j:slf4j-api:1.7.36=compileClasspath,runtimeClasspath +empty=annotationProcessor diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/AbstractServiceTalkSender.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/AbstractServiceTalkSender.java new file mode 100644 index 0000000000..059d571c1c --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/AbstractServiceTalkSender.java @@ -0,0 +1,151 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.buffer.api.Buffer; +import io.servicetalk.http.api.HttpClient; +import io.servicetalk.http.api.HttpHeaders; +import io.servicetalk.http.api.HttpRequest; +import io.servicetalk.http.api.HttpResponseStatus; + +import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.common.export.Compressor; +import io.opentelemetry.sdk.common.export.MessageWriter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.OutputStream; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.Consumer; +import java.util.function.Supplier; +import javax.annotation.Nullable; + +abstract class AbstractServiceTalkSender { + + private static final Logger LOGGER = LoggerFactory.getLogger(AbstractServiceTalkSender.class); + + private final HttpClient httpClient; + private final AtomicBoolean isShutdown; + private final String requestTarget; + @Nullable + private final Supplier>> headersSupplier; + + protected AbstractServiceTalkSender(HttpClient httpClient, String requestTarget, + @Nullable Supplier>> headersSupplier) { + this.httpClient = httpClient; + this.requestTarget = requestTarget; + this.headersSupplier = headersSupplier; + this.isShutdown = new AtomicBoolean(false); + } + + protected abstract Response buildResponse(HttpResponseStatus status, HttpHeaders headers, byte[] responseBody); + + protected final void doSend(MessageWriter messageWriter, + Consumer onResponse, + Consumer onError) { + try { + LOGGER.debug("Preparing request to: {}", requestTarget); + // both http and grpc are post requests, it's just a matter of what exactly. + io.servicetalk.http.api.HttpRequest request = httpClient.post(requestTarget); + prepareMessage(messageWriter, request); + + LOGGER.debug("Executing request to: {}", requestTarget); + // Handle response asynchronously + httpClient.request(request).subscribe(httpResponse -> { + byte[] responseBody = readResponseBody(httpResponse.payloadBody()); + LOGGER.debug("Received response: status={}, size={} bytes", + httpResponse.status(), responseBody.length); + Response response = buildResponse(httpResponse.status(), httpResponse.headers(), responseBody); + onResponse.accept(response); + }, error -> { + LOGGER.debug("Request failed with error", error); + onError.accept(error); + }); + + } catch (Exception e) { + LOGGER.debug("Exception during request preparation", e); + onError.accept(e); + } + } + + public CompletableResultCode shutdown() { + if (!isShutdown.compareAndSet(false, true)) { + LOGGER.debug("Sender already shutdown"); + return CompletableResultCode.ofSuccess(); + } + + LOGGER.debug("Shutting down sender"); + CompletableResultCode result = new CompletableResultCode(); + try { + httpClient.closeAsync().subscribe( + () -> { + LOGGER.debug("Sender shutdown complete"); + result.succeed(); + }, + throwable -> { + LOGGER.debug("Sender shutdown failed", throwable); + result.fail(); + } + ); + } catch (Exception e) { + LOGGER.debug("Exception during sender shutdown", e); + result.fail(); + } + return result; + } + + protected abstract void prepareMessage(MessageWriter messageWriter, HttpRequest request) throws IOException; + + final void applyHeaders(HttpRequest request) { + // Apply dynamic headers from config + if (headersSupplier != null) { + Map> headers = headersSupplier.get(); + LOGGER.debug("Applying {} dynamic headers", headers.size()); + headers.forEach((key, values) -> { + if (values == null || values.isEmpty()) { + return; + } + if (values.size() == 1) { + request.setHeader(key, values.get(0)); + } else { + request.headers().set(key, values); + } + }); + } + } + + private static byte[] readResponseBody(Buffer responseBuffer) { + byte[] responseBody = new byte[responseBuffer.readableBytes()]; + responseBuffer.getBytes(responseBuffer.readerIndex(), responseBody); + return responseBody; + } + + static byte[] readMessage(MessageWriter messageWriter, @Nullable Compressor compressor) throws IOException { + int contentLength = messageWriter.getContentLength(); + LOGGER.debug("Converting message writer to buffer: contentLength={}", contentLength); + ByteArrayOutputStream baos = new ByteArrayOutputStream(contentLength); + try (OutputStream toWrite = compressor == null ? baos : compressor.compress(baos)) { + messageWriter.writeMessage(toWrite); + } + byte[] bytes = baos.toByteArray(); + LOGGER.debug("Message written: originalSize={}, finalSize={}", contentLength, bytes.length); + return bytes; + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSender.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSender.java new file mode 100644 index 0000000000..2503483ad0 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSender.java @@ -0,0 +1,141 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.buffer.api.Buffer; +import io.servicetalk.buffer.api.BufferAllocator; +import io.servicetalk.grpc.api.GrpcHeaderValues; +import io.servicetalk.http.api.HttpClient; +import io.servicetalk.http.api.HttpHeaders; +import io.servicetalk.http.api.HttpRequest; +import io.servicetalk.http.api.HttpResponseStatus; + +import io.opentelemetry.sdk.common.export.Compressor; +import io.opentelemetry.sdk.common.export.GrpcResponse; +import io.opentelemetry.sdk.common.export.GrpcSender; +import io.opentelemetry.sdk.common.export.MessageWriter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Supplier; +import javax.annotation.Nullable; + +import static io.servicetalk.grpc.api.GrpcHeaderNames.GRPC_MESSAGE_ENCODING; +import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_TYPE; + +final class ServiceTalkGrpcSender extends AbstractServiceTalkSender implements GrpcSender { + + private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTalkGrpcSender.class); + + private static final byte UNCOMPRESSED_FLAG = 0; + private static final byte COMPRESSED_FLAG = 1; + private static final int HEADER_LENGTH = 5; + + @Nullable + private final Compressor compressor; + private final BufferAllocator bufferAllocator; + + ServiceTalkGrpcSender(HttpClient httpClient, + @Nullable Compressor compressor, + String fullMethodName, + @Nullable Supplier>> headersSupplier) { + super(httpClient, fullMethodName, headersSupplier); + this.compressor = compressor; + this.bufferAllocator = httpClient.executionContext().bufferAllocator(); + LOGGER.debug("Created ServiceTalkGrpcSender: fullMethodName={}, compression={}", + fullMethodName, compressor != null); + } + + @Override + public void send(MessageWriter messageWriter, Consumer onResponse, Consumer onError) { + LOGGER.debug("Sending gRPC request via ServiceTalk"); + doSend(messageWriter, onResponse, onError); + } + + @Override + protected void prepareMessage(MessageWriter messageWriter, HttpRequest request) throws IOException { + applyHeaders(request); + int messageSize = messageWriter.getContentLength(); + // We need to write the gRPC message header, which depends on whether it's compressed or not. + Buffer buffer; + if (compressor == null) { + buffer = bufferAllocator.newBuffer(messageSize + HEADER_LENGTH); + buffer.writeByte(UNCOMPRESSED_FLAG); + buffer.writeInt(messageSize); + buffer.writeBytes(readMessage(messageWriter, null)); + } else { + request.setHeader(GRPC_MESSAGE_ENCODING, compressor.getEncoding()); + byte[] compressedBody = readMessage(messageWriter, compressor); + buffer = bufferAllocator.newBuffer(compressedBody.length + HEADER_LENGTH); + buffer.writeByte(COMPRESSED_FLAG); + buffer.writeInt(compressedBody.length); + buffer.writeBytes(compressedBody); + } + + request.payloadBody(buffer); + request.setHeader("te", "trailers"); + request.setHeader(CONTENT_TYPE, GrpcHeaderValues.APPLICATION_GRPC); + LOGGER.debug("Prepared gRPC request: method={}, path={}, contentType={}", + request.method(), request.requestTarget(), GrpcHeaderValues.APPLICATION_GRPC); + } + + @Override + protected GrpcResponse buildResponse(HttpResponseStatus status, HttpHeaders headers, byte[] responseBody) { + CharSequence grpcStatusSeq = headers.get("grpc-status"); + String grpcStatusStr = grpcStatusSeq != null ? grpcStatusSeq.toString() : "0"; // 0 = OK + CharSequence grpcMessageSeq = headers.get("grpc-message"); + String grpcMessage = grpcMessageSeq != null ? grpcMessageSeq.toString() : ""; + + int grpcStatusValue = Integer.parseInt(grpcStatusStr); + io.opentelemetry.sdk.common.export.GrpcStatusCode statusCode = + io.opentelemetry.sdk.common.export.GrpcStatusCode.fromValue(grpcStatusValue); + LOGGER.debug("Received gRPC response: grpcStatus={}, grpcMessage={}, responseSize={} bytes", + statusCode, grpcMessage, responseBody.length); + return new GrpcResponseImpl(statusCode, grpcMessage, responseBody); + } + + private static final class GrpcResponseImpl implements GrpcResponse { + private final io.opentelemetry.sdk.common.export.GrpcStatusCode statusCode; + private final String statusDescription; + private final byte[] responseMessage; + + GrpcResponseImpl(io.opentelemetry.sdk.common.export.GrpcStatusCode statusCode, + String statusDescription, byte[] responseMessage) { + this.statusCode = statusCode; + this.statusDescription = statusDescription; + this.responseMessage = responseMessage; + } + + @Override + public io.opentelemetry.sdk.common.export.GrpcStatusCode getStatusCode() { + return statusCode; + } + + @Override + public String getStatusDescription() { + return statusDescription; + } + + @Override + public byte[] getResponseMessage() { + return responseMessage; + } + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSenderProvider.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSenderProvider.java new file mode 100644 index 0000000000..fa633f00a8 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkGrpcSenderProvider.java @@ -0,0 +1,45 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.http.api.HttpClient; + +import io.opentelemetry.sdk.common.export.GrpcSender; +import io.opentelemetry.sdk.common.export.GrpcSenderConfig; +import io.opentelemetry.sdk.common.export.GrpcSenderProvider; + +public final class ServiceTalkGrpcSenderProvider implements GrpcSenderProvider { + + @Override + public GrpcSender createSender(GrpcSenderConfig config) { + // Build HttpClient using shared factory with gRPC-specific configuration + HttpClient httpClient = ServiceTalkHttpClientFactory.buildGrpcClient( + config.getEndpoint(), + config.getTimeout(), + config.getConnectTimeout(), + config.getSslContext(), + config.getRetryPolicy() + ); + + // Create and return ServiceTalkGrpcSender with configured client + return new ServiceTalkGrpcSender( + httpClient, + config.getCompressor(), + config.getFullMethodName(), + config.getHeadersSupplier() + ); + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpClientFactory.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpClientFactory.java new file mode 100644 index 0000000000..25ad2724eb --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpClientFactory.java @@ -0,0 +1,279 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.http.api.HttpClient; +import io.servicetalk.http.api.ProxyConfigBuilder; +import io.servicetalk.http.api.SingleAddressHttpClientBuilder; +import io.servicetalk.http.netty.HttpClients; +import io.servicetalk.http.netty.HttpProtocolConfigs; +import io.servicetalk.http.netty.RetryingHttpRequesterFilter; +import io.servicetalk.http.utils.TimeoutHttpRequesterFilter; +import io.servicetalk.transport.api.ClientSslConfig; +import io.servicetalk.transport.api.ClientSslConfigBuilder; +import io.servicetalk.transport.api.HostAndPort; +import io.servicetalk.transport.api.ServiceTalkSocketOptions; + +import io.opentelemetry.sdk.common.export.ProxyOptions; +import io.opentelemetry.sdk.common.export.RetryPolicy; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.URI; +import java.time.Duration; +import java.util.function.Predicate; +import javax.annotation.Nullable; +import javax.net.ssl.SSLContext; + +final class ServiceTalkHttpClientFactory { + + private ServiceTalkHttpClientFactory() { + } + + static HttpClient buildGrpcClient( + URI endpoint, + @Nullable Duration timeout, + @Nullable Duration connectTimeout, + @Nullable SSLContext sslContext, + @Nullable RetryPolicy retryPolicy) { + + SingleAddressHttpClientBuilder builder = createBaseBuilder(endpoint); + + // Protocol: Always HTTP/2 for gRPC + builder.protocols(HttpProtocolConfigs.h2Default()); + applyConnectTimeout(builder, connectTimeout); + applySslConfiguration(builder, endpoint, sslContext); + applyRequestTimeout(builder, timeout); + applyRetryPolicy(builder, retryPolicy); + + return builder.build(); + } + + static HttpClient buildHttpClient( + URI endpoint, + @Nullable Duration timeout, + @Nullable Duration connectTimeout, + @Nullable SSLContext sslContext, + @Nullable ProxyOptions proxyOptions, + @Nullable RetryPolicy retryPolicy) { + + SingleAddressHttpClientBuilder builder = createBaseBuilder(endpoint); + + // Protocol: HTTP/1.1 by default (most compatible) + builder.protocols(HttpProtocolConfigs.h1Default()); + applyConnectTimeout(builder, connectTimeout); + applySslConfiguration(builder, endpoint, sslContext); + applyProxyConfiguration(builder, endpoint, proxyOptions); + applyRequestTimeout(builder, timeout); + applyRetryPolicy(builder, retryPolicy); + + return builder.build(); + } + + private static SingleAddressHttpClientBuilder createBaseBuilder(URI endpoint) { + String host = endpoint.getHost(); + if (host == null || host.isEmpty()) { + throw new IllegalArgumentException("Endpoint must have a host: " + endpoint); + } + + int port = endpoint.getPort(); + if (port <= 0) { + // Use default ports based on scheme + String scheme = endpoint.getScheme(); + if ("https".equalsIgnoreCase(scheme) || "grpcs".equalsIgnoreCase(scheme)) { + port = 443; + } else { + // Default to port 80 for http, grpc, or unknown schemes + port = 80; + } + } + + return HttpClients.forSingleAddress(host, port); + } + + private static void applyConnectTimeout( + SingleAddressHttpClientBuilder builder, + @Nullable Duration connectTimeout) { + + if (connectTimeout == null) { + return; + } + + // Connect timeout is configured via socket options + builder.socketOption(ServiceTalkSocketOptions.CONNECT_TIMEOUT, (int) connectTimeout.toMillis()); + } + + private static void applyRequestTimeout( + SingleAddressHttpClientBuilder builder, + @Nullable Duration timeout) { + + if (timeout == null) { + return; + } + + // Request timeout is applied via a filter + // The second parameter (true) means the timeout applies to the full request/response cycle + builder.appendClientFilter(new TimeoutHttpRequesterFilter(timeout, true)); + } + + private static void applySslConfiguration( + SingleAddressHttpClientBuilder builder, + URI endpoint, + @Nullable SSLContext sslContext) { + + String scheme = endpoint.getScheme(); + if (!"https".equalsIgnoreCase(scheme) && !"grpcs".equalsIgnoreCase(scheme)) { + return; // No SSL needed + } + + try { + ClientSslConfigBuilder sslConfigBuilder; + + if (sslContext != null) { + // Priority 1: Use explicit SSLContext if configured. This specifies both trust and keys + sslConfigBuilder = new ClientSslConfigBuilder(sslContext); + } else { + // Priority 2: Use system default trust managers + sslConfigBuilder = new ClientSslConfigBuilder(); + } + + // Configure SNI hostname for proper TLS handshake + String host = endpoint.getHost(); + if (host != null && !host.isEmpty()) { + sslConfigBuilder.sniHostname(host); + sslConfigBuilder.peerHost(host); + + int port = endpoint.getPort(); + if (port > 0) { + sslConfigBuilder.peerPort(port); + } + } + + ClientSslConfig sslConfig = sslConfigBuilder.build(); + builder.sslConfig(sslConfig); + } catch (Exception e) { + throw new IllegalStateException("Failed to configure SSL", e); + } + } + + private static void applyProxyConfiguration( + SingleAddressHttpClientBuilder builder, + URI endpoint, + @Nullable ProxyOptions proxyOptions) { + if (proxyOptions == null) { + return; + } + + try { + // Extract proxy address from ProxyOptions using ProxySelector + java.net.ProxySelector proxySelector = proxyOptions.getProxySelector(); + if (proxySelector == null) { + return; // No proxy configured + } + java.util.List proxies = proxySelector.select(endpoint); + + if (proxies == null || proxies.isEmpty() || proxies.get(0).type() == java.net.Proxy.Type.DIRECT) { + return; // No proxy or direct connection + } + + java.net.Proxy proxy = proxies.get(0); + if (proxy.type() != java.net.Proxy.Type.HTTP) { + // ServiceTalk only supports HTTP proxies for CONNECT tunneling + return; + } + + java.net.SocketAddress proxySocketAddress = proxy.address(); + if (!(proxySocketAddress instanceof InetSocketAddress)) { + throw new IllegalArgumentException("Proxy address must be InetSocketAddress, got: " + + proxySocketAddress.getClass().getName()); + } + + HostAndPort proxyAddress = HostAndPort.of((InetSocketAddress) proxySocketAddress); + ProxyConfigBuilder proxyConfigBuilder = new ProxyConfigBuilder<>(proxyAddress); + + // Check for proxy authentication credentials + // OpenTelemetry may provide username/password for Basic auth + // Note: The exact API for credentials in ProxyOptions may vary + // Common patterns include getUsername()/getPassword() methods + + // Attempt to get authentication info if ProxyOptions exposes it + // For OpenTelemetry 1.59.0, check if there are methods like: + // - proxyOptions.getUsername() / proxyOptions.getPassword() + // - proxyOptions.getAuthenticator() returning java.net.Authenticator + // Since these may not exist or vary, we'll use reflection or conditional checks + + // For now, create basic proxy config without auth + // Users can extend this with custom authentication if needed via: + // - System properties: http.proxyUser, http.proxyPassword + // - java.net.Authenticator.setDefault() + builder.proxyConfig(proxyConfigBuilder.build()); + } catch (Exception e) { + throw new IllegalStateException("Failed to configure proxy", e); + } + } + + private static void applyRetryPolicy( + SingleAddressHttpClientBuilder builder, + @Nullable RetryPolicy retryPolicy) { + + if (retryPolicy == null || retryPolicy.getMaxAttempts() == 1) { + return; + } + + // OpenTelemetry's maxAttempts includes the initial attempt + // ServiceTalk's maxTotalRetries is the number of retries (not including initial) + final int maxRetries = Math.max(0, retryPolicy.getMaxAttempts() - 1); + + // Get backoff parameters from OpenTelemetry RetryPolicy + final Duration initialBackoff = retryPolicy.getInitialBackoff(); + final Duration maxBackoff = retryPolicy.getMaxBackoff(); + final double backoffMultiplier = retryPolicy.getBackoffMultiplier(); + final Predicate retryExceptionPredicate = retryPolicy.getRetryExceptionPredicate(); + + // Build ServiceTalk retry filter + RetryingHttpRequesterFilter.Builder retryBuilder = new RetryingHttpRequesterFilter.Builder() + .maxTotalRetries(maxRetries); + + // Configure retry for exceptions that match OpenTelemetry's predicate + if (retryExceptionPredicate != null) { + retryBuilder.retryRetryableExceptions((metadata, throwable) -> { + // Check if the exception is an IOException and matches the predicate + if (throwable instanceof IOException && retryExceptionPredicate.test((IOException) throwable)) { + // Use exponential backoff if multiplier > 1.0, otherwise use initial delay as constant + if (backoffMultiplier > 1.0) { + // Calculate jitter as a fraction of the initial backoff + // OpenTelemetry uses full jitter, so we'll use full jitter here too + return RetryingHttpRequesterFilter.BackOffPolicy.ofExponentialBackoffFullJitter( + initialBackoff, + maxBackoff, + maxRetries + ); + } else { + // Constant backoff with full jitter + return RetryingHttpRequesterFilter.BackOffPolicy.ofConstantBackoffFullJitter( + initialBackoff, + maxRetries + ); + } + } + // Don't retry this exception + return RetryingHttpRequesterFilter.BackOffPolicy.ofNoRetries(); + }); + } + + builder.appendClientFilter(retryBuilder.build()); + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSender.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSender.java new file mode 100644 index 0000000000..3672985945 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSender.java @@ -0,0 +1,118 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.buffer.api.Buffer; +import io.servicetalk.buffer.api.BufferAllocator; +import io.servicetalk.http.api.HttpClient; +import io.servicetalk.http.api.HttpHeaders; +import io.servicetalk.http.api.HttpRequest; +import io.servicetalk.http.api.HttpResponseStatus; + +import io.opentelemetry.sdk.common.export.Compressor; +import io.opentelemetry.sdk.common.export.HttpResponse; +import io.opentelemetry.sdk.common.export.HttpSender; +import io.opentelemetry.sdk.common.export.MessageWriter; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.function.Supplier; +import javax.annotation.Nullable; + +import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_ENCODING; +import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_TYPE; + +final class ServiceTalkHttpSender extends AbstractServiceTalkSender implements HttpSender { + + private static final Logger LOGGER = LoggerFactory.getLogger(ServiceTalkHttpSender.class); + + private final String contentType; + @Nullable + private final Compressor compressor; + private final BufferAllocator bufferAllocator; + + ServiceTalkHttpSender(HttpClient httpClient, + @Nullable Compressor compressor, + String contentType, + String requestTarget, + @Nullable Supplier>> headersSupplier) { + super(httpClient, requestTarget, headersSupplier); + this.contentType = contentType; + this.compressor = compressor; + this.bufferAllocator = httpClient.executionContext().bufferAllocator(); + LOGGER.debug("Created ServiceTalkHttpSender: requestTarget={}, contentType={}, compression={}", + requestTarget, contentType, compressor != null); + } + + @Override + public void send(MessageWriter messageWriter, Consumer onResponse, Consumer onError) { + LOGGER.debug("Sending HTTP request via ServiceTalk"); + doSend(messageWriter, onResponse, onError); + } + + @Override + protected Object buildResponse(HttpResponseStatus status, HttpHeaders headers, byte[] responseBody) { + return new HttpResponseImpl(status.code(), status.reasonPhrase(), responseBody); + } + + @Override + protected void prepareMessage(MessageWriter messageWriter, HttpRequest request) throws IOException { + Buffer payload = bufferAllocator.wrap(readMessage(messageWriter, compressor)); + request.payloadBody(payload); + LOGGER.debug("Prepared message payload: size={} bytes", payload.readableBytes()); + + String compressorEncoding = compressor == null ? null : compressor.getEncoding(); + if (compressorEncoding != null) { + request.setHeader(CONTENT_ENCODING, compressorEncoding); + LOGGER.debug("Applied compression: encoding={}", compressorEncoding); + } + applyHeaders(request); + request.setHeader(CONTENT_TYPE, contentType); + LOGGER.debug("Prepared HTTP request: method={}, target={}, contentType={}", + request.method(), request.requestTarget(), contentType); + } + + private static final class HttpResponseImpl implements HttpResponse { + private final int statusCode; + private final String statusMessage; + private final byte[] responseBody; + + HttpResponseImpl(int statusCode, String statusMessage, byte[] responseBody) { + this.statusCode = statusCode; + this.statusMessage = statusMessage; + this.responseBody = responseBody; + } + + @Override + public int getStatusCode() { + return statusCode; + } + + @Override + public String getStatusMessage() { + return statusMessage; + } + + @Override + public byte[] getResponseBody() { + return responseBody; + } + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSenderProvider.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSenderProvider.java new file mode 100644 index 0000000000..f029ff18cb --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/ServiceTalkHttpSenderProvider.java @@ -0,0 +1,56 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.http.api.HttpClient; + +import io.opentelemetry.sdk.common.export.HttpSender; +import io.opentelemetry.sdk.common.export.HttpSenderConfig; +import io.opentelemetry.sdk.common.export.HttpSenderProvider; + +import java.net.URI; + +public final class ServiceTalkHttpSenderProvider implements HttpSenderProvider { + + @Override + public HttpSender createSender(HttpSenderConfig config) { + // Build HttpClient using shared factory with HTTP-specific configuration + HttpClient httpClient = ServiceTalkHttpClientFactory.buildHttpClient( + config.getEndpoint(), + config.getTimeout(), + config.getConnectTimeout(), + config.getSslContext(), + config.getProxyOptions(), + config.getRetryPolicy() + ); + + // Extract request target from endpoint path + URI endpoint = config.getEndpoint(); + String requestTarget = endpoint.getPath(); + if (requestTarget == null || requestTarget.isEmpty()) { + requestTarget = "/"; + } + + // Create and return ServiceTalkHttpSender with configured client + return new ServiceTalkHttpSender( + httpClient, + config.getCompressor(), + config.getContentType(), + requestTarget, + config.getHeadersSupplier() + ); + } +} diff --git a/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/package-info.java b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/package-info.java new file mode 100755 index 0000000000..d5f26c63f6 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/java/io/servicetalk/opentelemetry/client/package-info.java @@ -0,0 +1,20 @@ +/* + * Copyright © 2022 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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. + */ + +@ElementsAreNonnullByDefault +package io.servicetalk.opentelemetry.client; + +import io.servicetalk.annotations.ElementsAreNonnullByDefault; diff --git a/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.GrpcSenderProvider b/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.GrpcSenderProvider new file mode 100644 index 0000000000..838a5494c0 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.GrpcSenderProvider @@ -0,0 +1 @@ +io.servicetalk.opentelemetry.client.ServiceTalkGrpcSenderProvider diff --git a/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.HttpSenderProvider b/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.HttpSenderProvider new file mode 100644 index 0000000000..856e2c5ee5 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/main/resources/META-INF/services/io.opentelemetry.sdk.common.export.HttpSenderProvider @@ -0,0 +1 @@ +io.servicetalk.opentelemetry.client.ServiceTalkHttpSenderProvider diff --git a/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/GrpcSenderIntegrationTest.java b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/GrpcSenderIntegrationTest.java new file mode 100644 index 0000000000..89b988af43 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/GrpcSenderIntegrationTest.java @@ -0,0 +1,270 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.test.resources.DefaultTestCerts; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter; +import io.opentelemetry.proto.trace.v1.ResourceSpans; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.export.GrpcSenderProvider; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.ArrayList; +import java.util.List; +import java.util.ServiceLoader; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; +import javax.net.ssl.X509TrustManager; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.isA; +import static org.hamcrest.Matchers.notNullValue; + +final class GrpcSenderIntegrationTest { + + @Nullable + private MockOtlpCollector collector; + @Nullable + private OpenTelemetrySdk openTelemetry; + + @AfterEach + void tearDown() throws Exception { + if (openTelemetry != null) { + openTelemetry.close(); + } + if (collector != null) { + collector.close(); + } + } + + @Test + void onlyServiceTalkGrpcProvidersOnClassPath() { + // Verify ServiceTalk GrpcSender is available + ServiceLoader loader = ServiceLoader.load(GrpcSenderProvider.class); + List results = new ArrayList<>(); + for (GrpcSenderProvider spi : loader) { + results.add(spi); + } + assertThat(results, hasSize(1)); + assertThat(results.get(0), isA(ServiceTalkGrpcSenderProvider.class)); + } + + @Test + void cleartextGrpcSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .build(); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("http://localhost:" + collector.getPort()) + .build(); + + exportSpan(spanExporter, "cleartext-grpc-test-span"); + + assertGrpcSpanReceived(); + } + + @Test + void tlsGrpcSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .securityMode(MockOtlpCollector.SecurityMode.TLS) + .build(); + + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + SSLContext sslContext = TestUtils.createTlsSslContext(tmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + collector.getPort()) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "tls-grpc-test-span"); + + assertGrpcSpanReceived(); + } + + @Test + void mutualTlsGrpcSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .securityMode(MockOtlpCollector.SecurityMode.MUTUAL_TLS) + .build(); + + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + KeyManagerFactory kmf = TestUtils.createKeyManagerFactory( + DefaultTestCerts::loadClientPem, DefaultTestCerts::loadClientKey); + SSLContext sslContext = TestUtils.createMutualTlsSslContext(tmf, kmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + collector.getPort()) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "mtls-grpc-test-span"); + + assertGrpcSpanReceived(); + } + + @Test + void mutualTlsRejectsClientWithoutCertificate() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .securityMode(MockOtlpCollector.SecurityMode.MUTUAL_TLS) + .build(); + + // Client configured with server trust only — no client certificate presented. + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + SSLContext sslContext = TestUtils.createTlsSslContext(tmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + collector.getPort()) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "rejected-span"); + + assertThat("mTLS server should reject a client that presents no certificate", + collector.getRequestCount(), is(0)); + } + + @Test + void grpcMessageIsProperlyFramedAndDecoded() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .build(); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("http://localhost:" + collector.getPort()) + .build(); + + String spanName = "test-span-for-framing-and-decoding"; + exportSpan(spanExporter, spanName); + + // Wait for request to be received + boolean received = TestUtils.waitFor(() -> collector.getRequestCount() > 0, 5_000, 100); + assertThat("Collector should have received at least one request", received, is(true)); + + MockOtlpCollector.ReceivedRequest request = collector.getReceivedRequests().get(0); + + // Verify it's a gRPC request + assertThat("Request should be identified as gRPC", request.isGrpc(), is(true)); + + // Verify gRPC frame metadata is present and valid + MockOtlpCollector.GrpcFrameMetadata metadata = request.getGrpcMetadata(); + assertThat("gRPC metadata should be present", metadata, notNullValue()); + assertThat("Message should not be compressed by default", metadata.isCompressed(), is(false)); + assertThat("Message length should be positive", metadata.getMessageLength(), greaterThan(0)); + assertThat("Decoded message should be present", request.getGrpcMessage(), notNullValue()); + + assertThat("Export request should have resource spans", + request.getGrpcMessage().getResourceSpansCount(), greaterThan(0)); + + ResourceSpans resourceSpans = request.getGrpcMessage().getResourceSpans(0); + assertThat("Resource spans should have scope spans", + resourceSpans.getScopeSpansCount(), greaterThan(0)); + + io.opentelemetry.proto.trace.v1.ScopeSpans scopeSpans = resourceSpans.getScopeSpans(0); + assertThat("Scope spans should have spans", + scopeSpans.getSpansCount(), greaterThan(0)); + + io.opentelemetry.proto.trace.v1.Span span = scopeSpans.getSpans(0); + assertThat("Span name should match what was sent", span.getName(), equalTo(spanName)); + } + + @Test + void grpcPathIsCorrect() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .build(); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("http://localhost:" + collector.getPort()) + .build(); + + exportSpan(spanExporter, "path-test-span"); + + boolean received = TestUtils.waitFor(() -> collector.getRequestCount() > 0, 5_000, 100); + assertThat("Collector should have received at least one request", received, is(true)); + + MockOtlpCollector.ReceivedRequest request = collector.getReceivedRequests().get(0); + // The gRPC path for OTLP traces should be the full method name + assertThat("Request path should be the gRPC method path", + request.getPath(), containsString("TraceService")); + } + + @Test + void grpcHeadersAreCorrect() throws Exception { + collector = new MockOtlpCollector.Builder() + .protocolMode(MockOtlpCollector.ProtocolMode.GRPC) + .build(); + + OtlpGrpcSpanExporter spanExporter = OtlpGrpcSpanExporter.builder() + .setEndpoint("http://localhost:" + collector.getPort()) + .build(); + + exportSpan(spanExporter, "headers-test-span"); + + boolean received = TestUtils.waitFor(() -> collector.getRequestCount() > 0, 5_000, 100); + assertThat("Collector should have received at least one request", received, is(true)); + + MockOtlpCollector.ReceivedRequest request = collector.getReceivedRequests().get(0); + + // Verify gRPC-specific headers + assertThat("Content-Type should be gRPC", + request.getHeaders().get("content-type").toString(), + containsString("application/grpc")); + assertThat("TE header should be present", + request.getHeaders().contains("te"), is(true)); + } + + private void exportSpan(SpanExporter spanExporter, String spanName) { + openTelemetry = TestUtils.createOpenTelemetry(spanExporter); + Tracer tracer = openTelemetry.getTracer("test"); + Span span = TestUtils.createTestSpan(tracer, spanName); + span.end(); + openTelemetry.getSdkTracerProvider() + .forceFlush() + .join(10, TimeUnit.SECONDS); + } + + private void assertGrpcSpanReceived() throws InterruptedException { + boolean received = TestUtils.waitFor(() -> collector.getRequestCount() > 0, 5_000, 100); + + assertThat("Collector should have received at least one request", received, is(true)); + + MockOtlpCollector.ReceivedRequest request = collector.getReceivedRequests().get(0); + assertThat("Request should be gRPC", request.isGrpc(), is(true)); + assertThat("Request should have gRPC metadata", request.getGrpcMetadata(), notNullValue()); + assertThat("Request payload should not be empty", request.getGrpcMessage(), is(notNullValue())); + } +} diff --git a/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/HttpSenderIntegrationTest.java b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/HttpSenderIntegrationTest.java new file mode 100644 index 0000000000..cb35a899ca --- /dev/null +++ b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/HttpSenderIntegrationTest.java @@ -0,0 +1,179 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.test.resources.DefaultTestCerts; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.common.export.HttpSenderProvider; +import io.opentelemetry.sdk.trace.export.SpanExporter; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.ServiceLoader; +import java.util.concurrent.TimeUnit; +import javax.annotation.Nullable; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; +import javax.net.ssl.X509TrustManager; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.hasSize; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.isA; + +final class HttpSenderIntegrationTest { + + @Nullable + private MockOtlpCollector collector; + @Nullable + private OpenTelemetrySdk openTelemetry; + + @AfterEach + void tearDown() throws Exception { + if (openTelemetry != null) { + openTelemetry.close(); + } + if (collector != null) { + collector.close(); + } + } + + @Test + void onlyServiceTalkProvidersOnClassPath() { + // It's tough to verify the underlying transport, but if the ServiceTalk sender is the only one + // available, then that has got to be the right one. + ServiceLoader loader = ServiceLoader.load(HttpSenderProvider.class); + List results = new ArrayList<>(); + for (HttpSenderProvider spi : loader) { + results.add(spi); + } + assertThat(results, hasSize(1)); + assertThat(results.get(0), isA(ServiceTalkHttpSenderProvider.class)); + } + + @Test + void cleartextHttpSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder().build(); + + OtlpHttpSpanExporter spanExporter = OtlpHttpSpanExporter.builder() + .setEndpoint("http://localhost:" + collector.getPort() + "/v1/traces") + .setTimeout(Duration.ofSeconds(10)) + .build(); + + exportSpan(spanExporter, "cleartext-test-span"); + + assertSpanReceived(); + } + + @Test + void tlsHttpSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder() + .securityMode(MockOtlpCollector.SecurityMode.TLS) + .build(); + + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + SSLContext sslContext = TestUtils.createTlsSslContext(tmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpHttpSpanExporter spanExporter = OtlpHttpSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + + collector.getPort() + "/v1/traces") + .setTimeout(Duration.ofSeconds(10)) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "tls-test-span"); + + assertSpanReceived(); + } + + @Test + void mutualTlsHttpSendsSpansToCollector() throws Exception { + collector = new MockOtlpCollector.Builder() + .securityMode(MockOtlpCollector.SecurityMode.MUTUAL_TLS) + .build(); + + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + KeyManagerFactory kmf = TestUtils.createKeyManagerFactory( + DefaultTestCerts::loadClientPem, DefaultTestCerts::loadClientKey); + SSLContext sslContext = TestUtils.createMutualTlsSslContext(tmf, kmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpHttpSpanExporter spanExporter = OtlpHttpSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + + collector.getPort() + "/v1/traces") + .setTimeout(Duration.ofSeconds(10)) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "mtls-test-span"); + + assertSpanReceived(); + } + + @Test + void mutualTlsRejectsClientWithoutCertificate() throws Exception { + collector = new MockOtlpCollector.Builder() + .securityMode(MockOtlpCollector.SecurityMode.MUTUAL_TLS) + .build(); + + // Client configured with server trust only — no client certificate presented. + TrustManagerFactory tmf = TestUtils.createTrustManagerFactory(DefaultTestCerts::loadServerCAPem); + SSLContext sslContext = TestUtils.createTlsSslContext(tmf); + X509TrustManager trustManager = TestUtils.extractTrustManager(tmf); + + OtlpHttpSpanExporter spanExporter = OtlpHttpSpanExporter.builder() + .setEndpoint("https://" + DefaultTestCerts.serverPemHostname() + ":" + + collector.getPort() + "/v1/traces") + .setTimeout(Duration.ofSeconds(5)) + .setSslContext(sslContext, trustManager) + .build(); + + exportSpan(spanExporter, "rejected-span"); + + assertThat("mTLS server should reject a client that presents no certificate", + collector.getRequestCount(), is(0)); + } + + private void exportSpan(SpanExporter spanExporter, String spanName) throws Exception { + openTelemetry = TestUtils.createOpenTelemetry(spanExporter); + Tracer tracer = openTelemetry.getTracer("test"); + Span span = TestUtils.createTestSpan(tracer, spanName); + span.end(); + openTelemetry.getSdkTracerProvider() + .forceFlush() + .join(10, TimeUnit.SECONDS); + } + + private void assertSpanReceived() throws InterruptedException { + boolean received = TestUtils.waitFor(() -> collector.getRequestCount() > 0, 5_000, 100); + + assertThat("Collector should have received at least one request", received, is(true)); + + MockOtlpCollector.ReceivedRequest request = collector.getReceivedRequests().get(0); + assertThat("Request path should be /v1/traces", request.getPath(), is("/v1/traces")); + assertThat("Request payload should not be empty", request.getPayload().length, is(greaterThan(0))); + } +} diff --git a/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/MockOtlpCollector.java b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/MockOtlpCollector.java new file mode 100644 index 0000000000..87f5e29d5c --- /dev/null +++ b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/MockOtlpCollector.java @@ -0,0 +1,386 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.servicetalk.buffer.api.Buffer; +import io.servicetalk.buffer.api.BufferAllocator; +import io.servicetalk.concurrent.api.Single; +import io.servicetalk.grpc.api.GrpcHeaderValues; +import io.servicetalk.grpc.api.GrpcSerializationProvider; +import io.servicetalk.grpc.protobuf.ProtoBufSerializationProviderBuilder; +import io.servicetalk.http.api.DefaultHttpHeadersFactory; +import io.servicetalk.http.api.HttpDeserializer; +import io.servicetalk.http.api.HttpHeaders; +import io.servicetalk.http.api.HttpRequest; +import io.servicetalk.http.api.HttpRequestMetaData; +import io.servicetalk.http.api.HttpResponse; +import io.servicetalk.http.api.HttpResponseFactory; +import io.servicetalk.http.api.HttpResponseStatus; +import io.servicetalk.http.api.HttpServerBuilder; +import io.servicetalk.http.api.HttpService; +import io.servicetalk.http.api.HttpServiceContext; +import io.servicetalk.http.netty.HttpProtocolConfigs; +import io.servicetalk.http.netty.HttpServers; +import io.servicetalk.test.resources.DefaultTestCerts; +import io.servicetalk.transport.api.HostAndPort; +import io.servicetalk.transport.api.ServerContext; +import io.servicetalk.transport.api.ServerSslConfigBuilder; +import io.servicetalk.transport.api.SslClientAuthMode; + +import com.google.protobuf.MessageLite; +import com.google.protobuf.Parser; +import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest; + +import java.net.InetSocketAddress; +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.atomic.AtomicInteger; +import javax.annotation.Nullable; + +import static io.servicetalk.concurrent.api.Single.succeeded; +import static io.servicetalk.encoding.api.Identity.identity; +import static io.servicetalk.http.api.HttpHeaderNames.CONTENT_TYPE; +import static io.servicetalk.http.api.HttpResponseStatus.SERVICE_UNAVAILABLE; + +final class MockOtlpCollector implements AutoCloseable { + + enum SecurityMode { + CLEARTEXT, + TLS, + MUTUAL_TLS + } + + enum ProtocolMode { + HTTP, + GRPC + } + + @Nullable + private ServerContext serverContext; + private final Queue receivedRequests = new ConcurrentLinkedQueue<>(); + private final AtomicInteger requestCount = new AtomicInteger(0); + private final AtomicInteger failFirstNRequests; + private final HttpResponseStatus failureStatus; + @Nullable + private final Duration responseDelay; + + private MockOtlpCollector(int failFirstNRequests, + HttpResponseStatus failureStatus, + @Nullable Duration responseDelay) { + this.failFirstNRequests = new AtomicInteger(failFirstNRequests); + this.failureStatus = failureStatus; + this.responseDelay = responseDelay; + } + + int getPort() { + InetSocketAddress address = (InetSocketAddress) serverContext.listenAddress(); + return address.getPort(); + } + + String getHostname() { + InetSocketAddress address = (InetSocketAddress) serverContext.listenAddress(); + return address.getHostString(); + } + + HostAndPort getAddress() { + return HostAndPort.of(getHostname(), getPort()); + } + + int getRequestCount() { + return requestCount.get(); + } + + List getReceivedRequests() { + return new ArrayList<>(receivedRequests); + } + + void reset() { + receivedRequests.clear(); + requestCount.set(0); + failFirstNRequests.set(0); + } + + @Override + public void close() throws Exception { + serverContext.close(); + } + + static final class ReceivedRequest { + private final String path; + private final HttpHeaders headers; + private final byte[] payload; + private final boolean isGrpc; + @Nullable + private final GrpcFrameMetadata grpcMetadata; + @Nullable + private final ExportTraceServiceRequest grpcMessage; + + ReceivedRequest(String path, HttpHeaders headers, byte[] payload) { + this.path = path; + this.headers = headers; + this.payload = payload; + this.isGrpc = false; + this.grpcMetadata = null; + this.grpcMessage = null; + } + + ReceivedRequest(String path, HttpHeaders headers, byte[] payload, + boolean isGrpc, GrpcFrameMetadata grpcMetadata, + ExportTraceServiceRequest grpcMessage) { + this.path = path; + this.headers = headers; + this.payload = payload; + this.isGrpc = isGrpc; + this.grpcMetadata = grpcMetadata; + this.grpcMessage = grpcMessage; + } + + public String getPath() { + return path; + } + + public HttpHeaders getHeaders() { + return headers; + } + + public byte[] getPayload() { + return payload; + } + + public boolean isGrpc() { + return isGrpc; + } + + @Nullable + public ExportTraceServiceRequest getGrpcMessage() { + return grpcMessage; + } + + @Nullable + public GrpcFrameMetadata getGrpcMetadata() { + return grpcMetadata; + } + } + + static final class GrpcFrameMetadata { + private final boolean compressed; + private final int messageLength; + + GrpcFrameMetadata(boolean compressed, int messageLength) { + this.compressed = compressed; + this.messageLength = messageLength; + } + + public boolean isCompressed() { + return compressed; + } + + public int getMessageLength() { + return messageLength; + } + } + + static final class Builder { + private SecurityMode securityMode = SecurityMode.CLEARTEXT; + private ProtocolMode protocolMode = ProtocolMode.HTTP; + private int failFirstNRequests; + private HttpResponseStatus failureStatus = SERVICE_UNAVAILABLE; + @Nullable + private Duration responseDelay; + + Builder securityMode(SecurityMode securityMode) { + this.securityMode = securityMode; + return this; + } + + Builder protocolMode(ProtocolMode protocolMode) { + this.protocolMode = protocolMode; + return this; + } + + Builder failFirstNRequests(int count, HttpResponseStatus status) { + this.failFirstNRequests = count; + this.failureStatus = status; + return this; + } + + Builder responseDelay(Duration delay) { + this.responseDelay = delay; + return this; + } + + MockOtlpCollector build() throws Exception { + HttpServerBuilder serverBuilder = HttpServers.forAddress(new InetSocketAddress("localhost", 0)); + + // Configure for HTTP/1.1 or HTTP/2 (for gRPC) + if (protocolMode == ProtocolMode.GRPC) { + serverBuilder.protocols(HttpProtocolConfigs.h2Default()); + } else { + serverBuilder.protocols(HttpProtocolConfigs.h1Default()); + } + + // Configure SSL/TLS + if (securityMode != SecurityMode.CLEARTEXT) { + ServerSslConfigBuilder sslConfigBuilder = new ServerSslConfigBuilder( + DefaultTestCerts::loadServerPem, + DefaultTestCerts::loadServerKey); + + if (securityMode == SecurityMode.MUTUAL_TLS) { + // Require client certificates + sslConfigBuilder.trustManager(DefaultTestCerts::loadClientCAPem) + .clientAuthMode(SslClientAuthMode.REQUIRE); + } + + serverBuilder.sslConfig(sslConfigBuilder.build()); + } + + // Create the collector first so the handler and the returned instance are the same object. + MockOtlpCollector collector = new MockOtlpCollector(failFirstNRequests, failureStatus, responseDelay); + + ServerContext serverContext = serverBuilder.listenAndAwait( + new OtlpRequestHandler(collector, protocolMode)); + + collector.serverContext = serverContext; + return collector; + } + } + + private static final class OtlpRequestHandler implements HttpService { + private static final int GRPC_HEADER_LENGTH = 5; + private final MockOtlpCollector collector; + private final ProtocolMode protocolMode; + + OtlpRequestHandler(MockOtlpCollector collector, ProtocolMode protocolMode) { + this.collector = collector; + this.protocolMode = protocolMode; + } + + @Override + public Single handle(HttpServiceContext ctx, + HttpRequest request, + HttpResponseFactory factory) { + // Check if we should fail this request + if (collector.failFirstNRequests.getAndDecrement() > 0) { + return succeeded(factory.newResponse(collector.failureStatus)); + } + + // Add delay if configured + if (collector.responseDelay != null) { + // Delay the response + return ctx.executionContext().executor().timer(collector.responseDelay) + .concat(handleRequest(ctx, request, factory)); + } else { + return handleRequest(ctx, request, factory); + } + } + + private Single handleRequest(HttpServiceContext ctx, + HttpRequest request, + HttpResponseFactory factory) { + Buffer payload = request.payloadBody(); + byte[] payloadBytes = new byte[payload.readableBytes()]; + payload.getBytes(payload.readerIndex(), payloadBytes); + + ReceivedRequest captured; + if (protocolMode == ProtocolMode.GRPC) { + // Parse gRPC frame to extract the message + captured = parseGrpcRequest(ctx, request, payload, payloadBytes); + } else { + captured = new ReceivedRequest( + request.path(), + request.headers(), + payloadBytes + ); + } + + collector.receivedRequests.add(captured); + collector.requestCount.incrementAndGet(); + + // Create success response + HttpResponse response = factory.ok(); + if (protocolMode == ProtocolMode.GRPC) { + // gRPC trailers-only response (no body) with grpc-status in headers + response.headers().set(CONTENT_TYPE, GrpcHeaderValues.APPLICATION_GRPC); + response.headers().set("grpc-status", "0"); // 0 = OK (sent in headers for trailers-only) + } else { + response.headers().set(CONTENT_TYPE, "application/x-protobuf"); + } + return succeeded(response); + } + + private ReceivedRequest parseGrpcRequest(HttpServiceContext ctx, + HttpRequestMetaData request, + Buffer payload, + byte[] payloadBytes) { + try { + if (payload.readableBytes() < GRPC_HEADER_LENGTH) { + throw new IllegalStateException("Not enough data for grpc message: " + payload.readableBytes()); + } + // First do a lightweight frame header validation to extract metadata + byte compressionFlag = payload.getByte(payload.readerIndex()); + if (compressionFlag != 0x0 && compressionFlag != 0x1) { + throw new IllegalArgumentException("Compression flag must be 0 or 1 but was: " + compressionFlag); + } + boolean compressed = compressionFlag == 0x1; + int frameLength = payload.getInt(payload.readerIndex() + 1); + if (frameLength < 0) { + throw new IllegalArgumentException("Message-Length invalid: " + frameLength); + } else if (frameLength != payload.readableBytes() - 5) { + throw new IllegalStateException("Invalid message size. Expected " + frameLength); + } + + GrpcFrameMetadata metadata = new GrpcFrameMetadata(compressed, frameLength); + + // Now validate using ServiceTalk's actual gRPC deserialization + // This will throw if the framing doesn't match what ServiceTalk expects + io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest validated = + validateAndDecodeGrpcFrame( + payloadBytes, + io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceRequest.parser(), + ctx.executionContext().bufferAllocator()); + return new ReceivedRequest( + request.path(), + request.headers(), + payloadBytes, + true, + metadata, + validated + ); + } catch (Exception e) { + // If ServiceTalk's gRPC deserialization fails, the framing is invalid + throw new RuntimeException("Failed to validate gRPC frame using ServiceTalk's deserialization - " + + "framing is invalid", e); + } + } + + private static T validateAndDecodeGrpcFrame( + byte[] payload, + Parser parser, + BufferAllocator allocator) throws Exception { + @SuppressWarnings("unchecked") + Class messageClass = (Class) parser.parseFrom(new byte[0]).getClass(); + GrpcSerializationProvider grpcSerializationProvider = new ProtoBufSerializationProviderBuilder() + .registerMessageType(messageClass, parser) + .build(); + HttpDeserializer httpDeserializer = grpcSerializationProvider.deserializerFor(identity(), messageClass); + Buffer buffer = allocator.wrap(payload); + return httpDeserializer.deserialize(DefaultHttpHeadersFactory.INSTANCE.newHeaders(), buffer); + } + } +} diff --git a/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/TestUtils.java b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/TestUtils.java new file mode 100644 index 0000000000..5c0ade56b8 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/test/java/io/servicetalk/opentelemetry/client/TestUtils.java @@ -0,0 +1,265 @@ +/* + * Copyright © 2026 Apple Inc. and the ServiceTalk project authors + * + * Licensed 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 io.servicetalk.opentelemetry.client; + +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.export.SpanExporter; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.security.KeyFactory; +import java.security.KeyStore; +import java.security.PrivateKey; +import java.security.SecureRandom; +import java.security.cert.Certificate; +import java.security.cert.CertificateFactory; +import java.security.cert.X509Certificate; +import java.security.spec.PKCS8EncodedKeySpec; +import java.util.Base64; +import java.util.Collection; +import java.util.concurrent.TimeUnit; +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManager; +import javax.net.ssl.TrustManagerFactory; +import javax.net.ssl.X509TrustManager; + +/** + * Utility methods for ServiceTalk OpenTelemetry integration tests. + */ +final class TestUtils { + + private TestUtils() { + // Utility class + } + + /** + * Create an OpenTelemetry SDK instance configured with the given span exporter. + * + * @param spanExporter the span exporter to use + * @return configured OpenTelemetrySdk instance + */ + static OpenTelemetrySdk createOpenTelemetry(SpanExporter spanExporter) { + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(io.opentelemetry.sdk.trace.export.BatchSpanProcessor.builder(spanExporter) + .setScheduleDelay(100, TimeUnit.MILLISECONDS) + .build()) + .build(); + + return OpenTelemetrySdk.builder() + .setTracerProvider(tracerProvider) + .build(); + } + + /** + * Create a test span with the given tracer. + * + * @param tracer the tracer to use + * @param spanName the span name + * @return the created span + */ + static Span createTestSpan(Tracer tracer, String spanName) { + return tracer.spanBuilder(spanName) + .setSpanKind(SpanKind.CLIENT) + .setAttribute("test.key", "test.value") + .setAttribute("service.name", "test-service") + .startSpan(); + } + + /** + * End a span and wait for it to be exported. + * + * @param span the span to end + * @throws InterruptedException if interrupted while waiting + */ + static void endSpanAndWait(Span span) throws InterruptedException { + span.end(); + // Give some time for the batch processor to export + Thread.sleep(500); + } + + /** + * Create a TrustManagerFactory from PEM-encoded certificate. + * + * @param pemSupplier supplier of PEM-encoded certificate + * @return configured TrustManagerFactory + * @throws Exception if trust manager creation fails + */ + static TrustManagerFactory createTrustManagerFactory( + java.util.function.Supplier pemSupplier) throws Exception { + try (InputStream pemStream = pemSupplier.get()) { + CertificateFactory certFactory = CertificateFactory.getInstance("X.509"); + Collection certificates = certFactory.generateCertificates(pemStream); + + KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType()); + keyStore.load(null, null); + + int i = 0; + for (Certificate cert : certificates) { + keyStore.setCertificateEntry("cert-" + i++, cert); + } + + TrustManagerFactory trustManagerFactory = + TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm()); + trustManagerFactory.init(keyStore); + + return trustManagerFactory; + } + } + + /** + * Create a KeyManagerFactory from PEM-encoded certificate and private key. + * + * @param certPemSupplier supplier of PEM-encoded certificate + * @param keyPemSupplier supplier of PEM-encoded private key + * @return configured KeyManagerFactory + * @throws Exception if key manager creation fails + */ + static KeyManagerFactory createKeyManagerFactory( + java.util.function.Supplier certPemSupplier, + java.util.function.Supplier keyPemSupplier) throws Exception { + + // Load certificate + X509Certificate cert; + try (InputStream certStream = certPemSupplier.get()) { + CertificateFactory certFactory = CertificateFactory.getInstance("X.509"); + cert = (X509Certificate) certFactory.generateCertificate(certStream); + } + + // Load private key + PrivateKey privateKey = loadPrivateKey(keyPemSupplier.get()); + + // Create KeyStore + KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType()); + keyStore.load(null, null); + keyStore.setKeyEntry("client-key", privateKey, new char[0], new Certificate[]{cert}); + + // Create KeyManagerFactory + KeyManagerFactory keyManagerFactory = + KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm()); + keyManagerFactory.init(keyStore, new char[0]); + + return keyManagerFactory; + } + + /** + * Load a private key from PEM-encoded input stream. + * + * @param pemStream PEM-encoded private key stream + * @return the private key + * @throws Exception if key loading fails + */ + private static PrivateKey loadPrivateKey(InputStream pemStream) throws Exception { + // Read all bytes from stream + byte[] pemBytes = readAllBytes(pemStream); + String pemContent = new String(pemBytes, StandardCharsets.US_ASCII); + // Remove PEM headers and footers + pemContent = pemContent + .replaceAll("-----BEGIN PRIVATE KEY-----", "") + .replaceAll("-----END PRIVATE KEY-----", "") + .replaceAll("-----BEGIN RSA PRIVATE KEY-----", "") + .replaceAll("-----END RSA PRIVATE KEY-----", "") + .replaceAll("\\s", ""); + + byte[] keyBytes = Base64.getDecoder().decode(pemContent); + PKCS8EncodedKeySpec keySpec = new PKCS8EncodedKeySpec(keyBytes); + KeyFactory keyFactory = KeyFactory.getInstance("RSA"); + return keyFactory.generatePrivate(keySpec); + } + + /** + * Read all bytes from an InputStream. + * + * @param inputStream the input stream + * @return all bytes from the stream + * @throws IOException if reading fails + */ + static byte[] readAllBytes(InputStream inputStream) throws IOException { + byte[] buffer = new byte[8192]; + int bytesRead; + java.io.ByteArrayOutputStream output = new java.io.ByteArrayOutputStream(); + while ((bytesRead = inputStream.read(buffer)) != -1) { + output.write(buffer, 0, bytesRead); + } + return output.toByteArray(); + } + + /** + * Create an SSLContext for client authentication (mutual TLS). + * + * @param trustManagerFactory trust manager for server certificate validation + * @param keyManagerFactory key manager for client certificate + * @return configured SSLContext + * @throws Exception if SSL context creation fails + */ + static SSLContext createMutualTlsSslContext( + TrustManagerFactory trustManagerFactory, + KeyManagerFactory keyManagerFactory) throws Exception { + SSLContext sslContext = SSLContext.getInstance("TLS"); + sslContext.init( + keyManagerFactory.getKeyManagers(), + trustManagerFactory.getTrustManagers(), + new SecureRandom()); + return sslContext; + } + + /** + * Create an SSLContext for server authentication only (TLS). + * + * @param trustManagerFactory trust manager for server certificate validation + * @return configured SSLContext + * @throws Exception if SSL context creation fails + */ + static SSLContext createTlsSslContext(TrustManagerFactory trustManagerFactory) throws Exception { + SSLContext sslContext = SSLContext.getInstance("TLS"); + sslContext.init(null, trustManagerFactory.getTrustManagers(), new SecureRandom()); + return sslContext; + } + + /** + * Extract the X509TrustManager from a TrustManagerFactory. + * + * @param trustManagerFactory the trust manager factory + * @return the X509TrustManager + * @throws IllegalStateException if no X509TrustManager is found + */ + static X509TrustManager extractTrustManager(TrustManagerFactory trustManagerFactory) { + for (TrustManager tm : trustManagerFactory.getTrustManagers()) { + if (tm instanceof X509TrustManager) { + return (X509TrustManager) tm; + } + } + throw new IllegalStateException("No X509TrustManager found in TrustManagerFactory"); + } + + static boolean waitFor(java.util.function.BooleanSupplier condition, + long timeoutMs, + long checkIntervalMs) throws InterruptedException { + long endTime = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < endTime) { + if (condition.getAsBoolean()) { + return true; + } + Thread.sleep(checkIntervalMs); + } + return false; + } +} diff --git a/servicetalk-opentelemetry-client/src/test/resources/log4j2-test.xml b/servicetalk-opentelemetry-client/src/test/resources/log4j2-test.xml new file mode 100644 index 0000000000..f9b9efe872 --- /dev/null +++ b/servicetalk-opentelemetry-client/src/test/resources/log4j2-test.xml @@ -0,0 +1,43 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/settings.gradle b/settings.gradle index 5013a4cd0c..2bfd8ef4c5 100755 --- a/settings.gradle +++ b/settings.gradle @@ -70,6 +70,7 @@ include "servicetalk-annotations", "servicetalk-examples:http:jaxrs", "servicetalk-examples:http:metadata", "servicetalk-examples:http:opentelemetry-tracing", + "servicetalk-examples:http:opentelemetry-jaeger", "servicetalk-examples:http:opentracing", "servicetalk-examples:http:observer", "servicetalk-examples:http:retry", @@ -118,6 +119,7 @@ include "servicetalk-annotations", "servicetalk-oio-api", "servicetalk-oio-api-internal", "servicetalk-opentelemetry-asynccontext", + "servicetalk-opentelemetry-client", "servicetalk-opentelemetry-http", "servicetalk-opentracing-inmemory", "servicetalk-opentracing-inmemory-api", @@ -160,6 +162,7 @@ project(":servicetalk-examples:http:jaxrs").name = "servicetalk-examples-http-ja project(":servicetalk-examples:http:loadbalancer").name = "servicetalk-examples-http-loadbalancer" project(":servicetalk-examples:http:metadata").name = "servicetalk-examples-http-metadata" project(":servicetalk-examples:http:opentelemetry-tracing").name = "servicetalk-examples-http-opentelemetry-tracing" +project(":servicetalk-examples:http:opentelemetry-jaeger").name = "servicetalk-examples-http-opentelemetry-jaeger" project(":servicetalk-examples:http:opentracing").name = "servicetalk-examples-http-opentracing" project(":servicetalk-examples:http:observer").name = "servicetalk-examples-http-observer" project(":servicetalk-examples:http:retry").name = "servicetalk-examples-http-retry"