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.cluster.flow.rule;
  
  import java.util.ArrayList;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Set;
  import java.util.concurrent.ConcurrentHashMap;
  
  import com.alibaba.csp.sentinel.cluster.flow.statistic.ClusterParamMetricStatistics;
  import com.alibaba.csp.sentinel.cluster.flow.statistic.metric.ClusterParamMetric;
  import com.alibaba.csp.sentinel.cluster.server.ServerConstants;
  import com.alibaba.csp.sentinel.cluster.server.connection.ConnectionManager;
  import com.alibaba.csp.sentinel.cluster.server.util.ClusterRuleUtil;
  import com.alibaba.csp.sentinel.log.RecordLog;
  import com.alibaba.csp.sentinel.property.DynamicSentinelProperty;
  import com.alibaba.csp.sentinel.property.PropertyListener;
  import com.alibaba.csp.sentinel.property.SentinelProperty;
  import com.alibaba.csp.sentinel.slots.block.RuleConstant;
  import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowClusterConfig;
  import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRule;
  import com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowRuleUtil;
  import com.alibaba.csp.sentinel.util.AssertUtil;
  import com.alibaba.csp.sentinel.util.StringUtil;
  import com.alibaba.csp.sentinel.util.function.Function;
  import com.alibaba.csp.sentinel.util.function.Predicate;
  
  /**
   * Manager for cluster parameter flow rules.
   *
   * @author Eric Zhao
   * @since 1.4.0
   */
  public final class ClusterParamFlowRuleManager {
  
      /**
       * The default cluster parameter flow rule property supplier that creates a new
       * dynamic property for a specific namespace to manually do rule management.
       */
      public static final Function<String, SentinelProperty<List<ParamFlowRule>>> DEFAULT_PROPERTY_SUPPLIER =
          new Function<String, SentinelProperty<List<ParamFlowRule>>>() {
              @Override
              public SentinelProperty<List<ParamFlowRule>> apply(String namespace) {
                  return new DynamicSentinelProperty<>();
              }
          };
  
      /**
       * (id, clusterParamRule)
       */
      private static final Map<Long, ParamFlowRule> PARAM_RULES = new ConcurrentHashMap<>();
      /**
       * (namespace, [flowId...])
       */
      private static final Map<String, Set<Long>> NAMESPACE_FLOW_ID_MAP = new ConcurrentHashMap<>();
      /**
       * (flowId, namespace)
       */
      private static final Map<Long, String> FLOW_NAMESPACE_MAP = new ConcurrentHashMap<>();
  
      /**
       * (namespace, property-listener wrapper)
       */
      private static final Map<String, NamespaceFlowProperty<ParamFlowRule>> PROPERTY_MAP = new ConcurrentHashMap<>();
      /**
       * Cluster parameter flow rule property supplier for a specific namespace.
       */
      private static volatile Function<String, SentinelProperty<List<ParamFlowRule>>> propertySupplier
          = DEFAULT_PROPERTY_SUPPLIER;
  
      private static final Object UPDATE_LOCK = new Object();
  
      static {
          initDefaultProperty();
      }
  
      private static void initDefaultProperty() {
          SentinelProperty<List<ParamFlowRule>> defaultProperty = new DynamicSentinelProperty<>();
          String defaultNamespace = ServerConstants.DEFAULT_NAMESPACE;
          registerPropertyInternal(defaultNamespace, defaultProperty);
      }
  
      public static void setPropertySupplier(
          Function<String, SentinelProperty<List<ParamFlowRule>>> propertySupplier) {
          ClusterParamFlowRuleManager.propertySupplier = propertySupplier;
      }
  
      public static String getNamespace(long flowId) {
          return FLOW_NAMESPACE_MAP.get(flowId);
      }
  
      /**
       * Listen to the {@link SentinelProperty} for cluster {@link ParamFlowRule}s.
       * The property is the source of cluster {@link ParamFlowRule}s for a specific namespace.
       *
       * @param namespace namespace to register
       */
      public static void register2Property(String namespace) {
          AssertUtil.notEmpty(namespace, "namespace cannot be empty");
          if (propertySupplier == null) {
              RecordLog.warn(
                  "[ClusterParamFlowRuleManager] Cluster param rule property supplier is absent, cannot register "
                      + "property");
              return;
          }
          SentinelProperty<List<ParamFlowRule>> property = propertySupplier.apply(namespace);
          if (property == null) {
              RecordLog.warn(
                  "[ClusterParamFlowRuleManager] Wrong created property from cluster param rule property supplier, "
                      + "ignoring");
              return;
          }
          synchronized (UPDATE_LOCK) {
              RecordLog.info("[ClusterParamFlowRuleManager] Registering new property to cluster param rule manager"
                  + " for namespace <{}>", namespace);
              registerPropertyInternal(namespace, property);
          }
      }
  
      public static void registerPropertyIfAbsent(String namespace) {
          AssertUtil.notEmpty(namespace, "namespace cannot be empty");
          if (!PROPERTY_MAP.containsKey(namespace)) {
              synchronized (UPDATE_LOCK) {
                  if (!PROPERTY_MAP.containsKey(namespace)) {
                      register2Property(namespace);
                  }
              }
          }
      }
  
      private static void registerPropertyInternal(/*@NonNull*/ String namespace, /*@Valid*/
                                                                SentinelProperty<List<ParamFlowRule>> property) {
          NamespaceFlowProperty<ParamFlowRule> oldProperty = PROPERTY_MAP.get(namespace);
          if (oldProperty != null) {
              oldProperty.getProperty().removeListener(oldProperty.getListener());
          }
          PropertyListener<List<ParamFlowRule>> listener = new ParamRulePropertyListener(namespace);
          property.addListener(listener);
          PROPERTY_MAP.put(namespace, new NamespaceFlowProperty<>(namespace, property, listener));
          Set<Long> flowIdSet = NAMESPACE_FLOW_ID_MAP.get(namespace);
          if (flowIdSet == null) {
              resetNamespaceFlowIdMapFor(namespace);
          }
      }
  
      public static void removeProperty(String namespace) {
          AssertUtil.notEmpty(namespace, "namespace cannot be empty");
          synchronized (UPDATE_LOCK) {
              NamespaceFlowProperty<ParamFlowRule> property = PROPERTY_MAP.get(namespace);
              if (property != null) {
                  property.getProperty().removeListener(property.getListener());
                  PROPERTY_MAP.remove(namespace);
              }
              RecordLog.info("[ClusterParamFlowRuleManager] Removing property from cluster flow rule manager"
                  + " for namespace <{}>", namespace);
          }
      }
  
      private static void removePropertyListeners() {
          for (NamespaceFlowProperty<ParamFlowRule> property : PROPERTY_MAP.values()) {
              property.getProperty().removeListener(property.getListener());
          }
      }
  
      private static void restorePropertyListeners() {
          for (NamespaceFlowProperty<ParamFlowRule> p : PROPERTY_MAP.values()) {
              p.getProperty().removeListener(p.getListener());
              p.getProperty().addListener(p.getListener());
          }
      }
  
      private static void resetNamespaceFlowIdMapFor(/*@Valid*/ String namespace) {
          NAMESPACE_FLOW_ID_MAP.put(namespace, new HashSet<Long>());
      }
  
      private static void clearAndResetRulesFor(/*@Valid*/ String namespace) {
          Set<Long> flowIdSet = NAMESPACE_FLOW_ID_MAP.get(namespace);
          if (flowIdSet != null && !flowIdSet.isEmpty()) {
              for (Long flowId : flowIdSet) {
                  PARAM_RULES.remove(flowId);
                  FLOW_NAMESPACE_MAP.remove(flowId);
              }
              flowIdSet.clear();
          } else {
              resetNamespaceFlowIdMapFor(namespace);
          }
      }
  
      private static void clearAndResetRulesConditional(/*@Valid*/ String namespace, Predicate<Long> predicate) {
          Set<Long> oldIdSet = NAMESPACE_FLOW_ID_MAP.get(namespace);
          if (oldIdSet != null && !oldIdSet.isEmpty()) {
              for (Long flowId : oldIdSet) {
                  if (predicate.test(flowId)) {
                      PARAM_RULES.remove(flowId);
                      FLOW_NAMESPACE_MAP.remove(flowId);
                      ClusterParamMetricStatistics.removeMetric(flowId);
                  }
              }
              oldIdSet.clear();
          }
      }
  
      public static ParamFlowRule getParamRuleById(Long id) {
          if (!ClusterRuleUtil.validId(id)) {
              return null;
          }
          return PARAM_RULES.get(id);
      }
  
      public static Set<Long> getFlowIdSet(String namespace) {
          if (StringUtil.isEmpty(namespace)) {
              return new HashSet<>();
          }
          Set<Long> set = NAMESPACE_FLOW_ID_MAP.get(namespace);
          if (set == null) {
              return new HashSet<>();
          }
          return new HashSet<>(set);
      }
  
      public static List<ParamFlowRule> getAllParamRules() {
          return new ArrayList<>(PARAM_RULES.values());
      }
  
      /**
       * Get all cluster parameter flow rules within a specific namespace.
       *
       * @param namespace a valid namespace
       * @return cluster parameter flow rules within the provided namespace
       */
      public static List<ParamFlowRule> getParamRules(String namespace) {
          if (StringUtil.isEmpty(namespace)) {
              return new ArrayList<>();
          }
          List<ParamFlowRule> rules = new ArrayList<>();
          Set<Long> flowIdSet = NAMESPACE_FLOW_ID_MAP.get(namespace);
          if (flowIdSet == null || flowIdSet.isEmpty()) {
              return rules;
          }
          for (Long flowId : flowIdSet) {
              ParamFlowRule rule = PARAM_RULES.get(flowId);
              if (rule != null) {
                  rules.add(rule);
              }
          }
          return rules;
      }
  
      /**
       * Load parameter flow rules for a specific namespace. The former rules of the namespace will be replaced.
       *
       * @param namespace a valid namespace
       * @param rules rule list
       */
      public static void loadRules(String namespace, List<ParamFlowRule> rules) {
          AssertUtil.notEmpty(namespace, "namespace cannot be empty");
          NamespaceFlowProperty<ParamFlowRule> property = PROPERTY_MAP.get(namespace);
          if (property != null) {
              property.getProperty().updateValue(rules);
          }
      }
  
      /**
       * Get connected count for associated namespace of given {@code flowId}.
       *
       * @param flowId existing rule ID
       * @return connected count
       */
      public static int getConnectedCount(long flowId) {
          if (flowId <= 0) {
              return 0;
          }
          String namespace = FLOW_NAMESPACE_MAP.get(flowId);
          if (namespace == null) {
              return 0;
          }
          return ConnectionManager.getConnectedCount(namespace);
      }
  
      private static class ParamRulePropertyListener implements PropertyListener<List<ParamFlowRule>> {
  
          private final String namespace;
  
          public ParamRulePropertyListener(String namespace) {
              this.namespace = namespace;
          }
  
          @Override
          public void configLoad(List<ParamFlowRule> conf) {
              applyClusterParamRules(conf, namespace);
              RecordLog.info("[ClusterParamFlowRuleManager] Cluster parameter rules loaded for namespace <{}>: {}",
                  namespace, PARAM_RULES);
          }
  
          @Override
          public void configUpdate(List<ParamFlowRule> conf) {
              applyClusterParamRules(conf, namespace);
              RecordLog.info("[ClusterParamFlowRuleManager] Cluster parameter rules received for namespace <{}>: {}",
                  namespace, PARAM_RULES);
          }
      }
  
      private static void applyClusterParamRules(List<ParamFlowRule> list, /*@Valid*/ String namespace) {
          if (list == null || list.isEmpty()) {
              clearAndResetRulesFor(namespace);
              return;
          }
          final ConcurrentHashMap<Long, ParamFlowRule> ruleMap = new ConcurrentHashMap<>();
  
          Set<Long> flowIdSet = new HashSet<>();
  
          for (ParamFlowRule rule : list) {
              if (!rule.isClusterMode()) {
                  continue;
              }
              if (!ParamFlowRuleUtil.isValidRule(rule)) {
                  RecordLog.warn(
                      "[ClusterParamFlowRuleManager] Ignoring invalid param flow rule when loading new flow rules: "
                          + rule);
                  continue;
              }
              if (StringUtil.isBlank(rule.getLimitApp())) {
                  rule.setLimitApp(RuleConstant.LIMIT_APP_DEFAULT);
              }
  
              ParamFlowRuleUtil.fillExceptionFlowItems(rule);
  
              ParamFlowClusterConfig clusterConfig = rule.getClusterConfig();
              // Flow id should not be null after filtered.
              Long flowId = clusterConfig.getFlowId();
              if (flowId == null) {
                  continue;
              }
              ruleMap.put(flowId, rule);
              FLOW_NAMESPACE_MAP.put(flowId, namespace);
              flowIdSet.add(flowId);
  
              // Prepare cluster parameter metric from valid rule ID.
              ClusterParamMetricStatistics.putMetricIfAbsent(flowId,
                  new ClusterParamMetric(clusterConfig.getSampleCount(), clusterConfig.getWindowIntervalMs()));
          }
  
          // Cleanup unused cluster parameter metrics.
          clearAndResetRulesConditional(namespace, new Predicate<Long>() {
              @Override
              public boolean test(Long flowId) {
                  return !ruleMap.containsKey(flowId);
              }
          });
  
          PARAM_RULES.putAll(ruleMap);
          NAMESPACE_FLOW_ID_MAP.put(namespace, flowIdSet);
      }
  
      private ClusterParamFlowRuleManager() {}
  }
  