Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_lambda_metrics method in localstack

Best Python code snippet using localstack_python

test_integration.py

Source:test_integration.py

...143    if len(EVENTS) != num_events:144        LOGGER.warning('DynamoDB and Kinesis updates retrieved (actual/expected): %s/%s' % (len(EVENTS), num_events))145    assert len(EVENTS) == num_events146    # check cloudwatch notifications147    stats1 = get_lambda_metrics(TEST_LAMBDA_NAME_STREAM)148    assert len(stats1['Datapoints']) == 2 + num_events_sns149    stats2 = get_lambda_metrics(TEST_LAMBDA_NAME_STREAM, 'Errors')150    assert len(stats2['Datapoints']) == 1151    stats3 = get_lambda_metrics(TEST_LAMBDA_NAME_DDB)152    assert len(stats3['Datapoints']) == 10153def test_kinesis_lambda_forward_chain():154    kinesis = aws_stack.connect_to_service('kinesis')155    s3 = aws_stack.connect_to_service('s3')156    aws_stack.create_kinesis_stream(TEST_CHAIN_STREAM1_NAME, delete=True)157    aws_stack.create_kinesis_stream(TEST_CHAIN_STREAM2_NAME, delete=True)158    s3.create_bucket(Bucket=TEST_BUCKET_NAME)159    # deploy test lambdas connected to Kinesis streams160    zip_file = testutil.create_lambda_archive(load_file(TEST_LAMBDA_PYTHON), get_content=True,161        libs=TEST_LAMBDA_LIBS, runtime=LAMBDA_RUNTIME_PYTHON27)162    testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA1_NAME, zip_file=zip_file,163        event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM1_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)164    testutil.create_lambda_function(func_name=TEST_CHAIN_LAMBDA2_NAME, zip_file=zip_file,165        event_source_arn=get_event_source_arn(TEST_CHAIN_STREAM2_NAME), runtime=LAMBDA_RUNTIME_PYTHON27)166    # publish test record167    test_data = {'test_data': 'forward_chain_data_%s' % short_uid()}168    data = clone(test_data)169    data[lambda_integration.MSG_BODY_MESSAGE_TARGET] = 'kinesis:%s' % TEST_CHAIN_STREAM2_NAME170    kinesis.put_record(Data=to_bytes(json.dumps(data)), PartitionKey='testId', StreamName=TEST_CHAIN_STREAM1_NAME)171    # check results172    time.sleep(5)173    all_objects = testutil.list_all_s3_objects()174    testutil.assert_objects(test_data, all_objects)175# ---------------176# HELPER METHODS177# ---------------178def get_event_source_arn(stream_name):179    kinesis = aws_stack.connect_to_service('kinesis')180    return kinesis.describe_stream(StreamName=stream_name)['StreamDescription']['StreamARN']181def get_lambda_metrics(func_name, metric='Invocations'):182    return cloudwatch_util.get_metric_statistics(183        Namespace='AWS/Lambda',184        MetricName=metric,185        Dimensions=[{'Name': 'FunctionName', 'Value': func_name}],186        Period=60,187        StartTime=datetime.now() - timedelta(minutes=1),188        EndTime=datetime.now(),189        Statistics=['Sum']...

aws_lambda_metrics.py

Source:aws_lambda_metrics.py

