diff --git a/.licenserc.yaml b/.licenserc.yaml index 8d13be57..b5a6f7b8 100644 --- a/.licenserc.yaml +++ b/.licenserc.yaml @@ -35,7 +35,6 @@ header: - 'mvnw.cmd' - 'LICENSE' - 'pnpm-lock.yaml' - - '**/*.proto' - '**/*.txt' comment: on-failure diff --git a/bigtop-manager-agent/pom.xml b/bigtop-manager-agent/pom.xml index eb1bae1a..a18da7bc 100644 --- a/bigtop-manager-agent/pom.xml +++ b/bigtop-manager-agent/pom.xml @@ -46,18 +46,19 @@ org.apache.bigtop bigtop-manager-common - ${revision} - + + org.apache.bigtop + bigtop-manager-grpc + org.apache.bigtop bigtop-manager-stack-core - ${revision} org.springframework.boot - spring-boot-starter-websocket + spring-boot-starter-web @@ -108,6 +109,10 @@ oshi-core + + net.devh + grpc-server-spring-boot-starter + diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/AgentApplication.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/AgentApplication.java index cf3c29f1..39acaf52 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/AgentApplication.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/AgentApplication.java @@ -25,10 +25,14 @@ import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.context.annotation.Bean; +import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.annotation.EnableScheduling; import io.micrometer.core.instrument.MeterRegistry; import io.micrometer.core.instrument.MultiGauge; +@EnableAsync +@EnableScheduling @SpringBootApplication(scanBasePackages = {"org.apache.bigtop.manager.agent", "org.apache.bigtop.manager.common"}) public class AgentApplication { diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/AbstractCommandExecutor.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/AbstractCommandExecutor.java index e54bd614..8a44bb46 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/AbstractCommandExecutor.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/AbstractCommandExecutor.java @@ -18,26 +18,25 @@ */ package org.apache.bigtop.manager.agent.executor; -import org.apache.bigtop.manager.agent.holder.SpringContextHolder; import org.apache.bigtop.manager.common.constants.MessageConstants; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandResponseMessage; import org.apache.bigtop.manager.common.shell.ShellResult; import org.apache.bigtop.manager.common.utils.Environments; +import org.apache.bigtop.manager.grpc.generated.CommandReply; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; import lombok.extern.slf4j.Slf4j; @Slf4j public abstract class AbstractCommandExecutor implements CommandExecutor { - protected CommandRequestMessage commandRequestMessage; + protected CommandRequest commandRequest; - protected CommandResponseMessage commandResponseMessage; + protected CommandReply.Builder commandReplyBuilder; @Override - public void execute(CommandRequestMessage message) { - commandRequestMessage = message; - commandResponseMessage = new CommandResponseMessage(); + public CommandReply execute(CommandRequest request) { + commandRequest = request; + commandReplyBuilder = CommandReply.newBuilder(); try { if (Environments.isDevMode()) { @@ -46,24 +45,23 @@ public void execute(CommandRequestMessage message) { doExecute(); } } catch (Exception e) { - commandResponseMessage.setCode(MessageConstants.FAIL_CODE); - commandResponseMessage.setResult(e.getMessage()); + commandReplyBuilder.setCode(MessageConstants.FAIL_CODE); + commandReplyBuilder.setResult(e.getMessage()); - log.error("Run command failed, {}", message, e); + log.error("Run command failed, {}", request, e); } - commandResponseMessage.setCommandMessageType(message.getCommandMessageType()); - commandResponseMessage.setMessageId(message.getMessageId()); - commandResponseMessage.setHostname(message.getHostname()); - commandResponseMessage.setTaskId(message.getTaskId()); - commandResponseMessage.setStageId(message.getStageId()); - commandResponseMessage.setJobId(message.getJobId()); - SpringContextHolder.getAgentWebSocket().sendMessage(commandResponseMessage); + commandReplyBuilder.setType(request.getType()); + commandReplyBuilder.setHostname(request.getHostname()); + commandReplyBuilder.setTaskId(request.getTaskId()); + commandReplyBuilder.setStageId(request.getStageId()); + commandReplyBuilder.setJobId(request.getJobId()); + return commandReplyBuilder.build(); } protected void doExecuteOnDevMode() { - commandResponseMessage.setCode(MessageConstants.SUCCESS_CODE); - commandResponseMessage.setResult(ShellResult.success().getResult()); + commandReplyBuilder.setCode(MessageConstants.SUCCESS_CODE); + commandReplyBuilder.setResult(ShellResult.success().getResult()); } protected abstract void doExecute(); diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CacheDistributeCommandExecutor.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CacheDistributeCommandExecutor.java index 44018298..2900f171 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CacheDistributeCommandExecutor.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CacheDistributeCommandExecutor.java @@ -20,9 +20,9 @@ import org.apache.bigtop.manager.common.constants.Constants; import org.apache.bigtop.manager.common.constants.MessageConstants; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; import org.apache.bigtop.manager.common.message.entity.payload.CacheMessagePayload; import org.apache.bigtop.manager.common.utils.JsonUtils; +import org.apache.bigtop.manager.grpc.generated.CommandType; import org.apache.bigtop.manager.stack.common.utils.linux.LinuxFileUtils; import org.springframework.beans.factory.config.ConfigurableBeanFactory; @@ -46,15 +46,15 @@ public class CacheDistributeCommandExecutor extends AbstractCommandExecutor { @Override - public CommandMessageType getCommandMessageType() { - return CommandMessageType.CACHE_DISTRIBUTE; + public CommandType getCommandType() { + return CommandType.CACHE_DISTRIBUTE; } @Override public void doExecute() { CacheMessagePayload cacheMessagePayload = - JsonUtils.readFromString(commandRequestMessage.getMessagePayload(), CacheMessagePayload.class); - log.info("[agent executeTask] taskEvent is: {}", commandRequestMessage); + JsonUtils.readFromString(commandRequest.getPayload(), CacheMessagePayload.class); + log.info("[agent executeTask] taskEvent is: {}", commandRequest); String cacheDir = Constants.STACK_CACHE_DIR; LinuxFileUtils.createDirectories(cacheDir, "root", "root", "rwxr-xr-x", false); @@ -67,8 +67,8 @@ public void doExecute() { JsonUtils.writeToFile(cacheDir + REPOS_INFO, cacheMessagePayload.getRepoInfo()); JsonUtils.writeToFile(cacheDir + CLUSTER_INFO, cacheMessagePayload.getClusterInfo()); - commandResponseMessage.setCode(MessageConstants.SUCCESS_CODE); - commandResponseMessage.setResult( - MessageFormat.format("Host [{0}] cached successful!!!", commandRequestMessage.getHostname())); + commandReplyBuilder.setCode(MessageConstants.SUCCESS_CODE); + commandReplyBuilder.setResult( + MessageFormat.format("Host [{0}] cached successful!!!", commandRequest.getHostname())); } } diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutor.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutor.java index 04437595..d11d0e63 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutor.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutor.java @@ -18,8 +18,9 @@ */ package org.apache.bigtop.manager.agent.executor; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; +import org.apache.bigtop.manager.grpc.generated.CommandReply; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandType; /** * Interface for executing commands on agent. @@ -27,14 +28,14 @@ public interface CommandExecutor { /** - * Get the type of the command message. - * @return CommandMessageType - the type of the command message. + * Get the type of the command. + * @return CommandType - the type of the command. */ - CommandMessageType getCommandMessageType(); + CommandType getCommandType(); /** * Execute the command. - * @param message - the message for command that needs to be executed. + * @param request - the request for command that needs to be executed. */ - void execute(CommandRequestMessage message); + CommandReply execute(CommandRequest request); } diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutors.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutors.java index e1a8ed12..01704410 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutors.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/CommandExecutors.java @@ -19,7 +19,7 @@ package org.apache.bigtop.manager.agent.executor; import org.apache.bigtop.manager.agent.holder.SpringContextHolder; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; +import org.apache.bigtop.manager.grpc.generated.CommandType; import lombok.extern.slf4j.Slf4j; @@ -32,14 +32,14 @@ public class CommandExecutors { private static final AtomicBoolean LOADED = new AtomicBoolean(false); - private static final Map COMMAND_EXECUTORS = new HashMap<>(); + private static final Map COMMAND_EXECUTORS = new HashMap<>(); - public static CommandExecutor getCommandExecutor(CommandMessageType commandMessageType) { + public static CommandExecutor getCommandExecutor(CommandType commandType) { if (!LOADED.get()) { load(); } - String beanName = COMMAND_EXECUTORS.get(commandMessageType); + String beanName = COMMAND_EXECUTORS.get(commandType); return SpringContextHolder.getApplicationContext().getBean(beanName, CommandExecutor.class); } @@ -52,16 +52,16 @@ private static synchronized void load() { SpringContextHolder.getCommandExecutors().entrySet()) { String beanName = entry.getKey(); CommandExecutor commandExecutor = entry.getValue(); - if (COMMAND_EXECUTORS.containsKey(commandExecutor.getCommandMessageType())) { - log.error("Duplicate CommandExecutor with message type: {}", commandExecutor.getCommandMessageType()); + if (COMMAND_EXECUTORS.containsKey(commandExecutor.getCommandType())) { + log.error("Duplicate CommandExecutor with message type: {}", commandExecutor.getCommandType()); continue; } - COMMAND_EXECUTORS.put(commandExecutor.getCommandMessageType(), beanName); + COMMAND_EXECUTORS.put(commandExecutor.getCommandType(), beanName); log.info( "Load JobRunner: {} with identifier: {}", commandExecutor.getClass().getName(), - commandExecutor.getCommandMessageType()); + commandExecutor.getCommandType()); } LOADED.set(true); diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/ComponentCommandExecutor.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/ComponentCommandExecutor.java index c7af4743..9ec70206 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/ComponentCommandExecutor.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/ComponentCommandExecutor.java @@ -18,10 +18,10 @@ */ package org.apache.bigtop.manager.agent.executor; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; import org.apache.bigtop.manager.common.message.entity.payload.CommandPayload; import org.apache.bigtop.manager.common.shell.ShellResult; import org.apache.bigtop.manager.common.utils.JsonUtils; +import org.apache.bigtop.manager.grpc.generated.CommandType; import org.apache.bigtop.manager.stack.core.executor.StackExecutor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; @@ -35,8 +35,8 @@ public class ComponentCommandExecutor extends AbstractCommandExecutor { @Override - public CommandMessageType getCommandMessageType() { - return CommandMessageType.COMPONENT; + public CommandType getCommandType() { + return CommandType.COMPONENT; } @Override @@ -46,12 +46,11 @@ protected void doExecuteOnDevMode() { @Override public void doExecute() { - CommandPayload commandPayload = - JsonUtils.readFromString(commandRequestMessage.getMessagePayload(), CommandPayload.class); - log.info("[agent executeTask] taskEvent is: {}", commandRequestMessage); + CommandPayload commandPayload = JsonUtils.readFromString(commandRequest.getPayload(), CommandPayload.class); + log.info("[agent executeTask] taskEvent is: {}", commandRequest); ShellResult shellResult = StackExecutor.execute(commandPayload); - commandResponseMessage.setCode(shellResult.getExitCode()); - commandResponseMessage.setResult(shellResult.getResult()); + commandReplyBuilder.setCode(shellResult.getExitCode()); + commandReplyBuilder.setResult(shellResult.getResult()); } } diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/HostCheckCommandExecutor.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/HostCheckCommandExecutor.java index fa85b099..82eb1e93 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/HostCheckCommandExecutor.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/executor/HostCheckCommandExecutor.java @@ -19,9 +19,9 @@ package org.apache.bigtop.manager.agent.executor; import org.apache.bigtop.manager.common.constants.MessageConstants; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; import org.apache.bigtop.manager.common.shell.ShellResult; import org.apache.bigtop.manager.common.utils.os.TimeSyncDetection; +import org.apache.bigtop.manager.grpc.generated.CommandType; import org.springframework.beans.factory.config.ConfigurableBeanFactory; import org.springframework.context.annotation.Scope; @@ -38,15 +38,15 @@ public class HostCheckCommandExecutor extends AbstractCommandExecutor { @Override - public CommandMessageType getCommandMessageType() { - return CommandMessageType.HOST_CHECK; + public CommandType getCommandType() { + return CommandType.HOST_CHECK; } @Override public void doExecute() { ShellResult shellResult = runChecks(List.of(this::checkTimeSync)); - commandResponseMessage.setCode(shellResult.getExitCode()); - commandResponseMessage.setResult(shellResult.getResult()); + commandReplyBuilder.setCode(shellResult.getExitCode()); + commandReplyBuilder.setResult(shellResult.getResult()); } private ShellResult runChecks(List> suppliers) { diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/holder/SpringContextHolder.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/holder/SpringContextHolder.java index c062a8eb..7c2ab255 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/holder/SpringContextHolder.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/holder/SpringContextHolder.java @@ -19,7 +19,6 @@ package org.apache.bigtop.manager.agent.holder; import org.apache.bigtop.manager.agent.executor.CommandExecutor; -import org.apache.bigtop.manager.agent.ws.AgentWebSocketHandler; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; @@ -41,10 +40,6 @@ public void setApplicationContext(@Nonnull ApplicationContext applicationContext SpringContextHolder.applicationContext = applicationContext; } - public static AgentWebSocketHandler getAgentWebSocket() { - return applicationContext.getBean(AgentWebSocketHandler.class); - } - public static Map getCommandExecutors() { return applicationContext.getBeansOfType(CommandExecutor.class); } diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/metrics/MetricsCollector.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/metrics/MetricsCollector.java index 7238f042..7028af56 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/metrics/MetricsCollector.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/metrics/MetricsCollector.java @@ -23,8 +23,6 @@ import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.scheduling.annotation.Async; -import org.springframework.scheduling.annotation.EnableAsync; -import org.springframework.scheduling.annotation.EnableScheduling; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; @@ -35,21 +33,19 @@ @Slf4j @Component -@EnableScheduling -@EnableAsync public class MetricsCollector { - @Qualifier("diskMultiGauge") @Resource - private MultiGauge diskMultiGauge; + @Resource + @Qualifier("diskMultiGauge") private MultiGauge diskMultiGauge; - @Qualifier("memMultiGauge") @Resource - private MultiGauge memMultiGauge; + @Resource + @Qualifier("memMultiGauge") private MultiGauge memMultiGauge; - @Qualifier("cpuMultiGauge") @Resource - private MultiGauge cpuMultiGauge; + @Resource + @Qualifier("cpuMultiGauge") private MultiGauge cpuMultiGauge; - @Qualifier("zookeeperMultiGauge") @Resource - private MultiGauge zookeeperMultiGauge; + @Resource + @Qualifier("zookeeperMultiGauge") private MultiGauge zookeeperMultiGauge; @Async @Scheduled(cron = "*/10 * * * * ?") diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/monitoring/AgentHostMonitoring.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/monitoring/AgentHostMonitoring.java index f3a3cde0..703f28b1 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/monitoring/AgentHostMonitoring.java +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/monitoring/AgentHostMonitoring.java @@ -18,9 +18,6 @@ */ package org.apache.bigtop.manager.agent.monitoring; -import org.apache.bigtop.manager.agent.enums.AgentExceptionStatus; -import org.apache.bigtop.manager.agent.exception.AgentException; - import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; @@ -333,7 +330,7 @@ public static void diskMultiGaugeUpdateData(MultiGauge diskMultiGauge) { Map, Map, Double>> diskGauge = getDiskGauge(getHostInfo()); multiGaugeUpdateData(diskMultiGauge, diskGauge); } catch (UnknownHostException e) { - throw new AgentException(AgentExceptionStatus.AGENT_MONITORING_ERROR); + throw new RuntimeException("Get agent host monitoring info failed"); } } @@ -342,7 +339,7 @@ public static void memMultiGaugeUpdateData(MultiGauge memMultiGauge) { Map, Map, Double>> diskGauge = getMEMGauge(getHostInfo()); multiGaugeUpdateData(memMultiGauge, diskGauge); } catch (UnknownHostException e) { - throw new AgentException(AgentExceptionStatus.AGENT_MONITORING_ERROR); + throw new RuntimeException("Get agent host monitoring info failed"); } } @@ -351,7 +348,7 @@ public static void cpuMultiGaugeUpdateData(MultiGauge cpuMultiGauge) { Map, Map, Double>> diskGauge = getCPUGauge(getHostInfo()); multiGaugeUpdateData(cpuMultiGauge, diskGauge); } catch (UnknownHostException e) { - throw new AgentException(AgentExceptionStatus.AGENT_MONITORING_ERROR); + throw new RuntimeException("Get agent host monitoring info failed"); } } } diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/CommandScheduler.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/CommandScheduler.java deleted file mode 100644 index 0d1a022a..00000000 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/CommandScheduler.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.scheduler; - -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; - -/** - * Interface for scheduling commands on agent. - */ -public interface CommandScheduler { - - /** - * Submit a command request message to the scheduler. - * @param message - the command request message that needs to be submitted. - */ - void submit(CommandRequestMessage message); - - /** - * Start the command scheduler. - */ - void start(); - - /** - * Stop the command scheduler. - */ - void stop(); -} diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/DefaultCommandScheduler.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/DefaultCommandScheduler.java deleted file mode 100644 index 55eb8ba1..00000000 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/scheduler/DefaultCommandScheduler.java +++ /dev/null @@ -1,83 +0,0 @@ -/* - * 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.scheduler; - -import org.apache.bigtop.manager.agent.executor.CommandExecutor; -import org.apache.bigtop.manager.agent.executor.CommandExecutors; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; - -import org.slf4j.MDC; -import org.springframework.stereotype.Component; - -import lombok.extern.slf4j.Slf4j; - -import jakarta.annotation.PostConstruct; -import java.util.concurrent.Executor; -import java.util.concurrent.Executors; -import java.util.concurrent.LinkedBlockingQueue; - -@Slf4j -@Component -public class DefaultCommandScheduler implements CommandScheduler { - - private final LinkedBlockingQueue queue = new LinkedBlockingQueue<>(); - - private final Executor executor = Executors.newSingleThreadExecutor(); - - private volatile boolean running = true; - - @Override - public void submit(CommandRequestMessage message) { - queue.offer(() -> { - try { - MDC.put("taskId", message.getTaskId().toString()); - CommandExecutor commandExecutor = CommandExecutors.getCommandExecutor(message.getCommandMessageType()); - commandExecutor.execute(message); - } catch (Exception e) { - log.error("Error when running command", e); - } finally { - MDC.clear(); - } - }); - } - - @Override - public void start() { - running = true; - } - - @Override - public void stop() { - running = false; - } - - @PostConstruct - public void init() { - executor.execute(() -> { - while (running) { - try { - Runnable runnable = queue.take(); - runnable.run(); - } catch (InterruptedException e) { - log.warn("Error when running command", e); - } - } - }); - } -} diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/CommandServiceGrpcImpl.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/CommandServiceGrpcImpl.java new file mode 100644 index 00000000..42591bab --- /dev/null +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/CommandServiceGrpcImpl.java @@ -0,0 +1,54 @@ +/* + * 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.agent.executor.CommandExecutor; +import org.apache.bigtop.manager.agent.executor.CommandExecutors; +import org.apache.bigtop.manager.grpc.generated.CommandReply; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandServiceGrpc; + +import org.slf4j.MDC; + +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 CommandServiceGrpcImpl extends CommandServiceGrpc.CommandServiceImplBase { + + @Override + public void exec(CommandRequest request, StreamObserver responseObserver) { + try { + MDC.put("taskId", String.valueOf(request.getTaskId())); + CommandExecutor commandExecutor = CommandExecutors.getCommandExecutor(request.getType()); + CommandReply reply = commandExecutor.execute(request); + responseObserver.onNext(reply); + responseObserver.onCompleted(); + } catch (Exception e) { + log.error("Error when running command", e); + Status status = Status.UNKNOWN.withDescription(e.getMessage()); + responseObserver.onError(status.asRuntimeException()); + } finally { + MDC.clear(); + } + } +} diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/HostInfoServiceGrpcImpl.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/HostInfoServiceGrpcImpl.java new file mode 100644 index 00000000..c5a92446 --- /dev/null +++ b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/service/HostInfoServiceGrpcImpl.java @@ -0,0 +1,74 @@ +/* + * 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.utils.os.OSDetection; +import org.apache.bigtop.manager.grpc.generated.HostInfoReply; +import org.apache.bigtop.manager.grpc.generated.HostInfoRequest; +import org.apache.bigtop.manager.grpc.generated.HostInfoServiceGrpc; + +import com.sun.management.OperatingSystemMXBean; +import io.grpc.Status; +import io.grpc.stub.StreamObserver; +import lombok.extern.slf4j.Slf4j; +import net.devh.boot.grpc.server.service.GrpcService; + +import java.lang.management.ManagementFactory; +import java.net.InetAddress; + +@Slf4j +@GrpcService +public class HostInfoServiceGrpcImpl extends HostInfoServiceGrpc.HostInfoServiceImplBase { + + @Override + public void getHostInfo(HostInfoRequest request, StreamObserver responseObserver) { + HostInfoReply.Builder builder = HostInfoReply.newBuilder(); + + try { + InetAddress addr = InetAddress.getLocalHost(); + builder.setHostname(addr.getHostName()); + builder.setIpv4(addr.getHostAddress()); + + OperatingSystemMXBean osmxb = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); + builder.setOs(OSDetection.getOS()); + builder.setVersion(OSDetection.getVersion()); + builder.setArch(OSDetection.getArch()); + builder.setAvailableProcessors(osmxb.getAvailableProcessors()); + builder.setProcessCpuTime(osmxb.getProcessCpuTime()); + builder.setTotalMemorySize(osmxb.getTotalMemorySize()); + builder.setFreeMemorySize(osmxb.getFreeMemorySize()); + builder.setTotalSwapSpaceSize(osmxb.getTotalSwapSpaceSize()); + builder.setFreeSwapSpaceSize(osmxb.getFreeSwapSpaceSize()); + builder.setCommittedVirtualMemorySize(osmxb.getCommittedVirtualMemorySize()); + + builder.setCpuLoad(String.valueOf(osmxb.getCpuLoad())); + builder.setProcessCpuLoad(String.valueOf(osmxb.getProcessCpuLoad())); + builder.setSystemLoadAverage(String.valueOf(osmxb.getSystemLoadAverage())); + + builder.setFreeDisk(OSDetection.freeDisk()); + builder.setTotalDisk(OSDetection.totalDisk()); + responseObserver.onNext(builder.build()); + responseObserver.onCompleted(); + } catch (Exception e) { + log.error("Error getting host info", e); + Status status = Status.UNKNOWN.withDescription(e.getMessage()); + responseObserver.onError(status.asRuntimeException()); + } + } +} diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/ws/AgentWebSocketHandler.java b/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/ws/AgentWebSocketHandler.java deleted file mode 100644 index fba40103..00000000 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/ws/AgentWebSocketHandler.java +++ /dev/null @@ -1,217 +0,0 @@ -/* - * 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.ws; - -import org.apache.bigtop.manager.agent.holder.SpringContextHolder; -import org.apache.bigtop.manager.agent.scheduler.CommandScheduler; -import org.apache.bigtop.manager.common.config.ApplicationConfig; -import org.apache.bigtop.manager.common.constants.Constants; -import org.apache.bigtop.manager.common.message.entity.BaseMessage; -import org.apache.bigtop.manager.common.message.entity.HeartbeatMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; -import org.apache.bigtop.manager.common.message.entity.pojo.HostInfo; -import org.apache.bigtop.manager.common.message.serializer.MessageDeserializer; -import org.apache.bigtop.manager.common.utils.os.OSDetection; -import org.apache.bigtop.manager.common.ws.AbstractBinaryWebSocketHandler; - -import org.springframework.boot.context.event.ApplicationStartedEvent; -import org.springframework.context.ApplicationListener; -import org.springframework.stereotype.Component; -import org.springframework.web.socket.BinaryMessage; -import org.springframework.web.socket.CloseStatus; -import org.springframework.web.socket.WebSocketSession; -import org.springframework.web.socket.client.standard.StandardWebSocketClient; - -import com.sun.management.OperatingSystemMXBean; -import lombok.Getter; -import lombok.Setter; -import lombok.extern.slf4j.Slf4j; - -import jakarta.annotation.Nonnull; -import jakarta.annotation.Resource; -import java.lang.management.ManagementFactory; -import java.math.BigDecimal; -import java.net.InetAddress; -import java.text.MessageFormat; -import java.util.concurrent.Executors; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; - -import static org.apache.bigtop.manager.common.constants.Constants.WS_BINARY_MESSAGE_SIZE_LIMIT; - -@Slf4j -@Component -public class AgentWebSocketHandler extends AbstractBinaryWebSocketHandler - implements ApplicationListener { - - @Resource - private ApplicationConfig applicationConfig; - - @Resource - private MessageDeserializer deserializer; - - @Resource - private CommandScheduler commandScheduler; - - @Getter - @Setter - private WebSocketSession session; - - private final ScheduledExecutorService executor = Executors.newScheduledThreadPool(1); - - private HostInfo hostInfo; - - public void sendMessage(BaseMessage message) { - log.info("send message to server: {}", message); - if (session == null) { - log.warn("session is null, can't send message to server"); - return; - } - - try { - super.sendMessage(session, message); - } catch (Exception e) { - log.error("Error sending message to server: {}", message, e); - } - } - - @Override - protected void handleBinaryMessage(@Nonnull WebSocketSession session, BinaryMessage message) { - BaseMessage baseMessage = deserializer.deserialize(message.getPayload().array()); - - handleMessage(session, baseMessage); - } - - private void handleMessage(WebSocketSession session, BaseMessage baseMessage) { - log.info( - "Received message type: {}, session: {}", baseMessage.getClass().getSimpleName(), session); - - if (baseMessage instanceof CommandRequestMessage commandRequestMessage) { - commandScheduler.submit(commandRequestMessage); - } else { - log.error("Unrecognized message type: {}", baseMessage.getClass().getSimpleName()); - } - } - - @Override - public void afterConnectionEstablished(@Nonnull WebSocketSession session) { - this.setSession(session); - executor.scheduleAtFixedRate( - () -> { - try { - HeartbeatMessage heartbeatMessage = new HeartbeatMessage(); - heartbeatMessage.setHostInfo(hostInfo); - WebSocketSession innerSession = - SpringContextHolder.getAgentWebSocket().getSession(); - if (null == innerSession || !innerSession.isOpen()) { - connectToServer(); - } - innerSession = SpringContextHolder.getAgentWebSocket().getSession(); - if (null != innerSession && innerSession.isOpen()) { - super.sendMessage(innerSession, heartbeatMessage); - } - } catch (Exception e) { - log.error(MessageFormat.format("Error sending heartbeat to server: {0}", e.getMessage())); - } - }, - 3, - 10, - TimeUnit.SECONDS); - } - - @Override - public void afterConnectionClosed(@Nonnull WebSocketSession session, @Nonnull CloseStatus status) { - log.info("WebSocket connection closed unexpectedly, reconnecting..."); - - this.session = null; - connectToServer(); - } - - @Override - public void onApplicationEvent(@Nonnull ApplicationStartedEvent event) { - executor.scheduleAtFixedRate(this::readHostInfo, 0, 30, TimeUnit.SECONDS); - log.info("Bootstrap successfully, connecting to server websocket endpoint..."); - executor.scheduleAtFixedRate(this::connectToServer, 0, 30, TimeUnit.SECONDS); - } - - private void readHostInfo() { - hostInfo = new HostInfo(); - - try { - InetAddress addr = InetAddress.getLocalHost(); - hostInfo.setHostname(addr.getHostName()); - hostInfo.setIpv4(addr.getHostAddress()); - - OperatingSystemMXBean osmxb = (OperatingSystemMXBean) ManagementFactory.getOperatingSystemMXBean(); - hostInfo.setOs(OSDetection.getOS()); - hostInfo.setVersion(OSDetection.getVersion()); - hostInfo.setArch(OSDetection.getArch()); - hostInfo.setAvailableProcessors(osmxb.getAvailableProcessors()); - hostInfo.setProcessCpuTime(osmxb.getProcessCpuTime()); - hostInfo.setTotalMemorySize(osmxb.getTotalMemorySize()); - hostInfo.setFreeMemorySize(osmxb.getFreeMemorySize()); - hostInfo.setTotalSwapSpaceSize(osmxb.getTotalSwapSpaceSize()); - hostInfo.setFreeSwapSpaceSize(osmxb.getFreeSwapSpaceSize()); - hostInfo.setCommittedVirtualMemorySize(osmxb.getCommittedVirtualMemorySize()); - - hostInfo.setCpuLoad(new BigDecimal(String.valueOf(osmxb.getCpuLoad()))); - hostInfo.setProcessCpuLoad(new BigDecimal(String.valueOf(osmxb.getProcessCpuLoad()))); - hostInfo.setSystemLoadAverage(new BigDecimal(String.valueOf(osmxb.getSystemLoadAverage()))); - - hostInfo.setFreeDisk(OSDetection.freeDisk()); - hostInfo.setTotalDisk(OSDetection.totalDisk()); - } catch (Exception e) { - log.error("Error getting host info", e); - throw new RuntimeException(e); - } - } - - private void connectToServer() { - String host = applicationConfig.getServer().getHost(); - Integer port = applicationConfig.getServer().getPort(); - String uri = MessageFormat.format("ws://{0}:{1,number,#}/ws/server", host, port); - StandardWebSocketClient webSocketClient = new StandardWebSocketClient(); - int retryTime = 0; - while (true) { - try { - if (retryTime >= 3) break; - AgentWebSocketHandler agentWebSocket = SpringContextHolder.getAgentWebSocket(); - WebSocketSession contextSession = agentWebSocket.getSession(); - if (null == contextSession || !contextSession.isOpen()) { - WebSocketSession webSocketSession = - webSocketClient.execute(this, uri).get(5, TimeUnit.SECONDS); - webSocketSession.setBinaryMessageSizeLimit(WS_BINARY_MESSAGE_SIZE_LIMIT); - agentWebSocket.setSession(webSocketSession); - log.info(MessageFormat.format("Connect to server: {0} successfully", uri)); - } - ++retryTime; - break; - } catch (Exception e) { - log.error(MessageFormat.format( - "Error connecting to server: {0}, retry time: {1}", e.getMessage(), ++retryTime)); - // retry after 5 seconds - try { - TimeUnit.MILLISECONDS.sleep(Constants.REGISTRY_SESSION_TIMEOUT); - } catch (InterruptedException ex) { - throw new RuntimeException(ex); - } - } - } - } -} diff --git a/bigtop-manager-agent/src/main/resources/application.yml b/bigtop-manager-agent/src/main/resources/application.yml index d2de797c..41356a20 100644 --- a/bigtop-manager-agent/src/main/resources/application.yml +++ b/bigtop-manager-agent/src/main/resources/application.yml @@ -19,9 +19,8 @@ bigtop: manager: - server: - host: localhost - port: 8080 + grpc: + port: 8835 spring: banner: @@ -31,6 +30,10 @@ spring: main: banner-mode: log +grpc: + server: + port: ${bigtop.manager.grpc.port} + server: port: 8081 diff --git a/bigtop-manager-bom/pom.xml b/bigtop-manager-bom/pom.xml index de710374..6db0c611 100644 --- a/bigtop-manager-bom/pom.xml +++ b/bigtop-manager-bom/pom.xml @@ -34,7 +34,6 @@ 3.1.1 2.2.0 2.3.32 - 5.4.0 3.12.0 2.0 2.13.0 @@ -48,7 +47,7 @@ 1.11.0 1.0.0 6.4.11 - 1.12.2 + 1.12.4 8.1.2.192 @@ -98,12 +97,6 @@ ${freemarker.version} - - com.esotericsoftware - kryo - ${kryo.version} - - org.apache.commons commons-lang3 @@ -196,6 +189,12 @@ ${micrometer.version} + + io.micrometer + micrometer-core + ${micrometer.version} + + com.github.oshi oshi-core @@ -216,6 +215,33 @@ ${jdbc.dm.version} provided + + + + io.grpc + grpc-stub + ${grpc.version} + + + io.grpc + grpc-protobuf + ${grpc.version} + + + com.google.protobuf + protobuf-java-util + ${protobuf-java-util.version} + + + net.devh + grpc-server-spring-boot-starter + ${grpc-spring-boot.version} + + + net.devh + grpc-client-spring-boot-starter + ${grpc-spring-boot.version} + diff --git a/bigtop-manager-common/pom.xml b/bigtop-manager-common/pom.xml index e759a656..ad71ab9d 100644 --- a/bigtop-manager-common/pom.xml +++ b/bigtop-manager-common/pom.xml @@ -54,22 +54,12 @@ true - - org.springframework - spring-websocket - - org.projectlombok lombok true - - com.esotericsoftware - kryo - - org.apache.commons commons-lang3 diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/ApplicationConfig.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/ApplicationConfig.java index 54b38bc2..80e1d9e8 100644 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/ApplicationConfig.java +++ b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/ApplicationConfig.java @@ -18,8 +18,7 @@ */ package org.apache.bigtop.manager.common.config; -import org.apache.bigtop.manager.common.config.application.SerializerConfig; -import org.apache.bigtop.manager.common.config.application.ServerConfig; +import org.apache.bigtop.manager.common.config.application.GrpcConfig; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.boot.context.properties.NestedConfigurationProperty; @@ -33,8 +32,5 @@ public class ApplicationConfig { @NestedConfigurationProperty - private ServerConfig server; - - @NestedConfigurationProperty - private SerializerConfig serializer = new SerializerConfig(); + private GrpcConfig grpc; } diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/MessageConfig.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/MessageConfig.java deleted file mode 100644 index f365e1e0..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/MessageConfig.java +++ /dev/null @@ -1,54 +0,0 @@ -/* - * 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.common.config; - -import org.apache.bigtop.manager.common.message.serializer.MessageDeserializer; -import org.apache.bigtop.manager.common.message.serializer.MessageSerializer; - -import org.apache.commons.lang3.StringUtils; - -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -import jakarta.annotation.Resource; - -@Configuration -public class MessageConfig { - - @Resource - private ApplicationConfig applicationConfig; - - @Bean - public MessageSerializer messageSerializer() throws Exception { - String serializerType = applicationConfig.getSerializer().getType(); - String packageName = "org.apache.bigtop.manager.common.message.serializer"; - String className = packageName + "." + StringUtils.capitalize(serializerType) + "MessageSerializer"; - return (MessageSerializer) - Class.forName(className).getDeclaredConstructor().newInstance(); - } - - @Bean - public MessageDeserializer messageDeserializer() throws Exception { - String deserializerType = applicationConfig.getSerializer().getType(); - String packageName = "org.apache.bigtop.manager.common.message.serializer"; - String className = packageName + "." + StringUtils.capitalize(deserializerType) + "MessageDeserializer"; - return (MessageDeserializer) - Class.forName(className).getDeclaredConstructor().newInstance(); - } -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/ServerConfig.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/GrpcConfig.java similarity index 94% rename from bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/ServerConfig.java rename to bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/GrpcConfig.java index d79afba7..4f6c774a 100644 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/ServerConfig.java +++ b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/GrpcConfig.java @@ -21,9 +21,7 @@ import lombok.Data; @Data -public class ServerConfig { - - private String host; +public class GrpcConfig { private Integer port; } diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/constants/Constants.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/constants/Constants.java index 556f2c81..66c12b96 100644 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/constants/Constants.java +++ b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/constants/Constants.java @@ -42,26 +42,6 @@ private Constants() { */ public static final String ALL_HOST_KEY = "all"; - /** - * registry session timeout - */ - public static final long REGISTRY_SESSION_TIMEOUT = 5 * 1000L; - - /** - * kryo buffer size - */ - public static final int KRYO_BUFFER_SIZE = 65536; - - /** - * websocket binary message size limit - */ - public static final int WS_BINARY_MESSAGE_SIZE_LIMIT = 65536; - - /** - * timeout for command message to wait for response - */ - public static final long COMMAND_MESSAGE_RESPONSE_TIMEOUT = 15 * 60 * 1000L; - /** * permission 755 */ diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseMessage.java deleted file mode 100644 index b32defad..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseMessage.java +++ /dev/null @@ -1,34 +0,0 @@ -/* - * 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.common.message.entity; - -import lombok.Data; -import lombok.NoArgsConstructor; - -import java.sql.Timestamp; -import java.util.UUID; - -@Data -@NoArgsConstructor -public class BaseMessage { - - private Timestamp timestamp = new Timestamp(System.currentTimeMillis()); - - private String messageId = UUID.randomUUID().toString(); -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseRequestMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseRequestMessage.java deleted file mode 100644 index 7d67e5b0..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseRequestMessage.java +++ /dev/null @@ -1,21 +0,0 @@ -/* - * 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.common.message.entity; - -public class BaseRequestMessage extends BaseMessage {} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseResponseMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseResponseMessage.java deleted file mode 100644 index 805525eb..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/BaseResponseMessage.java +++ /dev/null @@ -1,21 +0,0 @@ -/* - * 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.common.message.entity; - -public class BaseResponseMessage extends BaseMessage {} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/HeartbeatMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/HeartbeatMessage.java deleted file mode 100644 index b47754a2..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/HeartbeatMessage.java +++ /dev/null @@ -1,35 +0,0 @@ -/* - * 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.common.message.entity; - -import org.apache.bigtop.manager.common.message.entity.pojo.HostInfo; - -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.NoArgsConstructor; -import lombok.ToString; - -@Data -@EqualsAndHashCode(callSuper = true) -@ToString(callSuper = true) -@NoArgsConstructor -public class HeartbeatMessage extends BaseMessage { - - private HostInfo hostInfo; -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandMessageType.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandMessageType.java deleted file mode 100644 index cdabdad5..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandMessageType.java +++ /dev/null @@ -1,30 +0,0 @@ -/* - * 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.common.message.entity.command; - -public enum CommandMessageType { - COMPONENT, - - HOST_CHECK, - - CACHE_DISTRIBUTE, - - TASK_LOG, - ; -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandRequestMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandRequestMessage.java deleted file mode 100644 index 7d6925a3..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandRequestMessage.java +++ /dev/null @@ -1,43 +0,0 @@ -/* - * 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.common.message.entity.command; - -import org.apache.bigtop.manager.common.message.entity.BaseRequestMessage; - -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.ToString; - -@EqualsAndHashCode(callSuper = true) -@ToString(callSuper = true) -@Data -public class CommandRequestMessage extends BaseRequestMessage { - - private CommandMessageType commandMessageType; - - private String hostname; - - private String messagePayload; - - private Long jobId; - - private Long stageId; - - private Long taskId; -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandResponseMessage.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandResponseMessage.java deleted file mode 100644 index 76a51368..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/command/CommandResponseMessage.java +++ /dev/null @@ -1,49 +0,0 @@ -/* - * 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.common.message.entity.command; - -import org.apache.bigtop.manager.common.message.entity.BaseResponseMessage; - -import lombok.AllArgsConstructor; -import lombok.Data; -import lombok.EqualsAndHashCode; -import lombok.NoArgsConstructor; -import lombok.ToString; - -@Data -@EqualsAndHashCode(callSuper = true) -@ToString(callSuper = true) -@NoArgsConstructor -@AllArgsConstructor -public class CommandResponseMessage extends BaseResponseMessage { - - private Integer code; - - private String result; - - private String hostname; - - private CommandMessageType commandMessageType; - - private Long jobId; - - private Long stageId; - - private Long taskId; -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/pojo/HostInfo.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/pojo/HostInfo.java deleted file mode 100644 index 953092b9..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/entity/pojo/HostInfo.java +++ /dev/null @@ -1,64 +0,0 @@ -/* - * 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.common.message.entity.pojo; - -import lombok.Data; - -import java.io.Serializable; -import java.math.BigDecimal; - -@Data -public class HostInfo implements Serializable { - - private String hostname; - - private String ipv4; - - private String ipv6; - - private String os; - - private String version; - - private String arch; - - private BigDecimal cpuLoad; - - private Integer availableProcessors; - - private BigDecimal processCpuLoad; - - private Long processCpuTime; - - private Long totalMemorySize; - - private Long freeMemorySize; - - private Long totalSwapSpaceSize; - - private Long freeSwapSpaceSize; - - private Long committedVirtualMemorySize; - - private BigDecimal systemLoadAverage; - - private Long freeDisk; - - private Long totalDisk; -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageDeserializer.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageDeserializer.java deleted file mode 100644 index 2abb240f..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageDeserializer.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * 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.common.message.serializer; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Input; - -public class KryoMessageDeserializer implements MessageDeserializer { - - @Override - public BaseMessage deserialize(byte[] bytes) { - Input input = new Input(bytes); - Kryo kryo = KryoPoolHolder.obtainKryo(); - BaseMessage baseMessage = (BaseMessage) kryo.readClassAndObject(input); - input.close(); - KryoPoolHolder.freeKryo(kryo); - - return baseMessage; - } -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageSerializer.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageSerializer.java deleted file mode 100644 index 7bd8174b..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoMessageSerializer.java +++ /dev/null @@ -1,44 +0,0 @@ -/* - * 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.common.message.serializer; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.io.Output; - -import java.io.ByteArrayOutputStream; - -import static org.apache.bigtop.manager.common.constants.Constants.KRYO_BUFFER_SIZE; - -public class KryoMessageSerializer implements MessageSerializer { - - @Override - public byte[] serialize(BaseMessage message) { - ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); - Output output = new Output(outputStream, KRYO_BUFFER_SIZE); - Kryo kryo = KryoPoolHolder.obtainKryo(); - kryo.writeClassAndObject(output, message); - output.flush(); - output.close(); - KryoPoolHolder.freeKryo(kryo); - - return output.getBuffer(); - } -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoPoolHolder.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoPoolHolder.java deleted file mode 100644 index b690d379..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/KryoPoolHolder.java +++ /dev/null @@ -1,99 +0,0 @@ -/* - * 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.common.message.serializer; - -import org.apache.bigtop.manager.common.enums.Command; -import org.apache.bigtop.manager.common.message.entity.BaseMessage; -import org.apache.bigtop.manager.common.message.entity.BaseRequestMessage; -import org.apache.bigtop.manager.common.message.entity.BaseResponseMessage; -import org.apache.bigtop.manager.common.message.entity.HeartbeatMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandResponseMessage; -import org.apache.bigtop.manager.common.message.entity.pojo.ClusterInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.ComponentInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.CustomCommandInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.HostCheckType; -import org.apache.bigtop.manager.common.message.entity.pojo.HostInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.OSSpecificInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.RepoInfo; -import org.apache.bigtop.manager.common.message.entity.pojo.ScriptInfo; - -import com.esotericsoftware.kryo.Kryo; -import com.esotericsoftware.kryo.util.Pool; - -import java.math.BigDecimal; -import java.sql.Timestamp; -import java.util.ArrayList; -import java.util.HashMap; -import java.util.HashSet; -import java.util.LinkedHashMap; - -public class KryoPoolHolder { - - private static final Pool KRYO_POOL = new Pool<>(true, false, 16) { - - @Override - protected Kryo create() { - Kryo kryo = new Kryo(); - kryo.setCopyReferences(true); - - // message types - kryo.register(BaseMessage.class); - kryo.register(BaseRequestMessage.class); - kryo.register(BaseResponseMessage.class); - kryo.register(HeartbeatMessage.class); - kryo.register(CommandResponseMessage.class); - kryo.register(CommandRequestMessage.class); - - // message pojo - kryo.register(HostInfo.class); - kryo.register(OSSpecificInfo.class); - kryo.register(ClusterInfo.class); - kryo.register(RepoInfo.class); - kryo.register(HostCheckType.class); - kryo.register(HostCheckType[].class); - kryo.register(CommandMessageType.class); - kryo.register(Command.class); - kryo.register(ScriptInfo.class); - kryo.register(ComponentInfo.class); - kryo.register(CustomCommandInfo.class); - - // java classes - kryo.register(BigDecimal.class); - kryo.register(Timestamp.class); - kryo.register(ArrayList.class); - kryo.register(Integer.class); - kryo.register(String.class); - kryo.register(HashMap.class); - kryo.register(LinkedHashMap.class); - kryo.register(HashSet.class); - - return kryo; - } - }; - - public static Kryo obtainKryo() { - return KRYO_POOL.obtain(); - } - - public static void freeKryo(Kryo kryo) { - KRYO_POOL.free(kryo); - } -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageDeserializer.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageDeserializer.java deleted file mode 100644 index ef5d8b74..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageDeserializer.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * 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.common.message.serializer; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; - -public interface MessageDeserializer { - - BaseMessage deserialize(byte[] bytes); -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageSerializer.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageSerializer.java deleted file mode 100644 index 5d0a6eb6..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/message/serializer/MessageSerializer.java +++ /dev/null @@ -1,26 +0,0 @@ -/* - * 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.common.message.serializer; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; - -public interface MessageSerializer { - - byte[] serialize(BaseMessage message); -} diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/ws/AbstractBinaryWebSocketHandler.java b/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/ws/AbstractBinaryWebSocketHandler.java deleted file mode 100644 index 0d4b8053..00000000 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/ws/AbstractBinaryWebSocketHandler.java +++ /dev/null @@ -1,111 +0,0 @@ -/* - * 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.common.ws; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; -import org.apache.bigtop.manager.common.message.entity.BaseRequestMessage; -import org.apache.bigtop.manager.common.message.entity.BaseResponseMessage; -import org.apache.bigtop.manager.common.message.serializer.MessageDeserializer; -import org.apache.bigtop.manager.common.message.serializer.MessageSerializer; - -import org.springframework.web.socket.BinaryMessage; -import org.springframework.web.socket.WebSocketSession; -import org.springframework.web.socket.handler.BinaryWebSocketHandler; - -import lombok.extern.slf4j.Slf4j; - -import jakarta.annotation.Resource; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; - -@Slf4j -public class AbstractBinaryWebSocketHandler extends BinaryWebSocketHandler { - - @Resource - protected MessageSerializer serializer; - - @Resource - protected MessageDeserializer deserializer; - - private final ConcurrentHashMap requests = new ConcurrentHashMap<>(); - - private final ConcurrentHashMap responses = new ConcurrentHashMap<>(); - - protected void sendMessage(WebSocketSession session, BaseMessage message) { - try { - sendMessageWithRetry(session, message); - } catch (Exception e) { - log.error("Error sending message: {}", message, e); - } - } - - protected BaseResponseMessage sendRequestMessage(WebSocketSession session, BaseRequestMessage request) { - requests.put(request.getMessageId(), request); - - try { - sendMessageWithRetry(session, request); - } catch (Exception e) { - log.error("Error sending message: {}", request, e); - } - - for (int i = 0; i <= 300; i++) { - BaseResponseMessage response = responses.get(request.getMessageId()); - if (response == null) { - try { - TimeUnit.SECONDS.sleep(5); - } catch (InterruptedException e) { - log.error("Error waiting for message response, messageId: {}", request.getMessageId(), e); - } - } else { - requests.remove(request.getMessageId()); - responses.remove(request.getMessageId()); - return response; - } - } - - requests.remove(request.getMessageId()); - return null; - } - - protected void handleResponseMessage(BaseResponseMessage response) { - if (requests.containsKey(response.getMessageId())) { - responses.put(response.getMessageId(), response); - } else { - log.warn("Message is timed out or unexpected, drop it: {}", response); - } - } - - private void sendMessageWithRetry(WebSocketSession session, BaseMessage message) throws Exception { - int retryCount = 3; - int retryInterval = 5000; - for (int i = 0; i < retryCount; i++) { - try { - session.sendMessage(new BinaryMessage(serializer.serialize(message))); - break; - } catch (Exception e) { - log.error("Error sending message: {}, retry count: {}", message, i + 1, e); - if (i + 1 == retryCount) { - throw e; - } else { - Thread.sleep(retryInterval); - } - } - } - } -} diff --git a/bigtop-manager-dao/pom.xml b/bigtop-manager-dao/pom.xml index f26dc14c..8bf7a516 100644 --- a/bigtop-manager-dao/pom.xml +++ b/bigtop-manager-dao/pom.xml @@ -43,6 +43,11 @@ + + org.apache.bigtop + bigtop-manager-common + + org.springframework.data spring-data-jpa @@ -64,10 +69,6 @@ org.springframework spring-web - - org.apache.bigtop - bigtop-manager-common - org.apache.tomcat.embed tomcat-embed-core diff --git a/bigtop-manager-grpc/pom.xml b/bigtop-manager-grpc/pom.xml new file mode 100644 index 00000000..82c8b1de --- /dev/null +++ b/bigtop-manager-grpc/pom.xml @@ -0,0 +1,89 @@ + + + + 4.0.0 + + org.apache.bigtop + bigtop-manager + ${revision} + ../pom.xml + + + bigtop-manager-grpc + ${project.artifactId} + Bigtop Manager gRPC + + + + + org.apache.bigtop + bigtop-manager-bom + ${project.version} + pom + import + + + + + + + io.grpc + grpc-stub + + + io.grpc + grpc-protobuf + + + com.google.protobuf + protobuf-java-util + + + + jakarta.annotation + jakarta.annotation-api + 1.3.5 + true + + + + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + + ${project.basedir}/src/main/resources/proto + com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier} + grpc-java + io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier} + + + + + compile + compile-custom + + + + + + + diff --git a/bigtop-manager-grpc/src/main/java/org/apache/bigtop/manager/grpc/utils/ProtobufUtil.java b/bigtop-manager-grpc/src/main/java/org/apache/bigtop/manager/grpc/utils/ProtobufUtil.java new file mode 100644 index 00000000..dc212e9b --- /dev/null +++ b/bigtop-manager-grpc/src/main/java/org/apache/bigtop/manager/grpc/utils/ProtobufUtil.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 + * + * 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.grpc.utils; + +import com.google.protobuf.Message; +import com.google.protobuf.MessageOrBuilder; +import com.google.protobuf.util.JsonFormat; + +public class ProtobufUtil { + + @SuppressWarnings("unchecked") + public static T fromJson(String json, Class clazz) { + try { + Message.Builder builder = + (Message.Builder) clazz.getMethod("newBuilder").invoke(null); + JsonFormat.parser().ignoringUnknownFields().merge(json, builder); + return (T) builder.build(); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + + public static String toJson(T message) { + try { + return JsonFormat.printer().print(message); + } catch (Exception e) { + throw new RuntimeException(e); + } + } +} diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/enums/AgentExceptionStatus.java b/bigtop-manager-grpc/src/main/resources/proto/command.proto similarity index 54% rename from bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/enums/AgentExceptionStatus.java rename to bigtop-manager-grpc/src/main/resources/proto/command.proto index 98727481..c86c571a 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/enums/AgentExceptionStatus.java +++ b/bigtop-manager-grpc/src/main/resources/proto/command.proto @@ -16,23 +16,38 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.bigtop.manager.agent.enums; +syntax = "proto3"; -import lombok.Getter; +option java_multiple_files = true; +option java_package = "org.apache.bigtop.manager.grpc.generated"; +option java_outer_classname = "CommandProto"; -@Getter -public enum AgentExceptionStatus { - AGENT_MONITORING_ERROR(10001, "Get agent host monitoring info failed"), - - COMMAND_FAILED(10002, "Run command failed"), - ; - - private final Integer code; +service CommandService { + rpc Exec (CommandRequest) returns (CommandReply) {} +} - private final String message; +enum CommandType { + COMPONENT = 0; + HOST_CHECK = 1; + CACHE_DISTRIBUTE = 2; + TASK_LOG = 3; +} - AgentExceptionStatus(Integer code, String message) { - this.code = code; - this.message = message; - } +message CommandRequest { + string payload = 1; + string hostname = 2; + int64 job_id = 3; + int64 stage_id = 4; + int64 task_id = 5; + CommandType type = 6; } + +message CommandReply { + int32 code = 1; + string result = 2; + string hostname = 3; + int64 job_id = 4; + int64 stage_id = 5; + int64 task_id = 6; + CommandType type = 7; +} \ No newline at end of file diff --git a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/exception/AgentException.java b/bigtop-manager-grpc/src/main/resources/proto/host_info.proto similarity index 50% rename from bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/exception/AgentException.java rename to bigtop-manager-grpc/src/main/resources/proto/host_info.proto index 9f321b7b..51c6ce9b 100644 --- a/bigtop-manager-agent/src/main/java/org/apache/bigtop/manager/agent/exception/AgentException.java +++ b/bigtop-manager-grpc/src/main/resources/proto/host_info.proto @@ -16,20 +16,36 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.bigtop.manager.agent.exception; +syntax = "proto3"; -import org.apache.bigtop.manager.agent.enums.AgentExceptionStatus; +option java_multiple_files = true; +option java_package = "org.apache.bigtop.manager.grpc.generated"; +option java_outer_classname = "HostInfoProto"; -import lombok.Data; -import lombok.EqualsAndHashCode; - -@Data -@EqualsAndHashCode(callSuper = false) -public class AgentException extends RuntimeException { - - private AgentExceptionStatus ex; +service HostInfoService { + rpc GetHostInfo (HostInfoRequest) returns (HostInfoReply) {} +} - public AgentException(AgentExceptionStatus ex) { - this.ex = ex; - } +message HostInfoRequest { } + +message HostInfoReply { + string hostname = 1; + string ipv4 = 2; + string ipv6 = 3; + string os = 4; + string version = 5; + string arch = 6; + string cpu_load = 7; + int32 available_processors = 8; + string process_cpu_load = 9; + int64 process_cpu_time = 10; + int64 total_memory_size = 11; + int64 free_memory_size = 12; + int64 total_swap_space_size = 13; + int64 free_swap_space_size = 14; + int64 committed_virtual_memory_size = 15; + string system_load_average = 16; + int64 free_disk = 17; + int64 total_disk = 18; +} \ No newline at end of file diff --git a/bigtop-manager-server/pom.xml b/bigtop-manager-server/pom.xml index 0a16c6de..acbe9ffb 100644 --- a/bigtop-manager-server/pom.xml +++ b/bigtop-manager-server/pom.xml @@ -47,12 +47,14 @@ org.apache.bigtop bigtop-manager-common + + org.apache.bigtop + bigtop-manager-grpc + org.apache.bigtop bigtop-manager-dao - ${project.version} - org.apache.bigtop bigtop-manager-ui @@ -89,11 +91,6 @@ spring-boot-starter-web - - org.springframework.boot - spring-boot-starter-websocket - - org.springframework.boot spring-boot-starter-webflux @@ -172,6 +169,11 @@ org.glassfish.jaxb jaxb-runtime + + + net.devh + grpc-client-spring-boot-starter + diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ServerApplication.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ServerApplication.java index 44ed5b71..5eafb01a 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ServerApplication.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ServerApplication.java @@ -26,8 +26,8 @@ import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.scheduling.annotation.EnableScheduling; -@EnableScheduling @EnableAsync +@EnableScheduling @EnableJpaAuditing @EntityScan("org.apache.bigtop.manager.dao") @EnableJpaRepositories("org.apache.bigtop.manager.dao") diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/validator/ClusterHostValidator.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/validator/ClusterHostValidator.java index de434ea2..586d506b 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/validator/ClusterHostValidator.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/validator/ClusterHostValidator.java @@ -25,7 +25,7 @@ import org.apache.bigtop.manager.server.enums.ApiExceptionEnum; import org.apache.bigtop.manager.server.enums.CommandLevel; import org.apache.bigtop.manager.server.exception.ApiException; -import org.apache.bigtop.manager.server.holder.SpringContextHolder; +import org.apache.bigtop.manager.server.grpc.GrpcClient; import org.apache.bigtop.manager.server.model.dto.command.ClusterCommandDTO; import org.apache.commons.collections4.CollectionUtils; @@ -51,20 +51,16 @@ public void validate(ValidatorContext context) { ClusterCommandDTO clusterCommand = context.getCommandDTO().getClusterCommand(); List hostnames = clusterCommand.getHostnames(); - List connectedHosts = SpringContextHolder.getServerWebSocket().getConnectedHosts(); - if (CollectionUtils.isNotEmpty(connectedHosts)) { - List notConnectedHostnames = hostnames.stream() - .filter(hostname -> !connectedHosts.contains(hostname)) - .toList(); - if (CollectionUtils.isNotEmpty(notConnectedHostnames)) { - throw new ApiException(ApiExceptionEnum.HOST_NOT_CONNECTED, String.join(",", notConnectedHostnames)); - } - } - List hosts = hostRepository.findAllByHostnameIn(hostnames); if (CollectionUtils.isNotEmpty(hosts)) { List existsHostnames = hosts.stream().map(Host::getHostname).toList(); throw new ApiException(ApiExceptionEnum.HOST_ASSIGNED, String.join(",", existsHostnames)); } + + for (String hostname : hostnames) { + if (!GrpcClient.isChannelAlive(hostname)) { + GrpcClient.createChannel(hostname); + } + } } } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/component/AbstractComponentStageFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/component/AbstractComponentStageFactory.java index d685be8a..508f68b0 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/component/AbstractComponentStageFactory.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/component/AbstractComponentStageFactory.java @@ -19,8 +19,6 @@ package org.apache.bigtop.manager.server.command.stage.factory.component; import org.apache.bigtop.manager.common.enums.Command; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; import org.apache.bigtop.manager.common.message.entity.payload.CommandPayload; import org.apache.bigtop.manager.common.message.entity.pojo.CustomCommandInfo; import org.apache.bigtop.manager.common.message.entity.pojo.OSSpecificInfo; @@ -29,6 +27,9 @@ import org.apache.bigtop.manager.dao.entity.Cluster; import org.apache.bigtop.manager.dao.entity.Task; import org.apache.bigtop.manager.dao.repository.ClusterRepository; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandType; +import org.apache.bigtop.manager.grpc.utils.ProtobufUtil; import org.apache.bigtop.manager.server.command.stage.factory.AbstractStageFactory; import org.apache.bigtop.manager.server.model.dto.ComponentDTO; import org.apache.bigtop.manager.server.model.dto.CustomCommandDTO; @@ -78,10 +79,8 @@ public void doCreateStage() { task.setCustomCommands(JsonUtils.writeAsString(componentDTO.getCustomCommands())); task.setCommandScript(JsonUtils.writeAsString(componentDTO.getCommandScript())); - CommandRequestMessage commandRequestMessage = getMessage(serviceDTO, componentDTO, hostname, command); - task.setContent(JsonUtils.writeAsString(commandRequestMessage)); - task.setMessageId(commandRequestMessage.getMessageId()); - + CommandRequest request = getMessage(serviceDTO, componentDTO, hostname, command); + task.setContent(ProtobufUtil.toJson(request)); tasks.add(task); } @@ -90,7 +89,7 @@ public void doCreateStage() { protected abstract Command getCommand(); - private CommandRequestMessage getMessage( + private CommandRequest getMessage( ServiceDTO serviceDTO, ComponentDTO componentDTO, String hostname, Command command) { CommandPayload commandPayload = new CommandPayload(); commandPayload.setServiceName(serviceDTO.getServiceName()); @@ -107,12 +106,12 @@ private CommandRequestMessage getMessage( commandPayload.setOsSpecifics(convertOSSpecificInfo(serviceDTO.getOsSpecifics())); commandPayload.setCommandScript(convertScriptInfo(componentDTO.getCommandScript())); - CommandRequestMessage commandRequestMessage = new CommandRequestMessage(); - commandRequestMessage.setCommandMessageType(CommandMessageType.COMPONENT); - commandRequestMessage.setHostname(hostname); - commandRequestMessage.setMessagePayload(JsonUtils.writeAsString(commandPayload)); + CommandRequest.Builder builder = CommandRequest.newBuilder(); + builder.setType(CommandType.COMPONENT); + builder.setHostname(hostname); + builder.setPayload(JsonUtils.writeAsString(commandPayload)); - return commandRequestMessage; + return builder.build(); } private ScriptInfo convertScriptInfo(ScriptDTO scriptDTO) { diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/host/HostCheckStageFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/host/HostCheckStageFactory.java index c4238d23..895ed7a2 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/host/HostCheckStageFactory.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/factory/host/HostCheckStageFactory.java @@ -19,13 +19,14 @@ package org.apache.bigtop.manager.server.command.stage.factory.host; import org.apache.bigtop.manager.common.enums.Command; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; import org.apache.bigtop.manager.common.message.entity.payload.HostCheckPayload; import org.apache.bigtop.manager.common.utils.JsonUtils; import org.apache.bigtop.manager.dao.entity.Cluster; import org.apache.bigtop.manager.dao.entity.Task; import org.apache.bigtop.manager.dao.repository.ClusterRepository; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandType; +import org.apache.bigtop.manager.grpc.utils.ProtobufUtil; import org.apache.bigtop.manager.server.command.stage.factory.AbstractStageFactory; import org.apache.bigtop.manager.server.command.stage.factory.StageType; @@ -76,9 +77,8 @@ public void doCreateStage() { task.setCommand(Command.CUSTOM); task.setCustomCommand("check_host"); - CommandRequestMessage commandRequestMessage = createMessage(hostname); - task.setContent(JsonUtils.writeAsString(commandRequestMessage)); - task.setMessageId(commandRequestMessage.getMessageId()); + CommandRequest request = createMessage(hostname); + task.setContent(ProtobufUtil.toJson(request)); tasks.add(task); } @@ -86,15 +86,15 @@ public void doCreateStage() { stage.setTasks(tasks); } - private CommandRequestMessage createMessage(String hostname) { + private CommandRequest createMessage(String hostname) { HostCheckPayload messagePayload = new HostCheckPayload(); messagePayload.setHostname(hostname); - CommandRequestMessage commandRequestMessage = new CommandRequestMessage(); - commandRequestMessage.setCommandMessageType(CommandMessageType.HOST_CHECK); - commandRequestMessage.setHostname(hostname); - commandRequestMessage.setMessagePayload(JsonUtils.writeAsString(messagePayload)); + CommandRequest.Builder builder = CommandRequest.newBuilder(); + builder.setType(CommandType.HOST_CHECK); + builder.setHostname(hostname); + builder.setPayload(JsonUtils.writeAsString(messagePayload)); - return commandRequestMessage; + return builder.build(); } } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/AbstractStageRunner.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/AbstractStageRunner.java index d7c085bf..f508fddd 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/AbstractStageRunner.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/AbstractStageRunner.java @@ -20,15 +20,16 @@ import org.apache.bigtop.manager.common.constants.MessageConstants; import org.apache.bigtop.manager.common.enums.JobState; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandResponseMessage; -import org.apache.bigtop.manager.common.utils.JsonUtils; import org.apache.bigtop.manager.dao.entity.Stage; import org.apache.bigtop.manager.dao.entity.Task; import org.apache.bigtop.manager.dao.repository.StageRepository; import org.apache.bigtop.manager.dao.repository.TaskRepository; +import org.apache.bigtop.manager.grpc.generated.CommandReply; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandServiceGrpc; +import org.apache.bigtop.manager.grpc.utils.ProtobufUtil; import org.apache.bigtop.manager.server.command.stage.factory.StageContext; -import org.apache.bigtop.manager.server.holder.SpringContextHolder; +import org.apache.bigtop.manager.server.grpc.GrpcClient; import org.apache.bigtop.manager.server.service.CommandLogService; import lombok.extern.slf4j.Slf4j; @@ -37,9 +38,6 @@ import java.util.ArrayList; import java.util.List; import java.util.concurrent.CompletableFuture; -import java.util.concurrent.TimeUnit; - -import static org.apache.bigtop.manager.common.constants.Constants.COMMAND_MESSAGE_RESPONSE_TIMEOUT; @Slf4j public abstract class AbstractStageRunner implements StageRunner { @@ -75,18 +73,21 @@ public void run() { for (Task task : stage.getTasks()) { beforeRunTask(task); - CommandRequestMessage message = JsonUtils.readFromString(task.getContent(), CommandRequestMessage.class); - message.setTaskId(task.getId()); - message.setStageId(stage.getId()); - message.setJobId(stage.getJob().getId()); + CommandRequest protoRequest = ProtobufUtil.fromJson(task.getContent(), CommandRequest.class); + CommandRequest.Builder builder = CommandRequest.newBuilder(protoRequest); + builder.setTaskId(task.getId()); + builder.setStageId(stage.getId()); + builder.setJobId(stage.getJob().getId()); + CommandRequest request = builder.build(); futures.add(CompletableFuture.supplyAsync(() -> { commandLogService.onLogStarted(task.getId(), task.getHostname()); - CommandResponseMessage res = - SpringContextHolder.getServerWebSocket().sendRequestMessage(task.getHostname(), message); + CommandServiceGrpc.CommandServiceBlockingStub stub = GrpcClient.getBlockingStub( + task.getHostname(), CommandServiceGrpc.CommandServiceBlockingStub.class); + CommandReply reply = stub.exec(request); - log.info("Execute task {} completed: {}", task.getId(), res); - boolean taskSuccess = res != null && res.getCode() == MessageConstants.SUCCESS_CODE; + log.info("Execute task {} completed: {}", task.getId(), reply); + boolean taskSuccess = reply != null && reply.getCode() == MessageConstants.SUCCESS_CODE; if (taskSuccess) { commandLogService.onLogReceived(task.getId(), task.getHostname(), "Success!"); @@ -104,7 +105,7 @@ public void run() { List taskResults = futures.stream() .map((future) -> { try { - return future.get(COMMAND_MESSAGE_RESPONSE_TIMEOUT, TimeUnit.MILLISECONDS); + return future.get(); } catch (Exception e) { log.error("Error running task", e); return false; diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/config/CacheDistributeStageRunner.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/config/CacheDistributeStageRunner.java index 6ccae8c2..50cfeb8e 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/config/CacheDistributeStageRunner.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/stage/runner/config/CacheDistributeStageRunner.java @@ -19,8 +19,6 @@ package org.apache.bigtop.manager.server.command.stage.runner.config; import org.apache.bigtop.manager.common.constants.Constants; -import org.apache.bigtop.manager.common.message.entity.command.CommandMessageType; -import org.apache.bigtop.manager.common.message.entity.command.CommandRequestMessage; import org.apache.bigtop.manager.common.message.entity.payload.CacheMessagePayload; import org.apache.bigtop.manager.common.message.entity.pojo.ClusterInfo; import org.apache.bigtop.manager.common.message.entity.pojo.ComponentInfo; @@ -44,6 +42,9 @@ import org.apache.bigtop.manager.dao.repository.ServiceConfigRepository; import org.apache.bigtop.manager.dao.repository.ServiceRepository; import org.apache.bigtop.manager.dao.repository.SettingRepository; +import org.apache.bigtop.manager.grpc.generated.CommandRequest; +import org.apache.bigtop.manager.grpc.generated.CommandType; +import org.apache.bigtop.manager.grpc.utils.ProtobufUtil; import org.apache.bigtop.manager.server.command.stage.factory.StageType; import org.apache.bigtop.manager.server.command.stage.runner.AbstractStageRunner; import org.apache.bigtop.manager.server.model.dto.PropertyDTO; @@ -135,9 +136,8 @@ private void updateTask(Task task) { genCaches(); } - CommandRequestMessage commandRequestMessage = getMessage(task.getHostname()); - task.setContent(JsonUtils.writeAsString(commandRequestMessage)); - task.setMessageId(commandRequestMessage.getMessageId()); + CommandRequest request = getMessage(task.getHostname()); + task.setContent(ProtobufUtil.toJson(request)); taskRepository.save(task); } @@ -256,7 +256,7 @@ private void genEmptyCaches() { } } - private CommandRequestMessage getMessage(String hostname) { + private CommandRequest getMessage(String hostname) { CacheMessagePayload messagePayload = new CacheMessagePayload(); messagePayload.setHostname(hostname); messagePayload.setClusterInfo(clusterInfo); @@ -267,11 +267,11 @@ private CommandRequestMessage getMessage(String hostname) { messagePayload.setUserInfo(userMap); messagePayload.setComponentInfo(componentInfoMap); - CommandRequestMessage commandRequestMessage = new CommandRequestMessage(); - commandRequestMessage.setCommandMessageType(CommandMessageType.CACHE_DISTRIBUTE); - commandRequestMessage.setHostname(hostname); - commandRequestMessage.setMessagePayload(JsonUtils.writeAsString(messagePayload)); + CommandRequest.Builder builder = CommandRequest.newBuilder(); + builder.setType(CommandType.CACHE_DISTRIBUTE); + builder.setHostname(hostname); + builder.setPayload(JsonUtils.writeAsString(messagePayload)); - return commandRequestMessage; + return builder.build(); } } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/config/WebSocketConfig.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/config/WebSocketConfig.java deleted file mode 100644 index d69839e1..00000000 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/config/WebSocketConfig.java +++ /dev/null @@ -1,56 +0,0 @@ -/* - * 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.server.config; - -import org.apache.bigtop.manager.server.ws.DefaultWebSocketHandler; -import org.apache.bigtop.manager.server.ws.ServerWebSocketHandler; - -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.web.socket.config.annotation.EnableWebSocket; -import org.springframework.web.socket.config.annotation.WebSocketConfigurer; -import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; -import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean; - -import jakarta.annotation.Resource; - -import static org.apache.bigtop.manager.common.constants.Constants.WS_BINARY_MESSAGE_SIZE_LIMIT; - -@Configuration -@EnableWebSocket -public class WebSocketConfig implements WebSocketConfigurer { - - @Resource - private DefaultWebSocketHandler defaultWebSocketHandler; - - @Resource - private ServerWebSocketHandler serverWebSocketHandler; - - public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { - registry.addHandler(defaultWebSocketHandler, "/ws/default"); - registry.addHandler(serverWebSocketHandler, "/ws/server"); - } - - @Bean - public ServletServerContainerFactoryBean createWebSocketContainer() { - ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean(); - container.setMaxBinaryMessageBufferSize(WS_BINARY_MESSAGE_SIZE_LIMIT); - return container; - } -} diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/controller/HostController.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/controller/HostController.java index c2eb2ad8..4babe181 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/controller/HostController.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/controller/HostController.java @@ -21,6 +21,7 @@ import org.apache.bigtop.manager.server.model.dto.HostDTO; import org.apache.bigtop.manager.server.model.mapper.HostMapper; import org.apache.bigtop.manager.server.model.req.HostReq; +import org.apache.bigtop.manager.server.model.req.HostnamesReq; import org.apache.bigtop.manager.server.model.vo.HostVO; import org.apache.bigtop.manager.server.service.HostService; import org.apache.bigtop.manager.server.utils.ResponseEntity; @@ -28,6 +29,7 @@ import org.springframework.validation.annotation.Validated; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @@ -70,4 +72,11 @@ public ResponseEntity update(@PathVariable Long id, @RequestBody @Valida public ResponseEntity delete(@PathVariable Long id) { return ResponseEntity.success(hostService.delete(id)); } + + @Operation(summary = "Check connection", description = "Check connection for hosts") + @PostMapping("/check-connection") + public ResponseEntity checkConnection( + @PathVariable Long clusterId, @RequestBody @Validated HostnamesReq hostnamesReq) { + return ResponseEntity.success(hostService.checkConnection(hostnamesReq.getHostnames())); + } } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/ApiExceptionEnum.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/ApiExceptionEnum.java index e52e1ec1..97c8766a 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/ApiExceptionEnum.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/ApiExceptionEnum.java @@ -37,7 +37,8 @@ public enum ApiExceptionEnum { // Host Exceptions -- 12000 ~ 12999 HOST_NOT_FOUND(12000, LocaleKeys.HOST_NOT_FOUND), HOST_ASSIGNED(12001, LocaleKeys.HOST_ASSIGNED), - HOST_NOT_CONNECTED(12002, LocaleKeys.HOST_NOT_FOUND), + HOST_NOT_CONNECTED(12002, LocaleKeys.HOST_NOT_CONNECTED), + HOST_UNABLE_TO_CONNECT(12003, LocaleKeys.HOST_UNABLE_TO_CONNECT), // Stack Exceptions -- 13000 ~ 13999 STACK_NOT_FOUND(13000, LocaleKeys.STACK_NOT_FOUND), diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/LocaleKeys.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/LocaleKeys.java index 0a24e608..2a9ad0f0 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/LocaleKeys.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/enums/LocaleKeys.java @@ -40,6 +40,7 @@ public enum LocaleKeys { HOST_NOT_FOUND("host.not.found"), HOST_ASSIGNED("host.assigned"), HOST_NOT_CONNECTED("host.not.connected"), + HOST_UNABLE_TO_CONNECT("host.unable.to.connect"), STACK_NOT_FOUND("stack.not.found"), diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/grpc/GrpcClient.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/grpc/GrpcClient.java new file mode 100644 index 00000000..61422456 --- /dev/null +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/grpc/GrpcClient.java @@ -0,0 +1,140 @@ +/* + * 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.server.grpc; + +import org.apache.bigtop.manager.server.enums.ApiExceptionEnum; +import org.apache.bigtop.manager.server.exception.ApiException; +import org.apache.bigtop.manager.server.holder.SpringContextHolder; + +import io.grpc.CallOptions; +import io.grpc.Channel; +import io.grpc.ConnectivityState; +import io.grpc.ManagedChannel; +import io.grpc.ManagedChannelBuilder; +import io.grpc.stub.AbstractAsyncStub; +import io.grpc.stub.AbstractBlockingStub; +import io.grpc.stub.AbstractFutureStub; +import io.grpc.stub.AbstractStub; +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; + +@Slf4j +public class GrpcClient { + + private static final Map CHANNELS = new ConcurrentHashMap<>(); + + // The key of outer map is hostname, inner map is stub class name + private static final Map>> BLOCKING_STUBS = new ConcurrentHashMap<>(); + private static final Map>> ASYNC_STUBS = new ConcurrentHashMap<>(); + private static final Map>> FUTURE_STUBS = new ConcurrentHashMap<>(); + + public static ManagedChannel createChannel(String host) { + int port = SpringContextHolder.getApplicationContext() + .getEnvironment() + .getRequiredProperty("bigtop.manager.grpc.port", Integer.class); + return createChannel(host, port); + } + + public static Boolean isChannelAlive(String host) { + ManagedChannel channel = CHANNELS.get(host); + return channel != null && !channel.isShutdown() && !channel.isTerminated(); + } + + @SuppressWarnings("unchecked") + public static > T getBlockingStub(String host, Class clazz) { + Map> innerMap = BLOCKING_STUBS.computeIfAbsent(host, k -> new HashMap<>()); + return (T) innerMap.computeIfAbsent(clazz.getName(), k -> { + T instance = T.newStub(getFactory(clazz), getChannel(host)); + log.info("Instance: {} created.", k); + return instance; + }); + } + + @SuppressWarnings("unchecked") + public static > T getAsyncStub(String host, Class clazz) { + Map> innerMap = ASYNC_STUBS.computeIfAbsent(host, k -> new HashMap<>()); + return (T) innerMap.computeIfAbsent(clazz.getName(), k -> { + T instance = T.newStub(getFactory(clazz), getChannel(host)); + log.info("Instance: {} created.", k); + return instance; + }); + } + + @SuppressWarnings("unchecked") + public static > T getFutureStub(String host, Class clazz) { + Map> innerMap = FUTURE_STUBS.computeIfAbsent(host, k -> new HashMap<>()); + return (T) innerMap.computeIfAbsent(clazz.getName(), k -> { + T instance = T.newStub(getFactory(clazz), getChannel(host)); + log.info("Instance: {} created.", k); + return instance; + }); + } + + private static ManagedChannel createChannel(String host, Integer port) { + ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port) + .usePlaintext() + .keepAliveTime(60, TimeUnit.SECONDS) + .keepAliveWithoutCalls(true) + .build(); + + ConnectivityState state = channel.getState(true); + while (state == ConnectivityState.IDLE || state == ConnectivityState.CONNECTING) { + try { + Thread.sleep(1000); + state = channel.getState(true); + } catch (Exception e) { + log.warn("Error ignored when creating channel", e); + } + } + + if (state != ConnectivityState.READY) { + channel.shutdown(); + log.error("Unable to connect to host: {}", host); + throw new ApiException(ApiExceptionEnum.HOST_UNABLE_TO_CONNECT, host); + } else { + CHANNELS.put(host, channel); + return channel; + } + } + + private static ManagedChannel getChannel(String host) { + if (isChannelAlive(host)) { + return CHANNELS.get(host); + } else { + throw new ApiException(ApiExceptionEnum.HOST_NOT_CONNECTED, host); + } + } + + private static > AbstractStub.StubFactory getFactory(Class clazz) { + return (channel, callOptions) -> { + try { + Constructor constructor = clazz.getDeclaredConstructor(Channel.class, CallOptions.class); + constructor.setAccessible(true); + return constructor.newInstance(channel, callOptions); + } catch (Exception e) { + throw new RuntimeException(e); + } + }; + } +} diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/holder/SpringContextHolder.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/holder/SpringContextHolder.java index c3c6e216..4143720d 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/holder/SpringContextHolder.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/holder/SpringContextHolder.java @@ -23,7 +23,6 @@ import org.apache.bigtop.manager.server.command.job.validator.CommandValidator; import org.apache.bigtop.manager.server.command.stage.factory.StageFactory; import org.apache.bigtop.manager.server.command.stage.runner.StageRunner; -import org.apache.bigtop.manager.server.ws.ServerWebSocketHandler; import org.springframework.context.ApplicationContext; import org.springframework.context.ApplicationContextAware; @@ -45,10 +44,6 @@ public void setApplicationContext(@Nonnull ApplicationContext applicationContext SpringContextHolder.applicationContext = applicationContext; } - public static ServerWebSocketHandler getServerWebSocket() { - return applicationContext.getBean(ServerWebSocketHandler.class); - } - public static Map getCommandValidators() { return applicationContext.getBeansOfType(CommandValidator.class); } diff --git a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/SerializerConfig.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/model/req/HostnamesReq.java similarity index 74% rename from bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/SerializerConfig.java rename to bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/model/req/HostnamesReq.java index 4f476b33..487cd312 100644 --- a/bigtop-manager-common/src/main/java/org/apache/bigtop/manager/common/config/application/SerializerConfig.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/model/req/HostnamesReq.java @@ -16,12 +16,18 @@ * specific language governing permissions and limitations * under the License. */ -package org.apache.bigtop.manager.common.config.application; +package org.apache.bigtop.manager.server.model.req; +import io.swagger.v3.oas.annotations.media.Schema; import lombok.Data; +import jakarta.validation.constraints.NotEmpty; +import java.util.List; + @Data -public class SerializerConfig { +public class HostnamesReq { - private String type = "kryo"; + @NotEmpty + @Schema(example = "[host1, host2]") + private List hostnames; } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduled/HostHeartbeatScheduled.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduled/HostHeartbeatScheduled.java deleted file mode 100644 index 917b515b..00000000 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduled/HostHeartbeatScheduled.java +++ /dev/null @@ -1,81 +0,0 @@ -/* - * 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.server.scheduled; - -import org.apache.bigtop.manager.common.message.entity.HeartbeatMessage; -import org.apache.bigtop.manager.common.message.entity.pojo.HostInfo; -import org.apache.bigtop.manager.dao.entity.Host; -import org.apache.bigtop.manager.dao.repository.HostRepository; -import org.apache.bigtop.manager.server.ws.ServerWebSocketHandler; - -import org.springframework.scheduling.annotation.Async; -import org.springframework.scheduling.annotation.Scheduled; -import org.springframework.stereotype.Component; - -import lombok.extern.slf4j.Slf4j; - -import jakarta.annotation.Resource; -import java.util.List; -import java.util.Map; -import java.util.stream.Collectors; - -@Slf4j -@Component -public class HostHeartbeatScheduled { - - @Resource - private HostRepository hostRepository; - - @Async - @Scheduled(cron = "0/30 * * * * ? ") - public void execute() { - List hosts = hostRepository.findAll(); - Map hostMap = hosts.stream().collect(Collectors.toMap(Host::getHostname, host -> host)); - for (Map.Entry entry : ServerWebSocketHandler.HEARTBEAT_MESSAGE_MAP.entrySet()) { - String hostname = entry.getKey(); - HeartbeatMessage heartbeatMessage = entry.getValue(); - HostInfo hostInfo = heartbeatMessage.getHostInfo(); - if (hostMap.containsKey(hostname)) { - Host host = hostMap.get(hostname); - if (hostInfo != null) { - host.setArch(hostInfo.getArch()); - host.setAvailableProcessors(hostInfo.getAvailableProcessors()); - host.setIpv4(hostInfo.getIpv4()); - host.setIpv6(hostInfo.getIpv6()); - host.setOs(hostInfo.getOs()); - host.setFreeMemorySize(hostInfo.getFreeMemorySize()); - host.setTotalMemorySize(hostInfo.getTotalMemorySize()); - host.setFreeDisk(hostInfo.getFreeDisk()); - host.setTotalDisk(hostInfo.getTotalDisk()); - // heartbeat HEALTHY - } - hostRepository.save(host); - hostMap.remove(hostname); - } - } - - if (!hostMap.isEmpty()) { - for (Map.Entry entry : hostMap.entrySet()) { - Host host = entry.getValue(); - // heartbeat LOST - hostRepository.save(host); - } - } - } -} diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduler/HostInfoScheduler.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduler/HostInfoScheduler.java new file mode 100644 index 00000000..81705abe --- /dev/null +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/scheduler/HostInfoScheduler.java @@ -0,0 +1,83 @@ +/* + * 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.server.scheduler; + +import org.apache.bigtop.manager.common.enums.MaintainState; +import org.apache.bigtop.manager.dao.entity.Host; +import org.apache.bigtop.manager.dao.repository.HostRepository; +import org.apache.bigtop.manager.grpc.generated.HostInfoReply; +import org.apache.bigtop.manager.grpc.generated.HostInfoRequest; +import org.apache.bigtop.manager.grpc.generated.HostInfoServiceGrpc; +import org.apache.bigtop.manager.server.grpc.GrpcClient; + +import org.springframework.scheduling.annotation.Async; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import lombok.extern.slf4j.Slf4j; + +import jakarta.annotation.Resource; +import java.util.List; + +@Slf4j +@Component +public class HostInfoScheduler { + + @Resource + private HostRepository hostRepository; + + @Async + @Scheduled(cron = "0/30 * * * * ? ") + public void execute() { + List hosts = hostRepository.findAll(); + for (Host host : hosts) { + getHostInfo(host); + } + } + + private void getHostInfo(Host host) { + String hostname = host.getHostname(); + try { + if (!GrpcClient.isChannelAlive(hostname)) { + GrpcClient.createChannel(hostname); + } + + HostInfoServiceGrpc.HostInfoServiceBlockingStub stub = + GrpcClient.getBlockingStub(hostname, HostInfoServiceGrpc.HostInfoServiceBlockingStub.class); + HostInfoReply reply = stub.getHostInfo(HostInfoRequest.newBuilder().build()); + + host.setArch(reply.getArch()); + host.setAvailableProcessors(reply.getAvailableProcessors()); + host.setIpv4(reply.getIpv4()); + host.setIpv6(reply.getIpv6()); + host.setOs(reply.getOs()); + host.setFreeMemorySize(reply.getFreeMemorySize()); + host.setTotalMemorySize(reply.getTotalMemorySize()); + host.setFreeDisk(reply.getFreeDisk()); + host.setTotalDisk(reply.getTotalDisk()); + + host.setState(MaintainState.STARTED); + } catch (Exception e) { + log.error("Error getting host info", e); + host.setState(MaintainState.STOPPED); + } + + hostRepository.save(host); + } +} diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/HostService.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/HostService.java index 08dd788e..d6206bbc 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/HostService.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/HostService.java @@ -59,4 +59,12 @@ public interface HostService { * @return Host */ Boolean delete(Long id); + + /** + * Check hosts connection + * + * @param hostnames hostname list + * @return true if all hosts are able to connect + */ + Boolean checkConnection(List hostnames); } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/impl/HostServiceImpl.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/impl/HostServiceImpl.java index 87a5222d..b24edd04 100644 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/impl/HostServiceImpl.java +++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/service/impl/HostServiceImpl.java @@ -25,6 +25,7 @@ import org.apache.bigtop.manager.dao.repository.HostRepository; import org.apache.bigtop.manager.server.enums.ApiExceptionEnum; import org.apache.bigtop.manager.server.exception.ApiException; +import org.apache.bigtop.manager.server.grpc.GrpcClient; import org.apache.bigtop.manager.server.model.dto.HostDTO; import org.apache.bigtop.manager.server.model.mapper.HostMapper; import org.apache.bigtop.manager.server.model.vo.HostVO; @@ -110,4 +111,17 @@ public Boolean delete(Long id) { hostRepository.deleteById(id); return true; } + + @Override + public Boolean checkConnection(List hostnames) { + for (String hostname : hostnames) { + if (!GrpcClient.isChannelAlive(hostname)) { + // An api exception will throw if connection fails to establish, we don't need to handle the return + // value. + GrpcClient.createChannel(hostname); + } + } + + return true; + } } diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/DefaultWebSocketHandler.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/DefaultWebSocketHandler.java deleted file mode 100644 index b3f8cb22..00000000 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/DefaultWebSocketHandler.java +++ /dev/null @@ -1,53 +0,0 @@ -/* - * 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.server.ws; - -import org.springframework.stereotype.Component; -import org.springframework.web.socket.CloseStatus; -import org.springframework.web.socket.TextMessage; -import org.springframework.web.socket.WebSocketSession; -import org.springframework.web.socket.handler.TextWebSocketHandler; - -import lombok.extern.slf4j.Slf4j; - -/** - * WebSocket Endpoint for frontend. - */ -@Slf4j -@Component -public class DefaultWebSocketHandler extends TextWebSocketHandler { - - @Override - public void afterConnectionEstablished(WebSocketSession session) throws Exception { - super.afterConnectionEstablished(session); - System.out.println("Connection established"); - } - - @Override - protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { - super.handleTextMessage(session, message); - System.out.println("Received message: " + message.getPayload()); - } - - @Override - public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { - super.afterConnectionClosed(session, status); - System.out.println("Connection closed"); - } -} diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/ServerWebSocketHandler.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/ServerWebSocketHandler.java deleted file mode 100644 index b1c163ca..00000000 --- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/ws/ServerWebSocketHandler.java +++ /dev/null @@ -1,112 +0,0 @@ -/* - * 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.server.ws; - -import org.apache.bigtop.manager.common.message.entity.BaseMessage; -import org.apache.bigtop.manager.common.message.entity.BaseRequestMessage; -import org.apache.bigtop.manager.common.message.entity.HeartbeatMessage; -import org.apache.bigtop.manager.common.message.entity.command.CommandResponseMessage; -import org.apache.bigtop.manager.common.message.entity.pojo.HostInfo; -import org.apache.bigtop.manager.common.ws.AbstractBinaryWebSocketHandler; - -import org.springframework.stereotype.Component; -import org.springframework.web.socket.BinaryMessage; -import org.springframework.web.socket.CloseStatus; -import org.springframework.web.socket.WebSocketSession; - -import lombok.extern.slf4j.Slf4j; - -import jakarta.annotation.Nonnull; -import java.util.ArrayList; -import java.util.List; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -/** - * WebSocket Endpoint for agent. - */ -@Slf4j -@Component -public class ServerWebSocketHandler extends AbstractBinaryWebSocketHandler { - - private final Map sessions = new ConcurrentHashMap<>(); - - public static final Map HEARTBEAT_MESSAGE_MAP = new ConcurrentHashMap<>(); - - public void sendMessage(String hostname, BaseMessage message) { - WebSocketSession session = sessions.get(hostname); - if (session == null) { - log.warn("host: {}, is not connected, can't send message: {}", hostname, message); - return; - } - - try { - super.sendMessage(session, message); - } catch (Exception e) { - log.error("Error sending message {} to host: {}", message, hostname, e); - } - } - - public CommandResponseMessage sendRequestMessage(String hostname, BaseRequestMessage message) { - WebSocketSession session = sessions.get(hostname); - if (session == null) { - return null; - } else { - return (CommandResponseMessage) super.sendRequestMessage(session, message); - } - } - - @Override - protected void handleBinaryMessage(@Nonnull WebSocketSession session, BinaryMessage message) throws Exception { - BaseMessage baseMessage = deserializer.deserialize(message.getPayload().array()); - - handleMessage(session, baseMessage); - - log.debug(baseMessage.toString()); - } - - private void handleMessage(WebSocketSession session, BaseMessage baseMessage) { - log.info("Received message type: {}", baseMessage.getClass().getSimpleName()); - if (baseMessage instanceof HeartbeatMessage heartbeatMessage) { - handleHeartbeatMessage(session, heartbeatMessage); - } else if (baseMessage instanceof CommandResponseMessage commandResponseMessage) { - super.handleResponseMessage(commandResponseMessage); - } else { - log.error("Unrecognized message type: {}", baseMessage.getClass().getSimpleName()); - } - } - - private void handleHeartbeatMessage(WebSocketSession session, HeartbeatMessage heartbeatMessage) { - HostInfo hostInfo = heartbeatMessage.getHostInfo(); - sessions.putIfAbsent(hostInfo.getHostname(), session); - HEARTBEAT_MESSAGE_MAP.put(hostInfo.getHostname(), heartbeatMessage); - } - - @Override - public void afterConnectionClosed(WebSocketSession session, @Nonnull CloseStatus status) throws Exception { - log.error("session closed: {}, remove it!!!", session.getId()); - sessions.values().removeIf(value -> value.getId().equals(session.getId())); - HEARTBEAT_MESSAGE_MAP.clear(); - log.info("latest ServerWebSocketSessionManager.sessions: {}", sessions); - } - - public List getConnectedHosts() { - return new ArrayList<>(sessions.keySet()); - } -} diff --git a/bigtop-manager-server/src/main/resources/application.yml b/bigtop-manager-server/src/main/resources/application.yml index 06c1baba..8cb54a72 100644 --- a/bigtop-manager-server/src/main/resources/application.yml +++ b/bigtop-manager-server/src/main/resources/application.yml @@ -22,6 +22,8 @@ bigtop: orm: # hibernate/eclipselink(default=eclipselink) type: eclipselink + grpc: + port: 8835 spring: banner: diff --git a/bigtop-manager-server/src/main/resources/i18n/messages_en_US.properties b/bigtop-manager-server/src/main/resources/i18n/messages_en_US.properties index 3d565a11..6560c05a 100644 --- a/bigtop-manager-server/src/main/resources/i18n/messages_en_US.properties +++ b/bigtop-manager-server/src/main/resources/i18n/messages_en_US.properties @@ -34,6 +34,7 @@ cluster.is.installed=Cluster is installed host.not.found=Host not exist host.assigned=Hosts [{0}] already assigned to another cluster host.not.connected=Hosts [{0}] not connected +host.unable.to.connect=Unable to connect to host [{0}] stack.not.found=Stack not exist diff --git a/bigtop-manager-server/src/main/resources/i18n/messages_zh_CN.properties b/bigtop-manager-server/src/main/resources/i18n/messages_zh_CN.properties index 5648d66c..60d9ed7c 100644 --- a/bigtop-manager-server/src/main/resources/i18n/messages_zh_CN.properties +++ b/bigtop-manager-server/src/main/resources/i18n/messages_zh_CN.properties @@ -34,6 +34,7 @@ cluster.is.installed=集群已创建 host.not.found=主机不存在 host.assigned=主机 [{0}] 已属于其他集群 host.not.connected=主机 [{0}] 未连接 +host.unable.to.connect=无法连接到主机 [{0}] stack.not.found=组件栈不存在 diff --git a/bigtop-manager-spi/pom.xml b/bigtop-manager-spi/pom.xml index 18e41f94..e48b38a4 100644 --- a/bigtop-manager-spi/pom.xml +++ b/bigtop-manager-spi/pom.xml @@ -26,12 +26,8 @@ bigtop-manager-spi - - - 8 - 8 - UTF-8 - + ${project.artifactId} + Bigtop Manager Spi @@ -50,6 +46,7 @@ org.apache.bigtop bigtop-manager-common + org.slf4j slf4j-api diff --git a/bigtop-manager-stack/bigtop-manager-stack-bigtop/pom.xml b/bigtop-manager-stack/bigtop-manager-stack-bigtop/pom.xml index 904524d1..ad85d1ae 100644 --- a/bigtop-manager-stack/bigtop-manager-stack-bigtop/pom.xml +++ b/bigtop-manager-stack/bigtop-manager-stack-bigtop/pom.xml @@ -26,19 +26,15 @@ bigtop-manager-stack-bigtop - - - 8 - 8 - UTF-8 - + ${project.artifactId} + Bigtop Manager Stack Bigtop org.apache.bigtop bigtop-manager-stack-common - ${revision} + com.google.auto.service auto-service diff --git a/bigtop-manager-stack/bigtop-manager-stack-common/pom.xml b/bigtop-manager-stack/bigtop-manager-stack-common/pom.xml index 27fb90c3..07b46b74 100644 --- a/bigtop-manager-stack/bigtop-manager-stack-common/pom.xml +++ b/bigtop-manager-stack/bigtop-manager-stack-common/pom.xml @@ -26,12 +26,9 @@ bigtop-manager-stack-common + ${project.artifactId} + Bigtop Manager Stack Common - - 8 - 8 - UTF-8 - com.google.auto.service @@ -45,7 +42,6 @@ commons-io commons-io - diff --git a/bigtop-manager-stack/bigtop-manager-stack-core/pom.xml b/bigtop-manager-stack/bigtop-manager-stack-core/pom.xml index 98325fb5..7d0aa057 100644 --- a/bigtop-manager-stack/bigtop-manager-stack-core/pom.xml +++ b/bigtop-manager-stack/bigtop-manager-stack-core/pom.xml @@ -26,24 +26,18 @@ bigtop-manager-stack-core - - - 8 - 8 - UTF-8 - + ${project.artifactId} + Bigtop Manager Stack Core org.apache.bigtop bigtop-manager-stack-nop - ${revision} org.apache.bigtop bigtop-manager-stack-bigtop - ${revision} diff --git a/bigtop-manager-stack/bigtop-manager-stack-nop/pom.xml b/bigtop-manager-stack/bigtop-manager-stack-nop/pom.xml index 0a0dbcf2..e4ede864 100644 --- a/bigtop-manager-stack/bigtop-manager-stack-nop/pom.xml +++ b/bigtop-manager-stack/bigtop-manager-stack-nop/pom.xml @@ -26,19 +26,15 @@ bigtop-manager-stack-nop - - - 8 - 8 - UTF-8 - + ${project.artifactId} + Bigtop Manager Stack NOP org.apache.bigtop bigtop-manager-stack-common - ${revision} + com.google.auto.service auto-service diff --git a/bigtop-manager-stack/pom.xml b/bigtop-manager-stack/pom.xml index d3dcafba..98556bb2 100644 --- a/bigtop-manager-stack/pom.xml +++ b/bigtop-manager-stack/pom.xml @@ -28,8 +28,9 @@ bigtop-manager-stack - ${revision} pom + ${project.artifactId} + Bigtop Manager Stack Parent bigtop-manager-stack-core @@ -39,9 +40,6 @@ - 8 - 8 - UTF-8 1.0.1 @@ -71,6 +69,7 @@ org.apache.bigtop bigtop-manager-spi + org.projectlombok lombok diff --git a/bigtop-manager-ui/.env.development b/bigtop-manager-ui/.env.development index 4af78b52..65962102 100644 --- a/bigtop-manager-ui/.env.development +++ b/bigtop-manager-ui/.env.development @@ -19,6 +19,3 @@ VITE_APP_BASE='/' #VITE_APP_BASE_URL='http://172.29.40.96:8080' VITE_APP_BASE_URL='http://localhost:8080' VITE_APP_BASE_API='/api' -#VITE_APP_BASE_WS_URL='ws://172.29.40.96:8080' -VITE_APP_BASE_WS_URL='ws://localhost:8080' -VITE_APP_BASE_WS_API='/ws' \ No newline at end of file diff --git a/bigtop-manager-ui/.env.production b/bigtop-manager-ui/.env.production index 229c89d7..7fba695f 100644 --- a/bigtop-manager-ui/.env.production +++ b/bigtop-manager-ui/.env.production @@ -18,5 +18,3 @@ NODE_ENV=production VITE_APP_BASE='/ui/' VITE_APP_BASE_URL='' VITE_APP_BASE_API='/api' -VITE_APP_BASE_WS_URL='' -VITE_APP_BASE_WS_API='/ws' \ No newline at end of file diff --git a/bigtop-manager-ui/src/api/request.ts b/bigtop-manager-ui/src/api/request.ts index 0d54acc5..3a471b52 100644 --- a/bigtop-manager-ui/src/api/request.ts +++ b/bigtop-manager-ui/src/api/request.ts @@ -80,6 +80,7 @@ request.interceptors.response.use( } else if (error.code === AxiosError.ETIMEDOUT) { message.error(i18n.global.t('common.error_timeout')) } else { + console.log(error) message.error(i18n.global.t('common.error_unknown')) } diff --git a/bigtop-manager-ui/src/locales/en_US/common.ts b/bigtop-manager-ui/src/locales/en_US/common.ts index f2bccb46..ad0a1fc8 100644 --- a/bigtop-manager-ui/src/locales/en_US/common.ts +++ b/bigtop-manager-ui/src/locales/en_US/common.ts @@ -52,7 +52,6 @@ export default { error_timeout: 'Request timeout', desc_404: 'Sorry, the page you visited does not exist.', back_home: 'Back Home', - websocket_disconnected: 'WebSocket disconnected, please reload the page', create_time: 'Create Time', update_time: 'Update Time', notification: 'Notification', diff --git a/bigtop-manager-ui/src/locales/zh_CN/common.ts b/bigtop-manager-ui/src/locales/zh_CN/common.ts index 7a74eb26..281bb177 100644 --- a/bigtop-manager-ui/src/locales/zh_CN/common.ts +++ b/bigtop-manager-ui/src/locales/zh_CN/common.ts @@ -52,7 +52,6 @@ export default { error_timeout: '请求超时', desc_404: '抱歉，您查看的页面不存在', back_home: '返回首页', - websocket_disconnected: 'WebSocket 连接异常断开，请重新加载页面', create_time: '创建时间', update_time: '更新时间', notification: '通知', diff --git a/bigtop-manager-ui/src/utils/constant.ts b/bigtop-manager-ui/src/utils/constant.ts index 8148e4eb..9d8b8f32 100644 --- a/bigtop-manager-ui/src/utils/constant.ts +++ b/bigtop-manager-ui/src/utils/constant.ts @@ -17,28 +17,8 @@ * under the License. */ -import { message } from 'ant-design-vue' -import i18n from '@/locales' - export const API_RETRY_TIME = 3 -export const API_EXPIRE_TIME = 3 * 1000 +export const API_EXPIRE_TIME = 30 * 1000 export const JOB_SCHEDULE_INTERVAL = 1000 export const MONITOR_SCHEDULE_INTERVAL = 10 * 1000 export const DEFAULT_PAGE_SIZE = 10 - -export const WS_URL = 'ws://' + window.location.host + '/ws/default' - -export const WS_DEFAULT_OPTIONS = { - autoReconnect: { - retries: API_RETRY_TIME, - delay: API_EXPIRE_TIME, - async onFailed() { - message.error(i18n.global.t('common.websocket_disconnected')) - } - }, - heartbeat: { - message: 'ping', - interval: API_EXPIRE_TIME, - pongTimeout: API_EXPIRE_TIME - } -} diff --git a/bigtop-manager-ui/vite.config.ts b/bigtop-manager-ui/vite.config.ts index 1ec5b142..933e6249 100644 --- a/bigtop-manager-ui/vite.config.ts +++ b/bigtop-manager-ui/vite.config.ts @@ -51,10 +51,6 @@ export default defineConfig(({ mode }) => { [env.VITE_APP_BASE_API]: { target: env.VITE_APP_BASE_URL, changeOrigin: true - }, - [env.VITE_APP_BASE_WS_API]: { - target: env.VITE_APP_BASE_WS_URL, - ws: true } } } diff --git a/pom.xml b/pom.xml index 78b89cf4..4e2d3477 100644 --- a/pom.xml +++ b/pom.xml @@ -42,6 +42,7 @@ bigtop-manager-ui bigtop-manager-dao bigtop-manager-spi + bigtop-manager-grpc @@ -50,11 +51,20 @@ 17 1.4.0 3.3.0 + 3.6.0 3.8.1 3.2.0 3.1.1 2.43.0 + + 1.7.0 + 0.6.1 + 3.24.0 + 1.63.0 + 3.25.3 + 3.1.0.RELEASE + 1.5.5.Final 1.18.30 0.2.0 @@ -68,12 +78,48 @@ ${project.version} + + org.apache.bigtop + bigtop-manager-dao + ${project.version} + + + + org.apache.bigtop + bigtop-manager-grpc + ${project.version} + + org.apache.bigtop bigtop-manager-spi ${project.version} + + org.apache.bigtop + bigtop-manager-stack-common + ${project.version} + + + + org.apache.bigtop + bigtop-manager-stack-core + ${project.version} + + + + org.apache.bigtop + bigtop-manager-stack-bigtop + ${project.version} + + + + org.apache.bigtop + bigtop-manager-stack-nop + ${project.version} + + org.apache.bigtop bigtop-manager-ui @@ -115,6 +161,18 @@ maven-assembly-plugin ${maven-assembly-plugin.version} + + + org.codehaus.mojo + build-helper-maven-plugin + ${maven-assembly-plugin.version} + + + + org.xolstice.maven.plugins + protobuf-maven-plugin + ${protobuf-maven-plugin.version} + @@ -188,5 +246,12 @@ + + + kr.motd.maven + os-maven-plugin + ${os-maven-plugin.version} + +