Allow using patterns to match multiple Secret or ConfigMap fields

Signed-off-by: Jakub Scholz <[email protected]>

README.md

@@ -23,55 +23,55 @@ The following example shows how to use it with Kafka Connect and Connectors:
     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnect
     metadata:
-      name: my-connect
-      annotations:
-        strimzi.io/use-connector-resources: "true"
+      name: my-connect
+      annotations:
+        strimzi.io/use-connector-resources: "true"
     spec:
-      # ...
-      config:
-        # ...
-        config.providers: secrets,configmaps
-        config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
-        config.providers.configmaps.class: io.strimzi.kafka.KubernetesConfigMapConfigProvider
-      # ...
+      # ...
+      config:
+        # ...
+        config.providers: secrets,configmaps
+        config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
+        config.providers.configmaps.class: io.strimzi.kafka.KubernetesConfigMapConfigProvider
+      # ...
     ```
 
 2) Create a configuration Config Map
     ```yaml
     apiVersion: v1
     kind: ConfigMap
     metadata:
-      name: connector-configuration
+      name: connector-configuration
     data:
-      option1: value1
-      option2: value2
+      option1: value1
+      option2: value2
     ```
 
 3) Create the Role and RoleBinding:
     ```yaml
     apiVersion: rbac.authorization.k8s.io/v1
     kind: Role
     metadata:
-      name: connector-configuration-role
+      name: connector-configuration-role
     rules:
     - apiGroups: [""]
-      resources: ["configmaps"]
-      resourceNames: ["connector-configuration"]
-      verbs: ["get"]
+      resources: ["configmaps"]
+      resourceNames: ["connector-configuration"]
+      verbs: ["get"]
     ---
 
     apiVersion: rbac.authorization.k8s.io/v1
     kind: RoleBinding
     metadata:
-      name: connector-configuration-role-binding
+      name: connector-configuration-role-binding
     subjects:
     - kind: ServiceAccount
-      name: my-connect-connect
-      namespace: myproject
+      name: my-connect-connect
+      namespace: myproject
     roleRef:
-      kind: Role
-      name: connector-configuration-role
-      apiGroup: rbac.authorization.k8s.io
+      kind: Role
+      name: connector-configuration-role
+      apiGroup: rbac.authorization.k8s.io
     ```
 
     Use the Service Account already used by your Kafka Connect deployment, which is named `<CLUSTER_NAME>-connect` where `<CLUSTER_NAME>` is the name of your KafkaConnect custom resource.
@@ -81,14 +81,14 @@ The following example shows how to use it with Kafka Connect and Connectors:
     apiVersion: kafka.strimzi.io/v1beta2
     kind: KafkaConnector
     metadata:
-      name: my-connector
-      labels:
-        strimzi.io/cluster: my-connect
+      name: my-connector
+      labels:
+        strimzi.io/cluster: my-connect
     spec:
-      # ...
-      config:
-        option: ${configmaps:myproject/connector-configuration:option1}
-        # ...
+      # ...
+      config:
+        option: ${configmaps:myproject/connector-configuration:option1}
+        # ...
     ```
 
 ## Adding the Kubernetes Configuration Provider to Apache Kafka clients
