From 447b885baefe7273ea9557330e668d6fe94d5ddb Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Wed, 21 Sep 2022 15:21:08 +0200 Subject: [PATCH 01/10] chore: update to SNAPSHOT version of fabric8 client --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index de985d53ab..daf1db265a 100644 --- a/pom.xml +++ b/pom.xml @@ -43,7 +43,7 @@ https://sonarcloud.io 5.9.0 - 5.12.3 + 5.12-SNAPSHOT 1.7.36 2.18.0 4.8.0 From 99618850d353fb47584c1fb5b4786984aa5bdaa6 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Wed, 21 Sep 2022 18:21:10 +0200 Subject: [PATCH 02/10] feat: enable configuring a handler to listen to informers stopping --- .../api/config/ConfigurationService.java | 3 ++ .../config/ConfigurationServiceOverrider.java | 12 ++++++ .../api/config/InformerStoppedHandler.java | 9 +++++ .../source/informer/InformerWrapper.java | 13 +++++++ .../operator/MockKubernetesClient.java | 22 +++++++++++ .../ConfigurationServiceOverriderTest.java | 4 ++ .../informer/InformerEventSourceTest.java | 39 ++++++++++--------- 7 files changed, 83 insertions(+), 19 deletions(-) create mode 100644 operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/InformerStoppedHandler.java diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java index a7f50bcf6c..00e0b84526 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationService.java @@ -148,4 +148,7 @@ default Optional getLeaderElectionConfiguration() { return Optional.empty(); } + default Optional getInformerStoppedHandler() { + return Optional.empty(); + } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java index 26e1de7bb1..ee442c07fa 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverrider.java +++ b/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 getLeaderElectionConfiguration() { return leaderElectionConfiguration != null ? Optional.of(leaderElectionConfiguration) : original.getLeaderElectionConfiguration(); } + + @Override + public Optional getInformerStoppedHandler() { + return informerStoppedHandler != null ? Optional.of(informerStoppedHandler) + : original.getInformerStoppedHandler(); + } }; } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/InformerStoppedHandler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/InformerStoppedHandler.java new file mode 100644 index 0000000000..d204c86664 --- /dev/null +++ b/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); +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java index 24930b6c0c..d6e6c7ce77 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java +++ b/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 public InformerWrapper(SharedIndexInformer 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) informer.getStore(); } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java index de8d5b6e9d..c7232c6b24 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/MockKubernetesClient.java +++ b/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 KubernetesClient client(Class clazz) { + return client(clazz, null); + } + + public static KubernetesClient client(Class clazz, + Consumer informerRunBehavior) { final var client = mock(GenericKubernetesClient.class); MixedOperation, Resource> resources = mock(MixedOperation.class); @@ -34,6 +43,19 @@ public static KubernetesClient client(Class clazz) { when(resources.inAnyNamespace()).thenReturn(inAnyNamespace); when(inAnyNamespace.withLabelSelector(nullable(String.class))).thenReturn(filterable); SharedIndexInformer informer = mock(SharedIndexInformer.class); + CompletableFuture 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); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java index 8b973b8159..878118ba25 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java @@ -112,6 +112,8 @@ public 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 clone(R object) { assertNotEquals(config.getObjectMapper(), overridden.getObjectMapper()); assertNotEquals(config.getLeaderElectionConfiguration(), overridden.getLeaderElectionConfiguration()); + assertNotEquals(config.getInformerStoppedHandler(), + overridden.getLeaderElectionConfiguration()); } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java index d134ea30e5..112d02dc66 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java +++ b/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 informerEventSource; - private final KubernetesClient clientMock = mock(KubernetesClient.class); + private final KubernetesClient clientMock = MockKubernetesClient.client(Deployment.class); private final TemporaryResourceCache 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 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()); From 41c06479eef9c1fa8349ca4c236aac7427fa03c2 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Wed, 28 Sep 2022 15:04:23 +0200 Subject: [PATCH 03/10] chore: temporarily activate snapshot repository --- pom.xml | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/pom.xml b/pom.xml index daf1db265a..bb830fe36f 100644 --- a/pom.xml +++ b/pom.xml @@ -196,6 +196,17 @@ + + + + ossrh + https://oss.sonatype.org/content/repositories/snapshots + + true + + + + ossrh From 8aac329a7456665641dc041c27a9104f3f8a6199 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Wed, 28 Sep 2022 16:38:16 +0200 Subject: [PATCH 04/10] fix: properly reset ConfigurationServiceProvider after test --- .../informer/InformerEventSourceTest.java | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java index 112d02dc66..e941e8390d 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerEventSourceTest.java @@ -245,16 +245,21 @@ void filtersOnDeleteEvents() { @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)); + try { + 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)); + } finally { + ConfigurationServiceProvider.reset(); + } } Deployment testDeployment() { From f4be98df921de7a7d2a8a050a177007785a85564 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Thu, 29 Sep 2022 09:53:40 +0200 Subject: [PATCH 05/10] fix: fail fast if stopped future is null --- .../event/source/informer/InformerWrapper.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java index d6e6c7ce77..abbc0fcad1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java @@ -31,12 +31,10 @@ public InformerWrapper(SharedIndexInformer informer) { ConfigurationServiceProvider.instance().getInformerStoppedHandler() .ifPresent(ish -> { final var stopped = informer.stopped(); - if (stopped != null) { - stopped.handle((res, ex) -> { - ish.onStop(informer, ex); - return null; - }); - } + stopped.handle((res, ex) -> { + ish.onStop(informer, ex); + return null; + }); }); this.cache = (Cache) informer.getStore(); } From c5cdd55919bc86e7da649ec0e49cd8d7704a0e55 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Thu, 29 Sep 2022 15:01:21 +0200 Subject: [PATCH 06/10] feat: make it clearer when the CRD file isn't found (#1503) --- .../operator/junit/LocallyRunOperatorExtension.java | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java index e2f4234453..f7627a5555 100644 --- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java +++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java @@ -134,7 +134,10 @@ protected void before(ExtensionContext context) { ref.controllerConfigurationOverrider.accept(oconfig); } - applyCrd(config.getResourceTypeName()); + // only try to apply a CRD for the reconciler if it is associated to a CR + if (CustomResource.class.isAssignableFrom(config.getResourceClass())) { + applyCrd(config.getResourceTypeName()); + } if (ref.reconciler instanceof KubernetesClientAware) { ((KubernetesClientAware) ref.reconciler).setKubernetesClient(kubernetesClient); @@ -151,6 +154,9 @@ protected void before(ExtensionContext context) { private void applyCrd(String resourceTypeName) { String path = "/META-INF/fabric8/" + resourceTypeName + "-v1.yml"; try (InputStream is = getClass().getResourceAsStream(path)) { + if (is == null) { + throw new IllegalStateException("Cannot find CRD at " + path); + } final var crd = getKubernetesClient().load(is); crd.createOrReplace(); Thread.sleep(CRD_READY_WAIT); // readiness is not applicable for CRD, just wait a little From 233b95460780520faf3bcfa68fbf9f04a87797e7 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Thu, 29 Sep 2022 18:02:32 +0200 Subject: [PATCH 07/10] feat: make it possible to override the configuration service --- .../operator/junit/AbstractOperatorExtension.java | 14 ++++---------- .../junit/ClusterDeployedOperatorExtension.java | 5 +---- .../junit/LocallyRunOperatorExtension.java | 8 +++----- 3 files changed, 8 insertions(+), 19 deletions(-) diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/AbstractOperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/AbstractOperatorExtension.java index 3f7e2225a5..7dac1ad674 100644 --- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/AbstractOperatorExtension.java +++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/AbstractOperatorExtension.java @@ -7,6 +7,7 @@ import java.util.Locale; import java.util.UUID; import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; import org.awaitility.Awaitility; import org.junit.jupiter.api.extension.*; @@ -22,7 +23,7 @@ import io.fabric8.kubernetes.client.dsl.Resource; import io.fabric8.kubernetes.client.utils.KubernetesResourceUtil; import io.fabric8.kubernetes.client.utils.Utils; -import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider; public abstract class AbstractOperatorExtension implements HasKubernetesClient, @@ -35,7 +36,6 @@ public abstract class AbstractOperatorExtension implements HasKubernetesClient, public static final int CRD_READY_WAIT = 2000; private final KubernetesClient kubernetesClient = new DefaultKubernetesClient(); - protected final ConfigurationService configurationService; protected final List infrastructure; protected Duration infrastructureTimeout; protected final boolean oneNamespacePerClass; @@ -45,14 +45,11 @@ public abstract class AbstractOperatorExtension implements HasKubernetesClient, protected String namespace; protected AbstractOperatorExtension( - ConfigurationService configurationService, List infrastructure, Duration infrastructureTimeout, boolean oneNamespacePerClass, boolean preserveNamespaceOnError, boolean waitForNamespaceDeletion) { - - this.configurationService = configurationService; this.infrastructure = infrastructure; this.infrastructureTimeout = infrastructureTimeout; this.oneNamespacePerClass = oneNamespacePerClass; @@ -214,7 +211,6 @@ protected void deleteOperator() { @SuppressWarnings("unchecked") public static abstract class AbstractBuilder> { - protected ConfigurationService configurationService; protected final List infrastructure; protected Duration infrastructureTimeout; protected boolean preserveNamespaceOnError; @@ -222,8 +218,6 @@ public static abstract class AbstractBuilder> { protected boolean oneNamespacePerClass; protected AbstractBuilder() { - this.configurationService = ConfigurationServiceProvider.instance(); - this.infrastructure = new ArrayList<>(); this.infrastructureTimeout = Duration.ofMinutes(1); @@ -255,8 +249,8 @@ public T oneNamespacePerClass(boolean value) { return (T) this; } - public T withConfigurationService(ConfigurationService value) { - configurationService = value; + public T withConfigurationService(Consumer overrider) { + ConfigurationServiceProvider.overrideCurrent(overrider); return (T) this; } diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/ClusterDeployedOperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/ClusterDeployedOperatorExtension.java index 4b23fe0805..2541ae1858 100644 --- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/ClusterDeployedOperatorExtension.java +++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/ClusterDeployedOperatorExtension.java @@ -18,7 +18,6 @@ import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.api.model.rbac.ClusterRoleBinding; -import io.javaoperatorsdk.operator.api.config.ConfigurationService; public class ClusterDeployedOperatorExtension extends AbstractOperatorExtension { @@ -29,7 +28,6 @@ public class ClusterDeployedOperatorExtension extends AbstractOperatorExtension private final Duration operatorDeploymentTimeout; private ClusterDeployedOperatorExtension( - ConfigurationService configurationService, List operatorDeployment, Duration operatorDeploymentTimeout, List infrastructure, @@ -37,7 +35,7 @@ private ClusterDeployedOperatorExtension( boolean preserveNamespaceOnError, boolean waitForNamespaceDeletion, boolean oneNamespacePerClass) { - super(configurationService, infrastructure, infrastructureTimeout, oneNamespacePerClass, + super(infrastructure, infrastructureTimeout, oneNamespacePerClass, preserveNamespaceOnError, waitForNamespaceDeletion); this.operatorDeployment = operatorDeployment; @@ -137,7 +135,6 @@ public Builder withOperatorDeployment(HasMetadata... hms) { public ClusterDeployedOperatorExtension build() { return new ClusterDeployedOperatorExtension( - configurationService, operatorDeployment, deploymentTimeout, infrastructure, diff --git a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java index f7627a5555..656d160acf 100644 --- a/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java +++ b/operator-framework-junit5/src/main/java/io/javaoperatorsdk/operator/junit/LocallyRunOperatorExtension.java @@ -20,7 +20,7 @@ import io.javaoperatorsdk.operator.Operator; import io.javaoperatorsdk.operator.ReconcilerUtils; import io.javaoperatorsdk.operator.RegisteredController; -import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceProvider; import io.javaoperatorsdk.operator.api.config.ControllerConfigurationOverrider; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; import io.javaoperatorsdk.operator.processing.retry.Retry; @@ -40,7 +40,6 @@ public class LocallyRunOperatorExtension extends AbstractOperatorExtension { private final Map registeredControllers; private LocallyRunOperatorExtension( - ConfigurationService configurationService, List reconcilers, List infrastructure, List portForwards, @@ -50,7 +49,6 @@ private LocallyRunOperatorExtension( boolean waitForNamespaceDeletion, boolean oneNamespacePerClass) { super( - configurationService, infrastructure, infrastructureTimeout, oneNamespacePerClass, @@ -60,7 +58,7 @@ private LocallyRunOperatorExtension( this.portForwards = portForwards; this.localPortForwards = new ArrayList<>(portForwards.size()); this.additionalCustomResourceDefinitions = additionalCustomResourceDefinitions; - this.operator = new Operator(getKubernetesClient(), this.configurationService); + this.operator = new Operator(getKubernetesClient()); this.registeredControllers = new HashMap<>(); } @@ -123,6 +121,7 @@ protected void before(ExtensionContext context) { additionalCustomResourceDefinitions .forEach(cr -> applyCrd(ReconcilerUtils.getResourceTypeName(cr))); + final var configurationService = ConfigurationServiceProvider.instance(); for (var ref : reconcilers) { final var config = configurationService.getConfigurationFor(ref.reconciler); final var oconfig = override(config).settingNamespace(namespace); @@ -252,7 +251,6 @@ public Builder withAdditionalCustomResourceDefinition( public LocallyRunOperatorExtension build() { return new LocallyRunOperatorExtension( - configurationService, reconcilers, infrastructure, portForwards, From 165ab9369ddae59c797c4884b4e84c6a8073572f Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Thu, 29 Sep 2022 19:24:26 +0200 Subject: [PATCH 08/10] feat: throw an exception if stopped future is null --- .../source/informer/InformerWrapper.java | 31 +++++++++++++------ 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java index abbc0fcad1..cb1ad28674 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapper.java @@ -26,16 +26,6 @@ class InformerWrapper public InformerWrapper(SharedIndexInformer informer) { this.informer = informer; - - // register - ConfigurationServiceProvider.instance().getInformerStoppedHandler() - .ifPresent(ish -> { - final var stopped = informer.stopped(); - stopped.handle((res, ex) -> { - ish.onStop(informer, ex); - return null; - }); - }); this.cache = (Cache) informer.getStore(); } @@ -43,6 +33,27 @@ public InformerWrapper(SharedIndexInformer informer) { public void start() throws OperatorException { try { informer.run(); + + // register stopped handler if we have one defined + ConfigurationServiceProvider.instance().getInformerStoppedHandler() + .ifPresent(ish -> { + final var stopped = informer.stopped(); + if (stopped != null) { + stopped.handle((res, ex) -> { + ish.onStop(informer, ex); + return null; + }); + } else { + final var apiTypeClass = informer.getApiTypeClass(); + final var fullResourceName = + HasMetadata.getFullResourceName(apiTypeClass); + final var version = HasMetadata.getVersion(apiTypeClass); + throw new IllegalStateException( + "Cannot retrieve 'stopped' callback to listen to informer stopping for informer for " + + fullResourceName + "/" + version); + } + }); + } catch (Exception e) { ReconcilerUtils.handleKubernetesClientException(e, HasMetadata.getFullResourceName(informer.getApiTypeClass())); From 0d1d38f9fb01fed39e4c77bd1341bfd3b164090d Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Fri, 30 Sep 2022 15:11:18 +0200 Subject: [PATCH 09/10] feat: retrofit test to showcase the InformerStoppedHandler --- .../operator/MultiVersionCRDIT.java | 93 +++++++++++++++++-- 1 file changed, 87 insertions(+), 6 deletions(-) diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/MultiVersionCRDIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/MultiVersionCRDIT.java index 65bf839ff4..2126f129c7 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/MultiVersionCRDIT.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/MultiVersionCRDIT.java @@ -6,7 +6,12 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import io.fabric8.kubernetes.api.model.HasMetadata; import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.fabric8.kubernetes.client.WatcherException; +import io.fabric8.kubernetes.client.informers.SharedIndexInformer; +import io.fabric8.kubernetes.client.utils.Serialization; +import io.javaoperatorsdk.operator.api.config.InformerStoppedHandler; import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; import io.javaoperatorsdk.operator.sample.multiversioncrd.MultiVersionCRDTestCustomResource1; import io.javaoperatorsdk.operator.sample.multiversioncrd.MultiVersionCRDTestCustomResource2; @@ -15,6 +20,8 @@ import io.javaoperatorsdk.operator.sample.multiversioncrd.MultiVersionCRDTestReconciler1; import io.javaoperatorsdk.operator.sample.multiversioncrd.MultiVersionCRDTestReconciler2; +import com.fasterxml.jackson.core.JsonProcessingException; + import static com.google.common.truth.Truth.assertThat; import static org.awaitility.Awaitility.await; @@ -28,8 +35,73 @@ class MultiVersionCRDIT { LocallyRunOperatorExtension.builder() .withReconciler(MultiVersionCRDTestReconciler1.class) .withReconciler(MultiVersionCRDTestReconciler2.class) + .withConfigurationService( + overrider -> overrider.withInformerStoppedHandler(informerStoppedHandler)) .build(); + private static class TestInformerStoppedHandler implements InformerStoppedHandler { + private String resourceClassName; + private String resourceCreateAsVersion; + + private String failedResourceVersion; + private String errorMessage; + + @Override + @SuppressWarnings("rawtypes") + public void onStop(SharedIndexInformer informer, Throwable ex) { + if (ex instanceof WatcherException) { + WatcherException watcherEx = (WatcherException) ex; + watcherEx.getRawWatchMessage().ifPresent(raw -> { + try { + // extract the resource at which the version is attempted to be created (i.e. the stored + // version) + final var unmarshal = Serialization.jsonMapper().readTree(raw); + final var object = unmarshal.get("object"); + resourceCreateAsVersion = acceptOnlyIfUnsetOrEqualToAlreadySet(resourceCreateAsVersion, + object.get("apiVersion").asText()); + // extract the asked resource version + failedResourceVersion = acceptOnlyIfUnsetOrEqualToAlreadySet(failedResourceVersion, + object.get("metadata").get("managedFields").get(0).get("apiVersion").asText()); + } catch (JsonProcessingException e) { + throw new RuntimeException(e); + } + }); + + // extract error message + errorMessage = + acceptOnlyIfUnsetOrEqualToAlreadySet(errorMessage, watcherEx.getCause().getMessage()); + } + final var apiTypeClass = informer.getApiTypeClass(); + resourceClassName = + acceptOnlyIfUnsetOrEqualToAlreadySet(resourceClassName, apiTypeClass.getName()); + System.out.println("Informer for " + HasMetadata.getFullResourceName(apiTypeClass) + + " stopped due to: " + ex.getMessage()); + } + + public String getResourceClassName() { + return resourceClassName; + } + + public String getResourceCreateAsVersion() { + return resourceCreateAsVersion; + } + + public String getErrorMessage() { + return errorMessage; + } + + public String getFailedResourceVersion() { + return failedResourceVersion; + } + + private String acceptOnlyIfUnsetOrEqualToAlreadySet(String existing, String newValue) { + return (existing == null || existing.equals(newValue)) ? newValue : null; + } + } + + private final static TestInformerStoppedHandler informerStoppedHandler = + new TestInformerStoppedHandler(); + @Test void multipleCRDVersions() { operator.create(createTestResourceV1WithoutLabel()); @@ -52,7 +124,7 @@ void multipleCRDVersions() { } @Test - void invalidEventsDoesNotBreakEventHandling() { + void invalidEventsShouldStopInformerAndCallInformerStoppedHandler() { var v2res = createTestResourceV2WithLabel(); v2res.getMetadata().getLabels().clear(); operator.create(v2res); @@ -60,13 +132,22 @@ void invalidEventsDoesNotBreakEventHandling() { operator.create(v1res); await() - .atMost(Duration.ofSeconds(2)) + .atMost(Duration.ofSeconds(1)) .pollInterval(Duration.ofMillis(50)) .untilAsserted(() -> { - var crV1Now = operator.get(MultiVersionCRDTestCustomResource1.class, CR_V1_NAME); - assertThat(crV1Now.getStatus()).isNotNull(); - assertThat(crV1Now.getStatus().getReconciledBy()) - .containsExactly(MultiVersionCRDTestReconciler1.class.getSimpleName()); + // v1 is the stored version so trying to create a v2 version should fail because we cannot + // convert a String (as defined by the spec of the v2 CRD) to an int (which is what the + // spec of the v1 CRD defines) + assertThat(informerStoppedHandler.getResourceCreateAsVersion()) + .isEqualTo(HasMetadata.getApiVersion( + MultiVersionCRDTestCustomResource1.class)); + assertThat(informerStoppedHandler.getResourceClassName()) + .isEqualTo(MultiVersionCRDTestCustomResource1.class.getName()); + assertThat(informerStoppedHandler.getFailedResourceVersion()) + .isEqualTo(HasMetadata.getApiVersion( + MultiVersionCRDTestCustomResource2.class)); + assertThat(informerStoppedHandler.getErrorMessage()).contains( + "Cannot deserialize value of type `int` from String \"string value\": not a valid `int` value"); }); assertThat( operator From 8310ca33a7b0853927d3d1bf5a0e7924d1e36574 Mon Sep 17 00:00:00 2001 From: Chris Laprun Date: Fri, 30 Sep 2022 15:56:35 +0200 Subject: [PATCH 10/10] chore: update to Fabric8 client 5.12.4 --- pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pom.xml b/pom.xml index bb830fe36f..8950cf6dac 100644 --- a/pom.xml +++ b/pom.xml @@ -43,7 +43,7 @@ https://sonarcloud.io 5.9.0 - 5.12-SNAPSHOT + 5.12.4 1.7.36 2.18.0 4.8.0