Allow configuring ownerReference on K8s resources created by worker service (#174)

fantapsody · web-flow · commit 7d4383ac170b · 2021-05-27T14:14:30.000+08:00
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/FunctionsImpl.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/FunctionsImpl.java
@@ -124,7 +124,8 @@ public void registerFunction(final String tenant,
                 version,
                 functionName,
                 functionPkgUrl,
-                functionConfig
+                functionConfig,
+                worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs()
         );
         // override namespace by configuration file
         v1alpha1Function.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
@@ -188,7 +189,8 @@ public void updateFunction(final String tenant,
                     version,
                     functionName,
                     functionPkgUrl,
-                    functionConfig
+                    functionConfig,
+                    worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs()
             );
             v1alpha1Function.getMetadata().setResourceVersion(oldFn.getMetadata().getResourceVersion());
             this.upsertFunction(tenant, namespace, functionName, functionConfig, v1alpha1Function, clientAuthenticationDataHttps);
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/SinksImpl.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/SinksImpl.java
@@ -132,7 +132,8 @@ public void registerSink(
                         sinkPkgUrl,
                         uploadedInputStream,
                         sinkConfig,
-                        this.meshWorkerServiceSupplier.get().getConnectorsManager());
+                        this.meshWorkerServiceSupplier.get().getConnectorsManager(),
+                        worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs());
         // override namesapce by configuration
         v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
         try {
@@ -207,7 +208,8 @@ public void updateSink(
                             sinkName,
                             sinkPkgUrl,
                             uploadedInputStream,
-                            sinkConfig, this.meshWorkerServiceSupplier.get().getConnectorsManager());
+                            sinkConfig, this.meshWorkerServiceSupplier.get().getConnectorsManager(),
+                            worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs());
             this.upsertSink(tenant, namespace, sinkName, sinkConfig, v1alpha1Sink, clientAuthenticationDataHttps);
             v1alpha1Sink.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
             v1alpha1Sink
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/SourcesImpl.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/rest/api/SourcesImpl.java
@@ -141,7 +141,8 @@ public void registerSource(final String tenant,
                             sourcePkgUrl,
                             uploadedInputStream,
                             sourceConfig,
-                            this.meshWorkerServiceSupplier.get().getConnectorsManager());
+                            this.meshWorkerServiceSupplier.get().getConnectorsManager(),
+                            worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs());
             // override namesapce by configuration
             v1alpha1Source.getMetadata().setNamespace(KubernetesUtils.getNamespace(worker().getFactoryConfig()));
             Map<String, String> customLabels = Maps.newHashMap();
@@ -207,7 +208,8 @@ public void updateSource(final String tenant,
                     sourcePkgUrl,
                     uploadedInputStream,
                     sourceConfig,
-                    this.meshWorkerServiceSupplier.get().getConnectorsManager()
+                    this.meshWorkerServiceSupplier.get().getConnectorsManager(),
+                    worker().getWorkerConfig().getFunctionsWorkerServiceCustomConfigs()
             );
             v1alpha1Source.getMetadata().setResourceVersion(oldRes.getMetadata().getResourceVersion());
             this.upsertSource(tenant, namespace, sourceName, sourceConfig, v1alpha1Source, clientAuthenticationDataHttps);
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/util/CommonUtil.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/util/CommonUtil.java
@@ -19,13 +19,13 @@
 package io.functionmesh.compute.util;
 
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
+import io.kubernetes.client.openapi.models.V1OwnerReference;
+import java.util.Collections;
 import org.apache.commons.codec.digest.DigestUtils;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import java.util.Map;
 import java.util.stream.Collectors;
 
