diff --git a/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java b/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java index d77e5afab7..4b084f780a 100644 --- a/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java +++ b/util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java @@ -19,8 +19,10 @@ import io.kubernetes.client.informer.ListerWatcher; import io.kubernetes.client.informer.exception.WatchExpiredException; import io.kubernetes.client.openapi.ApiException; +import io.kubernetes.client.openapi.JSON; import io.kubernetes.client.openapi.models.V1ListMeta; import io.kubernetes.client.openapi.models.V1ObjectMeta; +import io.kubernetes.client.openapi.models.V1Status; import io.kubernetes.client.util.CallGeneratorParams; import io.kubernetes.client.util.ModelMapper; import io.kubernetes.client.util.Strings; @@ -182,7 +184,7 @@ public void run() { } } } catch (ApiException e) { - if (e.getCode() == HttpURLConnection.HTTP_GONE) { + if (e.getCode() == HttpURLConnection.HTTP_GONE || isTooLargeResourceVersion(e)) { log.info( "ResourceVersion {} expired, will retry w/o resourceVersion at the next time", getRelistResourceVersion()); @@ -195,6 +197,24 @@ public void run() { } } + private static boolean isTooLargeResourceVersion(ApiException exception) { + if (Strings.isNullOrEmpty(exception.getResponseBody())) { + return false; + } + V1Status status; + try { + status = JSON.deserialize(exception.getResponseBody(), V1Status.class); + } catch (RuntimeException e) { + // Generated status adapters can throw unchecked exceptions for malformed response bodies. + return false; + } + return status != null + && status.getDetails() != null + && status.getDetails().getCauses() != null + && status.getDetails().getCauses().stream() + .anyMatch(cause -> cause != null && "ResourceVersionTooLarge".equals(cause.getReason())); + } + public void stop() { try { isActive.set(false); diff --git a/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java b/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java index d4c81a26ca..62ba4b6115 100644 --- a/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java +++ b/util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java @@ -43,6 +43,10 @@ import org.awaitility.Awaitility; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -212,6 +216,113 @@ void reflectorRunnableShouldCaptureListNon410ApiException() throws ApiException } } + @ParameterizedTest + @ValueSource(ints = {410, 504}) + void reflectorRecoversUnavailableResourceVersion(int statusCode) throws ApiException { + V1Pod initialPod = new V1Pod().metadata(new V1ObjectMeta().name("initial")); + V1Pod recoveredPod = new V1Pod().metadata(new V1ObjectMeta().name("recovered")); + ApiException unavailable = + statusCode == 410 + ? new ApiException(410, "Gone") + : new ApiException( + statusCode, + null, + "{\"kind\":\"Status\",\"apiVersion\":\"v1\",\"reason\":\"Timeout\"," + + "\"details\":{\"causes\":[{\"reason\":\"Other\"}," + + "{\"reason\":\"ResourceVersionTooLarge\"}]},\"code\":504}"); + when(listerWatcher.list(any())) + .thenReturn( + new V1PodList() + .metadata(new V1ListMeta().resourceVersion("100")) + .items(Arrays.asList(initialPod))) + .thenThrow(unavailable) + .thenReturn( + new V1PodList() + .metadata(new V1ListMeta().resourceVersion("200")) + .items(Arrays.asList(recoveredPod))); + MockWatch initialWatch = + new MockWatch<>(new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500))); + MockWatch recoveredWatch = + new MockWatch<>( + new Watch.Response<>( + EventType.BOOKMARK.name(), + new V1Pod().metadata(new V1ObjectMeta().resourceVersion("201"))), + new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500))); + when(listerWatcher.watch(any())).thenReturn(initialWatch, recoveredWatch); + ReflectorRunnable reflector = + new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO, exceptionHandler); + + reflector.run(); + assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100"); + reflector.run(); + assertThat(reflector.isLastSyncResourceVersionUnavailable()).isTrue(); + assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100"); + verify(exceptionHandler, never()).accept(V1Pod.class, unavailable); + reflector.run(); + + ArgumentCaptor listParams = + ArgumentCaptor.forClass(CallGeneratorParams.class); + verify(listerWatcher, times(3)).list(listParams.capture()); + assertThat(listParams.getAllValues()) + .extracting(params -> params.resourceVersion) + .containsExactly("0", "100", ""); + ArgumentCaptor watchParams = + ArgumentCaptor.forClass(CallGeneratorParams.class); + verify(listerWatcher, times(2)).watch(watchParams.capture()); + assertThat(watchParams.getAllValues()) + .extracting(params -> params.resourceVersion) + .containsExactly("100", "200"); + verify(deltaFIFO).replace(Arrays.asList(initialPod), "100"); + verify(deltaFIFO).replace(Arrays.asList(recoveredPod), "200"); + assertThat(reflector.isLastSyncResourceVersionUnavailable()).isFalse(); + assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("201"); + assertThat(initialWatch.isClosed()).isTrue(); + assertThat(recoveredWatch.isClosed()).isTrue(); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource( + strings = { + "not json", + "null", + "[]", + "{}", + "{\"details\":null}", + "{\"details\":{\"causes\":null}}", + "{\"details\":{\"causes\":[]}}", + "{\"details\":{\"causes\":[null,{}]}}", + "{\"details\":{\"causes\":\"invalid\"}}", + "{\"details\":{\"causes\":[{\"reason\":{}}]}}", + "{\"reason\":\"Timeout\",\"message\":\"Too large resource version\"}", + "{\"reason\":\"Timeout\",\"details\":{\"causes\":[{\"reason\":\"Other\"}]}}" + }) + void reflectorReportsOrdinaryTimeoutsAndMalformedBodies(String body) throws ApiException { + ApiException timeout = new ApiException(504, null, body); + when(listerWatcher.list(any())) + .thenReturn(new V1PodList().metadata(new V1ListMeta().resourceVersion("100"))) + .thenThrow(timeout); + when(listerWatcher.watch(any())) + .thenReturn( + new MockWatch<>( + new Watch.Response<>(EventType.ERROR.name(), new V1Status().code(500)))); + ReflectorRunnable reflector = + new ReflectorRunnable<>(V1Pod.class, listerWatcher, deltaFIFO, exceptionHandler); + + reflector.run(); + reflector.run(); + reflector.run(); + + ArgumentCaptor params = ArgumentCaptor.forClass(CallGeneratorParams.class); + verify(listerWatcher, times(3)).list(params.capture()); + assertThat(params.getAllValues()) + .extracting(value -> value.resourceVersion) + .containsExactly("0", "100", "100"); + verify(exceptionHandler, times(2)).accept(V1Pod.class, timeout); + assertThat(reflector.isLastSyncResourceVersionUnavailable()).isFalse(); + assertThat(reflector.getLastSyncResourceVersion()).isEqualTo("100"); + } + @Test void reflectorRunnableCaptureWatchException() throws ApiException { RuntimeException expectedException = new RuntimeException("noxu");