Custom Circuit Breaker Policy for Mule 4

 

1.​ Introduction

 This document is to show the details installation and configuration for custom circuit breaker policy.

 Circuit Breaker monitors API calls. When everything is working as expected, it is in the state “closed”. When the number of fails, like timeout, bad request, backend api failure reaches a specified threshold, Circuit Breaker will stop processing further requests. We call it the open state. As a result, API clients will receive instant information that something went wrong without waiting for the timeout to come.

 

​2.​ Detailed Policy Design

 This policy implements the Circuit Breaker pattern by monitoring a configurable number of exceptions or errors returned from the target endpoint.

When the number of consecutive failures crosses a threshold, the circuit breaker trips, and for the duration of a timeout period all attempts to invoke the API will fail immediately. After the timeout expires the circuit breaker allows the API request to pass through. If the request succeed, the circuit breaker resumes normal operation. Otherwise, the cycle repeats. 

This policy works with HTTP response code of the API. So please make sure response status is proper in case any error.

 

​3.​ Installation and Configuration

​ The policy configuration contains several input parameters:

 Max Error - number of errors/exceptions upon which the breaker trips. Accepts value between 1 and 1000 , defaulted to 3 

2.     Wait Time - Amount of time the breaker will be on until it gets opened again (ms). Accepts value beween 1 and 360000 , defaulted to 60000 

3.     Error Code - An mule expression used to trap for errors in the resulting message 

4.     Object Store Name for the Policy - The policy uses Object Store internally. The default objectStore name used by this policy is ```circuit-breaker-os```. This can be overriden. Please provide an unique name for API’s. 

5.     Response Code - The response http status code returned , once circuit is tripped. Defaulted to 503

Error Response – An mule error expression used to define the error payload (response) returned once the circuit is tripped. A default is provided , but this can be overriden.

 Policy configuration Diagram mentioned below

4.​ Implementation

<?xml version="1.0" encoding="UTF-8"?>

<mule xmlns="http://www.mulesoft.org/schema/mule/core"

      xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

      xmlns:http-policy="http://www.mulesoft.org/schema/mule/http-policy"

      xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"

      xmlns:os="http://www.mulesoft.org/schema/mule/os"

      xmlns:validation="http://www.mulesoft.org/schema/mule/validation"

  xmlns:http-transform="http://www.mulesoft.org/schema/mule/http-policy-transform"

      xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd

               http://www.mulesoft.org/schema/mule/http-policy http://www.mulesoft.org/schema/mule/http-policy/current/mule-http-policy.xsd

               http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd

               http://www.mulesoft.org/schema/mule/os http://www.mulesoft.org/schema/mule/os/current/mule-os.xsd

http://www.mulesoft.org/schema/mule/validation http://www.mulesoft.org/schema/mule/validation/current/mule-validation.xsd

http://www.mulesoft.org/schema/mule/http-policy-transform http://www.mulesoft.org/schema/mule/http-policy-transform/current/mule-http-policy-transform.xsd">

    <os:object-store name="{{{objectstoreName}}}" />

<validation:config name="Validation_Config" />

    <http-policy:proxy name="{{{policyId}}}-custom-policy">

        <http-policy:source>

            

            <try>

                <flow-ref name="circuit-breaker-evalTrip-flow" />

<validation:is-false config-ref="Validation_Config" expression="#[vars.breakerTripped]" message="#['Service Not available. Please try again later. ETA {{{wait_time}}} ms']"/>

                    

                <http-policy:execute-next/>  

                

                <flow-ref name="circuit-breaker-evalTrip-flow"/>

<logger level="INFO" message="============= Entry point logger ===================="/>

                <logger level="INFO" message="#[isEmpty(payload.error)]"/>

<logger level="INFO" message="#[attributes.statusCode]"/>

<logger level="INFO" message="#[payload.error.errorCode]"/>

                <set-variable value="{{{err_msg_trap_expr}}}" variableName="evalExpression"/>

                <try>

                    <validation:is-false config-ref="Validation_Config" message="#['Service Unavailable..']" expression="#[vars.evalExpression]" />

