This is an automated email from the ASF dual-hosted git repository.
amaechler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new f2f26f5c15d feat: Add Kubernetes node-label dynamic config provider
(#20404)
f2f26f5c15d is described below
commit f2f26f5c15d4f4c0ac6e9a40383b7cb746559de3
Author: Andreas Maechler <[email protected]>
AuthorDate: Tue Sep 29 05:20:28 2026 -0600
feat: Add Kubernetes node-label dynamic config provider (#20404)
Adds K8sNodeLabelDynamicConfigProvider to druid-kubernetes-extensions,
registered as the "k8sNodeLabel" DynamicConfigProvider subtype. It resolves
configuration values from labels on the Kubernetes node a process runs on,
which lets a spec express configuration that depends on where a task is
scheduled rather than on the spec itself.
The motivating case is Kafka's client.rack. An ingestion task is a
short-lived
pod that can land in any zone, and the downward API exposes the labels of
the
pod but not of the node, so the zone has to be read from the API server
where
the task ends up. The provider is generic, though: any property can be
filled
from any node label.
The pod names its own node through a downward-API environment variable
(HOST_NODE_NAME by default) and its service account needs a ClusterRole
granting get on nodes. The node is read with the extension's existing
io.kubernetes client: ClientBuilder.cluster() already supplies the endpoint
from the environment, the trust material from the projected cluster CA, and
a
token source that re-reads the projected token as the kubelet rotates it.
---
docs/development/extensions-core/kubernetes.md | 2 +
docs/operations/dynamic-config-provider.md | 97 ++++-
.../druid/k8s/config/ApiNodeLabelReader.java | 126 ++++++
.../druid/k8s/config/CachingNodeLabelReader.java | 62 +++
.../config/K8sNodeLabelDynamicConfigProvider.java | 159 ++++++++
.../apache/druid/k8s/config/NodeLabelReader.java | 37 ++
.../druid/k8s/discovery/K8sDiscoveryModule.java | 12 +-
.../K8sNodeLabelDynamicConfigProviderTest.java | 438 +++++++++++++++++++++
website/.spelling | 6 +
9 files changed, 936 insertions(+), 3 deletions(-)
diff --git a/docs/development/extensions-core/kubernetes.md
b/docs/development/extensions-core/kubernetes.md
index 776545ff21b..46d4a1b1e8e 100644
--- a/docs/development/extensions-core/kubernetes.md
+++ b/docs/development/extensions-core/kubernetes.md
@@ -26,6 +26,8 @@ Consider this an [EXPERIMENTAL](../experimental.md) feature
mostly because it ha
Apache® Druid extension to enable using the Kubernetes® API
Server for node discovery and leader election. This extension allows you to
deploy a Druid cluster on Kubernetes without Zookeeper and run multiple Druid
clusters within the same Kubernetes cluster. For more information, see the
`clusterIdentifier` configuration below.
+This extension also provides the `k8sNodeLabel` [dynamic config
provider](../../operations/dynamic-config-provider.md#kubernetes-node-label-dynamic-config-provider),
which resolves configuration values from the labels of the Kubernetes node a
Druid process runs on. You can load the extension for that provider alone. The
discovery and leader election bindings above are only instantiated when
`druid.discovery.type` is set to `k8s`.
+
## Configuration
diff --git a/docs/operations/dynamic-config-provider.md
b/docs/operations/dynamic-config-provider.md
index 92b9a47494d..f9193b43127 100644
--- a/docs/operations/dynamic-config-provider.md
+++ b/docs/operations/dynamic-config-provider.md
@@ -76,4 +76,99 @@ When you define the consumer properties in the supervisor
spec, use the dynamic
},
...
```
-When connecting to Kafka, Druid replaces the environment variables with their
corresponding values.
\ No newline at end of file
+When connecting to Kafka, Druid replaces the environment variables with their
corresponding values.
+
+## Kubernetes node label dynamic config provider
+
+The
[`druid-kubernetes-extensions`](../development/extensions-core/kubernetes.md)
extension provides a dynamic config provider
(`K8sNodeLabelDynamicConfigProvider`) that reads configuration values from the
labels of the Kubernetes node that the Druid process runs on. Use it for values
that depend on Kubernetes runtime information, for example where a process was
scheduled, which cannot be written into a spec ahead of time.
+
+The Kubernetes node label dynamic config provider uses the following syntax:
+
+```json
+druid.dynamic.config.provider={"type": "k8sNodeLabel","labels":{"property1":
"example.com/label-one","property2": "example.com/label-two"}}
+```
+
+|Field|Type|Description|Required|
+|-----|----|-----------|--------|
+|`type`|String|dynamic config provider type|Yes: `k8sNodeLabel`|
+|`labels`|Map|configuration keys to resolve, each mapped to the name of the
node label that holds its value|Yes|
+|`nodeNameVariable`|String|environment variable that holds the name of the
node the process runs on|No (default: `HOST_NODE_NAME`)|
+
+You can use it anywhere Druid accepts a dynamic config provider, for example
Kafka [`consumerProperties`](../ingestion/kafka-ingestion.md), Iceberg
[`catalogProperties`](../development/extensions-contrib/iceberg.md), and Schema
Registry [`config` and `headers`](../ingestion/data-formats.md).
+
+### Prerequisites
+
+Include `druid-kubernetes-extensions` in the [extensions load
list](../configuration/extensions.md#loading-extensions) of every service that
resolves the spec. For a supervisor spec, that is the Overlord and the Peon
services.
+
+Each pod must know the node it runs on. Expose the node name from the downward
API as the environment variable named by `nodeNameVariable`, or set
`nodeNameVariable` to a variable your pod template already provides:
+
+```yaml
+env:
+ - name: HOST_NODE_NAME
+ valueFrom:
+ fieldRef:
+ fieldPath: spec.nodeName
+```
+
+Nodes are cluster-scoped, so the service account the Druid pods run as needs a
ClusterRole granting `get` on `nodes`:
+
+```yaml
+kind: ClusterRole
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: druid-node-reader
+rules:
+- apiGroups:
+ - ""
+ resources:
+ - nodes
+ verbs:
+ - get
+---
+kind: ClusterRoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+ name: druid-node-reader
+subjects:
+- kind: ServiceAccount
+ name: default # the service account your Druid pods use
+ namespace: druid # their namespace
+roleRef:
+ kind: ClusterRole
+ name: druid-node-reader
+ apiGroup: rbac.authorization.k8s.io
+```
+
+### Behavior
+
+- If the node name variable is unset, the API server cannot be reached or
refuses the request, or the node does not carry the label, Druid logs a warning
and omits the key. The consumer then applies its own default. A label with an
empty value counts as missing. As with other dynamic config providers, a
resolved value takes precedence over the same key specified directly.
+- Druid reads the node once per process and keeps its labels until the process
restarts. A failed lookup is not kept, so it is retried the next time the
configuration is resolved.
+
+### Examples
+
+Kafka consumers fetch from the nearest replica when `client.rack` matches a
broker's `broker.rack`
([KIP-392](https://cwiki.apache.org/confluence/display/KAFKA/KIP-392%3A+Allow+consumers+to+fetch+from+closest+replica)).
On EKS, `topology.k8s.aws/zone-id` holds the zone ID that MSK uses for
`broker.rack`; on GKE, `topology.kubernetes.io/zone` holds the zone name that
Google Cloud Managed Service for Apache Kafka uses:
+
+```json
+"consumerProperties": {
+ "bootstrap.servers": "localhost:9092",
+ "druid.dynamic.config.provider": {
+ "type": "k8sNodeLabel",
+ "labels": { "client.rack": "topology.k8s.aws/zone-id" }
+ }
+}
+```
+
+An Iceberg REST catalog can read from S3 in the task's own region by taking
`client.region` from the node's region label:
+
+```json
+"icebergCatalog": {
+ "type": "rest",
+ "catalogUri": "https://iceberg-rest.example.com";,
+ "catalogProperties": {
+ "druid.dynamic.config.provider": {
+ "type": "k8sNodeLabel",
+ "labels": { "client.region": "topology.kubernetes.io/region" }
+ }
+ }
+}
+```
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/ApiNodeLabelReader.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/ApiNodeLabelReader.java
new file mode 100644
index 00000000000..dcb0545d79c
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/ApiNodeLabelReader.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.k8s.config;
+
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Node;
+import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.util.ClientBuilder;
+import org.apache.druid.java.util.common.logger.Logger;
+
+import javax.annotation.Nullable;
+
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Reads the labels of a Kubernetes node from the in-cluster API server.
+ * <p>
+ * {@link ClientBuilder#cluster()} supplies the endpoint from the environment,
the trust material from the projected
+ * cluster CA, and a token source that re-reads the projected token as the
kubelet rotates it.
+ */
+class ApiNodeLabelReader implements NodeLabelReader
+{
+ private static final Logger log = new Logger(ApiNodeLabelReader.class);
+
+ // A missing label is survivable, a slow task start is not.
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private final CoreV1Api api;
+
+ ApiNodeLabelReader(ApiClient apiClient, Duration timeout)
+ {
+ apiClient.setHttpClient(
+ apiClient.getHttpClient()
+ .newBuilder()
+ // Bounds the whole call. A read timeout alone lets a stalled
response body run on far longer.
+ .callTimeout(timeout)
+ // Keeps the bearer token from reaching any host but the API
server.
+ .followRedirects(false)
+ .followSslRedirects(false)
+ .build()
+ );
+ this.api = new CoreV1Api(apiClient);
+ }
+
+ /**
+ * Returns the labels of the named node. Returns null if the read failed, or
if the client could not be built,
+ * which is the case outside Kubernetes.
+ */
+ @Nullable
+ static Map<String, String> readFromCluster(String nodeName)
+ {
+ final ApiClient apiClient;
+ try {
+ apiClient = ClientBuilder.cluster().build();
+ }
+ catch (Exception e) {
+ // Missing credentials or endpoint is configuration rather than a bug,
so the cause is named without a trace.
+ log.warn("Could not build a Kubernetes API client [%s], so the labels of
node [%s] cannot be read.", e, nodeName);
+ return null;
+ }
+
+ try {
+ return new ApiNodeLabelReader(apiClient, TIMEOUT).readLabels(nodeName);
+ }
+ finally {
+ // OkHttp is not closeable, and this client made its one call, so drop
the pooled socket now rather than
+ // leave it reachable from the shared connection pool until it idles out.
+ apiClient.getHttpClient().connectionPool().evictAll();
+ }
+ }
+
+ @Nullable
+ @Override
+ public Map<String, String> readLabels(String nodeName)
+ {
+ try {
+ final V1Node node = api.readNode(nodeName, null);
+ final V1ObjectMeta metadata = node == null ? null : node.getMetadata();
+ if (metadata == null) {
+ // Not a node, so a failure. An empty map here would be cached as if
the node had no labels.
+ log.warn("Kubernetes API server returned no object metadata for node
[%s].", nodeName);
+ return null;
+ }
+
+ final Map<String, String> labels = metadata.getLabels();
+ return labels == null ? Collections.emptyMap() :
Collections.unmodifiableMap(new HashMap<>(labels));
+ }
+ catch (Exception e) {
+ // An ApiException without a status code is a connection failure or the
call timeout, not an answer.
+ if (e instanceof ApiException apiException && apiException.getCode() !=
0) {
+ log.warn(
+ "Kubernetes API server answered [%d] when asked for node [%s]: %s."
+ + " Reading a node requires a ClusterRole granting 'get' on
'nodes'.",
+ apiException.getCode(),
+ nodeName,
+ apiException.getResponseBody()
+ );
+ } else {
+ log.warn(e, "Could not read the labels of node [%s] from the
Kubernetes API server.", nodeName);
+ }
+ return null;
+ }
+ }
+}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/CachingNodeLabelReader.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/CachingNodeLabelReader.java
new file mode 100644
index 00000000000..056e8a8fd05
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/CachingNodeLabelReader.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.k8s.config;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Remembers what a delegate read, so that a node is asked about once per
process.
+ * <p>
+ * Druid builds a {@link org.apache.druid.metadata.DynamicConfigProvider} for
every use and a task builds several
+ * consumers, so without this each of them would call the API server again. A
failed read is not remembered, so a
+ * transient error is retried by the next caller.
+ */
+class CachingNodeLabelReader implements NodeLabelReader
+{
+ private final NodeLabelReader delegate;
+ private final ConcurrentHashMap<String, Map<String, String>> labelsByNode =
new ConcurrentHashMap<>();
+
+ CachingNodeLabelReader(NodeLabelReader delegate)
+ {
+ this.delegate = delegate;
+ }
+
+ @Nullable
+ @Override
+ public Map<String, String> readLabels(String nodeName)
+ {
+ final Map<String, String> cached = labelsByNode.get(nodeName);
+ if (cached != null) {
+ return cached;
+ }
+
+ // Not computeIfAbsent: the delegate blocks on the network, and a mapping
function must not hold the map's bin
+ // lock for that long. Two callers racing here cost one extra request and
agree on the result.
+ final Map<String, String> read = delegate.readLabels(nodeName);
+ if (read == null) {
+ return null;
+ }
+ labelsByNode.put(nodeName, read);
+ return read;
+ }
+}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProvider.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProvider.java
new file mode 100644
index 00000000000..fffbe07d8be
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProvider.java
@@ -0,0 +1,159 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.k8s.config;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableMap;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.metadata.DynamicConfigProvider;
+
+import javax.annotation.Nullable;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * Resolves configuration from the labels of the Kubernetes node that this
process runs on. For example, a Kafka
+ * supervisor can set {@code client.rack} from the node's zone label, which is
not known until the task is scheduled.
+ * <p>
+ * {@link #getConfig()} reads the node the first time it runs and keeps the
answer for the life of the process, so a
+ * node relabelled later is not picked up until restart.
+ */
+public class K8sNodeLabelDynamicConfigProvider implements
DynamicConfigProvider<String>
+{
+ private static final Logger log = new
Logger(K8sNodeLabelDynamicConfigProvider.class);
+
+ @VisibleForTesting
+ static final String DEFAULT_NODE_NAME_VARIABLE = "HOST_NODE_NAME";
+
+ /**
+ * Shared so that the labels of this process's node are read once, however
many providers are built.
+ */
+ private static final NodeLabelReader IN_CLUSTER_READER =
+ new CachingNodeLabelReader(ApiNodeLabelReader::readFromCluster);
+
+ private final ImmutableMap<String, String> labels;
+ private final String nodeNameVariable;
+ private final NodeLabelReader nodeLabelReader;
+ private final Function<String, String> env;
+
+ @JsonCreator
+ public K8sNodeLabelDynamicConfigProvider(
+ @JsonProperty("labels") Map<String, String> labels,
+ @JsonProperty("nodeNameVariable") @Nullable String nodeNameVariable
+ )
+ {
+ this(labels, nodeNameVariable, IN_CLUSTER_READER, System::getenv);
+ }
+
+ @VisibleForTesting
+ K8sNodeLabelDynamicConfigProvider(
+ Map<String, String> labels,
+ @Nullable String nodeNameVariable,
+ NodeLabelReader nodeLabelReader,
+ Function<String, String> env
+ )
+ {
+ this.labels = ImmutableMap.copyOf(Preconditions.checkNotNull(labels,
"labels"));
+ this.nodeNameVariable = nodeNameVariable == null ?
DEFAULT_NODE_NAME_VARIABLE : nodeNameVariable;
+ this.nodeLabelReader = nodeLabelReader;
+ this.env = env;
+ }
+
+ @JsonProperty("labels")
+ public Map<String, String> getLabels()
+ {
+ return labels;
+ }
+
+ @JsonProperty("nodeNameVariable")
+ public String getNodeNameVariable()
+ {
+ return nodeNameVariable;
+ }
+
+ @Override
+ public Map<String, String> getConfig()
+ {
+ final String nodeName = env.apply(nodeNameVariable);
+ if (nodeName == null || nodeName.isBlank()) {
+ log.warn(
+ "Environment variable [%s] is not set, so the node this process runs
on is unknown. Skipping %s.",
+ nodeNameVariable,
+ labels.keySet()
+ );
+ return Collections.emptyMap();
+ }
+
+ final Map<String, String> nodeLabels =
nodeLabelReader.readLabels(nodeName);
+ if (nodeLabels == null) {
+ // The reader has already said why.
+ return Collections.emptyMap();
+ }
+
+ final Map<String, String> config = new HashMap<>();
+ for (Map.Entry<String, String> entry : labels.entrySet()) {
+ final String value = nodeLabels.get(entry.getValue());
+ // Kubernetes allows a label with an empty value, which marks the node
without saying anything about it.
+ if (value == null || value.isEmpty()) {
+ log.warn("Node [%s] has no value for label [%s], skipping key [%s].",
nodeName, entry.getValue(), entry.getKey());
+ continue;
+ }
+ config.put(entry.getKey(), value);
+ }
+ return config;
+ }
+
+ @Override
+ public String toString()
+ {
+ return "K8sNodeLabelDynamicConfigProvider{" +
+ "labels=" + labels +
+ ", nodeNameVariable='" + nodeNameVariable + '\'' +
+ '}';
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ K8sNodeLabelDynamicConfigProvider that =
(K8sNodeLabelDynamicConfigProvider) o;
+
+ return Objects.equals(labels, that.labels) &&
Objects.equals(nodeNameVariable, that.nodeNameVariable);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(labels, nodeNameVariable);
+ }
+}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/NodeLabelReader.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/NodeLabelReader.java
new file mode 100644
index 00000000000..66a41b8a030
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/config/NodeLabelReader.java
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.k8s.config;
+
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
+/**
+ * Reads the labels of a single Kubernetes node.
+ */
+@FunctionalInterface
+interface NodeLabelReader
+{
+ /**
+ * Returns every label of the named node, or null if the labels could not be
read.
+ */
+ @Nullable
+ Map<String, String> readLabels(String nodeName);
+}
diff --git
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDiscoveryModule.java
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDiscoveryModule.java
index b18f03d7f5b..7f80c830952 100644
---
a/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDiscoveryModule.java
+++
b/extensions-core/kubernetes-extensions/src/main/java/org/apache/druid/k8s/discovery/K8sDiscoveryModule.java
@@ -20,6 +20,9 @@
package org.apache.druid.k8s.discovery;
import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import com.google.inject.Key;
import io.kubernetes.client.openapi.ApiClient;
@@ -33,9 +36,9 @@ import org.apache.druid.guice.JsonConfigProvider;
import org.apache.druid.guice.LazySingleton;
import org.apache.druid.guice.PolyBind;
import org.apache.druid.initialization.DruidModule;
+import org.apache.druid.k8s.config.K8sNodeLabelDynamicConfigProvider;
import java.io.IOException;
-import java.util.Collections;
import java.util.List;
public class K8sDiscoveryModule implements DruidModule
@@ -45,7 +48,12 @@ public class K8sDiscoveryModule implements DruidModule
@Override
public List<? extends Module> getJacksonModules()
{
- return Collections.emptyList();
+ return ImmutableList.of(
+ new SimpleModule("K8sDynamicConfigProviderModule")
+ .registerSubtypes(
+ new NamedType(K8sNodeLabelDynamicConfigProvider.class,
"k8sNodeLabel")
+ )
+ );
}
@Override
diff --git
a/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProviderTest.java
b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProviderTest.java
new file mode 100644
index 00000000000..14e0de3331e
--- /dev/null
+++
b/extensions-core/kubernetes-extensions/src/test/java/org/apache/druid/k8s/config/K8sNodeLabelDynamicConfigProviderTest.java
@@ -0,0 +1,438 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.k8s.config;
+
+import com.fasterxml.jackson.databind.Module;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.collect.ImmutableMap;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+import io.kubernetes.client.openapi.ApiClient;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.k8s.discovery.K8sDiscoveryModule;
+import org.apache.druid.metadata.DynamicConfigProvider;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class K8sNodeLabelDynamicConfigProviderTest
+{
+ private static final String NODE_NAME =
"ip-10-0-0-1.us-west-2.compute.internal";
+ private static final String ZONE_ID_LABEL = "topology.k8s.aws/zone-id";
+ private static final String ZONE_NAME_LABEL = "topology.kubernetes.io/zone";
+ private static final String CLIENT_RACK = "client.rack";
+ private static final String TOKEN = "test-token";
+ private static final String DEFAULT_VARIABLE =
K8sNodeLabelDynamicConfigProvider.DEFAULT_NODE_NAME_VARIABLE;
+
+ private static final Map<String, String> RACK_FROM_ZONE_ID =
ImmutableMap.of(CLIENT_RACK, ZONE_ID_LABEL);
+
+ private static final String LABELED_NODE =
+ "{\"metadata\":{\"name\":\"node\",\"labels\":{\"" + ZONE_ID_LABEL +
"\":\"usw2-az1\","
+ +
"\"kubernetes.io/os\":\"linux\",\"node-role.kubernetes.io/worker\":\"\"}}}";
+ private static final String UNLABELED_NODE =
"{\"metadata\":{\"labels\":{\"kubernetes.io/os\":\"linux\"}}}";
+
+ private final AtomicInteger requestCount = new AtomicInteger();
+ private final List<ApiClient> apiClients = new ArrayList<>();
+ private final CountDownLatch teardown = new CountDownLatch(1);
+
+ private HttpServer server;
+ private URI apiServerUri;
+ private volatile String requestPath;
+ private volatile String authorizationHeader;
+
+ @BeforeEach
+ public void setUp() throws IOException
+ {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ apiServerUri = URI.create("http://127.0.0.1:"; +
server.getAddress().getPort());
+ }
+
+ @AfterEach
+ public void tearDown()
+ {
+ // Lets a stalling handler finish so that stop() does not join a sleeping
thread.
+ teardown.countDown();
+ server.stop(0);
+ apiClients.forEach(c -> c.getHttpClient().connectionPool().evictAll());
+ }
+
+ @Test
+ public void testFullNodeResponse()
+ {
+ // The whole node object, which is what the API server sends when nothing
trims the representation.
+ serve(200,
"{\"kind\":\"Node\",\"apiVersion\":\"v1\",\"metadata\":{\"name\":\"node\",\"resourceVersion\":\"123\",\"creationTimestamp\":\"2024-01-01T00:00:00Z\",\"labels\":{\"topology.k8s.aws/zone-id\":\"usw2-az1\",\"kubernetes.io/os\":\"linux\"}},\"spec\":{\"podCIDR\":\"10.0.0.0/24\",\"providerID\":\"aws:///us-west-2a/i-0abc\"},\"status\":{\"capacity\":{\"cpu\":\"8\",\"memory\":\"32Gi\"},\"images\":[{\"names\":[\"apache/druid:latest\"],\"sizeBytes\":123456789}],\"conditions\":[{\"typ
[...]
+
+ Assertions.assertEquals(ImmutableMap.of(CLIENT_RACK, "usw2-az1"),
provider(RACK_FROM_ZONE_ID).getConfig());
+ }
+
+ @Test
+ public void testNodeIsReadWithBearerToken()
+ {
+ serve(200, LABELED_NODE);
+
+ provider(RACK_FROM_ZONE_ID).getConfig();
+
+ Assertions.assertEquals("/api/v1/nodes/" + NODE_NAME, requestPath);
+ Assertions.assertEquals("Bearer " + TOKEN, authorizationHeader);
+ }
+
+ @Test
+ public void testLabelAbsent()
+ {
+ serve(200, UNLABELED_NODE);
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID).getConfig());
+ }
+
+ @Test
+ public void testOnlyTheLabelsTheNodeHasAreReturned()
+ {
+ serve(200, LABELED_NODE);
+
+ Assertions.assertEquals(
+ ImmutableMap.of(CLIENT_RACK, "usw2-az1"),
+ provider(ImmutableMap.of(CLIENT_RACK, ZONE_ID_LABEL, "some.property",
ZONE_NAME_LABEL)).getConfig()
+ );
+ }
+
+ @Test
+ public void testEmptyLabelValueIsSkipped()
+ {
+ serve(200, LABELED_NODE);
+
+ Assertions.assertEquals(
+ Collections.emptyMap(),
+ provider(ImmutableMap.of("some.property",
"node-role.kubernetes.io/worker")).getConfig()
+ );
+ }
+
+ @Test
+ public void testForbidden()
+ {
+ serve(403, "{\"kind\":\"Status\",\"code\":403}");
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID).getConfig());
+ }
+
+ @Test
+ public void testUnparseableResponse()
+ {
+ serve(200, "not json");
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID).getConfig());
+ }
+
+ @Test
+ public void testResponseWithoutMetadataIsTreatedAsAFailure()
+ {
+ serve(200, "{\"kind\":\"Node\",\"apiVersion\":\"v1\"}");
+ final NodeLabelReader reader = new
CachingNodeLabelReader(reader(Duration.ofSeconds(5)));
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+
+ // A 200 that does not describe a node is a failure, not an answer, so it
must not be cached.
+ Assertions.assertEquals(2, requestCount.get());
+ }
+
+ /**
+ * A server that answers and then stalls mid-body. The client's own read
timeout would eventually catch this, but
+ * only after its default, so the reader bounds the whole call instead. The
elapsed assertion is what holds that
+ * bound in place: without it the test still passes, just ten seconds later.
+ */
+ @Test
+ @Timeout(value = 30_000, unit = TimeUnit.MILLISECONDS)
+ public void testStalledResponseBodyTimesOut()
+ {
+ server.createContext("/", exchange -> {
+ requestCount.incrementAndGet();
+ exchange.sendResponseHeaders(200, LABELED_NODE.length());
+ exchange.getResponseBody().write("{".getBytes(StandardCharsets.UTF_8));
+ exchange.getResponseBody().flush();
+ awaitTeardown();
+ });
+ server.start();
+
+ // Built before the clock starts so that class loading is not counted
against the bound.
+ final K8sNodeLabelDynamicConfigProvider provider =
provider(RACK_FROM_ZONE_ID, reader(Duration.ofMillis(200)));
+
+ final long startMillis = System.currentTimeMillis();
+ Assertions.assertEquals(Collections.emptyMap(), provider.getConfig());
+ final long elapsedMillis = System.currentTimeMillis() - startMillis;
+
+ Assertions.assertEquals(1, requestCount.get());
+ Assertions.assertTrue(elapsedMillis < 3000, "gave up only after " +
elapsedMillis + " ms");
+ }
+
+ @Test
+ public void testNodeWithNoLabelsAtAll()
+ {
+ serve(200, "{\"metadata\":{\"name\":\"node\"}}");
+ final NodeLabelReader reader = new
CachingNodeLabelReader(reader(Duration.ofSeconds(5)));
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+
+ // A node carrying no labels is an answer, so it is cached like any other.
+ Assertions.assertEquals(1, requestCount.get());
+ }
+
+ @Test
+ public void testEmptyResponseBody()
+ {
+ serve(200, "");
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID).getConfig());
+ }
+
+ @Test
+ public void testRedirectIsNotFollowed()
+ {
+ server.createContext("/", exchange -> {
+ requestCount.incrementAndGet();
+ exchange.getResponseHeaders().set("Location",
"http://example.invalid/api/v1/nodes/"; + NODE_NAME);
+ exchange.sendResponseHeaders(302, -1);
+ exchange.close();
+ });
+ server.start();
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID).getConfig());
+
+ // Following it would carry the bearer token to whatever host the redirect
names.
+ Assertions.assertEquals(1, requestCount.get());
+ }
+
+ @Test
+ public void testNodeNameVariableUnsetOrBlank()
+ {
+ serve(200, LABELED_NODE);
+ final NodeLabelReader reader = reader(Duration.ofSeconds(5));
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, DEFAULT_VARIABLE, null, reader).getConfig());
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, DEFAULT_VARIABLE, " ", reader).getConfig());
+ Assertions.assertEquals(0, requestCount.get());
+ }
+
+ @Test
+ public void testCustomNodeNameVariable()
+ {
+ serve(200, LABELED_NODE);
+
+ Assertions.assertEquals(
+ ImmutableMap.of(CLIENT_RACK, "usw2-az1"),
+ provider(RACK_FROM_ZONE_ID, "NODE_NAME", NODE_NAME,
reader(Duration.ofSeconds(5))).getConfig()
+ );
+ }
+
+ @Test
+ public void testSuccessfulLookupIsCached()
+ {
+ serve(200, LABELED_NODE);
+ final NodeLabelReader reader = new
CachingNodeLabelReader(reader(Duration.ofSeconds(5)));
+
+ final Map<String, String> expected = ImmutableMap.of(CLIENT_RACK,
"usw2-az1");
+ Assertions.assertEquals(expected, provider(RACK_FROM_ZONE_ID,
reader).getConfig());
+ Assertions.assertEquals(expected, provider(RACK_FROM_ZONE_ID,
reader).getConfig());
+
+ Assertions.assertEquals(1, requestCount.get());
+ }
+
+ @Test
+ public void testNodeWithoutTheLabelIsStillCached()
+ {
+ serve(200, UNLABELED_NODE);
+ final NodeLabelReader reader = new
CachingNodeLabelReader(reader(Duration.ofSeconds(5)));
+
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+ Assertions.assertEquals(Collections.emptyMap(),
provider(RACK_FROM_ZONE_ID, reader).getConfig());
+
+ // The node answered; it simply has no such label. That is an answer, so
it is not asked again.
+ Assertions.assertEquals(1, requestCount.get());
+ }
+
+ @Test
+ public void testFailedLookupIsNotCached()
+ {
+ serve(403, "{\"kind\":\"Status\",\"code\":403}");
+ final NodeLabelReader reader = new
CachingNodeLabelReader(reader(Duration.ofSeconds(5)));
+
+ provider(RACK_FROM_ZONE_ID, reader).getConfig();
+ provider(RACK_FROM_ZONE_ID, reader).getConfig();
+
+ Assertions.assertEquals(2, requestCount.get());
+ }
+
+ @Test
+ public void testSerde() throws Exception
+ {
+ final ObjectMapper mapper = new DefaultObjectMapper();
+ for (Module module : new K8sDiscoveryModule().getJacksonModules()) {
+ mapper.registerModule(module);
+ }
+
+ final DynamicConfigProvider<?> defaulted = mapper.readValue(
+
"{\"type\":\"k8sNodeLabel\",\"labels\":{\"client.rack\":\"topology.k8s.aws/zone-id\"}}",
+ DynamicConfigProvider.class
+ );
+ Assertions.assertEquals(new
K8sNodeLabelDynamicConfigProvider(RACK_FROM_ZONE_ID, null), defaulted);
+ Assertions.assertEquals(
+ DEFAULT_VARIABLE,
+ ((K8sNodeLabelDynamicConfigProvider) defaulted).getNodeNameVariable()
+ );
+ Assertions.assertEquals(
+ defaulted,
+ mapper.readValue(mapper.writeValueAsString(defaulted),
DynamicConfigProvider.class)
+ );
+
+ final DynamicConfigProvider<?> explicit = mapper.readValue(
+
"{\"type\":\"k8sNodeLabel\",\"labels\":{\"client.rack\":\"topology.kubernetes.io/zone\"},"
+ + "\"nodeNameVariable\":\"NODE_NAME\"}",
+ DynamicConfigProvider.class
+ );
+ Assertions.assertEquals(
+ new K8sNodeLabelDynamicConfigProvider(ImmutableMap.of(CLIENT_RACK,
ZONE_NAME_LABEL), "NODE_NAME"),
+ explicit
+ );
+ Assertions.assertEquals(
+ explicit,
+ mapper.readValue(mapper.writeValueAsString(explicit),
DynamicConfigProvider.class)
+ );
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ final K8sNodeLabelDynamicConfigProvider provider =
+ new K8sNodeLabelDynamicConfigProvider(RACK_FROM_ZONE_ID, null);
+ final K8sNodeLabelDynamicConfigProvider same =
+ new K8sNodeLabelDynamicConfigProvider(RACK_FROM_ZONE_ID,
DEFAULT_VARIABLE);
+ final K8sNodeLabelDynamicConfigProvider otherLabel =
+ new K8sNodeLabelDynamicConfigProvider(ImmutableMap.of(CLIENT_RACK,
ZONE_NAME_LABEL), null);
+ final K8sNodeLabelDynamicConfigProvider otherVariable =
+ new K8sNodeLabelDynamicConfigProvider(RACK_FROM_ZONE_ID, "NODE_NAME");
+
+ Assertions.assertEquals(provider, same);
+ Assertions.assertEquals(provider.hashCode(), same.hashCode());
+ Assertions.assertNotEquals(provider, otherLabel);
+ Assertions.assertNotEquals(provider, otherVariable);
+ Assertions.assertNotEquals(provider, null);
+ }
+
+ @Test
+ public void testLabelsAreRequired()
+ {
+ Assertions.assertThrows(
+ NullPointerException.class,
+ () -> new K8sNodeLabelDynamicConfigProvider(null, null)
+ );
+ }
+
+ private void serve(int status, String body)
+ {
+ server.createContext("/", exchange -> {
+ requestCount.incrementAndGet();
+ requestPath = exchange.getRequestURI().getPath();
+ authorizationHeader =
exchange.getRequestHeaders().getFirst("Authorization");
+ respond(exchange, status, body);
+ });
+ server.start();
+ }
+
+ private static void respond(HttpExchange exchange, int status, String body)
throws IOException
+ {
+ final byte[] bytes = body.getBytes(StandardCharsets.UTF_8);
+ exchange.getResponseHeaders().set("Content-Type", "application/json");
+ exchange.sendResponseHeaders(status, bytes.length);
+ try (OutputStream out = exchange.getResponseBody()) {
+ out.write(bytes);
+ }
+ }
+
+ /**
+ * Holds a handler open until the test is over, so a stalling server does
not also stall the suite.
+ */
+ private void awaitTeardown()
+ {
+ try {
+ teardown.await(30, TimeUnit.SECONDS);
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private K8sNodeLabelDynamicConfigProvider provider(Map<String, String>
labels)
+ {
+ return provider(labels, reader(Duration.ofSeconds(5)));
+ }
+
+ private K8sNodeLabelDynamicConfigProvider provider(Map<String, String>
labels, NodeLabelReader reader)
+ {
+ return provider(labels, DEFAULT_VARIABLE, NODE_NAME, reader);
+ }
+
+ /**
+ * The fake environment answers only for {@code variable}, with {@code
nodeName}, or null for an unset variable.
+ */
+ private K8sNodeLabelDynamicConfigProvider provider(
+ Map<String, String> labels,
+ String variable,
+ @Nullable String nodeName,
+ NodeLabelReader reader
+ )
+ {
+ return new K8sNodeLabelDynamicConfigProvider(labels, variable, reader, var
-> variable.equals(var) ? nodeName : null);
+ }
+
+ private NodeLabelReader reader(Duration timeout)
+ {
+ // Production gets its token from an OkHttp interceptor installed by
ClientBuilder, not from a default header,
+ // so the test uses an interceptor too. A default header would survive the
client swap either way.
+ final ApiClient apiClient = new
ApiClient().setBasePath(apiServerUri.toString());
+ apiClient.setHttpClient(
+ apiClient.getHttpClient()
+ .newBuilder()
+ .addInterceptor(chain -> chain.proceed(
+ chain.request().newBuilder().header("Authorization",
"Bearer " + TOKEN).build()
+ ))
+ .build()
+ );
+ apiClients.add(apiClient);
+ return new ApiNodeLabelReader(apiClient, timeout);
+ }
+}
diff --git a/website/.spelling b/website/.spelling
index 8d9fa585d9a..5c7d8ba7d5c 100644
--- a/website/.spelling
+++ b/website/.spelling
@@ -67,6 +67,7 @@ CentralizedDatasourceSchema
Ceph
circledR
CloudWatch
+ClusterRole
ColumnDescriptor
Corretto
CLI
@@ -104,6 +105,7 @@ DynamicConfigProvider
EC2
EC2ContainerCredentialsProviderWrapper
ECS
+EKS
EMR
EMRFS
ETL
@@ -118,6 +120,7 @@ Float.POSITIVE_INFINITY
Float.POSITIVE_INFINITY.
ForwardedRequestCustomizer
GC
+GKE
GPG
GoogleCloudStorageInputSource
GSSAPI
@@ -179,6 +182,7 @@ Joda
JsonProperty
Jupyter
JupyterLab
+KIP-392
KMS
Kerberized
Kerberos
@@ -205,6 +209,7 @@ Marquez
MiddleManager
MiddleManagers
Montréal
+MSK
MSQ
Murmur3
MVCC
@@ -422,6 +427,7 @@ kbps
kerberos
keystore
keytool
+kubelet
keytab
kubeconfig
kubernetes
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
