feat: add rocketmq check and nacos check

apache · Mar 7, 2024 · 8cd05d2 · 8cd05d2
1 parent 702b96a
commit 8cd05d2
Show file tree

Hide file tree

Showing 11 changed files with 917 additions and 8 deletions.
diff --git a/.../apache/eventmesh/dashboard/console/function/health/check/impl/meta/NacosConfigCheck.java b/.../apache/eventmesh/dashboard/console/function/health/check/impl/meta/NacosConfigCheck.java
@@ -0,0 +1,102 @@
+/*
+ * 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.dashboard.console.function.health.check.impl.meta;
+
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_SUBTYPE_NACOS_CONFIG;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_TYPE_META;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthConstant.NACOS_CHECK_CONTENT;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthConstant.NACOS_CHECK_DATA_ID;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthConstant.NACOS_CHECK_GROUP;
+
+import org.apache.eventmesh.dashboard.console.function.health.annotation.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.function.health.callback.HealthCheckCallback;
+import org.apache.eventmesh.dashboard.console.function.health.check.AbstractHealthCheckService;
+import org.apache.eventmesh.dashboard.console.function.health.check.config.HealthCheckObjectConfig;
+
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+
+import com.alibaba.nacos.api.NacosFactory;
+import com.alibaba.nacos.api.config.ConfigService;
+import com.alibaba.nacos.api.exception.NacosException;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@HealthCheckType(type = HEALTH_CHECK_TYPE_META, subType = HEALTH_CHECK_SUBTYPE_NACOS_CONFIG)
+public class NacosConfigCheck extends AbstractHealthCheckService {
+
+ private ConfigService configService;
+
+ public NacosConfigCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
+ super(healthCheckObjectConfig);
+ }
+
+ @Override
+ public void doCheck(HealthCheckCallback callback) {
+ CompletableFuture.runAsync(() -> {
+ try {
+ String content = configService.getConfig(NACOS_CHECK_DATA_ID, NACOS_CHECK_GROUP, getConfig().getRequestTimeoutMillis());
+ if (NACOS_CHECK_CONTENT.equals(content)) {
+ callback.onSuccess();
+ } else {
+ callback.onFail(new RuntimeException("NacosCheck failed. Content is wrong."));
+ }
+ } catch (NacosException e) {
+ callback.onFail(e);
+ }
+ });
+ }
+
+ @Override
+ public void init() {
+ //create a config
+ try {
+ Properties properties = new Properties();
+ properties.put("serverAddr", getConfig().getConnectUrl());
+ ConfigService configService = NacosFactory.createConfigService(properties);
+ boolean isPublishOk = configService.publishConfig(NACOS_CHECK_DATA_ID, NACOS_CHECK_GROUP,
+ NACOS_CHECK_CONTENT);
+ if (!isPublishOk) {
+ log.error("NacosCheck init failed caused by crate config failed");
+ }
+ } catch (NacosException e) {
+ log.error("NacosCheck init failed caused by {}", e.getErrMsg());
+ }
+
+ try {
+ Properties properties = new Properties();
+ properties.put("serverAddr", getConfig().getConnectUrl());
+ configService = NacosFactory.createConfigService(properties);
+ } catch (NacosException e) {
+ log.error("NacosCheck init failed caused by {}", e.getErrMsg());
+ }
+ }
+
+ @Override
+ public void destroy() {
+ if (configService != null) {
+ try {
+ configService.removeConfig(NACOS_CHECK_DATA_ID, NACOS_CHECK_GROUP);
+ } catch (NacosException e) {
+ log.error("NacosCheck destroy failed caused by {}", e.getErrMsg());
+ }
+
+ }
+ }
+}
diff --git a/...pache/eventmesh/dashboard/console/function/health/check/impl/meta/NacosRegisterCheck.java b/...pache/eventmesh/dashboard/console/function/health/check/impl/meta/NacosRegisterCheck.java
@@ -0,0 +1,87 @@
+/*
+ * 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.dashboard.console.function.health.check.impl.meta;
+
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_SUBTYPE_NACOS_REGISTER;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_TYPE_META;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthConstant.NACOS_CHECK_SERVICE_CLUSTER_NAME;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthConstant.NACOS_CHECK_SERVICE_NAME;
+
+import org.apache.eventmesh.dashboard.console.function.health.annotation.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.function.health.callback.HealthCheckCallback;
+import org.apache.eventmesh.dashboard.console.function.health.check.AbstractHealthCheckService;
+import org.apache.eventmesh.dashboard.console.function.health.check.config.HealthCheckObjectConfig;
+
+import java.util.Properties;
+import java.util.concurrent.CompletableFuture;
+
+import com.alibaba.nacos.api.exception.NacosException;
+import com.alibaba.nacos.api.naming.NamingFactory;
+import com.alibaba.nacos.api.naming.NamingService;
+import com.alibaba.nacos.api.naming.pojo.Instance;
+
+import lombok.extern.slf4j.Slf4j;
+
+
+@Slf4j
+@HealthCheckType(type = HEALTH_CHECK_TYPE_META, subType = HEALTH_CHECK_SUBTYPE_NACOS_REGISTER)
+public class NacosRegisterCheck extends AbstractHealthCheckService {
+
+ private NamingService namingService;
+
+ public NacosRegisterCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
+ super(healthCheckObjectConfig);
+ }
+
+ @Override
+ public void doCheck(HealthCheckCallback callback) {
+ CompletableFuture.runAsync(() -> {
+ try {
+ Instance result = namingService.selectOneHealthyInstance(NACOS_CHECK_SERVICE_NAME);
+ if (result.isHealthy()) {
+ callback.onSuccess();
+ } else {
+ callback.onFail(new RuntimeException("NacosCheck failed. Service is not healthy."));
+ }
+ } catch (NacosException e) {
+ callback.onFail(e);
+ }
+ });
+ }
+
+ @Override
+ public void init() {
+ try {
+ Properties properties = new Properties();
+ properties.put("serverAddr", getConfig().getConnectUrl());
+ namingService = NamingFactory.createNamingService(properties);
+ namingService.registerInstance(NACOS_CHECK_SERVICE_NAME, "11.11.11.11", 8888, NACOS_CHECK_SERVICE_CLUSTER_NAME);
+ } catch (NacosException e) {
+ log.error("NacosRegisterCheck init failed", e);
+ }
+ }
+
+ @Override
+ public void destroy() {
+ try {
+ namingService.deregisterInstance(NACOS_CHECK_SERVICE_NAME, "11.11.11.11", 8888, NACOS_CHECK_SERVICE_CLUSTER_NAME);
+ } catch (NacosException e) {
+ log.error("NacosRegisterCheck destroy failed", e);
+ }
+ }
+}
diff --git a/.../health/check/impl/StorageRedisCheck.java → ...health/check/impl/storage/RedisCheck.java b/.../health/check/impl/StorageRedisCheck.java → ...health/check/impl/storage/RedisCheck.java
@@ -15,12 +15,14 @@
  * limitations under the License.
  */
 