<flow-ref name="circuit-breaker-reset-flow"/>

                    <error-handler >

                        <on-error-continue enableNotifications="true" logException="true" >

                            <flow-ref name="circuit-breaker-increment-breaker-count"/>

                        </on-error-continue>

                    </error-handler>

                </try>

                <error-handler >

                    <on-error-propagate enableNotifications="true" logException="true" type="X_CRCTBRKPOLICY:CIRCUIT_BREAK_ERROR , VALIDATION:INVALID_BOOLEAN">

<set-variable value="{{{err_response_expr}}}" variableName="errorResponseExpression"/>

<set-variable value="{{{httpStatusValue}}}" variableName="httpStatusValue"/>

<http-transform:set-response statusCode="{{{httpStatusValue}}}">

                                <http-transform:body>#[

                                    output application/json

                                    ---

                                    vars.errorResponseExpression

                                    ]</http-transform:body>

                            </http-transform:set-response>

                    </on-error-propagate>

                </error-handler>

            </try>

        </http-policy:source>

    </http-policy:proxy>

    

    <!-- Initialization Flow - START -->

    <flow name="circuit-breaker-init-flow" >

<flow-ref name="circuit-breaker-init-os-flow"/>

<flow-ref name="circuit-breaker-init-var-flow"/>

</flow>

<flow name="circuit-breaker-init-var-flow" >

<os:retrieve objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-tripped']" target="breakerTripped"/>

<os:retrieve objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" target="breakerCount"/>

<os:retrieve objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ &quot;breaker-expiry-ts&quot;]" target="breakerExpiryTS"/>

</flow>

    <flow name="circuit-breaker-init-os-flow" >

<ee:transform  >

<ee:message >

</ee:message>

<ee:variables >

<ee:set-variable variableName="prefix" ><![CDATA[%dw 2.0

output application/java

---

app.name as String default ""  ++ mule.nodeId as String default "" ]]></ee:set-variable>

</ee:variables>

</ee:transform>

<ee:transform>

<ee:message >

</ee:message>

<ee:variables >

<ee:set-variable variableName="breakerTrippedVar" ><![CDATA[%dw 2.0

output application/java

---

vars.prefix ++ "breaker-tripped"]]></ee:set-variable>

<ee:set-variable variableName="breakerCountVar" ><![CDATA[%dw 2.0

output application/java

---

vars.prefix ++ "breaker-count"]]></ee:set-variable>

<ee:set-variable variableName="breakerExpiryTsVar" ><![CDATA[%dw 2.0

output application/java

---

vars.prefix ++ "breaker-expiry-ts"]]></ee:set-variable>

</ee:variables>

</ee:transform>

<os:contains objectStore="{{{objectstoreName}}}" key="#[vars.breakerTrippedVar]" target="containsBreakTripped"/>

<choice>

<when expression="#[not (vars.containsBreakTripped default false)]">

<os:store objectStore="{{{objectstoreName}}}" key="#[vars.breakerTrippedVar]">

