This is an automated email from the ASF dual-hosted git repository.
gian 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 a8da3814ec1 k8s: Update fabric8 to 6.13.1, switch to Vert.x http
client. (#17913)
a8da3814ec1 is described below
commit a8da3814ec1af24f06e14cc785df1a8f2e02d1d3
Author: Gian Merlino <[email protected]>
AuthorDate: Tue Apr 15 22:16:54 2025 -0700
k8s: Update fabric8 to 6.13.1, switch to Vert.x http client. (#17913)
* k8s: Update fabric8 to 6.13.1, switch to Vert.x http client.
fabric8 6.x uses the OkHttp client by default, which itself uses an
unbounded-size cached pool, with 60s timeout, for making requests.
In production with large task counts, this leads to lots of thread churn.
The Vert.x client is the default in fabric8 7.x, and is better behaved.
By default it uses a worker pool of 20 threads (see javadoc for
VertxOptions).
This patch uses fabric8 6.13.1 rather than some later version, because
higher versions are not compatible with Jackson 2.12.
* Fix call.
* Update licenses.yaml. Override thread pool configs.
* Remove annotation.
* Fix style.
* Update licenses.yaml.
* Add dep.
---
.../kubernetes-overlord-extensions/pom.xml | 33 +++++++--
.../k8s/overlord/KubernetesOverlordModule.java | 20 ++++--
.../k8s/overlord/common/DruidKubernetesClient.java | 20 +++---
.../common/DruidKubernetesHttpClientConfig.java | 50 +++++++++++++
.../common/DruidKubernetesHttpClientFactory.java | 82 ++++++++++++++++++++++
.../overlord/KubernetesTaskRunnerFactoryTest.java | 5 +-
.../DruidPeonClientIntegrationTest.java | 4 +-
licenses.yaml | 17 ++++-
8 files changed, 204 insertions(+), 27 deletions(-)
diff --git a/extensions-core/kubernetes-overlord-extensions/pom.xml
b/extensions-core/kubernetes-overlord-extensions/pom.xml
index eb2fc63f8b2..80e61cec9ee 100644
--- a/extensions-core/kubernetes-overlord-extensions/pom.xml
+++ b/extensions-core/kubernetes-overlord-extensions/pom.xml
@@ -34,6 +34,13 @@
<relativePath>../../pom.xml</relativePath>
</parent>
+ <properties>
+ <!-- fabric8 6.13.2+ is incompatible with jackson 2.12
+ ~ due to fix for
https://github.com/fabric8io/kubernetes-client/issues/6110.
+ ~ GoIntegerDeserializer uses a _parseInteger method that does not exist
in 2.12 (introduced in 2.14) -->
+ <fabric8.version>6.13.1</fabric8.version>
+ </properties>
+
<dependencyManagement>
<dependencies>
<!-- snakeyaml explicitly pinned to version 1.33 as it is
@@ -111,7 +118,7 @@
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-model-core</artifactId>
- <version>6.7.2</version>
+ <version>${fabric8.version}</version>
</dependency>
<dependency>
<groupId>jakarta.validation</groupId>
@@ -121,18 +128,34 @@
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-model-batch</artifactId>
- <version>6.7.2</version>
+ <version>${fabric8.version}</version>
</dependency>
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-client-api</artifactId>
- <version>6.7.2</version>
+ <version>${fabric8.version}</version>
</dependency>
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-client</artifactId>
- <version>6.7.2</version>
+ <version>${fabric8.version}</version>
<scope>runtime</scope>
+ <exclusions>
+ <exclusion>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-httpclient-okhttp</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>io.fabric8</groupId>
+ <artifactId>kubernetes-httpclient-vertx</artifactId>
+ <version>${fabric8.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>io.vertx</groupId>
+ <artifactId>vertx-core</artifactId>
+ <version>4.5.8</version>
</dependency>
<dependency>
<groupId>javax.ws.rs</groupId>
@@ -169,7 +192,7 @@
<dependency>
<groupId>io.fabric8</groupId>
<artifactId>kubernetes-server-mock</artifactId>
- <version>6.4.1</version>
+ <version>${fabric8.version}</version>
<scope>test</scope>
</dependency>
<dependency>
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
index 1b07a34b43c..6320eb7df20 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/KubernetesOverlordModule.java
@@ -57,6 +57,7 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.lifecycle.Lifecycle;
import org.apache.druid.java.util.common.logger.Logger;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
+import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
import
org.apache.druid.k8s.overlord.execution.KubernetesTaskExecutionConfigResource;
import
org.apache.druid.k8s.overlord.execution.KubernetesTaskRunnerDynamicConfig;
import org.apache.druid.k8s.overlord.runnerstrategy.RunnerStrategy;
@@ -85,6 +86,7 @@ public class KubernetesOverlordModule implements DruidModule
+
".k8sAndWorker";
private static final String RUNNERSTRATEGY_PROPERTIES_FORMAT_STRING =
K8SANDWORKER_PROPERTIES_PREFIX
+
".runnerStrategy.%s";
+ private static final String HTTPCLIENT_PROPERITES_PREFIX =
K8SANDWORKER_PROPERTIES_PREFIX + ".http";
@Override
public void configure(Binder binder)
@@ -119,22 +121,28 @@ public class KubernetesOverlordModule implements
DruidModule
configureTaskLogs(binder);
Jerseys.addResource(binder, KubernetesTaskExecutionConfigResource.class);
+
+ JsonConfigProvider.bind(binder, HTTPCLIENT_PROPERITES_PREFIX,
DruidKubernetesHttpClientConfig.class);
}
@Provides
@LazySingleton
- public DruidKubernetesClient makeKubernetesClient(KubernetesTaskRunnerConfig
kubernetesTaskRunnerConfig, Lifecycle lifecycle)
+ public DruidKubernetesClient makeKubernetesClient(
+ KubernetesTaskRunnerConfig kubernetesTaskRunnerConfig,
+ DruidKubernetesHttpClientConfig httpClientConfig,
+ Lifecycle lifecycle
+ )
{
- DruidKubernetesClient client;
+ final DruidKubernetesClient client;
+ final Config config = new ConfigBuilder().build();
+
if (kubernetesTaskRunnerConfig.isDisableClientProxy()) {
- Config config = new ConfigBuilder().build();
config.setHttpsProxy(null);
config.setHttpProxy(null);
- client = new DruidKubernetesClient(config);
- } else {
- client = new DruidKubernetesClient();
}
+ client = new DruidKubernetesClient(httpClientConfig, config);
+
lifecycle.addHandler(
new Lifecycle.Handler()
{
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
index f1f81bed8d8..9c1940bb1f4 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesClient.java
@@ -20,25 +20,19 @@
package org.apache.druid.k8s.overlord.common;
import io.fabric8.kubernetes.client.Config;
-import io.fabric8.kubernetes.client.ConfigBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.KubernetesClientBuilder;
public class DruidKubernetesClient implements KubernetesClientApi
{
-
- private final Config config;
private final KubernetesClient kubernetesClient;
- public DruidKubernetesClient()
- {
- this(new ConfigBuilder().build());
- }
-
- public DruidKubernetesClient(Config config)
+ public DruidKubernetesClient(DruidKubernetesHttpClientConfig
httpClientConfig, Config kubernetesClientConfig)
{
- this.config = config;
- this.kubernetesClient = new
KubernetesClientBuilder().withConfig(config).build();
+ this.kubernetesClient = new KubernetesClientBuilder()
+ .withHttpClientFactory(new
DruidKubernetesHttpClientFactory(httpClientConfig))
+ .withConfig(kubernetesClientConfig)
+ .build();
}
@Override
@@ -47,8 +41,10 @@ public class DruidKubernetesClient implements
KubernetesClientApi
return executor.executeRequest(kubernetesClient);
}
- /** This client automatically gets closed by the druid lifecycle, it should
not be closed when used as it is
+ /**
+ * This client automatically gets closed by the druid lifecycle, it should
not be closed when used as it is
* meant to be reused.
+ *
* @return re-useable KubernetesClient
*/
@Override
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
new file mode 100644
index 00000000000..b659f8e9af5
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientConfig.java
@@ -0,0 +1,50 @@
+/*
+ * 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.overlord.common;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.vertx.core.VertxOptions;
+
+public class DruidKubernetesHttpClientConfig
+{
+ @JsonProperty
+ private int workerPoolSize = VertxOptions.DEFAULT_WORKER_POOL_SIZE;
+
+ @JsonProperty
+ private int eventLoopPoolSize = VertxOptions.DEFAULT_EVENT_LOOP_POOL_SIZE;
+
+ @JsonProperty
+ private int internalBlockingPoolSize =
VertxOptions.DEFAULT_INTERNAL_BLOCKING_POOL_SIZE;
+
+ public int getWorkerPoolSize()
+ {
+ return workerPoolSize;
+ }
+
+ public int getEventLoopPoolSize()
+ {
+ return eventLoopPoolSize;
+ }
+
+ public int getInternalBlockingPoolSize()
+ {
+ return internalBlockingPoolSize;
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
new file mode 100644
index 00000000000..1924eb11d81
--- /dev/null
+++
b/extensions-core/kubernetes-overlord-extensions/src/main/java/org/apache/druid/k8s/overlord/common/DruidKubernetesHttpClientFactory.java
@@ -0,0 +1,82 @@
+/*
+ * 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.overlord.common;
+
+import io.fabric8.kubernetes.client.http.HttpClient;
+import io.fabric8.kubernetes.client.vertx.VertxHttpClientBuilder;
+import io.fabric8.kubernetes.client.vertx.VertxHttpClientFactory;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.file.FileSystemOptions;
+import io.vertx.core.spi.resolver.ResolverProvider;
+
+/**
+ * Similar to {@link VertxHttpClientFactory} but allows us to override thread
pool configurations.
+ */
+public class DruidKubernetesHttpClientFactory implements HttpClient.Factory
+{
+ private final Vertx vertx;
+
+ public DruidKubernetesHttpClientFactory(final
DruidKubernetesHttpClientConfig httpClientConfig)
+ {
+ this.vertx = createVertxInstance(httpClientConfig);
+ }
+
+ @Override
+ public VertxHttpClientBuilder<DruidKubernetesHttpClientFactory> newBuilder()
+ {
+ return new VertxHttpClientBuilder<>(this, vertx);
+ }
+
+ /**
+ * Adapted from fabric8 kubernetes-client 7.1.0. We bring this here so we
can customize thread pool sizes
+ * and force usage of daemon threads.
+ */
+ private static Vertx createVertxInstance(final
DruidKubernetesHttpClientConfig httpClientConfig)
+ {
+ // fabric8 disables the async DNS resolver while creating Vertx.
+ // I'm not sure if we really need to do this, but I'm keeping it to align
behavior with upstream.
+ final String originalDnsResolverProperty =
System.getProperty(ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME);
+ Vertx vertx;
+ try {
+ System.setProperty(ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME,
"true");
+ vertx = Vertx.vertx(
+ new VertxOptions()
+ .setFileSystemOptions(
+ new FileSystemOptions().setFileCachingEnabled(false)
+ .setClassPathResolvingEnabled(false)
+ )
+ .setWorkerPoolSize(httpClientConfig.getWorkerPoolSize())
+ .setEventLoopPoolSize(httpClientConfig.getEventLoopPoolSize())
+
.setInternalBlockingPoolSize(httpClientConfig.getInternalBlockingPoolSize())
+ .setUseDaemonThread(true)
+ );
+ }
+ finally {
+ if (originalDnsResolverProperty == null) {
+ System.clearProperty(ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME);
+ } else {
+ System.setProperty(ResolverProvider.DISABLE_DNS_RESOLVER_PROP_NAME,
originalDnsResolverProperty);
+ }
+ }
+
+ return vertx;
+ }
+}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
index d12aae08a55..49720642216 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/KubernetesTaskRunnerFactoryTest.java
@@ -21,10 +21,12 @@ package org.apache.druid.k8s.overlord;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.fabric8.kubernetes.api.model.batch.v1.Job;
+import io.fabric8.kubernetes.client.ConfigBuilder;
import org.apache.druid.indexing.common.TestUtils;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.java.util.emitter.service.ServiceEmitter;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
+import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.taskadapter.TaskAdapter;
import org.apache.druid.tasklogs.NoopTaskLogs;
@@ -54,7 +56,8 @@ public class KubernetesTaskRunnerFactoryTest
.withCapacity(1)
.build();
taskLogs = new NoopTaskLogs();
- druidKubernetesClient = new DruidKubernetesClient();
+ druidKubernetesClient =
+ new DruidKubernetesClient(new DruidKubernetesHttpClientConfig(), new
ConfigBuilder().build());
taskAdapter = new TestTaskAdapter();
}
diff --git
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
index cd865e256ee..446c2502fa1 100644
---
a/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
+++
b/extensions-core/kubernetes-overlord-extensions/src/test/java/org/apache/druid/k8s/overlord/taskadapter/DruidPeonClientIntegrationTest.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.databind.jsontype.NamedType;
import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodSpec;
import io.fabric8.kubernetes.api.model.batch.v1.Job;
+import io.fabric8.kubernetes.client.ConfigBuilder;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.druid.indexing.common.TestUtils;
@@ -34,6 +35,7 @@ import org.apache.druid.indexing.common.task.Task;
import
org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexTuningConfig;
import org.apache.druid.k8s.overlord.KubernetesTaskRunnerConfig;
import org.apache.druid.k8s.overlord.common.DruidKubernetesClient;
+import org.apache.druid.k8s.overlord.common.DruidKubernetesHttpClientConfig;
import org.apache.druid.k8s.overlord.common.JobResponse;
import org.apache.druid.k8s.overlord.common.K8sTaskId;
import org.apache.druid.k8s.overlord.common.K8sTestUtils;
@@ -85,7 +87,7 @@ public class DruidPeonClientIntegrationTest
new NamedType(ParallelIndexTuningConfig.class, "index_parallel"),
new NamedType(IndexTask.IndexTuningConfig.class, "index")
);
- k8sClient = new DruidKubernetesClient();
+ k8sClient = new DruidKubernetesClient(new
DruidKubernetesHttpClientConfig(), new ConfigBuilder().build());
peonClient = new KubernetesPeonClient(k8sClient, "default", false, new
NoopServiceEmitter());
druidNode = new DruidNode(
"test",
diff --git a/licenses.yaml b/licenses.yaml
index eb47a93aada..33c09795182 100644
--- a/licenses.yaml
+++ b/licenses.yaml
@@ -855,7 +855,7 @@ name: kubernetes fabric java client
license_category: binary
module: extensions-core/kubernetes-overlord-extensions
license_name: Apache License version 2.0
-version: 6.7.2
+version: 6.13.1
libraries:
- io.fabric8: kubernetes-client-api
- io.fabric8: kubernetes-model-batch
@@ -880,6 +880,19 @@ libraries:
- io.fabric8: kubernetes-model-resource
- io.fabric8: kubernetes-model-scheduling
- io.fabric8: kubernetes-model-storageclass
+ - io.fabric8: kubernetes-httpclient-vertx
+---
+
+name: vertx
+license_category: binary
+module: extensions-core/kubernetes-overlord-extensions
+license_name: Apache License version 2.0
+version: 4.5.8
+libraries:
+ - io.vertx: vertx-auth-common
+ - io.vertx: vertx-core
+ - io.vertx: vertx-web-client
+ - io.vertx: vertx-web-common
---
name: kubernetes official java client
@@ -1051,7 +1064,7 @@ name: org.snakeyaml snakeyaml-engine
license_category: binary
module: extensions-core/druid-kubernetes-overlord-extensions
license_name: Apache License version 2.0
-version: 2.6
+version: 2.7
libraries:
- org.snakeyaml: snakeyaml-engine
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