-import static org.apache.commons.lang.StringUtils.left;
-
 public class CommonUtil {
     private static final String CLUSTER_NAME_ENV = "clusterName";
 
@@ -49,10 +49,29 @@ private static String toValidResourceName(String ori) {
         return ori.toLowerCase().replaceAll("[^a-z0-9-\\.]", "-");
     }
 
-    public static V1ObjectMeta makeV1ObjectMeta(String name, String k8sNamespace, String pulsarNamespace, String tenant, String cluster) {
+    public static V1OwnerReference getOwnerReferenceFromCustomConfigs(Map<String, Object> customConfigs) {
+        if (customConfigs == null) {
+            return null;
+        }
+        Map<String, Object> ownerRef = (Map<String, Object>) customConfigs.get("ownerReference");
+        if (ownerRef == null) {
+            return null;
+        }
+        return new V1OwnerReference()
+                .apiVersion(String.valueOf(ownerRef.get("apiVersion")))
+                .kind(String.valueOf(ownerRef.get("kind")))
+                .name(String.valueOf(ownerRef.get("name")))
+                .uid(String.valueOf(ownerRef.get("uid")));
+    }
+
+    public static V1ObjectMeta makeV1ObjectMeta(String name, String k8sNamespace, String pulsarNamespace, String tenant,
+                                                String cluster, V1OwnerReference ownerReference) {
         V1ObjectMeta v1ObjectMeta = new V1ObjectMeta();
         v1ObjectMeta.setName(createObjectName(cluster, tenant, pulsarNamespace, name));
         v1ObjectMeta.setNamespace(k8sNamespace);
+        if (ownerReference != null) {
+            v1ObjectMeta.setOwnerReferences(Collections.singletonList(ownerReference));
+        }
 
         return v1ObjectMeta;
     }
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/util/FunctionsUtil.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/util/FunctionsUtil.java
@@ -58,7 +58,8 @@ public class FunctionsUtil {
     public final static String sourceKey = "source";
 
     public static V1alpha1Function createV1alpha1FunctionFromFunctionConfig(String kind, String group, String version
-            , String functionName, String functionPkgUrl, FunctionConfig functionConfig) {
+            , String functionName, String functionPkgUrl, FunctionConfig functionConfig
+            , Map<String, Object> customConfigs) {
         String customRuntimeOptionsJSON = functionConfig.getCustomRuntimeOptions();
         CustomRuntimeOptions customRuntimeOptions = null;
         if (Strings.isEmpty(customRuntimeOptionsJSON)) {
@@ -98,7 +99,8 @@ public static V1alpha1Function createV1alpha1FunctionFromFunctionConfig(String k
                 functionConfig.getNamespace(),
                 functionDetails.getNamespace(),
                 functionDetails.getTenant(),
-                clusterName));
+                clusterName,
+                CommonUtil.getOwnerReferenceFromCustomConfigs(customConfigs)));
 
         V1alpha1FunctionSpec v1alpha1FunctionSpec = new V1alpha1FunctionSpec();
         v1alpha1FunctionSpec.setClassName(functionConfig.getClassName());
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/util/SinksUtil.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/util/SinksUtil.java
@@ -56,7 +56,8 @@ public class SinksUtil {
 
     public static V1alpha1Sink createV1alpha1SkinFromSinkConfig(String kind, String group, String version
             , String sinkName, String sinkPkgUrl, InputStream uploadedInputStream, SinkConfig sinkConfig,
-                                                                MeshConnectorsManager connectorsManager) {
+                                                                MeshConnectorsManager connectorsManager,
+                                                                Map<String, Object> customConfigs) {
         String customRuntimeOptionsJSON = sinkConfig.getCustomRuntimeOptions();
         CustomRuntimeOptions customRuntimeOptions = null;
         if (Strings.isEmpty(customRuntimeOptionsJSON)) {
@@ -107,7 +108,8 @@ public static V1alpha1Sink createV1alpha1SkinFromSinkConfig(String kind, String
                 sinkConfig.getNamespace(),
                 functionDetails.getNamespace(),
                 functionDetails.getTenant(),
-                clusterName));
+                clusterName,
+                CommonUtil.getOwnerReferenceFromCustomConfigs(customConfigs)));
 
         V1alpha1SinkSpec v1alpha1SinkSpec = new V1alpha1SinkSpec();
         v1alpha1SinkSpec.setClassName(sinkConfig.getClassName());
diff --git a/mesh-worker-service/src/main/java/io/functionmesh/compute/util/SourcesUtil.java b/mesh-worker-service/src/main/java/io/functionmesh/compute/util/SourcesUtil.java
@@ -61,7 +61,8 @@ public static V1alpha1Source createV1alpha1SourceFromSourceConfig(String kind, S
                                                                       String sourceName, String sourcePkgUrl,
                                                                       InputStream uploadedInputStream,
                                                                       SourceConfig sourceConfig,
-                                                                      MeshConnectorsManager connectorsManager) {
+                                                                      MeshConnectorsManager connectorsManager,
+                                                                      Map<String, Object> customConfigs) {
         String customRuntimeOptionsJSON = sourceConfig.getCustomRuntimeOptions();
         CustomRuntimeOptions customRuntimeOptions = null;
         if (Strings.isEmpty(customRuntimeOptionsJSON)) {
@@ -110,7 +111,8 @@ public static V1alpha1Source createV1alpha1SourceFromSourceConfig(String kind, S
                 sourceConfig.getNamespace(),
                 functionDetails.getNamespace(),
                 functionDetails.getTenant(),
-                clusterName));
+                clusterName,
+                CommonUtil.getOwnerReferenceFromCustomConfigs(customConfigs)));
 
         V1alpha1SourceSpec v1alpha1SourceSpec = new V1alpha1SourceSpec();
         v1alpha1SourceSpec.setClassName(sourceConfig.getClassName());
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/FunctionsImplTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/FunctionsImplTest.java
@@ -30,6 +30,7 @@
 import io.kubernetes.client.openapi.JSON;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
 import io.kubernetes.client.openapi.apis.CustomObjectsApi;
+import java.util.Collections;
 import okhttp3.Call;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
@@ -278,7 +279,7 @@ public void registerFunctionTest() throws ApiException, IOException, PulsarAdmin
         PowerMockito.when(tenants.getTenantInfo(tenant)).thenReturn(null);
 
         V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(kind, group,
-                version, functionName, null, functionConfig);
+                version, functionName, null, functionConfig, Collections.emptyMap());
 
         Map<String, String> customLabels = Maps.newHashMap();
         customLabels.put("pulsar-tenant", tenant);