<os:value ><![CDATA[#[false]]]></os:value>

</os:store>

</when>

</choice>

<os:contains objectStore="{{{objectstoreName}}}" key="#[vars.breakerCountVar]" target="containsBreakerCount"/>

<choice>

<when expression="#[not (vars.containsBreakerCount default false)]">

<os:store objectStore="{{{objectstoreName}}}" key="#[vars.breakerCountVar]">

<os:value ><![CDATA[#[0]]]></os:value>

</os:store>

</when>

</choice>

<os:contains objectStore="{{{objectstoreName}}}" key="#[vars.breakerExpiryTsVar]" target="containsBreakerExpiryTs"/>

<choice>

<when expression="#[not (vars.containsBreakerExpiryTs default false)]">

<os:store objectStore="{{{objectstoreName}}}" key="#[vars.breakerExpiryTsVar]">

<os:value ><![CDATA[#[0]]]></os:value>

</os:store>

</when>

</choice>

</flow>

    <!-- Initialization Flow - END -->

    <!-- Circuit Reset Flow - START-->

    <flow name="circuit-breaker-reset-flow">

<os:contains key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-tripped']" objectStore="{{{objectstoreName}}}" target="containsBreakTripped"/>

<choice>

<when expression="#[vars.containsBreakTripped]">

<os:remove objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-tripped']" />

</when>

</choice>

<os:contains objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" target="containsBreakerCount"/>

<choice >

<when expression="#[vars.containsBreakerCount]">

<os:remove key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" objectStore="{{{objectstoreName}}}" />

</when>

</choice>

<os:contains key="#[(app.name default '') ++ (mule.nodeId default '') ++ &quot;breaker-expiry-ts&quot;]" objectStore="{{{objectstoreName}}}" target="containsBreakerExpiryTs"/>

<choice>

<when expression="#[vars.containsBreakerExpiryTs]">

<os:remove key="#[(app.name default '') ++ (mule.nodeId default '') ++ &quot;breaker-expiry-ts&quot;]" objectStore="{{{objectstoreName}}}" />

</when>

</choice>

<logger level="INFO" message="#['Breaker Reset.']"/>

</flow>

    <!-- Circuit Reset Flow - END-->

    <!-- Circuit Trip Flow - START-->

    <flow name="circuit-breaker-trip-flow" >

<os:store objectStore="{{{objectstoreName}}}" key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-tripped']">

<os:value ><![CDATA[#[true]]]></os:value>

</os:store>

<os:store key="#[(app.name default '') ++ (mule.nodeId default '') ++ &quot;breaker-expiry-ts&quot;]" objectStore="{{{objectstoreName}}}">

<os:value ><![CDATA[#[now() + ("PT$({{{wait_time}}} / 1000)S" as Period)]]]></os:value>

</os:store>

</flow>

    <!-- Circuit Trip Flow - END-->

    <flow name="circuit-breaker-log-flow" >

<logger level="INFO" message="#['breakerTripped : ' ++ vars.breakerTripped]" />

<logger level="INFO" message="#['breakerCount : ' ++ vars.breakerCount]" />

<logger level="INFO" message="#['breakerExpiryTS : ' ++ vars.breakerExpiryTS]" />

</flow>

    <!-- Current Failure Count Increment Flow - START-->

<flow name="circuit-breaker-increment-breaker-count" >

<logger level="DEBUG" message="Increment Counts.."/>

<os:retrieve key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" target="currentCount" objectStore="{{{objectstoreName}}}">

<os:default-value ><![CDATA[#[0]]]></os:default-value>

</os:retrieve>

<os:store key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" objectStore="{{{objectstoreName}}}">

<os:value ><![CDATA[#[(vars.currentCount default 0 as Number) + 1]]]></os:value>

</os:store>

<os:retrieve key="#[(app.name default '') ++ (mule.nodeId default '') ++ 'breaker-count']" target="newCount" objectStore="{{{objectstoreName}}}"/>

<logger level="INFO" message="#[vars.newCount]"/>

</flow>

    <!-- Current Failure Count Increment Flow - END-->

<!-- Circuit Breaker Evaluation Flow - START-->

    <flow name="circuit-breaker-evalTrip-flow" >

<logger level="INFO" message="============= Entry point logger 2 ===================="/>

<logger level="INFO" message="#[attributes.statusCode]"/>

<logger level="INFO" message="#[payload.error.errorCode]"/>

<flow-ref name="circuit-breaker-init-flow"/>

<flow-ref name="circuit-breaker-log-flow" />

<set-variable value="{{{max_error_count}}}" variableName="maxCount"/>

<choice>

<when expression="#[vars.breakerTripped and (now() &lt;= vars.breakerExpiryTS)]">

<logger level="INFO" message="============= Condition 1 ===================="/>

<raise-error type="X_CRCTBRKPOLICY:CIRCUIT_BREAK_ERROR" description="Service Not available. Please try again later. ETA {{{wait_time}}} ms"/>

</when>

<when expression="#[vars.breakerTripped and (now() &gt; vars.breakerExpiryTS)]">

<logger level="INFO" message="=================== Condition 2 =======================" />

<flow-ref name="circuit-breaker-reset-flow"/>

<set-variable value="false" variableName="breakerTripped"/>

</when>

<when expression="#[vars.breakerCount &gt;= vars.maxCount]">

<logger level="INFO"  message="===================== Condition 3 ======================"/>

<flow-ref name="circuit-breaker-trip-flow"/>

</when>

<otherwise >

<logger level="INFO" message="NOOP"/>

</otherwise>

</choice>

</flow>

    <!-- Circuit Breaker Evaluation Flow - END-->

</mule>

5.​ Pom file

<?xml version="1.0" encoding="UTF-8" standalone="no"?>

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>mycompany</groupId>

    <artifactId>circuit-breaker-mule4</artifactId>

    <version>1.0.0</version>

    <name>Circuit Breaker Policy (Mule 4)</name>

    <description>A custom policy to implement a lightweight Circuit Breaker pattern</description>

    <packaging>mule-policy</packaging>

    <properties>

        <mule.maven.plugin.version>3.3.5</mule.maven.plugin.version>

        <exchange.url>https://maven.anypoint.mulesoft.com/api/v1/organizations/${groupId}/maven</exchange.url>

        <httpPolicyTransformVersion>3.1.0</httpPolicyTransformVersion>

        <app.runtime>4.1.4-hf3</app.runtime>

    </properties>

    <build>

        <plugins>

            <plugin>

                <groupId>org.mule.tools.maven</groupId>

                <artifactId>mule-maven-plugin</artifactId>

                <version>${mule.maven.plugin.version}</version>

                <extensions>true</extensions>

            </plugin>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-deploy-plugin</artifactId>

                <executions>

                    <execution>

                        <id>upload-template</id>

                        <phase>deploy</phase>

                        <goals>

                            <goal>deploy-file</goal>

                        </goals>

                        <configuration>

                            <repositoryId>ExchangeRepository</repositoryId>

                            <url>${exchange.url}</url>

                            <file>${project.basedir}/${project.artifactId}.yaml</file>

                            <generatePom>false</generatePom>

                            <groupId>${project.groupId}</groupId>

                            <artifactId>${project.artifactId}</artifactId>

                            <version>${project.version}</version>

                            <packaging>yaml</packaging>

                            <classifier>policy-definition</classifier>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>

    <dependencies>

        <dependency>

        <groupId>com.mulesoft.anypoint</groupId>

        <artifactId>mule-http-policy-transform-extension</artifactId>

        <version>${httpPolicyTransformVersion}</version>

        <classifier>mule-plugin</classifier>

        <!--

        <exclusions>

            <exclusion>

                <groupId>org.mule.connectors</groupId>

                <artifactId>mule-http-connector</artifactId>

            </exclusion>

        </exclusions>

        -->

    </dependency>

<dependency>

            <groupId>org.mule.connectors</groupId>

            <artifactId>mule-objectstore-connector</artifactId>

            <version>1.1.3</version>

            <classifier>mule-plugin</classifier>

        </dependency>

    <dependency>

            <groupId>org.mule.modules</groupId>

            <artifactId>mule-validation-module</artifactId>

            <version>1.4.0</version>

            <classifier>mule-plugin</classifier>

        </dependency>

    </dependencies>

    <distributionManagement>

        <repository>

            <id>ExchangeRepository</id>

            <name>Corporate Repository</name>

            <url>https://maven.anypoint.mulesoft.com/api/v1/organizations/${groupId}/maven</url>

            <layout>default</layout>

        </repository>        

    </distributionManagement>

     <pluginRepositories>

        <pluginRepository>

            <id>mule-plugin</id>

            <name>Mule Repository</name>

            <url>https://repository.mulesoft.org/nexus/content/repositories/public/</url>

        </pluginRepository>

    </pluginRepositories>

     

</project>

6.​

Publishing the policy to the Exchange

1. Check out this project on to your local desktop

2. Configure the ```exchange-server``` credentials to the Maven settings.xml

3. Navigate to the Project folder and execute the below command

./deploy.sh <YOUR_ORG_ID>

Please let me know if you need any help contact @sumitfrndz@gmail.com