How does Sentinel intercept abnormal traffic?

[[342079]]

When you use electricity at home, you must have experienced "tripping". This "gate" is used to protect our electricity safety when the power exceeds the load. It is also called a "circuit breaker" and has a resounding English name - CircuitBreaker.

Just like electricity safety, you and I should all be familiar with "current limiting", "downgrading", "fusing"... In order to avoid being overwhelmed by abnormal traffic, various software, systems, and Internet applications we develop also need a circuit breaker.

In Spring applications, it is convenient to use circuit breakers. We can use Spring Cloud CircuitBreaker.

What is Spring Cloud Circuit Breaker? If you are familiar with Spring, you can probably guess what it is. Similar to Spring Data JPA, Spring has come up with an abstract, standard API. This time, it abstracts the "circuit breaker" for downgrade circuit breaking. With this layer, the specific implementation can be easily replaced, and the code we use basically has to be changed to zero.

Let's first get a preliminary impression from the official Demo:

 @RestController
 public class DemoController {
  private CircuitBreakerFactory circuitBreakerFactory;
  private HttpBinService httpBin;
 public DemoController(CircuitBreakerFactory circuitBreakerFactory, HttpBinService httpBinService) {
    this.circuitBreakerFactory = circuitBreakerFactory;
    this.httpBin = httpBinService;
 }
  @GetMapping( "/delay/{seconds}" )
 public Map delay(@PathVariable int seconds) {
 return circuitBreakerFactory. create ( "delay" ).run(httpBin.delaySuppplier(seconds), t -> {
      Map<String, String> fallback = new HashMap<>();
      fallback.put( "hello" , "world" );
 return fallback;
 });
 }
 }

Thousands of words, summed up in one sentence circuitBreakerFactory.create("delay").run()

Because it is abstract, there are many corresponding implementations.

Currently supported implementations are:

• Hystrix
• Resilience4j
• Sentinel
• Spring Retry

Abstraction is equivalent to setting a standard. Like JDBC, no matter we change the database to MySQL, Oracle or SQLite, the interface and other non-specific code do not need to be changed. The same is true for circuit breakers.

The circuit breaker factory here has standard creation methods. How the circuit breaker implementation intercepts and degrades when executing business logic here can be left to the specific implementation to complete.

This time, we take the open source Sentinel as an example to see how they block abnormal traffic.

First of all, because it is Spring Cloud, it will also be based on Spring Boot's Autoconfiguration. The following is the configuration class, where we can see that a factory is generated.

 public class SentinelCircuitBreakerAutoConfiguration {
 @Bean
  @ConditionalOnMissingBean(CircuitBreakerFactory.class)
 public CircuitBreakerFactory sentinelCircuitBreakerFactory() {
 return new SentinelCircuitBreakerFactory();
 }
 }

When we actually execute the logic in the code, what is created?

It is a circuit breaker CircuitBreaker, used to execute code.

 public interface CircuitBreaker { 
 
 default <T> T run(Supplier<T> toRun) {
 return run(toRun, throwable -> {
      throw new NoFallbackAvailableException( "No fallback available." , throwable);
 });
 };
  <T> T run(Supplier<T> toRun, Function <Throwable, T> fallback);
 }

Contains two execution methods, and can specify fallback logic when needed. Specifically for Sentinel:

 public CircuitBreaker create (String id) {
    SentinelConfigBuilder.SentinelCircuitBreakerConfiguration conf = getConfigurations()
        .computeIfAbsent(id, defaultConfiguration);
 return new SentinelCircuitBreaker(id, conf.getEntryType(), conf.getRules());
 }

You will see that a SentinelCircuitBreaker is created. Our business logic will be executed in this circuit breaker, and the run method is the stage for each specific implementation.

 @Override
 public <T> T run(Supplier<T> toRun, Function <Throwable, T> fallback) {
 Entry entry = null ;
 try {
      entry = SphU.entry(resourceName, entryType);
      // If the SphU.entry() does not throw `BlockException`, it means that the
 // request can pass.
 return toRun.get();
 }
    catch (BlockException ex) {
      // SphU.entry() may throw BlockException which indicates that
      // the request was rejected (flow control or circuit breaking triggered).
      // So it should not be counted as the business exception.
 return fallback.apply(ex);
 }
 catch (Exception ex) {
      // For other kinds of exceptions, we'll trace the exception count via
 // Tracer.trace(ex).
 Tracer.trace(ex);
 return fallback.apply(ex);
 }
 finally {
      // Guarantee the invocation has been completed.
 if (entry != null ) {
 entry.exit();
 }
 }
 }

