Skip to content

Commit d69ea8a

Browse files
robobariorobobario
authored
Merge pull request kroxylicious#2102 from robobario/ignore-stale-referents
operator: VKC reconciler ignores events from stale referents
2 parents 8059379 + 5fb5961 commit d69ea8a

File tree

4 files changed

+322
-27
lines changed

4 files changed

+322
-27
lines changed

‎kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/ResourcesUtil.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -290,7 +290,13 @@ static <O extends HasMetadata, R extends HasMetadata> Set<ResourceID> findReferr
290290
return ResourcesUtil.filteredResourceIdsInSameNamespace(context,
291291
referent,
292292
owner,
293-
primary -> refAccessor.apply(primary).stream().anyMatch(ref -> ResourcesUtil.isReferent(ref, referent)));
293+
primary -> {
294+
Collection<? extends LocalRef<R>> refs = refAccessor.apply(primary);
295+
if (refs == null) {
296+
return false;
297+
}
298+
return refs.stream().anyMatch(ref -> ResourcesUtil.isReferent(ref, referent));
299+
});
294300
}
295301

296302
public static String namespacedSlug(LocalRef<?> ref, HasMetadata resource) {

‎kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/operator/VirtualKafkaClusterReconciler.java‎

Lines changed: 52 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
import org.slf4j.LoggerFactory;
2020

2121
import io.fabric8.kubernetes.api.model.ConfigMap;
22+
import io.fabric8.kubernetes.api.model.HasMetadata;
2223
import io.fabric8.kubernetes.api.model.OwnerReference;
2324
import io.fabric8.kubernetes.api.model.Secret;
2425
import io.fabric8.kubernetes.api.model.Service;
@@ -306,10 +307,7 @@ public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSource
306307
.withName(SERVICES_EVENT_SOURCE_NAME)
307308
.withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefAsResourceId(cluster,
308309
cluster.getSpec().getTargetKafkaServiceRef()))
309-
.withSecondaryToPrimaryMapper(service -> ResourcesUtil.findReferrers(context,
310-
service,
311-
VirtualKafkaCluster.class,
312-
cluster -> Optional.of(cluster.getSpec().getTargetKafkaServiceRef())))
310+
.withSecondaryToPrimaryMapper(kafkaServiceSecondaryToPrimaryMapper(context))
313311
.build();
314312

315313
InformerEventSourceConfiguration<KafkaProxyIngress> clusterToIngresses = InformerEventSourceConfiguration.from(
@@ -321,10 +319,7 @@ public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSource
321319
.map(VirtualKafkaClusterSpec::getIngresses)
322320
.stream().flatMap(List::stream)
323321
.map(Ingresses::getIngressRef).toList()))
324-
.withSecondaryToPrimaryMapper(ingress -> ResourcesUtil.findReferrersMulti(context,
325-
ingress,
326-
VirtualKafkaCluster.class,
327-
cluster -> cluster.getSpec().getIngresses().stream().map(Ingresses::getIngressRef).toList()))
322+
.withSecondaryToPrimaryMapper(ingressSecondaryToPrimaryMapper(context))
328323
.build();
329324

330325
InformerEventSourceConfiguration<KafkaProtocolFilter> clusterToFilters = InformerEventSourceConfiguration.from(
@@ -333,10 +328,7 @@ public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSource
333328
.withName(FILTERS_EVENT_SOURCE_NAME)
334329
.withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefsAsResourceIds(cluster,
335330
Optional.ofNullable(cluster.getSpec()).map(VirtualKafkaClusterSpec::getFilterRefs).orElse(List.of())))
336-
.withSecondaryToPrimaryMapper(filter -> ResourcesUtil.findReferrersMulti(context,
337-
filter,
338-
VirtualKafkaCluster.class,
339-
cluster -> cluster.getSpec().getFilterRefs()))
331+
.withSecondaryToPrimaryMapper(filterSecondaryToPrimaryMapper(context))
340332
.build();
341333

342334
InformerEventSourceConfiguration<Service> clusterToKubeService = InformerEventSourceConfiguration.from(
@@ -394,6 +386,54 @@ static SecondaryToPrimaryMapper<Secret> secretToVirtualKafkaCluster(EventSourceC
394386
.map(Tls::getCertificateRef).toList());
395387
}
396388

