Skip to content

Commit 9077a06

Browse files
brendandburnsbrendandburns
committed
Add exponential backoff for informer watch reconnects
Retry transient informer watch connection failures with bounded exponential backoff, interruption handling, and coverage.
1 parent 7601bd2 commit 9077a06

3 files changed

Lines changed: 196 additions & 4 deletions

File tree

‎e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,14 +15,25 @@
1515
import static org.assertj.core.api.Assertions.assertThat;
1616
import static org.awaitility.Awaitility.await;
1717

18+
import io.kubernetes.client.informer.ListerWatcher;
19+
import io.kubernetes.client.informer.ResourceEventHandler;
1820
import io.kubernetes.client.informer.SharedIndexInformer;
1921
import io.kubernetes.client.informer.SharedInformerFactory;
2022
import io.kubernetes.client.informer.cache.Lister;
2123
import io.kubernetes.client.openapi.ApiClient;
24+
import io.kubernetes.client.openapi.ApiException;
25+
import io.kubernetes.client.openapi.apis.CoreV1Api;
2226
import io.kubernetes.client.openapi.models.V1Namespace;
2327
import io.kubernetes.client.openapi.models.V1NamespaceList;
28+
import io.kubernetes.client.openapi.models.V1ObjectMeta;
29+
import io.kubernetes.client.util.CallGeneratorParams;
2430
import io.kubernetes.client.util.ClientBuilder;
31+
import io.kubernetes.client.util.Watchable;
2532
import io.kubernetes.client.util.generic.GenericKubernetesApi;
33+
import io.kubernetes.client.util.generic.options.ListOptions;
34+
import java.util.concurrent.CountDownLatch;
35+
import java.util.concurrent.TimeUnit;
36+
import java.util.concurrent.atomic.AtomicInteger;
2637
import org.junit.jupiter.api.Test;
2738

2839
class NamespaceInformerTest {
@@ -57,4 +68,69 @@ void listWatchingNamespaces() throws Exception {
5768
informerFactory.stopAllRegisteredInformers(true);
5869
}
5970
}
71+
72+
@Test
73+
void listWatchingNamespacesRecoversFromInitialConnectExceptions() throws Exception {
74+
ApiClient client = ClientBuilder.defaultClient();
75+
CoreV1Api coreV1Api = new CoreV1Api(client);
76+
SharedInformerFactory informerFactory = new SharedInformerFactory(client);
77+
String namespaceName = "e2e-informer-retry";
78+
AtomicInteger watchAttempts = new AtomicInteger(0);
79+
GenericKubernetesApi<V1Namespace, V1NamespaceList> api =
80+
new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);
81+
82+
ListerWatcher<V1Namespace, V1NamespaceList> flakyWatcher =
83+
new ListerWatcher<V1Namespace, V1NamespaceList>() {
84+
@Override
85+
public V1NamespaceList list(CallGeneratorParams params) {
86+
return api
87+
.list(
88+
new ListOptions()
89+
.resourceVersion(params.resourceVersion)
90+
.timeoutSeconds(params.timeoutSeconds))
91+
.getObject();
92+
}
93+
94+
@Override
95+
public Watchable<V1Namespace> watch(CallGeneratorParams params) throws ApiException {
96+
if (watchAttempts.incrementAndGet() <= 2) {
97+
throw new RuntimeException(new java.net.ConnectException("simulated transient failure"));
98+
}
99+
return api.watch(
100+
new ListOptions()
101+
.resourceVersion(params.resourceVersion)
102+
.timeoutSeconds(params.timeoutSeconds));
103+
}
104+
};
105+
106+
SharedIndexInformer<V1Namespace> nsInformer =
107+
informerFactory.sharedIndexInformerFor(flakyWatcher, V1Namespace.class, 0);
108+
CountDownLatch selectedSeen = new CountDownLatch(1);
109+
try {
110+
nsInformer.addEventHandler(
111+
new ResourceEventHandler<V1Namespace>() {
112+
@Override
113+
public void onAdd(V1Namespace obj) {
114+
if (namespaceName.equals(obj.getMetadata().getName())) {
115+
selectedSeen.countDown();
116+
}
117+
}
118+
119+
@Override
120+
public void onUpdate(V1Namespace oldObj, V1Namespace newObj) {}
121+
122+
@Override
123+
public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {}
124+
});
125+
126+
informerFactory.startAllRegisteredInformers();
127+
await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue());
128+
coreV1Api.createNamespace(new V1Namespace().metadata(new V1ObjectMeta().name(namespaceName))).execute();
129+
assertThat(selectedSeen.await(45, TimeUnit.SECONDS)).isTrue();
130+
assertThat(watchAttempts.get()).isGreaterThanOrEqualTo(3);
131+
} finally {
132+
informerFactory.stopAllRegisteredInformers(true);
133+
coreV1Api.deleteNamespace(namespaceName).execute();
134+
}
135+
}
60136
}