OK, so far, Spring Cloud CircuitBreaker has been presented. Other details are put in the specific implementation "box". Now let's open this box.

Sentinel is a circuit breaker and downgrade framework. The official introduction is as follows:

The high-availability flow control component for distributed service architecture mainly takes flow as the entry point, and helps users ensure the stability of microservices from multiple dimensions such as flow control, circuit breaking and degradation, and system adaptive protection.

This screenshot of the code on the official website succinctly explains how it works.

It stands in front of the business code, and when something happens, it will rush to it first. Only after it can pass, the business logic will be followed. It is really similar to various levels.

In the run method of CircuitBreaker above, we must have noticed this sentence

 entry = SphU.entry(resourceName, entryType);

This is the secret of all interceptions.

Whether we use the CircuitBreaker method mentioned above, the @SentinelResource annotation method, or the Interceptor method, there is no essential difference. The only difference is the trigger point. In the end, it is all done through SphU.

Since it is an interception, it must be stopped and checked in one way or another.

During the actual inspection, the core codes in the entry are as follows:

 Entry entryWithPriority(ResourceWrapper resourceWrapper, ...)
        throws BlockException {
        ProcessorSlot<Object> chain = lookProcessChain(resourceWrapper);
        Entry e = new CtEntry(resourceWrapper, chain, context);
 try {
            chain.entry(context, resourceWrapper,...);
        } catch (BlockException e1) {
            e.exit( count , args);
 throw e1;
 }
 return e;
 }

Note that ProcessorSlot chain = lookProcessChain(resourceWrapper); will be initialized when a request comes in for processing. If the processing chain is not initialized, various first and next settings will be set, and subsequent requests will be processed according to this. All slots that need to be intercepted will be added to this chain, and then the slots in the chain will be executed one by one. Similar to Servlet Filter.

