Skip to content

Commit 66aa670

Browse files
cbuschkacbuschka
committed
#1629 Added injection of exception handler into shared informer.
1 parent 5a8c753 commit 66aa670

File tree

8 files changed

+556
-352
lines changed

8 files changed

+556
-352
lines changed

‎util/src/main/java/io/kubernetes/client/informer/SharedInformerFactory.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import com.google.gson.reflect.TypeToken;
1616
import io.kubernetes.client.common.KubernetesListObject;
1717
import io.kubernetes.client.common.KubernetesObject;
18+
import io.kubernetes.client.informer.cache.Cache;
1819
import io.kubernetes.client.informer.impl.DefaultSharedIndexInformer;
1920
import io.kubernetes.client.openapi.ApiClient;
2021
import io.kubernetes.client.openapi.ApiException;
@@ -32,6 +33,7 @@
3233
import java.util.concurrent.ExecutorService;
3334
import java.util.concurrent.Executors;
3435
import java.util.concurrent.Future;
36+
import java.util.function.BiConsumer;
3537
import okhttp3.Call;
3638
import org.apache.commons.collections4.MapUtils;
3739

@@ -140,8 +142,19 @@ SharedIndexInformer<ApiType> sharedIndexInformerFor(
140142
ListerWatcher<ApiType, ApiListType> listerWatcher,
141143
Class<ApiType> apiTypeClass,
142144
long resyncPeriodInMillis) {
145+
return sharedIndexInformerFor(listerWatcher, apiTypeClass, resyncPeriodInMillis, null);
146+
}
147+
148+
public synchronized <ApiType extends KubernetesObject, ApiListType extends KubernetesListObject>
149+
SharedIndexInformer<ApiType> sharedIndexInformerFor(
150+
ListerWatcher<ApiType, ApiListType> listerWatcher,
151+
Class<ApiType> apiTypeClass,
152+
long resyncPeriodInMillis,
153+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
154+
143155
SharedIndexInformer<ApiType> informer =
144-
new DefaultSharedIndexInformer<>(apiTypeClass, listerWatcher, resyncPeriodInMillis);
156+
new DefaultSharedIndexInformer<>(
157+
apiTypeClass, listerWatcher, resyncPeriodInMillis, new Cache<>(), exceptionHandler);
145158
this.informers.putIfAbsent(TypeToken.get(apiTypeClass).getType(), informer);
146159
return informer;
147160
}

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

Lines changed: 21 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import java.util.concurrent.ScheduledExecutorService;
2424
import java.util.concurrent.ScheduledFuture;
2525
import java.util.concurrent.TimeUnit;
26+
import java.util.function.BiConsumer;
2627
import java.util.function.Consumer;
2728
import java.util.function.Supplier;
2829
import org.apache.commons.lang3.tuple.MutablePair;
@@ -65,19 +66,34 @@ public class Controller<
6566

6667
private ScheduledFuture reflectorFuture;
6768

