sijie closed pull request #2856: Added ability for the kubernetes to poll a
configmap to look out for changes
URL: https://github.com/apache/pulsar/pull/2856
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
index 1a180aee0c..c6d5d02645 100644
---
a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
+++
b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeFactory.java
@@ -24,13 +24,20 @@
import io.kubernetes.client.Configuration;
import io.kubernetes.client.apis.AppsV1Api;
import io.kubernetes.client.apis.CoreV1Api;
+import io.kubernetes.client.models.V1ConfigMap;
import io.kubernetes.client.util.Config;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.functions.instance.AuthenticationConfig;
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
+import java.lang.reflect.Field;
import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
import static org.apache.commons.lang3.StringUtils.isEmpty;
@@ -40,24 +47,33 @@
@Slf4j
public class KubernetesRuntimeFactory implements RuntimeFactory {
- private final String k8Uri;
- private final String jobNamespace;
- private final String pulsarDockerImageName;
- private final String pulsarRootDir;
+ @Getter
+ @Setter
+ @NoArgsConstructor
+ class KubernetesInfo {
+ private String k8Uri;
+ private String jobNamespace;
+ private String pulsarDockerImageName;
+ private String pulsarRootDir;
+ private String pulsarAdminUrl;
+ private String pulsarServiceUrl;
+ private String pythonDependencyRepository;
+ private String pythonExtraDependencyRepository;
+ private String changeConfigMap;
+ private String changeConfigMapNamespace;
+ }
+ private final KubernetesInfo kubernetesInfo;
private final Boolean submittingInsidePod;
private final Boolean installUserCodeDependencies;
- private final String pythonDependencyRepository;
- private final String pythonExtraDependencyRepository;
private final Map<String, String> customLabels;
- private final String pulsarAdminUri;
- private final String pulsarServiceUri;
+ private final Integer expectedMetricsCollectionInterval;
private final String stateStorageServiceUri;
private final AuthenticationConfig authConfig;
private final String javaInstanceJarFile;
private final String pythonInstanceFile;
private final String prometheusMetricsServerJarFile;
private final String logDirectory = "logs/functions";
- private final Integer expectedMetricsInterval;
+ private Timer changeConfigMapTimer;
private AppsV1Api appsClient;
private CoreV1Api coreClient;
@@ -75,36 +91,41 @@ public KubernetesRuntimeFactory(String k8Uri,
String pulsarAdminUri,
String stateStorageServiceUri,
AuthenticationConfig authConfig,
- Integer expectedMetricsInterval) {
- this.k8Uri = k8Uri;
+ Integer expectedMetricsCollectionInterval,
+ String changeConfigMap,
+ String changeConfigMapNamespace) {
+ this.kubernetesInfo = new KubernetesInfo();
+ this.kubernetesInfo.setK8Uri(k8Uri);
if (!isEmpty(jobNamespace)) {
- this.jobNamespace = jobNamespace;
+ this.kubernetesInfo.setJobNamespace(jobNamespace);
} else {
- this.jobNamespace = "default";
+ this.kubernetesInfo.setJobNamespace("default");
}
if (!isEmpty(pulsarDockerImageName)) {
- this.pulsarDockerImageName = pulsarDockerImageName;
+
this.kubernetesInfo.setPulsarDockerImageName(pulsarDockerImageName);
} else {
- this.pulsarDockerImageName = "apachepulsar/pulsar";
+
this.kubernetesInfo.setPulsarDockerImageName("apachepulsar/pulsar");
}
if (!isEmpty(pulsarRootDir)) {
- this.pulsarRootDir = pulsarRootDir;
+ this.kubernetesInfo.setPulsarRootDir(pulsarRootDir);
} else {
- this.pulsarRootDir = "/pulsar";
+ this.kubernetesInfo.setPulsarRootDir("/pulsar");
}
+
this.kubernetesInfo.setPythonDependencyRepository(pythonDependencyRepository);
+
this.kubernetesInfo.setPythonExtraDependencyRepository(pythonExtraDependencyRepository);
+ this.kubernetesInfo.setPulsarServiceUrl(pulsarServiceUri);
+ this.kubernetesInfo.setPulsarAdminUrl(pulsarAdminUri);
+ this.kubernetesInfo.setChangeConfigMap(changeConfigMap);
+
this.kubernetesInfo.setChangeConfigMapNamespace(changeConfigMapNamespace);
this.submittingInsidePod = submittingInsidePod;
this.installUserCodeDependencies = installUserCodeDependencies;
- this.pythonDependencyRepository = pythonDependencyRepository;
- this.pythonExtraDependencyRepository = pythonExtraDependencyRepository;
this.customLabels = customLabels;
- this.pulsarServiceUri = pulsarServiceUri;
- this.pulsarAdminUri = pulsarAdminUri;
this.stateStorageServiceUri = stateStorageServiceUri;
this.authConfig = authConfig;
- this.javaInstanceJarFile = this.pulsarRootDir +
"/instances/java-instance.jar";
- this.pythonInstanceFile = this.pulsarRootDir +
"/instances/python-instance/python_instance_main.py";
- this.prometheusMetricsServerJarFile = this.pulsarRootDir +
"/instances/PrometheusMetricsServer.jar";
- this.expectedMetricsInterval = expectedMetricsInterval == null ? -1 :
expectedMetricsInterval;
+ this.javaInstanceJarFile = this.kubernetesInfo.getPulsarRootDir() +
"/instances/java-instance.jar";
+ this.pythonInstanceFile = this.kubernetesInfo.getPulsarRootDir() +
"/instances/python-instance/python_instance_main.py";
+ this.prometheusMetricsServerJarFile =
this.kubernetesInfo.getPulsarRootDir() +
"/instances/PrometheusMetricsServer.jar";
+ this.expectedMetricsCollectionInterval =
expectedMetricsCollectionInterval == null ? -1 :
expectedMetricsCollectionInterval;
}
@Override
@@ -131,24 +152,24 @@ public KubernetesRuntime createContainer(InstanceConfig
instanceConfig, String c
return new KubernetesRuntime(
appsClient,
coreClient,
- jobNamespace,
+ this.kubernetesInfo.getJobNamespace(),
customLabels,
installUserCodeDependencies,
- pythonDependencyRepository,
- pythonExtraDependencyRepository,
- pulsarDockerImageName,
- pulsarRootDir,
+ this.kubernetesInfo.getPythonDependencyRepository(),
+ this.kubernetesInfo.getPythonExtraDependencyRepository(),
+ this.kubernetesInfo.getPulsarDockerImageName(),
+ this.kubernetesInfo.getPulsarRootDir(),
instanceConfig,
instanceFile,
prometheusMetricsServerJarFile,
logDirectory,
codePkgUrl,
originalCodeFileName,
- pulsarServiceUri,
- pulsarAdminUri,
+ this.kubernetesInfo.getPulsarServiceUrl(),
+ this.kubernetesInfo.getPulsarAdminUrl(),
stateStorageServiceUri,
authConfig,
- expectedMetricsInterval);
+ expectedMetricsCollectionInterval);
}
@Override
@@ -163,7 +184,7 @@ public void doAdmissionChecks(Function.FunctionDetails
functionDetails) {
@VisibleForTesting
void setupClient() throws Exception {
if (appsClient == null) {
- if (k8Uri == null) {
+ if (this.kubernetesInfo.getK8Uri() == null) {
log.info("k8Uri is null thus going by defaults");
ApiClient cli;
if (submittingInsidePod) {
@@ -177,11 +198,44 @@ void setupClient() throws Exception {
appsClient = new AppsV1Api();
coreClient = new CoreV1Api();
} else {
- log.info("Setting up k8Client using uri " + k8Uri);
- final ApiClient apiClient = new ApiClient().setBasePath(k8Uri);
+ log.info("Setting up k8Client using uri " +
this.kubernetesInfo.getK8Uri());
+ final ApiClient apiClient = new
ApiClient().setBasePath(this.kubernetesInfo.getK8Uri());
appsClient = new AppsV1Api(apiClient);
coreClient = new CoreV1Api(apiClient);
}
+
+ // Setup a timer to change stuff.
+ if (!isEmpty(this.kubernetesInfo.getChangeConfigMap())) {
+ changeConfigMapTimer = new Timer();
+ changeConfigMapTimer.scheduleAtFixedRate(new TimerTask() {
+ @Override
+ public void run() {
+ fetchConfigMap();
+ }
+ }, 300000, 300000);
+ }
+ }
+ }
+
+ void fetchConfigMap() {
+ try {
+ V1ConfigMap v1ConfigMap =
coreClient.readNamespacedConfigMap(kubernetesInfo.getChangeConfigMap(),
kubernetesInfo.getChangeConfigMapNamespace(), null, true, false);
+ Map<String, String> data = v1ConfigMap.getData();
+ if (data != null) {
+ overRideKubernetesConfig(data);
+ }
+ } catch (Exception e) {
+ log.error("Error while trying to fetch configmap {} at namespace
{}", kubernetesInfo.getChangeConfigMap(),
kubernetesInfo.getChangeConfigMapNamespace(), e);
+ }
+ }
+
+ void overRideKubernetesConfig(Map<String, String> data) throws Exception {
+ for (Field field : KubernetesInfo.class.getDeclaredFields()) {
+ field.setAccessible(true);
+ if (data.containsKey(field.getName()) &&
!data.get(field.getName()).equals(field.get(kubernetesInfo))) {
+ log.info("Kubernetes Config {} changed from {} to {}",
field.getName(), field.get(kubernetesInfo), data.get(field.getName()));
+ field.set(kubernetesInfo, data.get(field.getName()));
+ }
}
}
}
diff --git
a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
index e82c75febf..02886604c5 100644
---
a/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
+++
b/pulsar-functions/runtime/src/test/java/org/apache/pulsar/functions/runtime/KubernetesRuntimeTest.java
@@ -72,7 +72,9 @@ public KubernetesRuntimeTest() throws Exception {
this.stateStorageServiceUrl = "bk://localhost:4181";
this.logDirectory = "logs/functions";
this.factory = spy(new KubernetesRuntimeFactory(null, null, null,
pulsarRootDir,
- false, true, "myrepo", "anotherrepo", null, pulsarServiceUrl,
pulsarAdminUrl, stateStorageServiceUrl, null, null));
+ false, true, "myrepo", "anotherrepo",
+ null, pulsarServiceUrl, pulsarAdminUrl,
stateStorageServiceUrl, null,
+ null, null, null));
doNothing().when(this.factory).setupClient();
}
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
index f7dd4bc07e..d8bf252905 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/FunctionRuntimeManager.java
@@ -140,7 +140,9 @@ public FunctionRuntimeManager(WorkerConfig workerConfig,
WorkerService workerSer
StringUtils.isEmpty(workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl())
? workerConfig.getPulsarWebServiceUrl() :
workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl(),
workerConfig.getStateStorageServiceUrl(),
authConfig,
-
workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval()
== null ? -1 :
workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval());
+
workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval()
== null ? -1 :
workerConfig.getKubernetesContainerFactory().getExpectedMetricsCollectionInterval(),
+
workerConfig.getKubernetesContainerFactory().getChangeConfigMap(),
+
workerConfig.getKubernetesContainerFactory().getChangeConfigMapNamespace());
} else {
throw new RuntimeException("Either Thread, Process or Kubernetes
Container Factory need to be set");
}
diff --git
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
index 2b3e816a79..587f3117bf 100644
---
a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
+++
b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/WorkerConfig.java
@@ -144,6 +144,11 @@
private String pythonExtraDependencyRepository;
private Map<String, String> customLabels;
private Integer expectedMetricsCollectionInterval;
+ // Kubernetes Runtime will periodically checkback on
+ // this configMap if defined and if there are any changes
+ // to the kubernetes specific stuff, we apply those changes
+ private String changeConfigMap;
+ private String changeConfigMapNamespace;
}
private KubernetesContainerFactory kubernetesContainerFactory;
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
