Skip to content

Commit a3bf8af

Browse files
jack-bergjack-berg
authored
Break out GrpcSender, GrpcSenderProvider (#5617)
1 parent c5cdc80 commit a3bf8af

File tree

71 files changed

+1026
-753
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

71 files changed

+1026
-753
lines changed

‎README.md

Lines changed: 13 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -249,18 +249,19 @@ dependency as follows, replacing `{{artifact-id}}` with the value from the "Arti
249249

250250
### SDK Exporters
251251

252-
| Component | Description | Artifact ID | Version | Javadoc |
253-
|-----------------------------------------------------|--------------------------------------------------------------------|----------------------------------------|-------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
254-
| [OTLP Exporters](./exporters/otlp/all) | OTLP gRPC & HTTP exporters, including traces, metrics, and logs | `opentelemetry-exporter-otlp` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-otlp.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-otlp) |
255-
| [OTLP Common](./exporters/otlp/common) | Shared OTLP components (internal) | `opentelemetry-exporter-otlp-common` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-otlp-common.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-otlp-common) |
256-
| [Jaeger gRPC Exporter](./exporters/jaeger) | Jaeger gRPC trace exporter (deprecated [1]) | `opentelemetry-exporter-jaeger` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-jaeger.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-jaeger) |
257-
| [Jaeger Thrift Exporter](./exporters/jaeger-thrift) | Jaeger thrift trace exporter (deprecated [1]) | `opentelemetry-exporter-jaeger-thift` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-jaeger-thrift.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-jaeger-thrift) |
258-
| [Logging Exporter](./exporters/logging) | Logging exporters, including metrics, traces, and logs | `opentelemetry-exporter-logging` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-logging.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-logging) |
259-
| [Zipkin Exporter](./exporters/zipkin) | Zipkin trace exporter | `opentelemetry-exporter-zipkin` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-zipkin.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-zipkin) |
260-
| [Prometheus Exporter](./exporters/prometheus) | Prometheus metric exporter | `opentelemetry-exporter-prometheus` | <!--VERSION_UNSTABLE-->1.28.0-alpha<!--/VERSION_UNSTABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-prometheus.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-prometheus) |
261-
| [Exporter Common](./exporters/common) | Shared exporter components (internal) | `opentelemetry-exporter-common` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-common.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-common) |
262-
| [OkHttp Sender](./exporters/sender/okhttp) | OkHttp implementation of HttpSender (internal) | `opentelemetry-exporter-sender-okhttp` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-sender-okhttp.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-sender-okhttp) |
263-
| [JDK Sender](./exporters/sender/okhttp) | Java 11+ native HttpClient implementation of HttpSender (internal) | `opentelemetry-exporter-sender-jdk` | <!--VERSION_UNSTABLE-->1.28.0-alpha<!--/VERSION_UNSTABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-sender-jdk.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-sender-jdk) | |
252+
| Component | Description | Artifact ID | Version | Javadoc |
253+
|-----------------------------------------------------------------------|--------------------------------------------------------------------|------------------------------------------------------|-------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
254+
| [OTLP Exporters](./exporters/otlp/all) | OTLP gRPC & HTTP exporters, including traces, metrics, and logs | `opentelemetry-exporter-otlp` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-otlp.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-otlp) |
255+
| [OTLP Common](./exporters/otlp/common) | Shared OTLP components (internal) | `opentelemetry-exporter-otlp-common` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-otlp-common.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-otlp-common) |
256+
| [Jaeger gRPC Exporter](./exporters/jaeger) | Jaeger gRPC trace exporter (deprecated [1]) | `opentelemetry-exporter-jaeger` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-jaeger.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-jaeger) |
257+
| [Jaeger Thrift Exporter](./exporters/jaeger-thrift) | Jaeger thrift trace exporter (deprecated [1]) | `opentelemetry-exporter-jaeger-thift` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-jaeger-thrift.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-jaeger-thrift) |
258+
| [Logging Exporter](./exporters/logging) | Logging exporters, including metrics, traces, and logs | `opentelemetry-exporter-logging` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-logging.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-logging) |
259+
| [Zipkin Exporter](./exporters/zipkin) | Zipkin trace exporter | `opentelemetry-exporter-zipkin` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-zipkin.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-zipkin) |
260+
| [Prometheus Exporter](./exporters/prometheus) | Prometheus metric exporter | `opentelemetry-exporter-prometheus` | <!--VERSION_UNSTABLE-->1.28.0-alpha<!--/VERSION_UNSTABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-prometheus.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-prometheus) |
261+
| [Exporter Common](./exporters/common) | Shared exporter components (internal) | `opentelemetry-exporter-common` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-common.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-common) |
262+
| [OkHttp Sender](./exporters/sender/okhttp) | OkHttp implementation of HttpSender (internal) | `opentelemetry-exporter-sender-okhttp` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-sender-okhttp.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-sender-okhttp) |
263+
| [JDK Sender](./exporters/sender/okhttp) | Java 11+ native HttpClient implementation of HttpSender (internal) | `opentelemetry-exporter-sender-jdk` | <!--VERSION_UNSTABLE-->1.28.0-alpha<!--/VERSION_UNSTABLE--> | [![Javadocs](https://www.javadoc.io/badge/io.opentelemetry/opentelemetry-exporter-sender-jdk.svg)](https://www.javadoc.io/doc/io.opentelemetry/opentelemetry-exporter-sender-jdk) | |
264+
| [gRPC ManagedChannel Sender](./exporters/sender/grpc-managed-channel) | gRPC ManagedChannel implementation of GrpcSender (internal) | `opentelemetry-exporter-sender-grpc-managed-channel` | <!--VERSION_STABLE-->1.28.0<!--/VERSION_STABLE--> | TODO: add link after 1.29.0 | |
264265

265266
**[1]**: Jaeger now
266267
has [native support for OTLP](https://opentelemetry.io/blog/2022/jaeger-native-otlp/) and jaeger

‎docs/apidiffs/current_vs_latest/opentelemetry-exporter-sender-grpc-managed-channel.txt

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
Comparing source compatibility of against
2+
No changes.

‎exporters/common/build.gradle.kts

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,13 +21,11 @@ dependencies {
2121
// We include helpers shared by gRPC exporters but do not want to impose these
2222
// dependency on all of our consumers.
2323
compileOnly("com.fasterxml.jackson.core:jackson-core")
24-
compileOnly("com.squareup.okhttp3:okhttp")
2524
compileOnly("io.grpc:grpc-stub")
2625

2726
testImplementation(project(":sdk:common"))
2827

2928
testImplementation("com.google.protobuf:protobuf-java-util")
30-
testImplementation("com.squareup.okhttp3:okhttp")
3129
testImplementation("com.linecorp.armeria:armeria-junit5")
3230
testImplementation("org.skyscreamer:jsonassert")
3331
testImplementation("com.google.api.grpc:proto-google-common-protos")
@@ -53,6 +51,17 @@ testing {
5351
}
5452
}
5553
}
54+
suites {
55+
register<JvmTestSuite>("testGrpcSenderProvider") {
56+
dependencies {
57+
implementation(project(":exporters:sender:okhttp"))
58+
implementation(project(":exporters:sender:grpc-managed-channel"))
59+
60+
implementation("io.grpc:grpc-stub")
61+
implementation("io.grpc:grpc-netty")
62+
}
63+
}
64+
}
5665
}
5766

5867
tasks {

‎exporters/common/src/main/java/io/opentelemetry/exporter/internal/RetryUtil.java

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,50 @@
1+
/*
2+
* Copyright The OpenTelemetry Authors
3+
* SPDX-License-Identifier: Apache-2.0
4+
*/
5+
6+
package io.opentelemetry.exporter.internal;
7+
8+
import io.opentelemetry.exporter.internal.grpc.GrpcExporterUtil;
9+
import java.util.Arrays;
10+
import java.util.Collections;
11+
import java.util.HashSet;
12+
import java.util.Set;
13+
import java.util.stream.Collectors;
14+
15+
/**
16+
* This class is internal and is hence not for public use. Its APIs are unstable and can change at
17+
* any time.
18+
*/
19+
public final class RetryUtil {
20+
21+
private static final Set<String> RETRYABLE_GRPC_STATUS_CODES;
22+
private static final Set<Integer> RETRYABLE_HTTP_STATUS_CODES =
23+
Collections.unmodifiableSet(new HashSet<>(Arrays.asList(429, 502, 503, 504)));
24+
25+
static {
26+
Set<Integer> retryableGrpcStatusCodes = new HashSet<>();
27+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_CANCELLED);
28+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_DEADLINE_EXCEEDED);
29+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_RESOURCE_EXHAUSTED);
30+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_ABORTED);
31+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_OUT_OF_RANGE);
32+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_UNAVAILABLE);
33+
retryableGrpcStatusCodes.add(GrpcExporterUtil.GRPC_STATUS_DATA_LOSS);
34+
RETRYABLE_GRPC_STATUS_CODES =
35+
Collections.unmodifiableSet(
36+
retryableGrpcStatusCodes.stream().map(Object::toString).collect(Collectors.toSet()));
37+
}
38+
39+
private RetryUtil() {}
40+
41+
/** Returns the retryable gRPC status codes. */
42+
public static Set<String> retryableGrpcStatusCodes() {
43+
return RETRYABLE_GRPC_STATUS_CODES;
44+
}
45+
46+
/** Returns the retryable HTTP status codes. */
47+
public static Set<Integer> retryableHttpResponseCodes() {
48+
return RETRYABLE_HTTP_STATUS_CODES;
49+
}
50+
}

