Implementing circuit breaker pattern from scratch in Python

Bhavesh Praveen — Mon, 14 Dec 2020 17:41:06 +0000

We’ll briefly look into circuit breaker pattern before jumping to code.

What is circuit breaking?

In real world applications, services might go down and start back up all the time (or they might just stay down). The idea is that when you make a remote call(HTTP Request/RPC) to another service, there are chances that the remote call might fail. After a certain number of failed remote calls, we stop making the remote calls and send a cached response or an error as a response. After a certain delay, we allow one remote call to be made to the failing server, if it succeeds, we allow the subsequent remote calls to the server, if it did not succeed, we will continue sending a cached response or an error and will not make any remote calls to the failing service for some time.

When all services were working and the remote calls were returning without any errors, we call this state - "Closed".

When the remote calls continued to fail and when we stopped making any more remote calls to the failing service, we call this state - "Open"

After a certain delay, when we make a remote call to the failing service, the state transitions from "Open" to "Half Open". If the remote call does not fail, then we transition the state from "Half Open" to "Closed" and the subsequent remote calls are allowed to be made. In case the remote call failed, we transition the state from "Half Open", back to "Open" state and we wait for a certain period of time till we can make the next remote call (in Half Open state)

State Transition Diagram: image src

To know more, read this and this

Why do you need it?

To prevent a network or service failure from cascading to other services.
Saves bandwidth by not making requests over a network when the service you’re requesting is down.
Gives time for the failing service to recover or start backup.

Code Marathon

Let's now try to build a simple circuit-breaker using Python

Disclaimer: This is in no way production ready. There are some excellent libraries that are available online and well tested. I've mentioned two of them here: circuit-breaker and pybreaker.

Let's first decide on the api for the circuit breaker that we are going to build and also define the expected behavior.

I'm a big fan of retry library syntax. Let's try to use that here. We can it to this api towards the end of the blog post.

def circuit_breaker(exceptions=(Exception,), threshold=5, delay=60):
      """Returns a circuit decorator.

    :param exceptions: an exception or a tuple of exceptions to catch. default: Exception.
    :param threshold: the number of failed attempts before changing the state to Open
    :param delay: delay in seconds between Closed and Half Open state
    :param logger: logger.warning(fmt, error, delay) will be called on failed attempts.
                   default: retry.logging_logger. if None, logging is disabled.
    :returns: a circuit_breaker decorator.
    """

@circuit_breaker(exceptions=Exception, threshold=5, delay=60)
def make_api_call(url, data):
  # function that makes an api-request to another server/application
  pass

Let's define all the possible states

# circuit_breaker.py
class StateChoices:
    OPEN = "open"
    CLOSED = "closed"
    HALF_OPEN = "half_open"

Let's create a class that handles all of the circuit breaker logic.

# circuit_breaker.py
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s,%(msecs)d %(levelname)s: %(message)s",
    datefmt="%H:%M:%S",
)

class CircuitBreaker:
    def __init__(self, func, exceptions, threshold, delay):
        """
        :param func: method that makes the remote call
        :param exceptions: an exception or a tuple of exceptions to catch (ideally should be network exceptions)
        :param threshold: number of failed attempts before the state is changed to "Open"
        :param delay: delay in seconds between "Closed" and "Half-Open" state
        """
        self.func = func
        self.exceptions_to_catch = exceptions
        self.threshold = threshold
        self.delay = delay

        # by default set the state to closed
        self.state = StateChoices.CLOSED


        self.last_attempt_timestamp = None
        # keep track of failed attemp count
        self._failed_attempt_count = 0

    def update_last_attempt_timestamp(self):
        self.last_attempt_timestamp = datetime.utcnow().timestamp()

    def set_state(self, state):
        """To track the state changes by logging the information"""
        prev_state = self.state
        self.state = state
        logging.info(f"Changed state from {prev_state} to {self.state}")

    def handle_closed_state(self, *args, **kwargs):
        pass

    def handle_open_state(self, *args, **kwargs):
        pass

    def make_remote_call(self, *args, **kwargs):
        if self.state == StateChoices.CLOSED:
            return self.handle_closed_state(*args, **kwargs)
        if self.state == StateChoices.OPEN:
            return self.handle_open_state(*args, **kwargs)

Constructor takes the following parameters

func - method/function that makes the remote call
exceptions - an exception or a tuple of exceptions to catch (ideally should be network exceptions)
threshold - number of failed attempts before the state is changed to "Open"
delay - delay in seconds between "Closed" and "Half-Open" state

make_remote_call takes the parameters that the underlying remote call needs (func)