...44aws_secret_access_key = config.get('xx', 'aws_secret_access_key')45aws_session_token = config.get('xx', 'aws_session_token')46CW_WINDOW = 5 # Min47PERIOD = 300 # Secs48def get_lambda_metrics(function_name,REGION_NAME):49    influx_client = InfluxDBClient(host='localhost', port=8086, database='telegraf')50    client = boto3.client('cloudwatch', region_name=REGION_NAME, aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, aws_session_token=aws_session_token)51    metrics = ['Invocations','Errors','Dead Letter Error','Duration','Throttles','IteratorAge','ConcurrentExecutions','UnreservedConcurrentExecutions']52    for metric in metrics:53        response = client.get_metric_statistics(54            Namespace = 'AWS/Lambda',55            MetricName = metric,56            Dimensions = [57                {58                    'Name': 'FunctionName',59                    'Value': function_name60                },61            ],62            StartTime = datetime.utcnow() - timedelta(minutes=CW_WINDOW),63            EndTime = datetime.utcnow(),64            Period = PERIOD,65            Statistics = [66                'Maximum',67                'Minimum',68                'Average',69                'Sum',70                'SampleCount'71            ],72        )73       74        recs = len(response['Datapoints'])75        #pprint.pprint(response, indent=2)76        count = 077        if recs > 0:78            while count < recs:79                aws_avg = response['Datapoints'][count]['Average']80                aws_max = response['Datapoints'][count]['Maximum']81                aws_min = response['Datapoints'][count]['Minimum']82                aws_sum = response['Datapoints'][count]['Sum']83                aws_smp = response['Datapoints'][count]['SampleCount']84                aws_tsp = response['Datapoints'][count]['Timestamp']85                aws_unt = response['Datapoints'][count]['Unit']86                #print line87                if aws_tsp is not None:88                    pattern = '%Y-%m-%d %H:%M:%S+00:00'89                    aws_tsp = str(int(time.mktime(time.strptime(str(aws_tsp), pattern))))                90                    line = function_name + ',' + metric + ',' + aws_unt + ' ' + 'minimum=' + str(aws_min) + ',' + 'maximum=' + str(aws_max) + ',' + 'average=' + str(aws_avg) + 'sum=' + str(aws_sum) + ',' + 'sample_count=' + str(aws_smp) + ' ' + aws_tsp + '000000000'91                    data = [92                        {93                        "measurement": "LambdaMetrics",94                        "tags": {95                                    "name": function_name,96                                    "unit": aws_unt,97                                    "env": "qa",98                                    "region": REGION_NAME,99                                    "metric": metric,100                                },101                        "time": str(aws_tsp),102                        "fields": 103                                {104                                    "minimum": float(aws_min),105                                    "maximum": float(aws_max),106                                    "average": float(aws_avg),107                                    "sum": float(aws_sum),108                                    "sample": float(aws_smp)109                                }110                            }111                        ]112                    print line113                    print data114                    #time.sleep(1)115                    #influx_client.write_points(data)116                    #print line117                    logger.info('Sending: %s', line)118                count += 1119    return None120#@app.periodic_task(run_every=timedelta(seconds=60))121@app.task122def main():123    REGIONS = ['us-east-1','us-west-2']124    for REGION in REGIONS:125        logger.info('Region: %s', REGION)126        lambda_client = boto3.client('lambda', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key,aws_session_token=aws_session_token,region_name=REGION)127        funcs = lambda_client.list_functions()128        #pprint.pprint(funcs)129        count = 0130        total_fcn = len(funcs['Functions'])131    for key, value in funcs.items():132        if key == 'Functions':133            while count < total_fcn:134                #print 'Function Name: %s Description: %s | Runtime: %s | MemorySize: %s' %(value[count]['FunctionName'],value[count]['Description'],value[count]['Runtime'],value[count]['MemorySize'])135                #print value[count]['FunctionName']136                if re.findall(r'h2o',value[count]['FunctionName'],re.IGNORECASE):137                    logger.info('Invoking get_lambda_metrics for %s', value[count]['FunctionName'])138                    get_lambda_metrics(value[count]['FunctionName'],REGION)139                count += 1140if __name__=='__main__':...

verify_lambda.py

Source:verify_lambda.py

...11    }12    if 'AWS_SESSION_TOKEN' in os.environ:13        options['aws_session_token'] = os.environ['AWS_SESSION_TOKEN']14    return boto3.client(service, **options)15def get_lambda_metrics(metric):16    lambda_name = os.environ['LAMBDA']17    metrics = cw.get_metric_statistics(18        Namespace='AWS/Lambda',19        MetricName=metric,20        Dimensions=[{'Name': 'FunctionName', 'Value': lambda_name}],21        Period=60,22        Statistics=['Sum'],23        StartTime=datetime.today() - timedelta(hours=1),24        EndTime=datetime.today(),25    )26    return sum([x['Sum'] for x in metrics['Datapoints']])27def check_instance(instance, launch_config):28    return all([29        launch_config == instance.get('LaunchConfigurationName'),30        instance['HealthStatus'] == 'Healthy',31        instance['LifecycleState'] == 'InService'32    ])33autoscaling = get_client('autoscaling')34cw = get_client('cloudwatch')35print("Checking Autoscaling Group instances...\n")36asg_name = os.environ['ASG']37endtime = datetime.today() + timedelta(minutes=10)38while datetime.today() < endtime:39    response = autoscaling.describe_auto_scaling_groups(AutoScalingGroupNames=[asg_name])40    asg = response['AutoScalingGroups'][0]41    instance_count = len([x for x in asg['Instances'] if check_instance(x, asg['LaunchConfigurationName'])])42    if instance_count == asg['DesiredCapacity']:43        print("{} instances updated and healthy.  Proceeding to invocation checks...\n".format(instance_count))44        break45    print("{} of {} instances updated and healthy.  Timeout in {}".format(instance_count, asg['DesiredCapacity'],46                                                                          endtime - datetime.today()))47    time.sleep(15)48else:49    sys.exit("Error: Timed out waiting for ASG instances to update")50lambda_errors = get_lambda_metrics('Errors')51lambda_invocations = get_lambda_metrics('Invocations')52print("{} out of {} invocations generated errors\n".format(lambda_errors, lambda_invocations))53if lambda_errors > 0:54    sys.exit("Error: Lambda function encountered errors.")55else:...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.