BIGTOP-4162: Add health check for components (#19)

bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/AgentApplication.java

@@ -19,7 +19,6 @@
 package org.apache.bigtop.manager.agent;
 
 import org.apache.bigtop.manager.agent.monitoring.AgentHostMonitoring;
-import org.apache.bigtop.manager.agent.monitoring.ZookeeperHealthyMonitoring;
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.boot.SpringApplication;
@@ -54,9 +53,4 @@ public static void main(String[] args) {
     @Qualifier("memMultiGauge") public MultiGauge memMultiGauge(MeterRegistry meterRegistry) {
         return AgentHostMonitoring.newMemMultiGauge(meterRegistry);
     }
-
-    @Bean
-    @Qualifier("zookeeperMultiGauge") public MultiGauge zookeeperMultiGauge(MeterRegistry meterRegistry) {
-        return ZookeeperHealthyMonitoring.registerZookeeperHealthyGauge(meterRegistry);
-    }
 }

bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/metrics/MetricsCollector.java

@@ -19,7 +19,6 @@
 package org.apache.bigtop.manager.agent.metrics;
 
 import org.apache.bigtop.manager.agent.monitoring.AgentHostMonitoring;
-import org.apache.bigtop.manager.agent.monitoring.ZookeeperHealthyMonitoring;
 
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.scheduling.annotation.Async;
@@ -44,9 +43,6 @@ public class MetricsCollector {
     @Resource
     @Qualifier("cpuMultiGauge") private MultiGauge cpuMultiGauge;
 
-    @Resource
-    @Qualifier("zookeeperMultiGauge") private MultiGauge zookeeperMultiGauge;
-
     @Async
     @Scheduled(cron = "*/10 * *  * * ?")
     public void collect() {
@@ -58,6 +54,5 @@ private void scrape() {
         AgentHostMonitoring.diskMultiGaugeUpdateData(diskMultiGauge);
         AgentHostMonitoring.memMultiGaugeUpdateData(memMultiGauge);
         AgentHostMonitoring.cpuMultiGaugeUpdateData(cpuMultiGauge);
-        ZookeeperHealthyMonitoring.zookeeperUpdateStatus(zookeeperMultiGauge);
     }
 }

bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/ComponentStatusServiceGrpcImpl.java

@@ -0,0 +1,66 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.agent.service;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.common.message.entity.payload.CommandPayload;
+import org.apache.bigtop.manager.common.message.entity.pojo.ScriptInfo;
+import org.apache.bigtop.manager.common.shell.ShellResult;
+import org.apache.bigtop.manager.common.utils.JsonUtils;
+import org.apache.bigtop.manager.grpc.generated.ComponentStatusReply;
+import org.apache.bigtop.manager.grpc.generated.ComponentStatusRequest;
+import org.apache.bigtop.manager.grpc.generated.ComponentStatusServiceGrpc;
+import org.apache.bigtop.manager.stack.core.executor.StackExecutor;
+
+import io.grpc.Status;
+import io.grpc.stub.StreamObserver;
+import lombok.extern.slf4j.Slf4j;
+import net.devh.boot.grpc.server.service.GrpcService;
+
+@Slf4j
+@GrpcService
+public class ComponentStatusServiceGrpcImpl extends ComponentStatusServiceGrpc.ComponentStatusServiceImplBase {
+
+    @Override
+    public void getComponentStatus(
+            ComponentStatusRequest request, StreamObserver<ComponentStatusReply> responseObserver) {
+
+        try {
+            CommandPayload commandPayload = new CommandPayload();
+            commandPayload.setCommand(Command.STATUS);
+            commandPayload.setServiceName(request.getServiceName());
+            commandPayload.setComponentName(request.getComponentName());
+            commandPayload.setCommandScript(JsonUtils.readFromString(request.getCommandScript(), ScriptInfo.class));
+            commandPayload.setRoot(request.getRoot());
+            commandPayload.setStackName(request.getStackName());
+            commandPayload.setStackVersion(request.getStackVersion());
+            ShellResult shellResult = StackExecutor.execute(commandPayload);
+
+            ComponentStatusReply reply = ComponentStatusReply.newBuilder()
+                    .setStatus(shellResult.getExitCode())
+                    .build();
+            responseObserver.onNext(reply);
+            responseObserver.onCompleted();
+        } catch (Exception e) {
+            log.error("Error getting component status", e);
+            Status status = Status.UNKNOWN.withDescription(e.getMessage());
+            responseObserver.onError(status.asRuntimeException());
+        }
+    }
+}

bigtop-manager-grpc/src/main/resources/proto/component_status.proto

@@ -0,0 +1,41 @@
+/*
+ * 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
+ *
+ *    https://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.
+ */
+syntax = "proto3";
+
+option java_multiple_files = true;
+option java_package = "org.apache.bigtop.manager.grpc.generated";
+option java_outer_classname = "ComponentStatusProto";
+
+service ComponentStatusService {
+  rpc GetComponentStatus (ComponentStatusRequest) returns (ComponentStatusReply) {}
+}
+
+message ComponentStatusRequest {
+  string service_name = 1;
+  string component_name = 2;
+  string command_script = 3;
+  // TODO Unnecessary, should be removed in the future
+  string root = 4;
+  string stack_name = 5;
+  string stack_version = 6;
+}
+
+message ComponentStatusReply {
+  int32 status = 1;
+}

bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/grpc/GrpcClient.java

@@ -34,7 +34,6 @@
 import lombok.extern.slf4j.Slf4j;
 
 import java.lang.reflect.Constructor;
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
@@ -63,7 +62,8 @@ public static Boolean isChannelAlive(String host) {
 
     @SuppressWarnings("unchecked")
     public static <T extends AbstractBlockingStub<T>> T getBlockingStub(String host, Class<T> clazz) {
-        Map<String, AbstractBlockingStub<?>> innerMap = BLOCKING_STUBS.computeIfAbsent(host, k -> new HashMap<>());
+        Map<String, AbstractBlockingStub<?>> innerMap =
+                BLOCKING_STUBS.computeIfAbsent(host, k -> new ConcurrentHashMap<>());
         return (T) innerMap.computeIfAbsent(clazz.getName(), k -> {
             T instance = T.newStub(getFactory(clazz), getChannel(host));
             log.info("Instance: {} created.", k);
@@ -73,7 +73,7 @@ public static <T extends AbstractBlockingStub<T>> T getBlockingStub(String host,
 
     @SuppressWarnings("unchecked")
     public static <T extends AbstractAsyncStub<T>> T getAsyncStub(String host, Class<T> clazz) {
-        Map<String, AbstractAsyncStub<?>> innerMap = ASYNC_STUBS.computeIfAbsent(host, k -> new HashMap<>());
+        Map<String, AbstractAsyncStub<?>> innerMap = ASYNC_STUBS.computeIfAbsent(host, k -> new ConcurrentHashMap<>());
         return (T) innerMap.computeIfAbsent(clazz.getName(), k -> {
             T instance = T.newStub(getFactory(clazz), getChannel(host));
             log.info("Instance: {} created.", k);
@@ -83,7 +83,8 @@ public static <T extends AbstractAsyncStub<T>> T getAsyncStub(String host, Class
 
     @SuppressWarnings("unchecked")
     public static <T extends AbstractFutureStub<T>> T getFutureStub(String host, Class<T> clazz) {
-        Map<String, AbstractFutureStub<?>> innerMap = FUTURE_STUBS.computeIfAbsent(host, k -> new HashMap<>());
+        Map<String, AbstractFutureStub<?>> innerMap =
+                FUTURE_STUBS.computeIfAbsent(host, k -> new ConcurrentHashMap<>());
         return (T) innerMap.computeIfAbsent(clazz.getName(), k -> {
             T instance = T.newStub(getFactory(clazz), getChannel(host));
             log.info("Instance: {} created.", k);