-
Notifications
You must be signed in to change notification settings - Fork 2.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #23826 from michalszynkiewicz/collect-grpc-stats-f…
…or-stork gRPC - gather call statistics
- Loading branch information
Showing
30 changed files
with
659 additions
and
63 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
.../grpc/runtime/src/main/java/io/quarkus/grpc/runtime/config/GrpcClientBuildTimeConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package io.quarkus.grpc.runtime.config; | ||
|
|
||
| import io.quarkus.runtime.annotations.ConfigItem; | ||
| import io.quarkus.runtime.annotations.ConfigPhase; | ||
| import io.quarkus.runtime.annotations.ConfigRoot; | ||
|
|
||
| @ConfigRoot(phase = ConfigPhase.BUILD_TIME) | ||
| public class GrpcClientBuildTimeConfig { | ||
|
|
||
| /** | ||
| * If set to true, and a Stork load balancer is used, connections with all available service instances will be | ||
| * requested proactively. This means better load balancing at the cost of having multiple active connections. | ||
| */ | ||
| @ConfigItem(defaultValue = "true") | ||
| public boolean storkProactiveConnections; | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
99 changes: 99 additions & 0 deletions
99
...pc/runtime/src/main/java/io/quarkus/grpc/runtime/stork/StorkMeasuringGrpcInterceptor.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package io.quarkus.grpc.runtime.stork; | ||
|
|
||
| import java.util.concurrent.atomic.AtomicReference; | ||
|
|
||
| import javax.enterprise.context.ApplicationScoped; | ||
| import javax.enterprise.inject.spi.Prioritized; | ||
|
|
||
| import io.grpc.CallOptions; | ||
| import io.grpc.Channel; | ||
| import io.grpc.ClientCall; | ||
| import io.grpc.ClientInterceptor; | ||
| import io.grpc.Context; | ||
| import io.grpc.ForwardingClientCall; | ||
| import io.grpc.ForwardingClientCallListener; | ||
| import io.grpc.Metadata; | ||
| import io.grpc.MethodDescriptor; | ||
| import io.grpc.Status; | ||
| import io.smallrye.stork.api.ServiceInstance; | ||
|
|
||
| @ApplicationScoped | ||
| public class StorkMeasuringGrpcInterceptor implements ClientInterceptor, Prioritized { | ||
|
|
||
| public static final Context.Key<AtomicReference<ServiceInstance>> STORK_SERVICE_INSTANCE = Context | ||
| .key("stork.service-instance"); | ||
| public static final Context.Key<Boolean> STORK_MEASURE_TIME = Context.key("stork.measure-time"); | ||
|
|
||
| @Override | ||
| public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(MethodDescriptor<ReqT, RespT> method, CallOptions callOptions, | ||
| Channel next) { | ||
| return new StorkMeasuringCall<>(next.newCall(method, callOptions), method.getType()); | ||
| } | ||
|
|
||
| @Override | ||
| public int getPriority() { | ||
| return Integer.MAX_VALUE - 100; | ||
| } | ||
|
|
||
| private static class StorkMeasuringCall<ReqT, RespT> extends ForwardingClientCall.SimpleForwardingClientCall<ReqT, RespT> { | ||
| ServiceInstance serviceInstance; | ||
| final boolean recordTime; | ||
|
|
||
| protected StorkMeasuringCall(ClientCall<ReqT, RespT> delegate, | ||
| MethodDescriptor.MethodType type) { | ||
| super(delegate); | ||
| this.recordTime = type == MethodDescriptor.MethodType.UNARY; | ||
| } | ||
|
|
||
| @Override | ||
| public void start(final ClientCall.Listener<RespT> responseListener, final Metadata metadata) { | ||
| Context context = Context.current().withValues(STORK_SERVICE_INSTANCE, new AtomicReference<>(), | ||
| STORK_MEASURE_TIME, recordTime); | ||
| Context oldContext = context.attach(); | ||
| try { | ||
| super.start(new StorkMeasuringCallListener<>(responseListener, this), metadata); | ||
| serviceInstance = STORK_SERVICE_INSTANCE.get().get(); | ||
| } finally { | ||
| context.detach(oldContext); | ||
| } | ||
| } | ||
|
|
||
| void recordReply() { | ||
| if (serviceInstance != null && recordTime) { | ||
| serviceInstance.recordReply(); | ||
| } | ||
| } | ||
|
|
||
| void recordEnd(Throwable error) { | ||
| if (serviceInstance != null) { | ||
| serviceInstance.recordEnd(error); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| private static class StorkMeasuringCallListener<RespT> | ||
| extends ForwardingClientCallListener.SimpleForwardingClientCallListener<RespT> { | ||
| final StorkMeasuringCall<?, ?> collector; | ||
|
|
||
| public StorkMeasuringCallListener(ClientCall.Listener<RespT> responseListener, StorkMeasuringCall<?, ?> collector) { | ||
| super(responseListener); | ||
| this.collector = collector; | ||
| } | ||
|
|
||
| @Override | ||
| public void onMessage(RespT message) { | ||
| collector.recordReply(); | ||
| super.onMessage(message); | ||
| } | ||
|
|
||
| @Override | ||
| public void onClose(Status status, Metadata trailers) { | ||
| Exception error = null; | ||
| if (!status.isOk()) { | ||
| error = status.asException(trailers); | ||
| } | ||
| collector.recordEnd(error); | ||
| super.onClose(status, trailers); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 0 additions & 19 deletions
19
.../runtime/src/main/java/io/quarkus/rest/client/reactive/runtime/SmallRyeStorkRecorder.java
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.