Skip to content

Commit 79c3b3f

Browse files
tombentleytombentley
authored
Generate proxy configuration honouring the KafkaService tls object (kroxylicious#2070)
* Hook up client and ca certificates in the proxy and in the volumes and mounts * cipher suites and protocols * Add an IT Signed-off-by: Tom Bentley <[email protected]>
1 parent 8033296 commit 79c3b3f

19 files changed

+675
-51
lines changed

‎kroxylicious-operator/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -168,6 +168,11 @@
168168
<artifactId>kroxylicious-runtime</artifactId>
169169
<scope>compile</scope>
170170
</dependency>
171+
<dependency>
172+
<groupId>io.kroxylicious</groupId>
173+
<artifactId>kroxylicious-api</artifactId>
174+
<scope>compile</scope>
175+
</dependency>
171176

172177
<dependency>
173178
<groupId>io.micrometer</groupId>

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

Lines changed: 48 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,16 +6,58 @@
66

77
package io.kroxylicious.kubernetes.operator;
88

9+
import java.util.Optional;
910
import java.util.Set;
11+
import java.util.function.BiFunction;
12+
import java.util.function.Function;
13+
import java.util.stream.Collectors;
14+
import java.util.stream.Stream;
1015

1116
import io.fabric8.kubernetes.api.model.Volume;
1217
import io.fabric8.kubernetes.api.model.VolumeMount;
1318

1419
/**
15-
* A piece of proxy configuration which references files on the proxy container filesystem
16-
* @param fragment The piece of proxy configuration
17-
* @param volumes The volumes the configuration depends on
18-
* @param mounts The mount the configuration depends on
19-
* @param <F> The type of fragment
20+
* A piece of proxy configuration which references files on the proxy container filesystem.
21+
* @param fragment The piece of proxy configuration.
22+
* @param volumes The volumes the configuration depends on.
23+
* @param mounts The mount the configuration depends on.
24+
* @param <F> The type of fragment.
2025
*/
21-
public record ConfigurationFragment<F>(F fragment, Set<Volume> volumes, Set<VolumeMount> mounts) {}
26+
record ConfigurationFragment<F>(F fragment, Set<Volume> volumes, Set<VolumeMount> mounts) {
27+
28+
/**
29+
* @return An empty optional fragment.
30+
* @param <F> The fragment type.
31+
*/
32+
static <F> ConfigurationFragment<Optional<F>> empty() {
33+
return new ConfigurationFragment<>(Optional.empty(), Set.of(), Set.of());
34+
}
35+
36+
/**
37+
* Combine two fragments into a new fragment by applying the given function and taking the union of their volumes and mounts.
38+
* @param x The first fragment to combine.
39+
* @param y The first fragment to combine.
40+
* @param fn The combining function.
41+
* @return The new fragment.
42+
* @param <X> The type of the first fragment.
43+
* @param <Y> The type of the second fragment.
44+
* @param <F> The type of the result fragment.
45+
*/
46+
public static <X, Y, F> ConfigurationFragment<F> combine(ConfigurationFragment<X> x, ConfigurationFragment<Y> y, BiFunction<X, Y, F> fn) {
47+
var t = fn.apply(x.fragment(), y.fragment());
48+
return new ConfigurationFragment<>(t,
49+
Stream.concat(x.volumes.stream(), y.volumes.stream()).collect(Collectors.toSet()),
50+
Stream.concat(x.mounts.stream(), y.mounts.stream()).collect(Collectors.toSet()));
51+
}
52+
53+
/**
54+
* Apply the given mapping function to the fragment, returning a new instance with the same volumes and mounts.
55+
* @param mapper The mapping function.
56+
* @return A new fragment.
57+
* @param <G> The type of the new fragment.
58+
*/
59+
public <G> ConfigurationFragment<G> map(Function<F, G> mapper) {
60+
return new ConfigurationFragment<>(mapper.apply(fragment), volumes, mounts);
61+
}
62+
63+
}

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

Lines changed: 112 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
*/
66
package io.kroxylicious.kubernetes.operator;
77

8+
import java.nio.file.Path;
89
import java.time.Clock;
910
import java.util.ArrayList;
1011
import java.util.Collection;
@@ -19,12 +20,15 @@
1920
import java.util.function.Function;
2021
import java.util.function.Predicate;
2122
import java.util.stream.Collectors;
23+
import java.util.stream.Stream;
2224

2325
import org.slf4j.Logger;
2426
import org.slf4j.LoggerFactory;
2527

2628
import io.fabric8.kubernetes.api.model.Volume;
29+
import io.fabric8.kubernetes.api.model.VolumeBuilder;
2730
import io.fabric8.kubernetes.api.model.VolumeMount;
31+
import io.fabric8.kubernetes.api.model.VolumeMountBuilder;
2832
import io.javaoperatorsdk.operator.OperatorException;
2933
import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration;
3034
import io.javaoperatorsdk.operator.api.reconciler.Context;
@@ -41,14 +45,17 @@
4145
import io.javaoperatorsdk.operator.processing.event.source.SecondaryToPrimaryMapper;
4246
import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource;
4347

48+
import io.kroxylicious.kubernetes.api.common.AnyLocalRef;
4449
import io.kroxylicious.kubernetes.api.common.Condition;
4550
import io.kroxylicious.kubernetes.api.common.FilterRef;
4651
import io.kroxylicious.kubernetes.api.common.LocalRef;
4752
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxy;
4853
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaProxyIngress;
4954
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
55+
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceSpec;
5056
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
5157
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterSpec;
58+
import io.kroxylicious.kubernetes.api.v1alpha1.kafkaservicespec.tls.TrustAnchorRef;
5259
import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilter;
5360
import io.kroxylicious.kubernetes.filter.api.v1alpha1.KafkaProtocolFilterSpec;
5461
import io.kroxylicious.kubernetes.operator.model.ProxyModel;
@@ -63,8 +70,17 @@
6370
import io.kroxylicious.proxy.config.admin.EndpointsConfiguration;
6471
import io.kroxylicious.proxy.config.admin.ManagementConfiguration;
6572
import io.kroxylicious.proxy.config.admin.PrometheusMetricsConfig;
73+
import io.kroxylicious.proxy.config.tls.AllowDeny;
74+
import io.kroxylicious.proxy.config.tls.KeyPair;
75+
import io.kroxylicious.proxy.config.tls.KeyProvider;
76+
import io.kroxylicious.proxy.config.tls.PlatformTrustProvider;
77+
import io.kroxylicious.proxy.config.tls.Tls;
78+
import io.kroxylicious.proxy.config.tls.TrustProvider;
79+
import io.kroxylicious.proxy.config.tls.TrustStore;
6680
import io.kroxylicious.proxy.tag.VisibleForTesting;
6781

82+
import edu.umd.cs.findbugs.annotations.Nullable;
83+
6884
import static io.kroxylicious.kubernetes.operator.ResourcesUtil.name;
6985
import static io.kroxylicious.kubernetes.operator.ResourcesUtil.namespace;
7086
import static io.kroxylicious.kubernetes.operator.ResourcesUtil.toLocalRef;
@@ -104,6 +120,7 @@ public class KafkaProxyReconciler implements
104120
public static final String CONFIG_DEP = "config";
105121
public static final String DEPLOYMENT_DEP = "deployment";
106122
public static final String CLUSTERS_DEP = "clusters";
123+
public static final Path TARGET_CLUSTER_MOUNTS_BASE_DIR = Path.of("/opt/kroxylicious/target-cluster");
107124

108125
private final Clock clock;
109126
private final SecureConfigInterpolator secureConfigInterpolator;
@@ -145,26 +162,25 @@ private ConfigurationFragment<Configuration> generateProxyConfig(ProxyModel mode
145162

146163
var virtualClusters = buildVirtualClusters(namedDefinitions.keySet(), model);
147164

148-
List<NamedFilterDefinition> referencedFilters = virtualClusters.stream().flatMap(c -> Optional.ofNullable(c.filters()).stream().flatMap(Collection::stream))
165+
List<NamedFilterDefinition> referencedFilters = virtualClusters.stream()
166+
.flatMap(vcFragment -> Optional.ofNullable(vcFragment.fragment().filters()).stream().flatMap(Collection::stream))
149167
.distinct()
150168
.map(filterName -> namedDefinitions.get(filterName).fragment()).toList();
151169

152-
var allVolumes = allFilterDefinitions.stream()
170+
var allVolumes = Stream.concat(allFilterDefinitions.stream(), virtualClusters.stream())
153171
.flatMap(fd -> fd.volumes().stream())
154172
.collect(Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(Volume::getName).reversed())));
155173

