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

How to use put_metric_filter method in localstack

Best Python code snippet using localstack_python

test_createlogmetricfilterandalarm.py

Source:test_createlogmetricfilterandalarm.py

...83        }84    )85    logs_stubber.activate()86    mocker.patch('CreateLogMetricFilterAndAlarm.get_service_client', return_value = logs )87    logMetricAlarm.put_metric_filter(88        event['LogGroupName'], event['FilterName'], event['FilterPattern'],89        event['MetricName'], event['MetricNamespace'], event['MetricValue']90    )91    assert logs_stubber.assert_no_pending_responses() is None92    logs_stubber.deactivate()93def test_put_metric_filter_error(mocker):94    event = {95        'FilterName': 'test_filter',96        'FilterPattern': 'test_pattern',97        'MetricName': 'test_metric',98        'MetricNamespace': 'test_metricnamespace',99        'MetricValue': 'test_metric_value',100        'AlarmName': 'test_alarm',101        'AlarmDesc': 'alarm_desc',102        'AlarmThreshold': 'alarm_threshold',103        'LogGroupName': 'test_log',104        'TopicArn': 'arn:aws:sns:us-east-1:111111111111:test-topic-name'105    }106    BOTO_CONFIG = Config(107        retries={108            'mode': 'standard'109        },110        region_name=my_region111    )112    logs = botocore.session.get_session().create_client('logs', config=BOTO_CONFIG)113    logs_stubber = Stubber(logs)114    logs_stubber.add_client_error(115        'put_metric_filter',116        'CannotAddFilter'117    )118    logs_stubber.activate()119    mocker.patch('CreateLogMetricFilterAndAlarm.get_service_client', return_value=logs)120    with pytest.raises(SystemExit) as pytest_wrapped_exception:121        logMetricAlarm.put_metric_filter(122            event['LogGroupName'], event['FilterName'], event['FilterPattern'],123            event['MetricName'], event['MetricNamespace'], event['MetricValue']124        )125    assert pytest_wrapped_exception.type == SystemExit126def test_put_metric_alarm(mocker):127    event = {128        'FilterName': 'test_filter',129        'FilterPattern': 'test_pattern',130        'MetricName': 'test_metric',131        'MetricNamespace': 'test_metricnamespace',132        'MetricValue': 'test_metric_value',133        'AlarmName': 'test_alarm',134        'AlarmDesc': 'alarm_desc',135        'AlarmThreshold': 1,...

CreateLogMetricFilterAndAlarm.py

Source:CreateLogMetricFilterAndAlarm.py

...33    :return: service client34    """35    log.debug("Getting the service client for service: {}".format(service_name))36    return boto3.client(service_name, config=boto_config)37def put_metric_filter(cw_log_group, filter_name, filter_pattern, metric_name, metric_namespace, metric_value):38    """39    Puts the metric filter on the CloudWatch log group with provided values40    :param cw_log_group: Name of the CloudWatch log group41    :param filter_name: Name of the filter42    :param filter_pattern: Pattern for the filter43    :param metric_name: Name of the metric44    :param metric_namespace: Namespace where metric is logged45    :param metric_value: Value to be logged for the metric46    """47    logs_client = get_service_client('logs')48    log.debug("Putting the metric filter with values: {}".format([49        cw_log_group, filter_name, filter_pattern, metric_name, metric_namespace, metric_value]))50    try:51        logs_client.put_metric_filter(52            logGroupName=cw_log_group,53            filterName=filter_name,54            filterPattern=filter_pattern,55            metricTransformations=[56                {57                    'metricName': metric_name,58                    'metricNamespace': metric_namespace,59                    'metricValue': str(metric_value),60                    'unit': 'Count'61                }62            ]63        )64    except Exception as e:65        exit("Exception occurred while putting metric filter: " + str(e))66    log.debug("Successfully added the metric filter.")67def put_metric_alarm(alarm_name, alarm_desc, alarm_threshold, metric_name, metric_namespace, topic_arn):68    """69    Puts the metric alarm for the metric name with provided values70    :param alarm_name: Name for the alarm71    :param alarm_desc: Description for the alarm72    :param alarm_threshold: Threshold value for the alarm73    :param metric_name: Name of the metric74    :param metric_namespace: Namespace where metric is logged75    """76    cw_client = get_service_client('cloudwatch')77    log.debug("Putting the metric alarm with values {}".format(78        [alarm_name, alarm_desc, alarm_threshold, metric_name, metric_namespace]))79    try:80        cw_client.put_metric_alarm(81            AlarmName=alarm_name,82            AlarmDescription=alarm_desc,83            ActionsEnabled=True,84            OKActions=[85                topic_arn86            ],87            AlarmActions=[88                topic_arn89            ],90            MetricName=metric_name,91            Namespace=metric_namespace,92            Statistic='Sum',93            Period=300,94            Unit='Count',95            EvaluationPeriods=12,96            DatapointsToAlarm=1,97            Threshold=alarm_threshold,98            ComparisonOperator='GreaterThanOrEqualToThreshold',99            TreatMissingData='notBreaching'100        )101    except Exception as e:102        exit("Exception occurred while putting metric alarm: " + str(e))103    log.debug("Successfully added metric alarm.")104def verify(event, context):105    log.info("Begin handler")106    log.debug("====Print Event====")107    log.debug(event)108    filter_name = event['FilterName']109    filter_pattern = event['FilterPattern']110    metric_name = event['MetricName']111    metric_namespace = event['MetricNamespace']112    metric_value = event['MetricValue']113    alarm_name = event['AlarmName']114    alarm_desc = event['AlarmDesc']115    alarm_threshold = event['AlarmThreshold']116    cw_log_group = event['LogGroupName']117    topic_arn = event['TopicArn']118    put_metric_filter(cw_log_group, filter_name, filter_pattern, metric_name, metric_namespace, metric_value)119    put_metric_alarm(alarm_name, alarm_desc, alarm_threshold, metric_name, metric_namespace, topic_arn)120    return {121        "response": {122            "message": f'Created filter {event["FilterName"]} for metric {event["MetricName"]}, and alarm {event["AlarmName"]}',123            "status": "Success"124        }...

filter.py

Source:filter.py

1# Copyright 2015 Isotoma Limited2#3# Licensed under the Apache License, Version 2.0 (the "License");4# you may not use this file except in compliance with the License.5# You may obtain a copy of the License at6#7#   http://www.apache.org/licenses/LICENSE-2.08#9# Unless required by applicable law or agreed to in writing, software10# distributed under the License is distributed on an "AS IS" BASIS,11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12# See the License for the specific language governing permissions and13# limitations under the License.14from touchdown.core import argument, serializers15from touchdown.core.plan import Plan, Present16from ..common import Resource, SimpleApply, SimpleDescribe, SimpleDestroy17from .group import LogGroup18class Transformation(Resource):19    resource_name = "transformation"20    name = argument.String(field="metricName")21    namespace = argument.String(field="metricNamespace")22    value = argument.String(field="metricValue")23class Filter(Resource):24    resource_name = "filter"25    name = argument.String(min=1, max=512, field="filterName")26    log_group = argument.Resource(LogGroup, field="logGroupName", update=False)27    pattern = argument.String(min=1, max=512, field="filterPattern")28    transformations = argument.ResourceList(29        Transformation,30        min=1,31        field="metricTransformations",32        serializer=serializers.List(serializers.Resource()),33    )34class Describe(SimpleDescribe, Plan):35    resource = Filter36    service_name = "logs"37    api_version = "2014-03-28"38    describe_action = "describe_metric_filters"39    describe_notfound_exception = "ResourceNotFoundException"40    describe_envelope = "metricFilters"41    key = "filterName"42    def get_describe_filters(self):43        return {44            "logGroupName": self.resource.log_group.name,45            "filterNamePrefix": self.resource.name,46        }47class Apply(SimpleApply, Describe):48    signature = (Present("name"), Present("pattern"), Present("transformations"))49    create_action = "put_metric_filter"50    update_action = "put_metric_filter"51    create_response = "nothing-useful"52class Destroy(SimpleDestroy, Describe):53    destroy_action = "delete_metric_filter"54    def get_destroy_serializer(self):55        return serializers.Dict(56            logGroupName=self.resource.log_group.name, filterName=self.resource.name...

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.