[SCB-2807]add new instance isolation feature based on instance-isolation-consumer (#3932)

Author: liubao68

handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/ConsumerInstanceIsolationHandler.java

@@ -17,19 +17,21 @@
 
 package org.apache.servicecomb.handler.governance;
 
+import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.function.Supplier;
 
+import javax.ws.rs.core.Response.Status;
+
 import org.apache.servicecomb.core.Handler;
 import org.apache.servicecomb.core.Invocation;
 import org.apache.servicecomb.core.governance.MatchType;
+import org.apache.servicecomb.foundation.common.event.EventManager;
 import org.apache.servicecomb.foundation.common.utils.BeanUtils;
 import org.apache.servicecomb.governance.handler.InstanceIsolationHandler;
 import org.apache.servicecomb.governance.marker.GovernanceRequestExtractor;
-import org.apache.servicecomb.registry.api.MicroserviceKey;
-import org.apache.servicecomb.registry.api.event.MicroserviceInstanceChangedEvent;
-import org.apache.servicecomb.registry.api.event.ServiceCenterEventBus;
+import org.apache.servicecomb.governance.policy.CircuitBreakerPolicy;
 import org.apache.servicecomb.swagger.invocation.AsyncResponse;
 import org.apache.servicecomb.swagger.invocation.Response;
 import org.apache.servicecomb.swagger.invocation.exception.CommonExceptionData;
@@ -57,6 +59,12 @@ public void handle(Invocation invocation, AsyncResponse asyncResp) throws Except
     DecorateCompletionStage<Response> dcs = Decorators.ofCompletionStage(next);
     GovernanceRequestExtractor request = MatchType.createGovHttpRequest(invocation);
 
+    CircuitBreakerPolicy circuitBreakerPolicy = instanceIsolationHandler.matchPolicy(request);
+    if (circuitBreakerPolicy != null && circuitBreakerPolicy.isForceOpen()) {
+      asyncResp.consumerFail(new InvocationException(Status.SERVICE_UNAVAILABLE,
+          "Policy " + circuitBreakerPolicy.getName() + " forced open and deny requests"));
+      return;
+    }
     addCircuitBreaker(dcs, request);
 
     dcs.get().whenComplete((r, e) -> {
@@ -67,7 +75,7 @@ public void handle(Invocation invocation, AsyncResponse asyncResp) throws Except
 
       if (e instanceof CallNotPermittedException) {
         LOGGER.warn("instance isolation circuitBreaker is open by policy : {}", e.getMessage());
-        ServiceCenterEventBus.getEventBus().post(createMicroserviceInstanceChangedEvent(invocation));
+        EventManager.post(createInstanceIsolatedEvent(circuitBreakerPolicy, request));
         // return 503 so that consumer can retry
         asyncResp.complete(
             Response.failResp(new InvocationException(503, "instance isolation circuitBreaker is open.",
@@ -78,13 +86,10 @@ public void handle(Invocation invocation, AsyncResponse asyncResp) throws Except
     });
   }
 
-  private Object createMicroserviceInstanceChangedEvent(Invocation invocation) {
-    MicroserviceInstanceChangedEvent event = new MicroserviceInstanceChangedEvent();
-    MicroserviceKey key = new MicroserviceKey();
-    key.setAppId(invocation.getAppId());
-    key.setServiceName(invocation.getMicroserviceName());
-    event.setKey(key);
-    return event;
+  private Object createInstanceIsolatedEvent(CircuitBreakerPolicy circuitBreakerPolicy,
+      GovernanceRequestExtractor requestExtractor) {
+    return new InstanceIsolatedEvent(requestExtractor.instanceId(),
+        Duration.parse(circuitBreakerPolicy.getWaitDurationInOpenState()));
   }
 
   private void addCircuitBreaker(DecorateCompletionStage<Response> dcs, GovernanceRequestExtractor request) {

handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/InstanceIsolatedEvent.java

@@ -0,0 +1,38 @@
+/*
+ * 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.apache.servicecomb.handler.governance;
+
+import java.time.Duration;
+
+public class InstanceIsolatedEvent {
+  private final String instanceId;
+
+  private final Duration waitDurationInHalfOpenState;
+
+  public InstanceIsolatedEvent(String instanceId, Duration waitDurationInHalfOpenState) {
+    this.instanceId = instanceId;
+    this.waitDurationInHalfOpenState = waitDurationInHalfOpenState;
+  }
+
+  public String getInstanceId() {
+    return instanceId;
+  }
+
+  public Duration getWaitDurationInHalfOpenState() {
+    return waitDurationInHalfOpenState;
+  }
+}

handlers/handler-governance/src/main/java/org/apache/servicecomb/handler/governance/InstanceIsolationDiscoveryFilter.java

@@ -0,0 +1,120 @@
+/*
+ * 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.apache.servicecomb.handler.governance;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Map.Entry;
+
+import org.apache.servicecomb.foundation.common.concurrent.ConcurrentHashMapEx;
+import org.apache.servicecomb.foundation.common.event.EventManager;
+import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
+import org.apache.servicecomb.registry.discovery.DiscoveryContext;
+import org.apache.servicecomb.registry.discovery.DiscoveryFilter;
+import org.apache.servicecomb.registry.discovery.DiscoveryTreeNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.eventbus.Subscribe;
+import com.netflix.config.DynamicPropertyFactory;
+
+public class InstanceIsolationDiscoveryFilter implements DiscoveryFilter {
+  private static final Logger LOGGER = LoggerFactory.getLogger(InstanceIsolationDiscoveryFilter.class);
+
+  private static final String KEY_ISOLATED = "isolated";
+
+  private final Object lock = new Object();
+
+  private final Map<String, Long> isolatedInstances = new ConcurrentHashMapEx<>();
+
+  public InstanceIsolationDiscoveryFilter() {
+    EventManager.register(this);
+  }
+
+  @Subscribe
+  public void onInstanceIsolatedEvent(InstanceIsolatedEvent event) {
+    synchronized (lock) {
+      for (Iterator<String> iterator = isolatedInstances.keySet().iterator(); iterator.hasNext(); ) {
+        Long duration = isolatedInstances.get(iterator.next());
+        if (System.currentTimeMillis() - duration > 0) {
+          iterator.remove();
+        }
+      }
+
+      isolatedInstances.put(event.getInstanceId(),
+          System.currentTimeMillis() + event.getWaitDurationInHalfOpenState().toMillis());
+      LOGGER.info("isolate instance {} for {}ms", event.getInstanceId(),
+          event.getWaitDurationInHalfOpenState().toMillis());
+    }
+  }
+
+  @Override
+  public boolean enabled() {
+    return DynamicPropertyFactory.getInstance().getBooleanProperty(
+        "servicecomb.loadbalance.filter.instance.isolation.enabled", true).get();
+  }
+
+  @Override
+  public int getOrder() {
+    return Short.MAX_VALUE - 1;
+  }
+
+  @Override
+  public boolean isGroupingFilter() {
+    return true;
+  }
+
+  @Override
+  public DiscoveryTreeNode discovery(DiscoveryContext context, DiscoveryTreeNode parent) {
+    Map<String, MicroserviceInstance> instances = parent.data();
+    if (isolatedInstances.isEmpty() || instances.isEmpty()) {
+      return parent;
+    }
+
+    boolean changed = false;
+    Map<String, MicroserviceInstance> result = new HashMap<>(instances.size());
+    for (Entry<String, MicroserviceInstance> item : instances.entrySet()) {
+      Long duration = isolatedInstances.get(item.getKey());
+      if (duration == null) {
+        result.put(item.getKey(), item.getValue());
+        continue;
+      }
+
+      if (System.currentTimeMillis() - duration < 0) {
+        changed = true;
+        continue;
+      }
+
+      synchronized (lock) {
+        isolatedInstances.remove(item.getKey());
+        LOGGER.info("try to recover instance {}", item.getKey());
+      }
+      result.put(item.getKey(), item.getValue());
+    }
+
+    if (!changed || result.size() == 0) {
+      return parent;
+    }
+
+    // Create new child. And all later DiscoveryFilter will re-calculate based on this result.
+    DiscoveryTreeNode child = new DiscoveryTreeNode().subName(parent, KEY_ISOLATED).data(result);
+    parent.child(KEY_ISOLATED, child);
+    return child;
+  }
+}

handlers/handler-governance/src/main/resources/META-INF/services/org.apache.servicecomb.registry.discovery.DiscoveryFilter

@@ -15,3 +15,4 @@
 # limitations under the License.
 #
 
+org.apache.servicecomb.handler.governance.InstanceIsolationDiscoveryFilter

handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/event/IsolationServerEvent.java

@@ -20,7 +20,7 @@
 import org.apache.servicecomb.core.Invocation;
 import org.apache.servicecomb.foundation.common.event.AlarmEvent;
 import org.apache.servicecomb.loadbalance.ServiceCombServerStats;
-import org.apache.servicecomb.loadbalance.filterext.IsolationDiscoveryFilter;
+import org.apache.servicecomb.loadbalance.filterext.IsolationServerListFilterExt;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
 
 public class IsolationServerEvent extends AlarmEvent {
@@ -52,7 +52,7 @@ public class IsolationServerEvent extends AlarmEvent {
 
   public IsolationServerEvent(Invocation invocation, MicroserviceInstance instance,
       ServiceCombServerStats serverStats,
-      IsolationDiscoveryFilter.Settings settings, Type type, Endpoint endpoint) {
+      IsolationServerListFilterExt.Settings settings, Type type, Endpoint endpoint) {
     super(type);
     this.microserviceName = invocation.getMicroserviceName();
     this.endpoint = endpoint;

…/filterext/IsolationDiscoveryFilter.java …terext/IsolationServerListFilterExt.javahandlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/filterext/IsolationDiscoveryFilter.java renamed to handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/filterext/IsolationServerListFilterExt.java handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/filterext/IsolationDiscoveryFilter.java renamed to handlers/handler-loadbalance/src/main/java/org/apache/servicecomb/loadbalance/filterext/IsolationServerListFilterExt.java

@@ -39,9 +39,9 @@
 /**
  * Isolate instances by error metrics
  */
-public class IsolationDiscoveryFilter implements ServerListFilterExt {
+public class IsolationServerListFilterExt implements ServerListFilterExt {
 
-  private static final Logger LOGGER = LoggerFactory.getLogger(IsolationDiscoveryFilter.class);
+  private static final Logger LOGGER = LoggerFactory.getLogger(IsolationServerListFilterExt.class);
 
   private final DynamicBooleanProperty emptyProtection = DynamicPropertyFactory.getInstance()
       .getBooleanProperty(EMPTY_INSTANCE_PROTECTION, false);
@@ -65,7 +65,7 @@ public int getOrder() {
     return ORDER_ISOLATION;
   }
 
-  public IsolationDiscoveryFilter() {
+  public IsolationServerListFilterExt() {
     emptyProtection.addCallback(() -> {
       boolean newValue = emptyProtection.get();
       LOGGER.info("{} changed from {} to {}", EMPTY_INSTANCE_PROTECTION, emptyProtection, newValue);

handlers/handler-loadbalance/src/main/resources/META-INF/services/org.apache.servicecomb.loadbalance.ServerListFilterExt

@@ -15,5 +15,5 @@
 # limitations under the License.
 #
 
-org.apache.servicecomb.loadbalance.filterext.IsolationDiscoveryFilter
+org.apache.servicecomb.loadbalance.filterext.IsolationServerListFilterExt
 org.apache.servicecomb.loadbalance.filterext.ZoneAwareDiscoveryFilter

…filter/IsolationDiscoveryFilterTest.java …er/IsolationServerListFilterExtTest.javahandlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/filter/IsolationDiscoveryFilterTest.java renamed to handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/filter/IsolationServerListFilterExtTest.java handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/filter/IsolationDiscoveryFilterTest.java renamed to handlers/handler-loadbalance/src/test/java/org/apache/servicecomb/loadbalance/filter/IsolationServerListFilterExtTest.java

@@ -28,7 +28,7 @@
 import org.apache.servicecomb.loadbalance.ServiceCombServer;
 import org.apache.servicecomb.loadbalance.ServiceCombServerStats;
 import org.apache.servicecomb.loadbalance.TestServiceCombServerStats;
-import org.apache.servicecomb.loadbalance.filterext.IsolationDiscoveryFilter;
+import org.apache.servicecomb.loadbalance.filterext.IsolationServerListFilterExt;
 import org.apache.servicecomb.registry.api.registry.MicroserviceInstance;
 import org.apache.servicecomb.registry.cache.CacheEndpoint;
 import org.junit.jupiter.api.AfterEach;
@@ -42,9 +42,9 @@
 import mockit.Mocked;
 
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
-public class IsolationDiscoveryFilterTest {
+public class IsolationServerListFilterExtTest {
 
-  private IsolationDiscoveryFilter filter;
+  private IsolationServerListFilterExt filter;
 
   private List<ServiceCombServer> servers;
 
@@ -73,7 +73,7 @@ public void before() {
       ServiceCombLoadBalancerStats.INSTANCE.getServiceCombServerStats(serviceCombServer);
     }
 
-    filter = new IsolationDiscoveryFilter();
+    filter = new IsolationServerListFilterExt();
     TestServiceCombServerStats.releaseTryingChance();
   }