If it seems confusing, please take a look at the following snippet

def make_request(url):
  print(f"Url is {url}")

obj = CircuitBreaker(make_request, exceptions=(Exception,), threshold=5, delay=10)
obj.make_remote_call(url="www.google.com")

make_request is passed as a first class function to CircuitBreaker class. The params required by make_request are sent through make_remote_call

Let's now try to complete handle_closed_state and handle_open_state

# circuit_breaker.py
class RemoteCallFailedException(Exception):
    pass

class CircuitBreaker:

    def handle_closed_state(self, *args, **kwargs):
        allowed_exceptions = self.exceptions_to_catch
        try:
            ret_val = self.func(*args, **kwargs)
            logging.info("Success: Remote call")
            self.update_last_attempt_timestamp()
            return ret_val
        except allowed_exceptions as e:
            # remote call has failed
            logging.info("Failure: Remote call")
            # increment the failed attempt count
            self._failed_attempt_count += 1

            # update last_attempt_timestamp
            self.update_last_attempt_timestamp()

            # if the failed attempt count is more than the threshold
            # then change the state to OPEN
            if self._failed_attempt_count >= self.threshold:
                self.set_state(StateChoices.OPEN)
            # re-raise the exception
            raise RemoteCallFailedException from e


    def make_remote_call(self, *args, **kwargs):
        if self.state == StateChoices.CLOSED:
            return self.handle_closed_state(*args, **kwargs)
        if self.state == StateChoices.OPEN:
            return self.handle_open_state(*args, **kwargs)

handle_closed_state makes the remote call, if it is a success, then we update last_attempt_timestamp and return the result of the remote call. If the remote call fails, then _failed_attempt_count is incremented. If _failed_attempt_count has not reached the threshold, then simple raise an exception. If _failed_attempt_count is greater than or equal to the threshold, we change the state to Open and finally an exception is raised.

# circuit_breaker.py
class CircuitBreaker:

    def handle_open_state(self, *args, **kwargs):
        current_timestamp = datetime.utcnow().timestamp()
        # if `delay` seconds have not elapsed since the last attempt, raise an exception
        if self.last_attempt_timestamp + self.delay >= current_timestamp:
            raise RemoteCallFailedException(f"Retry after {self.last_attempt_timestamp+self.delay-current_timestamp} secs")

        # after `delay` seconds have elapsed since the last attempt, try making the remote call
        # update the state to half open state
        self.set_state(StateChoices.HALF_OPEN)
        allowed_exceptions = self.exceptions_to_catch
        try:
            ret_val = self.func(*args, **kwargs)
            # the remote call was successful
            # now reset the state to Closed
            self.set_state(StateChoices.CLOSED)
            # reset the failed attempt counter
            self._failed_attempt_count = 0
            # update the last_attempt_timestamp
            self.update_last_attempt_timestamp()
            # return the remote call's response
            return ret_val
        except allowed_exceptions as e:
            # the remote call failed again
            # increment the failed attempt count
            self._failed_attempt_count += 1

            # update last_attempt_timestamp
            self.update_last_attempt_timestamp()

            # set the state to "OPEN"
            self.set_state(StateChoices.OPEN)

            # raise the error
            raise RemoteCallFailedException from e

    def make_remote_call(self, *args, **kwargs):
        if self.state == StateChoices.CLOSED:
            return self.handle_closed_state(*args, **kwargs)
        if self.state == StateChoices.OPEN:
            return self.handle_open_state(*args, **kwargs)

handle_open_state first checks if the delay seconds has elapsed since the last attempt to make a remote call. If not, then it raises an exception. If delay seconds has elapsed since the last attempt then we change the state ot "Half Open". Now we try to make one remote call to the failing service. If the remote call was successful, then we change the state to "Closed" and reset the _failed_attempt_count to 0 and return the response of the remote call. If the remote call failed, when it was in "Half Open" state, then state is again set to "Open" and we raise an exception.

Complete code

# circuit_breaker.py

import functools
import http
import logging
from datetime import datetime

import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s,%(msecs)d %(levelname)s: %(message)s",
    datefmt="%H:%M:%S",
)


class StateChoices:
    OPEN = "open"
    CLOSED = "closed"
    HALF_OPEN = "half_open"


class RemoteCallFailedException(Exception):
    pass