@@ -170,6 +170,41 @@ The following example shows how to use it in a Kafka Consumer consuming from Apa
     ssl.truststore.certificates=${secrets:myproject/my-cluster-cluster-ca-cert:ca.crt}
     ```
 
+## Using patterns
+
+You can also use [Glob patterns](https://en.wikipedia.org/wiki/Glob_(programming)) to specify the keys in the Secret or Config Map that should be used.
+The Kubernetes Config Provider will find all fields matching the pattern and join them together into some value.
+This is useful for example when you want to load multiple certificates from a Secret into a single field.
+With an example Kubernetes Secret container multiple certificate files:
+
+```yaml
+apiVersion: v1
+kind: Secret
+metadata:
+  name: certificates
+data:
+  ca.crt: ...
+  ca2.crt: ...
+  ca3.crt: ...
+```
+
+You can use the provider to load all of the certificates by using the key pattern `*.crt`:
+
+```properties
+config.providers=secrets
+config.providers.secrets.class=io.strimzi.kafka.KubernetesSecretConfigProvider
+...
+ssl.truststore.certificates=${secrets:myproject/certificates:*.crt}
+```
+
+By default, a new line will be used as a separator when joining the different values.
+But you can configure the separator if you want to use a different one:
+  ```properties
+  config.providers=secrets,configmaps
+  config.providers.secrets.class=io.strimzi.kafka.KubernetesSecretConfigProvider
+  config.providers.secrets.param.separator=,
+  ```
+
 ## Configuring the Kubernetes client
 
 The Kubernetes Config Provider is using the [Fabric8 Kubernetes Client](https://github.com/fabric8io/kubernetes-client).

src/main/java/io/strimzi/kafka/AbstractKubernetesConfigProvider.java

@@ -18,9 +18,13 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.nio.file.FileSystems;
+import java.nio.file.PathMatcher;
+import java.nio.file.Paths;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 /**
  * Abstract class for Kafka configuration providers using Kubernetes resources
@@ -31,10 +35,12 @@
  */
 abstract class AbstractKubernetesConfigProvider<T extends HasMetadata, L extends KubernetesResourceList<T>, R extends Resource<T>> implements ConfigProvider {
     private static final Logger LOG = LoggerFactory.getLogger(AbstractKubernetesConfigProvider.class);
+    private static final String SEPARATOR_CONFIG_NAME = "separator";
 
     protected final String kind;
 
     protected KubernetesClient client;
+    private String separator = System.lineSeparator();
 
     /**
      * Creates the configuration provider
@@ -58,8 +64,13 @@ public void close() throws IOException {
     }
 
     @Override
-    public void configure(Map<String, ?> map) {
-        LOG.info("Configuring Kubernetes {} config provider", kind);
+    public void configure(Map<String, ?> config) {
+        LOG.info("Configuring Kubernetes {} config provider with configuration {}", kind, config);
+
+        if (config.get(SEPARATOR_CONFIG_NAME) != null) {
+            separator = (String) config.get(SEPARATOR_CONFIG_NAME);
+        }
+
         client = new KubernetesClientBuilder().build();
     }
 
@@ -88,10 +99,9 @@ private ConfigData getValues(String path, Set<String> keys)    {
         if (keys == null)   {
             configs.putAll(values);
         } else {
-            for (Map.Entry<String, String> entry : values.entrySet())   {
-                if (keys.contains(entry.getKey()))  {
-                    configs.put(entry.getKey(), entry.getValue());
-                }
+            for (String key : keys) {
+                PathMatcher pathMatcher = FileSystems.getDefault().getPathMatcher("glob:" + key);
+                configs.put(key, values.entrySet().stream().filter(entry -> pathMatcher.matches(Paths.get(entry.getKey()))).sorted(Map.Entry.comparingByKey()).map(Map.Entry::getValue).collect(Collectors.joining(separator)));
             }
         }
 

src/test/java/io/strimzi/kafka/KubernetesConfigMapProviderIT.java

@@ -45,9 +45,9 @@ public static void beforeAll()   {
                     .withName(RESOURCE_NAME)
                     .withNamespace(namespace)
                 .endMetadata()
-                .addToData("test-key-1", "test-value-1")
-                .addToData("test-key-2", "test-value-2")
-                .addToData("test-key-3", "test-value-3")
+                .addToData("test.config", "test-value-1")
+                .addToData("test2.config", "test-value-2")
+                .addToData("test.properties", "test-value-3")
                 .build();
 
         client.configMaps().resource(cm).create();
@@ -65,28 +65,47 @@ public void testAllValues() {
         Map<String, String> data = config.data();
 
         assertThat(data.size(), is(3));
-        assertThat(data.get("test-key-1"), is("test-value-1"));
-        assertThat(data.get("test-key-2"), is("test-value-2"));
-        assertThat(data.get("test-key-3"), is("test-value-3"));
+        assertThat(data.get("test.config"), is("test-value-1"));
+        assertThat(data.get("test2.config"), is("test-value-2"));
+        assertThat(data.get("test.properties"), is("test-value-3"));
     }
 
     @Test
     public void testSomeValues() {
-        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, new HashSet<>(Arrays.asList("test-key-1", "test-key-3")));
+        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, new HashSet<>(Arrays.asList("test.config", "test.properties", "*.config")));
         Map<String, String> data = config.data();
 
-        assertThat(data.size(), is(2));
-        assertThat(data.get("test-key-1"), is("test-value-1"));
-        assertThat(data.get("test-key-3"), is("test-value-3"));
+        assertThat(data.size(), is(3));
+        assertThat(data.get("test.config"), is("test-value-1"));
+        assertThat(data.get("test.properties"), is("test-value-3"));
+        assertThat(data.get("*.config"), is("test-value-1" + System.lineSeparator() + "test-value-2"));
     }
 
     @Test
     public void testOneValue() {
-        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, Collections.singleton("test-key-2"));
+        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, Collections.singleton("test2.config"));
+        Map<String, String> data = config.data();
+
+        assertThat(data.size(), is(1));
+        assertThat(data.get("test2.config"), is("test-value-2"));
+    }
+
+    @Test
+    public void testPatternValue() {
+        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, Collections.singleton("*.config"));
+        Map<String, String> data = config.data();
+
+        assertThat(data.size(), is(1));
+        assertThat(data.get("*.config"), is("test-value-1" + System.lineSeparator() + "test-value-2"));
+    }
+
+    @Test
+    public void testNotMatchingPatternValue() {
+        ConfigData config = provider.get(namespace + "/" + RESOURCE_NAME, Collections.singleton("*.cfg"));
         Map<String, String> data = config.data();
 
         assertThat(data.size(), is(1));
-        assertThat(data.get("test-key-2"), is("test-value-2"));
+        assertThat(data.get("*.cfg"), is(""));
     }
 
     @Test
@@ -95,9 +114,9 @@ public void testDefaultNamespace() {
         Map<String, String> data = config.data();
 
         assertThat(data.size(), is(3));
-        assertThat(data.get("test-key-1"), is("test-value-1"));
-        assertThat(data.get("test-key-2"), is("test-value-2"));
-        assertThat(data.get("test-key-3"), is("test-value-3"));
+        assertThat(data.get("test.config"), is("test-value-1"));
+        assertThat(data.get("test2.config"), is("test-value-2"));
+        assertThat(data.get("test.properties"), is("test-value-3"));
     }
 
     @Test
@@ -106,4 +125,16 @@ public void testNonExistentConfigMap() {
         assertThrows(ConfigException.class, () -> provider.get("i-do-not-exist/i-do-not-exist-either"));
         assertThrows(ConfigException.class, () -> provider.get("i-do-not-exist"));
     }
+
+    @Test
+    public void testCustomSeparator() {
+        KubernetesConfigMapConfigProvider customProvider = new KubernetesConfigMapConfigProvider();
+        customProvider.configure(Map.of("separator", ";"));
+
+        ConfigData config = customProvider.get(namespace + "/" + RESOURCE_NAME, Collections.singleton("*.config"));
+        Map<String, String> data = config.data();
+
+        assertThat(data.size(), is(1));
+        assertThat(data.get("*.config"), is("test-value-1;test-value-2"));
+    }
 }