feat: enable configuring a handler to listen to informers stopping

Author: metacosm

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java

@@ -148,4 +148,7 @@ default Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
     return Optional.empty();
   }
 
+  default Optional<InformerStoppedHandler> getInformerStoppedHandler() {
+    return Optional.empty();
+  }
 }

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java

@@ -26,6 +26,7 @@ public class ConfigurationServiceOverrider {
   private ExecutorService executorService;
   private ExecutorService workflowExecutorService;
   private LeaderElectionConfiguration leaderElectionConfiguration;
+  private InformerStoppedHandler informerStoppedHandler;
 
   ConfigurationServiceOverrider(ConfigurationService original) {
     this.original = original;
@@ -93,6 +94,11 @@ public ConfigurationServiceOverrider withLeaderElectionConfiguration(
     return this;
   }
 
+  public ConfigurationServiceOverrider withInformerStoppedHandler(InformerStoppedHandler handler) {
+    this.informerStoppedHandler = handler;
+    return this;
+  }
+
   public ConfigurationService build() {
     return new BaseConfigurationService(original.getVersion(), cloner, objectMapper) {
       @Override
@@ -159,6 +165,12 @@ public Optional<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
         return leaderElectionConfiguration != null ? Optional.of(leaderElectionConfiguration)
             : original.getLeaderElectionConfiguration();
       }
+
+      @Override
+      public Optional<InformerStoppedHandler> getInformerStoppedHandler() {
+        return informerStoppedHandler != null ? Optional.of(informerStoppedHandler)
+            : original.getInformerStoppedHandler();
+      }
     };
   }
 

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/InformerStoppedHandler.java

@@ -0,0 +1,9 @@
+package io.javaoperatorsdk.operator.api.config;
+
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+
+public interface InformerStoppedHandler {
+
+  @SuppressWarnings("rawtypes")
+  void onStop(SharedIndexInformer informer, Throwable ex);
+}

operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java

@@ -13,6 +13,7 @@
 import io.fabric8.kubernetes.client.informers.cache.Cache;
 import io.javaoperatorsdk.operator.OperatorException;
 import io.javaoperatorsdk.operator.ReconcilerUtils;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
 import io.javaoperatorsdk.operator.processing.LifecycleAware;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
 import io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache;
@@ -25,6 +26,18 @@ class InformerWrapper<T extends HasMetadata>
 
   public InformerWrapper(SharedIndexInformer<T> informer) {
     this.informer = informer;
+
+    // register
+    ConfigurationServiceProvider.instance().getInformerStoppedHandler()
+        .ifPresent(ish -> {
+          final var stopped = informer.stopped();
+          if (stopped != null) {
+            stopped.handle((res, ex) -> {
+              ish.onStop(informer, ex);
+              return null;
+            });
+          }
+        });
     this.cache = (Cache<T>) informer.getStore();
   }
 

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java

@@ -1,5 +1,8 @@
 package io.javaoperatorsdk.operator;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
 import io.fabric8.kubernetes.api.model.HasMetadata;
 import io.fabric8.kubernetes.api.model.KubernetesResourceList;
 import io.fabric8.kubernetes.client.GenericKubernetesClient;
@@ -14,12 +17,18 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.ArgumentMatchers.nullable;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class MockKubernetesClient {
 
   public static <T extends HasMetadata> KubernetesClient client(Class<T> clazz) {
+    return client(clazz, null);
+  }
+
+  public static <T extends HasMetadata> KubernetesClient client(Class<T> clazz,
+      Consumer<Void> informerRunBehavior) {
     final var client = mock(GenericKubernetesClient.class);
     MixedOperation<T, KubernetesResourceList<T>, Resource<T>> resources =
         mock(MixedOperation.class);
@@ -34,6 +43,19 @@ public static <T extends HasMetadata> KubernetesClient client(Class<T> clazz) {
     when(resources.inAnyNamespace()).thenReturn(inAnyNamespace);
     when(inAnyNamespace.withLabelSelector(nullable(String.class))).thenReturn(filterable);
     SharedIndexInformer<T> informer = mock(SharedIndexInformer.class);
+    CompletableFuture<Void> stopped = new CompletableFuture<>();
+    when(informer.stopped()).thenReturn(stopped);
+    if (informerRunBehavior != null) {
+      doAnswer(invocation -> {
+        try {
+          informerRunBehavior.accept(null);
+        } catch (Exception e) {
+          stopped.completeExceptionally(e);
+        }
+        return null;
+      }).when(informer).run();
+    }
+    doAnswer(invocation -> null).when(informer).stop();
     Indexer mockIndexer = mock(Indexer.class);
     when(informer.getIndexer()).thenReturn(mockIndexer);
     when(filterable.runnableInformer(anyLong())).thenReturn(informer);

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java

@@ -112,6 +112,8 @@ public <R extends HasMetadata> R clone(R object) {
         .withTerminationTimeoutSeconds(100)
         .withMetrics(new Metrics() {})
         .withLeaderElectionConfiguration(new LeaderElectionConfiguration("newLease", "newLeaseNS"))
+        .withInformerStoppedHandler((informer, ex) -> {
+        })
         .build();
 
     assertNotEquals(config.closeClientOnStop(), overridden.closeClientOnStop());
@@ -128,5 +130,7 @@ public <R extends HasMetadata> R clone(R object) {
     assertNotEquals(config.getObjectMapper(), overridden.getObjectMapper());
     assertNotEquals(config.getLeaderElectionConfiguration(),
         overridden.getLeaderElectionConfiguration());
+    assertNotEquals(config.getInformerStoppedHandler(),
+        overridden.getLeaderElectionConfiguration());
   }
 }

operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java

@@ -9,11 +9,9 @@
 import io.fabric8.kubernetes.api.model.ObjectMeta;
 import io.fabric8.kubernetes.api.model.apps.Deployment;
 import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
-import io.fabric8.kubernetes.client.dsl.FilterWatchListMultiDeletable;
-import io.fabric8.kubernetes.client.dsl.MixedOperation;
-import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
-import io.fabric8.kubernetes.client.informers.cache.Indexer;
+import io.javaoperatorsdk.operator.MockKubernetesClient;
+import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider;
+import io.javaoperatorsdk.operator.api.config.InformerStoppedHandler;
 import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration;
 import io.javaoperatorsdk.operator.processing.event.EventHandler;
 import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -22,6 +20,8 @@
 
 import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_NAMESPACES_SET;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.times;
@@ -36,28 +36,15 @@ class InformerEventSourceTest {
   private static final String NEXT_RESOURCE_VERSION = "2";
 
   private InformerEventSource<Deployment, TestCustomResource> informerEventSource;
-  private final KubernetesClient clientMock = mock(KubernetesClient.class);
+  private final KubernetesClient clientMock = MockKubernetesClient.client(Deployment.class);
   private final TemporaryResourceCache<Deployment> temporaryResourceCacheMock =
       mock(TemporaryResourceCache.class);
   private final EventHandler eventHandlerMock = mock(EventHandler.class);
-  private final MixedOperation crClientMock = mock(MixedOperation.class);
-  private final FilterWatchListMultiDeletable specificResourceClientMock =
-      mock(FilterWatchListMultiDeletable.class);
-  private final FilterWatchListDeletable labeledResourceClientMock =
-      mock(FilterWatchListDeletable.class);
-  private final SharedIndexInformer informer = mock(SharedIndexInformer.class);
   private final InformerConfiguration<Deployment> informerConfiguration =
       mock(InformerConfiguration.class);
 
   @BeforeEach
   void setup() {
-    when(clientMock.resources(any())).thenReturn(crClientMock);
-    when(crClientMock.inAnyNamespace()).thenReturn(specificResourceClientMock);
-    when(specificResourceClientMock.withLabelSelector((String) null))
-        .thenReturn(labeledResourceClientMock);
-    when(labeledResourceClientMock.runnableInformer(0)).thenReturn(informer);
-    when(informer.getIndexer()).thenReturn(mock(Indexer.class));
-
     when(informerConfiguration.getEffectiveNamespaces())
         .thenReturn(DEFAULT_NAMESPACES_SET);
     when(informerConfiguration.getSecondaryToPrimaryMapper())
@@ -256,6 +243,20 @@ void filtersOnDeleteEvents() {
     verify(eventHandlerMock, never()).handleEvent(any());
   }
 
+  @Test
+  void informerStoppedHandlerShouldBeCalledWhenInformerStops() {
+    final var exception = new RuntimeException("Informer stopped exceptionally!");
+    final var informerStoppedHandler = mock(InformerStoppedHandler.class);
+    ConfigurationServiceProvider
+        .overrideCurrent(overrider -> overrider.withInformerStoppedHandler(informerStoppedHandler));
+    informerEventSource = new InformerEventSource<>(informerConfiguration,
+        MockKubernetesClient.client(Deployment.class, unused -> {
+          throw exception;
+        }));
+    informerEventSource.start();
+    verify(informerStoppedHandler, atLeastOnce()).onStop(any(), eq(exception));
+  }
+
   Deployment testDeployment() {
     Deployment deployment = new Deployment();
     deployment.setMetadata(new ObjectMeta());