Skip to content

Add exponential backoff for informer watch reconnects #4895

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
brendandburns wants to merge 1 commit into
base: master
Choose a base branch
from
+196 −4
Open

Add exponential backoff for informer watch reconnects #4895

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
76 changes: 76 additions & 0 deletions e2e/src/test/java/io/kubernetes/client/e2e/informer/NamespaceInformerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,14 +15,25 @@
import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;

import io.kubernetes.client.informer.ListerWatcher;
import io.kubernetes.client.informer.ResourceEventHandler;
import io.kubernetes.client.informer.SharedIndexInformer;
import io.kubernetes.client.informer.SharedInformerFactory;
import io.kubernetes.client.informer.cache.Lister;
import io.kubernetes.client.openapi.ApiClient;
import io.kubernetes.client.openapi.ApiException;
import io.kubernetes.client.openapi.apis.CoreV1Api;
import io.kubernetes.client.openapi.models.V1Namespace;
import io.kubernetes.client.openapi.models.V1NamespaceList;
import io.kubernetes.client.openapi.models.V1ObjectMeta;
import io.kubernetes.client.util.CallGeneratorParams;
import io.kubernetes.client.util.ClientBuilder;
import io.kubernetes.client.util.Watchable;
import io.kubernetes.client.util.generic.GenericKubernetesApi;
import io.kubernetes.client.util.generic.options.ListOptions;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.jupiter.api.Test;

class NamespaceInformerTest {
Expand Down Expand Up @@ -57,4 +68,69 @@ void listWatchingNamespaces() throws Exception {
informerFactory.stopAllRegisteredInformers(true);
}
}

@Test
void listWatchingNamespacesRecoversFromInitialConnectExceptions() throws Exception {
ApiClient client = ClientBuilder.defaultClient();
CoreV1Api coreV1Api = new CoreV1Api(client);
SharedInformerFactory informerFactory = new SharedInformerFactory(client);
String namespaceName = "e2e-informer-retry";
AtomicInteger watchAttempts = new AtomicInteger(0);
GenericKubernetesApi<V1Namespace, V1NamespaceList> api =
new GenericKubernetesApi<>(V1Namespace.class, V1NamespaceList.class, "", "v1", "namespaces", client);

ListerWatcher<V1Namespace, V1NamespaceList> flakyWatcher =
new ListerWatcher<V1Namespace, V1NamespaceList>() {
@Override
public V1NamespaceList list(CallGeneratorParams params) {
return api
.list(
new ListOptions()
.resourceVersion(params.resourceVersion)
.timeoutSeconds(params.timeoutSeconds))
.getObject();
}

@Override
public Watchable<V1Namespace> watch(CallGeneratorParams params) throws ApiException {
if (watchAttempts.incrementAndGet() <= 2) {
throw new RuntimeException(new java.net.ConnectException("simulated transient failure"));
}
return api.watch(
new ListOptions()
.resourceVersion(params.resourceVersion)
.timeoutSeconds(params.timeoutSeconds));
}
};

SharedIndexInformer<V1Namespace> nsInformer =
informerFactory.sharedIndexInformerFor(flakyWatcher, V1Namespace.class, 0);
CountDownLatch selectedSeen = new CountDownLatch(1);
try {
nsInformer.addEventHandler(
new ResourceEventHandler<V1Namespace>() {
@Override
public void onAdd(V1Namespace obj) {
if (namespaceName.equals(obj.getMetadata().getName())) {
selectedSeen.countDown();
}
}

@Override
public void onUpdate(V1Namespace oldObj, V1Namespace newObj) {}

@Override
public void onDelete(V1Namespace obj, boolean deletedFinalStateUnknown) {}
});

informerFactory.startAllRegisteredInformers();
await().untilAsserted(() -> assertThat(nsInformer.hasSynced()).isTrue());
coreV1Api.createNamespace(new V1Namespace().metadata(new V1ObjectMeta().name(namespaceName))).execute();
assertThat(selectedSeen.await(45, TimeUnit.SECONDS)).isTrue();
assertThat(watchAttempts.get()).isGreaterThanOrEqualTo(3);
} finally {
informerFactory.stopAllRegisteredInformers(true);
coreV1Api.deleteNamespace(namespaceName).execute();
}
}
}
46 changes: 42 additions & 4 deletions util/src/main/java/io/kubernetes/client/informer/cache/ReflectorRunnable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.function.LongConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

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

private String lastSyncResourceVersion;

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

private Method setKindMethod;
private Method setApiVersionMethod;
private final LongConsumer connectExceptionSleeper;
private long watchRetryBackoffMillis;

