[ISSUE #1028] support Etcd Registry

build.gradle

@@ -473,10 +473,10 @@ subprojects {
             dependency "io.cloudevents:cloudevents-core:2.2.0"
             dependency "io.cloudevents:cloudevents-json-jackson:2.2.0"
 
-            dependency "io.grpc:grpc-protobuf:1.15.0"
-            dependency "io.grpc:grpc-stub:1.15.0"
-            dependency "io.grpc:grpc-netty:1.15.0"
-            dependency "io.grpc:grpc-netty-shaded:1.15.0"
+            dependency "io.grpc:grpc-protobuf:1.17.1"
+            dependency "io.grpc:grpc-stub:1.17.1"
+            dependency "io.grpc:grpc-netty:1.17.1"
+            dependency "io.grpc:grpc-netty-shaded:1.17.1"
 
             dependency "javax.annotation:javax.annotation-api:1.3.2"
 

eventmesh-common/build.gradle

@@ -39,8 +39,8 @@ dependencies {
 
     implementation "io.netty:netty-all"
 
-    implementation "io.grpc:grpc-protobuf:1.15.0"
-    implementation "io.grpc:grpc-stub:1.15.0"
+    implementation "io.grpc:grpc-protobuf:1.17.1"
+    implementation "io.grpc:grpc-stub:1.17.1"
     implementation "javax.annotation:javax.annotation-api:1.3.2"
 
     implementation "com.github.stefanbirkner:system-rules"

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/protos/ConsumerServiceGrpc.java

@@ -31,7 +31,7 @@
  */
 @SuppressWarnings({"all"})
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.15.0)",
+    value = "by gRPC proto compiler (version 1.17.1)",
     comments = "Source: eventmesh-client.proto")
 public final class ConsumerServiceGrpc {
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/protos/HeartbeatServiceGrpc.java

@@ -28,7 +28,7 @@
  */
 @SuppressWarnings({"all"})
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.15.0)",
+    value = "by gRPC proto compiler (version 1.17.1)",
     comments = "Source: eventmesh-client.proto")
 public final class HeartbeatServiceGrpc {
 

eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/protos/PublisherServiceGrpc.java

@@ -28,7 +28,7 @@
  */
 @SuppressWarnings({"all"})
 @javax.annotation.Generated(
-    value = "by gRPC proto compiler (version 1.15.0)",
+    value = "by gRPC proto compiler (version 1.17.1)",
     comments = "Source: eventmesh-client.proto")
 public final class PublisherServiceGrpc {
 

eventmesh-examples/build.gradle

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-def grpcVersion = '1.15.0'
+def grpcVersion = '1.17.1'
 
 configurations {
     implementation.exclude group: 'org.springframework.boot', module: 'spring-boot-starter-logging'

eventmesh-protocol-plugin/eventmesh-protocol-cloudevents/build.gradle

@@ -20,5 +20,5 @@ dependencies {
     implementation "io.cloudevents:cloudevents-core"
     implementation "com.google.guava:guava"
     implementation "io.cloudevents:cloudevents-json-jackson"
-    implementation "io.grpc:grpc-protobuf:1.15.0"
+    implementation "io.grpc:grpc-protobuf:1.17.1"
 }

eventmesh-protocol-plugin/eventmesh-protocol-grpc/build.gradle

@@ -24,7 +24,7 @@ repositories {
     mavenCentral()
 }
 
-def grpcVersion = '1.15.0' // CURRENT_GRPC_VERSION
+def grpcVersion = '1.17.1' // CURRENT_GRPC_VERSION
 def protobufVersion = '3.5.1'
 def protocVersion = protobufVersion
 

eventmesh-protocol-plugin/eventmesh-protocol-http/build.gradle

@@ -20,5 +20,5 @@ dependencies {
     implementation "io.cloudevents:cloudevents-core"
     implementation "com.google.guava:guava"
     implementation "io.cloudevents:cloudevents-json-jackson"
-    implementation "io.grpc:grpc-protobuf:1.15.0"
+    implementation "io.grpc:grpc-protobuf:1.17.1"
 }

eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/build.gradle

@@ -18,7 +18,7 @@
 dependencies {
     implementation project(":eventmesh-protocol-plugin:eventmesh-protocol-api")
     implementation "io.cloudevents:cloudevents-core"
-    implementation "io.grpc:grpc-protobuf:1.15.0"
+    implementation "io.grpc:grpc-protobuf:1.17.1"
 
     testImplementation project(":eventmesh-protocol-plugin:eventmesh-protocol-api")
     testImplementation "io.cloudevents:cloudevents-core"

eventmesh-registry-plugin/eventmesh-registry-api/src/main/java/org/apache/eventmesh/api/registry/dto/EventMeshDataInfo.java

@@ -30,6 +30,10 @@ public class EventMeshDataInfo {
 
     private Map<String, String> metadata;
 
+    public EventMeshDataInfo() {
+
+    }
+
     public EventMeshDataInfo(String eventMeshClusterName, String eventMeshName, String endpoint, long lastUpdateTimestamp,
                              Map<String, String> metadata) {
         this.eventMeshClusterName = eventMeshClusterName;

eventmesh-registry-plugin/eventmesh-registry-etcd/build.gradle

@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+dependencies {
+    implementation ("io.etcd:jetcd-core:0.3.0")
+    implementation project(":eventmesh-registry-plugin:eventmesh-registry-api")
+    implementation project(":eventmesh-common")
+    testImplementation "org.mockito:mockito-core"
+}

eventmesh-registry-plugin/eventmesh-registry-etcd/gradle.properties

@@ -0,0 +1,19 @@
+#
+# 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.
+#
+
+pluginType=registry
+pluginName=etcd

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/constant/EtcdConstant.java

@@ -0,0 +1,36 @@
+/*
+ * 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.eventmesh.registry.etcd.constant;
+
+/**
+ * EtcdConstant.
+ */
+public class EtcdConstant {
+
+    public static final String SERVER_ADDR = "serverAddr";
+
+    public static final String USERNAME = "username";
+
+    public static final String PASSWORD = "password";
+
+    public static final String KEY_SEPARATOR = "/";
+
+    public static final long TTL = 15L;
+
+
+}

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientFactory.java

@@ -0,0 +1,116 @@
+/*
+ * 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.eventmesh.registry.etcd.factory;
+
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.ClientBuilder;
+import io.etcd.jetcd.options.LeaseOption;
+
+
+public class EtcdClientFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(EtcdClientFactory.class);
+
+    private static final Map<String, EtcdLeaseId> etcdLeaseIdMap = new ConcurrentHashMap<>();
+
+
+    public static Client createClient(Properties properties) {
+        String serverAddr = properties.getProperty(EtcdConstant.SERVER_ADDR);
+        String username = properties.getProperty(EtcdConstant.USERNAME);
+        String password = properties.getProperty(EtcdConstant.PASSWORD);
+
+        EtcdLeaseId etcdLeaseId = etcdLeaseIdMap.get(serverAddr);
+        if (Objects.nonNull(etcdLeaseId)) {
+            return etcdLeaseId.getClientWrapper();
+        }
+        ClientBuilder clientBuilder = Client.builder();
+        String[] addresses = serverAddr.split(",");
+        String[] httpAddress = new String[addresses.length];
+        for (int i = 0; i < addresses.length; i++) {
+            if (!addresses[i].startsWith(Constants.HTTP_PROTOCOL_PREFIX)) {
+                httpAddress[i] = Constants.HTTP_PROTOCOL_PREFIX + addresses[i];
+            }
+        }
+        etcdLeaseId = new EtcdLeaseId();
+        try {
+            etcdLeaseId.setUrl(serverAddr);
+            etcdLeaseId.setClientBuilder(clientBuilder.endpoints(httpAddress));
+            if (StringUtils.isNoneBlank(username)) {
+                etcdLeaseId.getClientBuilder().user(ByteSequence.from(username.getBytes()));
+            }
+            if (StringUtils.isNoneBlank(password)) {
+                etcdLeaseId.getClientBuilder().password(ByteSequence.from(password.getBytes()));
+            }
+            etcdLeaseId.setClientWrapper(new EtcdClientWrapper(etcdLeaseId.getClientBuilder().build()));
+            EtcdClientWrapper client = etcdLeaseId.getClientWrapper();
+            long leaseId = client.getLeaseClient().grant(EtcdConstant.TTL).get().getID();
+            etcdLeaseId.setLeaseId(leaseId);
+            EtcdStreamObserver etcdStreamObserver = new EtcdStreamObserver();
+            etcdStreamObserver.setEtcdLeaseId(etcdLeaseId);
+            etcdLeaseId.setEtcdStreamObserver(etcdStreamObserver);
+            client.getLeaseClient().keepAlive(leaseId, etcdStreamObserver);
+
+            etcdLeaseIdMap.put(serverAddr, etcdLeaseId);
+        } catch (Throwable e) {
+            logger.error("createClient failed, address: {}", serverAddr, e);
+            throw new RegistryException("createClient failed", e);
+        }
+        return etcdLeaseId.getClientWrapper();
+    }
+
+
+    public static void renewalLeaseId(EtcdLeaseId etcdLeaseId) {
+        logger.info("renewal of contract. server url: {}", etcdLeaseId.getUrl());
+        Client client = etcdLeaseId.getClientWrapper();
+        try {
+            long ttl = client.getLeaseClient().timeToLive(etcdLeaseId.getLeaseId(), LeaseOption.DEFAULT).get().getTTl();
+            if (ttl < 1) {
+                long leaseId = client.getLeaseClient().grant(EtcdConstant.TTL).get().getID();
+                client.getLeaseClient().keepAlive(leaseId, etcdLeaseId.getEtcdStreamObserver());
+                etcdLeaseId.setLeaseId(leaseId);
+            }
+        } catch (Throwable e) {
+            logger.error("renewal error, server url: {}", etcdLeaseId.getUrl(), e);
+            client.getLeaseClient().keepAlive(System.currentTimeMillis(), etcdLeaseId.getEtcdStreamObserver());
+        }
+    }
+
+    public static Long getLeaseId(String url) {
+        return getEtcdLeaseId(url).getLeaseId();
+    }
+
+    public static EtcdLeaseId getEtcdLeaseId(String url) {
+        return etcdLeaseIdMap.get(url);
+    }
+
+}

eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientWrapper.java

@@ -0,0 +1,79 @@
+/*
+ * 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.eventmesh.registry.etcd.factory;
+
+
+import io.etcd.jetcd.Auth;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.Cluster;
+import io.etcd.jetcd.KV;
+import io.etcd.jetcd.Lease;
+import io.etcd.jetcd.Lock;
+import io.etcd.jetcd.Maintenance;
+import io.etcd.jetcd.Watch;
+
+
+class EtcdClientWrapper implements Client {
+
+    private volatile Client client;
+
+    public EtcdClientWrapper(Client client) {
+        this.client = client;
+    }
+
+    @Override
+    public Auth getAuthClient() {
+        return client.getAuthClient();
+    }
+
+    @Override
+    public KV getKVClient() {
+        return client.getKVClient();
+    }
+
+    @Override
+    public Cluster getClusterClient() {
+        return client.getClusterClient();
+    }
+
+    @Override
+    public Maintenance getMaintenanceClient() {
+        return client.getMaintenanceClient();
+    }
+
+    @Override
+    public Lease getLeaseClient() {
+        return client.getLeaseClient();
+    }
+
+    @Override
+    public Watch getWatchClient() {
+        return client.getWatchClient();
+    }
+
+    @Override
+    public Lock getLockClient() {
+        return client.getLockClient();
+    }
+
+    @Override
+    public void close() {
+        client.close();
+    }
+
+}