-package org.apache.eventmesh.dashboard.console.health.check.impl;
+package org.apache.eventmesh.dashboard.console.function.health.check.impl.storage;
 
-import org.apache.eventmesh.dashboard.console.health.annotation.HealthCheckType;
-import org.apache.eventmesh.dashboard.console.health.callback.HealthCheckCallback;
-import org.apache.eventmesh.dashboard.console.health.check.AbstractHealthCheckService;
-import org.apache.eventmesh.dashboard.console.health.check.config.HealthCheckObjectConfig;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_TYPE_STORAGE;
+
+import org.apache.eventmesh.dashboard.console.function.health.annotation.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.function.health.callback.HealthCheckCallback;
+import org.apache.eventmesh.dashboard.console.function.health.check.AbstractHealthCheckService;
+import org.apache.eventmesh.dashboard.console.function.health.check.config.HealthCheckObjectConfig;
 
 import java.time.Duration;
 import java.util.Objects;
@@ -30,13 +32,15 @@
 import io.lettuce.core.RedisURI.Builder;
 import io.lettuce.core.api.async.RedisAsyncCommands;
 
+import lombok.extern.slf4j.Slf4j;
 
-@HealthCheckType(type = "storage", subType = "redis")
-public class StorageRedisCheck extends AbstractHealthCheckService {
+@Slf4j
+@HealthCheckType(type = HEALTH_CHECK_TYPE_STORAGE, subType = "redis")
+public class RedisCheck extends AbstractHealthCheckService {
 
  private RedisClient redisClient;
 
- public StorageRedisCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
+ public RedisCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
  super(healthCheckObjectConfig);
  }
 
@@ -55,6 +59,7 @@ public void doCheck(HealthCheckCallback callback) {
  return null;
  });
  } catch (Exception e) {
+ log.error(e.toString());
  callback.onFail(e);
  }
  }
