CloudEon云原生大数据平台，构建于kubernetes集群之上的大数据集群管理平台

/* * Licensed to the Apache Software Foundation (ASF) under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The ASF licenses this file to You under the Apache License, Version 2.0 * (the "License"); you may not use this file except in compliance with * the License. You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package org.dromara.cloudeon.controller; import cn.hutool.core.bean.BeanUtil; import cn.hutool.core.collection.ListUtil; import cn.hutool.core.lang.Dict; import cn.hutool.core.thread.ThreadUtil; import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson.JSONObject; import com.google.common.collect.Lists; import freemarker.cache.StringTemplateLoader; import freemarker.template.Configuration; import freemarker.template.Template; import freemarker.template.TemplateException; import io.vertx.core.Vertx; import lombok.extern.slf4j.Slf4j; import org.dromara.cloudeon.config.CloudeonConfigProp; import org.dromara.cloudeon.controller.request.InitServiceRequest; import org.dromara.cloudeon.controller.request.ServiceConfUpgradeRequest; import org.dromara.cloudeon.controller.response.*; import org.dromara.cloudeon.dao.*; import org.dromara.cloudeon.dto.*; import org.dromara.cloudeon.entity.*; import org.dromara.cloudeon.enums.*; import org.dromara.cloudeon.processor.TaskParam; import org.dromara.cloudeon.service.CommandHandler; import org.dromara.cloudeon.utils.DAG; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.bind.annotation.*; import javax.annotation.Resource; import javax.persistence.EntityManager; import java.io.File; import java.io.IOException; import java.io.StringWriter; import java.io.Writer; import java.util.*; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; import java.util.function.Predicate; import java.util.stream.Collectors; import static org.dromara.cloudeon.utils.Constant.*; /** * 集群服务相关接口 */ @RestController @RequestMapping("/service") @Slf4j public class ClusterServiceController { ExecutorService flowSchedulerThreadPool = ThreadUtil.newExecutor(5, 10, 1024); @Resource private CloudeonConfigProp cloudeonConfigProp; @Resource(name = "cloudeonVertx") private Vertx cloudeonVertx; @Resource private AlertMessageRepository alertMessageRepository; @Resource private ServiceInstanceRepository serviceInstanceRepository; @Resource private ServiceRoleInstanceRepository roleInstanceRepository; @Resource private StackServiceRoleRepository stackServiceRoleRepository; @Resource private StackServiceRepository stackServiceRepository; @Resource private ServiceRoleInstanceWebuisRepository roleInstanceWebuisRepository; @Resource private ServiceInstanceConfigRepository serviceInstanceConfigRepository; @Resource private EntityManager entityManager; @Resource private CommandRepository commandRepository; @Resource private CommandTaskRepository commandTaskRepository; @Resource private CommandHandler commandHandler; @Resource private ClusterNodeRepository clusterNodeRepository; @Resource private StackServiceConfRepository stackServiceConfRepository; @Resource private ServiceInstanceSeqRepository serviceInstanceSeqRepository; @PostMapping("/initService") public ResultDTO<Void> initService(@RequestBody InitServiceRequest req) { Integer clusterId = req.getClusterId(); Integer stackId = req.getStackId(); List<Integer> installedServiceInstanceIds = new ArrayList<>(); List<InitServiceRequest.ServiceInfo> serviceInfos = req.getServiceInfos(); // 校验该集群是否已经安装过相同的服务了 String errorServiceInstanceNames = serviceInfos.stream().map(info -> { ServiceInstanceEntity sameStackServiceInstance = serviceInstanceRepository.findByClusterIdAndStackServiceId(clusterId, info.getStackServiceId()); if (sameStackServiceInstance != null) { return sameStackServiceInstance.getServiceName(); } return null; }).filter(StrUtil::isNotBlank).collect(Collectors.joining(",")); if (StrUtil.isNotBlank(errorServiceInstanceNames)) { return ResultDTO.failed("该集群已经安装过相同的服务实例：" + errorServiceInstanceNames); } // 根据服务间依赖关系调整安装顺序 List<InitServiceRequest.ServiceInfo> sortedServiceInfos = changeInstallSortByDependence(req); for (InitServiceRequest.ServiceInfo serviceInfo : sortedServiceInfos) { Integer stackServiceId = serviceInfo.getStackServiceId(); // 查询实例表获取新增的实例序号 ServiceInstanceSeqEntity serviceInstanceSeqEntity = serviceInstanceSeqRepository.findByStackServiceId(stackServiceId); Integer maxInstanceSeq; // 不存在该服务的序号，说明之前没安装过 if (serviceInstanceSeqEntity == null) { maxInstanceSeq = 1; serviceInstanceSeqEntity = ServiceInstanceSeqEntity.builder().maxSeq(maxInstanceSeq).stackServiceId(stackServiceId).build(); } else { maxInstanceSeq = serviceInstanceSeqEntity.getMaxSeq() + 1; serviceInstanceSeqEntity.setMaxSeq(maxInstanceSeq); } ServiceInstanceEntity serviceInstanceEntity = new ServiceInstanceEntity(); serviceInstanceEntity.setInstanceSequence(maxInstanceSeq); String stackServiceName = serviceInfo.getStackServiceName().toLowerCase(); String serviceName = stackServiceName + maxInstanceSeq; serviceInstanceEntity.setServiceName(serviceName); serviceInstanceEntity.setLabel(serviceInfo.getStackServiceLabel()); serviceInstanceEntity.setClusterId(clusterId); serviceInstanceEntity.setCreateTime(new Date()); serviceInstanceEntity.setUpdateTime(new Date()); serviceInstanceEntity.setEnableKerberos(req.getEnableKerberos()); serviceInstanceEntity.setStackServiceId(stackServiceId); serviceInstanceEntity.setServiceState(ServiceState.INIT_SERVICE); // 生成持久化宿主机路径 String persistencePaths = stackServiceRepository.findById(stackServiceId).get().getPersistencePaths(); serviceInstanceEntity.setPersistencePaths(genPersistencePaths(persistencePaths, serviceInstanceEntity)); // 持久化service信息 serviceInstanceRepository.save(serviceInstanceEntity); // 持久化service instance 序号 serviceInstanceSeqRepository.save(serviceInstanceSeqEntity); // 获取持久化后的service 实例id Integer serviceInstanceEntityId = serviceInstanceEntity.getId(); installedServiceInstanceIds.add(serviceInstanceEntityId); List<ServiceInstanceConfigEntity> serviceInstanceConfigEntities = new ArrayList<>(); // 页面上的预设配置 List<ServicePresetConf> presetConfList = serviceInfo.getPresetConfList(); // 用户自定义配置 List<ServiceCustomConf> customConfList = serviceInfo.getCustomConfList(); fillInstanceConfigEntities(stackId, stackServiceId, serviceInstanceEntityId, serviceInstanceConfigEntities,