content: /*
   * Copyright 1999-2020 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.nacos.core.controller;
  
  import com.alibaba.nacos.api.exception.NacosException;
  import com.alibaba.nacos.api.remote.RequestCallBack;
  import com.alibaba.nacos.api.remote.request.RequestMeta;
  import com.alibaba.nacos.api.remote.request.ServerLoaderInfoRequest;
  import com.alibaba.nacos.api.remote.request.ServerReloadRequest;
  import com.alibaba.nacos.api.remote.response.Response;
  import com.alibaba.nacos.api.remote.response.ServerLoaderInfoResponse;
  import com.alibaba.nacos.auth.annotation.Secured;
  import com.alibaba.nacos.common.utils.StringUtils;
  import com.alibaba.nacos.core.cluster.Member;
  import com.alibaba.nacos.core.cluster.MemberUtil;
  import com.alibaba.nacos.core.cluster.ServerMemberManager;
  import com.alibaba.nacos.core.cluster.remote.ClusterRpcClientProxy;
  import com.alibaba.nacos.core.controller.compatibility.Compatibility;
  import com.alibaba.nacos.core.remote.Connection;
  import com.alibaba.nacos.core.remote.ConnectionManager;
  import com.alibaba.nacos.core.remote.core.ServerLoaderInfoRequestHandler;
  import com.alibaba.nacos.core.remote.core.ServerReloaderRequestHandler;
  import com.alibaba.nacos.core.utils.Commons;
  import com.alibaba.nacos.core.utils.RemoteUtils;
  import com.alibaba.nacos.plugin.auth.constant.ActionTypes;
  import com.alibaba.nacos.plugin.auth.constant.ApiType;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  import org.springframework.http.ResponseEntity;
  import org.springframework.web.bind.annotation.GetMapping;
  import org.springframework.web.bind.annotation.RequestMapping;
  import org.springframework.web.bind.annotation.RequestParam;
  import org.springframework.web.bind.annotation.RestController;
  
  import jakarta.servlet.http.HttpServletRequest;
  import java.util.ArrayList;
  import java.util.Comparator;
  import java.util.HashMap;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.Executor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  
  /**
   * controller to control server loader.
   *
   * @author liuzunfei
   * @version $Id: ServerLoaderController.java, v 0.1 2020年07月22日 4:28 PM liuzunfei Exp $
   */
  @RestController
  @RequestMapping(Commons.NACOS_CORE_CONTEXT_V2 + "/loader")
  @Deprecated
  @SuppressWarnings("PMD.MethodTooLongRule")
  public class ServerLoaderController {
      
      private static final Logger LOGGER = LoggerFactory.getLogger(ServerLoaderController.class);
      
      private static final String X_REAL_IP = "X-Real-IP";
      
      private static final String X_FORWARDED_FOR = "X-Forwarded-For";
      
      private static final String X_FORWARDED_FOR_SPLIT_SYMBOL = ",";
      
      private static final String SUCCESS_RESULT = "Ok";
      
      private static final String FAIL_RESULT = "Fail";
      
      private static final String SDK_CONNECTION_COUNT_METRIC = "sdkConCount";
      
      private final ConnectionManager connectionManager;
      
      private final ServerMemberManager serverMemberManager;
      
      private final ClusterRpcClientProxy clusterRpcClientProxy;
      
      private final ServerReloaderRequestHandler serverReloaderRequestHandler;
      
      private final ServerLoaderInfoRequestHandler serverLoaderInfoRequestHandler;
      
      public ServerLoaderController(ConnectionManager connectionManager, ServerMemberManager serverMemberManager,
              ClusterRpcClientProxy clusterRpcClientProxy, ServerReloaderRequestHandler serverReloaderRequestHandler,
              ServerLoaderInfoRequestHandler serverLoaderInfoRequestHandler) {
          this.connectionManager = connectionManager;
          this.serverMemberManager = serverMemberManager;
          this.clusterRpcClientProxy = clusterRpcClientProxy;
          this.serverReloaderRequestHandler = serverReloaderRequestHandler;
          this.serverLoaderInfoRequestHandler = serverLoaderInfoRequestHandler;
      }
      
      /**
       * Get current clients.
       *
       * @return state json.
       */
      @Secured(resource = Commons.NACOS_CORE_CONTEXT_V2 + "/loader", action = ActionTypes.READ)
      @GetMapping("/current")
      @Compatibility(apiType = ApiType.ADMIN_API, alternatives = "GET {contextPath:nacos}/v3/admin/core/loader/current")
      public ResponseEntity<Map<String, Connection>> currentClients() {
          Map<String, Connection> stringConnectionMap = connectionManager.currentClients();
          return ResponseEntity.ok().body(stringConnectionMap);
      }
      
      /**
       * Rebalance the number of sdk connections on the current server.
       *
       * @return state json.
       */
      @Secured(resource = Commons.NACOS_CORE_CONTEXT_V2 + "/loader", action = ActionTypes.WRITE)
      @GetMapping("/reloadCurrent")
      @Compatibility(apiType = ApiType.ADMIN_API, alternatives = "GET {contextPath:nacos}/v3/admin/core/loader/reloadCurrent")
      public ResponseEntity<String> reloadCount(@RequestParam Integer count,
              @RequestParam(value = "redirectAddress", required = false) String redirectAddress) {
          connectionManager.loadCount(count, redirectAddress);
          return ResponseEntity.ok().body("success");
      }
      
      /**
       * According to the total number of sdk connections of all nodes in the nacos cluster, intelligently balance the
       * number of sdk connections of each node in the nacos cluster.
       *
       * @return state json.
       */
      @Secured(resource = Commons.NACOS_CORE_CONTEXT_V2 + "/loader", action = ActionTypes.WRITE)
      @GetMapping("/smartReloadCluster")
      @Compatibility(apiType = ApiType.ADMIN_API, alternatives = "GET {contextPath:nacos}/v3/admin/core/loader/smartReloadCluster")
      public ResponseEntity<String> smartReload(HttpServletRequest request,
              @RequestParam(value = "loaderFactor", required = false) String loaderFactorStr,
              @RequestParam(value = "force", required = false) String force) {
          
          LOGGER.info("Smart reload request receive,requestIp={}", getRemoteIp(request));
          
          Map<String, Object> serverLoadMetrics = getServerLoadMetrics();
          Object avgString = serverLoadMetrics.get("avg");
          List<ServerLoaderMetrics> details = (List<ServerLoaderMetrics>) serverLoadMetrics.get("detail");
          int avg = Integer.parseInt(avgString.toString());
          float loaderFactor =
                  StringUtils.isBlank(loaderFactorStr) ? RemoteUtils.LOADER_FACTOR : Float.parseFloat(loaderFactorStr);
          int overLimitCount = (int) (avg * (1 + loaderFactor));
          int lowLimitCount = (int) (avg * (1 - loaderFactor));
          
          List<ServerLoaderMetrics> overLimitServer = new ArrayList<>();
          List<ServerLoaderMetrics> lowLimitServer = new ArrayList<>();
          
          for (ServerLoaderMetrics metrics : details) {
              int sdkCount = Integer.parseInt(metrics.getMetric().get(SDK_CONNECTION_COUNT_METRIC));
              if (sdkCount > overLimitCount) {
                  overLimitServer.add(metrics);
              }
              if (sdkCount < lowLimitCount) {
                  lowLimitServer.add(metrics);
              }
          }
          
          // desc by sdkConCount
          overLimitServer.sort((o1, o2) -> {
              Integer sdkCount1 = Integer.valueOf(o1.getMetric().get(SDK_CONNECTION_COUNT_METRIC));
              Integer sdkCount2 = Integer.valueOf(o2.getMetric().get(SDK_CONNECTION_COUNT_METRIC));
              return sdkCount1.compareTo(sdkCount2) * -1;
          });
          
          LOGGER.info("Over load limit server list ={}", overLimitServer);
          
          //asc by sdkConCount
          lowLimitServer.sort((o1, o2) -> {
              Integer sdkCount1 = Integer.valueOf(o1.getMetric().get(SDK_CONNECTION_COUNT_METRIC));
              Integer sdkCount2 = Integer.valueOf(o2.getMetric().get(SDK_CONNECTION_COUNT_METRIC));
              return sdkCount1.compareTo(sdkCount2);
          });
          
          LOGGER.info("Low load limit server list ={}", lowLimitServer);
          AtomicBoolean result = new AtomicBoolean(true);
          
          for (int i = 0; i < overLimitServer.size() & i < lowLimitServer.size(); i++) {
              ServerReloadRequest serverLoaderInfoRequest = new ServerReloadRequest();
              serverLoaderInfoRequest.setReloadCount(overLimitCount);
              serverLoaderInfoRequest.setReloadServer(lowLimitServer.get(i).address);
              Member member = serverMemberManager.find(overLimitServer.get(i).address);
              
              LOGGER.info("Reload task submit ,fromServer ={},toServer={}, ", overLimitServer.get(i).address,
                      lowLimitServer.get(i).address);
              
              if (serverMemberManager.getSelf().equals(member)) {
                  try {
                      serverReloaderRequestHandler.handle(serverLoaderInfoRequest, new RequestMeta());
                  } catch (NacosException e) {
                      LOGGER.error("Fail to loader self server", e);
                      result.set(false);
                  }
              } else {
                  
                  try {
                      clusterRpcClientProxy.asyncRequest(member, serverLoaderInfoRequest, new RequestCallBack() {
                          @Override
                          public Executor getExecutor() {
                              return null;
                          }
                          
                          @Override
                          public long getTimeout() {
                              return 100L;
                          }
                          
                          @Override
                          public void onResponse(Response response) {
                              if (response == null || !response.isSuccess()) {
                                  LOGGER.error("Fail to loader member={},response={}", member.getAddress(), response);
                                  result.set(false);
                                  
                              }
                          }
                          
                          @Override
                          public void onException(Throwable e) {
                              LOGGER.error("Fail to loader member={}", member.getAddress(), e);
                              result.set(false);
                          }
                      });
                  } catch (NacosException e) {
                      LOGGER.error("Fail to loader member={}", member.getAddress(), e);
                      result.set(false);
                  }
              }
          }
          
          return ResponseEntity.ok().body(result.get() ? SUCCESS_RESULT : FAIL_RESULT);
      }
      
      
      /**
       * Send a ConnectResetRequest to this connection according to the sdk connection ID.
       *
       * @return state json.
       */
      @Secured(resource = Commons.NACOS_CORE_CONTEXT_V2 + "/loader", action = ActionTypes.WRITE)
      @GetMapping("/reloadClient")
      @Compatibility(apiType = ApiType.ADMIN_API, alternatives = "GET {contextPath:nacos}/v3/admin/core/loader/reloadClient")
      public ResponseEntity<String> reloadSingle(@RequestParam String connectionId,
              @RequestParam(value = "redirectAddress", required = false) String redirectAddress) {
          connectionManager.loadSingle(connectionId, redirectAddress);
          return ResponseEntity.ok().body("success");
      }
      
      /**
       * Get current clients.
       *
       * @return state json.
       */
      @Secured(resource = Commons.NACOS_CORE_CONTEXT_V2 + "/loader", action = ActionTypes.READ)
      @GetMapping("/cluster")
      @Compatibility(apiType = ApiType.ADMIN_API, alternatives = "GET {contextPath:nacos}/v3/admin/core/loader/cluster")
      public ResponseEntity<Map<String, Object>> loaderMetrics() {
          
          Map<String, Object> serverLoadMetrics = getServerLoadMetrics();
          
          return ResponseEntity.ok().body(serverLoadMetrics);
      }
      
      private Map<String, Object> getServerLoadMetrics() {
          
          List<ServerLoaderMetrics> responseList = new LinkedList<>();
          
          // default include self.
          int memberSize = serverMemberManager.allMembersWithoutSelf().size();
          CountDownLatch countDownLatch = new CountDownLatch(memberSize);
          for (Member member : serverMemberManager.allMembersWithoutSelf()) {
              if (MemberUtil.isSupportedLongCon(member)) {
                  ServerLoaderInfoRequest serverLoaderInfoRequest = new ServerLoaderInfoRequest();
                  
                  try {
                      clusterRpcClientProxy.asyncRequest(member, serverLoaderInfoRequest, new RequestCallBack() {
                          @Override
                          public Executor getExecutor() {
                              return null;
                          }
                          
                          @Override
                          public long getTimeout() {
                              return 200L;
                          }
                          
                          @Override
                          public void onResponse(Response response) {
                              if (response instanceof ServerLoaderInfoResponse) {
                                  ServerLoaderMetrics metrics = new ServerLoaderMetrics();
                                  metrics.setAddress(member.getAddress());
                                  metrics.setMetric(((ServerLoaderInfoResponse) response).getLoaderMetrics());
                                  responseList.add(metrics);
                              }
                              countDownLatch.countDown();
                          }
                          
                          @Override
                          public void onException(Throwable e) {
                              LOGGER.error("Get metrics fail,member={}", member.getAddress(), e);
                              countDownLatch.countDown();
                          }
                      });
                  } catch (NacosException e) {
                      LOGGER.error("Get metrics fail,member={}", member.getAddress(), e);
                      countDownLatch.countDown();
                  }
              } else {
                  countDownLatch.countDown();
              }
          }
          
          try {
              ServerLoaderInfoResponse handle = serverLoaderInfoRequestHandler.handle(new ServerLoaderInfoRequest(),
                      new RequestMeta());
              ServerLoaderMetrics metrics = new ServerLoaderMetrics();
              metrics.setAddress(serverMemberManager.getSelf().getAddress());
              metrics.setMetric(handle.getLoaderMetrics());
              responseList.add(metrics);
          } catch (NacosException e) {
              LOGGER.error("Get self metrics fail", e);
          }
          
          try {
              countDownLatch.await(1000, TimeUnit.MILLISECONDS);
          } catch (InterruptedException e) {
              LOGGER.warn("Get  metrics timeout,metrics info may not complete.");
          }
          int max = 0;
          int min = -1;
          int total = 0;
          
          for (ServerLoaderMetrics serverLoaderMetrics : responseList) {
              String sdkConCountStr = serverLoaderMetrics.getMetric().get("sdkConCount");
              
              if (StringUtils.isNotBlank(sdkConCountStr)) {
                  int sdkConCount = Integer.parseInt(sdkConCountStr);
                  if (max == 0 || max < sdkConCount) {
                      max = sdkConCount;
                  }
                  if (min == -1 || sdkConCount < min) {
                      min = sdkConCount;
                  }
                  total += sdkConCount;
              }
          }
          Map<String, Object> responseMap = new HashMap<>(9);
          responseList.sort(Comparator.comparing(ServerLoaderMetrics::getAddress));
          responseMap.put("detail", responseList);
          responseMap.put("memberCount", serverMemberManager.allMembers().size());
          responseMap.put("metricsCount", responseList.size());
          responseMap.put("completed", responseList.size() == serverMemberManager.allMembers().size());
          responseMap.put("max", max);
          responseMap.put("min", min);
          responseMap.put("avg", total / responseList.size());
          responseMap.put("threshold", total / responseList.size() * 1.1);
          responseMap.put("total", total);
          return responseMap;
          
      }
      
      class ServerLoaderMetrics {
          
          String address;
          
          Map<String, String> metric = new HashMap<>();
          
          /**
           * Getter method for property <tt>address</tt>.
           *
           * @return property value of address
           */
          public String getAddress() {
              return address;
          }
          
          /**
           * Setter method for property <tt>address</tt>.
           *
           * @param address value to be assigned to property address
           */
          public void setAddress(String address) {
              this.address = address;
          }
          
          /**
           * Getter method for property <tt>metric</tt>.
           *
           * @return property value of metric
           */
          public Map<String, String> getMetric() {
              return metric;
          }
          
          /**
           * Setter method for property <tt>metric</tt>.
           *
           * @param metric value to be assigned to property metric
           */
          public void setMetric(Map<String, String> metric) {
              this.metric = metric;
          }
      }
      
      private static String getRemoteIp(HttpServletRequest request) {
          String xForwardedFor = request.getHeader(X_FORWARDED_FOR);
          if (!StringUtils.isBlank(xForwardedFor)) {
              return xForwardedFor.split(X_FORWARDED_FOR_SPLIT_SYMBOL)[0].trim();
          }
          String nginxHeader = request.getHeader(X_REAL_IP);
          return StringUtils.isBlank(nginxHeader) ? request.getRemoteAddr() : nginxHeader;
      }
  }
  