diff --git a/extensions-core/kubernetes-extensions/pom.xml b/extensions-core/kubernetes-extensions/pom.xml index 7decd1a881f7..b1340affb034 100644 --- a/extensions-core/kubernetes-extensions/pom.xml +++ b/extensions-core/kubernetes-extensions/pom.xml @@ -111,6 +111,12 @@ easymock test + + com.squareup.okhttp3 + mockwebserver3 + ${okhttp.version} + test + diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java new file mode 100644 index 000000000000..345ec04fed92 --- /dev/null +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/ChannelResetException.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.k8s.discovery; + +import java.io.IOException; + +public class ChannelResetException extends IOException +{ + public ChannelResetException(Throwable ex) + { + super(ex); + } +} diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java index b7992a3fb2ad..95c8dfd4154b 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/DefaultK8sApiClient.java @@ -40,6 +40,7 @@ import org.apache.druid.java.util.common.logger.Logger; import java.io.IOException; +import java.io.InterruptedIOException; import java.net.SocketTimeoutException; import java.util.HashMap; import java.util.List; @@ -194,7 +195,7 @@ public WatchResult watchPods(String namespace, String labelSelector, String last private Watch.Response obj; @Override - public boolean hasNext() throws SocketTimeoutException + public boolean hasNext() throws IOException { try { while (watch.hasNext()) { @@ -263,8 +264,14 @@ public boolean hasNext() throws SocketTimeoutException return true; } else if (item != null && item.type != null && item.type.equals(WatchResult.BOOKMARK)) { // Events with type BOOKMARK will only contain resourceVersion and no metadata. See - // Kubernetes API documentation for details. - LOGGER.debug("BOOKMARK event fired, no nothing, only update resourceVersion"); + // Kubernetes API documentation for details. Build a response carrying just the + //updated resourceVersion; otherwise obj would still hold the previous event (or + // null on the first event), corrupting the caller's resourceVersion tracking. + LOGGER.debug("BOOKMARK event fired, only updating resourceVersion"); + obj = new Watch.Response<>( + WatchResult.BOOKMARK, + new DiscoveryDruidNodeAndResourceVersion( + item.object.getMetadata().getResourceVersion(), null)); return true; } else { LOGGER.error("WTH! item or item.type is NULL"); @@ -274,9 +281,11 @@ public boolean hasNext() throws SocketTimeoutException catch (RuntimeException ex) { if (ex.getCause() instanceof SocketTimeoutException) { throw (SocketTimeoutException) ex.getCause(); - } else { - throw ex; } + if (ex.getCause() instanceof IOException && !(ex.getCause() instanceof InterruptedIOException)) { + throw new ChannelResetException(ex.getCause()); + } + throw ex; } return false; diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java index 8dfc595db742..522b7aa42417 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProvider.java @@ -254,50 +254,51 @@ private void watch() private void keepWatching(String labelSelector, String resourceVersion) { String nextResourceVersion = resourceVersion; - while (lifecycleLock.isStarted()) { - try { - WatchResult iter = - k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole); + while (lifecycleLock.isStarted()) { + try (WatchResult iter = k8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, nextResourceVersion, nodeRole)) { if (iter == null) { // history not available, we need to start from scratch return; } - try { - while (iter.hasNext()) { - Watch.Response item = iter.next(); - if (item != null && item.type != null && item.object != null) { - switch (item.type) { - case WatchResult.ADDED: - baseNodeRoleWatcher.childAdded(item.object.getNode()); - break; - case WatchResult.DELETED: - case WatchResult.NOT_READY: - // Use skipIfUnknown=true for all k8s discovery removals. - // DELETED can fire after NOT_READY (so the service is already removed), or before ADDED (pod deleted before becoming ready). - // NOT_READY can repeat during CrashLoopBackOff. None of these warrant the error-level logging that - // comes with trying to remove an unknown service. - baseNodeRoleWatcher.childRemoved(item.object.getNode(), true); - break; - default: - } - - // This should be updated after the action has been dealt with successfully - nextResourceVersion = item.object.getResourceVersion(); - - } else { - // Try again by starting the watch from the beginning. This can happen if the - // watch goes bad. - LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole); - return; - } + while (iter.hasNext()) { + Watch.Response item = iter.next(); + + if (item == null || item.type == null || item.object == null) { + LOGGER.debug("Received NULL item while watching role[%s]. Restarting watch.", this.nodeRole); + return; } - } - finally { - iter.close(); + + switch (item.type) { + case WatchResult.ADDED: + baseNodeRoleWatcher.childAdded(item.object.getNode()); + break; + case WatchResult.DELETED: + case WatchResult.NOT_READY: + // Use skipIfUnknown=true for all k8s discovery removals. + // DELETED can fire after NOT_READY (so the service is already removed), or before ADDED (pod deleted before becoming ready). + // NOT_READY can repeat during CrashLoopBackOff. None of these warrant the error-level logging that + // comes with trying to remove an unknown service. + baseNodeRoleWatcher.childRemoved(item.object.getNode(), true); + break; + default: + } + + // This should be updated after the action has been dealt with successfully + nextResourceVersion = item.object.getResourceVersion(); } + LOGGER.trace("Watch closed normally for role[%s]", this.nodeRole); + return; + } + catch (ChannelResetException ex) { + LOGGER.debug("Watch stream terminated normally for role[%s], restarting", this.nodeRole); + // Apply the same backoff as other retry paths below: a persistently reset stream + // (flaky proxy, LB idle timeout shorter than expected) would otherwise cause an + // unbounded relist/watch loop against the API server. + sleep(watcherErrorRetryWaitMS); + return; } catch (SocketTimeoutException ex) { // socket read timeout can happen normally due to k8s not having anything new to push leading to socket diff --git a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java index 2b3ef7b584a9..37cac3768c56 100644 --- a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java +++ b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/WatchResult.java @@ -21,7 +21,7 @@ import io.kubernetes.client.util.Watch; -import java.net.SocketTimeoutException; +import java.io.IOException; /** * Iterator over k8s pod watch events that is aligned with the needs of Druid service discovery rather than @@ -29,7 +29,7 @@ * a Kubernetes MODIFIED event for a pod whose containers are no longer ready (according to k8s readiness state) is surfaced as the * synthetic {@link #NOT_READY} type so that consumers can handle it as a removal from k8s service discovery. */ -public interface WatchResult +public interface WatchResult extends AutoCloseable { String ADDED = "ADDED"; String MODIFIED = "MODIFIED"; @@ -44,9 +44,7 @@ public interface WatchResult */ String NOT_READY = "NOT_READY"; - boolean hasNext() throws SocketTimeoutException; + boolean hasNext() throws IOException; Watch.Response next(); - - void close(); } diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java new file mode 100644 index 000000000000..33a73d972e32 --- /dev/null +++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/DefaultK8sApiClientWatchPodsTest.java @@ -0,0 +1,164 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.druid.k8s.discovery; + +import com.fasterxml.jackson.databind.InjectableValues; +import com.fasterxml.jackson.databind.ObjectMapper; +import io.kubernetes.client.openapi.ApiClient; +import io.kubernetes.client.util.Watch; +import mockwebserver3.MockResponse; +import mockwebserver3.MockWebServer; +import mockwebserver3.SocketEffect; +import okhttp3.Headers; +import org.apache.druid.discovery.DiscoveryDruidNode; +import org.apache.druid.discovery.NodeRole; +import org.apache.druid.jackson.DefaultObjectMapper; +import org.apache.druid.java.util.common.StringUtils; +import org.apache.druid.server.DruidNode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; + +/** + * Exercises {@link DefaultK8sApiClient#watchPods} end to end against a real (but local, canned) + * HTTP watch stream: BOOKMARK events must surface their resourceVersion to the caller instead of + * a stale or null response, and a mid-stream connection reset must surface as + * {@link ChannelResetException} rather than an unrelated exception type. + */ +public class DefaultK8sApiClientWatchPodsTest +{ + private static final ObjectMapper JSON_MAPPER = new DefaultObjectMapper(); + + static { + // DiscoveryDruidNode's @JsonCreator takes a @JacksonInject ObjectMapper, so the mapper + // used to deserialize it must know how to inject itself. + JSON_MAPPER.setInjectableValues(new InjectableValues.Std().addValue(ObjectMapper.class, JSON_MAPPER)); + } + + private final DiscoveryDruidNode testNode = new DiscoveryDruidNode( + new DruidNode("druid/router", "test-host", true, 80, null, true, false), NodeRole.ROUTER, null); + + private MockWebServer server; + private DefaultK8sApiClient client; + + @BeforeEach + public void setUp() throws IOException + { + server = new MockWebServer(); + server.start(); + + ApiClient apiClient = new ApiClient().setBasePath(server.url("/").toString()); + client = new DefaultK8sApiClient(apiClient, JSON_MAPPER); + } + + @AfterEach + public void tearDown() throws IOException + { + server.close(); + } + + @Test + public void testBookmarkAsFirstEventSurfacesResourceVersion() throws Exception + { + enqueueWatchStream(bookmarkLine("100")); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertTrue(result.hasNext()); + Watch.Response response = result.next(); + + Assertions.assertNotNull(response, "BOOKMARK event must not surface as a null response"); + Assertions.assertEquals(WatchResult.BOOKMARK, response.type); + Assertions.assertNotNull(response.object); + Assertions.assertEquals("100", response.object.getResourceVersion()); + } + } + + @Test + public void testBookmarkAfterRealEventSurfacesItsOwnResourceVersion() throws Exception + { + enqueueWatchStream(addedLine("200", testNode) + bookmarkLine("201")); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertTrue(result.hasNext()); + Watch.Response added = result.next(); + Assertions.assertEquals(WatchResult.ADDED, added.type); + Assertions.assertEquals("200", added.object.getResourceVersion()); + + Assertions.assertTrue(result.hasNext()); + Watch.Response bookmark = result.next(); + + Assertions.assertNotNull(bookmark, "BOOKMARK event must not surface as a null response"); + Assertions.assertEquals(WatchResult.BOOKMARK, bookmark.type); + Assertions.assertNotNull(bookmark.object); + Assertions.assertEquals( + "201", + bookmark.object.getResourceVersion(), + "BOOKMARK must carry its own resourceVersion, not the preceding event's"); + } + } + + @Test + public void testConnectionResetMidStreamSurfacesAsChannelResetException() throws Exception + { + // Truncate the body mid-object and sever the connection while it's being read, simulating + // a server-side stream reset. okhttp surfaces this as a ProtocolException, which + // DefaultK8sApiClient must remap to ChannelResetException rather than an unrelated + // RuntimeException, so callers can distinguish it from other IO failures. + server.enqueue(new MockResponse.Builder() + .code(200) + .headers(new Headers.Builder() + .add("Content-Type", "application/json") + .build()) + .body(addedLine("200", testNode).substring(0, 20)) + .onResponseBody(new SocketEffect.CloseSocket(true, true, true)) + .build()); + + try (WatchResult result = client.watchPods("testns", null, "99", NodeRole.ROUTER)) { + Assertions.assertThrows(ChannelResetException.class, result::hasNext); + } + } + + private void enqueueWatchStream(String body) + { + server.enqueue(new MockResponse( + 200, + new Headers.Builder().add("Content-Type", "application/json").build(), + body)); + } + + private static String bookmarkLine(String resourceVersion) + { + return "{\"type\":\"BOOKMARK\",\"object\":{\"kind\":\"Pod\",\"apiVersion\":\"v1\",\"metadata\":{\"resourceVersion\":\"" + + resourceVersion + "\"}}}\n"; + } + + private static String addedLine(String resourceVersion, DiscoveryDruidNode node) throws Exception + { + String infoAnnotation = K8sDruidNodeAnnouncer.getInfoAnnotation(node.getNodeRole()); + String nodeJson = StringUtils.replace(JSON_MAPPER.writeValueAsString(node), "\"", "\\\""); + return "{\"type\":\"ADDED\",\"object\":{\"kind\":\"Pod\",\"apiVersion\":\"v1\",\"metadata\":{" + + "\"name\":\"test-pod\",\"resourceVersion\":\"" + resourceVersion + "\"," + + "\"annotations\":{\"" + infoAnnotation + "\":\"" + nodeJson + "\"}}," + + "\"status\":{\"containerStatuses\":[{\"name\":\"main\",\"ready\":true}]}}}\n"; + } +} diff --git a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java index 6417edb4b275..445247575fa8 100644 --- a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java +++ b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/discovery/K8sDruidNodeDiscoveryProviderTest.java @@ -34,6 +34,7 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; +import java.io.IOException; import java.net.SocketTimeoutException; import java.util.ArrayList; import java.util.Collection; @@ -486,6 +487,122 @@ public void testStopDoesNotWaitForErrorRetryWhenWatchIsInterrupted() throws Exce ); } + @Test + @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS) + public void testNodeRoleWatcherHandlesChannelResetException() throws Exception + { + String labelSelector = "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true"; + K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class); + + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn( + new DiscoveryDruidNodeList( + "v1", + ImmutableMap.of( + testNode1.getDruidNode().getHostAndPortToUse(), testNode1, + testNode2.getDruidNode().getHostAndPortToUse(), testNode2 + ) + ) + ); + + EasyMock.expect(mockK8sApiClient.watchPods( + podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)).andReturn( + new MockWatchResult(Collections.emptyList(), false, false, true) + ); + + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)).andReturn( + new DiscoveryDruidNodeList( + "v2", + ImmutableMap.of( + testNode2.getDruidNode().getHostAndPortToUse(), testNode2, + testNode3.getDruidNode().getHostAndPortToUse(), testNode3 + ) + ) + ); + + EasyMock.expect(mockK8sApiClient.watchPods( + podInfo.getPodNamespace(), labelSelector, "v2", NodeRole.ROUTER)).andReturn( + new MockWatchResult(Collections.emptyList(), false, false) + ); + + EasyMock.replay(mockK8sApiClient); + + K8sDruidNodeDiscoveryProvider discoveryProvider = new K8sDruidNodeDiscoveryProvider( + podInfo, + discoveryConfig, + mockK8sApiClient, + 1 + ); + discoveryProvider.start(); + + K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = + discoveryProvider.getForNodeRole(NodeRole.ROUTER, false); + + MockListener testListener = new MockListener( + ImmutableList.of( + MockListener.Event.added(testNode1), + MockListener.Event.added(testNode2), + MockListener.Event.inited(), + MockListener.Event.added(testNode3), + MockListener.Event.deleted(testNode1) + ) + ); + nodeDiscovery.registerListener(testListener); + nodeDiscovery.start(); + testListener.assertSuccess(); + discoveryProvider.stop(); + } + + @Test + @Timeout(value = 10_000, unit = TimeUnit.MILLISECONDS) + public void testPersistentChannelResetAppliesBackoffBetweenRelists() throws Exception + { + String labelSelector = + "druidDiscoveryAnnouncement-cluster-identifier=druid-cluster,druidDiscoveryAnnouncement-router=true"; + K8sApiClient mockK8sApiClient = EasyMock.createMock(K8sApiClient.class); + + // Three consecutive watch attempts are immediately reset, followed by a clean one. Without + // backoff on the ChannelResetException path, these three relist/watch cycles would complete + // almost instantly; with backoff, each reset costs watcherErrorRetryWaitMS. + for (int i = 0; i < 3; i++) { + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)) + .andReturn(new DiscoveryDruidNodeList("v1", ImmutableMap.of())); + EasyMock.expect(mockK8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, "v1", NodeRole.ROUTER)) + .andReturn(new MockWatchResult(Collections.emptyList(), false, false, true)); + } + + EasyMock.expect(mockK8sApiClient.listPods(podInfo.getPodNamespace(), labelSelector, NodeRole.ROUTER)) + .andReturn(new DiscoveryDruidNodeList( + "v2", ImmutableMap.of(testNode1.getDruidNode().getHostAndPortToUse(), testNode1))); + EasyMock.expect(mockK8sApiClient.watchPods(podInfo.getPodNamespace(), labelSelector, "v2", NodeRole.ROUTER)) + .andReturn(new MockWatchResult(Collections.emptyList(), false, false)); + + EasyMock.replay(mockK8sApiClient); + + long watcherErrorRetryWaitMS = 1500; + K8sDruidNodeDiscoveryProvider discoveryProvider = + new K8sDruidNodeDiscoveryProvider(podInfo, discoveryConfig, mockK8sApiClient, watcherErrorRetryWaitMS); + discoveryProvider.start(); + + K8sDruidNodeDiscoveryProvider.NodeRoleWatcher nodeDiscovery = + discoveryProvider.getForNodeRole(NodeRole.ROUTER, false); + + MockListener testListener = + new MockListener(ImmutableList.of(MockListener.Event.inited(), MockListener.Event.added(testNode1))); + nodeDiscovery.registerListener(testListener); + + long start = System.currentTimeMillis(); + nodeDiscovery.start(); + testListener.assertSuccess(); + long elapsed = System.currentTimeMillis() - start; + + discoveryProvider.stop(); + + Assertions.assertTrue( + elapsed >= 3 * watcherErrorRetryWaitMS, + "Expected at least 3 backoff waits (" + (3 * watcherErrorRetryWaitMS) + "ms) for 3 persistent " + + "channel resets, but only " + elapsed + "ms elapsed"); + } + private static class MockListener implements DruidNodeDiscovery.Listener { List events; @@ -654,6 +771,7 @@ private static class MockWatchResult implements WatchResult private volatile boolean timeoutOnStart; private volatile boolean timeooutOnEmptyResults; + private volatile boolean channelResetOnStart; private volatile boolean closeCalled = false; public MockWatchResult( @@ -667,13 +785,28 @@ public MockWatchResult( this.timeooutOnEmptyResults = timeooutOnEmptyResults; } + public MockWatchResult( + List> results, + boolean timeoutOnStart, + boolean timeooutOnEmptyResults, + boolean channelResetOnStart + ) + { + this(results, timeoutOnStart, timeooutOnEmptyResults); + this.channelResetOnStart = channelResetOnStart; + } + @Override - public boolean hasNext() throws SocketTimeoutException + public boolean hasNext() throws IOException { if (timeoutOnStart) { throw new SocketTimeoutException("testing timeout on start!!!"); } + if (channelResetOnStart) { + throw new ChannelResetException(new RuntimeException("simulated StreamResetException")); + } + if (results.isEmpty()) { if (timeooutOnEmptyResults) { throw new SocketTimeoutException("testing timeout on end!!!");