This is a rate-limiter module which leverages DynamoDB to enforce resource limits.

This module offers two different methods for consuming and replenishing resource capacity. The typical use calls for a time driven method for distributing and replenishing tokens, e.g. 10 requests per second. However, there are cases which require both an acquisition and return of a specific token, e.g. wait for an EMR cluster to complete.

Bearing this in mind, resource capacity is represented by either fungible or non-fungible tokens. The former are interchangeable and thus can make use of time-based expiration to restore capacity. The latter are unique and require explicit removal to restore capacity.

For fungible tokens, the limiter leverages the token bucket algorithm, tracking resource usage by acquiring and replenishing tokens for each unit of capacity.

For non-fungible tokens, the limiter creates a new token for each resource. Limiting is enforced by disallowing token creation beyond the specified capacity.

Fungible token rate-limiting requirements and usage is detailed below.

The expected usage is each rate limiter will use the same multi-tenant token and limit tables, created and managed by a separate service. However, a private token and/or limit table can be used when instantiating the middleware.

The tokens for a single resource are stored in a single DynamoDB row, representing the "bucket". The expected table schema is detailed below.

These are all the expected table attributes, including the keys.

The key data type and description can be found in the above, attributes table.

The limit and window for a specific account on a specific resource are stored in a single DynamoDB row. The expected table schema is detailed below.

These are all the expected table attributes, including the keys.

The key data type and description can be found in the above, attributes table.

The service limits global secondary index is used when updating/loading service limits.

Each of the fungible token limiter implementations require the names of the token and limit tables. These values can be passed directly to the limiter or set via environment variables.

The only other value required by each implementation is account id. Each implementation handles specifying this value differently.

The function decorator has a minimum of four arguments:

1. The name of the resource being rate-limited.

2. How to access the account id from the arguments of the function being decorated. This can be either a positional argument numeric index or a keyword argument key.

3. The default limit, used when no limit is found in the limits table.

4. The default window, used when no window is found in the limits table.

The examples below assume the table name, limit and window have been set via environment variables.

from limiter import rate_limit

@rate_limit('my-resource', 1, 10, account_id_pos=1)
def invoke_my_resource(arg_1, account_id):
  # If I am here, I was not rate limited

from limiter import rate_limit

@rate_limit('my-resource', 1, 10, account_id_key='foo')
def invoke_my_resource(arg_1, foo='account-1234'):
  # If I am here, I was not rate limited

# The default keyword argument is "account_id", to make the decorator more succinct:
@rate_limit('my-resource', 1, 10)
def invoke_my_resource(arg_1, account_id='account-1234'):
  # If I am here, I was not rate limited

The context manager has a minimum of four arguments:

1. The name of the resource being rate-limited.

2. The account id.

3. The default limit, used when no limit is found in the limits table.

4. The default window, used when no window is found in the limits table.

The example below assume the table name, limit and window have been set via environment variables.

from limiter import fungible_limiter

def invoke_my_resource(account_id):
  with fungible_limiter('my-resource', account_id, 1, 10):
    # If I am here, I was not rate limited

Directly creating an instance of the fungible limiter has a minimum of four arguments:

1. The name of the resource being rate-limited.

2. The account id.

3. The default limit, used when no limit is found in the limits table.

4. The default window, used when no window is found in the limits table.

The example below assume the table name, limit and window have been set via environment variables.

from limiter import fungible_limiter

def invoke_my_resource(account_id):
  limiter = fungible_limiter('my-resource', account_id, 1, 10)
  limiter.get_token()
  # If I am here, I was not rate limited

Non-fungible token rate-limiting requirements and usage is detailed below.

Each token is represented as a single row in DynamoDB. The expected table schema is detailed below.

These are all the expected table attributes, including the keys.

The key data type and description can be found in the above, attributes table.

A global secondary index is used to locate tokens using only resourceId. This will be needed to locate tokens based on the resource id provided in CloudWatch events.

Each of the fungible token limiter implementations require the names of the token and limit tables. These values can be passed directly to the limiter or set via environment variables.

The only other value required by each implementation is account id. Each implementation handles specifying this value differently.

Each implementation example assumes the table name and limit have been set via environment variables.

The context manager has a minimum of three arguments:

1. The name of the resource being rate-limited.

2. The account id.

3. The default limit, used when no limit is found in the limits table.

from limiter import non_fungible_limiter

def invoke_my_resource(account_id):
  with non_fungible_limiter('my-resource', account_id, 10) as reservation:
    emr_cluster_id = create_emr_cluster() # Create an instance of the resource
    reservation.create_token(emr_cluster_id) # Create a token for this unique resource

Directly creating an instance of the non-fungible limiter has a minimum of three arguments:

1. The name of the resource being rate-limited.

2. The account id.

3. The default limit, used when no limit is found in the limits table.

from limiter import non_fungible_limiter

def invoke_my_resource(account_id):
  limiter = non_fungible_limiter('my-resource', account_id, 10)
  reservation = limiter.get_reservation()

  emr_cluster_id = create_emr_cluster() # Create an instance of the resource
  reservation.create_token(emr_cluster_id) # Create a token for this unique resource

The recommended approach for detecting the termination of and removing tokenized resources is via a Lambda triggered by CloudWatch. The CloudWatch rules should be as precise as practical to avoid unnecessarily executing the lambda.

The event_processors module contains all the logic necessary to consume, test and remove tokens from CloudWatch events.

The EventProcessorManager class removes non-fungible tokens from DynamoDB represented by CloudWatch events. The manager is composed of multiple EventProcessors, one for specific event "source", e.g. 'aws.emr'. The processors are responsible for determining if an event references a tokenized resource and if so, extracting its resource id. Each processor is composed of zero to many predicates, which determine if an event references a tokenized resource. If a processor is not configured with any predicates it will just extract the resource id.

from limiter.event_processors import EventProcessorManager, EventProcessor, ProcessorPredicate

predicate = ProcessorPredicate('detail.name', lambda name: 'debugging' not in name)
processor = EventProcessor('aws.emr', 'detail.clusterId', predicate=predicate)
manager = EventProcessorManager(table_name='table', index_name='idx', processors=[processor])

def handler(event, context):
  manager.process_event(event)

EventProcessors can also be configured with the event detail-type. The EventProcessor in the above example can be tweaked to process EMR step changes.

...
processor = EventProcessor('aws.emr', 'detail.stepId', type='EMR Step Status Change', predicate=predicate)
...

The dependencies needed for local development (running unit tests, etc.) are contained in dev_requirements.txt and can be installed via pip: pip install -r dev_requirements.txt.

Running the unit tests is done via a recipe in the makefile, the command: make test. The unit tests are run with nose inside a virtual environment managed by tox. The test_requirements.txt contains all the testing dependencies and is used to pip install everything needed by the tests in the tox environments (tox installs these dependencies).

The make check command will run pylint with standards defined in pylintrc. This is a measurable way to enforce style and standards.

Run make clean to remove artifacts leftover by tox and pylint.