support sofa registry kubernetes

sofastack · Jan 24, 2024 · a477e3e · a477e3e
1 parent f37465b
commit a477e3e
Show file tree

Hide file tree

Showing 11 changed files with 580 additions and 0 deletions.
diff --git a/bom/pom.xml b/bom/pom.xml
@@ -31,6 +31,8 @@
         <asm.version>7.0</asm.version>
         <guava.version>32.0.0-jre</guava.version>
         <prometheus.client.version>0.16.0</prometheus.client.version>
+        <!-- Fabric8 for Kubernetes -->
+        <fabric8_kubernetes_version>6.9.2</fabric8_kubernetes_version>
         <!-- serialization -->
         <hessian.version>3.5.2</hessian.version>
         <thrift.version>0.9.2</thrift.version>
@@ -598,6 +600,19 @@
                 <version>0.16.0</version>
                 <scope>test</scope>
             </dependency>
+
+            <!-- Fabric8 for Kubernetes -->
+            <dependency>
+                <groupId>io.fabric8</groupId>
+                <artifactId>kubernetes-client</artifactId>
+                <version>${fabric8_kubernetes_version}</version>
+            </dependency>
+            <dependency>
+                <groupId>io.fabric8</groupId>
+                <artifactId>kubernetes-server-mock</artifactId>
+                <scope>test</scope>
+                <version>${fabric8_kubernetes_version}</version>
+            </dependency>
         </dependencies>
     </dependencyManagement>
 

diff --git a/registry/pom.xml b/registry/pom.xml
@@ -22,6 +22,7 @@
         <module>registry-multicast</module>
         <module>registry-sofa</module>
         <module>registry-polaris</module>
+        <module>registry-kubernetes</module>
     </modules>
 
     <dependencyManagement>

