content: /*
   * Copyright 1999-2018 Alibaba Group Holding Ltd.
   *
   * Licensed 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 com.alibaba.csp.sentinel.dashboard.service;
  
  import java.util.HashSet;
  import java.util.List;
  import java.util.Objects;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  
  import com.alibaba.csp.sentinel.cluster.ClusterStateManager;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.state.ClusterUniversalStatePairVO;
  import com.alibaba.csp.sentinel.util.AssertUtil;
  import com.alibaba.csp.sentinel.util.function.Tuple2;
  
  import com.alibaba.csp.sentinel.dashboard.client.SentinelApiClient;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.ClusterAppAssignResultVO;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.ClusterGroupEntity;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.config.ClusterClientConfig;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.config.ServerFlowConfig;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.config.ServerTransportConfig;
  import com.alibaba.csp.sentinel.dashboard.domain.cluster.request.ClusterAppAssignMap;
  import com.alibaba.csp.sentinel.dashboard.util.MachineUtils;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Service;
  
  /**
   * @author Eric Zhao
   * @since 1.4.1
   */
  @Service
  public class ClusterAssignServiceImpl implements ClusterAssignService {
  
      private final Logger LOGGER = LoggerFactory.getLogger(ClusterAssignServiceImpl.class);
  
      @Autowired
      private SentinelApiClient sentinelApiClient;
      @Autowired
      private ClusterConfigService clusterConfigService;
  
      private boolean isMachineInApp(/*@NonEmpty*/ String machineId) {
          return machineId.contains(":");
      }
  
      private ClusterAppAssignResultVO handleUnbindClusterServerNotInApp(String app, String machineId) {
          Set<String> failedSet = new HashSet<>();
          try {
              List<ClusterUniversalStatePairVO> list = clusterConfigService.getClusterUniversalState(app)
                  .get(10, TimeUnit.SECONDS);
              Set<String> toModifySet = list.stream()
                  .filter(e -> e.getState().getStateInfo().getMode() == ClusterStateManager.CLUSTER_CLIENT)
                  .filter(e -> machineId.equals(e.getState().getClient().getClientConfig().getServerHost() + ':' +
                      e.getState().getClient().getClientConfig().getServerPort()))
                  .map(e -> e.getIp() + '@' + e.getCommandPort())
                  .collect(Collectors.toSet());
              // Modify mode to NOT-STARTED for all associated token clients.
              modifyToNonStarted(toModifySet, failedSet);
          } catch (Exception ex) {
              Throwable e = ex instanceof ExecutionException ? ex.getCause() : ex;
              LOGGER.error("Failed to unbind machine <{}>", machineId, e);
              failedSet.add(machineId);
          }
          return new ClusterAppAssignResultVO()
              .setFailedClientSet(failedSet)
              .setFailedServerSet(new HashSet<>());
      }
  
      private void modifyToNonStarted(Set<String> toModifySet, Set<String> failedSet) {
          toModifySet.parallelStream()
              .map(MachineUtils::parseCommandIpAndPort)
              .filter(Optional::isPresent)
              .map(Optional::get)
              .map(e -> {
                  CompletableFuture<Void> f = modifyMode(e.r1, e.r2, ClusterStateManager.CLUSTER_NOT_STARTED);
                  return Tuple2.of(e.r1 + '@' + e.r2, f);
              })
              .forEach(f -> handleFutureSync(f, failedSet));
      }
  
      @Override
      public ClusterAppAssignResultVO unbindClusterServer(String app, String machineId) {
          AssertUtil.assertNotBlank(app, "app cannot be blank");
          AssertUtil.assertNotBlank(machineId, "machineId cannot be blank");
  
          if (isMachineInApp(machineId)) {
              return handleUnbindClusterServerNotInApp(app, machineId);
          }
          Set<String> failedSet = new HashSet<>();
          try {
              ClusterGroupEntity entity = clusterConfigService.getClusterUniversalStateForAppMachine(app, machineId)
                  .get(10, TimeUnit.SECONDS);
              Set<String> toModifySet = new HashSet<>();
              toModifySet.add(machineId);
              if (entity.getClientSet() != null) {
                  toModifySet.addAll(entity.getClientSet());
              }
              // Modify mode to NOT-STARTED for all chosen token servers and associated token clients.
              modifyToNonStarted(toModifySet, failedSet);
          } catch (Exception ex) {
              Throwable e = ex instanceof ExecutionException ? ex.getCause() : ex;
              LOGGER.error("Failed to unbind machine <{}>", machineId, e);
              failedSet.add(machineId);
          }
          return new ClusterAppAssignResultVO()
              .setFailedClientSet(failedSet)
              .setFailedServerSet(new HashSet<>());
      }
  
      @Override
      public ClusterAppAssignResultVO unbindClusterServers(String app, Set<String> machineIdSet) {
          AssertUtil.assertNotBlank(app, "app cannot be blank");
          AssertUtil.isTrue(machineIdSet != null && !machineIdSet.isEmpty(), "machineIdSet cannot be empty");
          ClusterAppAssignResultVO result = new ClusterAppAssignResultVO()
              .setFailedClientSet(new HashSet<>())
              .setFailedServerSet(new HashSet<>());
          for (String machineId : machineIdSet) {
              ClusterAppAssignResultVO resultVO = unbindClusterServer(app, machineId);
              result.getFailedClientSet().addAll(resultVO.getFailedClientSet());
              result.getFailedServerSet().addAll(resultVO.getFailedServerSet());
          }
          return result;
      }
  
      @Override
      public ClusterAppAssignResultVO applyAssignToApp(String app, List<ClusterAppAssignMap> clusterMap,
                                                       Set<String> remainingSet) {
          AssertUtil.assertNotBlank(app, "app cannot be blank");
          AssertUtil.notNull(clusterMap, "clusterMap cannot be null");
          Set<String> failedServerSet = new HashSet<>();
          Set<String> failedClientSet = new HashSet<>();
  
          // Assign server and apply config.
          clusterMap.stream()
              .filter(Objects::nonNull)
              .filter(ClusterAppAssignMap::getBelongToApp)
              .map(e -> {
                  String ip = e.getIp();
                  int commandPort = parsePort(e);
                  CompletableFuture<Void> f = modifyMode(ip, commandPort, ClusterStateManager.CLUSTER_SERVER)
                      .thenCompose(v -> applyServerConfigChange(app, ip, commandPort, e));
                  return Tuple2.of(e.getMachineId(), f);
              })
              .forEach(t -> handleFutureSync(t, failedServerSet));
  
          // Assign client of servers and apply config.
          clusterMap.parallelStream()
              .filter(Objects::nonNull)
              .forEach(e -> applyAllClientConfigChange(app, e, failedClientSet));
  
          // Unbind remaining (unassigned) machines.
          applyAllRemainingMachineSet(app, remainingSet, failedClientSet);
  
          return new ClusterAppAssignResultVO()
              .setFailedClientSet(failedClientSet)
              .setFailedServerSet(failedServerSet);
      }
  
      private void applyAllRemainingMachineSet(String app, Set<String> remainingSet, Set<String> failedSet) {
          if (remainingSet == null || remainingSet.isEmpty()) {
              return;
          }
          remainingSet.parallelStream()
              .filter(Objects::nonNull)
              .map(MachineUtils::parseCommandIpAndPort)
              .filter(Optional::isPresent)
              .map(Optional::get)
              .map(ipPort -> {
                  String ip = ipPort.r1;
                  int commandPort = ipPort.r2;
                  CompletableFuture<Void> f = modifyMode(ip, commandPort, ClusterStateManager.CLUSTER_NOT_STARTED);
                  return Tuple2.of(ip + '@' + commandPort, f);
              })
              .forEach(t -> handleFutureSync(t, failedSet));
      }
  
      private void applyAllClientConfigChange(String app, ClusterAppAssignMap assignMap,
                                              Set<String> failedSet) {
          Set<String> clientSet = assignMap.getClientSet();
          if (clientSet == null || clientSet.isEmpty()) {
              return;
          }
          final String serverIp = assignMap.getIp();
          final int serverPort = assignMap.getPort();
          clientSet.stream()
              .map(MachineUtils::parseCommandIpAndPort)
              .filter(Optional::isPresent)
              .map(Optional::get)
              .map(ipPort -> {
                  CompletableFuture<Void> f = sentinelApiClient
                      .modifyClusterMode(ipPort.r1, ipPort.r2, ClusterStateManager.CLUSTER_CLIENT)
                      .thenCompose(v -> sentinelApiClient.modifyClusterClientConfig(app, ipPort.r1, ipPort.r2,
                          new ClusterClientConfig().setRequestTimeout(20)
                              .setServerHost(serverIp)
                              .setServerPort(serverPort)
                      ));
                  return Tuple2.of(ipPort.r1 + '@' + ipPort.r2, f);
              })
              .forEach(t -> handleFutureSync(t, failedSet));
      }
  
      private void handleFutureSync(Tuple2<String, CompletableFuture<Void>> t, Set<String> failedSet) {
          try {
              t.r2.get(10, TimeUnit.SECONDS);
          } catch (Exception ex) {
              if (ex instanceof ExecutionException) {
                  LOGGER.error("Request for <{}> failed", t.r1, ex.getCause());
              } else {
                  LOGGER.error("Request for <{}> failed", t.r1, ex);
              }
              failedSet.add(t.r1);
          }
      }
  
      private CompletableFuture<Void> applyServerConfigChange(String app, String ip, int commandPort,
                                                              ClusterAppAssignMap assignMap) {
          ServerTransportConfig transportConfig = new ServerTransportConfig()
              .setPort(assignMap.getPort())
              .setIdleSeconds(600);
          return sentinelApiClient.modifyClusterServerTransportConfig(app, ip, commandPort, transportConfig)
              .thenCompose(v -> applyServerFlowConfigChange(app, ip, commandPort, assignMap))
              .thenCompose(v -> applyServerNamespaceSetConfig(app, ip, commandPort, assignMap));
      }
  
      private CompletableFuture<Void> applyServerFlowConfigChange(String app, String ip, int commandPort,
                                                                  ClusterAppAssignMap assignMap) {
          Double maxAllowedQps = assignMap.getMaxAllowedQps();
          if (maxAllowedQps == null || maxAllowedQps <= 0 || maxAllowedQps > 20_0000) {
              return CompletableFuture.completedFuture(null);
          }
          return sentinelApiClient.modifyClusterServerFlowConfig(app, ip, commandPort,
              new ServerFlowConfig().setMaxAllowedQps(maxAllowedQps));
      }
  
      private CompletableFuture<Void> applyServerNamespaceSetConfig(String app, String ip, int commandPort,
                                                                    ClusterAppAssignMap assignMap) {
          Set<String> namespaceSet = assignMap.getNamespaceSet();
          if (namespaceSet == null || namespaceSet.isEmpty()) {
              return CompletableFuture.completedFuture(null);
          }
          return sentinelApiClient.modifyClusterServerNamespaceSet(app, ip, commandPort, namespaceSet);
      }
  
      private CompletableFuture<Void> modifyMode(String ip, int port, int mode) {
          return sentinelApiClient.modifyClusterMode(ip, port, mode);
      }
  
      private int parsePort(ClusterAppAssignMap assignMap) {
          return MachineUtils.parseCommandPort(assignMap.getMachineId())
              .orElse(ServerTransportConfig.DEFAULT_PORT);
      }
  }
  