KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use get_metric_widget_image method in localstack

Best Python code snippet using localstack_python

cloudwatch.py

Source:cloudwatch.py

...58        ec2_image["metrics"].append(59            ["AWS/EC2", "CPUUtilization", "InstanceId", instanceid])60        ec2_status_instance["metrics"].append(61            ["AWS/EC2", "StatusCheckFailed_Instance", "InstanceId", instanceid, {"stat": "Sum"}])62    response = cloudwatch.get_metric_widget_image(63        MetricWidget=json.dumps(ec2_image))64    response_status = cloudwatch.get_metric_widget_image(65        MetricWidget=json.dumps(ec2_status_instance))66    ec2_image_encode = (base64.b64encode(bytes(response["MetricWidgetImage"])))67    ec2_image_final = ec2_image_encode.decode()68    ec2_instance_image_encode = (base64.b64encode(69        bytes(response_status["MetricWidgetImage"])))70    ec2_instance_image_final = ec2_instance_image_encode.decode()71    metrics.append({"cpuutilization": ec2_image_final,72                    "instance_status": ec2_instance_image_final})73    return jsonify(metrics)74@cloudwatch_bp.route('/cloudwatch/metrics/rds')75def get_metric_widget_image_rds():76    region = request.args.get('region')77    cloudwatch = boto3.client('cloudwatch', region_name=region)78    metrics = []79    # rds_connections_image = """{{80    #     "metrics": [["AWS/RDS", "DatabaseConnections"]],81    #     "view": "singleValue",82    #     "stacked": false,83    #     "region": "{region}",84    #     "stat": "Average",85    #     "period": 300,86    #     "start": "-PT3H",87    #      "end": "P0D",88    #     "title": "RDS DB Connections"89    # }}""".format(region=region)90    rds_connections_image = {91        "view": "timeSeries",92        "stacked": False,93        "metrics": [],94        "width": 600,95        "height": 300,96        "start": "-PT3H",97        "end": "P0D",98        "title": "RDS DB Connections",99    }100    rds_cpu_image = {101        "view": "timeSeries",102        "stacked": False,103        "metrics": [],104        "width": 600,105        "height": 300,106        "start": "-PT3H",107        "end": "P0D",108        "title": "RDS CPU Utilization",109    }110    for rds_identifier in get_rds_id(region):111        rds_connections_image["metrics"].append(112            ["AWS/RDS", "DatabaseConnections", "DBInstanceIdentifier", rds_identifier])113        rds_cpu_image["metrics"].append(114            ["AWS/RDS", "CPUUtilization", "DBInstanceIdentifier", rds_identifier])115    response_conn = cloudwatch.get_metric_widget_image(116        MetricWidget=json.dumps(rds_connections_image))117    response_cpu = cloudwatch.get_metric_widget_image(118        MetricWidget=json.dumps(rds_cpu_image))119    rds_cpu_image_encode = (120        base64.b64encode(bytes(response_cpu["MetricWidgetImage"])))121    rds_cpu_image_final = rds_cpu_image_encode.decode()122    rds_connections_image_encode = (123        base64.b64encode(bytes(response_conn["MetricWidgetImage"])))124    rds_connections_image_final = rds_connections_image_encode.decode()125    metrics.append({"DbConnections": rds_connections_image_final,126                    "CPUUtilization": rds_cpu_image_final})...

workerListService.py

Source:workerListService.py

...15        instances = []16        instance_id = "i-0b1961797c1693607"17        instances.append(instance_id)18        CPUUtilization_REQUEST["metrics"][0][3] = instance_id19        cpu_response = self.CLOUD_WATCH.get_metric_widget_image(MetricWidget=json.dumps(CPUUtilization_REQUEST))20        cpu_chart = base64.b64encode(cpu_response['MetricWidgetImage']).decode("utf-8")21        cpu_charts.append(cpu_chart)22        Requests_bulabula["metrics"][0][3] = instance_id23        Requests_in_response = self.CLOUD_WATCH.get_metric_widget_image(24            MetricWidget=json.dumps(Requests_bulabula))25        RequestCount_in_chart = base64.b64encode(Requests_in_response['MetricWidgetImage']).decode("utf-8")26        request_charts.append(RequestCount_in_chart)27        return instances, cpu_charts, request_charts28    def get_charts(self):29        target_group = self.CLIENT.describe_target_health(TargetGroupArn=current_app.config["TARGET_GROUP_ARN"])30        cpu_charts = []31        request_charts = []32        instances = []33        for target in target_group['TargetHealthDescriptions']:34            if target['TargetHealth']['State'] != 'draining':35                instance_id = target['Target']['Id']36                instances.append(instance_id)37                CPUUtilization_REQUEST["metrics"][0][3] = instance_id38                cpu_response = self.CLOUD_WATCH.get_metric_widget_image(MetricWidget=json.dumps(CPUUtilization_REQUEST))39                cpu_chart = base64.b64encode(cpu_response['MetricWidgetImage']).decode("utf-8")40                cpu_charts.append(cpu_chart)41                # Hidden42                # NETWORK_PACKETS_IN_REQUEST["metrics"][0][3] = instance_id43                # network_packets_in_response = self.CLOUD_WATCH.get_metric_widget_image(44                # MetricWidget=json.dumps(NETWORK_PACKETS_IN_REQUEST))45                # network_packet_in_chart = base64.b64encode(network_packets_in_response['MetricWidgetImage']).decode("utf-8")46                # network_packets_in_charts.append(network_packet_in_chart)47                #LoadBalancer RequestCount48                #RequestCount_IN_REQUEST["metrics"][0][3] = instance_id49                # RequestCount_in_response = self.CLOUD_WATCH.get_metric_widget_image(50                # MetricWidget=json.dumps(RequestCount_IN_REQUEST))51                # RequestCount_in_chart = base64.b64encode(RequestCount_in_response['MetricWidgetImage']).decode("utf-8")52                # RequestCount_in_charts.append(RequestCount_in_chart)53                Requests_bulabula["metrics"][0][3] = instance_id54                Requests_in_response = self.CLOUD_WATCH.get_metric_widget_image(55                    MetricWidget=json.dumps(Requests_bulabula))56                RequestCount_in_chart = base64.b64encode(Requests_in_response['MetricWidgetImage']).decode("utf-8")57                request_charts.append(RequestCount_in_chart)...

app.py

Source:app.py

...19    bucket.download_file(config_key, config_path)20    with open(config_path, "r") as f:21        config = yaml.load(f, Loader=yaml.SafeLoader)22        return config23def get_metric_widget_image(item):24    params = item.copy()25    if params.get('start'):26        params['start'] = (datetime.datetime.now() - datetime.timedelta(minutes=int(item.get('start')))).isoformat()27    return cw.get_metric_widget_image(MetricWidget=json.dumps(params))28def get_message(format, item, image) -> str:29    return f"{{{format}}}".format(hex_image=image.hex(), title=item.get('title') or "")30def notify(message: str, topic: str):31    res = sns.publish(32        TopicArn=topic,33        Message=message,34    )35    logger.info(res)36    return res37def lambda_handler(event, context):38    config = get_config(config_bucket, config_key)39    global logger40    logger = log.get_logger(config.get('log_level'))41    items = config['items']42    format = config['message_format']43    for item in items:44        metric_image = get_metric_widget_image(item)45        message = get_message(format, item, metric_image['MetricWidgetImage'])46        notify(message, target_topic)47    return {48        'isBase64Encoded': False,49        'statusCode': 200,50        'headers': {},51        'body': {}...

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.