class CircuitBreaker:
    def __init__(self, func, exceptions, threshold, delay):
        """
        :param func: method that makes the remote call
        :param exceptions: an exception or a tuple of exceptions to catch (ideally should be network exceptions)
        :param threshold: number of failed attempts before the state is changed to "Open"
        :param delay: delay in seconds between "Closed" and "Half-Open" state
        """
        self.func = func
        self.exceptions_to_catch = exceptions
        self.threshold = threshold
        self.delay = delay

        # by default set the state to closed
        self.state = StateChoices.CLOSED


        self.last_attempt_timestamp = None
        # keep track of failed attemp count
        self._failed_attempt_count = 0

    def update_last_attempt_timestamp(self):
        self.last_attempt_timestamp = datetime.utcnow().timestamp()

    def set_state(self, state):
        prev_state = self.state
        self.state = state
        logging.info(f"Changed state from {prev_state} to {self.state}")

    def handle_closed_state(self, *args, **kwargs):
        allowed_exceptions = self.exceptions_to_catch
        try:
            ret_val = self.func(*args, **kwargs)
            logging.info("Success: Remote call")
            self.update_last_attempt_timestamp()
            return ret_val
        except allowed_exceptions as e:
            # remote call has failed
            logging.info("Failure: Remote call")
            # increment the failed attempt count
            self._failed_attempt_count += 1

            # update last_attempt_timestamp
            self.update_last_attempt_timestamp()

            # if the failed attempt count is more than the threshold
            # then change the state to OPEN
            if self._failed_attempt_count >= self.threshold:
                self.set_state(StateChoices.OPEN)
            # re-raise the exception
            raise RemoteCallFailedException from e

    def handle_open_state(self, *args, **kwargs):
        current_timestamp = datetime.utcnow().timestamp()
        # if `delay` seconds have not elapsed since the last attempt, raise an exception
        if self.last_attempt_timestamp + self.delay >= current_timestamp:
            raise RemoteCallFailedException(f"Retry after {self.last_attempt_timestamp+self.delay-current_timestamp} secs")

        # after `delay` seconds have elapsed since the last attempt, try making the remote call
        # update the state to half open state
        self.set_state(StateChoices.HALF_OPEN)
        allowed_exceptions = self.exceptions_to_catch
        try:
            ret_val = self.func(*args, **kwargs)
            # the remote call was successful
            # now reset the state to Closed
            self.set_state(StateChoices.CLOSED)
            # reset the failed attempt counter
            self._failed_attempt_count = 0
            # update the last_attempt_timestamp
            self.update_last_attempt_timestamp()
            # return the remote call's response
            return ret_val
        except allowed_exceptions as e:
            # the remote call failed again
            # increment the failed attempt count
            self._failed_attempt_count += 1

            # update last_attempt_timestamp
            self.update_last_attempt_timestamp()

            # set the state to "OPEN"
            self.set_state(StateChoices.OPEN)

            # raise the error
            raise RemoteCallFailedException from e

    def make_remote_call(self, *args, **kwargs):
        if self.state == StateChoices.CLOSED:
            return self.handle_closed_state(*args, **kwargs)
        if self.state == StateChoices.OPEN:
            return self.handle_open_state(*args, **kwargs)

Now to test it out. Let's create a mock server.

Install Flask and requests. Ipython is optional

pip install requests
pip install Flask
pip install ipython

Let's create some endpoints to mock the server

# main.py

import random
import time

from flask import Flask
app = Flask(__name__)


@app.route('/success')
def success_endpoint():
    return {
        "msg": "Call to this endpoint was a smashing success."
    }, 200


@app.route('/failure')
def faulty_endpoint():
    r = random.randint(0, 1)
    if r == 0:
        time.sleep(2)

    return {
        "msg": "I will fail."
    }, 500


@app.route('/random')
def fail_randomly_endpoint():
    r = random.randint(0, 1)
    if r == 0:
        return {
            "msg": "Success msg"
        }, 200

    return {
        "msg": "I will fail (sometimes)."
    }, 500

Run the development server

export FLASK_APP=main.py; flask run

By default it runs on port 5000

Now to test it out. You can use these snippets to test it out.

# snippets.py

faulty_endpoint = "http://localhost:5000/failure"
success_endpoint = "http://localhost:5000/success"
random_status_endpoint = "http://localhost:5000/random"

def make_request(url):
    try:
        response = requests.get(url, timeout=0.3)
        if response.status_code == http.HTTPStatus.OK:
            print(f"Call to {url} succeed with status code = {response.status_code}")
            return response
        if 500 <= response.status_code < 600:
            print(f"Call to {url} failed with status code = {response.status_code}")
            raise Exception("Server Issue")
    except Exception:
        print(f"Call to {url} failed")
        raise

(circuit-breaker) ➜  circuit-breaker git:(master) ✗ ipython

In [1]: from circuit_breaker import CircuitBreaker

