Skip to content

Commit 2cb78a0

Browse files
tombentleytombentley
authored
VKCReconciler part two (kroxylicious#1993)
* Resolve VKC service, ingress and filter references Signed-off-by: Tom Bentley <tbentley@redhat.com>
1 parent 86d2e61 commit 2cb78a0

File tree

10 files changed

+566
-64
lines changed

10 files changed

+566
-64
lines changed

‎kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/api/common/FilterRef.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@
1010

1111
import io.fabric8.kubernetes.api.model.KubernetesResource;
1212

13-
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
13+
import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
1414

1515
/**
1616
* A reference, used in a kubernetes resource, to a KafkaProxy resource in the same namespace.
@@ -31,7 +31,7 @@
3131
@io.sundr.builder.annotations.BuildableReference(io.fabric8.kubernetes.api.model.VolumeMount.class)
3232
})
3333
public class FilterRef
34-
extends LocalRef<KafkaProxy>
34+
extends LocalRef<KafkaProtocolFilter>
3535
implements io.fabric8.kubernetes.api.builder.Editable<FilterRefBuilder>,
3636
KubernetesResource {
3737

‎kroxylicious-operator/src/main/java/io/kroxylicious/kubernetes/api/common/LocalRef.java‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,14 +6,19 @@
66

77
package io.kroxylicious.kubernetes.api.common;
88

9+
import java.util.Comparator;
910
import java.util.Objects;
1011

1112
/**
1213
* Abstraction for references in one kubernetes resource to some kubernetes resource in the same namespace.
1314
* Two LocalRefs are equal iff they have the same group, kind and name (they don't need to have the same class)
1415
* @param <T> The Java type of the resource
1516
*/
16-
public abstract class LocalRef<T> {
17+
public abstract class LocalRef<T> implements Comparable<LocalRef<T>> {
18+
19+
public static final Comparator<LocalRef<?>> COMPARATOR = Comparator.<LocalRef<?>, String> comparing(LocalRef::getKind)
20+
.thenComparing(LocalRef::getGroup)
21+
.thenComparing(LocalRef::getName);
1722

1823
public abstract String getGroup();
1924

@@ -40,4 +45,9 @@ public final boolean equals(Object obj) {
4045
&& Objects.equals(getKind(), other.getKind())
4146
&& Objects.equals(getName(), other.getName());
4247
}
48+
49+
@Override
50+
public int compareTo(LocalRef<T> o) {
51+
return COMPARATOR.compare(this, o);
52+
}
4353
}

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

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@
3131
import io.fabric8.kubernetes.api.model.OwnerReference;
3232
import io.fabric8.kubernetes.api.model.OwnerReferenceBuilder;
3333
import io.fabric8.kubernetes.client.CustomResource;
34+
import io.fabric8.kubernetes.model.annotation.Group;
35+
import io.fabric8.kubernetes.model.annotation.Singular;
3436
import io.javaoperatorsdk.operator.api.reconciler.Context;
3537
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
3638
import io.javaoperatorsdk.operator.processing.event.ResourceID;
@@ -204,8 +206,8 @@ static <T extends HasMetadata> Stream<T> resourcesInSameNamespace(EventSourceCon
204206
.stream();
205207
}
206208

207-
static <T> boolean isReferent(LocalRef<T> ref, HasMetadata proxy) {
208-
return Objects.equals(ResourcesUtil.name(proxy), ref.getName());
209+
static <T> boolean isReferent(LocalRef<T> ref, HasMetadata resource) {
210+
return Objects.equals(ResourcesUtil.name(resource), ref.getName());
209211
}
210212

211213
/**
@@ -221,6 +223,13 @@ static <O extends HasMetadata, R extends HasMetadata> Set<ResourceID> localRefAs
221223
return Set.of(new ResourceID(ref.getName(), owner.getMetadata().getNamespace()));
222224
}
223225

226+
@NonNull
227+
static <O extends HasMetadata, R extends HasMetadata> Set<ResourceID> localRefsAsResourceIds(O owner, Optional<List<? extends LocalRef<R>>> refs) {
228+
return refs.orElse(List.of()).stream()
229+
.map(ref -> new ResourceID(ref.getName(), owner.getMetadata().getNamespace()))
230+
.collect(Collectors.toSet());
231+
}
232+
224233
/**
225234
* Finds the (ids of) the resources which reference the given referent
226235
* This is the inverse of {@link #localRefAsResourceId(HasMetadata, LocalRef)}}.
@@ -243,6 +252,27 @@ static <O extends HasMetadata, R extends HasMetadata> Set<ResourceID> findReferr
243252
primary -> ResourcesUtil.isReferent(refAccessor.apply(primary), referent));
244253
}
245254

255+
/**
256+
* Like {@link #findReferrers(EventSourceContext, HasMetadata, Class, Function)}
257+
* except for the case where the owner is able to reference multiple referents (i.e. {@code refAccessor} returns a Collection.
258+
* @param context The context
259+
* @param referent The potential referent
260+
* @param owner The type of the owner of the reference
261+
* @param refAccessor A function which returns the references from a given owner.
262+
* @return The ids of reference owners which refer to the referent.
263+
* @param <O> The type of the reference owner
264+
* @param <R> The type of the referent
265+
*/
266+
static <O extends HasMetadata, R extends HasMetadata> Set<ResourceID> findReferrersMulti(EventSourceContext<?> context,
267+
R referent,
268+
Class<O> owner,
269+
Function<O, Collection<? extends LocalRef<R>>> refAccessor) {
270+
return ResourcesUtil.filteredResourceIdsInSameNamespace(context,
271+
referent,
272+
owner,
273+
primary -> refAccessor.apply(primary).stream().anyMatch(ref -> ResourcesUtil.isReferent(ref, referent)));
274+
}
275+
246276
static List<Condition> maybeAddOrUpdateCondition(List<Condition> conditions, Condition condition) {
247277
var type = Objects.requireNonNull(condition.getType());
248278

@@ -422,4 +452,15 @@ static Condition resolvedRefsUnknown(Clock clock,
422452
return newUnknownCondition(clock, observedGenerationSource, Condition.Type.ResolvedRefs, e);
423453
}
424454

455+
static String slug(String singular, String group, String name) {
456+
return singular + "." + group + "/" + name;
457+
}
458+
459+
static String slug(Class<? extends CustomResource<?, ?>> annotatedCrdClass, String crName) {
460+
return slug(
461+
annotatedCrdClass.getAnnotation(Singular.class).value(),
462+
annotatedCrdClass.getAnnotation(Group.class).value(),
463+
crName);
464+
}
465+
425466
}

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

Lines changed: 168 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,18 @@
77
package io.kroxylicious.kubernetes.operator;
88

99
import java.time.Clock;
10+
import java.util.Collection;
1011
import java.util.List;
12+
import java.util.Optional;
13+
import java.util.TreeSet;
14+
import java.util.function.Function;
15+
import java.util.stream.Collectors;
16+
import java.util.stream.Stream;
1117

1218
import org.slf4j.Logger;
1319
import org.slf4j.LoggerFactory;
1420

21+
import io.fabric8.kubernetes.client.CustomResource;
1522
import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
1623
import io.javaoperatorsdk.operator.api.reconciler.Context;
1724
import io.javaoperatorsdk.operator.api.reconciler.ErrorStatusUpdateControl;
@@ -22,17 +29,38 @@
2229
import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
2330

2431
import io.kroxylicious.kubernetes.api.common.Condition;
32+
import io.kroxylicious.kubernetes.api.common.LocalRef;
2533
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
34+
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
35+
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngressStatus;
36+
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
37+
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceStatus;
2638
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
39+
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterSpec;
40+
import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
41+
import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilterStatus;
42+
43+
import edu.umd.cs.findbugs.annotations.NonNull;
2744

2845
import static io.kroxylicious.kubernetes.operator.ResourcesUtil.name;
2946
import static io.kroxylicious.kubernetes.operator.ResourcesUtil.namespace;
3047

48+
/**
49+
* Reconciles a {@link VirtualKafkaCluster} by checking whether the resources
50+
* referenced by the {@code spec.proxyRef.name}, {@code spec.targetClusterRef.name},
51+
* {@code spec.ingressRefs[].name} and {@code spec.filterRefs[].name} actually exist,
52+
* setting a {@link Condition.Type#ResolvedRefs} {@link Condition} accordingly.
53+
*/
3154
public final class VirtualKafkaClusterReconciler implements
3255
Reconciler<VirtualKafkaCluster> {
3356

3457
private static final Logger LOGGER = LoggerFactory.getLogger(VirtualKafkaClusterReconciler.class);
3558
public static final String PROXY_EVENT_SOURCE_NAME = "proxy";
59+
public static final String SERVICES_EVENT_SOURCE_NAME = "services";
60+
public static final String INGRESSES_EVENT_SOURCE_NAME = "ingresses";
61+
public static final String FILTERS_EVENT_SOURCE_NAME = "filters";
62+
public static final String TRANSITIVELY_REFERENCED_RESOURCES_NOT_FOUND = "TransitivelyReferencedResourcesNotFound";
63+
public static final String REFERENCED_RESOURCES_NOT_FOUND = "ReferencedResourcesNotFound";
3664

3765
private final Clock clock;
3866

@@ -42,18 +70,85 @@ public VirtualKafkaClusterReconciler(Clock clock) {
4270

4371
@Override
4472
public UpdateControl<VirtualKafkaCluster> reconcile(VirtualKafkaCluster cluster, Context<VirtualKafkaCluster> context) {
45-
var proxyOpt = context.getSecondaryResource(KafkaProxy.class, PROXY_EVENT_SOURCE_NAME);
46-
LOGGER.debug("spec.proxyRef.name resolves to: {}", proxyOpt);
73+
var existingProxies = context.getSecondaryResource(KafkaProxy.class, PROXY_EVENT_SOURCE_NAME).stream().collect(Collectors.toSet());
74+
TreeSet<LocalRef<KafkaProxy>> missingProxies = Optional.ofNullable(cluster.getSpec())
75+
.map(VirtualKafkaClusterSpec::getProxyRef)
76+
.stream()
77+
.collect(Collectors.toCollection(TreeSet::new));
78+
missingProxies.removeAll(existingProxies.stream()
79+
.map(ResourcesUtil::toLocalRef)
80+
.toList());
81+
82+
var existingServices = context.getSecondaryResource(KafkaService.class, SERVICES_EVENT_SOURCE_NAME).stream().collect(Collectors.toSet());
83+
TreeSet<LocalRef<KafkaService>> missingServices = Optional.ofNullable(cluster.getSpec())
84+
.map(VirtualKafkaClusterSpec::getTargetKafkaServiceRef)
85+
.stream()
86+
.collect(Collectors.toCollection(TreeSet::new));
87+
missingServices.removeAll(existingServices.stream()
88+
.map(ResourcesUtil::toLocalRef)
89+
.toList());
90+
91+
var existingIngresses = context.getSecondaryResources(KafkaProxyIngress.class);
92+
TreeSet<LocalRef<KafkaProxyIngress>> missingIngresses = Optional.ofNullable(cluster.getSpec())
93+
.map(VirtualKafkaClusterSpec::getIngressRefs)
94+
.stream().flatMap(Collection::stream)
95+
.collect(Collectors.toCollection(TreeSet::new));
96+
missingIngresses.removeAll(existingIngresses.stream()
97+
.map(ResourcesUtil::toLocalRef)
98+
.toList());
99+
100+
var existingFilters = context.getSecondaryResources(KafkaProtocolFilter.class);
101+
TreeSet<LocalRef<KafkaProtocolFilter>> missingFilters = Optional.ofNullable(cluster.getSpec())
102+
.map(VirtualKafkaClusterSpec::getFilterRefs)
103+
.stream().flatMap(Collection::stream)
104+
.collect(Collectors.toCollection(TreeSet::new));
105+
missingFilters.removeAll(existingFilters.stream()
106+
.map(ResourcesUtil::toLocalRef)
107+
.toList());
47108

48109
Condition condition;
49-
if (proxyOpt.isPresent()) {
50-
condition = ResourcesUtil.newResolvedRefsTrue(clock, cluster);
110+
if (missingProxies.isEmpty()
111+
&& missingServices.isEmpty()
112+
&& missingIngresses.isEmpty()
113+
&& missingFilters.isEmpty()) {
114+
var unresolvedServices = existingServices.stream()
115+
.filter(ks -> hasAnyResolvedRefsFalse(Optional.ofNullable(ks.getStatus()).map(KafkaServiceStatus::getConditions).orElse(List.of())))
116+
.map(ResourcesUtil::toLocalRef)
117+
.collect(Collectors.toCollection(TreeSet::new));
118+
var unresolvedIngresses = existingIngresses.stream()
119+
.filter(kafkaProxyIngress -> hasAnyResolvedRefsFalse(
120+
Optional.ofNullable(kafkaProxyIngress.getStatus()).map(KafkaProxyIngressStatus::getConditions).orElse(List.of())))
121+
.map(ResourcesUtil::toLocalRef)
122+
.collect(Collectors.toCollection(TreeSet::new));
123+
var unresolvedFilters = existingFilters.stream()
124+
.filter(kpf -> hasAnyResolvedRefsFalse(Optional.ofNullable(kpf.getStatus()).map(KafkaProtocolFilterStatus::getConditions).orElse(List.of())))
125+
.map(ResourcesUtil::toLocalRef)
126+
.collect(Collectors.toCollection(TreeSet::new));
127+
if (unresolvedServices.isEmpty()
128+
&& unresolvedIngresses.isEmpty()
129+
&& unresolvedFilters.isEmpty()) {
130+
condition = ResourcesUtil.newResolvedRefsTrue(clock, cluster);
131+
}
132+
else {
133+
Stream<String> serviceMsg = refsMessage("spec.targetKafkaServiceRef references ", KafkaService.class, unresolvedServices);
134+
Stream<String> ingressMsg = refsMessage("spec.ingressRefs references ", KafkaProxyIngress.class, unresolvedIngresses);
135+
Stream<String> filterMsg = refsMessage("spec.filterRefs references ", KafkaProtocolFilter.class, unresolvedFilters);
136+
condition = ResourcesUtil.newResolvedRefsFalse(clock,
137+
cluster,
138+
TRANSITIVELY_REFERENCED_RESOURCES_NOT_FOUND,
139+
joiningMessages(serviceMsg, ingressMsg, filterMsg));
140+
}
51141
}
52142
else {
143+
Stream<String> proxyMsg = refsMessage("spec.proxyRef references ", KafkaProxy.class, missingProxies);
144+
Stream<String> serviceMsg = refsMessage("spec.targetKafkaServiceRef references ", KafkaService.class, missingServices);
145+
Stream<String> ingressMsg = refsMessage("spec.ingressRefs references ", KafkaProxyIngress.class, missingIngresses);
146+
Stream<String> filterMsg = refsMessage("spec.filterRefs references ", KafkaProtocolFilter.class, missingFilters);
147+
53148
condition = ResourcesUtil.newResolvedRefsFalse(clock,
54149
cluster,
55-
"spec.proxyRef.name",
56-
"KafkaProxy not found");
150+
REFERENCED_RESOURCES_NOT_FOUND,
151+
joiningMessages(proxyMsg, serviceMsg, ingressMsg, filterMsg));
57152
}
58153

59154
UpdateControl<VirtualKafkaCluster> uc = UpdateControl.patchStatus(ResourcesUtil.patchWithCondition(cluster, condition));
@@ -63,9 +158,33 @@ public UpdateControl<VirtualKafkaCluster> reconcile(VirtualKafkaCluster cluster,
63158
return uc;
64159
}
65160

161+
@NonNull
162+
private static String joiningMessages(
163+
Stream<String>... serviceMsg) {
164+
return Stream.of(serviceMsg).flatMap(Function.identity()).collect(Collectors.joining("; "));
165+
}
166+
167+
private static boolean hasAnyResolvedRefsFalse(List<Condition> conditions) {
168+
return conditions.stream()
169+
.anyMatch(c -> Condition.Type.ResolvedRefs.equals(c.getType())
170+
&& Condition.Status.FALSE.equals(c.getStatus()));
171+
}
172+
173+
@NonNull
174+
private static <R extends CustomResource<?, ?>> Stream<String> refsMessage(
175+
String prefix,
176+
Class<R> crdClass,
177+
TreeSet<? extends LocalRef<R>> refs) {
178+
return refs.isEmpty() ? Stream.of()
179+
: Stream.of(
180+
prefix + refs.stream()
181+
.map(ref -> ResourcesUtil.slug(crdClass, ref.getName()))
182+
.collect(Collectors.joining(", ")));
183+
}
184+
66185
@Override
67186
public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSourceContext<VirtualKafkaCluster> context) {
68-
InformerEventSourceConfiguration<KafkaProxy> configuration = InformerEventSourceConfiguration.from(
187+
InformerEventSourceConfiguration<KafkaProxy> clusterToProxy = InformerEventSourceConfiguration.from(
69188
KafkaProxy.class,
70189
VirtualKafkaCluster.class)
71190
.withName(PROXY_EVENT_SOURCE_NAME)
@@ -75,7 +194,48 @@ public List<EventSource<?, VirtualKafkaCluster>> prepareEventSources(EventSource
75194
VirtualKafkaCluster.class,
76195
cluster -> cluster.getSpec().getProxyRef()))
77196
.build();
78-
return List.of(new InformerEventSource<>(configuration, context));
197+
198+
InformerEventSourceConfiguration<KafkaService> clusterToService = InformerEventSourceConfiguration.from(
199+
KafkaService.class,
200+
VirtualKafkaCluster.class)
201+
.withName(SERVICES_EVENT_SOURCE_NAME)
202+
.withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefAsResourceId(cluster,
203+
cluster.getSpec().getTargetKafkaServiceRef()))
204+
.withSecondaryToPrimaryMapper(service -> ResourcesUtil.findReferrers(context,
205+
service,
206+
VirtualKafkaCluster.class,
207+
cluster -> cluster.getSpec().getTargetKafkaServiceRef()))
208+
.build();
209+
210+
InformerEventSourceConfiguration<KafkaProxyIngress> clusterToIngresses = InformerEventSourceConfiguration.from(
211+
KafkaProxyIngress.class,
212+
VirtualKafkaCluster.class)
213+
.withName(INGRESSES_EVENT_SOURCE_NAME)
214+
.withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefsAsResourceIds(cluster,
215+
Optional.ofNullable(cluster.getSpec()).map(VirtualKafkaClusterSpec::getIngressRefs)))
216+
.withSecondaryToPrimaryMapper(ingress -> ResourcesUtil.findReferrersMulti(context,
217+
ingress,
218+
VirtualKafkaCluster.class,
219+
cluster -> cluster.getSpec().getIngressRefs()))
220+
.build();
221+
222+
InformerEventSourceConfiguration<KafkaProtocolFilter> clusterToFilters = InformerEventSourceConfiguration.from(
223+
KafkaProtocolFilter.class,
224+
VirtualKafkaCluster.class)
225+
.withName(FILTERS_EVENT_SOURCE_NAME)
226+
.withPrimaryToSecondaryMapper((VirtualKafkaCluster cluster) -> ResourcesUtil.localRefsAsResourceIds(cluster,
227+
Optional.ofNullable(cluster.getSpec()).map(VirtualKafkaClusterSpec::getFilterRefs)))
228+
.withSecondaryToPrimaryMapper(filter -> ResourcesUtil.findReferrersMulti(context,
229+
filter,
230+
VirtualKafkaCluster.class,
231+
cluster -> cluster.getSpec().getFilterRefs()))
232+
.build();
233+
234+
return List.of(
235+
new InformerEventSource<>(clusterToProxy, context),
236+
new InformerEventSource<>(clusterToIngresses, context),
237+
new InformerEventSource<>(clusterToService, context),
238+
new InformerEventSource<>(clusterToFilters, context));
79239
}
80240

81241
@Override

‎kroxylicious-operator/src/test/java/io/kroxylicious/kubernetes/api/common/FilterRefTest.java‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,4 +25,22 @@ void shouldEqualAnyRefWithSameCoordinates() {
2525
assertThat(filterRefFoo).hasSameHashCodeAs(anyFoo);
2626
}
2727

28-
}
28+
@SuppressWarnings({ "rawtypes", "unchecked" })
29+
@Test
30+
void shouldCompareEqualAnyRefWithSameCoordinates() {
31+
LocalRef refFoo = new FilterRefBuilder().withName("foo").build();
32+
LocalRef anyFoo = new AnyLocalRefBuilder().withName("foo").withKind(refFoo.getKind()).withGroup(refFoo.getGroup()).build();
33+
assertThat(refFoo).isEqualByComparingTo(anyFoo);
34+
assertThat(anyFoo).isEqualByComparingTo(refFoo);
35+
}
36+
37+
@SuppressWarnings({ "rawtypes", "unchecked" })
38+
@Test
39+
void shouldCompareLessThatAnyRefWithSameCoordinates() {
40+
LocalRef refFoo = new FilterRefBuilder().withName("foo").build();
41+
LocalRef anyFoo = new AnyLocalRefBuilder().withName("fooa").withKind(refFoo.getKind()).withGroup(refFoo.getGroup()).build();
42+
assertThat(refFoo).isLessThan(anyFoo);
43+
assertThat(anyFoo).isGreaterThan(refFoo);
44+
}
45+
46+
}

0 commit comments

Comments
 (0)