Skip to content

Commit 299e536

Browse files
AgraVatorAgraVator
committed
interop: add signal handling and try-with-resources cleanups
1 parent cf6b688 commit 299e536

2 files changed

Lines changed: 112 additions & 94 deletions

File tree

‎interop-testing/src/main/java/io/grpc/testing/integration/TestServiceClient.java‎

Lines changed: 26 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -54,15 +54,12 @@
5454
import io.grpc.internal.JsonParser;
5555
import io.grpc.netty.InsecureFromHttp1ChannelCredentials;
5656
import io.grpc.netty.InternalNettyChannelBuilder;
57-
import io.grpc.opentelemetry.GrpcOpenTelemetry;
58-
import io.grpc.opentelemetry.GrpcTraceBinContextPropagator;
59-
import io.grpc.opentelemetry.InternalGrpcOpenTelemetry;
60-
import io.opentelemetry.context.propagation.TextMapPropagator;
61-
import io.opentelemetry.sdk.OpenTelemetrySdk;
62-
import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
6357
import io.grpc.netty.NettyChannelBuilder;
6458
import io.grpc.okhttp.InternalOkHttpChannelBuilder;
6559
import io.grpc.okhttp.OkHttpChannelBuilder;
60+
import io.grpc.opentelemetry.GrpcOpenTelemetry;
61+
import io.grpc.opentelemetry.GrpcTraceBinContextPropagator;
62+
import io.grpc.opentelemetry.InternalGrpcOpenTelemetry;
6663
import io.grpc.stub.ClientCalls;
6764
import io.grpc.stub.MetadataUtils;
6865
import io.grpc.stub.StreamObserver;
@@ -74,6 +71,9 @@
7471
import io.grpc.testing.integration.Messages.StreamingOutputCallRequest;
7572
import io.grpc.testing.integration.Messages.StreamingOutputCallResponse;
7673
import io.grpc.testing.integration.Messages.TestOrcaReport;
74+
import io.opentelemetry.context.propagation.TextMapPropagator;
75+
import io.opentelemetry.sdk.OpenTelemetrySdk;
76+
import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdk;
7777
import java.io.File;
7878
import java.io.FileInputStream;
7979
import java.io.InputStream;
@@ -448,28 +448,36 @@ private void runTest(TestCases testCase) throws Exception {
448448

449449
case SERVICE_ACCOUNT_CREDS: {
450450
String jsonKey = Files.asCharSource(new File(serviceAccountKeyFile), UTF_8).read();
451-
FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
452-
tester.serviceAccountCreds(jsonKey, credentialsStream, oauthScope);
451+
try (FileInputStream credentialsStream =
452+
new FileInputStream(new File(serviceAccountKeyFile))) {
453+
tester.serviceAccountCreds(jsonKey, credentialsStream, oauthScope);
454+
}
453455
break;
454456
}
455457

456458
case JWT_TOKEN_CREDS: {
457-
FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
458-
tester.jwtTokenCreds(credentialsStream);
459+
try (FileInputStream credentialsStream =
460+
new FileInputStream(new File(serviceAccountKeyFile))) {
461+
tester.jwtTokenCreds(credentialsStream);
462+
}
459463
break;
460464
}
461465

462466
case OAUTH2_AUTH_TOKEN: {
463467
String jsonKey = Files.asCharSource(new File(serviceAccountKeyFile), UTF_8).read();
464-
FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
465-
tester.oauth2AuthToken(jsonKey, credentialsStream, oauthScope);
468+
try (FileInputStream credentialsStream =
469+
new FileInputStream(new File(serviceAccountKeyFile))) {
470+
tester.oauth2AuthToken(jsonKey, credentialsStream, oauthScope);
471+
}
466472
break;
467473
}
468474

469475
case PER_RPC_CREDS: {
470476
String jsonKey = Files.asCharSource(new File(serviceAccountKeyFile), UTF_8).read();
471-
FileInputStream credentialsStream = new FileInputStream(new File(serviceAccountKeyFile));
472-
tester.perRpcCreds(jsonKey, credentialsStream, oauthScope);
477+
try (FileInputStream credentialsStream =
478+
new FileInputStream(new File(serviceAccountKeyFile))) {
479+
tester.perRpcCreds(jsonKey, credentialsStream, oauthScope);
480+
}
473481
break;
474482
}
475483

@@ -701,7 +709,8 @@ protected ManagedChannelBuilder<?> createChannelBuilder() {
701709
if (serverPort == 0) {
702710
nettyBuilder = NettyChannelBuilder.forTarget(serverHost, channelCredentials);
703711
} else {
704-
nettyBuilder = NettyChannelBuilder.forAddress(serverHost, serverPort, channelCredentials);
712+
nettyBuilder =
713+
NettyChannelBuilder.forAddress(serverHost, serverPort, channelCredentials);
705714
}
706715
nettyBuilder.flowControlWindow(AbstractInteropTest.TEST_FLOW_CONTROL_WINDOW);
707716
if (serverHostOverride != null) {
@@ -819,8 +828,8 @@ public void cacheableUnary() {
819828
}
820829

821830
/** Sends a large unary rpc with service account credentials. */
822-
public void serviceAccountCreds(String jsonKey, InputStream credentialsStream, String authScope)
823-
throws Exception {
831+
public void serviceAccountCreds(
832+
String jsonKey, InputStream credentialsStream, String authScope) throws Exception {
824833
// cast to ServiceAccountCredentials to double-check the right type of object was created.
825834
GoogleCredentials credentials =
826835
ServiceAccountCredentials.class.cast(GoogleCredentials.fromStream(credentialsStream));

‎interop-testing/src/main/java/io/grpc/testing/integration/TestServiceServer.java‎

Lines changed: 86 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -166,94 +166,103 @@ void parseArgs(String[] args) {
166166
@SuppressWarnings("AddressSelection")
167167
@VisibleForTesting
168168
void start() throws Exception {
169-
if (enableOpentelemetry) {
170-
AutoConfiguredOpenTelemetrySdk autoSdk = AutoConfiguredOpenTelemetrySdk.builder()
171-
.addPropagatorCustomizer(
172-
(previous, config) ->
173-
TextMapPropagator.composite(
174-
previous, GrpcTraceBinContextPropagator.defaultInstance()))
175-
.build();
176-
this.openTelemetrySdk = autoSdk.getOpenTelemetrySdk();
177-
GrpcOpenTelemetry.Builder grpcOpentelemetryBuilder = GrpcOpenTelemetry.newBuilder()
178-
.sdk(openTelemetrySdk);
179-
InternalGrpcOpenTelemetry.enableTracing(grpcOpentelemetryBuilder, true);
180-
GrpcOpenTelemetry grpcOpenTelemetry = grpcOpentelemetryBuilder.build();
181-
grpcOpenTelemetry.registerGlobal();
182-
}
183-
executor = Executors.newSingleThreadScheduledExecutor();
184-
ServerCredentials serverCreds;
185-
if (useAlts) {
186-
if (localHandshakerPort > -1) {
187-
serverCreds = AltsServerCredentials.newBuilder()
188-
.enableUntrustedAltsForTesting()
189-
.setHandshakerAddressForTesting("localhost:" + localHandshakerPort).build();
169+
try {
170+
if (enableOpentelemetry) {
171+
AutoConfiguredOpenTelemetrySdk autoSdk = AutoConfiguredOpenTelemetrySdk.builder()
172+
.addPropagatorCustomizer(
173+
(previous, config) ->
174+
TextMapPropagator.composite(
175+
previous, GrpcTraceBinContextPropagator.defaultInstance()))
176+
.build();
177+
this.openTelemetrySdk = autoSdk.getOpenTelemetrySdk();
178+
GrpcOpenTelemetry.Builder grpcOpentelemetryBuilder = GrpcOpenTelemetry.newBuilder()
179+
.sdk(openTelemetrySdk);
180+
InternalGrpcOpenTelemetry.enableTracing(grpcOpentelemetryBuilder, true);
181+
GrpcOpenTelemetry grpcOpenTelemetry = grpcOpentelemetryBuilder.build();
182+
grpcOpenTelemetry.registerGlobal();
183+
}
184+
executor = Executors.newSingleThreadScheduledExecutor();
185+
ServerCredentials serverCreds;
186+
if (useAlts) {
187+
if (localHandshakerPort > -1) {
188+
serverCreds = AltsServerCredentials.newBuilder()
189+
.enableUntrustedAltsForTesting()
190+
.setHandshakerAddressForTesting("localhost:" + localHandshakerPort).build();
191+
} else {
192+
serverCreds = AltsServerCredentials.create();
193+
}
194+
} else if (useTls) {
195+
serverCreds = TlsServerCredentials.create(
196+
TlsTesting.loadCert("server1.pem"), TlsTesting.loadCert("server1.key"));
190197
} else {
191-
serverCreds = AltsServerCredentials.create();
198+
serverCreds = InsecureServerCredentials.create();
192199
}
193-
} else if (useTls) {
194-
serverCreds = TlsServerCredentials.create(
195-
TlsTesting.loadCert("server1.pem"), TlsTesting.loadCert("server1.key"));
196-
} else {
197-
serverCreds = InsecureServerCredentials.create();
198-
}
199-
MetricRecorder metricRecorder = MetricRecorder.newInstance();
200-
BindableService orcaOobService =
201-
OrcaServiceImpl.createService(executor, metricRecorder, 1, TimeUnit.SECONDS);
200+
MetricRecorder metricRecorder = MetricRecorder.newInstance();
201+
BindableService orcaOobService =
202+
OrcaServiceImpl.createService(executor, metricRecorder, 1, TimeUnit.SECONDS);
202203

203-
// Create ServerBuilder with appropriate addresses
204-
// - IPV4_IPV6: bind to wildcard which covers all addresses on all interfaces of both families
205-
// - IPV4: bind to v4 address for local hostname + v4 localhost
206-
// - IPV6: bind to all v6 addresses for local hostname + v6 localhost
207-
ServerBuilder<?> serverBuilder;
208-
switch (addressType) {
209-
case IPV4_IPV6:
210-
serverBuilder = Grpc.newServerBuilderForPort(port, serverCreds);
211-
break;
212-
case IPV4:
213-
SocketAddress v4Address = Util.getV4Address(port);
214-
InetSocketAddress localV4Address = new InetSocketAddress("127.0.0.1", port);
215-
serverBuilder =
216-
NettyServerBuilder.forAddress(localV4Address, serverCreds);
217-
if (v4Address != null && !v4Address.equals(localV4Address)) {
218-
((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
219-
}
220-
if (mcsLimit != -1) {
221-
((NettyServerBuilder) serverBuilder).maxConcurrentCallsPerConnection(mcsLimit);
222-
}
223-
break;
224-
case IPV6:
225-
List<SocketAddress> v6Addresses = Util.getV6Addresses(port);
226-
InetSocketAddress localV6Address = new InetSocketAddress("::1", port);
227-
serverBuilder =
228-
NettyServerBuilder.forAddress(localV6Address, serverCreds);
229-
for (SocketAddress address : v6Addresses) {
230-
if (!address.equals(localV6Address)) {
231-
((NettyServerBuilder) serverBuilder).addListenAddress(address);
204+
// Create ServerBuilder with appropriate addresses
205+
// - IPV4_IPV6: bind to wildcard which covers all addresses on all interfaces of both families
206+
// - IPV4: bind to v4 address for local hostname + v4 localhost
207+
// - IPV6: bind to all v6 addresses for local hostname + v6 localhost
208+
ServerBuilder<?> serverBuilder;
209+
switch (addressType) {
210+
case IPV4_IPV6:
211+
serverBuilder = Grpc.newServerBuilderForPort(port, serverCreds);
212+
break;
213+
case IPV4:
214+
SocketAddress v4Address = Util.getV4Address(port);
215+
InetSocketAddress localV4Address = new InetSocketAddress("127.0.0.1", port);
216+
serverBuilder =
217+
NettyServerBuilder.forAddress(localV4Address, serverCreds);
218+
if (v4Address != null && !v4Address.equals(localV4Address)) {
219+
((NettyServerBuilder) serverBuilder).addListenAddress(v4Address);
232220
}
233-
}
234-
break;
235-
default:
236-
throw new AssertionError("Unknown address type: " + addressType);
221+
if (mcsLimit != -1) {
222+
((NettyServerBuilder) serverBuilder).maxConcurrentCallsPerConnection(mcsLimit);
223+
}
224+
break;
225+
case IPV6:
226+
List<SocketAddress> v6Addresses = Util.getV6Addresses(port);
227+
InetSocketAddress localV6Address = new InetSocketAddress("::1", port);
228+
serverBuilder =
229+
NettyServerBuilder.forAddress(localV6Address, serverCreds);
230+
for (SocketAddress address : v6Addresses) {
231+
if (!address.equals(localV6Address)) {
232+
((NettyServerBuilder) serverBuilder).addListenAddress(address);
233+
}
234+
}
235+
break;
236+
default:
237+
throw new AssertionError("Unknown address type: " + addressType);
238+
}
239+
server = serverBuilder
240+
.maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
241+
.addService(
242+
ServerInterceptors.intercept(
243+
new TestServiceImpl(executor, metricRecorder), TestServiceImpl.interceptors()))
244+
.addService(orcaOobService)
245+
.intercept(OrcaMetricReportingServerInterceptor.create(metricRecorder))
246+
.build()
247+
.start();
248+
} catch (Throwable t) {
249+
stop();
250+
throw t;
237251
}
238-
server = serverBuilder
239-
.maxInboundMessageSize(AbstractInteropTest.MAX_MESSAGE_SIZE)
240-
.addService(
241-
ServerInterceptors.intercept(
242-
new TestServiceImpl(executor, metricRecorder), TestServiceImpl.interceptors()))
243-
.addService(orcaOobService)
244-
.intercept(OrcaMetricReportingServerInterceptor.create(metricRecorder))
245-
.build()
246-
.start();
247252
}
248253

249254
@VisibleForTesting
250255
void stop() throws Exception {
251256
try {
252-
server.shutdownNow();
253-
if (!server.awaitTermination(5, TimeUnit.SECONDS)) {
254-
System.err.println("Timed out waiting for server shutdown");
257+
if (server != null) {
258+
server.shutdownNow();
259+
if (!server.awaitTermination(5, TimeUnit.SECONDS)) {
260+
System.err.println("Timed out waiting for server shutdown");
261+
}
262+
}
263+
if (executor != null) {
264+
MoreExecutors.shutdownAndAwaitTermination(executor, 5, TimeUnit.SECONDS);
255265
}
256-
MoreExecutors.shutdownAndAwaitTermination(executor, 5, TimeUnit.SECONDS);
257266
} finally {
258267
if (openTelemetrySdk != null) {
259268
openTelemetrySdk.close();

0 commit comments

Comments
 (0)