public ReflectorRunnable(
Class<ApiType> apiTypeClass,
Expand All @@ -78,11 +83,22 @@ public ReflectorRunnable(
ListerWatcher<ApiType, ApiListType> listerWatcher,
DeltaFIFO store,
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
this(apiTypeClass, listerWatcher, store, exceptionHandler, ReflectorRunnable::sleep);
}

ReflectorRunnable(
Class<ApiType> apiTypeClass,
ListerWatcher<ApiType, ApiListType> listerWatcher,
DeltaFIFO store,
BiConsumer<Class<ApiType>, Throwable> exceptionHandler,
LongConsumer connectExceptionSleeper) {
this.listerWatcher = listerWatcher;
this.store = store;
this.apiTypeClass = apiTypeClass;
this.exceptionHandler =
exceptionHandler == null ? ReflectorRunnable::defaultWatchErrorHandler : exceptionHandler;
this.connectExceptionSleeper = connectExceptionSleeper;
this.watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS;
try {
this.setKindMethod = apiTypeClass.getMethod("setKind", String.class);
this.setApiVersionMethod = apiTypeClass.getMethod("setApiVersion", String.class);
Expand All @@ -97,6 +113,9 @@ public ReflectorRunnable(
*/
public void run() {
log.info("{}#Start listing and watching...", apiTypeClass);
// run() can be invoked multiple times for the same reflector instance; always restart backoff
// from the initial value for each list-watch cycle.
resetWatchRetryBackoff();

try {
ApiListType list =
Expand Down Expand Up @@ -148,6 +167,7 @@ public void run() {
watch = newWatch;
}
watchHandler(newWatch);
resetWatchRetryBackoff();
} catch (WatchExpiredException e) {
// Watch calls were failed due to expired resource-version. Returning
// to unwind the list-watch loops so that we can respawn a new round
Expand All @@ -161,10 +181,9 @@ public void run() {
// objects because most likely we will be able to restart watch where
// we ended. If that's the case wait and resend watch request.
log.info("{}#Watch get connect exception, retry watch", this.apiTypeClass);
try {
Thread.sleep(1000L);
} catch (InterruptedException e) {
// no-op
sleepForConnectExceptionRetry();
if (Thread.currentThread().isInterrupted()) {
return;
}
continue;
}
Expand Down Expand Up @@ -364,4 +383,23 @@ private boolean isConnectException(Throwable t) {
Throwable cause = t.getCause();
return cause instanceof ConnectException;
}

private void sleepForConnectExceptionRetry() {
long currentBackoffMillis = watchRetryBackoffMillis;
watchRetryBackoffMillis =
Math.min(watchRetryBackoffMillis * 2, WATCH_RETRY_MAX_BACKOFF_MILLIS);
connectExceptionSleeper.accept(currentBackoffMillis);
}

private void resetWatchRetryBackoff() {
watchRetryBackoffMillis = WATCH_RETRY_INITIAL_BACKOFF_MILLIS;
}

private static void sleep(long durationMillis) {
try {
Thread.sleep(durationMillis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
78 changes: 78 additions & 0 deletions util/src/test/java/io/kubernetes/client/informer/cache/ReflectorRunnableTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,11 +33,13 @@
import io.kubernetes.client.util.Watchable;
import java.net.HttpURLConnection;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import org.awaitility.Awaitility;
Expand Down Expand Up @@ -361,6 +363,82 @@ void reflectorListShouldHandleExpiredResourceVersionFromWatchHandler()
}
}

@Test
void reflectorWatchConnectExceptionShouldUseExponentialBackoff()
throws ApiException, InterruptedException {
List<Long> retryBackoffs = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(3);
when(listerWatcher.list(any()))
.thenReturn(new V1PodList().metadata(new V1ListMeta().resourceVersion("100")));
when(listerWatcher.watch(any())).thenThrow(new RuntimeException(new java.net.ConnectException("refused")));
ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
new ReflectorRunnable<>(
V1Pod.class,
listerWatcher,
deltaFIFO,
exceptionHandler,
backoff -> {
if (retryBackoffs.size() < 3) {
retryBackoffs.add(backoff);
latch.countDown();
}
});
try {
Thread thread = new Thread(reflectorRunnable::run);
thread.setDaemon(true);
thread.start();
assertThat(latch.await(5, TimeUnit.SECONDS)).isTrue();
} finally {
reflectorRunnable.stop();
}
assertThat(retryBackoffs).containsExactly(1000L, 2000L, 4000L);
}

@Test
void reflectorWatchBackoffShouldResetAfterSuccessfulWatch() {
List<Long> retryBackoffs = new ArrayList<>();
CountDownLatch latch = new CountDownLatch(2);
AtomicInteger watchCount = new AtomicInteger();

ReflectorRunnable<V1Pod, V1PodList> reflectorRunnable =
new ReflectorRunnable<>(
V1Pod.class,
new ListerWatcher<V1Pod, V1PodList>() {
@Override
public V1PodList list(CallGeneratorParams params) {
return new V1PodList().metadata(new V1ListMeta().resourceVersion("100"));
}

@Override
public Watchable<V1Pod> watch(CallGeneratorParams params) {
int call = watchCount.incrementAndGet();
if (call == 2) {
return new MockWatch<>();
}
throw new RuntimeException(new java.net.ConnectException("refused"));
}
},
deltaFIFO,
exceptionHandler,
backoff -> {
if (retryBackoffs.size() < 2) {
retryBackoffs.add(backoff);
latch.countDown();
}
});
try {
Thread thread = new Thread(reflectorRunnable::run);
thread.setDaemon(true);
thread.start();
assertThat(latch.await(2, TimeUnit.SECONDS)).isTrue();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
reflectorRunnable.stop();
}
assertThat(retryBackoffs).containsExactly(1000L, 1000L);
}

@Test
void defaultExceptionHandlerSetPerDefault() {
ReflectorRunnable<V1Pod, V1PodList> reflector =
Expand Down
Loading