@@ -79,4 +84,11 @@ public void init() {
  }
  redisClient = RedisClient.create(redisUrl);
  }
+
+ @Override
+ public void destroy() {
+ if (redisClient != null) {
+ redisClient.shutdown();
+ }
+ }
 }
diff --git a/.../dashboard/console/function/health/check/impl/storage/rocketmq4/Rocketmq4BrokerCheck.java b/.../dashboard/console/function/health/check/impl/storage/rocketmq4/Rocketmq4BrokerCheck.java
@@ -0,0 +1,98 @@
+/*
+ * 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.dashboard.console.function.health.check.impl.storage.rocketmq4;
+
+
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_SUBTYPE_ROCKETMQ_BROKER;
+import static org.apache.eventmesh.dashboard.console.constant.health.HealthCheckTypeConstant.HEALTH_CHECK_TYPE_STORAGE;
+
+import org.apache.eventmesh.dashboard.console.function.health.annotation.HealthCheckType;
+import org.apache.eventmesh.dashboard.console.function.health.callback.HealthCheckCallback;
+import org.apache.eventmesh.dashboard.console.function.health.check.AbstractHealthCheckService;
+import org.apache.eventmesh.dashboard.console.function.health.check.config.HealthCheckObjectConfig;
+
+import org.apache.rocketmq.common.protocol.RequestCode;
+import org.apache.rocketmq.remoting.InvokeCallback;
+import org.apache.rocketmq.remoting.RemotingClient;
+import org.apache.rocketmq.remoting.netty.NettyClientConfig;
+import org.apache.rocketmq.remoting.netty.NettyRemotingClient;
+import org.apache.rocketmq.remoting.netty.ResponseFuture;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@HealthCheckType(type = HEALTH_CHECK_TYPE_STORAGE, subType = HEALTH_CHECK_SUBTYPE_ROCKETMQ_BROKER)
+public class Rocketmq4BrokerCheck extends AbstractHealthCheckService {
+
+ private RemotingClient remotingClient;
+
+
+ public Rocketmq4BrokerCheck(HealthCheckObjectConfig healthCheckObjectConfig) {
+ super(healthCheckObjectConfig);
+ }
+
+ @Override
+ public void doCheck(HealthCheckCallback callback) {
+ try {
+ RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_RUNTIME_INFO, null);
+ remotingClient.invokeAsync(getConfig().getRocketmqConfig().getBrokerUrl(), request, getConfig().getRequestTimeoutMillis(),
+ new InvokeCallback() {
+ @Override
+ public void operationComplete(ResponseFuture responseFuture) {
+ if (responseFuture.isSendRequestOK()) {
+ callback.onSuccess();
+ } else {
+ callback.onFail(new RuntimeException("RocketmqNameServerCheck failed caused by " + responseFuture.getCause()));
+ }
+ }
+
+ });
+ } catch (Exception e) {
+ log.error("RocketmqCheck failed.", e);
+ callback.onFail(e);
+ }
+ }
+
+ @Override
+ public void init() {
+ if (getConfig().getRocketmqConfig().getBrokerUrl() == null || getConfig().getRocketmqConfig().getBrokerUrl().isEmpty()) {
+ throw new IllegalArgumentException("RocketmqCheck failed. BrokerUrl is null.");
+ }
+
+ NettyClientConfig config = new NettyClientConfig();
+ config.setUseTLS(false);
+ remotingClient = new NettyRemotingClient(config);
+ remotingClient.start();
+
+ if (getConfig().getConnectUrl() == null || getConfig().getConnectUrl().isEmpty()) {
+ if (getConfig().getHost() != null && getConfig().getPort() != null) {
+ getConfig().setConnectUrl(getConfig().getHost() + ":" + getConfig().getPort());
+ }
+ }
+
+ if (getConfig().getConnectUrl() == null) {
+ log.error("RocketmqCheck failed. ConnectUrl is null.");
+ }
+ }
+
+ @Override
+ public void destroy() {
+
+ }
+}