In [2]: from snippets import make_request, faulty_endpoint, success_endpoint

In [3]: obj = CircuitBreaker(make_request, exceptions=(Exception,), threshold=5, delay=10)

In [4]: obj.make_remote_call(success_endpoint)
Call to http://localhost:5000/success succeed with status code = 200
06:07:51,255 INFO: Success: Remote call
Out[4]: <Response [200]>

In [5]: obj.make_remote_call(success_endpoint)
Call to http://localhost:5000/success succeed with status code = 200
06:07:53,610 INFO: Success: Remote call
Out[5]: <Response [200]>

In [6]: vars(obj)
Out[6]:
{'func': <function snippets.make_request(url)>,
 'exceptions_to_catch': (Exception,),
 'threshold': 5,
 'delay': 10,
 'state': 'closed',
 'last_attempt_timestamp': 1607800073.610199,
 '_failed_attempt_count': 0}

Line 1 and Line 2 are just imports. In line 3, we are creating a CircuitBreaker object for make_request. Here, we're setting exceptions=(Exception,), this will catch all the exceptions. We should ideally narrow down the exception to the one that we actually want to catch, in this case, Network Exceptions, but we're going to leave it there for this demo.

Now make successive calls to the faulty endpoint.

In [7]: obj.make_remote_call(faulty_endpoint)

In [8]: obj.make_remote_call(faulty_endpoint)

In [9]: obj.make_remote_call(faulty_endpoint)

In [10]: obj.make_remote_call(faulty_endpoint)

In [11]: obj.make_remote_call(faulty_endpoint)

In [12]: obj.make_remote_call(faulty_endpoint)
---------------------------------------------------------------------------
Traceback data ..........

RemoteCallFailedException: Retry after 8.688776969909668 secs  

In [13]: obj.make_remote_call(success_endpoint)
---------------------------------------------------------------------------
Traceback data......

RemoteCallFailedException: Retry after 6.096494913101196 secs

Try to make these calls as fast as possible. After the first five callls to the faulty_endpoint, the next call(Line 12) will not make an api-request to the flask server instead it will raise an Exception, mentioning to retry after a specified number of secs. Even if you make an api call to the success_endpoint endpoint (Line 13), it will still raise an error. It is in "Open" state.

Now, after the delay time has elapsed, if we make a call to the faulty endpoint, it will transition from Half-Open to Open state.

In [18]: obj.make_remote_call(faulty_endpoint)
06:21:24,959 INFO: Changed state from open to half_open
...
06:21:24,964 INFO: Changed state from half_open to open

Now, after the delay has elapsed, if we make a call to the success_endpoint, it will transition from Half-Open to Closed state

In [19]: obj.make_remote_call(success_endpoint)
06:25:10,673 INFO: Changed state from open to half_open
...
06:25:10,678 INFO: Changed state from half_open to closed
Out[19]: <Response [200]>

Now we have a working circuit breaker. We could introduce response caching, monitoring and make it threadsafe. Errors could be handled better. More Exception types could help. All of these features are left as an exercise for the readers.

Finally, improving the api shouldn't take a lot of time. I've added quick dirty version here

# circuit_breaker.py

class APICircuitBreaker:
    def __init__(self, exceptions=(Exception,), threshold=5, delay=60):
        self.obj = functools.partial(
            CircuitBreaker,
            exceptions=exceptions,
            threshold=threshold,
            delay=delay
        )

    def __call__(self, func):
        self.obj = self.obj(func=func)

        def decorator(*args, **kwargs):
            ret_val = self.obj.make_remote_call(*args, **kwargs)
            return ret_val

        return decorator

    def __getattr__(self, item):
        return getattr(self.obj, item)


circuit_breaker = APICircuitBreaker

# snippets.py

@circuit_breaker()
def make_request(url):
    try:
        response = requests.get(url, timeout=0.3)
        if response.status_code == http.HTTPStatus.OK:
            print(f"Call to {url} succeed with status code = {response.status_code}")
            return response
        if 500 <= response.status_code < 600:
            print(f"Call to {url} failed with status code = {response.status_code}")
            raise Exception("Server Issue")
    except Exception:
        print(f"Call to {url} failed")
        raise

All code samples can be found here

Now we have a working circuit breaker. We could introduce response caching, monitoring and make it thread-safe. Errors could be handled better. More Exception types could help. All of these features are left as an exercise for the readers.

Connect with me on Twitter

References:

DEV Community: Bhavesh Praveen

Implementing circuit breaker pattern from scratch in Python

What is circuit breaking?

Why do you need it?

Code Marathon