diff --git a/registry/registry-kubernetes/pom.xml b/registry/registry-kubernetes/pom.xml
@@ -0,0 +1,111 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>com.alipay.sofa</groupId>
+        <artifactId>sofa-rpc-registry</artifactId>
+        <version>${revision}</version>
+    </parent>
+
+    <artifactId>sofa-rpc-registry-kubernetes</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alipay.sofa</groupId>
+            <artifactId>sofa-rpc-log</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alipay.sofa</groupId>
+            <artifactId>sofa-rpc-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alipay.sofa</groupId>
+            <artifactId>sofa-rpc-codec-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.fabric8</groupId>
+            <artifactId>kubernetes-client</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.fabric8</groupId>
+            <artifactId>kubernetes-server-mock</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <sourceDirectory>src/main/java</sourceDirectory>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>false</filtering>
+                <includes>
+                    <include>**/**</include>
+                </includes>
+            </resource>
+        </resources>
+        <testSourceDirectory>src/test/java</testSourceDirectory>
+        <testResources>
+            <testResource>
+                <directory>src/test/resources</directory>
+                <filtering>false</filtering>
+                <includes>
+                    <include>**/**</include>
+                </includes>
+            </testResource>
+        </testResources>
+
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>${maven.compiler.source}</source>
+                    <target>${maven.compiler.target}</target>
+                    <encoding>${project.build.sourceEncoding}</encoding>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-install-plugin</artifactId>
+                <configuration>
+                    <skip>${module.install.skip}</skip>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-deploy-plugin</artifactId>
+                <configuration>
+                    <skip>${module.deploy.skip}</skip>
+                </configuration>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <skipTests>${skipTests}</skipTests>
+                    <includes>
+                        <!-- 这里需要根据自己的需要指定要跑的单元测试 -->
+                        <include>**/*Test.java</include>
+                    </includes>
+                    <!-- 如无特殊需求，将forkMode设置为once -->
+                    <forkMode>once</forkMode>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>
diff --git a/...-kubernetes/src/main/java/com/alipay/sofa/rpc/registry/kubernetes/KubernetesConstant.java b/...-kubernetes/src/main/java/com/alipay/sofa/rpc/registry/kubernetes/KubernetesConstant.java
@@ -0,0 +1,46 @@
+/*
+ * 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 com.alipay.sofa.rpc.registry.kubernetes;
+
+import io.fabric8.kubernetes.client.dsl.base.CustomResourceDefinitionContext;
+
+public class KubernetesConstant {
+
+    public static CustomResourceDefinitionContext getVsDefinition() {
+
+        return new CustomResourceDefinitionContext.Builder()
+                .withGroup("service.com.alipay.sofa.rpc")
+                .withVersion("v1alpha1")
+                .withScope("Namespaced")
+                .withName("virtualservices.service.com.alipay.sofa.rpc")
+                .withPlural("virtualservices")
+                .withKind("VirtualService")
+                .build();
+    }
+
+    public static CustomResourceDefinitionContext getDrDefinition() {
+
+        return new CustomResourceDefinitionContext.Builder()
+                .withGroup("service.com.alipay.sofa.rpc")
+                .withVersion("v1alpha1")
+                .withScope("Namespaced")
+                .withName("destinationrules.service.com.alipay.sofa.rpc")
+                .withPlural("destinationrules")
+                .withKind("DestinationRule")
+                .build();
+    }
+}
diff --git a/...-kubernetes/src/main/java/com/alipay/sofa/rpc/registry/kubernetes/KubernetesRegistry.java b/...-kubernetes/src/main/java/com/alipay/sofa/rpc/registry/kubernetes/KubernetesRegistry.java
@@ -0,0 +1,201 @@
+/*
+ * 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 com.alipay.sofa.rpc.registry.kubernetes;
+
+import com.alipay.sofa.rpc.client.ProviderGroup;
+import com.alipay.sofa.rpc.common.json.JSON;
+import com.alipay.sofa.rpc.config.ConsumerConfig;
+import com.alipay.sofa.rpc.config.ProviderConfig;
+import com.alipay.sofa.rpc.config.RegistryConfig;
+import com.alipay.sofa.rpc.ext.Extension;
+import com.alipay.sofa.rpc.log.LogCodes;
+import com.alipay.sofa.rpc.log.Logger;
+import com.alipay.sofa.rpc.log.LoggerFactory;
+import com.alipay.sofa.rpc.registry.Registry;
+import com.alipay.sofa.rpc.registry.kubernetes.util.KubernetesClientUtils;
+import com.alipay.sofa.rpc.registry.kubernetes.util.KubernetesConfigUtils;
+import io.fabric8.kubernetes.api.model.Endpoints;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.Service;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Extension("kubernetes")
+public class KubernetesRegistry extends Registry {
+
+    public static final String EXT_NAME = "KubernetesRegistry";
+
+    /**
+     * slf4j Logger for this class
+     */
+    private final static Logger LOGGER = LoggerFactory.getLogger(KubernetesRegistry.class);
+
+    private static final String kubernetesPropertiesKey = "io.sofa.rpc/metadata";
+
+    private KubernetesClient kubernetesClient;
+
+    private String currentHostname;
+
+    private String namespace;
+
+    private static final ConcurrentHashMap<String, SharedIndexInformer<Service>> serviceInformer = new ConcurrentHashMap<>(64);
+
+    private static final ConcurrentHashMap<String, SharedIndexInformer<Pod>> podsInformer = new ConcurrentHashMap<>(64);
+
+    private static final ConcurrentHashMap<String, SharedIndexInformer<Endpoints>> endpointsInformer = new ConcurrentHashMap<>(64);
+
+    /**
+     * Instantiates a new kubernetes registry.
+     *
+     * @param registryConfig
+     */
+    protected KubernetesRegistry(RegistryConfig registryConfig) {
+        super(registryConfig);
+    }
+
+    @Override
+    public boolean start() {
+        return true;
+    }
+
+    @Override
+    public void register(ProviderConfig config) {
+        String appName = config.getAppName();
+        if (!config.isRegister()) {
+            if (LOGGER.isInfoEnabled(appName)) {
+                LOGGER.infoWithApp(appName, LogCodes.getLog(LogCodes.INFO_REGISTRY_IGNORE));
+            }
+            return;
+        }
+
+        kubernetesClient
+                .pods()
+                .inNamespace(namespace)
+                .withName(currentHostname)
+                .edit(pod -> new PodBuilder(pod)
+                        .editOrNewMetadata()
+                        .addToAnnotations(
+                                kubernetesPropertiesKey, JSON.toJSONString(config.getApplication().getInsId()))
+                        .endMetadata()
+                        .build());
+    }
+
+    @Override
+    public void unRegister(ProviderConfig config) {
+        String appName = config.getAppName();
+        if (!config.isRegister()) {
+            if (LOGGER.isInfoEnabled(appName)) {
+                LOGGER.infoWithApp(appName, LogCodes.getLog(LogCodes.INFO_REGISTRY_IGNORE));
+            }
+            return;
+        }
+
+        kubernetesClient
+                .pods()
+                .inNamespace(namespace)
+                .withName(currentHostname)
+                .edit(pod -> new PodBuilder(pod)
+                        .editOrNewMetadata()
+                        .removeFromAnnotations(kubernetesPropertiesKey)
+                        .endMetadata()
+                        .build());
+
+    }
+
+    @Override
+    public void batchUnRegister(List<ProviderConfig> configs) {
+        // one by one
+        for (ProviderConfig config : configs) {
+            try {
+                this.unRegister(config);
+            } catch (Exception e) {
+                LOGGER.errorWithApp(config.getAppName(), "Batch unregister error", e);
+            }
+        }
+    }
+
+    @Override
+    public List<ProviderGroup> subscribe(ConsumerConfig config) {
+        config.getProviderInfoListener()
+        return null;
+    }
+
+    @Override
+    public void unSubscribe(ConsumerConfig config) {
+
+    }
+
+    @Override
+    public void batchUnSubscribe(List<ConsumerConfig> configs) {
+        // one by one
+        for (ConsumerConfig config : configs) {
+            try {
+                this.unSubscribe(config);
+            } catch (Exception e) {
+                LOGGER.errorWithApp(config.getAppName(), "Batch unSubscribe error", e);
+            }
+        }
+    }
+
+    @Override
+    public void destroy() {
+        // stop service
+        serviceInformer.forEach((k, v) -> v.close());
+        serviceInformer.clear();
+
+        // stop pod
+        podsInformer.forEach((k, v) -> v.close());
+        podsInformer.clear();
+
+        // stop endpoints
+        endpointsInformer.forEach((k, v) -> v.close());
+        endpointsInformer.clear();
+
+        // close kubernetes client
+        kubernetesClient.close();
+    }
+
+    @Override
+    public void init() {
+        // init kubernetes config
+        Config config = KubernetesConfigUtils.buildKubernetesConfig(registryConfig);
+        // init kubernetes client
+        this.kubernetesClient = KubernetesClientUtils.buildKubernetesClient(config);
+        this.currentHostname = System.getenv("HOSTNAME");
+        this.namespace = config.getNamespace();
+    }
+
+    private Map<String, String> getServiceSelector(String serviceName) {
+        Service service = kubernetesClient
+                .services()
+                .inNamespace(namespace)
+                .withName(serviceName)
+                .get();
+        if (service == null) {
+            return null;
+        }
+        return service.getSpec().getSelector();
+    }
+
+}