156-
var allMounts = allFilterDefinitions.stream()
174+
var allMounts = Stream.concat(allFilterDefinitions.stream(), virtualClusters.stream())
157175
.flatMap(fd -> fd.mounts().stream())
158-
.collect(Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(VolumeMount::getName).reversed())));
159-
160-
// TODO add the volume & mounts for each KafkaService's spec.tls
176+
.collect(Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(VolumeMount::getMountPath).reversed())));
161177

162178
return new ConfigurationFragment<>(
163179
new Configuration(
164180
new ManagementConfiguration(null, null, new EndpointsConfiguration(new PrometheusMetricsConfig())),
165181
referencedFilters,
166182
null, // no defaultFilters <= each of the virtualClusters specifies its own
167-
virtualClusters,
183+
virtualClusters.stream().map(ConfigurationFragment::fragment).toList(),
168184
List.of(),
169185
false,
170186
// micrometer
@@ -173,11 +189,11 @@ private ConfigurationFragment<Configuration> generateProxyConfig(ProxyModel mode
173189
allMounts);
174190
}
175191

176-
private static List<VirtualCluster> buildVirtualClusters(Set<String> successfullyBuiltFilterNames, ProxyModel model) {
192+
private static List<ConfigurationFragment<VirtualCluster>> buildVirtualClusters(Set<String> successfullyBuiltFilterNames, ProxyModel model) {
177193
return model.clustersWithValidIngresses().stream()
178194
.filter(cluster -> Optional.ofNullable(cluster.getSpec().getFilterRefs()).stream().flatMap(Collection::stream).allMatch(
179195
filters -> successfullyBuiltFilterNames.contains(filterDefinitionName(filters))))
180-
.map(cluster -> getVirtualCluster(cluster, model.resolutionResult().kafkaServiceRef(cluster).orElseThrow(), model.ingressModel()))
196+
.map(cluster -> buildVirtualCluster(cluster, model.resolutionResult().kafkaServiceRef(cluster).orElseThrow(), model.ingressModel()))
181197
.toList();
182198
}
183199

@@ -230,19 +246,95 @@ private SecureConfigInterpolator.InterpolationResult interpolateConfig(KafkaProt
230246
return secureConfigInterpolator.interpolate(configTemplate);
231247
}
232248

233-
private static VirtualCluster getVirtualCluster(VirtualKafkaCluster cluster,
234-
KafkaService kafkaServiceRef,
235-
ProxyIngressModel ingressModel) {
249+
private static ConfigurationFragment<VirtualCluster> buildVirtualCluster(VirtualKafkaCluster cluster,
250+
KafkaService kafkaServiceRef,
251+
ProxyIngressModel ingressModel) {
236252

237253
ProxyIngressModel.VirtualClusterIngressModel virtualClusterIngressModel = ingressModel.clusterIngressModel(cluster).orElseThrow();
238-
String bootstrap = kafkaServiceRef.getSpec().getBootstrapServers();
239-
return new VirtualCluster(
240-
ResourcesUtil.name(cluster), new TargetCluster(bootstrap, Optional.empty()),
254+
255+
return buildTargetCluster(kafkaServiceRef).map(targetCluster -> new VirtualCluster(
256+
ResourcesUtil.name(cluster),
257+
targetCluster,
241258
null,
242259
Optional.empty(),
243260
virtualClusterIngressModel.gateways(),
244-
false, false,
245-
filterNamesForCluster(cluster));
261+
false,
262+
false,
263+
filterNamesForCluster(cluster)));
264+
}
265+
266+
private static ConfigurationFragment<TargetCluster> buildTargetCluster(KafkaService kafkaServiceRef) {
267+
return buildTargetClusterTls(kafkaServiceRef)
268+
.map(tls -> new TargetCluster(kafkaServiceRef.getSpec().getBootstrapServers(), tls));
269+
}
270+
271+
private static ConfigurationFragment<Optional<Tls>> buildTargetClusterTls(KafkaService kafkaServiceRef) {
272+
return Optional.ofNullable(kafkaServiceRef.getSpec())
273+
.map(KafkaServiceSpec::getTls)
274+
.map(serviceTls -> ConfigurationFragment.combine(
275+
buildKeyProvider(serviceTls.getCertificateRef()),
276+
buildTrustProvider(serviceTls.getTrustAnchorRef()),
277+
(keyProviderOpt, trustProvider) -> Optional.of(
278+
new Tls(keyProviderOpt.orElse(null),
279+
trustProvider,
280+
Optional.ofNullable(serviceTls.getCipherSuites())
281+
.map(cipherSuites -> new AllowDeny<>(cipherSuites.getAllowed(), new HashSet<>(cipherSuites.getDenied())))
282+
.orElse(null),
283+
Optional.ofNullable(serviceTls.getProtocols())
284+
.map(protocols -> new AllowDeny<>(protocols.getAllowed(), new HashSet<>(protocols.getDenied())))
285+
.orElse(null)))))
286+
.orElse(ConfigurationFragment.empty());
287+
}
288+
289+
private static ConfigurationFragment<Optional<KeyProvider>> buildKeyProvider(@Nullable AnyLocalRef certificateRef) {
290+
return Optional.ofNullable(certificateRef)
291+
.filter(ResourcesUtil::isSecret)
292+
.map(ref -> {
293+
var volume = new VolumeBuilder()
294+
.withName(ResourcesUtil.volumeName("", "secrets", ref.getName()))
295+
.withNewSecret()
296+
.withSecretName(ref.getName())
297+
.endSecret()
298+
.build();
299+
Path mountPath = TARGET_CLUSTER_MOUNTS_BASE_DIR.resolve("client-certs").resolve(ref.getName());
300+
var mount = new VolumeMountBuilder()
301+
.withName(ResourcesUtil.volumeName("", "secrets", ref.getName()))
302+
.withMountPath(mountPath.toString())
303+
.withReadOnly(true)
304+
.build();
305+
var keyPath = mountPath.resolve("tls.key");
306+
var crtPath = mountPath.resolve("tls.crt");
307+
return new ConfigurationFragment<>(
308+
Optional.<KeyProvider> of(new KeyPair(keyPath.toString(), crtPath.toString(), null)),
309+
Set.of(volume),
310+
Set.of(mount));
311+
}).orElse(ConfigurationFragment.empty());
312+
}
313+
314+
private static ConfigurationFragment<TrustProvider> buildTrustProvider(@Nullable TrustAnchorRef trustAnchorRef) {
315+
return Optional.ofNullable(trustAnchorRef)
316+
.filter(ResourcesUtil::isConfigMap)
317+
.map(ref -> {
318+
var volume = new VolumeBuilder()
319+
.withName(ResourcesUtil.volumeName("", "configmaps", ref.getName()))
320+
.withNewConfigMap()
321+
.withName(ref.getName())
322+
.endConfigMap()
323+
.build();
324+
Path mountPath = TARGET_CLUSTER_MOUNTS_BASE_DIR.resolve("trusted-certs").resolve(ref.getName());
325+
var mount = new VolumeMountBuilder()
326+
.withName(ResourcesUtil.volumeName("", "configmaps", ref.getName()))
327+
.withMountPath(mountPath.toString())
328+
.withReadOnly(true)
329+
.build();
330+
TrustProvider trustProvider = new TrustStore(
331+
mountPath.resolve(ref.getKey()).toString(),
332+
null,
333+
"PEM");
334+
return new ConfigurationFragment<>(trustProvider,
335+
Set.of(volume),
336+
Set.of(mount));
337+
}).orElse(new ConfigurationFragment<>(PlatformTrustProvider.INSTANCE, Set.of(), Set.of()));
246338
}
247339

248340
/**
@@ -266,7 +358,9 @@ public ErrorStatusUpdateControl<KafkaProxy> updateErrorStatus(KafkaProxy proxy,
266358
Context<KafkaProxy> context,
267359
Exception e) {
268360
if (e instanceof StaleReferentStatusException || e instanceof OperatorException && e.getCause() instanceof StaleReferentStatusException) {
269-
LOGGER.debug("Completed reconciliation of {}/{} with stale referent", namespace(proxy), name(proxy), e);
361+
if (LOGGER.isDebugEnabled()) {
362+
LOGGER.debug("Completed reconciliation of {}/{} with stale referent", namespace(proxy), name(proxy), e);
363+
}
270364
return ErrorStatusUpdateControl.noStatusUpdate();
271365
}
272366
var uc = ErrorStatusUpdateControl.patchStatus(statusFactory.newUnknownConditionStatusPatch(proxy, Condition.Type.Ready, e));

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

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -131,8 +131,7 @@ private KafkaService checkTrustAnchorRef(KafkaService service,
131131
Context<KafkaService> context,
132132
TrustAnchorRef trustAnchorRef) {
133133
String path = "spec.tls.trustAnchorRef";
134-
if ("ConfigMap".equals(Optional.ofNullable(trustAnchorRef.getKind()).orElse("ConfigMap"))
135-
&& Optional.ofNullable(trustAnchorRef.getGroup()).orElse("").isEmpty()) {
134+
if (ResourcesUtil.isConfigMap(trustAnchorRef)) {
136135
Optional<ConfigMap> configMapOpt = context.getSecondaryResource(ConfigMap.class, CONFIG_MAPS_EVENT_SOURCE_NAME);
137136
if (configMapOpt.isEmpty()) {
138137
return statusFactory.newFalseConditionStatusPatch(service, ResolvedRefs,
@@ -173,8 +172,7 @@ private KafkaService checkCertRef(KafkaService service,
173172
Context<KafkaService> context,
174173
AnyLocalRef certRef) {
175174
String path = "spec.tls.certificateRef";
176-
if ("Secret".equals(Optional.ofNullable(certRef.getKind()).orElse("Secret"))
177-
&& Optional.ofNullable(certRef.getGroup()).orElse("").isEmpty()) {
175+
if (ResourcesUtil.isSecret(certRef)) {
178176
Optional<Secret> secretOpt = context.getSecondaryResource(Secret.class, SECRETS_EVENT_SOURCE_NAME);
179177
if (secretOpt.isEmpty()) {
180178
return statusFactory.newFalseConditionStatusPatch(service, ResolvedRefs,

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,15 @@ public class MountedResourceConfigProvider implements SecureConfigProvider {
2424
static final MountedResourceConfigProvider CONFIGMAP_PROVIDER = new MountedResourceConfigProvider("", "configmaps",
2525
(vb, resourceName) -> vb.withNewConfigMap().withName(resourceName).endConfigMap());
2626

27-
private final String volumeNamePrefix;
27+
private final String group;
28+
private final String plural;
2829
private final BiFunction<VolumeBuilder, String, VolumeBuilder> volumeBuilder;
2930

3031
MountedResourceConfigProvider(String group,
3132
String plural,
3233
BiFunction<VolumeBuilder, String, VolumeBuilder> volumeBuilder) {
33-
this.volumeNamePrefix = group.isEmpty() ? plural : group + "." + plural;
34+
this.group = group;
35+
this.plural = plural;
3436
this.volumeBuilder = volumeBuilder;
3537
}
3638

@@ -41,9 +43,7 @@ public ContainerFileReference containerFile(
4143
String key,
4244
Path mountPathBase) {
4345
try {
44-
String volumeName = volumeNamePrefix + "-" + resourceName;
45-
ResourcesUtil.requireIsDnsLabel(volumeName, true,
46-
"volume name would not be a DNS label: " + volumeName);
46+
String volumeName = ResourcesUtil.volumeName(group, plural, resourceName);
4747
Path mountPath = mountPathBase.resolve(providerName).resolve(resourceName);
4848
Path itemPath = mountPath.resolve(key);
4949
Volume volume = volumeBuilder.apply(new VolumeBuilder(), resourceName)
@@ -60,7 +60,7 @@ public ContainerFileReference containerFile(
6060
itemPath);
6161
}
6262
catch (IllegalArgumentException e) {
63-
throw new InterpolationException("Cannot construct mounted volume for ${%s:%s:%s}".formatted(
63+
throw new InterpolationException("Cannot construct mounted volume for ${%s:%s:%s}: %s".formatted(
6464
providerName,
6565
resourceName,
6666
key,

0 commit comments

Comments
 (0)