You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by GitBox <gi...@apache.org> on 2022/07/18 16:24:04 UTC
[GitHub] [incubator-eventmesh] JellyBo opened a new pull request, #1029: [ISSUE #1028] support Etcd Registry
JellyBo opened a new pull request, #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029
Fixes ISSUE #<1028>.
### Modifications
*Describe the modifications you've done.*
Support Etcd registry implementation:
Interconnecting with Etcd v3.6.0
- update grpc version from 1.15.0 to 1.17.1
- add etcd registry support
- how to use, add etcd registry properties, for example:
`eventMesh.registry.plugin.enabled=true`
`eventMesh.registry.plugin.type=etcd`
`eventMesh.registry.plugin.server-addr=127.0.0.1:2379`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
[GitHub] [incubator-eventmesh] xwm1992 commented on pull request #1029: [ISSUE #1028] support Etcd Registry
Posted by GitBox <gi...@apache.org>.
xwm1992 commented on PR #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029#issuecomment-1202543352
@JellyBo Hi, please fixed the UT's error，thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
[GitHub] [incubator-eventmesh] codecov[bot] commented on pull request #1029: [ISSUE #1028] support Etcd Registry
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029#issuecomment-1203452834
# [Codecov](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#1029](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (2e94c9a) into [registry-etcd](https://codecov.io/gh/apache/incubator-eventmesh/commit/24e5c19a4e3b442a8e6fe14e2e302d90182bce68?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (24e5c19) will **increase** coverage by `1.48%`.
> The diff coverage is `27.90%`.
```diff
@@ Coverage Diff @@
## registry-etcd #1029 +/- ##
==================================================
+ Coverage 8.30% 9.78% +1.48%
- Complexity 528 623 +95
==================================================
Files 362 368 +6
Lines 23118 23328 +210
Branches 2546 2567 +21
==================================================
+ Hits 1920 2283 +363
+ Misses 21048 20828 -220
- Partials 150 217 +67
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...mmon/protocol/grpc/protos/ConsumerServiceGrpc.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL2NvbW1vbi9wcm90b2NvbC9ncnBjL3Byb3Rvcy9Db25zdW1lclNlcnZpY2VHcnBjLmphdmE=) | `0.00% <ø> (ø)` | |
| [...mon/protocol/grpc/protos/HeartbeatServiceGrpc.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL2NvbW1vbi9wcm90b2NvbC9ncnBjL3Byb3Rvcy9IZWFydGJlYXRTZXJ2aWNlR3JwYy5qYXZh) | `0.00% <ø> (ø)` | |
| [...mon/protocol/grpc/protos/PublisherServiceGrpc.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL2NvbW1vbi9wcm90b2NvbC9ncnBjL3Byb3Rvcy9QdWJsaXNoZXJTZXJ2aWNlR3JwYy5qYXZh) | `0.00% <ø> (ø)` | |
| [...eventmesh/registry/etcd/constant/EtcdConstant.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2QvY29uc3RhbnQvRXRjZENvbnN0YW50LmphdmE=) | `0.00% <0.00%> (ø)` | |
| [...mesh/registry/etcd/factory/EtcdStreamObserver.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2QvZmFjdG9yeS9FdGNkU3RyZWFtT2JzZXJ2ZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...esh/registry/etcd/service/EtcdRegistryService.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2Qvc2VydmljZS9FdGNkUmVnaXN0cnlTZXJ2aWNlLmphdmE=) | `21.95% <21.95%> (ø)` | |
| [...tmesh/registry/etcd/factory/EtcdClientWrapper.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2QvZmFjdG9yeS9FdGNkQ2xpZW50V3JhcHBlci5qYXZh) | `33.33% <33.33%> (ø)` | |
| [...tmesh/registry/etcd/factory/EtcdClientFactory.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2QvZmFjdG9yeS9FdGNkQ2xpZW50RmFjdG9yeS5qYXZh) | `38.00% <38.00%> (ø)` | |
| [...e/eventmesh/registry/etcd/factory/EtcdLeaseId.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLXJlZ2lzdHJ5LXBsdWdpbi9ldmVudG1lc2gtcmVnaXN0cnktZXRjZC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL3JlZ2lzdHJ5L2V0Y2QvZmFjdG9yeS9FdGNkTGVhc2VJZC5qYXZh) | `50.00% <50.00%> (ø)` | |
| [...ava/org/apache/eventmesh/common/utils/IPUtils.java](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-ZXZlbnRtZXNoLWNvbW1vbi9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvZXZlbnRtZXNoL2NvbW1vbi91dGlscy9JUFV0aWxzLmphdmE=) | `33.33% <0.00%> (-6.49%)` | :arrow_down: |
| ... and [30 more](https://codecov.io/gh/apache/incubator-eventmesh/pull/1029/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
[GitHub] [incubator-eventmesh] githublaohu commented on a diff in pull request #1029: [ISSUE #1028] support Etcd Registry
Posted by GitBox <gi...@apache.org>.
githublaohu commented on code in PR #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029#discussion_r923645172
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.service;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.PutOption;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.api.registry.RegistryService;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EtcdRegistryService implements RegistryService {
+
+ private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class);
+
+ private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
+
+ private static final AtomicBoolean START_STATUS = new AtomicBoolean(false);
+
+ private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR;
+
+ private String serverAddr;
+
+ private String username;
+
+ private String password;
+
+ private EtcdClientFactory etcdClientFactory;
+
+ private Client etcdClient;
+
+ private Long leaseId;
+
+ private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
+
+ @Override
+ public void init() throws RegistryException {
+ boolean update = INIT_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ etcdClientFactory = new EtcdClientFactory();
+ eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size());
+ for (String key : ConfigurationContextUtil.KEYS) {
+ CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key);
+ if (null == commonConfiguration) {
+ continue;
+ }
+ if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) {
+ throw new RegistryException("namesrvAddr cannot be null");
+ }
+ this.serverAddr = commonConfiguration.namesrvAddr;
+ this.username = commonConfiguration.eventMeshRegistryPluginUsername;
+ this.password = commonConfiguration.eventMeshRegistryPluginPassword;
+ break;
+ }
+ }
+
+ @Override
+ public void start() throws RegistryException {
+ boolean update = START_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ try {
+ Properties properties = new Properties();
+ properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr);
+ properties.setProperty(EtcdConstant.USERNAME, username);
+ properties.setProperty(EtcdConstant.PASSWORD, password);
+ this.etcdClient = etcdClientFactory.createClient(properties);
+ this.leaseId = etcdClientFactory.getLeaseId(serverAddr);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][start] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void shutdown() throws RegistryException {
+ INIT_STATUS.compareAndSet(true, false);
+ START_STATUS.compareAndSet(true, false);
+ logger.info("EtcdRegistryService closed");
+ }
+
+ @Override
+ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
+ List<EventMeshDataInfo> eventMeshDataInfoList = new ArrayList<>();
+
+ try {
+ String keyPrefix = clusterName == null? KEY_PREFIX : KEY_PREFIX + EtcdConstant.KEY_SEPARATOR + clusterName;
+ GetOption getOption = GetOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix.getBytes())).build();
+ List<KeyValue> keyValues = getEtcdClient().getKVClient().get(ByteSequence.from(keyPrefix.getBytes()), getOption).get().getKvs();
+
+ if (CollectionUtils.isNotEmpty(keyValues)) {
+ for (KeyValue kv : keyValues) {
+ EventMeshDataInfo eventMeshDataInfo = JsonUtils.deserialize(new String(kv.getValue().getBytes()), EventMeshDataInfo.class);
+ eventMeshDataInfoList.add(eventMeshDataInfo);
+ }
+ }
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ return eventMeshDataInfoList;
+ }
+
+ @Override
+ public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException {
+ try {
+ return findEventMeshInfoByCluster(null);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose)
+ throws RegistryException {
+ // todo find metadata
+ return null;
+ }
+
+ @Override
+ public void registerMetadata(Map<String, String> metadataMap) {
+ for (Map.Entry<String, EventMeshRegisterInfo> eventMeshRegisterInfo : eventMeshRegisterInfoMap.entrySet()) {
+ EventMeshRegisterInfo registerInfo = eventMeshRegisterInfo.getValue();
+ registerInfo.setMetadata(metadataMap);
+ this.register(registerInfo);
+ }
+ }
+
+ @Override
+ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException {
+ try {
+ String eventMeshClusterName = eventMeshRegisterInfo.getEventMeshClusterName();
+ String eventMeshName = eventMeshRegisterInfo.getEventMeshName();
+ ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshRegisterInfo.getEndPoint()).getBytes());
+ EventMeshDataInfo eventMeshDataInfo =
+ new EventMeshDataInfo(eventMeshClusterName, eventMeshName,
+ eventMeshRegisterInfo.getEndPoint(), new Date().getTime(), eventMeshRegisterInfo.getMetadata());
+ ByteSequence etcdValue = ByteSequence.from(JsonUtils.serialize(eventMeshDataInfo).getBytes());
+ etcdClient.getKVClient().put(etcdKey, etcdValue, PutOption.newBuilder().withLeaseId(leaseId).build());
+ eventMeshRegisterInfoMap.put(eventMeshName, eventMeshRegisterInfo);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][register] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ logger.info("EventMesh successfully registered to etcd");
+ return true;
+ }
+
+ @Override
+ public boolean unRegister(EventMeshUnRegisterInfo eventMeshUnRegisterInfo) throws RegistryException {
+ try {
+ String eventMeshClusterName = eventMeshUnRegisterInfo.getEventMeshClusterName();
+ String eventMeshName = eventMeshUnRegisterInfo.getEventMeshName();
+ ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshUnRegisterInfo.getEndPoint()).getBytes());
+ etcdClient.getKVClient().delete(etcdKey);
+ eventMeshRegisterInfoMap.remove(eventMeshName);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][unRegister] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ logger.info("EventMesh successfully logout to etcd");
+ return true;
+ }
+
+ public String getServerAddr() {
Review Comment:
这个与下面三个方法有存在的意义吗？
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.service;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.PutOption;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.api.registry.RegistryService;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EtcdRegistryService implements RegistryService {
+
+ private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class);
+
+ private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
+
+ private static final AtomicBoolean START_STATUS = new AtomicBoolean(false);
+
+ private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR;
+
+ private String serverAddr;
+
+ private String username;
+
+ private String password;
+
+ private EtcdClientFactory etcdClientFactory;
+
+ private Client etcdClient;
+
+ private Long leaseId;
+
+ private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
+
+ @Override
+ public void init() throws RegistryException {
+ boolean update = INIT_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ etcdClientFactory = new EtcdClientFactory();
+ eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size());
+ for (String key : ConfigurationContextUtil.KEYS) {
+ CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key);
+ if (null == commonConfiguration) {
+ continue;
+ }
+ if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) {
+ throw new RegistryException("namesrvAddr cannot be null");
+ }
+ this.serverAddr = commonConfiguration.namesrvAddr;
+ this.username = commonConfiguration.eventMeshRegistryPluginUsername;
+ this.password = commonConfiguration.eventMeshRegistryPluginPassword;
+ break;
+ }
+ }
+
+ @Override
+ public void start() throws RegistryException {
+ boolean update = START_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ try {
+ Properties properties = new Properties();
+ properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr);
+ properties.setProperty(EtcdConstant.USERNAME, username);
+ properties.setProperty(EtcdConstant.PASSWORD, password);
+ this.etcdClient = etcdClientFactory.createClient(properties);
+ this.leaseId = etcdClientFactory.getLeaseId(serverAddr);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][start] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void shutdown() throws RegistryException {
Review Comment:
应该调用etcdclient.close();
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientFactory.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.*;
+import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
+import io.etcd.jetcd.options.LeaseOption;
+import io.grpc.stub.StreamObserver;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class EtcdClientFactory {
Review Comment:
是否应该是单列模式，否则etcdregistrySercie每一个都创建一次。如果每次都创建一次etcdLeaseIdMap，是否有存在的必要
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.service;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.PutOption;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.api.registry.RegistryService;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EtcdRegistryService implements RegistryService {
+
+ private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class);
+
+ private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
+
+ private static final AtomicBoolean START_STATUS = new AtomicBoolean(false);
+
+ private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR;
+
+ private String serverAddr;
+
+ private String username;
+
+ private String password;
+
+ private EtcdClientFactory etcdClientFactory;
+
+ private Client etcdClient;
+
+ private Long leaseId;
+
+ private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
+
+ @Override
+ public void init() throws RegistryException {
+ boolean update = INIT_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ etcdClientFactory = new EtcdClientFactory();
+ eventMeshRegisterInfoMap = new HashMap<>(ConfigurationContextUtil.KEYS.size());
+ for (String key : ConfigurationContextUtil.KEYS) {
+ CommonConfiguration commonConfiguration = ConfigurationContextUtil.get(key);
+ if (null == commonConfiguration) {
+ continue;
+ }
+ if (StringUtils.isBlank(commonConfiguration.namesrvAddr)) {
+ throw new RegistryException("namesrvAddr cannot be null");
+ }
+ this.serverAddr = commonConfiguration.namesrvAddr;
+ this.username = commonConfiguration.eventMeshRegistryPluginUsername;
+ this.password = commonConfiguration.eventMeshRegistryPluginPassword;
+ break;
+ }
+ }
+
+ @Override
+ public void start() throws RegistryException {
+ boolean update = START_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ try {
+ Properties properties = new Properties();
+ properties.setProperty(EtcdConstant.SERVER_ADDR, serverAddr);
+ properties.setProperty(EtcdConstant.USERNAME, username);
+ properties.setProperty(EtcdConstant.PASSWORD, password);
+ this.etcdClient = etcdClientFactory.createClient(properties);
+ this.leaseId = etcdClientFactory.getLeaseId(serverAddr);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][start] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ }
+
+ @Override
+ public void shutdown() throws RegistryException {
+ INIT_STATUS.compareAndSet(true, false);
+ START_STATUS.compareAndSet(true, false);
+ logger.info("EtcdRegistryService closed");
+ }
+
+ @Override
+ public List<EventMeshDataInfo> findEventMeshInfoByCluster(String clusterName) throws RegistryException {
+ List<EventMeshDataInfo> eventMeshDataInfoList = new ArrayList<>();
+
+ try {
+ String keyPrefix = clusterName == null? KEY_PREFIX : KEY_PREFIX + EtcdConstant.KEY_SEPARATOR + clusterName;
+ GetOption getOption = GetOption.newBuilder().withPrefix(ByteSequence.from(keyPrefix.getBytes())).build();
+ List<KeyValue> keyValues = getEtcdClient().getKVClient().get(ByteSequence.from(keyPrefix.getBytes()), getOption).get().getKvs();
+
+ if (CollectionUtils.isNotEmpty(keyValues)) {
+ for (KeyValue kv : keyValues) {
+ EventMeshDataInfo eventMeshDataInfo = JsonUtils.deserialize(new String(kv.getValue().getBytes()), EventMeshDataInfo.class);
+ eventMeshDataInfoList.add(eventMeshDataInfo);
+ }
+ }
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ return eventMeshDataInfoList;
+ }
+
+ @Override
+ public List<EventMeshDataInfo> findAllEventMeshInfo() throws RegistryException {
+ try {
+ return findEventMeshInfoByCluster(null);
+ } catch (Exception e) {
+ logger.error("[EtcdRegistryService][findEventMeshInfoByCluster] error", e);
+ throw new RegistryException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Map<String, Map<String, Integer>> findEventMeshClientDistributionData(String clusterName, String group, String purpose)
+ throws RegistryException {
+ // todo find metadata
+ return null;
+ }
+
+ @Override
+ public void registerMetadata(Map<String, String> metadataMap) {
+ for (Map.Entry<String, EventMeshRegisterInfo> eventMeshRegisterInfo : eventMeshRegisterInfoMap.entrySet()) {
+ EventMeshRegisterInfo registerInfo = eventMeshRegisterInfo.getValue();
+ registerInfo.setMetadata(metadataMap);
+ this.register(registerInfo);
+ }
+ }
+
+ @Override
+ public boolean register(EventMeshRegisterInfo eventMeshRegisterInfo) throws RegistryException {
+ try {
+ String eventMeshClusterName = eventMeshRegisterInfo.getEventMeshClusterName();
+ String eventMeshName = eventMeshRegisterInfo.getEventMeshName();
+ ByteSequence etcdKey = ByteSequence.from((KEY_PREFIX + eventMeshClusterName + EtcdConstant.KEY_SEPARATOR + eventMeshName + EtcdConstant.KEY_SEPARATOR + eventMeshRegisterInfo.getEndPoint()).getBytes());
Review Comment:
etcdkey的创建有多处，麻烦提取出来
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/constant/EtcdConstant.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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 {
Review Comment:
common包里面是否有这些变量？
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/factory/EtcdClientFactory.java:
##########
@@ -0,0 +1,243 @@
+/*
+ * 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.*;
+import io.etcd.jetcd.lease.LeaseKeepAliveResponse;
+import io.etcd.jetcd.options.LeaseOption;
+import io.grpc.stub.StreamObserver;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public class EtcdClientFactory {
+
+ private static final Logger logger = LoggerFactory.getLogger(EtcdClientFactory.class);
+
+ private Map<String, EtcdLeaseId> etcdLeaseIdMap = new ConcurrentHashMap<>();
+
+
+ public 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[] addresss = serverAddr.split(",");
+ String[] httpAddress = new String[addresss.length];
+ for (int i = 0; i < addresss.length; i++) {
+ if (!addresss[i].startsWith("http://")) {
+ httpAddress[i] = "http://" + addresss[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 ClientWrapper(etcdLeaseId.getClientBuilder().build()));
+ ClientWrapper client = etcdLeaseId.getClientWrapper();
+ long leaseId = client.getLeaseClient().grant(10L).get().getID();
+ etcdLeaseId.setLeaseId(leaseId);
+ EtcdStreamObserver etcdStreamObserver = new EtcdStreamObserver();
+ etcdStreamObserver.etcdLeaseId = etcdLeaseId;
+ etcdLeaseId.setEtcdStreamObserver(etcdStreamObserver);
+ client.getLeaseClient().keepAlive(leaseId, etcdStreamObserver);
+
+ etcdLeaseIdMap.put(serverAddr, etcdLeaseId);
+ } catch (Throwable e) {
+ logger.error(e.getMessage(), e);
Review Comment:
不抛出异常吗?
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/service/EtcdRegistryService.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * 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.service;
+
+import io.etcd.jetcd.ByteSequence;
+import io.etcd.jetcd.Client;
+import io.etcd.jetcd.KeyValue;
+import io.etcd.jetcd.options.GetOption;
+import io.etcd.jetcd.options.PutOption;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.eventmesh.api.exception.RegistryException;
+import org.apache.eventmesh.api.registry.RegistryService;
+import org.apache.eventmesh.api.registry.dto.EventMeshDataInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshRegisterInfo;
+import org.apache.eventmesh.api.registry.dto.EventMeshUnRegisterInfo;
+import org.apache.eventmesh.common.config.CommonConfiguration;
+import org.apache.eventmesh.common.utils.ConfigurationContextUtil;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.eventmesh.common.utils.JsonUtils;
+import org.apache.eventmesh.registry.etcd.constant.EtcdConstant;
+import org.apache.eventmesh.registry.etcd.factory.EtcdClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class EtcdRegistryService implements RegistryService {
+
+ private static final Logger logger = LoggerFactory.getLogger(EtcdRegistryService.class);
+
+ private static final AtomicBoolean INIT_STATUS = new AtomicBoolean(false);
+
+ private static final AtomicBoolean START_STATUS = new AtomicBoolean(false);
+
+ private static final String KEY_PREFIX = EtcdConstant.KEY_SEPARATOR + "eventMesh" + EtcdConstant.KEY_SEPARATOR + "registry" + EtcdConstant.KEY_SEPARATOR;
+
+ private String serverAddr;
+
+ private String username;
+
+ private String password;
+
+ private EtcdClientFactory etcdClientFactory;
+
+ private Client etcdClient;
+
+ private Long leaseId;
+
+ private Map<String, EventMeshRegisterInfo> eventMeshRegisterInfoMap;
+
+ @Override
+ public void init() throws RegistryException {
+ boolean update = INIT_STATUS.compareAndSet(false, true);
+ if (!update) {
+ return;
+ }
+ etcdClientFactory = new EtcdClientFactory();
Review Comment:
是否可以放到 start() 方法创建
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
[GitHub] [incubator-eventmesh] xwm1992 merged pull request #1029: [ISSUE #1028] support Etcd Registry
Posted by GitBox <gi...@apache.org>.
xwm1992 merged PR #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
[GitHub] [incubator-eventmesh] JellyBo commented on a diff in pull request #1029: [ISSUE #1028] support Etcd Registry
Posted by GitBox <gi...@apache.org>.
JellyBo commented on code in PR #1029:
URL: https://github.com/apache/incubator-eventmesh/pull/1029#discussion_r926255255
##########
eventmesh-registry-plugin/eventmesh-registry-etcd/src/main/java/org/apache/eventmesh/registry/etcd/constant/EtcdConstant.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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 {
Review Comment:
common没有etcd相关变量
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org