‎exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporter.java

Lines changed: 91 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -5,39 +5,110 @@
55

66
package io.opentelemetry.exporter.internal.grpc;
77

8-
import io.grpc.Channel;
8+
import static io.opentelemetry.exporter.internal.grpc.GrpcExporterUtil.GRPC_STATUS_UNAVAILABLE;
9+
import static io.opentelemetry.exporter.internal.grpc.GrpcExporterUtil.GRPC_STATUS_UNIMPLEMENTED;
10+
11+
import io.opentelemetry.api.metrics.MeterProvider;
12+
import io.opentelemetry.exporter.internal.ExporterMetrics;
913
import io.opentelemetry.exporter.internal.marshal.Marshaler;
1014
import io.opentelemetry.sdk.common.CompletableResultCode;
11-
import java.net.URI;
12-
import java.util.function.BiFunction;
15+
import io.opentelemetry.sdk.internal.ThrottlingLogger;
16+
import java.util.concurrent.atomic.AtomicBoolean;
1317
import java.util.function.Supplier;
18+
import java.util.logging.Level;
19+
import java.util.logging.Logger;
1420

1521
/**
16-
* An exporter of a {@link Marshaler} using the gRPC wire format.
22+
* Generic gRPC exporter.
1723
*
1824
* <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
1925
* at any time.
2026
*/
21-
public interface GrpcExporter<T extends Marshaler> {
27+
@SuppressWarnings("checkstyle:JavadocMethod")
28+
public final class GrpcExporter<T extends Marshaler> {
29+
30+
private static final Logger internalLogger = Logger.getLogger(GrpcExporter.class.getName());
31+
32+
private final ThrottlingLogger logger = new ThrottlingLogger(internalLogger);
2233

23-
/** Returns a new {@link GrpcExporterBuilder}. */
24-
static <T extends Marshaler> GrpcExporterBuilder<T> builder(
34+
// We only log unimplemented once since it's a configuration issue that won't be recovered.
35+
private final AtomicBoolean loggedUnimplemented = new AtomicBoolean();
36+
private final AtomicBoolean isShutdown = new AtomicBoolean();
37+
38+
private final String type;
39+
private final GrpcSender<T> grpcSender;
40+
private final ExporterMetrics exporterMetrics;
41+
42+
public GrpcExporter(
2543
String exporterName,
2644
String type,
27-
long defaultTimeoutSecs,
28-
URI defaultEndpoint,
29-
Supplier<BiFunction<Channel, String, MarshalerServiceStub<T, ?, ?>>> stubFactory,
30-
String grpcEndpointPath) {
31-
return new GrpcExporterBuilder<>(
32-
exporterName, type, defaultTimeoutSecs, defaultEndpoint, stubFactory, grpcEndpointPath);
45+
GrpcSender<T> grpcSender,
46+
Supplier<MeterProvider> meterProviderSupplier) {
47+
this.type = type;
48+
this.grpcSender = grpcSender;
49+
this.exporterMetrics = ExporterMetrics.createGrpc(exporterName, type, meterProviderSupplier);
3350
}
3451

35-
/**
36-
* Exports the {@code exportRequest} which is a request {@link Marshaler} for {@code numItems}
37-
* items.
38-
*/
39-
CompletableResultCode export(T exportRequest, int numItems);
52+
public CompletableResultCode export(T exportRequest, int numItems) {
53+
if (isShutdown.get()) {
54+
return CompletableResultCode.ofFailure();
55+
}
56+
57+
exporterMetrics.addSeen(numItems);
4058

41-
/** Shuts the exporter down. */
42-
CompletableResultCode shutdown();
59+
CompletableResultCode result = new CompletableResultCode();
60+
61+
grpcSender.send(
62+
exportRequest,
63+
() -> {
64+
exporterMetrics.addSuccess(numItems);
65+
result.succeed();
66+
},
67+
(response, throwable) -> {
68+
exporterMetrics.addFailed(numItems);
69+
switch (response.grpcStatusValue()) {
70+
case GRPC_STATUS_UNIMPLEMENTED:
71+
if (loggedUnimplemented.compareAndSet(false, true)) {
72+
GrpcExporterUtil.logUnimplemented(
73+
internalLogger, type, response.grpcStatusDescription());
74+
}
75+
break;
76+
case GRPC_STATUS_UNAVAILABLE:
77+
logger.log(
78+
Level.SEVERE,
79+
"Failed to export "
80+
+ type
81+
+ "s. Server is UNAVAILABLE. "
82+
+ "Make sure your collector is running and reachable from this network. "
83+
+ "Full error message:"
84+
+ response.grpcStatusDescription());
85+
break;
86+
default:
87+
logger.log(
88+
Level.WARNING,
89+
"Failed to export "
90+
+ type
91+
+ "s. Server responded with gRPC status code "
92+
+ response.grpcStatusValue()
93+
+ ". Error message: "
94+
+ response.grpcStatusDescription());
95+
break;
96+
}
97+
if (logger.isLoggable(Level.FINEST)) {
98+
logger.log(
99+
Level.FINEST, "Failed to export " + type + "s. Details follow: " + throwable);
100+
}
101+
result.fail();
102+
});
103+
104+
return result;
105+
}
106+
107+
public CompletableResultCode shutdown() {
108+
if (!isShutdown.compareAndSet(false, true)) {
109+
logger.log(Level.INFO, "Calling shutdown() multiple times.");
110+
return CompletableResultCode.ofSuccess();
111+
}
112+
return grpcSender.shutdown();
113+
}
43114
}

0 commit comments

Comments
 (0)