Automatic fault tolerance module change to extension mode. (sofastac…

…k#501)

core/api/src/main/java/com/alipay/sofa/rpc/common/RpcOptions.java

@@ -509,4 +509,25 @@ public class RpcOptions {
      * Whether to close lookout collection.
      */
     public static final String LOOKOUT_COLLECT_DISABLE                  = "lookout.collect.disable";
+
+    /**
+     * Automatic fault tolerance regulator
+     */
+    public static final String AFT_REGULATOR                            = "aft.regulator";
+    /**
+     * Automatic fault tolerance regulation strategy
+     */
+    public static final String AFT_REGULATION_STRATEGY                  = "aft.regulation.strategy";
+    /**
+     * Automatic fault tolerance recover strategy
+     */
+    public static final String AFT_RECOVER_STRATEGY                     = "aft.recover.strategy";
+    /**
+     * Automatic fault tolerance degrade strategy
+     */
+    public static final String AFT_DEGRADE_STRATEGY                     = "aft.degrade.strategy";
+    /**
+     * Automatic fault tolerance measure strategy
+     */
+    public static final String AFT_MEASURE_STRATEGY                     = "aft.measure.strategy";
 }

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/DegradeStrategy.java

@@ -23,7 +23,7 @@
  * 
  * @author <a href="mailto:[email protected]">liangen</a>
  */
-@Extensible
+@Extensible(singleton = false)
 public interface DegradeStrategy {
 
     /**

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/MeasureResultDetail.java

@@ -32,6 +32,7 @@ public class MeasureResultDetail {
     private double                        averageAbnormalRate;
     private double                        leastAbnormalRateMultiple;
     private boolean                       recoveredOriginWeight;
+    private transient boolean             logOnly;
 
     /**
      * Instantiates a new Measure result detail.
@@ -187,4 +188,24 @@ public boolean isRecoveredOriginWeight() {
     public void setRecoveredOriginWeight(boolean recoveredOriginWeight) {
         this.recoveredOriginWeight = recoveredOriginWeight;
     }
+
+    /**
+     * Is log only boolean.
+     *
+     * @return the boolean
+     */
+    public boolean isLogOnly() {
+        return logOnly;
+    }
+
+    /**
+     * Sets log only.
+     *
+     * @param logOnly the log only
+     * @return the log only
+     */
+    public MeasureResultDetail setLogOnly(boolean logOnly) {
+        this.logOnly = logOnly;
+        return this;
+    }
 }

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/MeasureStrategy.java

@@ -24,7 +24,7 @@
  * @author <a href="mailto:[email protected]">liangen</a>
  * @author <a href="mailto:[email protected]">GengZhang</a>
  */
-@Extensible
+@Extensible(singleton = false)
 public interface MeasureStrategy {
 
     /**

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/RecoverStrategy.java

@@ -23,7 +23,7 @@
  * 
  * @author <a href="mailto:[email protected]">liangen</a>
  */
-@Extensible
+@Extensible(singleton = false)
 public interface RecoverStrategy {
 
     /**

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/RegulationStrategy.java

@@ -23,7 +23,7 @@
  *
  * @author <a href="mailto:[email protected]">liangen</a>
  */
-@Extensible
+@Extensible(singleton = false)
 public interface RegulationStrategy {
 
     /**

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/Regulator.java

@@ -16,11 +16,14 @@
  */
 package com.alipay.sofa.rpc.client.aft;
 
+import com.alipay.sofa.rpc.ext.Extensible;
+
 /**
  * Regulator of adaptive fault tolerance
  *
  * @author <a href="mailto:[email protected]">GengZhang</a>
  */
+@Extensible(singleton = false)
 public interface Regulator {
 
     /**

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/impl/TimeWindowRegulator.java

@@ -30,9 +30,13 @@
 import com.alipay.sofa.rpc.client.aft.RecoverStrategy;
 import com.alipay.sofa.rpc.client.aft.RegulationStrategy;
 import com.alipay.sofa.rpc.client.aft.Regulator;
+import com.alipay.sofa.rpc.common.RpcConfigs;
+import com.alipay.sofa.rpc.common.RpcOptions;
 import com.alipay.sofa.rpc.common.struct.NamedThreadFactory;
 import com.alipay.sofa.rpc.common.struct.ScheduledService;
 import com.alipay.sofa.rpc.common.utils.ThreadPoolUtils;
+import com.alipay.sofa.rpc.ext.Extension;
+import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
 import com.alipay.sofa.rpc.log.LogCodes;
 import com.alipay.sofa.rpc.log.Logger;
 import com.alipay.sofa.rpc.log.LoggerFactory;
@@ -51,6 +55,7 @@
  * @author <a href="mailto:[email protected]">liangen</a>
  * @author <a href="mailto:[email protected]">GengZhang</a>
  */
+@Extension("timeWindow")
 public class TimeWindowRegulator implements Regulator {
 
     /** Logger for this class */
@@ -96,15 +101,12 @@ public class TimeWindowRegulator implements Regulator {
     private RegulationStrategy                       regulationStrategy;
 
     /**
-     * 降级策略: 调整权重 
+     * 降级策略: 例如调整权重 
      */
-    private DegradeStrategy                          weightDegradeStrategy;
-    /**
-     * 降级策略: 只打印日志 
-     */
-    private DegradeStrategy                          logDegradeStrategy;
+    private DegradeStrategy                          degradeStrategy;
+
     /**
-     * 恢复策略：调整权重 
+     * 恢复策略：例如调整权重 
      */
     private RecoverStrategy                          recoverStrategy;
 
@@ -115,11 +117,21 @@ public class TimeWindowRegulator implements Regulator {
 
     @Override
     public void init() {
-        measureStrategy = new ServiceHorizontalMeasureStrategy();
-        regulationStrategy = new ServiceHorizontalRegulationStrategy();
-        weightDegradeStrategy = new WeightDegradeStrategy();
-        logDegradeStrategy = new LogPrintDegradeStrategy();
-        recoverStrategy = new WeightRecoverStrategy();
+        String measureStrategyAlias = RpcConfigs
+            .getOrDefaultValue(RpcOptions.AFT_MEASURE_STRATEGY, "serviceHorizontal");
+        String regulationStrategyAlias = RpcConfigs.getOrDefaultValue(RpcOptions.AFT_REGULATION_STRATEGY,
+            "serviceHorizontal");
+        String degradeStrategyAlias = RpcConfigs.getOrDefaultValue(RpcOptions.AFT_DEGRADE_STRATEGY, "weight");
+        String recoverStrategyAlias = RpcConfigs.getOrDefaultValue(RpcOptions.AFT_RECOVER_STRATEGY, "weight");
+
+        measureStrategy = ExtensionLoaderFactory.getExtensionLoader(MeasureStrategy.class).getExtension(
+            measureStrategyAlias);
+        regulationStrategy = ExtensionLoaderFactory.getExtensionLoader(RegulationStrategy.class).getExtension(
+            regulationStrategyAlias);
+        degradeStrategy = ExtensionLoaderFactory.getExtensionLoader(DegradeStrategy.class).getExtension(
+            degradeStrategyAlias);
+        recoverStrategy = ExtensionLoaderFactory.getExtensionLoader(RecoverStrategy.class).getExtension(
+            recoverStrategyAlias);
 
         InvocationStatFactory.addListener(listener);
     }
@@ -151,8 +163,7 @@ public void destroy() {
         // release strategy
         measureStrategy = null;
         regulationStrategy = null;
-        weightDegradeStrategy = null;
-        logDegradeStrategy = null;
+        degradeStrategy = null;
         recoverStrategy = null;
     }
 
@@ -222,10 +233,11 @@ void doRegulate(MeasureResultDetail measureResultDetail) {
 
             boolean isDegradeEffective = regulationStrategy.isDegradeEffective(measureResultDetail);
             if (isDegradeEffective) {
+                measureResultDetail.setLogOnly(false);
                 if (measureState.equals(MeasureState.ABNORMAL)) {
                     boolean isReachMaxDegradeIpCount = regulationStrategy.isReachMaxDegradeIpCount(measureResultDetail);
                     if (!isReachMaxDegradeIpCount) {
-                        weightDegradeStrategy.degrade(measureResultDetail);
+                        degradeStrategy.degrade(measureResultDetail);
                     } else {
                         String appName = measureResult.getMeasureModel().getAppName();
                         if (LOGGER.isInfoEnabled(appName)) {
@@ -243,8 +255,9 @@ void doRegulate(MeasureResultDetail measureResultDetail) {
                     //没有被降级过，因此不需要被恢复。
                 }
             } else {
+                measureResultDetail.setLogOnly(true);
                 if (measureState.equals(MeasureState.ABNORMAL)) {
-                    logDegradeStrategy.degrade(measureResultDetail);
+                    degradeStrategy.degrade(measureResultDetail);
                     String appName = measureResult.getMeasureModel().getAppName();
                     if (LOGGER.isInfoEnabled(appName)) {
                         LOGGER.infoWithApp(appName, LogCodes.getLog(LogCodes.INFO_REGULATION_ABNORMAL_NOT_DEGRADE,

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/client/aft/impl/WeightDegradeStrategy.java

@@ -39,8 +39,13 @@ public class WeightDegradeStrategy extends LogPrintDegradeStrategy {
 
     @Override
     public void degrade(MeasureResultDetail measureResultDetail) {
+
         super.degrade(measureResultDetail);
 
+        if (measureResultDetail.isLogOnly()) {
+            return;
+        }
+
         InvocationStatDimension statDimension = measureResultDetail.getInvocationStatDimension();
         String appName = statDimension.getAppName();
 

extension-impl/fault-tolerance/src/main/java/com/alipay/sofa/rpc/module/FaultToleranceModule.java

@@ -18,11 +18,14 @@
 
 import com.alipay.sofa.rpc.client.aft.Regulator;
 import com.alipay.sofa.rpc.client.aft.impl.TimeWindowRegulator;
+import com.alipay.sofa.rpc.common.RpcConfigs;
+import com.alipay.sofa.rpc.common.RpcOptions;
 import com.alipay.sofa.rpc.event.ClientAsyncReceiveEvent;
 import com.alipay.sofa.rpc.event.ClientSyncReceiveEvent;
 import com.alipay.sofa.rpc.event.EventBus;
 import com.alipay.sofa.rpc.event.FaultToleranceSubscriber;
 import com.alipay.sofa.rpc.ext.Extension;
+import com.alipay.sofa.rpc.ext.ExtensionLoaderFactory;
 
 /**
  * FaultToleranceModule
@@ -53,7 +56,8 @@ public void install() {
         EventBus.register(ClientSyncReceiveEvent.class, subscriber);
         EventBus.register(ClientAsyncReceiveEvent.class, subscriber);
 
-        regulator = new TimeWindowRegulator();
+        String regulatorAlias = RpcConfigs.getOrDefaultValue(RpcOptions.AFT_REGULATOR, "timeWindow");
+        regulator = ExtensionLoaderFactory.getExtensionLoader(Regulator.class).getExtension(regulatorAlias);
         regulator.init();
     }
 

extension-impl/fault-tolerance/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.aft.DegradeStrategy

@@ -0,0 +1,2 @@
+log=com.alipay.sofa.rpc.client.aft.impl.LogPrintDegradeStrategy
+weight=com.alipay.sofa.rpc.client.aft.impl.WeightDegradeStrategy

extension-impl/fault-tolerance/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.aft.MeasureStrategy

@@ -0,0 +1 @@
+serviceHorizontal=com.alipay.sofa.rpc.client.aft.impl.ServiceHorizontalMeasureStrategy

extension-impl/fault-tolerance/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.aft.RecoverStrategy

@@ -0,0 +1 @@
+weight=com.alipay.sofa.rpc.client.aft.impl.WeightRecoverStrategy

extension-impl/fault-tolerance/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.aft.RegulationStrategy

@@ -0,0 +1 @@
+serviceHorizontal=com.alipay.sofa.rpc.client.aft.impl.ServiceHorizontalRegulationStrategy

extension-impl/fault-tolerance/src/main/resources/META-INF/services/sofa-rpc/com.alipay.sofa.rpc.client.aft.Regulator

@@ -0,0 +1 @@
+timeWindow=com.alipay.sofa.rpc.client.aft.impl.TimeWindowRegulator