389+
@VisibleForTesting
390+
static SecondaryToPrimaryMapper<KafkaProtocolFilter> filterSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
391+
return filter -> {
392+
if (!ResourcesUtil.isStatusFresh(filter)) {
393+
logIgnoredEvent(filter);
394+
return Set.of();
395+
}
396+
return ResourcesUtil.findReferrersMulti(context,
397+
filter,
398+
VirtualKafkaCluster.class,
399+
cluster -> cluster.getSpec().getFilterRefs());
400+
};
401+
}
402+
403+
@VisibleForTesting
404+
static SecondaryToPrimaryMapper<KafkaProxyIngress> ingressSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
405+
return ingress -> {
406+
if (!ResourcesUtil.isStatusFresh(ingress)) {
407+
logIgnoredEvent(ingress);
408+
return Set.of();
409+
}
410+
return ResourcesUtil.findReferrersMulti(context,
411+
ingress,
412+
VirtualKafkaCluster.class,
413+
cluster -> cluster.getSpec().getIngresses().stream().map(Ingresses::getIngressRef).toList());
414+
};
415+
}
416+
417+
@VisibleForTesting
418+
static SecondaryToPrimaryMapper<KafkaService> kafkaServiceSecondaryToPrimaryMapper(EventSourceContext<VirtualKafkaCluster> context) {
419+
return service -> {
420+
if (!ResourcesUtil.isStatusFresh(service)) {
421+
logIgnoredEvent(service);
422+
return Set.of();
423+
}
424+
return ResourcesUtil.findReferrers(context,
425+
service,
426+
VirtualKafkaCluster.class,
427+
cluster -> Optional.of(cluster.getSpec().getTargetKafkaServiceRef()));
428+
};
429+
}
430+
431+
private static void logIgnoredEvent(HasMetadata hasMetadata) {
432+
if (LOGGER.isDebugEnabled()) {
433+
LOGGER.debug("Ignoring event from {} with stale status: {}", HasMetadata.getKind(hasMetadata.getClass()), ResourcesUtil.toLocalRef(hasMetadata));
434+
}
435+
}
436+
397437
@Override
398438
public ErrorStatusUpdateControl<VirtualKafkaCluster> updateErrorStatus(VirtualKafkaCluster cluster, Context<VirtualKafkaCluster> context, Exception e) {
399439
// ResolvedRefs to UNKNOWN

‎kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/operator/ResourcesUtilTest.java‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -156,6 +156,99 @@ void findReferrers() {
156156
assertThat(resources).containsExactly(ResourceID.fromResource(cm));
157157
}
158158

159+
@Test
160+
void findReferrersMultiShouldMapEmptyRefsToEmptySet() {
161+
// Given
162+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
163+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
164+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
165+
166+
// When
167+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
168+
configMap -> Set.of());
169+
170+
// Then
171+
assertThat(resources).isEmpty();
172+
}
173+
174+
@Test
175+
void findReferrersMultiShouldMapSingleRefToResourceInContext() {
176+
// Given
177+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
178+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
179+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
180+
181+
// When
182+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
183+
configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
184+
185+
// Then
186+
assertThat(resources).containsExactly(ResourceID.fromResource(cm));
187+
}
188+
189+
@Test
190+
void findReferrersMultiShouldMapMultipleRefsToCorrespondingResourcesInContext() {
191+
// Given
192+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
193+
ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "primary").endMetadata().build();
194+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
195+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
196+
197+
// When
198+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
199+
configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
200+
201+
// Then
202+
assertThat(resources).containsExactlyInAnyOrder(ResourceID.fromResource(cm), ResourceID.fromResource(cm2));
203+
}
204+
205+
@Test
206+
void findReferrersMultiShouldExcludeResourcesThatDontReferencePrimary() {
207+
// Given
208+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "primary").endMetadata().build();
209+
ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "not-primary").endMetadata().build();
210+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
211+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
212+
213+
// When
214+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
215+
configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
216+
217+
// Then
218+
assertThat(resources).containsExactlyInAnyOrder(ResourceID.fromResource(cm));
219+
}
220+
221+
@Test
222+
void findReferrersMultiShouldHandleAllSecondariesDontReferencePrimary() {
223+
// Given
224+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "not-primary").endMetadata().build();
225+
ConfigMap cm2 = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("bar").addToAnnotations("ref", "not-primary").endMetadata().build();
226+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm, cm2), ConfigMap.class);
227+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
228+
229+
// When
230+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
231+
configMap -> Set.of(new AnyLocalRefBuilder().withName(configMap.getMetadata().getAnnotations().get("ref")).build()));
232+
233+
// Then
234+
assertThat(resources).isEmpty();
235+
}
236+
237+
@Test
238+
void findReferrersMultiShouldTolerateNullReferencesCollection() {
239+
// Given
240+
ConfigMap cm = new ConfigMapBuilder().withNewMetadata().withNamespace("ns").withName("foo").addToAnnotations("ref", "not-primary").endMetadata().build();
241+
EventSourceContext<?> eventSourceContext = prepareMockContextToProduceList(List.of(cm), ConfigMap.class);
242+
HasMetadata primary = new SecretBuilder().withNewMetadata().withNamespace("ns").withName("primary").endMetadata().build();
243+
244+
// When
245+
var resources = ResourcesUtil.findReferrersMulti(eventSourceContext, primary, ConfigMap.class,
246+
configMap -> null);
247+
248+
// Then
249+
assertThat(resources).isEmpty();
250+
}
251+
159252
@Test
160253
void findReferrersSupportsResourcesWithoutReferences() {
161254
// Given

0 commit comments

Comments
 (0)