@@ -460,7 +461,7 @@ public void updateFunctionTest() throws ApiException, IOException {
 
         FunctionConfig functionConfig = Generate.CreateJavaFunctionConfig(tenant, namespace, functionName);
         V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(kind, group,
-                version, functionName, null, functionConfig);
+                version, functionName, null, functionConfig, Collections.emptyMap());
         v1alpha1Function.getMetadata().setResourceVersion("24794021");
 
         PowerMockito.when(meshWorkerService.getCustomObjectsApi()
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/SinksImpTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/SinksImpTest.java
@@ -224,7 +224,8 @@ public void testRegisterSink()
 
         V1alpha1Sink v1alpha1Sink =
                 SinksUtil.createV1alpha1SkinFromSinkConfig(
-                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null);
+                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null,
+                        Collections.emptyMap());
 
         Map<String, String> customLabels = Maps.newHashMap();
         customLabels.put("pulsar-cluster", clusterName);
@@ -442,7 +443,8 @@ public void testUpdateSink()
 
         V1alpha1Sink v1alpha1Sink =
                 SinksUtil.createV1alpha1SkinFromSinkConfig(
-                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null);
+                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null,
+                        Collections.emptyMap());
         v1alpha1Sink.getMetadata().setResourceVersion("881033");
 
         PowerMockito.when(
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/SourcesImpTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/rest/api/SourcesImpTest.java
@@ -28,6 +28,7 @@
 import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.openapi.JSON;
 import io.kubernetes.client.openapi.apis.CustomObjectsApi;
+import java.util.Collections;
 import okhttp3.Call;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
@@ -227,7 +228,8 @@ public void testRegisterSource()
                         null,
                         uploadedInputStream,
                         sourceConfig,
-                        null);
+                        null,
+                        Collections.emptyMap());
         Map<String, String> customLabels = Maps.newHashMap();
         customLabels.put("pulsar-cluster", clusterName);
         customLabels.put("pulsar-tenant", tenant);