‎util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java‎

Lines changed: 42 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import java.util.Optional;
3535
import java.util.concurrent.atomic.AtomicBoolean;
3636
import java.util.function.BiConsumer;
37+
import java.util.function.LongConsumer;
3738
import org.slf4j.Logger;
3839
import org.slf4j.LoggerFactory;
3940

@@ -46,6 +47,8 @@ public class ReflectorRunnable<
4647
public static Duration REFLECTOR_WATCH_CLIENTSIDE_MAX_TIMEOUT = Duration.ofMinutes(5 * 2);
4748

4849
private static final Logger log = LoggerFactory.getLogger(ReflectorRunnable.class);
50+
private static final long WATCH_RETRY_INITIAL_BACKOFF_MILLIS = 1000L;
51+
private static final long WATCH_RETRY_MAX_BACKOFF_MILLIS = 30000L;
4952

5053
private String lastSyncResourceVersion;
5154

@@ -65,6 +68,8 @@ public class ReflectorRunnable<
6568

6669
private Method setKindMethod;
6770
private Method setApiVersionMethod;
71+
private final LongConsumer connectExceptionSleeper;
72+
private long watchRetryBackoffMillis;
6873

6974
public ReflectorRunnable(
7075
Class<ApiType> apiTypeClass,
@@ -78,11 +83,22 @@ public ReflectorRunnable(
7883
ListerWatcher<ApiType, ApiListType> listerWatcher,
7984
DeltaFIFO store,
8085
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
86+
this(apiTypeClass, listerWatcher, store, exceptionHandler, ReflectorRunnable::sleep);
87+
}
88+
89+
ReflectorRunnable(
90+
Class<ApiType> apiTypeClass,
91+
ListerWatcher<ApiType, ApiListType> listerWatcher,
92+
DeltaFIFO store,
93+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
94+
LongConsumer connectExceptionSleeper) {
8195
this.listerWatcher = listerWatcher;
8296
this.store = store;
8397
this.apiTypeClass = apiTypeClass;
8498
this.exceptionHandler =
8599
exceptionHandler == null ? ReflectorRunnable::defaultWatchErrorHandler : exceptionHandler;
100+
this.connectExceptionSleeper = connectExceptionSleeper;
101+
this.watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS;
86102
try {
87103
this.setKindMethod = apiTypeClass.getMethod("setKind", String.class);
88104
this.setApiVersionMethod = apiTypeClass.getMethod("setApiVersion", String.class);
@@ -97,6 +113,9 @@ public ReflectorRunnable(
97113
*/
98114
public void run() {
99115
log.info("{}#Start listing and watching...", apiTypeClass);
116+
// run() can be invoked multiple times for the same reflector instance; always restart backoff
117+
// from the initial value for each list-watch cycle.
118+
resetWatchRetryBackoff();
100119

101120
try {
102121
ApiListType list =
@@ -148,6 +167,7 @@ public void run() {
148167
watch = newWatch;
149168
}
150169
watchHandler(newWatch);
170+
resetWatchRetryBackoff();
151171
} catch (WatchExpiredException e) {
152172
// Watch calls were failed due to expired resource-version. Returning
153173
// to unwind the list-watch loops so that we can respawn a new round
@@ -161,10 +181,9 @@ public void run() {
161181
// objects because most likely we will be able to restart watch where
162182
// we ended. If that's the case wait and resend watch request.
163183
log.info("{}#Watch get connect exception, retry watch", this.apiTypeClass);
164-
try {
165-
Thread.sleep(1000L);
166-
} catch (InterruptedException e) {
167-
// no-op
184+
sleepForConnectExceptionRetry();
185+
if (Thread.currentThread().isInterrupted()) {
186+
return;
168187
}
169188
continue;
170189
}
@@ -364,4 +383,23 @@ private boolean isConnectException(Throwable t) {
364383
Throwable cause = t.getCause();
365384
return cause instanceof ConnectException;
366385
}
386+
387+
private void sleepForConnectExceptionRetry() {
388+
long currentBackoffMillis = watchRetryBackoffMillis;
389+
watchRetryBackoffMillis =
390+
Math.min(watchRetryBackoffMillis * 2, WATCH_RETRY_MAX_BACKOFF_MILLIS);
391+
connectExceptionSleeper.accept(currentBackoffMillis);
392+
}
393+
394+
private void resetWatchRetryBackoff() {
395+
watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS;
396+
}
397+
398+
private static void sleep(long durationMillis) {
399+
try {
400+
Thread.sleep(durationMillis);
401+
} catch (InterruptedException e) {
402+
Thread.currentThread().interrupt();
403+
}
404+
}
367405
}

‎util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,13 @@
3333
import io.kubernetes.client.util.Watchable;
3434
import java.net.HttpURLConnection;
3535
import java.time.Duration;
36+
import java.util.ArrayList;
3637
import java.util.Arrays;
3738
import java.util.List;
3839
import java.util.concurrent.CompletableFuture;
3940
import java.util.concurrent.CountDownLatch;
4041
import java.util.concurrent.TimeUnit;
42+
import java.util.concurrent.atomic.AtomicInteger;
4143
import java.util.concurrent.atomic.AtomicReference;
4244
import java.util.function.BiConsumer;
4345
import org.awaitility.Awaitility;
@@ -361,6 +363,82 @@ void reflectorListShouldHandleExpiredResourceVersionFromWatchHandler()
361363
}
362364
}
363365

366+
@Test
367+
void reflectorWatchConnectExceptionShouldUseExponentialBackoff()
368+
throws ApiException, InterruptedException {
369+
List<Long> retryBackoffs = new ArrayList<>();
370+
CountDownLatch latch = new CountDownLatch(3);
371+
when(listerWatcher.list(any()))
372+
.thenReturn(new V1PodList().metadata(new V1ListMeta().resourceVersion("100")));
373+
when(listerWatcher.watch(any())).thenThrow(new RuntimeException(new java.net.ConnectException("refused")));
374+
ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
375+
new ReflectorRunnable<>(
376+
V1Pod.class,
377+
listerWatcher,
378+
deltaFIFO,
379+
exceptionHandler,
380+
backoff -> {
381+
if (retryBackoffs.size() < 3) {
382+
retryBackoffs.add(backoff);
383+
latch.countDown();
384+
}
385+
});
386+
try {
387+
Thread thread = new Thread(reflectorRunnable::run);
388+
thread.setDaemon(true);
389+
thread.start();
390+
assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
391+
} finally {
392+
reflectorRunnable.stop();
393+
}
394+
assertThat(retryBackoffs).containsExactly(1000L, 2000L, 4000L);
395+
}
396+
397+
@Test
398+
void reflectorWatchBackoffShouldResetAfterSuccessfulWatch() {
399+
List<Long> retryBackoffs = new ArrayList<>();
400+
CountDownLatch latch = new CountDownLatch(2);
401+
AtomicInteger watchCount = new AtomicInteger();
402+
403+
ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
404+
new ReflectorRunnable<>(
405+
V1Pod.class,
406+
new ListerWatcher<V1Pod, V1PodList>() {
407+
@Override
408+
public V1PodList list(CallGeneratorParams params) {
409+
return new V1PodList().metadata(new V1ListMeta().resourceVersion("100"));
410+
}
411+
412+
@Override
413+
public Watchable<V1Pod> watch(CallGeneratorParams params) {
414+
int call = watchCount.incrementAndGet();
415+
if (call == 2) {
416+
return new MockWatch<>();
417+
}
418+
throw new RuntimeException(new java.net.ConnectException("refused"));
419+
}
420+
},
421+
deltaFIFO,
422+
exceptionHandler,
423+
backoff -> {
424+
if (retryBackoffs.size() < 2) {
425+
retryBackoffs.add(backoff);
426+
latch.countDown();
427+
}
428+
});
429+
try {
430+
Thread thread = new Thread(reflectorRunnable::run);
431+
thread.setDaemon(true);
432+
thread.start();
433+
assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue();
434+
} catch (InterruptedException e) {
435+
Thread.currentThread().interrupt();
436+
} finally {
437+
reflectorRunnable.stop();
438+
}
439+
assertThat(retryBackoffs).containsExactly(1000L, 1000L);
440+
}
441+
364442
@Test
365443
void defaultExceptionHandlerSetPerDefault() {
366444
ReflectorRunnable<V1Pod, V1PodList> reflector =

0 commit comments

Comments
 (0)