69+
/* visible for testing */ BiConsumer<Class<ApiType>, Throwable> exceptionHandler;
70+
6871
public Controller(
6972
Class<ApiType> apiTypeClass,
7073
DeltaFIFO queue,
7174
ListerWatcher<ApiType, ApiListType> listerWatcher,
7275
Consumer<Deque<MutablePair<DeltaFIFO.DeltaType, KubernetesObject>>> processFunc,
7376
Supplier<Boolean> resyncFunc,
7477
long fullResyncPeriod) {
78+
this(apiTypeClass, queue, listerWatcher, processFunc, resyncFunc, fullResyncPeriod, null);
79+
}
80+
81+
public Controller(
82+
Class<ApiType> apiTypeClass,
83+
DeltaFIFO queue,
84+
ListerWatcher<ApiType, ApiListType> listerWatcher,
85+
Consumer<Deque<MutablePair<DeltaFIFO.DeltaType, KubernetesObject>>> processFunc,
86+
Supplier<Boolean> resyncFunc,
87+
long fullResyncPeriod,
88+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
89+
7590
this.queue = queue;
7691
this.listerWatcher = listerWatcher;
7792
this.apiTypeClass = apiTypeClass;
7893
this.processFunc = processFunc;
7994
this.resyncFunc = resyncFunc;
8095
this.fullResyncPeriod = fullResyncPeriod;
96+
this.exceptionHandler = exceptionHandler;
8197

8298
// starts one daemon thread for reflector
8399
this.reflectExecutor =
@@ -113,7 +129,7 @@ public void run() {
113129

114130
synchronized (this) {
115131
// TODO(yue9944882): proper naming for reflector
116-
reflector = new ReflectorRunnable<ApiType, ApiListType>(apiTypeClass, listerWatcher, queue);
132+
reflector = newReflector();
117133
try {
118134
reflectorFuture =
119135
reflectExecutor.scheduleWithFixedDelay(
@@ -130,6 +146,10 @@ public void run() {
130146
this.processLoop();
131147
}
132148

149+
/* visible for testing */ ReflectorRunnable<ApiType, ApiListType> newReflector() {
150+
return new ReflectorRunnable<>(apiTypeClass, listerWatcher, queue, exceptionHandler);
151+
}
152+
133153
/** stops the resync thread pool firstly, then stop the reflector */
134154
public void stop() {
135155
synchronized (this) {

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

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -58,11 +58,11 @@ public class ReflectorRunnable<
5858

5959
private AtomicBoolean isActive = new AtomicBoolean(true);
6060

61-
private final BiConsumer<Class<ApiType>, Throwable> exceptionHandler;
61+
/* visible for testing */ final BiConsumer<Class<ApiType>, Throwable> exceptionHandler;
6262

6363
public ReflectorRunnable(
6464
Class<ApiType> apiTypeClass, ListerWatcher listerWatcher, DeltaFIFO store) {
65-
this(apiTypeClass, listerWatcher, store, ReflectorRunnable::defaultWatchErrorHandler);
65+
this(apiTypeClass, listerWatcher, store, null);
6666
}
6767

6868
public ReflectorRunnable(
@@ -73,7 +73,8 @@ public ReflectorRunnable(
7373
this.listerWatcher = listerWatcher;
7474
this.store = store;
7575
this.apiTypeClass = apiTypeClass;
76-
this.exceptionHandler = exceptionHandler;
76+
this.exceptionHandler =
77+
exceptionHandler == null ? ReflectorRunnable::defaultWatchErrorHandler : exceptionHandler;
7778
}
7879

7980
/**
@@ -277,7 +278,7 @@ private void watchHandler(Watchable<ApiType> watch) {
277278
}
278279
}
279280

280-
private static <ApiType extends KubernetesObject> void defaultWatchErrorHandler(
281+
static <ApiType extends KubernetesObject> void defaultWatchErrorHandler(
281282
Class<ApiType> watchingApiTypeClass, Throwable t) {
282283
log.error(String.format("%s#Reflector loop failed unexpectedly", watchingApiTypeClass), t);
283284
}

‎util/src/main/java/io/kubernetes/client/informer/impl/DefaultSharedIndexInformer.java‎

Lines changed: 37 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import java.util.Deque;
2727
import java.util.List;
2828
import java.util.Map;
29+
import java.util.function.BiConsumer;
2930
import java.util.function.Function;
3031
import org.apache.commons.collections4.CollectionUtils;
3132
import org.apache.commons.lang3.tuple.MutablePair;
@@ -82,7 +83,28 @@ public DefaultSharedIndexInformer(
8283
new DeltaFIFO(
8384
(Function<KubernetesObject, String>) cache.getKeyFunc(),
8485
(Cache<KubernetesObject>) cache),
85-
cache);
86+
cache,
87+
null);
88+
}
89+
90+
public DefaultSharedIndexInformer(
91+
Class<ApiType> apiTypeClass,
92+
ListerWatcher<ApiType, ApiListType> listerWatcher,
93+
long resyncPeriod,
94+
Cache<ApiType> cache,
95+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
96+
this(
97+
apiTypeClass,
98+
listerWatcher,
99+
resyncPeriod,
100+
// down-casting should be safe here because one delta FIFO instance only serves one
101+
// resource
102+
// type
103+
new DeltaFIFO(
104+
(Function<KubernetesObject, String>) cache.getKeyFunc(),
105+
(Cache<KubernetesObject>) cache),
106+
cache,
107+
exceptionHandler);
86108
}
87109

88110
public DefaultSharedIndexInformer(
@@ -91,19 +113,31 @@ public DefaultSharedIndexInformer(
91113
long resyncPeriod,
92114
DeltaFIFO deltaFIFO,
93115
Indexer<ApiType> indexer) {
116+
this(apiTypeClass, listerWatcher, resyncPeriod, deltaFIFO, indexer, null);
117+
}
118+
119+
public DefaultSharedIndexInformer(
120+
Class<ApiType> apiTypeClass,
121+
ListerWatcher<ApiType, ApiListType> listerWatcher,
122+
long resyncPeriod,
123+
DeltaFIFO deltaFIFO,
124+
Indexer<ApiType> indexer,
125+
BiConsumer<Class<ApiType>, Throwable> exceptionHandler) {
126+
94127
this.resyncCheckPeriodMillis = resyncPeriod;
95128
this.defaultEventHandlerResyncPeriod = resyncPeriod;
96129

97130
this.processor = new SharedProcessor<>();
98131
this.indexer = indexer;
99132
this.controller =
100-
new Controller<ApiType, ApiListType>(
133+
new Controller<>(
101134
apiTypeClass,
102135
deltaFIFO,
103136
listerWatcher,
104137
this::handleDeltas,
105138
processor::shouldResync,
106-
resyncCheckPeriodMillis);
139+
resyncCheckPeriodMillis,
140+
exceptionHandler);
107141

108142
controllerThread =
109143
new Thread(controller::run, "informer-controller-" + apiTypeClass.getSimpleName());

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

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -14,6 +14,7 @@
1414

1515
import static org.junit.Assert.*;
1616

17+
import io.kubernetes.client.common.KubernetesObject;
1718
import io.kubernetes.client.informer.EventType;
1819
import io.kubernetes.client.informer.ListerWatcher;
1920
import io.kubernetes.client.openapi.models.V1ListMeta;
@@ -23,13 +24,36 @@
2324
import io.kubernetes.client.util.Watch;
2425
import java.time.Duration;
2526
import java.util.Arrays;
27+
import java.util.Deque;
2628
import java.util.concurrent.atomic.AtomicInteger;
29+
import java.util.function.BiConsumer;
30+
import java.util.function.Consumer;
31+
import java.util.function.Supplier;
32+
import org.apache.commons.lang3.tuple.MutablePair;
2733
import org.awaitility.Awaitility;
2834
import org.hamcrest.core.IsEqual;
35+
import org.junit.Rule;
2936
import org.junit.Test;
37+
import org.mockito.Mock;
38+
import org.mockito.junit.MockitoJUnit;
39+
import org.mockito.junit.MockitoRule;
3040

3141
public class ControllerTest {
3242

43+
@Rule public MockitoRule mockitoRule = MockitoJUnit.rule();
44+
45+
private static final Class<V1Pod> anyApiTypeClass = V1Pod.class;
46+
private static final long anyFullResyncPeriod = 1000L;
47+
48+
@Mock private DeltaFIFO deltaFIFOMock;
49+
@Mock private ListerWatcher<V1Pod, V1PodList> listerWatcherMock;
50+
51+
@Mock
52+
private Consumer<Deque<MutablePair<DeltaFIFO.DeltaType, KubernetesObject>>> popProcessFuncMock;
53+
54+
@Mock private Supplier<Boolean> resyncFuncMock;
55+
@Mock private BiConsumer<Class<V1Pod>, Throwable> exceptionHandlerMock;
56+
3357
@Test
3458
public void testControllerProcessDeltas() {
3559

@@ -75,4 +99,37 @@ public void testControllerProcessDeltas() {
7599
controller.stop();
76100
}
77101
}
102+
103+
@Test
104+
public void testReflectorIsConstructedWithExeptionHandler() {
105+
Controller<V1Pod, V1PodList> controller =
106+
new Controller<>(
107+
anyApiTypeClass,
108+
deltaFIFOMock,
109+
listerWatcherMock,
110+
popProcessFuncMock,
111+
resyncFuncMock,
112+
anyFullResyncPeriod,
113+
exceptionHandlerMock);
114+
assertSame(exceptionHandlerMock, controller.exceptionHandler);
115+
116+
ReflectorRunnable<V1Pod, V1PodList> reflector = controller.newReflector();
117+
118+
assertSame(exceptionHandlerMock, reflector.exceptionHandler);
119+
}
120+
121+
@Test
122+
public void testControllerIsConstructedWithoutExceptionHandlerPerDefault() {
123+
124+
Controller<V1Pod, V1PodList> controller =
125+
new Controller<>(
126+
anyApiTypeClass,
127+
deltaFIFOMock,
128+
listerWatcherMock,
129+
popProcessFuncMock,
130+
resyncFuncMock,
131+
anyFullResyncPeriod);
132+
133+
assertNull(controller.exceptionHandler);
134+
}
78135
}

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

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,15 +12,15 @@
1212
*/
1313
package io.kubernetes.client.informer.cache;
1414

15-
import static org.junit.Assert.assertFalse;
16-
import static org.junit.Assert.assertTrue;
15+
import static org.junit.Assert.*;
1716
import static org.mockito.Mockito.any;
1817
import static org.mockito.Mockito.mock;
1918
import static org.mockito.Mockito.never;
2019
import static org.mockito.Mockito.times;
2120
import static org.mockito.Mockito.verify;
2221
import static org.mockito.Mockito.when;
2322

23+
import io.kubernetes.client.common.KubernetesListObject;
2424
import io.kubernetes.client.informer.EventType;
2525
import io.kubernetes.client.informer.ListerWatcher;
2626
import io.kubernetes.client.openapi.ApiException;
@@ -35,6 +35,7 @@
3535
import java.time.Duration;
3636
import java.util.concurrent.CompletableFuture;
3737
import java.util.concurrent.atomic.AtomicReference;
38+
import java.util.function.BiConsumer;
3839
import org.awaitility.Awaitility;
3940
import org.hamcrest.core.IsEqual;
4041
import org.junit.Test;
@@ -45,10 +46,14 @@
4546
@RunWith(MockitoJUnitRunner.class)
4647
public class ReflectorRunnableTest {
4748

49+
private static final Class<V1Pod> anyApiType = V1Pod.class;
50+
4851
@Mock private DeltaFIFO deltaFIFO;
4952

5053
@Mock private ListerWatcher<V1Pod, V1PodList> listerWatcher;
5154

55+
@Mock private BiConsumer<Class<V1Pod>, Throwable> exceptionHandler;
56+
5257
@Test
5358
public void testReflectorRunOnce() throws ApiException {
5459
String mockResourceVersion = "1000";
@@ -343,4 +348,20 @@ public void testReflectorListShouldHandleExpiredResourceVersionFromWatchHandler(
343348
reflectorRunnable.stop();
344349
}
345350
}
351+
352+
@Test
353+
public void testDefaultExceptionHandlerSetPerDefault() {
354+
ReflectorRunnable<V1Pod, KubernetesListObject> reflector =
355+
new ReflectorRunnable<>(anyApiType, listerWatcher, deltaFIFO);
356+
357+
assertNotNull(reflector.exceptionHandler);
358+
}
359+
360+
@Test
361+
public void testGivemExceptionHandlerSet() {
362+
ReflectorRunnable<V1Pod, KubernetesListObject> reflector =
363+
new ReflectorRunnable<>(anyApiType, listerWatcher, deltaFIFO, exceptionHandler);
364+
365+
assertSame(exceptionHandler, reflector.exceptionHandler);
366+
}
346367
}

0 commit comments

Comments
 (0)