@@ -459,7 +461,8 @@ public void testUpdateSource()
                         componentName,
                         null,
                         uploadedInputStream,
-                        sourceConfig, null);
+                        sourceConfig, null,
+                        Collections.emptyMap());
         v1alpha1Source.getMetadata().setResourceVersion("881033");
 
         PowerMockito.when(
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/util/FunctionsUtilTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/util/FunctionsUtilTest.java
@@ -21,6 +21,7 @@
 import io.functionmesh.compute.functions.models.V1alpha1Function;
 import io.functionmesh.compute.functions.models.V1alpha1FunctionSpec;
 import io.functionmesh.compute.testdata.Generate;
+import java.util.Collections;
 import org.apache.pulsar.common.functions.FunctionConfig;
 import org.junit.Assert;
 import org.junit.Test;
@@ -45,7 +46,7 @@ public void testCreateV1alpha1FunctionFromFunctionConfig() {
         FunctionConfig functionConfig = Generate.CreateJavaFunctionConfig(tenant, namespace, functionName);
 
         V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(kind, group, version,
-                functionName, null, functionConfig);
+                functionName, null, functionConfig, Collections.emptyMap());
 
         Assert.assertEquals(v1alpha1Function.getKind(), kind);
 
@@ -75,7 +76,7 @@ public void testCreateFunctionConfigFromV1alpha1Function() {
         FunctionConfig functionConfig = Generate.CreateJavaFunctionConfig(tenant, namespace, functionName);
 
         V1alpha1Function v1alpha1Function = FunctionsUtil.createV1alpha1FunctionFromFunctionConfig(kind, group, version,
-                functionName, null, functionConfig);
+                functionName, null, functionConfig, Collections.emptyMap());
 
         FunctionConfig newFunctionConfig = FunctionsUtil.createFunctionConfigFromV1alpha1Function(tenant, namespace,
                 functionName, v1alpha1Function);
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/util/SinksUtilTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/util/SinksUtilTest.java
@@ -23,6 +23,7 @@
 import io.functionmesh.compute.sinks.models.V1alpha1SinkSpec;
 import io.functionmesh.compute.testdata.Generate;
 import io.functionmesh.compute.worker.MeshConnectorsManager;
+import java.util.Collections;
 import org.apache.commons.io.FileUtils;
 import org.apache.pulsar.common.io.SinkConfig;
 import org.apache.pulsar.common.nar.NarClassLoader;
@@ -86,7 +87,8 @@ public void testCreateV1alpha1SinkFromSinkConfig()
 
         V1alpha1Sink actualV1alpha1Sink =
                 SinksUtil.createV1alpha1SkinFromSinkConfig(
-                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null);
+                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null,
+                        Collections.emptyMap());
 
         Assert.assertEquals(actualV1alpha1Sink.getKind(), kind);
         V1alpha1SinkSpec v1alpha1SinkSpec = actualV1alpha1Sink.getSpec();
@@ -128,7 +130,8 @@ public void testCreateSinkConfigFromV1alpha1Sink() throws ClassNotFoundException
 
         V1alpha1Sink actualV1alpha1Sink =
                 SinksUtil.createV1alpha1SkinFromSinkConfig(
-                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null);
+                        kind, group, version, componentName, null, uploadedInputStream, sinkConfig, null,
+                        Collections.emptyMap());
 
         SinkConfig newSinkConfig = SinksUtil.createSinkConfigFromV1alpha1Sink(tenant, namespace, componentName, actualV1alpha1Sink);
 