What's added to the chain?

 public class HotParamSlotChainBuilder implements SlotChainBuilder {
 public ProcessorSlotChain build() {
        ProcessorSlotChain chain = new DefaultProcessorSlotChain();
        chain.addLast(new NodeSelectorSlot());
        chain.addLast(new ClusterBuilderSlot());
        chain.addLast(new LogSlot());
        chain.addLast(new StatisticSlot());
        chain.addLast(new ParamFlowSlot());
        chain.addLast(new SystemSlot());
        chain.addLast(new AuthoritySlot());
        chain.addLast(new FlowSlot());
        chain.addLast(new DegradeSlot());
 return chain;
 }

Initially, first points to an anonymous inner class. These added slots will be used as the next in the chain each time addLast is called.

 AbstractLinkedProcessorSlot<?> end = first ; 
 
 @Override
 public void addFirst(AbstractLinkedProcessorSlot<?> protocolProcessor) {
        protocolProcessor.setNext( first .getNext());
 first .setNext(protocolProcessor);
 if ( end == first ) {
 end = protocolProcessor;
 }
 }
 @Override
 public void addLast(AbstractLinkedProcessorSlot<?> protocolProcessor) {
 end .setNext(protocolProcessor);
 end = protocolProcessor;
 }

Each slot has its own specific use. After processing its own logic, it will trigger the execution of the next slot through fireEntry.

Giving you a long thread call stack would make it too obvious:

 java.lang.Thread.State: RUNNABLE
 at com.alibaba.csp.sentinel.slots.block.flow.FlowSlot.checkFlow(FlowSlot.java:168)
 at com.alibaba.csp.sentinel.slots.block.flow.FlowSlot.entry(FlowSlot.java:161)
 at com.alibaba.csp.sentinel.slots.block.flow.FlowSlot.entry(FlowSlot.java:139)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot.entry(AuthoritySlot.java:39)
 at com.alibaba.csp.sentinel.slots.block.authority.AuthoritySlot.entry(AuthoritySlot.java:33)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.system.SystemSlot.entry(SystemSlot.java:36)
 at com.alibaba.csp.sentinel.slots.system.SystemSlot.entry(SystemSlot.java:30)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowSlot.entry(ParamFlowSlot.java:39)
 at com.alibaba.csp.sentinel.slots.block.flow.param.ParamFlowSlot.entry(ParamFlowSlot.java:33)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.statistic.StatisticSlot.entry(StatisticSlot.java:57)
 at com.alibaba.csp.sentinel.slots.statistic.StatisticSlot.entry(StatisticSlot.java:50)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.logger.LogSlot.entry(LogSlot.java:35)
 at com.alibaba.csp.sentinel.slots.logger.LogSlot.entry(LogSlot.java:29)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot.entry(ClusterBuilderSlot.java:101)
 at com.alibaba.csp.sentinel.slots.clusterbuilder.ClusterBuilderSlot.entry(ClusterBuilderSlot.java:47)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slots.nodeselector.NodeSelectorSlot.entry(NodeSelectorSlot.java:171)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.fireEntry(AbstractLinkedProcessorSlot.java:32)
 at com.alibaba.csp.sentinel.slotchain.DefaultProcessorSlotChain$1.entry(DefaultProcessorSlotChain.java:31)
 at com.alibaba.csp.sentinel.slotchain.AbstractLinkedProcessorSlot.transformEntry(AbstractLinkedProcessorSlot.java:40)
 at com.alibaba.csp.sentinel.slotchain.DefaultProcessorSlotChain.entry(DefaultProcessorSlotChain.java:75)
 at com.alibaba.csp.sentinel.CtSph.entryWithPriority(CtSph.java:148)
 at com.alibaba.csp.sentinel.CtSph.entryWithType(CtSph.java:347)
 at com.alibaba.csp.sentinel.CtSph.entryWithType(CtSph.java:340)
 at com.alibaba.csp.sentinel.SphU.entry(SphU.java:285)

There are three types of downgrades

Each type will be compared with the corresponding configuration item data. If it does not match, it will be interrupted. It cannot be interrupted forever. When will it be restored? According to the configured time window, a recovery thread will be started, and it will be scheduled to restore the interruption flag when the time comes.

 public boolean passCheck(Context context, DefaultNode node, int acquireCount, Object... args) {
 if (cut.get()) {
 return   false ;
 }
        ClusterNode clusterNode = ClusterBuilderSlot.getClusterNode(this.getResource());
        if (clusterNode == null ) {
 return   true ;
 }
        if (grade == RuleConstant.DEGRADE_GRADE_RT) {
 double rt = clusterNode.avgRt();
            if (rt < this. count ) {
                passCount.set (0);
 return   true ;
 }
            // Sentinel will degrade the service only if count exceeds.
            if (passCount.incrementAndGet() < rtSlowRequestAmount) {
 return   true ;
 }
        } else if (grade == RuleConstant.DEGRADE_GRADE_EXCEPTION_RATIO) {
 double exception = clusterNode.exceptionQps();
 double success = clusterNode.successQps();
 double total = clusterNode.totalQps();
            // If total amount is less than minRequestAmount, the request will pass.
            if (total < minRequestAmount) {
 return   true ;
 }
            // In the same aligned statistic time window,
            // "success" (aka. completed count ) = exception count + non-exception count (realSuccess)
 double realSuccess = success - exception;
            if (realSuccess <= 0 && exception < minRequestAmount) {
 return   true ;
 }
            if (exception / success < count ) {
 return   true ;
 }
        } else if (grade == RuleConstant.DEGRADE_GRADE_EXCEPTION_COUNT) {
 double exception = clusterNode.totalException();
            if (exception < count ) {
 return   true ;
 }
 }
        if (cut.compareAndSet( false , true )) {
            ResetTask resetTask = new ResetTask(this);
            pool.schedule(resetTask, timeWindow, TimeUnit.SECONDS);
 }
 return   false ;
 }

The recovery process does two things: 1. Set passCount to 0, 2. Restore the interrupt flag

The above introduces the interception and processing of requests. The core of these two, which are also our main configurations, are "flow control" and "downgrade". These two corresponding slots will be judged according to the configured "rules" when processing requests. For example, the time window, fuse time, etc., as well as the number of threads and QPS of flow control, etc.

These rules are configured in memory by default, and can also be loaded from different data sources. If the Sentinel console is enabled, rules can also be configured in the console. These rules will be sent via HTTP to the corresponding application instance node using Sentinel.

This article is reprinted from the WeChat public account "Tomcat Things", which can be followed through the following QR code. To reprint this article, please contact the Tomcat Things public account.