Siblings

Warning

This components has performance issues unless running with force=True. Can cause CloudWatch requests throttling. Requires some refactoring.

SiblingsManager provides Lambda executions an option of a “healthy” shutdown. They may pass the remaining payload to another execution automatically. See example:

import logging
import time
from sosw import Processor as SoswProcessor
from sosw.app import LambdaGlobals, get_lambda_handler
from sosw.components.siblings import SiblingsManager

logger = logging.getLogger()
logger.setLevel(logging.INFO)

class Processor(SoswProcessor):

    DEFAULT_CONFIG = {
        'init_clients':    ['Siblings'],    # Automatically initialize Siblings Manager
        'shutdown_period': 10,  # Some time to shutdown in a healthy manner.
    }

    siblings_client: SiblingsManager = None


    def __call__(self, event):

        cursor = event.get('cursor', 0)

        while self.sufficient_execution_time_left:
            self.process_data(cursor)
            cursor += 1
            if cursor == 20:
                return f"Reached the end of data"

        else:
            # Spawning another sibling to continue the processing
            payload = {'cursor': cursor}

            self.siblings_client.spawn_sibling(global_vars.lambda_context, payload=payload, force=True)
            self.stats['siblings_spawned'] += 1


    def process_data(self, cursor):
        """ Your custom logic respecting current cursor. """
        logger.info(f"Processing data at cursor: {cursor}")
        time.sleep(1)


    @property
    def sufficient_execution_time_left(self) -> bool:
        """ Return whether there is a sufficient execution time for processing ('shutdown period' is in seconds). """
        return global_vars.lambda_context.get_remaining_time_in_millis() > self.config['shutdown_period'] * 1000


global_vars = LambdaGlobals()
lambda_handler = get_lambda_handler(Processor, global_vars)

Here is an example use-case when you can store the remaining payload for example in S3 and call the sibling with a pointer to it.

View Licence Agreement

sosw - Serverless Orchestrator of Serverless Workers

The MIT License (MIT)
Copyright (C) 2024  sosw core contributors <info@sosw.app>

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

class sosw.components.siblings.SiblingsManager(custom_config=None, **kwargs)

This set of helpers can be used for Lambdas that want to invoke some siblings of self. Very useful for Lambdas processing queues and running out of time.

The Role of your Lambda must have the following extra permissions to run correctly. Please note that we hardcode the Arn in the policy to avoid circular dependency when parsing YAML. This dependency is absolutely valid, but CloudFormation doesn’t know how to parse it.

Read more:

https://aws.amazon.com/premiumsupport/knowledge-center/unable-validate-circular-dependency-cloudformation/

Policies:
- PolicyName: "YOUR_FUNCTION_NAME"
PolicyDocument:
  Version: "2012-10-17"
  Statement:
  - Effect: "Allow"
    Action: "cloudwatch:GetMetricStatistics"
    Resource: "*"
  - Effect: "Allow"
    Action: "lambda:InvokeFunction"
    Resource: "arn:aws:lambda:us-west-2:737060422660:function:YOUR_FUNCTION_NAME"

any_events_rules_enabled(lambda_context)

Checks the Status of CloudWatch Events Rules. It is very important to use this checker before launching siblings. Otherwise, you can create an infinite autorespawning loop and waste A LOT of money.

Parameters:

lambda_context – Context object from your lambda_handler.

Return type:

bool

Raises:

ResourceNotFoundException – If Rule with the given name doesn’t exist.

get_approximate_concurrent_executions(minutes_back=5, name=None)

Get approximate concurrent executions from CloudWatch Metrics. The value is very approximate and calculated as count of invocations during minutes_back divided by average duration in same period. Return value is rounded to integer using ceil.

We assume that the Role has permissions to read CloudWatch.

Parameters:

• minutes_back (int) – Aggregate statistics for this number of minutes.

• name (str) – Name of the function to check. Default: currently running lambda.

Return type:

int

Returns:

Approximate number of concurrent executions.

spawn_sibling(lambda_context, payload=None, force=False)

Asynchronously invokes a copy of same function to continue working. Should be called if there is still work left to do (ex: messages in the queue).

Can optionally send some payload for example remaining unprocessed rows of something. Should be formatted as dictionary.

Parameters:

• lambda_context – Context object from your lambda_handler.

• payload (dict) – The payload to be put to event.

• force (bool) – If specified True it will ignore the checks of enabled Events Rules.