@@ -177,7 +180,8 @@ public void testCreateV1alpha1SinkFromSinkConfigWithBuiltin()
 
         V1alpha1Sink actualV1alpha1Sink =
                 SinksUtil.createV1alpha1SkinFromSinkConfig(
-                        kind, group, version, componentName, null, null, sinkConfig, connectorsManager);
+                        kind, group, version, componentName, null, null, sinkConfig, connectorsManager,
+                        Collections.emptyMap());
 
         Assert.assertEquals(actualV1alpha1Sink.getKind(), kind);
         V1alpha1SinkSpec v1alpha1SinkSpec = actualV1alpha1Sink.getSpec();
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/util/SourcesUtilTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/util/SourcesUtilTest.java
@@ -23,6 +23,7 @@
 import io.functionmesh.compute.sources.models.V1alpha1SourceSpec;
 import io.functionmesh.compute.testdata.Generate;
 import io.functionmesh.compute.worker.MeshConnectorsManager;
+import java.util.Collections;
 import org.apache.commons.io.FileUtils;
 import org.apache.pulsar.common.io.SourceConfig;
 import org.apache.pulsar.common.nar.NarClassLoader;
@@ -83,7 +84,8 @@ public void testCreateV1alpha1SourceFromSourceConfig() throws ClassNotFoundExcep
         SourceConfig sourceConfig = Generate.CreateSourceConfig(tenant, namespace, componentName);
 
         V1alpha1Source v1alpha1Source = SourcesUtil.createV1alpha1SourceFromSourceConfig(kind, group, version,
-                componentName, null, uploadedInputStream, sourceConfig, null);
+                componentName, null, uploadedInputStream, sourceConfig, null,
+                Collections.emptyMap());
 
         Assert.assertEquals(v1alpha1Source.getKind(), kind);
         V1alpha1SourceSpec v1alpha1SourceSpec = v1alpha1Source.getSpec();
@@ -121,7 +123,8 @@ public void testCreateSourceConfigFromV1alpha1Source() throws ClassNotFoundExcep
         SourceConfig sourceConfig = Generate.CreateSourceConfig(tenant, namespace, componentName);
 
         V1alpha1Source v1alpha1Source = SourcesUtil.createV1alpha1SourceFromSourceConfig(kind, group, version,
-                componentName, null, uploadedInputStream, sourceConfig, null);
+                componentName, null, uploadedInputStream, sourceConfig, null,
+                Collections.emptyMap());
 
         SourceConfig newSourceConfig = SourcesUtil.createSourceConfigFromV1alpha1Source(tenant, namespace,
                 componentName, v1alpha1Source);
@@ -168,7 +171,8 @@ public void testCreateV1alpha1SourceFromSinkConfigWithBuiltin() throws IOExcepti
 
         V1alpha1Source v1alpha1Source =
                 SourcesUtil.createV1alpha1SourceFromSourceConfig(
-                        kind, group, version, componentName, null, null, sourceConfig, connectorsManager);
+                        kind, group, version, componentName, null, null, sourceConfig, connectorsManager,
+                        Collections.emptyMap());
 
         Assert.assertEquals(v1alpha1Source.getKind(), kind);
         V1alpha1SourceSpec v1alpha1SourceSpec = v1alpha1Source.getSpec();
diff --git a/mesh-worker-service/src/test/java/io/functionmesh/compute/util/WorkerConfigTest.java b/mesh-worker-service/src/test/java/io/functionmesh/compute/util/WorkerConfigTest.java
diff --git a/mesh-worker-service/src/test/resources/test_worker_config.yaml b/mesh-worker-service/src/test/resources/test_worker_config.yaml
