def _perform_canary()

in blogs/ecs-canary-deployments-pipeline/shared_stack/lambda_functions/start_canary/main.py [0:0]
68 lines of code
3 McCabe index (conditional complexity)

def _perform_canary(event, current_vn_weight, new_vn_weight, current_vn, new_vn):
    """ Function to perform Canary. """

    new_vn = event['MicroserviceName']+'-'+event['Sha']
    try:
        if event['Protocol'].lower() == 'http':
            spec={
                'httpRoute': {
                    'action': {
                        'weightedTargets': [
                            {
                                'virtualNode':current_vn,
                                'weight':current_vn_weight
                            },
                            {
                                'virtualNode':new_vn,
                                'weight':new_vn_weight
                            }
                        ]
                    },
                    "match": {
                                "prefix": "/"
                            },
                    'retryPolicy': {
                        'httpRetryEvents': [
                            'server-error',
                            'client-error',
                            'gateway-error'
                        ],
                        'maxRetries': 2,
                        'perRetryTimeout': {
                            'unit': 'ms',
                            'value': 2000
                        }
                    }
                }
            }
        else:
            spec={
                'tcpRoute': {
                    'action': {
                        'weightedTargets': [
                            {
                                'virtualNode':current_vn,
                                'weight':current_vn_weight
                            },
                            {
                                'virtualNode':new_vn,
                                'weight':new_vn_weight
                            }
                        ]
                    },
                    'timeout': {
                        'idle': {
                            'unit': 'ms',
                            'value': 2000
                        }
                    }
                }
            }

        APPMESH_CLIENT.update_route(
            meshName=event['EnvironmentName'],
            routeName=event['MicroserviceName']+'-'+'route',
            spec=spec,
            virtualRouterName=event['MicroserviceName']+'-'+'vr'
        )
        return True
    except ClientError as _ex:
        LOGGER.error("Update route failed during subsequent canary rollouts")
        return False