You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/10/03 19:06:24 UTC
[GitHub] sijie closed pull request #2702: Have the ability to specify custom
labels for k8 jobs
sijie closed pull request #2702: Have the ability to specify custom labels for k8 jobs
URL: https://github.com/apache/pulsar/pull/2702
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/KubernetesRuntime.java b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
index 369b5a62f2..a2cd71c44b 100644
--- a/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
+++ b/pulsar-functions/runtime/src/main/java/org/apache/pulsar/functions/runtime/KubernetesRuntime.java
@@ -83,6 +83,7 @@
private InstanceControlGrpc.InstanceControlFutureStub[] stub;
private InstanceConfig instanceConfig;
private final String jobNamespace;
+ private final Map<String, String> customLabels;
private final String pulsarDockerImageName;
private final String pulsarRootDir;
private final String userCodePkgUrl;
@@ -94,6 +95,7 @@
KubernetesRuntime(AppsV1Api appsClient,
CoreV1Api coreClient,
String jobNamespace,
+ Map<String, String> customLabels,
String pulsarDockerImageName,
String pulsarRootDir,
InstanceConfig instanceConfig,
@@ -109,6 +111,7 @@
this.coreClient = coreClient;
this.instanceConfig = instanceConfig;
this.jobNamespace = jobNamespace;
+ this.customLabels = customLabels;
this.pulsarDockerImageName = pulsarDockerImageName;
this.pulsarRootDir = pulsarRootDir;
this.userCodePkgUrl = userCodePkgUrl;
@@ -450,6 +453,9 @@ private V1StatefulSet createStatefulSet() {
labels.put("app", createJobName(functionDetails));
labels.put("namespace", functionDetails.getNamespace());
labels.put("tenant", functionDetails.getTenant());
+ if (customLabels != null && !customLabels.isEmpty()) {
+ labels.putAll(customLabels);
+ }
return labels;
}
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 5b902dba27..8ba750516d 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
@@ -30,6 +30,8 @@
import org.apache.pulsar.functions.instance.InstanceConfig;
import org.apache.pulsar.functions.proto.Function;
+import java.util.Map;
+
import static org.apache.commons.lang3.StringUtils.isEmpty;
/**
@@ -43,6 +45,7 @@
private final String pulsarDockerImageName;
private final String pulsarRootDir;
private final Boolean submittingInsidePod;
+ private final Map<String, String> customLabels;
private final String pulsarAdminUri;
private final String pulsarServiceUri;
private final String stateStorageServiceUri;
@@ -59,6 +62,7 @@ public KubernetesRuntimeFactory(String k8Uri,
String pulsarDockerImageName,
String pulsarRootDir,
Boolean submittingInsidePod,
+ Map<String, String> customLabels,
String pulsarServiceUri,
String pulsarAdminUri,
String stateStorageServiceUri,
@@ -80,6 +84,7 @@ public KubernetesRuntimeFactory(String k8Uri,
this.pulsarRootDir = "/pulsar";
}
this.submittingInsidePod = submittingInsidePod;
+ this.customLabels = customLabels;
this.pulsarServiceUri = pulsarServiceUri;
this.pulsarAdminUri = pulsarAdminUri;
this.stateStorageServiceUri = stateStorageServiceUri;
@@ -113,6 +118,7 @@ public KubernetesRuntime createContainer(InstanceConfig instanceConfig, String c
appsClient,
coreClient,
jobNamespace,
+ customLabels,
pulsarDockerImageName,
pulsarRootDir,
instanceConfig,
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 274d973659..6ed11b34fe 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
@@ -68,7 +68,7 @@ public KubernetesRuntimeTest() throws Exception {
this.stateStorageServiceUrl = "bk://localhost:4181";
this.logDirectory = "logs/functions";
this.factory = new KubernetesRuntimeFactory(null, null, null, null,
- false, pulsarServiceUrl, pulsarAdminUrl, stateStorageServiceUrl, null);
+ false, null, pulsarServiceUrl, pulsarAdminUrl, stateStorageServiceUrl, null);
}
@AfterMethod
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 992e5db23c..e5b4a6b8d1 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
@@ -125,6 +125,7 @@ public FunctionRuntimeManager(WorkerConfig workerConfig, WorkerService workerSer
workerConfig.getKubernetesContainerFactory().getPulsarDockerImageName(),
workerConfig.getKubernetesContainerFactory().getPulsarRootDir(),
workerConfig.getKubernetesContainerFactory().getSubmittingInsidePod(),
+ workerConfig.getKubernetesContainerFactory().getCustomLabels(),
StringUtils.isEmpty(workerConfig.getKubernetesContainerFactory().getPulsarServiceUrl()) ? workerConfig.getPulsarServiceUrl() : workerConfig.getKubernetesContainerFactory().getPulsarServiceUrl(),
StringUtils.isEmpty(workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl()) ? workerConfig.getPulsarWebServiceUrl() : workerConfig.getKubernetesContainerFactory().getPulsarAdminUrl(),
workerConfig.getStateStorageServiceUrl(),
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 81e9fd22ca..a3c393f510 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
@@ -27,6 +27,7 @@
import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
+import java.util.Map;
import java.util.Properties;
import java.util.Set;
@@ -138,6 +139,7 @@
private Boolean submittingInsidePod;
private String pulsarServiceUrl;
private String pulsarAdminUrl;
+ private Map<String, String> customLabels;
}
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:
users@infra.apache.org
With regards,
Apache Git Services