diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventFilteringIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventFilteringIT.java new file mode 100644 index 0000000000..75682b1e03 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventFilteringIT.java @@ -0,0 +1,121 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +import java.time.Duration; +import java.util.Map; +import java.util.Objects; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.annotation.Sample; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static io.javaoperatorsdk.operator.baseapi.statuseventfiltering.StatusEventPrimaryReconciler.PRIMARY_NAME_LABEL; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Sample( + tldr = "Cross-controller status event filtering reproducer", + description = + """ + Reproduces an issue where two independent controllers run in the same operator. \ + A secondary controller reconciles its primary resource and patches its status \ + (observedGeneration = generation) while the primary controller patches its own \ + status. The primary controller watches secondaries via InformerEventSource. \ + The concurrent status patches may cause the secondary's status-change event to \ + be lost — leaving the primary controller unaware the secondary is ready.\ + """) +class StatusEventFilteringIT { + + @RegisterExtension + static LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withReconciler(StatusEventSecondaryReconciler.class) + .withReconciler(StatusEventPrimaryReconciler.class) + .build(); + + @Test + void mapperShouldFireAfterIndependentControllerPatchesStatus() { + // Given: primary exists, secondary created and reaches steady state + var primary = new StatusEventPrimaryResource(); + primary.setMetadata( + new ObjectMetaBuilder() + .withName("test-primary") + .withNamespace(extension.getNamespace()) + .build()); + extension.create(primary); + + var secondary = new StatusEventSecondaryResource(); + secondary.setMetadata( + new ObjectMetaBuilder() + .withName("test-secondary") + .withNamespace(extension.getNamespace()) + .withLabels(Map.of(PRIMARY_NAME_LABEL, "test-primary")) + .build()); + secondary.setSpec(new StatusEventSecondaryResourceSpec()); + secondary.getSpec().setValue("initial"); + extension.create(secondary); + + await() + .atMost(Duration.ofSeconds(30)) + .until( + () -> { + var p = extension.get(StatusEventPrimaryResource.class, "test-primary"); + return Boolean.TRUE.equals(p.getStatus().getSecondaryReady()); + }); + + // Arm the stall so the next primary reconciliation blocks before returning patchStatus + var primaryReconciler = extension.getReconcilerOfType(StatusEventPrimaryReconciler.class); + primaryReconciler.stallReconcile(); + + // When: secondary spec changes, triggering the secondary controller to reconcile + // and patch its status via UpdateControl.patchStatus() (through + // eventFilteringUpdateAndCacheResource) + var current = extension.get(StatusEventSecondaryResource.class, "test-secondary"); + current.getSpec().setValue("updated"); + extension.replace(current); + + // Then: wait for secondary controller to patch its status (pure status-only WATCH event), + // then release the stalled primary. + await() + .atMost(Duration.ofSeconds(30)) + .until( + () -> { + var sec = extension.get(StatusEventSecondaryResource.class, "test-secondary"); + return Objects.equals( + sec.getStatus().getObservedGeneration(), sec.getMetadata().getGeneration()); + }); + primaryReconciler.releaseReconcile(); + + // The primary was stalled during the spec-change event (secondaryReady=false). + // After release it patches secondaryReady=false. The status-change event from + // the secondary controller's patchStatus should trigger a re-reconciliation + // where the primary sees secondaryReady=true. + await() + .atMost(Duration.ofSeconds(15)) + .pollInterval(Duration.ofMillis(500)) + .untilAsserted( + () -> { + var p = extension.get(StatusEventPrimaryResource.class, "test-primary"); + assertThat(p.getStatus().getSecondaryReady()) + .as("primary should see secondary as ready after controller status patch") + .isTrue(); + }); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryReconciler.java new file mode 100644 index 0000000000..e7ac101d32 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryReconciler.java @@ -0,0 +1,131 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; + +@ControllerConfiguration +public class StatusEventPrimaryReconciler implements Reconciler { + + private static final Logger log = LoggerFactory.getLogger(StatusEventPrimaryReconciler.class); + + public static final String PRIMARY_NAME_LABEL = "primary-name"; + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + private volatile CountDownLatch stallLatch; + + public void stallReconcile() { + stallLatch = new CountDownLatch(1); + } + + public void releaseReconcile() { + var latch = stallLatch; + if (latch != null) { + latch.countDown(); + } + stallLatch = null; + } + + @Override + public UpdateControl reconcile( + StatusEventPrimaryResource resource, Context context) { + + int count = numberOfExecutions.incrementAndGet(); + log.info( + "PrimaryReconciler reconciled '{}', execution #{}", + resource.getMetadata().getName(), + count); + + var secondaries = + context + .getClient() + .resources(StatusEventSecondaryResource.class) + .inNamespace(resource.getMetadata().getNamespace()) + .withLabel(PRIMARY_NAME_LABEL, resource.getMetadata().getName()) + .list() + .getItems(); + var latch = stallLatch; + if (latch != null) { + try { + log.info("PrimaryReconciler stalled before returning patchStatus..."); + latch.await(60, TimeUnit.SECONDS); + log.info("PrimaryReconciler released, returning patchStatus"); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + boolean allReady = + !secondaries.isEmpty() + && secondaries.stream() + .allMatch( + s -> + Objects.equals( + s.getMetadata().getGeneration(), + s.getStatus().getObservedGeneration())); + + log.info("PrimaryReconciler: secondaryReady={}, secondaries={}", allReady, secondaries.size()); + + resource.getStatus().setSecondaryReady(allReady); + return UpdateControl.patchStatus(resource); + } + + @Override + public List> prepareEventSources( + EventSourceContext context) { + + var config = + InformerEventSourceConfiguration.from( + StatusEventSecondaryResource.class, StatusEventPrimaryResource.class) + .withSecondaryToPrimaryMapper( + secondary -> { + var primaryName = secondary.getMetadata().getLabels().get(PRIMARY_NAME_LABEL); + if (primaryName == null) { + return Set.of(); + } + log.info( + "Mapper: secondary '{}' -> primary '{}'", + secondary.getMetadata().getName(), + primaryName); + return Set.of( + new ResourceID(primaryName, secondary.getMetadata().getNamespace())); + }) + .build(); + + return List.of(new InformerEventSource<>(config, context)); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResource.java new file mode 100644 index 0000000000..b4162c9dda --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResource.java @@ -0,0 +1,34 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("sepr") +public class StatusEventPrimaryResource + extends CustomResource implements Namespaced { + + @Override + protected StatusEventPrimaryResourceStatus initStatus() { + return new StatusEventPrimaryResourceStatus(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResourceStatus.java new file mode 100644 index 0000000000..caf4ddcf43 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventPrimaryResourceStatus.java @@ -0,0 +1,28 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +public class StatusEventPrimaryResourceStatus { + private Boolean secondaryReady; + + public Boolean getSecondaryReady() { + return secondaryReady; + } + + public void setSecondaryReady(Boolean secondaryReady) { + this.secondaryReady = secondaryReady; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryReconciler.java new file mode 100644 index 0000000000..351262a904 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryReconciler.java @@ -0,0 +1,52 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +import java.util.Objects; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; + +@ControllerConfiguration +public class StatusEventSecondaryReconciler implements Reconciler { + + private static final Logger log = LoggerFactory.getLogger(StatusEventSecondaryReconciler.class); + + @Override + public UpdateControl reconcile( + StatusEventSecondaryResource resource, Context context) { + + var generation = resource.getMetadata().getGeneration(); + var observedGeneration = resource.getStatus().getObservedGeneration(); + + if (!Objects.equals(generation, observedGeneration)) { + log.info( + "Patching secondary '{}' status: obsGen {} -> {}", + resource.getMetadata().getName(), + observedGeneration, + generation); + resource.getStatus().setObservedGeneration(generation); + return UpdateControl.patchStatus(resource); + } + + return UpdateControl.noUpdate(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResource.java new file mode 100644 index 0000000000..874825df62 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResource.java @@ -0,0 +1,40 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@ShortNames("sesr") +public class StatusEventSecondaryResource + extends CustomResource + implements Namespaced { + + @Override + protected StatusEventSecondaryResourceSpec initSpec() { + return new StatusEventSecondaryResourceSpec(); + } + + @Override + protected StatusEventSecondaryResourceStatus initStatus() { + return new StatusEventSecondaryResourceStatus(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceSpec.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceSpec.java new file mode 100644 index 0000000000..eb6bc0e734 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceSpec.java @@ -0,0 +1,28 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +public class StatusEventSecondaryResourceSpec { + private String value; + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceStatus.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceStatus.java new file mode 100644 index 0000000000..c7f14fa936 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/statuseventfiltering/StatusEventSecondaryResourceStatus.java @@ -0,0 +1,37 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.baseapi.statuseventfiltering; + +public class StatusEventSecondaryResourceStatus { + private Long observedGeneration; + private String message; + + public Long getObservedGeneration() { + return observedGeneration; + } + + public void setObservedGeneration(Long observedGeneration) { + this.observedGeneration = observedGeneration; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } +}