diff --git a/MODULE.bazel b/MODULE.bazel index 293df736527..7281827dc96 100644 --- a/MODULE.bazel +++ b/MODULE.bazel @@ -25,20 +25,20 @@ IO_GRPC_GRPC_JAVA_ARTIFACTS = [ "dev.cel:common:0.13.1", "com.squareup.okhttp:okhttp:2.7.5", "com.squareup.okio:okio:2.10.0", # 3.0+ needs swapping to -jvm; need work to avoid flag-day - "io.netty:netty-buffer:4.2.15.Final", - "io.netty:netty-codec-base:4.2.15.Final", - "io.netty:netty-codec-http2:4.2.15.Final", - "io.netty:netty-codec-http:4.2.15.Final", - "io.netty:netty-codec-socks:4.2.15.Final", - "io.netty:netty-common:4.2.15.Final", - "io.netty:netty-handler-proxy:4.2.15.Final", - "io.netty:netty-handler:4.2.15.Final", - "io.netty:netty-resolver:4.2.15.Final", - "io.netty:netty-tcnative-boringssl-static:2.0.75.Final", - "io.netty:netty-tcnative-classes:2.0.75.Final", - "io.netty:netty-transport-native-epoll:jar:linux-x86_64:4.2.15.Final", - "io.netty:netty-transport-native-unix-common:4.2.15.Final", - "io.netty:netty-transport:4.2.15.Final", + "io.netty:netty-buffer:4.2.16.Final", + "io.netty:netty-codec-base:4.2.16.Final", + "io.netty:netty-codec-http2:4.2.16.Final", + "io.netty:netty-codec-http:4.2.16.Final", + "io.netty:netty-codec-socks:4.2.16.Final", + "io.netty:netty-common:4.2.16.Final", + "io.netty:netty-handler-proxy:4.2.16.Final", + "io.netty:netty-handler:4.2.16.Final", + "io.netty:netty-resolver:4.2.16.Final", + "io.netty:netty-tcnative-boringssl-static:2.0.81.Final", + "io.netty:netty-tcnative-classes:2.0.81.Final", + "io.netty:netty-transport-native-epoll:jar:linux-x86_64:4.2.16.Final", + "io.netty:netty-transport-native-unix-common:4.2.16.Final", + "io.netty:netty-transport:4.2.16.Final", "io.opencensus:opencensus-api:0.31.0", "io.opencensus:opencensus-contrib-grpc-metrics:0.31.0", "io.perfmark:perfmark-api:0.27.0", diff --git a/SECURITY.md b/SECURITY.md index e710ceaabe1..e8f83c98394 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -400,7 +400,7 @@ grpc-netty version | netty-handler version | netty-tcnative-boringssl-static ver 1.75.x-1.76.x | 4.1.124.Final | 2.0.72.Final 1.77.x-1.78.x | 4.1.127.Final | 2.0.74.Final 1.79.x-1.80.x | 4.1.130.Final | 2.0.74.Final -1.81.x- | 4.1.132.Final | 2.0.75.Final +1.81.x- | 4.2.16.Final | 2.0.81.Final _(grpc-netty-shaded avoids issues with keeping these versions in sync.)_ diff --git a/api/src/context/java/io/grpc/Context.java b/api/src/context/java/io/grpc/Context.java index c19d2db9da3..b61a082c1b7 100644 --- a/api/src/context/java/io/grpc/Context.java +++ b/api/src/context/java/io/grpc/Context.java @@ -27,6 +27,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import java.util.logging.Level; import java.util.logging.Logger; @@ -554,6 +555,26 @@ public V call(Callable c) throws Exception { } } + /** + * Immediately gets a value from a {@link Supplier} with this context as the + * {@link #current} context. + * + *

This API is experimental and + * subject to change. + * + * @param supplier {@link Supplier} to use to produce the value. + * @see io.grpc.ExperimentalApi + * @return result of supplier. + */ + public V supply(Supplier supplier) { + Context previous = attach(); + try { + return supplier.get(); + } finally { + detach(previous); + } + } + /** * Wrap a {@link Runnable} so that it executes with this context as the {@link #current} context. */ diff --git a/api/src/test/java/io/grpc/ContextTest.java b/api/src/test/java/io/grpc/ContextTest.java index 7f24ff4461d..f1d5e019b82 100644 --- a/api/src/test/java/io/grpc/ContextTest.java +++ b/api/src/test/java/io/grpc/ContextTest.java @@ -26,6 +26,7 @@ import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -49,6 +50,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import java.util.logging.Handler; import java.util.logging.Level; import java.util.logging.LogRecord; @@ -553,6 +555,75 @@ public Object call() { current.detach(Context.ROOT); } + @Test + public void supply_executesSupplierUnderTargetContextAndRestoresPrevious() { + Context base = Context.current().withValue(PET, "cat"); + Context current = Context.current().withValue(PET, "fish"); + Context toRestore = current.attach(); + + final Object ret = new Object(); + Supplier supplier = new Supplier() { + @Override + public Object get() { + runner.run(); + return ret; + } + }; + + Object result = base.supply(supplier); + + assertSame(ret, result); + assertSame(base, observed); + assertSame(current, Context.current()); + + current.detach(toRestore); + } + + @Test + public void supply_whenContextIsAlreadyCurrent_executesAndMaintainsContext() { + Context current = Context.current().withValue(PET, "fish"); + Context toRestore = current.attach(); + + final Object ret = new Object(); + Supplier supplier = new Supplier() { + @Override + public Object get() { + runner.run(); + return ret; + } + }; + + Object result = current.supply(supplier); + + assertSame(ret, result); + assertSame(current, observed); + assertSame(current, Context.current()); + + current.detach(toRestore); + } + + @Test + public void supply_whenSupplierThrows_propagatesExceptionAndRestoresPreviousContext() { + Context base = Context.current().withValue(PET, "cat"); + Context current = Context.current().withValue(PET, "fish"); + Context toRestore = current.attach(); + + final TestError err = new TestError(); + Supplier supplier = new Supplier() { + @Override + public Object get() { + throw err; + } + }; + + TestError thrown = assertThrows(TestError.class, () -> base.supply(supplier)); + + assertSame(err, thrown); + assertSame(current, Context.current()); + + current.detach(toRestore); + } + @Test public void currentContextExecutor() { QueuedExecutor queuedExecutor = new QueuedExecutor(); diff --git a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransportFactory.java b/binder/src/main/java/io/grpc/binder/internal/BinderClientTransportFactory.java index 459e064ad9b..8188e4ecf71 100644 --- a/binder/src/main/java/io/grpc/binder/internal/BinderClientTransportFactory.java +++ b/binder/src/main/java/io/grpc/binder/internal/BinderClientTransportFactory.java @@ -101,6 +101,9 @@ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials ch @Override public void close() { + if (closed) { + return; + } closed = true; executorService = scheduledExecutorPool.returnObject(executorService); offloadExecutor = offloadExecutorPool.returnObject(offloadExecutor); diff --git a/buildscripts/make_dependencies.bat b/buildscripts/make_dependencies.bat index befb9c78baf..02461b9ae1f 100644 --- a/buildscripts/make_dependencies.bat +++ b/buildscripts/make_dependencies.bat @@ -1,4 +1,20 @@ -choco install -y pkgconfiglite --allow-empty-checksums +choco feature enable -n allowEmptyChecksums + +set RETRY=0 +:install_pkgconfig +choco install -y pkgconfiglite --allow-empty-checksums --force +if %ERRORLEVEL% neq 0 ( + if %RETRY% lss 3 ( + set /a RETRY=%RETRY%+1 + echo pkgconfiglite installation failed. Retrying %RETRY% of 3 in 5 seconds... + @rem Sleep for 5 seconds using the loopback ping trick (timeout command fails in non-interactive CI) + ping -n 6 127.0.0.1 >nul + goto :install_pkgconfig + ) + echo Failed to install pkgconfiglite after 3 attempts. + exit /b 1 +) + choco install -y openjdk --version=17.0 set PATH=%PATH%;"c:\Program Files\OpenJDK\jdk-17\bin" set PROTOBUF_VER=35.1 @@ -24,11 +40,11 @@ if not exist "%CMAKE_NAME%" ( set PATH=%PATH%;%cd%\%CMAKE_NAME%\bin :hasCmake @rem GitHub requires TLSv1.2, and for whatever reason our powershell doesn't have it enabled -powershell -command "$ProgressPreference = 'SilentlyContinue'; $ErrorActionPreference = 'stop'; & { [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12 ; iwr https://github.com/google/protobuf/releases/download/v%PROTOBUF_VER%/protobuf-%PROTOBUF_VER%.zip -OutFile protobuf.zip }" || exit /b 1 -powershell -command "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('protobuf.zip', '.') }" || exit /b 1 +call :RunPowershellWithRetry "$ProgressPreference = 'SilentlyContinue'; $ErrorActionPreference = 'stop'; & { [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12 ; iwr https://github.com/google/protobuf/releases/download/v%PROTOBUF_VER%/protobuf-%PROTOBUF_VER%.zip -OutFile protobuf.zip }" || exit /b 1 +call :RunPowershellWithRetry "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('protobuf.zip', '.') }" || exit /b 1 del protobuf.zip -powershell -command "$ProgressPreference = 'SilentlyContinue'; $ErrorActionPreference = 'stop'; & { [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12 ; iwr https://github.com/abseil/abseil-cpp/archive/refs/tags/%ABSL_VERSION%.zip -OutFile absl.zip }" || exit /b 1 -powershell -command "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('absl.zip', '.') }" || exit /b 1 +call :RunPowershellWithRetry "$ProgressPreference = 'SilentlyContinue'; $ErrorActionPreference = 'stop'; & { [Net.ServicePointManager]::SecurityProtocol = [Net.SecurityProtocolType]::Tls12 ; iwr https://github.com/abseil/abseil-cpp/archive/refs/tags/%ABSL_VERSION%.zip -OutFile absl.zip }" || exit /b 1 +call :RunPowershellWithRetry "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('absl.zip', '.') }" || exit /b 1 del absl.zip move abseil-cpp-%ABSL_VERSION% protobuf-%PROTOBUF_VER%\third_party\abseil-cpp mkdir protobuf-%PROTOBUF_VER%\build @@ -58,8 +74,25 @@ goto :eof :installCmake -powershell -command "$ErrorActionPreference = 'stop'; & { iwr https://cmake.org/files/v3.3/%CMAKE_NAME%.zip -OutFile cmake.zip }" || exit /b 1 -powershell -command "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('cmake.zip', '.') }" || exit /b 1 +call :RunPowershellWithRetry "$ErrorActionPreference = 'stop'; & { iwr https://cmake.org/files/v3.3/%CMAKE_NAME%.zip -OutFile cmake.zip }" || exit /b 1 +call :RunPowershellWithRetry "$ErrorActionPreference = 'stop'; & { Add-Type -AssemblyName System.IO.Compression.FileSystem; [System.IO.Compression.ZipFile]::ExtractToDirectory('cmake.zip', '.') }" || exit /b 1 del cmake.zip goto :eof +@rem Helper to retry powershell commands (e.g. for transient network failures during iwr) +:RunPowershellWithRetry +set "PS_CMD=%~1" +set PS_RETRY=0 +:ps_retry_loop +powershell -command "%PS_CMD%" +if %ERRORLEVEL% equ 0 exit /b 0 +if %PS_RETRY% lss 3 ( + set /a PS_RETRY=%PS_RETRY%+1 + echo PowerShell command failed. Retrying %PS_RETRY% of 3 in 5 seconds... + ping -n 6 127.0.0.1 >nul + goto :ps_retry_loop +) +echo PowerShell command failed after 3 attempts: %PS_CMD% +exit /b 1 + + diff --git a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java index 00df05a0c00..e533770da22 100644 --- a/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java +++ b/core/src/main/java/io/grpc/internal/ManagedChannelImpl.java @@ -173,7 +173,7 @@ public Result selectConfig(PickSubchannelArgs args) { private final NameResolverProvider nameResolverProvider; private final NameResolver.Args nameResolverArgs; private final LoadBalancerProvider loadBalancerFactory; - private final ClientTransportFactory originalTransportFactory; + private final RefCountedClientTransportFactory originalTransportFactory; @Nullable private final ChannelCredentials originalChannelCreds; private final ClientTransportFactory transportFactory; @@ -562,11 +562,15 @@ ClientStream newSubstream( this.executorPool = checkNotNull(builder.executorPool, "executorPool"); this.executor = checkNotNull(executorPool.getObject(), "executor"); this.originalChannelCreds = builder.channelCredentials; - this.originalTransportFactory = clientTransportFactory; + if (clientTransportFactory instanceof RefCountedClientTransportFactory) { + this.originalTransportFactory = (RefCountedClientTransportFactory) clientTransportFactory; + } else { + this.originalTransportFactory = new RefCountedClientTransportFactory(clientTransportFactory); + } this.offloadExecutorHolder = new ExecutorHolder(checkNotNull(builder.offloadExecutorPool, "offloadExecutorPool")); this.transportFactory = new CallCredentialsApplyingTransportFactory( - clientTransportFactory, builder.callCredentials, this.offloadExecutorHolder); + originalTransportFactory, builder.callCredentials, this.offloadExecutorHolder); this.scheduledExecutor = new RestrictedScheduledExecutor(transportFactory.getScheduledExecutorService()); maxTraceEvents = builder.maxTraceEvents; @@ -1462,7 +1466,10 @@ final class ResolvingOobChannelBuilder final ClientTransportFactory transportFactory; CallCredentials callCredentials; if (channelCreds instanceof DefaultChannelCreds) { - transportFactory = originalTransportFactory; + // TODO(kannanjgithub) We should eventually refactor ManagedChannelImplBuilder so + // callCredentials can be resolved lazily at build() time, allowing transport factory + // retention to happen strictly inside buildClientTransportFactory(). + transportFactory = originalTransportFactory.retain(); callCredentials = null; } else { SwapChannelCredentialsResult swapResult = diff --git a/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java new file mode 100644 index 00000000000..b1962b1e9f0 --- /dev/null +++ b/core/src/main/java/io/grpc/internal/RefCountedClientTransportFactory.java @@ -0,0 +1,75 @@ +/* + * Copyright 2026 The gRPC 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.grpc.internal; + +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; + +import io.grpc.ChannelCredentials; +import io.grpc.ChannelLogger; +import java.net.SocketAddress; +import java.util.Collection; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * A wrapper for {@link ClientTransportFactory} that reference-counts calls to {@link #retain()} and + * {@link #close()}, ensuring the delegate factory is closed only when all references are released. + */ +final class RefCountedClientTransportFactory implements ClientTransportFactory { + private final ClientTransportFactory delegate; + private final AtomicInteger refCount = new AtomicInteger(1); + + public RefCountedClientTransportFactory(ClientTransportFactory delegate) { + this.delegate = checkNotNull(delegate, "delegate"); + } + + public RefCountedClientTransportFactory retain() { + refCount.incrementAndGet(); + return this; + } + + @Override + public ConnectionClientTransport newClientTransport( + SocketAddress serverAddress, ClientTransportOptions options, ChannelLogger channelLogger) { + return delegate.newClientTransport(serverAddress, options, channelLogger); + } + + @Override + public ScheduledExecutorService getScheduledExecutorService() { + return delegate.getScheduledExecutorService(); + } + + @Override + public Collection> getSupportedSocketAddressTypes() { + return delegate.getSupportedSocketAddressTypes(); + } + + @Override + public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials channelCreds) { + return delegate.swapChannelCredentials(channelCreds); + } + + @Override + public void close() { + int count = refCount.decrementAndGet(); + checkState(count >= 0, "Reference count has gone negative: %s", count); + if (count == 0) { + delegate.close(); + } + } +} diff --git a/core/src/main/java/io/grpc/internal/SpiffeUtil.java b/core/src/main/java/io/grpc/internal/SpiffeUtil.java index 21232ad293b..102485ebd54 100644 --- a/core/src/main/java/io/grpc/internal/SpiffeUtil.java +++ b/core/src/main/java/io/grpc/internal/SpiffeUtil.java @@ -231,14 +231,8 @@ private static List extractCert(List> keysNode, for (Map keyNode : keysNode) { checkJwkEntry(keyNode, trustDomainName); List rawCerts = JsonUtil.getListOfStrings(keyNode, "x5c"); - if (rawCerts == null) { - throw new IllegalArgumentException(String.format("'x5c' parameter is required. Certificate " - + "loading for trust domain '%s' failed.", trustDomainName)); - } - if (rawCerts.size() != 1) { - throw new IllegalArgumentException(String.format("Exactly 1 certificate is expected, but " - + "%s found. Certificate loading for trust domain '%s' failed.", rawCerts.size(), - trustDomainName)); + if (rawCerts == null || rawCerts.isEmpty()) { + continue; } InputStream stream = new ByteArrayInputStream((CERTIFICATE_PREFIX + rawCerts.get(0) + "\n" + CERTIFICATE_SUFFIX) diff --git a/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java b/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java index e958fcdae00..d052cce137b 100644 --- a/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java +++ b/core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java @@ -4824,6 +4824,46 @@ public void run() { }); } + @Test + public void oobChannelTermination_doesNotCloseSharedTransportFactory() { + channelBuilder.nameResolverRegistry.register(new NameResolverProvider() { + @Override + public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) { + NameResolver resolver = mock(NameResolver.class); + when(resolver.getServiceAuthority()).thenReturn( + targetUri.getAuthority() != null ? targetUri.getAuthority() : targetUri.getPath()); + return resolver; + } + + @Override + public String getDefaultScheme() { + return expectedUri.getScheme(); + } + + @Override + protected boolean isAvailable() { + return true; + } + + @Override + protected int priority() { + return 10; + } + }); + createChannel(); + ManagedChannel oob = helper.createResolvingOobChannelBuilder("oobauthority").build(); + + // Shutting down OOB channel should release its reference but not close the + // shared transport factory + oob.shutdownNow(); + verify(mockTransportFactory, never()).close(); + + // Terminating the main channel releases the final reference and closes the + // transport factory + channel.shutdownNow(); + verify(mockTransportFactory).close(); + } + @SuppressWarnings("unchecked") private static Map parseConfig(String json) throws Exception { return (Map) JsonParser.parse(json); diff --git a/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java new file mode 100644 index 00000000000..3ead6396ce5 --- /dev/null +++ b/core/src/test/java/io/grpc/internal/RefCountedClientTransportFactoryTest.java @@ -0,0 +1,81 @@ +/* + * Copyright 2026 The gRPC 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.grpc.internal; + +import static org.junit.Assert.assertThrows; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnit; +import org.mockito.junit.MockitoRule; + +/** Unit tests for {@link RefCountedClientTransportFactory}. */ +@RunWith(JUnit4.class) +public class RefCountedClientTransportFactoryTest { + @Rule public final MockitoRule mocks = MockitoJUnit.rule(); + + @Mock private ClientTransportFactory mockDelegate; + + @Test + public void singleClose_closesDelegate() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.close(); + verify(mockDelegate).close(); + } + + @Test + public void retainAndClose_closesDelegateOnlyWhenCountReachesZero() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + RefCountedClientTransportFactory retained = factory.retain(); + + factory.close(); + verify(mockDelegate, never()).close(); + + retained.close(); + verify(mockDelegate).close(); + } + + @Test + public void multipleRetains_requiresEqualClosesToCloseDelegate() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.retain(); + factory.retain(); + + factory.close(); + verify(mockDelegate, never()).close(); + + factory.close(); + verify(mockDelegate, never()).close(); + + factory.close(); + verify(mockDelegate).close(); + } + + @Test + public void closeMoreThanRetain_throwsIllegalStateException() { + RefCountedClientTransportFactory factory = new RefCountedClientTransportFactory(mockDelegate); + factory.close(); + verify(mockDelegate).close(); + + assertThrows(IllegalStateException.class, factory::close); + } +} diff --git a/core/src/test/java/io/grpc/internal/SpiffeUtilTest.java b/core/src/test/java/io/grpc/internal/SpiffeUtilTest.java index 8888046c586..bea79704c66 100644 --- a/core/src/test/java/io/grpc/internal/SpiffeUtilTest.java +++ b/core/src/test/java/io/grpc/internal/SpiffeUtilTest.java @@ -231,6 +231,8 @@ public static class CertificateApiTest { private static final String SPIFFE_TRUST_BUNDLE_WRONG_ROOT = "spiffebundle_wrong_root.json"; private static final String SPIFFE_TRUST_BUNDLE_WRONG_SEQ = "spiffebundle_wrong_seq_type.json"; private static final String SPIFFE_TRUST_BUNDLE_MISSING_X5C = "spiffebundle_missing_x5c.json"; + private static final String SPIFFE_TRUST_BUNDLE_EMPTY_X5C = "spiffebundle_empty_x5c.json"; + private static final String SPIFFE_TRUST_BUNDLE_IGNORED_KEYS = "spiffebundle_ignored_keys.json"; private static final String DOMAIN_ERROR_MESSAGE = " Certificate loading for trust domain 'google.com' failed."; @@ -330,6 +332,54 @@ public void loadTrustBundleFromFileSuccessTest() throws Exception { assertEquals("foo.bar.com", spiffeId_ec.get().getTrustDomain()); } + @Test + public void loadTrustBundleFromFileWithMultiCertsSuccessTest() throws Exception { + SpiffeBundle tb = SpiffeUtil.loadTrustBundleFromFile( + copyFileToTmp(SPIFFE_TRUST_BUNDLE_WRONG_MULTI_CERTS)); + assertEquals(1, tb.getSequenceNumbers().size()); + assertEquals(123L, (long) tb.getSequenceNumbers().get("google.com")); + assertEquals(1, tb.getBundleMap().size()); + assertEquals(1, tb.getBundleMap().get("google.com").size()); + Optional spiffeId = SpiffeUtil.extractSpiffeId( + tb.getBundleMap().get("google.com").toArray(new X509Certificate[0])); + assertTrue(spiffeId.isPresent()); + assertEquals("foo.bar.com", spiffeId.get().getTrustDomain()); + } + + @Test + public void loadTrustBundleFromFileWithMissingX5cSuccessTest() throws Exception { + SpiffeBundle tb = SpiffeUtil.loadTrustBundleFromFile( + copyFileToTmp(SPIFFE_TRUST_BUNDLE_MISSING_X5C)); + assertEquals(1, tb.getBundleMap().size()); + assertEquals(1, tb.getBundleMap().get("google.com").size()); + } + + @Test + public void loadTrustBundleFromFileWithEmptyX5cSuccessTest() throws Exception { + SpiffeBundle tb = SpiffeUtil.loadTrustBundleFromFile( + copyFileToTmp(SPIFFE_TRUST_BUNDLE_EMPTY_X5C)); + assertEquals(1, tb.getBundleMap().size()); + assertEquals(1, tb.getBundleMap().get("google.com").size()); + Optional spiffeId = SpiffeUtil.extractSpiffeId( + tb.getBundleMap().get("google.com").toArray(new X509Certificate[0])); + assertTrue(spiffeId.isPresent()); + assertEquals("foo.bar.com", spiffeId.get().getTrustDomain()); + } + + @Test + public void loadTrustBundleFromFileWithIgnoredKeysSuccessTest() throws Exception { + SpiffeBundle tb = SpiffeUtil.loadTrustBundleFromFile( + copyFileToTmp(SPIFFE_TRUST_BUNDLE_IGNORED_KEYS)); + assertEquals(1, tb.getSequenceNumbers().size()); + assertEquals(123L, (long) tb.getSequenceNumbers().get("google.com")); + assertEquals(1, tb.getBundleMap().size()); + assertEquals(1, tb.getBundleMap().get("google.com").size()); + Optional spiffeId = SpiffeUtil.extractSpiffeId( + tb.getBundleMap().get("google.com").toArray(new X509Certificate[0])); + assertTrue(spiffeId.isPresent()); + assertEquals("foo.bar.com", spiffeId.get().getTrustDomain()); + } + @Test public void loadTrustBundleFromFileFailureTest() { // Check the exception if JSON root element is different from 'trust_domains' @@ -352,10 +402,6 @@ public void loadTrustBundleFromFileFailureTest() { iae = assertThrows(IllegalArgumentException.class, () -> SpiffeUtil .loadTrustBundleFromFile(copyFileToTmp(SPIFFE_TRUST_BUNDLE_CORRUPTED_CERT))); assertEquals("Certificate can't be parsed." + DOMAIN_ERROR_MESSAGE, iae.getMessage()); - // Check the exception if a key entry is missing the 'x5c' parameter - iae = assertThrows(IllegalArgumentException.class, () -> SpiffeUtil - .loadTrustBundleFromFile(copyFileToTmp(SPIFFE_TRUST_BUNDLE_MISSING_X5C))); - assertEquals("'x5c' parameter is required." + DOMAIN_ERROR_MESSAGE, iae.getMessage()); // Check the exception if 'kty' value differs from 'RSA' iae = assertThrows(IllegalArgumentException.class, () -> SpiffeUtil .loadTrustBundleFromFile(copyFileToTmp(SPIFFE_TRUST_BUNDLE_WRONG_KTY))); @@ -371,11 +417,6 @@ public void loadTrustBundleFromFileFailureTest() { .loadTrustBundleFromFile(copyFileToTmp(SPIFFE_TRUST_BUNDLE_WRONG_USE))); assertEquals("'use' parameter must be 'x509-svid' but 'i_am_not_x509-svid' found." + DOMAIN_ERROR_MESSAGE, iae.getMessage()); - // Check the exception if multiple certs are provided for 'x5c' - iae = assertThrows(IllegalArgumentException.class, () -> SpiffeUtil - .loadTrustBundleFromFile(copyFileToTmp(SPIFFE_TRUST_BUNDLE_WRONG_MULTI_CERTS))); - assertEquals("Exactly 1 certificate is expected, but 2 found." + DOMAIN_ERROR_MESSAGE, - iae.getMessage()); } @Test diff --git a/core/src/test/resources/io/grpc/internal/spiffebundle_empty_x5c.json b/core/src/test/resources/io/grpc/internal/spiffebundle_empty_x5c.json new file mode 100644 index 00000000000..70eabf2233c --- /dev/null +++ b/core/src/test/resources/io/grpc/internal/spiffebundle_empty_x5c.json @@ -0,0 +1,19 @@ +{ + "trust_domains": { + "google.com": { + "spiffe_sequence": 123, + "keys": [ + { + "kty": "RSA", + "use": "x509-svid", + "x5c": [] + }, + { + "kty": "RSA", + "use": "x509-svid", + "x5c": ["MIIFsjCCA5qgAwIBAgIURygVMMzdr+Q7rsUaz189JozyHMwwDQYJKoZIhvcNAQELBQAwTjELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMQwwCgYDVQQHDANTVkwxDTALBgNVBAoMBGdSUEMxFTATBgNVBAMMDHRlc3QtY2xpZW50MTAeFw0yMTEyMjMxODQyNTJaFw0zMTEyMjExODQyNTJaME4xCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJDQTEMMAoGA1UEBwwDU1ZMMQ0wCwYDVQQKDARnUlBDMRUwEwYDVQQDDAx0ZXN0LWNsaWVudDEwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQDJ4AqpGetyVSqGUuBJLVFla+7bEfca7UYzfVSSZLZ/X+JDmWIVN8UIPuFib5jhMEc3XaUnFXUmM7zEtz/ZG5hapwLwOb2C3ZxOP6PQjYCJxbkLie+b43UQrFu1xxd3vMhVJgcj/AIxEpmszuqOa6kUrkYifjJADQ+64kZgl66bsTdXMCzpxyFl9xUfff59L8OX+HUfAcoZz3emjg3ZJPYURQEmjdZTOau1EjFilwHgd989Jt7NKgx30NXoHmw7nusVBIY94fL2VKN3f1XVm0dHu5NI279Q6zr0ZBU7k5T3IeHnzsUesQS4NGlklDWoVTKk73Uv9Pna8yQsSW757PEbHOGp9Knu4bnoGPOlsG81yIPipO6hTgGFK24pF97M9kpGbWqYX4+2vLlrCAfcmsHqaUPmQlYeRVTT6vw7ctYo2kyUYGtnODXk76LqewRBVvkzx75QUhfjAyb740YcDmIenc56Tq6gebJHjhEmVSehR6xIpXP7SVeurTyhPsEQnpJHtgs4dcwWOZp7BvPNzHXmJqfr7vsshie3vS5kQ0u1e1yqAqXgyDjqKXOkx+dpgUTehSJHhPNHvTc5LXRsvvXKYz6FrwR/DZ8t7BNEvPeLjFgxpH7QVJFLCvCbXs5K6yYbsnLfxFIBPRnrbJkIsK+sQwnRdnsiUdPsTkG5B2lQfQIDAQABo4GHMIGEMB0GA1UdDgQWBBQ2lBp0PiRHHvQ5IRURm8aHsj4RETAfBgNVHSMEGDAWgBQ2lBp0PiRHHvQ5IRURm8aHsj4RETAPBgNVHRMBAf8EBTADAQH/MDEGA1UdEQQqMCiGJnNwaWZmZTovL2Zvby5iYXIuY29tL2NsaWVudC93b3JrbG9hZC8xMA0GCSqGSIb3DQEBCwUAA4ICAQA1mSkgRclAl+E/aS9zJ7t8+Y4n3T24nOKKveSIjxXm/zjhWqVsLYBI6kglWtih2+PELvU8JdPqNZK34Kl0Q6FWpVSGDdWN1i6NyORt2ocggL3ke3iXxRk3UpUKJmqwz81VhA2KUHnMlyE0IufFfZNwNWWHBv13uJfRbjeQpKPhU+yf4DeXrsWcvrZlGvAET+mcplafUzCp7Iv+PcISJtUerbxbVtuHVeZCLlgDXWkLAWJN8rf0dIG4x060LJ+j6j9uRVhb9sZn1HJV+j4XdIYm1VKilluhOtNwP2d3Ox/JuTBxf7hFHXZPfMagQE5k5PzmxRaCAEMJ1l2DvUbZw+shJfSNoWcBo2qadnUaWT3BmmJRBDh7ZReib/RQ1Rd4ygOyzP3E0vkV4/gqyjLdApXh5PZP8KLQZ+1JN/sdWt7VfIt9wYOpkIqujdll51ESHzwQeAK9WVCB4UvVz6zdhItB9CRbXPreWC+wCB1xDovIzFKOVsLs5+Gqs1m7VinG2LxbDqaKyo/FB0Hxx0acBNzezLWoDwXYQrN0T0S4pnqhKD1CYPpdArBkNezUYAjS725FkApuK+mnBX3U0msBffEaUEOkcyar1EW2m/33vpetD/k3eQQkmvQf4Hbiu9AF+9cNDm/hMuXEw5EXGA91fn0891b5eEW8BJHXX0jri0aN8g=="] + } + ] + } + } +} diff --git a/core/src/test/resources/io/grpc/internal/spiffebundle_ignored_keys.json b/core/src/test/resources/io/grpc/internal/spiffebundle_ignored_keys.json new file mode 100644 index 00000000000..f1daffba60e --- /dev/null +++ b/core/src/test/resources/io/grpc/internal/spiffebundle_ignored_keys.json @@ -0,0 +1,23 @@ +{ + "trust_domains": { + "google.com": { + "spiffe_sequence": 123, + "keys": [ + { + "kty": "RSA", + "use": "x509-svid" + }, + { + "kty": "RSA", + "use": "x509-svid", + "x5c": [] + }, + { + "kty": "RSA", + "use": "x509-svid", + "x5c": ["MIIFsjCCA5qgAwIBAgIURygVMMzdr+Q7rsUaz189JozyHMwwDQYJKoZIhvcNAQELBQAwTjELMAkGA1UEBhMCVVMxCzAJBgNVBAgMAkNBMQwwCgYDVQQHDANTVkwxDTALBgNVBAoMBGdSUEMxFTATBgNVBAMMDHRlc3QtY2xpZW50MTAeFw0yMTEyMjMxODQyNTJaFw0zMTEyMjExODQyNTJaME4xCzAJBgNVBAYTAlVTMQswCQYDVQQIDAJDQTEMMAoGA1UEBwwDU1ZMMQ0wCwYDVQQKDARnUlBDMRUwEwYDVQQDDAx0ZXN0LWNsaWVudDEwggIiMA0GCSqGSIb3DQEBAQUAA4ICDwAwggIKAoICAQDJ4AqpGetyVSqGUuBJLVFla+7bEfca7UYzfVSSZLZ/X+JDmWIVN8UIPuFib5jhMEc3XaUnFXUmM7zEtz/ZG5hapwLwOb2C3ZxOP6PQjYCJxbkLie+b43UQrFu1xxd3vMhVJgcj/AIxEpmszuqOa6kUrkYifjJADQ+64kZgl66bsTdXMCzpxyFl9xUfff59L8OX+HUfAcoZz3emjg3ZJPYURQEmjdZTOau1EjFilwHgd989Jt7NKgx30NXoHmw7nusVBIY94fL2VKN3f1XVm0dHu5NI279Q6zr0ZBU7k5T3IeHnzsUesQS4NGlklDWoVTKk73Uv9Pna8yQsSW757PEbHOGp9Knu4bnoGPOlsG81yIPipO6hTgGFK24pF97M9kpGbWqYX4+2vLlrCAfcmsHqaUPmQlYeRVTT6vw7ctYo2kyUYGtnODXk76LqewRBVvkzx75QUhfjAyb740YcDmIenc56Tq6gebJHjhEmVSehR6xIpXP7SVeurTyhPsEQnpJHtgs4dcwWOZp7BvPNzHXmJqfr7vsshie3vS5kQ0u1e1yqAqXgyDjqKXOkx+dpgUTehSJHhPNHvTc5LXRsvvXKYz6FrwR/DZ8t7BNEvPeLjFgxpH7QVJFLCvCbXs5K6yYbsnLfxFIBPRnrbJkIsK+sQwnRdnsiUdPsTkG5B2lQfQIDAQABo4GHMIGEMB0GA1UdDgQWBBQ2lBp0PiRHHvQ5IRURm8aHsj4RETAfBgNVHSMEGDAWgBQ2lBp0PiRHHvQ5IRURm8aHsj4RETAPBgNVHRMBAf8EBTADAQH/MDEGA1UdEQQqMCiGJnNwaWZmZTovL2Zvby5iYXIuY29tL2NsaWVudC93b3JrbG9hZC8xMA0GCSqGSIb3DQEBCwUAA4ICAQA1mSkgRclAl+E/aS9zJ7t8+Y4n3T24nOKKveSIjxXm/zjhWqVsLYBI6kglWtih2+PELvU8JdPqNZK34Kl0Q6FWpVSGDdWN1i6NyORt2ocggL3ke3iXxRk3UpUKJmqwz81VhA2KUHnMlyE0IufFfZNwNWWHBv13uJfRbjeQpKPhU+yf4DeXrsWcvrZlGvAET+mcplafUzCp7Iv+PcISJtUerbxbVtuHVeZCLlgDXWkLAWJN8rf0dIG4x060LJ+j6j9uRVhb9sZn1HJV+j4XdIYm1VKilluhOtNwP2d3Ox/JuTBxf7hFHXZPfMagQE5k5PzmxRaCAEMJ1l2DvUbZw+shJfSNoWcBo2qadnUaWT3BmmJRBDh7ZReib/RQ1Rd4ygOyzP3E0vkV4/gqyjLdApXh5PZP8KLQZ+1JN/sdWt7VfIt9wYOpkIqujdll51ESHzwQeAK9WVCB4UvVz6zdhItB9CRbXPreWC+wCB1xDovIzFKOVsLs5+Gqs1m7VinG2LxbDqaKyo/FB0Hxx0acBNzezLWoDwXYQrN0T0S4pnqhKD1CYPpdArBkNezUYAjS725FkApuK+mnBX3U0msBffEaUEOkcyar1EW2m/33vpetD/k3eQQkmvQf4Hbiu9AF+9cNDm/hMuXEw5EXGA91fn0891b5eEW8BJHXX0jri0aN8g=="] + } + ] + } + } +} diff --git a/cronet/src/main/java/io/grpc/cronet/CronetChannelBuilder.java b/cronet/src/main/java/io/grpc/cronet/CronetChannelBuilder.java index 7ea1bc891c2..3453d8ee405 100644 --- a/cronet/src/main/java/io/grpc/cronet/CronetChannelBuilder.java +++ b/cronet/src/main/java/io/grpc/cronet/CronetChannelBuilder.java @@ -246,6 +246,7 @@ static class CronetTransportFactory implements ClientTransportFactory { private final boolean usingSharedScheduler; private final boolean useGetForSafeMethods; private final boolean usePutForIdempotentMethods; + private boolean closed; private CronetTransportFactory( StreamBuilderFactory streamFactory, @@ -271,6 +272,9 @@ private CronetTransportFactory( @Override public ConnectionClientTransport newClientTransport( SocketAddress addr, ClientTransportOptions options, ChannelLogger channelLogger) { + if (closed) { + throw new IllegalStateException("The transport factory is closed."); + } InetSocketAddress inetSocketAddr = (InetSocketAddress) addr; return new CronetClientTransport(streamFactory, inetSocketAddr, options.getAuthority(), options.getUserAgent(), options.getEagAttributes(), executor, maxMessageSize, @@ -289,6 +293,10 @@ public SwapChannelCredentialsResult swapChannelCredentials(ChannelCredentials ch @Override public void close() { + if (closed) { + return; + } + closed = true; if (usingSharedScheduler) { SharedResourceHolder.release(GrpcUtil.TIMER_SERVICE, timeoutService); } diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 8cce5b9babe..c1f7d13d411 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -102,14 +102,14 @@ lincheck = "org.jetbrains.lincheck:lincheck:3.7" mockito-android = "org.mockito:mockito-android:4.4.0" # checkForUpdates: mockito-core:4.4.+ mockito-core = "org.mockito:mockito-core:4.4.0" -netty-codec-http2 = "io.netty:netty-codec-http2:4.2.15.Final" -netty-handler-proxy = "io.netty:netty-handler-proxy:4.2.15.Final" +netty-codec-http2 = "io.netty:netty-codec-http2:4.2.16.Final" +netty-handler-proxy = "io.netty:netty-handler-proxy:4.2.16.Final" # Keep the following references of tcnative version in sync whenever it's updated: # SECURITY.md -netty-tcnative = "io.netty:netty-tcnative-boringssl-static:2.0.75.Final" -netty-tcnative-classes = "io.netty:netty-tcnative-classes:2.0.75.Final" -netty-transport-epoll = "io.netty:netty-transport-native-epoll:4.2.15.Final" -netty-unix-common = "io.netty:netty-transport-native-unix-common:4.2.15.Final" +netty-tcnative = "io.netty:netty-tcnative-boringssl-static:2.0.81.Final" +netty-tcnative-classes = "io.netty:netty-tcnative-classes:2.0.81.Final" +netty-transport-epoll = "io.netty:netty-transport-native-epoll:4.2.16.Final" +netty-unix-common = "io.netty:netty-transport-native-unix-common:4.2.16.Final" okhttp = "com.squareup.okhttp:okhttp:2.7.5" # okio 3.5+ uses Kotlin 1.9+ which requires Android Gradle Plugin 9+ # checkForUpdates: okio:3.4.+ diff --git a/repositories.bzl b/repositories.bzl index 0bbdfb9c23d..22bb6cac5b0 100644 --- a/repositories.bzl +++ b/repositories.bzl @@ -30,20 +30,20 @@ IO_GRPC_GRPC_JAVA_ARTIFACTS = [ "dev.cel:common:0.13.1", "com.squareup.okhttp:okhttp:2.7.5", "com.squareup.okio:okio:2.10.0", # 3.0+ needs swapping to -jvm; need work to avoid flag-day - "io.netty:netty-buffer:4.2.15.Final", - "io.netty:netty-codec-base:4.2.15.Final", - "io.netty:netty-codec-http2:4.2.15.Final", - "io.netty:netty-codec-http:4.2.15.Final", - "io.netty:netty-codec-socks:4.2.15.Final", - "io.netty:netty-common:4.2.15.Final", - "io.netty:netty-handler-proxy:4.2.15.Final", - "io.netty:netty-handler:4.2.15.Final", - "io.netty:netty-resolver:4.2.15.Final", - "io.netty:netty-tcnative-boringssl-static:2.0.75.Final", - "io.netty:netty-tcnative-classes:2.0.75.Final", - "io.netty:netty-transport-native-epoll:jar:linux-x86_64:4.2.15.Final", - "io.netty:netty-transport-native-unix-common:4.2.15.Final", - "io.netty:netty-transport:4.2.15.Final", + "io.netty:netty-buffer:4.2.16.Final", + "io.netty:netty-codec-base:4.2.16.Final", + "io.netty:netty-codec-http2:4.2.16.Final", + "io.netty:netty-codec-http:4.2.16.Final", + "io.netty:netty-codec-socks:4.2.16.Final", + "io.netty:netty-common:4.2.16.Final", + "io.netty:netty-handler-proxy:4.2.16.Final", + "io.netty:netty-handler:4.2.16.Final", + "io.netty:netty-resolver:4.2.16.Final", + "io.netty:netty-tcnative-boringssl-static:2.0.81.Final", + "io.netty:netty-tcnative-classes:2.0.81.Final", + "io.netty:netty-transport-native-epoll:jar:linux-x86_64:4.2.16.Final", + "io.netty:netty-transport-native-unix-common:4.2.16.Final", + "io.netty:netty-transport:4.2.16.Final", "io.opencensus:opencensus-api:0.31.0", "io.opencensus:opencensus-contrib-grpc-metrics:0.31.0", "io.perfmark:perfmark-api:0.27.0", diff --git a/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java b/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java index ca3ec3b9db5..fd5865bea2f 100644 --- a/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java +++ b/rls/src/main/java/io/grpc/rls/CachingRlsLbClient.java @@ -324,7 +324,7 @@ private void periodicClean() { @GuardedBy("lock") private CachedRouteLookupResponse asyncRlsCall( RouteLookupRequestKey routeLookupRequestKey, @Nullable BackoffPolicy backoffPolicy, - RouteLookupRequest.Reason routeLookupReason) { + RouteLookupRequest.Reason routeLookupReason, @Nullable String staleHeaderData) { if (throttler.shouldThrottle()) { logger.log(ChannelLogLevel.DEBUG, "[RLS Entry {0}] Throttled RouteLookup", routeLookupRequestKey); @@ -336,7 +336,8 @@ private CachedRouteLookupResponse asyncRlsCall( } final SettableFuture response = SettableFuture.create(); io.grpc.lookup.v1.RouteLookupRequest routeLookupRequest = REQUEST_CONVERTER.convert( - RouteLookupRequest.create(routeLookupRequestKey.keyMap(), routeLookupReason)); + RouteLookupRequest.create( + routeLookupRequestKey.keyMap(), routeLookupReason, staleHeaderData)); logger.log(ChannelLogLevel.DEBUG, "[RLS Entry {0}] Starting RouteLookup: {1}", routeLookupRequestKey, routeLookupRequest); rlsStub.withDeadlineAfter(callTimeoutNanos, TimeUnit.NANOSECONDS) @@ -386,7 +387,7 @@ final CachedRouteLookupResponse get(final RouteLookupRequestKey routeLookupReque } return asyncRlsCall(routeLookupRequestKey, cacheEntry instanceof BackoffCacheEntry ? ((BackoffCacheEntry) cacheEntry).backoffPolicy : null, - RouteLookupRequest.Reason.REASON_MISS); + RouteLookupRequest.Reason.REASON_MISS, /* staleHeaderData= */ null); } if (cacheEntry instanceof DataCacheEntry) { @@ -717,7 +718,7 @@ void maybeRefresh() { logger.log(ChannelLogLevel.DEBUG, "[RLS Entry {0}] Cache entry is stale, refreshing", routeLookupRequestKey); asyncRlsCall(routeLookupRequestKey, /* backoffPolicy= */ null, - RouteLookupRequest.Reason.REASON_STALE); + RouteLookupRequest.Reason.REASON_STALE, getHeaderData()); } } diff --git a/rls/src/main/java/io/grpc/rls/RlsProtoConverters.java b/rls/src/main/java/io/grpc/rls/RlsProtoConverters.java index 70f9fb4d891..c5eb3eebf0b 100644 --- a/rls/src/main/java/io/grpc/rls/RlsProtoConverters.java +++ b/rls/src/main/java/io/grpc/rls/RlsProtoConverters.java @@ -65,18 +65,22 @@ static final class RouteLookupRequestConverter protected RlsProtoData.RouteLookupRequest doForward(RouteLookupRequest routeLookupRequest) { return RlsProtoData.RouteLookupRequest.create( ImmutableMap.copyOf(routeLookupRequest.getKeyMapMap()), - RlsProtoData.RouteLookupRequest.Reason.valueOf(routeLookupRequest.getReason().name()) + RlsProtoData.RouteLookupRequest.Reason.valueOf(routeLookupRequest.getReason().name()), + Strings.emptyToNull(routeLookupRequest.getStaleHeaderData()) ); } @Override protected RouteLookupRequest doBackward(RlsProtoData.RouteLookupRequest routeLookupRequest) { - return + RouteLookupRequest.Builder builder = RouteLookupRequest.newBuilder() .setTargetType("grpc") .setReason(RouteLookupRequest.Reason.valueOf(routeLookupRequest.reason().name())) - .putAllKeyMap(routeLookupRequest.keyMap()) - .build(); + .putAllKeyMap(routeLookupRequest.keyMap()); + if (routeLookupRequest.staleHeaderData() != null) { + builder.setStaleHeaderData(routeLookupRequest.staleHeaderData()); + } + return builder.build(); } } diff --git a/rls/src/main/java/io/grpc/rls/RlsProtoData.java b/rls/src/main/java/io/grpc/rls/RlsProtoData.java index 39c404870f9..2dfd075a6a8 100644 --- a/rls/src/main/java/io/grpc/rls/RlsProtoData.java +++ b/rls/src/main/java/io/grpc/rls/RlsProtoData.java @@ -61,8 +61,16 @@ enum Reason { /** Returns a map of key values extracted via key builders for the gRPC or HTTP request. */ abstract ImmutableMap keyMap(); + @Nullable + abstract String staleHeaderData(); + + static RouteLookupRequest create( + ImmutableMap keyMap, Reason reason, @Nullable String staleHeaderData) { + return new AutoValue_RlsProtoData_RouteLookupRequest(reason, keyMap, staleHeaderData); + } + static RouteLookupRequest create(ImmutableMap keyMap, Reason reason) { - return new AutoValue_RlsProtoData_RouteLookupRequest(reason, keyMap); + return create(keyMap, reason, null); } } diff --git a/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java b/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java index c5f06195964..86af63642e3 100644 --- a/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java +++ b/rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java @@ -277,6 +277,160 @@ public void get_noError_lifeCycle() throws Exception { inOrder.verifyNoMoreInteractions(); } + @Test + public void asyncRefresh_sendsStaleHeaderData() throws Exception { + setUpRlsLbClient(); + RlsProtoData.RouteLookupRequestKey routeLookupRequestKey = + RlsProtoData.RouteLookupRequestKey.create( + ImmutableMap.of( + "server", "bigtable.googleapis.com", "service-key", "foo", "method-key", "bar")); + rlsServerImpl.setLookupTable( + ImmutableMap.of( + routeLookupRequestKey, + RouteLookupResponse.create(ImmutableList.of("target"), "stale-header-v1"))); + + // Initial lookup: cache miss + CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequestKey); + assertThat(resp.isPending()).isTrue(); + + // RLS server response arrives + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + resp = getInSyncContext(routeLookupRequestKey); + assertThat(resp.hasData()).isTrue(); + assertThat(resp.getHeaderData()).isEqualTo("stale-header-v1"); + assertThat(rlsServerImpl.routeLookupReason).isEqualTo( + io.grpc.lookup.v1.RouteLookupRequest.Reason.REASON_MISS); + + // Advance fake clock past staleAge + fakeClock.forwardTime(ROUTE_LOOKUP_CONFIG.staleAgeInNanos(), TimeUnit.NANOSECONDS); + + rlsServerImpl.routeLookupReason = null; + rlsServerImpl.routeLookupStaleHeaderData = null; + + // Lookup on stale entry: returns cached response immediately + resp = getInSyncContext(routeLookupRequestKey); + assertThat(resp.hasData()).isTrue(); + assertThat(resp.getHeaderData()).isEqualTo("stale-header-v1"); + + // Async refresh finishes + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + assertThat(rlsServerImpl.routeLookupReason).isEqualTo( + io.grpc.lookup.v1.RouteLookupRequest.Reason.REASON_STALE); + assertThat(rlsServerImpl.routeLookupStaleHeaderData).isEqualTo("stale-header-v1"); + } + + @Test + public void updatedHeaderData_replacesCachedHeaderData() throws Exception { + setUpRlsLbClient(); + RlsProtoData.RouteLookupRequestKey routeLookupRequestKey = + RlsProtoData.RouteLookupRequestKey.create( + ImmutableMap.of( + "server", "bigtable.googleapis.com", "service-key", "foo", "method-key", "bar")); + rlsServerImpl.setLookupTable( + ImmutableMap.of( + routeLookupRequestKey, + RouteLookupResponse.create(ImmutableList.of("target"), "stale-header-v1"))); + + // Initial lookup + getInSyncContext(routeLookupRequestKey); + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + // RLS server updated to return new header data + rlsServerImpl.setLookupTable( + ImmutableMap.of( + routeLookupRequestKey, + RouteLookupResponse.create(ImmutableList.of("target"), "updated-header-v2"))); + + // Advance fake clock past staleAge + fakeClock.forwardTime(ROUTE_LOOKUP_CONFIG.staleAgeInNanos(), TimeUnit.NANOSECONDS); + + // Stale lookup triggers background refresh + getInSyncContext(routeLookupRequestKey); + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + // Verify cache entry is updated with new header data + CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequestKey); + assertThat(resp.getHeaderData()).isEqualTo("updated-header-v2"); + + // Advance past staleAge again + fakeClock.forwardTime(ROUTE_LOOKUP_CONFIG.staleAgeInNanos(), TimeUnit.NANOSECONDS); + rlsServerImpl.routeLookupStaleHeaderData = null; + + // Next stale refresh sends updated stale_header_data + getInSyncContext(routeLookupRequestKey); + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + assertThat(rlsServerImpl.routeLookupStaleHeaderData).isEqualTo("updated-header-v2"); + } + + @Test + public void expiredEntry_cacheMiss_clearsStaleHeaderData() throws Exception { + setUpRlsLbClient(); + RlsProtoData.RouteLookupRequestKey routeLookupRequestKey = + RlsProtoData.RouteLookupRequestKey.create( + ImmutableMap.of( + "server", "bigtable.googleapis.com", "service-key", "foo", "method-key", "bar")); + rlsServerImpl.setLookupTable( + ImmutableMap.of( + routeLookupRequestKey, + RouteLookupResponse.create(ImmutableList.of("target"), "stale-header-v1"))); + + // Initial lookup + getInSyncContext(routeLookupRequestKey); + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + // Advance fake clock past maxAge (expiration) + fakeClock.forwardTime(ROUTE_LOOKUP_CONFIG.maxAgeInNanos(), TimeUnit.NANOSECONDS); + + rlsServerImpl.routeLookupReason = null; + rlsServerImpl.routeLookupStaleHeaderData = null; + + // Expired entry triggers cache miss + CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequestKey); + assertThat(resp.isPending()).isTrue(); + + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + assertThat(rlsServerImpl.routeLookupReason).isEqualTo( + io.grpc.lookup.v1.RouteLookupRequest.Reason.REASON_MISS); + assertThat(rlsServerImpl.routeLookupStaleHeaderData).isEmpty(); + } + + @Test + public void rlsPicker_attachesHeaderDataToPickedRpcs() throws Exception { + setUpRlsLbClient(); + RlsProtoData.RouteLookupRequestKey routeLookupRequestKey = + RlsProtoData.RouteLookupRequestKey.create( + ImmutableMap.of( + "server", "bigtable.googleapis.com", "service-key", "service1", + "method-key", "create")); + rlsServerImpl.setLookupTable( + ImmutableMap.of( + routeLookupRequestKey, + RouteLookupResponse.create( + ImmutableList.of("primary.cloudbigtable.googleapis.com"), + "header-rls-data-value"))); + + // Populate cache and wait for server response + getInSyncContext(routeLookupRequestKey); + fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS); + + ArgumentCaptor pickerCaptor = + ArgumentCaptor.forClass(SubchannelPicker.class); + verify(helper, times(3)) + .updateBalancingState(any(ConnectivityState.class), pickerCaptor.capture()); + + Metadata headers = new Metadata(); + headers.put(RLS_DATA_KEY, "old-header-data"); + + PickResult pickResult = getPickResultForCreate(pickerCaptor, headers); + + assertThat(pickResult.getStatus().isOk()).isTrue(); + assertThat(headers.get(RLS_DATA_KEY)).isEqualTo("header-rls-data-value"); + } + + @Test public void rls_withCustomRlsChannelServiceConfig() throws Exception { Map routeLookupChannelServiceConfig = @@ -1120,6 +1274,7 @@ private static final class StaticFixedDelayRlsServerImpl private Map lookupTable = ImmutableMap.of(); io.grpc.lookup.v1.RouteLookupRequest.Reason routeLookupReason; + String routeLookupStaleHeaderData; public StaticFixedDelayRlsServerImpl( long responseDelayNano, ScheduledExecutorService scheduledExecutorService) { @@ -1143,6 +1298,7 @@ public void routeLookup(final io.grpc.lookup.v1.RouteLookupRequest request, @Override public void run() { routeLookupReason = request.getReason(); + routeLookupStaleHeaderData = request.getStaleHeaderData(); RouteLookupResponse response = lookupTable.get( RlsProtoData.RouteLookupRequestKey.create( diff --git a/rls/src/test/java/io/grpc/rls/RlsProtoConvertersTest.java b/rls/src/test/java/io/grpc/rls/RlsProtoConvertersTest.java index 82ad606c50d..0bb17a866a4 100644 --- a/rls/src/test/java/io/grpc/rls/RlsProtoConvertersTest.java +++ b/rls/src/test/java/io/grpc/rls/RlsProtoConvertersTest.java @@ -71,6 +71,65 @@ public void convert_toRequestObject() { assertThat(proto.getReason()).isEqualTo(RouteLookupRequest.Reason.REASON_MISS); } + @Test + public void convert_toRequestProto_staleHeaderData() { + Converter converter = + new RouteLookupRequestConverter(); + + // Non-null value + RouteLookupRequest protoWithStaleHeader = RouteLookupRequest.newBuilder() + .putKeyMap("key1", "val1") + .setStaleHeaderData("stale-header-v1") + .build(); + RlsProtoData.RouteLookupRequest objectWithStaleHeader = + converter.convert(protoWithStaleHeader); + assertThat(objectWithStaleHeader.staleHeaderData()).isEqualTo("stale-header-v1"); + + // Null value (unset) + RouteLookupRequest protoUnset = RouteLookupRequest.newBuilder() + .putKeyMap("key1", "val1") + .build(); + RlsProtoData.RouteLookupRequest objectUnset = converter.convert(protoUnset); + assertThat(objectUnset.staleHeaderData()).isNull(); + + // Empty string + RouteLookupRequest protoEmpty = RouteLookupRequest.newBuilder() + .putKeyMap("key1", "val1") + .setStaleHeaderData("") + .build(); + RlsProtoData.RouteLookupRequest objectEmpty = converter.convert(protoEmpty); + assertThat(objectEmpty.staleHeaderData()).isNull(); + } + + @Test + public void convert_toRequestObject_staleHeaderData() { + Converter converter = + new RouteLookupRequestConverter().reverse(); + + // Non-null value + RlsProtoData.RouteLookupRequest objectWithStaleHeader = + RlsProtoData.RouteLookupRequest.create( + ImmutableMap.of("key1", "val1"), + RlsProtoData.RouteLookupRequest.Reason.REASON_STALE, + "stale-header-v1"); + RouteLookupRequest protoWithStaleHeader = converter.convert(objectWithStaleHeader); + assertThat(protoWithStaleHeader.getStaleHeaderData()).isEqualTo("stale-header-v1"); + assertThat(protoWithStaleHeader.getReason()) + .isEqualTo(RouteLookupRequest.Reason.REASON_STALE); + + // Null value + RlsProtoData.RouteLookupRequest objectNull = + RlsProtoData.RouteLookupRequest.create( + ImmutableMap.of("key1", "val1"), + RlsProtoData.RouteLookupRequest.Reason.REASON_MISS, + null); + RouteLookupRequest protoNull = converter.convert(objectNull); + assertThat(protoNull.getStaleHeaderData()).isEmpty(); + assertThat(protoNull.getReason()) + .isEqualTo(RouteLookupRequest.Reason.REASON_MISS); + } + + @Test public void convert_toResponseProto() { Converter converter = diff --git a/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java b/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java index dffdf2c7476..ebab2eb4cc5 100644 --- a/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java +++ b/xds/src/main/java/io/grpc/xds/XdsServerWrapper.java @@ -118,6 +118,10 @@ public void uncaughtException(Thread t, Throwable e) { private final CountDownLatch internalTerminationLatch = new CountDownLatch(1); private final SettableFuture initialStartFuture = SettableFuture.create(); private boolean initialStarted; + // Must be accessed in syncContext. + // Guards the forceful-shutdown work in shutdownNow(), independently of the shutdown AtomicBoolean + // above, so it isn't skipped when shutdown() + private boolean shutdownNowed; private ScheduledHandle restartTimer; private ObjectPool xdsClientPool; private XdsClient xdsClient; @@ -408,16 +412,15 @@ public void run() { @Override public Server shutdownNow() { - if (!shutdown.compareAndSet(false, true)) { - return this; - } + shutdown(); syncContext.execute(new Runnable() { @Override public void run() { - if (!delegate.isShutdown()) { - delegate.shutdownNow(); + if (shutdownNowed) { + return; } - internalShutdown(); + shutdownNowed = true; + delegate.shutdownNow(); initialStartFuture.set(new IOException("server is forcefully shut down")); } }); diff --git a/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java b/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java index 47ac32cdc8a..e309df4f453 100644 --- a/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java +++ b/xds/src/test/java/io/grpc/xds/XdsServerWrapperTest.java @@ -487,6 +487,62 @@ public void run() { } } + @Test + public void shutdownNow_afterShutdown_stillUnblocksStartThread() throws Exception { + final SettableFuture start = SettableFuture.create(); + Executors.newSingleThreadExecutor() + .execute( + new Runnable() { + @Override + public void run() { + try { + start.set(xdsServerWrapper.start()); + } catch (Exception ex) { + start.setException(ex); + } + } + }); + assertThat(xdsClient.ldsResource.get(5, TimeUnit.SECONDS)) + .isEqualTo("grpc/server?udpa.resource.listening_address=0.0.0.0:1"); + xdsServerWrapper.shutdown(); + xdsServerWrapper.shutdownNow(); + try { + start.get(5, TimeUnit.SECONDS); + fail("should have thrown but not"); + } catch (ExecutionException ex) { + assertThat(ex).hasCauseThat().isInstanceOf(IOException.class); + assertThat(ex).hasCauseThat().hasMessageThat().isEqualTo("server is forcefully shut down"); + } + } + + @Test + public void shutdownNow_calledTwice_forcefullyShutsDownDelegateOnce() throws Exception { + final SettableFuture start = SettableFuture.create(); + Executors.newSingleThreadExecutor() + .execute( + new Runnable() { + @Override + public void run() { + try { + start.set(xdsServerWrapper.start()); + } catch (Exception ex) { + start.setException(ex); + } + } + }); + assertThat(xdsClient.ldsResource.get(5, TimeUnit.SECONDS)) + .isEqualTo("grpc/server?udpa.resource.listening_address=0.0.0.0:1"); + xdsServerWrapper.shutdownNow(); + xdsServerWrapper.shutdownNow(); + try { + start.get(5, TimeUnit.SECONDS); + fail("should have thrown but not"); + } catch (ExecutionException ex) { + assertThat(ex).hasCauseThat().isInstanceOf(IOException.class); + } + verify(mockServer, times(1)).shutdownNow(); + } + @Test public void initialStartIoException() throws Exception { final SettableFuture start = SettableFuture.create();