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

How to use describe_alarms_for_metric method in localstack

Best Python code snippet using localstack_python

Controls.py

Source:Controls.py

...1189                                    command='describe_alarms_for_metric',1190                                    kwargs=log_kwargs1191                                )1192                                #cwclient = boto3.client('cloudwatch', region_name=m)1193                                #response = cwclient.describe_alarms_for_metric(1194                                #     MetricName=p['metricTransformations'][0]['metricName'],1195                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1196                                # )1197                                # snsClient = boto3.client('sns', region_name=m)1198                                # subscribers = snsClient.list_subscriptions_by_topic(1199                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1200                                    #  Pagination not used since only 1 subscriber required1201                                sns_kwargs={1202                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1203                                }1204                                subscribers = self.connection_manager.call(1205                                    service='sns',1206                                    command='list_subscriptions_by_topic',1207                                    kwargs=sns_kwargs1208                                )1209                                if not len(subscribers['Subscriptions']) == 0:1210                                    result = True1211                except:1212                    pass1213        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1214                'Description': description, 'ControlId': control}1215    # 3.2 Ensure a log metric filter and alarm exist for Management Console sign-in without MFA (Scored)1216    def control_3_2_ensure_log_metric_filter_console_signin_no_mfa(self,cloudtrails):1217        """Summary1218        Returns:1219            TYPE: Description1220        """1221        result = False1222        failReason = ""1223        offenders = []1224        control = "3.2"1225        description = "Ensure a log metric filter and alarm exist for Management Console sign-in without MFA"1226        scored = True1227        failReason = "Incorrect log metric alerts for management console signin without MFA"1228        for m, n in cloudtrails.items():1229            self.setRegion(m, iam_role=None)1230            for o in n:1231                try:1232                    if o['CloudWatchLogsLogGroupArn']:1233                        metric_kwargs={1234                            "logGroupName": group1235                        }1236                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1237                        # client = boto3.client('logs', region_name=m)1238                        # filters = client.describe_metric_filters(1239                        #     logGroupName=group1240                        # )1241                        filters = self.connection_manager.call(1242                            service='logs',1243                            command='describe_metric_filters',1244                            kwargs=metric_kwargs1245                        )1246                        for p in filters['metricFilters']:1247                            patterns = ["\$\.eventName\s*=\s*\"?ConsoleLogin(\"|\)|\s)",1248                                        "\$\.additionalEventData\.MFAUsed\s*\!=\s*\"?Yes"]1249                            if find_in_string(patterns, str(p['filterPattern'])):1250                                log_kwargs = {1251                                    "MetricName": p['metricTransformations'][0]['metricName'],1252                                    "Namespace": p['metricTransformations'][0]['metricNamespace']1253                                }1254                                # cwclient = boto3.client('cloudwatch', region_name=m)1255                                # response = cwclient.describe_alarms_for_metric(1256                                #     MetricName=p['metricTransformations'][0]['metricName'],1257                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1258                                # )1259                                response = self.connection_manager.call(1260                                    service='cloudwatch',1261                                    command='describe_alarms_for_metric',1262                                    kwargs=log_kwargs1263                                )1264                                # snsClient = boto3.client('sns', region_name=m)1265                                # subscribers = snsClient.list_subscriptions_by_topic(1266                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1267                                #1268                                # )1269                                sns_kwargs = {1270                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1271                                }1272                                subscribers = self.connection_manager.call(1273                                    service='sns',1274                                    command='list_subscriptions_by_topic',1275                                    kwargs=sns_kwargs1276                                )1277                                #  Pagination not used since only 1 subscriber required1278                                if not len(subscribers['Subscriptions']) == 0:1279                                    result = True1280                except:1281                    pass1282        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1283                'Description': description, 'ControlId': control}1284    # 3.3 Ensure a log metric filter and alarm exist for usage of "root" account (Scored)1285    def control_3_3_ensure_log_metric_filter_root_usage(self,cloudtrails):1286        """Summary1287        Returns:1288            TYPE: Description1289        """1290        result = False1291        failReason = ""1292        offenders = []1293        control = "3.3"1294        description = "Ensure a log metric filter and alarm exist for root usage"1295        scored = True1296        failReason = "Incorrect log metric alerts for root usage"1297        for m, n in cloudtrails.items():1298            self.setRegion(m, iam_role=None)1299            for o in n:1300                try:1301                    if o['CloudWatchLogsLogGroupArn']:1302                        metric_kwargs = {1303                            "logGroupName": group1304                        }1305                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1306                        # client = boto3.client('logs', region_name=m)1307                        # filters = client.describe_metric_filters(1308                        #     logGroupName=group1309                        # )1310                        filters = self.connection_manager.call(1311                            service='logs',1312                            command='describe_metric_filters',1313                            kwargs=metric_kwargs1314                        )1315                        for p in filters['metricFilters']:1316                            patterns = ["\$\.userIdentity\.type\s*=\s*\"?Root",1317                                        "\$\.userIdentity\.invokedBy\s*NOT\s*EXISTS",1318                                        "\$\.eventType\s*\!=\s*\"?AwsServiceEvent(\"|\)|\s)"]1319                            if find_in_string(patterns, str(p['filterPattern'])):1320                                # cwclient = boto3.client('cloudwatch', region_name=m)1321                                # response = cwclient.describe_alarms_for_metric(1322                                #     MetricName=p['metricTransformations'][0]['metricName'],1323                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1324                                # )1325                                log_kwargs = {1326                                    "MetricName": p['metricTransformations'][0]['metricName'],1327                                    "Namespace": p['metricTransformations'][0]['metricNamespace']1328                                }1329                                response = self.connection_manager.call(1330                                    service='cloudwatch',1331                                    command='describe_alarms_for_metric',1332                                    kwargs=log_kwargs1333                                )1334                                # snsClient = boto3.client('sns', region_name=m)1335                                # subscribers = snsClient.list_subscriptions_by_topic(1336                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1337                                #     #  Pagination not used since only 1 subscriber required1338                                # )1339                                sns_kwargs = {1340                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1341                                }1342                                subscribers = self.connection_manager.call(1343                                    service='sns',1344                                    command='list_subscriptions_by_topic',1345                                    kwargs=sns_kwargs1346                                )1347                                #  Pagination not used since only 1 subscriber required1348                                if not len(subscribers['Subscriptions']) == 0:1349                                    result = True1350                except:1351                    pass1352        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1353                'Description': description, 'ControlId': control}1354    # 3.4 Ensure a log metric filter and alarm exist for IAM policy changes  (Scored)1355    def control_3_4_ensure_log_metric_iam_policy_change(self, cloudtrails):1356        """Summary1357        Returns:1358            TYPE: Description1359        """1360        result = False1361        failReason = ""1362        offenders = []1363        control = "3.4"1364        description = "Ensure a log metric filter and alarm exist for IAM changes"1365        scored = True1366        failReason = "Incorrect log metric alerts for IAM policy changes"1367        for m, n in cloudtrails.items():1368            self.setRegion(m, iam_role=None)1369            for o in n:1370                try:1371                    if o['CloudWatchLogsLogGroupArn']:1372                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1373                        cloud_kwargs = {1374                            'logGroupName': group,1375                        }1376                        filters = self.connection_manager.call(1377                            service='logs',1378                            command='describe_metric_filters',1379                            kwargs=cloud_kwargs1380                        )1381                        """1382                        client = connection_manager.client('logs', region_name=m)1383                        filters = client.describe_metric_filters(1384                            logGroupName=group1385                        )"""1386                        for p in filters['metricFilters']:1387                            patterns = ["\$\.eventName\s*=\s*\"?DeleteGroupPolicy(\"|\)|\s)",1388                                        "\$\.eventName\s*=\s*\"?DeleteRolePolicy(\"|\)|\s)",1389                                        "\$\.eventName\s*=\s*\"?DeleteUserPolicy(\"|\)|\s)",1390                                        "\$\.eventName\s*=\s*\"?PutGroupPolicy(\"|\)|\s)",1391                                        "\$\.eventName\s*=\s*\"?PutRolePolicy(\"|\)|\s)",1392                                        "\$\.eventName\s*=\s*\"?PutUserPolicy(\"|\)|\s)",1393                                        "\$\.eventName\s*=\s*\"?CreatePolicy(\"|\)|\s)",1394                                        "\$\.eventName\s*=\s*\"?DeletePolicy(\"|\)|\s)",1395                                        "\$\.eventName\s*=\s*\"?CreatePolicyVersion(\"|\)|\s)",1396                                        "\$\.eventName\s*=\s*\"?DeletePolicyVersion(\"|\)|\s)",1397                                        "\$\.eventName\s*=\s*\"?AttachRolePolicy(\"|\)|\s)",1398                                        "\$\.eventName\s*=\s*\"?DetachRolePolicy(\"|\)|\s)",1399                                        "\$\.eventName\s*=\s*\"?AttachUserPolicy(\"|\)|\s)",1400                                        "\$\.eventName\s*=\s*\"?DetachUserPolicy(\"|\)|\s)",1401                                        "\$\.eventName\s*=\s*\"?AttachGroupPolicy(\"|\)|\s)",1402                                        "\$\.eventName\s*=\s*\"?DetachGroupPolicy(\"|\)|\s)"]1403                            if find_in_string(patterns, str(p['filterPattern'])):1404                                MetricName = p['metricTransformations'][0]['metricName'],1405                                Namespace = p['metricTransformations'][0]['metricNamespace']1406                                cloud_kwargs = {1407                                    'MetricName': MetricName,1408                                    'Namespace': Namespace,1409                                }1410                                response = self.connection_manager.call(1411                                    service='cloudwatch',1412                                    command='describe_alarms_for_metric',1413                                    kwargs=cloud_kwargs1414                                )1415                                """1416                                cwclient = connection_manager.client('cloudwatch', region_name=m)1417                                response = cwclient.describe_alarms_for_metric(1418                                    MetricName=p['metricTransformations'][0]['metricName'],1419                                    Namespace=p['metricTransformations'][0]['metricNamespace']1420                                )"""1421                                TopicArn = response['MetricAlarms'][0]['AlarmActions'][0]1422                                cloud_kwargs = {1423                                    'TopicArn': TopicArn,1424                                }1425                                subscribers = self.connection_manager.call(1426                                    service='sns',1427                                    command='list_subscriptions_by_topic',1428                                    kwargs=cloud_kwargs1429                                )1430                                """1431                                snsClient = connection_manager.client('sns', region_name=m)1432                                subscribers = snsClient.list_subscriptions_by_topic(1433                                    TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1434                                    #  Pagination not used since only 1 subscriber required1435                                )"""1436                                if not len(subscribers['Subscriptions']) == 0:1437                                    result = True1438                except:1439                    pass1440        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1441                'Description': description, 'ControlId': control}1442    # 3.5 Ensure a log metric filter and alarm exist for CloudTrail configuration changes (Scored)1443    def control_3_5_ensure_log_metric_cloudtrail_configuration_changes(self, cloudtrails):1444        """Summary1445        Returns:1446            TYPE: Description1447        """1448        result = False1449        failReason = ""1450        offenders = []1451        control = "3.5"1452        description = "Ensure a log metric filter and alarm exist for CloudTrail configuration changes"1453        scored = True1454        failReason = "Incorrect log metric alerts for CloudTrail configuration changes"1455        for m, n in cloudtrails.items():1456            self.setRegion(m, iam_role=None)1457            for o in n:1458                try:1459                    if o['CloudWatchLogsLogGroupArn']:1460                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1461                        cloud_kwargs = {1462                            'logGroupName': group,1463                        }1464                        filters = self.connection_manager.call(1465                            service='logs',1466                            command='describe_metric_filters',1467                            kwargs=cloud_kwargs1468                        )1469                        """client = connection_manager.client('logs', region_name=m)1470                        filters = client.describe_metric_filters(1471                            logGroupName=group1472                        )"""1473                        for p in filters['metricFilters']:1474                            patterns = ["\$\.eventName\s*=\s*\"?CreateTrail(\"|\)|\s)",1475                                        "\$\.eventName\s*=\s*\"?UpdateTrail(\"|\)|\s)",1476                                        "\$\.eventName\s*=\s*\"?DeleteTrail(\"|\)|\s)",1477                                        "\$\.eventName\s*=\s*\"?StartLogging(\"|\)|\s)",1478                                        "\$\.eventName\s*=\s*\"?StopLogging(\"|\)|\s)"]1479                            if find_in_string(patterns, str(p['filterPattern'])):1480                                MetricName = p['metricTransformations'][0]['metricName'],1481                                Namespace = p['metricTransformations'][0]['metricNamespace']1482                                cloud_kwargs = {1483                                    'MetricName': MetricName,1484                                    'Namespace': Namespace1485                                }1486                                response = self.connection_manager.call(1487                                    service='cloudwatch',1488                                    command='describe_alarms_for_metric',1489                                    kwargs=cloud_kwargs1490                                )1491                                """cwclient = connection_manager.client('cloudwatch', region_name=m)1492                                response = cwclient.describe_alarms_for_metric(1493                                    MetricName=p['metricTransformations'][0]['metricName'],1494                                    Namespace=p['metricTransformations'][0]['metricNamespace']1495                                )"""1496                                TopicArn = response['MetricAlarms'][0]['AlarmActions'][0]1497                                m = 'us-east-1'1498                                cloud_kwargs = {1499                                    'TopicArn': TopicArn,1500                                    'region_name': m1501                                }1502                                subscribers = self.connection_manager.call(1503                                    service='sns',1504                                    command='list_subscriptions_by_topic',1505                                    kwargs=cloud_kwargs1506                                )1507                                """1508                                snsClient = connection_manager.client('sns', region_name=m)1509                                subscribers = snsClient.list_subscriptions_by_topic(1510                                    TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1511                                    #  Pagination not used since only 1 subscriber required1512                                )"""1513                                if not len(subscribers['Subscriptions']) == 0:1514                                    result = True1515                except:1516                    pass1517        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1518                'Description': description, 'ControlId': control}1519    # 3.6 Ensure a log metric filter and alarm exist for AWS Management Console authentication failures (Scored)1520    def control_3_6_ensure_log_metric_console_auth_failures(self, cloudtrails):1521        """Summary1522        Returns:1523            TYPE: Description1524        """1525        result = False1526        failReason = ""1527        offenders = []1528        control = "3.6"1529        description = "Ensure a log metric filter and alarm exist for console auth failures"1530        scored = True1531        failReason = "Ensure a log metric filter and alarm exist for console auth failures"1532        for m, n in cloudtrails.items():1533            self.setRegion(m, iam_role=None)1534            for o in n:1535                try:1536                    if o['CloudWatchLogsLogGroupArn']:1537                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1538                        cloud_kwargs = {1539                            'logGroupName': group,1540                        }1541                        filters = self.connection_manager.call(1542                            service='logs',1543                            command='describe_metric_filters',1544                            kwargs=cloud_kwargs1545                        )1546                        """1547                        client = connection_manager.client('logs', region_name=m)1548                        filters = client.describe_metric_filters(1549                            logGroupName=group1550                        )"""1551                        for p in filters['metricFilters']:1552                            patterns = ["\$\.eventName\s*=\s*\"?ConsoleLogin(\"|\)|\s)",1553                                        "\$\.errorMessage\s*=\s*\"?Failed authentication(\"|\)|\s)"]1554                            if find_in_string(patterns, str(p['filterPattern'])):1555                                MetricName = p['metricTransformations'][0]['metricName'],1556                                Namespace = p['metricTransformations'][0]['metricNamespace']1557                                cloud_kwargs = {1558                                    'MetricName': MetricName,1559                                    'Namespace': Namespace1560                                }1561                                response = self.connection_manager.call(1562                                    service='cloudwatch',1563                                    command='describe_alarms_for_metric',1564                                    kwargs=cloud_kwargs1565                                )1566                                """1567                                cwclient = connection_manager.client('cloudwatch', region_name=m)1568                                response = cwclient.describe_alarms_for_metric(1569                                    MetricName=p['metricTransformations'][0]['metricName'],1570                                    Namespace=p['metricTransformations'][0]['metricNamespace']1571                                )"""1572                                TopicArn = response['MetricAlarms'][0]['AlarmActions'][0]1573                                cloud_kwargs = {1574                                    'TopicArn': TopicArn,1575                                }1576                                subscribers = self.connection_manager.call(1577                                    service='sns',1578                                    command='list_subscriptions_by_topic',1579                                    kwargs=cloud_kwargs1580                                )1581                                """1582                                snsClient = connection_manager.client('sns', region_name=m)1583                                subscribers = snsClient.list_subscriptions_by_topic(1584                                    TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1585                                    #  Pagination not used since only 1 subscriber required1586                                )"""1587                                if not len(subscribers['Subscriptions']) == 0:1588                                    result = True1589                except:1590                    pass1591        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1592                'Description': description, 'ControlId': control}1593    # 3.7 Ensure a log metric filter and alarm exist for disabling or scheduled deletion of customer created CMKs (Scored)1594    def control_3_7_ensure_log_metric_disabling_scheduled_delete_of_kms_cmk(self, cloudtrails):1595        """Summary1596        Returns:1597            TYPE: Description1598        """1599        result = False1600        failReason = ""1601        offenders = []1602        control = "3.7"1603        description = "Ensure a log metric filter and alarm exist for disabling or scheduling deletion of KMS CMK"1604        scored = True1605        failReason = "Ensure a log metric filter and alarm exist for disabling or scheduling deletion of KMS CMK"1606        for m, n in cloudtrails.items():1607            self.setRegion(m, iam_role=None)1608            for o in n:1609                try:1610                    if o['CloudWatchLogsLogGroupArn']:1611                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1612                        cloud_kwargs = {1613                            'logGroupName': group,1614                        }1615                        filters = self.connection_manager.call(1616                            service='logs',1617                            command='describe_metric_filters',1618                            kwargs=cloud_kwargs1619                        )1620                        """1621                        client = connection_manager.client('logs', region_name=m)1622                        filters = client.describe_metric_filters(1623                            logGroupName=group1624                        )"""1625                        for p in filters['metricFilters']:1626                            patterns = ["\$\.eventSource\s*=\s*\"?kms\.amazonaws\.com(\"|\)|\s)",1627                                        "\$\.eventName\s*=\s*\"?DisableKey(\"|\)|\s)",1628                                        "\$\.eventName\s*=\s*\"?ScheduleKeyDeletion(\"|\)|\s)"]1629                            if find_in_string(patterns, str(p['filterPattern'])):1630                                MetricName = p['metricTransformations'][0]['metricName'],1631                                Namespace = p['metricTransformations'][0]['metricNamespace']1632                                cloud_kwargs = {1633                                    'MetricName': MetricName,1634                                    'Namespace': Namespace1635                                }1636                                response = self.connection_manager.call(1637                                    service='cloudwatch',1638                                    command='describe_alarms_for_metric',1639                                    kwargs=cloud_kwargs1640                                )1641                                """1642                                cwclient = connection_manager.client('cloudwatch', region_name=m)1643                                response = cwclient.describe_alarms_for_metric(1644                                    MetricName=p['metricTransformations'][0]['metricName'],1645                                    Namespace=p['metricTransformations'][0]['metricNamespace']1646                                )"""1647                                TopicArn = response['MetricAlarms'][0]['AlarmActions'][0]1648                                cloud_kwargs = {1649                                    'TopicArn': TopicArn,1650                                }1651                                subscribers = self.connection_manager.call(1652                                    service='sns',1653                                    command='list_subscriptions_by_topic',1654                                    kwargs=cloud_kwargs1655                                )1656                                """1657                                snsClient = connection_manager.client('sns', region_name=m)1658                                subscribers = snsClient.list_subscriptions_by_topic(1659                                    TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1660                                    #  Pagination not used since only 1 subscriber required1661                                )"""1662                                if not len(subscribers['Subscriptions']) == 0:1663                                    result = True1664                except:1665                    pass1666        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1667                'Description': description, 'ControlId': control}1668    # 3.8 Ensure a log metric filter and alarm exist for S3 bucket policy changes (Scored)1669    def control_3_8_ensure_log_metric_s3_bucket_policy_changes(self,cloudtrails):1670        """Summary1671        Returns:1672            TYPE: Description1673        """1674        result = False1675        failReason = ""1676        offenders = []1677        control = "3.8"1678        description = "Ensure a log metric filter and alarm exist for S3 bucket policy changes"1679        scored = True1680        failReason = "Ensure a log metric filter and alarm exist for S3 bucket policy changes"1681        for m, n in cloudtrails.items():1682            self.setRegion(m, iam_role=None)1683            for o in n:1684                try:1685                    if o['CloudWatchLogsLogGroupArn']:1686                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1687                        # client = boto3.client('logs', region_name=m)1688                        # filters = client.describe_metric_filters(1689                        #     logGroupName=group1690                        # )1691                        cloud_kwargs = {1692                            'logGroupName': group,1693                        }1694                        filters = self.connection_manager.call(1695                            service='logs',1696                            command='describe_metric_filters',1697                            kwargs=cloud_kwargs1698                        )1699                        for p in filters['metricFilters']:1700                            patterns = ["\$\.eventSource\s*=\s*\"?s3\.amazonaws\.com(\"|\)|\s)",1701                                        "\$\.eventName\s*=\s*\"?PutBucketAcl(\"|\)|\s)",1702                                        "\$\.eventName\s*=\s*\"?PutBucketPolicy(\"|\)|\s)",1703                                        "\$\.eventName\s*=\s*\"?PutBucketCors(\"|\)|\s)",1704                                        "\$\.eventName\s*=\s*\"?PutBucketLifecycle(\"|\)|\s)",1705                                        "\$\.eventName\s*=\s*\"?PutBucketReplication(\"|\)|\s)",1706                                        "\$\.eventName\s*=\s*\"?DeleteBucketPolicy(\"|\)|\s)",1707                                        "\$\.eventName\s*=\s*\"?DeleteBucketCors(\"|\)|\s)",1708                                        "\$\.eventName\s*=\s*\"?DeleteBucketLifecycle(\"|\)|\s)",1709                                        "\$\.eventName\s*=\s*\"?DeleteBucketReplication(\"|\)|\s)"]1710                            if find_in_string(patterns, str(p['filterPattern'])):1711                                # cwclient = boto3.client('cloudwatch', region_name=m)1712                                # response = cwclient.describe_alarms_for_metric(1713                                #     MetricName=p['metricTransformations'][0]['metricName'],1714                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1715                                # )1716                                cloud_kwargs = {1717                                    'MetricName': MetricName,1718                                    'Namespace': Namespace,1719                                }1720                                response = self.connection_manager.call(1721                                    service='cloudwatch',1722                                    command='describe_alarms_for_metric',1723                                    kwargs=cloud_kwargs1724                                )1725                                # snsClient = boto3.client('sns', region_name=m)1726                                # subscribers = snsClient.list_subscriptions_by_topic(1727                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1728                                #     #  Pagination not used since only 1 subscriber required1729                                # )1730                                sns_kwargs = {1731                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1732                                }1733                                subscribers = self.connection_manager.call(1734                                    service='sns',1735                                    command='list_subscriptions_by_topic',1736                                    kwargs=sns_kwargs1737                                )1738                                if not len(subscribers['Subscriptions']) == 0:1739                                    result = True1740                except:1741                    pass1742        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1743                'Description': description, 'ControlId': control}1744    # 3.9 Ensure a log metric filter and alarm exist for AWS Config configuration changes (Scored)1745    def control_3_9_ensure_log_metric_config_configuration_changes(self,cloudtrails):1746        """Summary1747        Returns:1748            TYPE: Description1749        """1750        result = False1751        failReason = ""1752        offenders = []1753        control = "3.9"1754        description = "Ensure a log metric filter and alarm exist for for AWS Config configuration changes"1755        scored = True1756        failReason = "Ensure a log metric filter and alarm exist for for AWS Config configuration changes"1757        for m, n in cloudtrails.items():1758            self.setRegion(m, iam_role=None)1759            for o in n:1760                try:1761                    if o['CloudWatchLogsLogGroupArn']:1762                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1763                        # client = boto3.client('logs', region_name=m)1764                        # filters = client.describe_metric_filters(1765                        #     logGroupName=group1766                        # )1767                        cloud_kwargs = {1768                            'logGroupName': group,1769                        }1770                        filters = self.connection_manager.call(1771                            service='logs',1772                            command='describe_metric_filters',1773                            kwargs=cloud_kwargs1774                        )1775                        for p in filters['metricFilters']:1776                            patterns = ["\$\.eventSource\s*=\s*\"?config\.amazonaws\.com(\"|\)|\s)",1777                                        "\$\.eventName\s*=\s*\"?StopConfigurationRecorder(\"|\)|\s)",1778                                        "\$\.eventName\s*=\s*\"?DeleteDeliveryChannel(\"|\)|\s)",1779                                        "\$\.eventName\s*=\s*\"?PutDeliveryChannel(\"|\)|\s)",1780                                        "\$\.eventName\s*=\s*\"?PutConfigurationRecorder(\"|\)|\s)"]1781                            if find_in_string(patterns, str(p['filterPattern'])):1782                                # cwclient = boto3.client('cloudwatch', region_name=m)1783                                # response = cwclient.describe_alarms_for_metric(1784                                #     MetricName=p['metricTransformations'][0]['metricName'],1785                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1786                                # )1787                                cloud_kwargs = {1788                                    'MetricName': MetricName,1789                                    'Namespace': Namespace,1790                                }1791                                response = self.connection_manager.call(1792                                    service='cloudwatch',1793                                    command='describe_alarms_for_metric',1794                                    kwargs=cloud_kwargs1795                                )1796                                # snsClient = boto3.client('sns', region_name=m)1797                                # subscribers = snsClient.list_subscriptions_by_topic(1798                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1799                                #     #  Pagination not used since only 1 subscriber required1800                                # )1801                                sns_kwargs = {1802                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1803                                }1804                                subscribers = self.connection_manager.call(1805                                    service='sns',1806                                    command='list_subscriptions_by_topic',1807                                    kwargs=sns_kwargs1808                                )1809                                if not len(subscribers['Subscriptions']) == 0:1810                                    result = True1811                except:1812                    pass1813        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1814                'Description': description, 'ControlId': control}1815    # 3.10 Ensure a log metric filter and alarm exist for security group changes (Scored)1816    def control_3_10_ensure_log_metric_security_group_changes(self,cloudtrails):1817        """Summary1818        Returns:1819            TYPE: Description1820        """1821        result = False1822        failReason = ""1823        offenders = []1824        control = "3.10"1825        description = "Ensure a log metric filter and alarm exist for security group changes"1826        scored = True1827        failReason = "Ensure a log metric filter and alarm exist for security group changes"1828        for m, n in cloudtrails.items():1829            self.setRegion(m, iam_role=None)1830            for o in n:1831                try:1832                    if o['CloudWatchLogsLogGroupArn']:1833                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1834                        # client = boto3.client('logs', region_name=m)1835                        # filters = client.describe_metric_filters(1836                        #     logGroupName=group1837                        # )1838                        cloud_kwargs = {1839                            'logGroupName': group,1840                        }1841                        filters = self.connection_manager.call(1842                            service='logs',1843                            command='describe_metric_filters',1844                            kwargs=cloud_kwargs1845                        )1846                        for p in filters['metricFilters']:1847                            patterns = ["\$\.eventName\s*=\s*\"?AuthorizeSecurityGroupIngress(\"|\)|\s)",1848                                        "\$\.eventName\s*=\s*\"?AuthorizeSecurityGroupEgress(\"|\)|\s)",1849                                        "\$\.eventName\s*=\s*\"?RevokeSecurityGroupIngress(\"|\)|\s)",1850                                        "\$\.eventName\s*=\s*\"?RevokeSecurityGroupEgress(\"|\)|\s)",1851                                        "\$\.eventName\s*=\s*\"?CreateSecurityGroup(\"|\)|\s)",1852                                        "\$\.eventName\s*=\s*\"?DeleteSecurityGroup(\"|\)|\s)"]1853                            if find_in_string(patterns, str(p['filterPattern'])):1854                                # cwclient = boto3.client('cloudwatch', region_name=m)1855                                # response = cwclient.describe_alarms_for_metric(1856                                #     MetricName=p['metricTransformations'][0]['metricName'],1857                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1858                                # )1859                                cloud_kwargs = {1860                                    'MetricName': MetricName,1861                                    'Namespace': Namespace,1862                                }1863                                response = self.connection_manager.call(1864                                    service='cloudwatch',1865                                    command='describe_alarms_for_metric',1866                                    kwargs=cloud_kwargs1867                                )1868                                # snsClient = boto3.client('sns', region_name=m)1869                                # subscribers = snsClient.list_subscriptions_by_topic(1870                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1871                                #     #  Pagination not used since only 1 subscriber required1872                                # )1873                                sns_kwargs = {1874                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1875                                }1876                                subscribers = self.connection_manager.call(1877                                    service='sns',1878                                    command='list_subscriptions_by_topic',1879                                    kwargs=sns_kwargs1880                                )1881                                if not len(subscribers['Subscriptions']) == 0:1882                                    result = True1883                except:1884                    pass1885        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1886                'Description': description, 'ControlId': control}1887    # 3.11 Ensure a log metric filter and alarm exist for changes to Network Access Control Lists (NACL) (Scored)1888    def control_3_11_ensure_log_metric_nacl(self,cloudtrails):1889        """Summary1890        Returns:1891            TYPE: Description1892        """1893        result = False1894        failReason = ""1895        offenders = []1896        control = "3.11"1897        description = "Ensure a log metric filter and alarm exist for changes to Network Access Control Lists (NACL)"1898        scored = True1899        failReason = "Ensure a log metric filter and alarm exist for changes to Network Access Control Lists (NACL)"1900        for m, n in cloudtrails.items():1901            self.setRegion(m, iam_role=None)1902            for o in n:1903                try:1904                    if o['CloudWatchLogsLogGroupArn']:1905                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1906                        # client = boto3.client('logs', region_name=m)1907                        # filters = client.describe_metric_filters(1908                        #     logGroupName=group1909                        # )1910                        cloud_kwargs = {1911                            'logGroupName': group,1912                        }1913                        filters = self.connection_manager.call(1914                            service='logs',1915                            command='describe_metric_filters',1916                            kwargs=cloud_kwargs1917                        )1918                        for p in filters['metricFilters']:1919                            patterns = ["\$\.eventName\s*=\s*\"?CreateNetworkAcl(\"|\)|\s)",1920                                        "\$\.eventName\s*=\s*\"?CreateNetworkAclEntry(\"|\)|\s)",1921                                        "\$\.eventName\s*=\s*\"?DeleteNetworkAcl(\"|\)|\s)",1922                                        "\$\.eventName\s*=\s*\"?DeleteNetworkAclEntry(\"|\)|\s)",1923                                        "\$\.eventName\s*=\s*\"?ReplaceNetworkAclEntry(\"|\)|\s)",1924                                        "\$\.eventName\s*=\s*\"?ReplaceNetworkAclAssociation(\"|\)|\s)"]1925                            if find_in_string(patterns, str(p['filterPattern'])):1926                                # cwclient = boto3.client('cloudwatch', region_name=m)1927                                # response = cwclient.describe_alarms_for_metric(1928                                #     MetricName=p['metricTransformations'][0]['metricName'],1929                                #     Namespace=p['metricTransformations'][0]['metricNamespace']1930                                # )1931                                cloud_kwargs = {1932                                    'MetricName': MetricName,1933                                    'Namespace': Namespace,1934                                }1935                                response = self.connection_manager.call(1936                                    service='cloudwatch',1937                                    command='describe_alarms_for_metric',1938                                    kwargs=cloud_kwargs1939                                )1940                                # snsClient = boto3.client('sns', region_name=m)1941                                # subscribers = snsClient.list_subscriptions_by_topic(1942                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]1943                                #     #  Pagination not used since only 1 subscriber required1944                                # )1945                                sns_kwargs = {1946                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]1947                                }1948                                subscribers = self.connection_manager.call(1949                                    service='sns',1950                                    command='list_subscriptions_by_topic',1951                                    kwargs=sns_kwargs1952                                )1953                                if not len(subscribers['Subscriptions']) == 0:1954                                    result = True1955                                if not len(subscribers['Subscriptions']) == 0:1956                                    result = True1957                except:1958                    pass1959        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,1960                'Description': description, 'ControlId': control}1961    # 3.12 Ensure a log metric filter and alarm exist for changes to network gateways (Scored)1962    def control_3_12_ensure_log_metric_changes_to_network_gateways(self,cloudtrails):1963        """Summary1964        Returns:1965            TYPE: Description1966        """1967        result = False1968        failReason = ""1969        offenders = []1970        control = "3.12"1971        description = "Ensure a log metric filter and alarm exist for changes to network gateways"1972        scored = True1973        failReason = "Ensure a log metric filter and alarm exist for changes to network gateways"1974        for m, n in cloudtrails.items():1975            self.setRegion(m, iam_role=None)1976            for o in n:1977                try:1978                    if o['CloudWatchLogsLogGroupArn']:1979                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)1980                        # client = boto3.client('logs', region_name=m)1981                        # filters = client.describe_metric_filters(1982                        #     logGroupName=group1983                        # )1984                        cloud_kwargs = {1985                            'logGroupName': group,1986                        }1987                        filters = self.connection_manager.call(1988                            service='logs',1989                            command='describe_metric_filters',1990                            kwargs=cloud_kwargs1991                        )1992                        for p in filters['metricFilters']:1993                            patterns = ["\$\.eventName\s*=\s*\"?CreateCustomerGateway(\"|\)|\s)",1994                                        "\$\.eventName\s*=\s*\"?DeleteCustomerGateway(\"|\)|\s)",1995                                        "\$\.eventName\s*=\s*\"?AttachInternetGateway(\"|\)|\s)",1996                                        "\$\.eventName\s*=\s*\"?CreateInternetGateway(\"|\)|\s)",1997                                        "\$\.eventName\s*=\s*\"?DeleteInternetGateway(\"|\)|\s)",1998                                        "\$\.eventName\s*=\s*\"?DetachInternetGateway(\"|\)|\s)"]1999                            if find_in_string(patterns, str(p['filterPattern'])):2000                                # cwclient = boto3.client('cloudwatch', region_name=m)2001                                # response = cwclient.describe_alarms_for_metric(2002                                #     MetricName=p['metricTransformations'][0]['metricName'],2003                                #     Namespace=p['metricTransformations'][0]['metricNamespace']2004                                # )2005                                cloud_kwargs = {2006                                    'MetricName': MetricName,2007                                    'Namespace': Namespace,2008                                }2009                                response = self.connection_manager.call(2010                                    service='cloudwatch',2011                                    command='describe_alarms_for_metric',2012                                    kwargs=cloud_kwargs2013                                )2014                                # snsClient = boto3.client('sns', region_name=m)2015                                # subscribers = snsClient.list_subscriptions_by_topic(2016                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]2017                                #     #  Pagination not used since only 1 subscriber required2018                                # )2019                                sns_kwargs = {2020                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]2021                                }2022                                subscribers = self.connection_manager.call(2023                                    service='sns',2024                                    command='list_subscriptions_by_topic',2025                                    kwargs=sns_kwargs2026                                )2027                                if not len(subscribers['Subscriptions']) == 0:2028                                    result = True2029                except:2030                    pass2031        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,2032                'Description': description, 'ControlId': control}2033    # 3.13 Ensure a log metric filter and alarm exist for route table changes (Scored)2034    def control_3_13_ensure_log_metric_changes_to_route_tables(self,cloudtrails):2035        """Summary2036        Returns:2037            TYPE: Description2038        """2039        result = False2040        failReason = ""2041        offenders = []2042        control = "3.13"2043        description = "Ensure a log metric filter and alarm exist for route table changes"2044        scored = True2045        failReason = "Ensure a log metric filter and alarm exist for route table changes"2046        for m, n in cloudtrails.items():2047            self.setRegion(m, iam_role=None)2048            for o in n:2049                try:2050                    if o['CloudWatchLogsLogGroupArn']:2051                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)2052                        # client = boto3.client('logs', region_name=m)2053                        # filters = client.describe_metric_filters(2054                        #     logGroupName=group2055                        # )2056                        cloud_kwargs = {2057                            'logGroupName': group,2058                        }2059                        filters = self.connection_manager.call(2060                            service='logs',2061                            command='describe_metric_filters',2062                            kwargs=cloud_kwargs2063                        )2064                        for p in filters['metricFilters']:2065                            patterns = ["\$\.eventName\s*=\s*\"?CreateRoute(\"|\)|\s)",2066                                        "\$\.eventName\s*=\s*\"?CreateRouteTable(\"|\)|\s)",2067                                        "\$\.eventName\s*=\s*\"?ReplaceRoute(\"|\)|\s)",2068                                        "\$\.eventName\s*=\s*\"?ReplaceRouteTableAssociation(\"|\)|\s)",2069                                        "\$\.eventName\s*=\s*\"?DeleteRouteTable(\"|\)|\s)",2070                                        "\$\.eventName\s*=\s*\"?DeleteRoute(\"|\)|\s)",2071                                        "\$\.eventName\s*=\s*\"?DisassociateRouteTable(\"|\)|\s)"]2072                            if find_in_string(patterns, str(p['filterPattern'])):2073                                # cwclient = boto3.client('cloudwatch', region_name=m)2074                                # response = cwclient.describe_alarms_for_metric(2075                                #     MetricName=p['metricTransformations'][0]['metricName'],2076                                #     Namespace=p['metricTransformations'][0]['metricNamespace']2077                                # )2078                                cloud_kwargs = {2079                                    'MetricName': MetricName,2080                                    'Namespace': Namespace,2081                                }2082                                response = self.connection_manager.call(2083                                    service='cloudwatch',2084                                    command='describe_alarms_for_metric',2085                                    kwargs=cloud_kwargs2086                                )2087                                # snsClient = boto3.client('sns', region_name=m)2088                                # subscribers = snsClient.list_subscriptions_by_topic(2089                                #     TopicArn=response['MetricAlarms'][0]['AlarmActions'][0]2090                                #     #  Pagination not used since only 1 subscriber required2091                                # )2092                                sns_kwargs = {2093                                    "TopicArn": response['MetricAlarms'][0]['AlarmActions'][0]2094                                }2095                                subscribers = self.connection_manager.call(2096                                    service='sns',2097                                    command='list_subscriptions_by_topic',2098                                    kwargs=sns_kwargs2099                                )2100                                if not len(subscribers['Subscriptions']) == 0:2101                                    result = True2102                except:2103                    pass2104        return {'Result': result, 'failReason': failReason, 'Offenders': offenders, 'ScoredControl': scored,2105                'Description': description, 'ControlId': control}2106    # 3.14 Ensure a log metric filter and alarm exist for VPC changes (Scored)2107    def control_3_14_ensure_log_metric_changes_to_vpc(self,cloudtrails):2108        """Summary2109        Returns:2110            TYPE: Description2111        """2112        result = False2113        failReason = ""2114        offenders = []2115        control = "3.14"2116        description = "Ensure a log metric filter and alarm exist for VPC changes"2117        scored = True2118        failReason = "Ensure a log metric filter and alarm exist for VPC changes"2119        for m, n in cloudtrails.items():2120            self.setRegion(m, iam_role=None)2121            for o in n:2122                try:2123                    if o['CloudWatchLogsLogGroupArn']:2124                        group = re.search('log-group:(.+?):', o['CloudWatchLogsLogGroupArn']).group(1)2125                        # client = boto3.client('logs', region_name=m)2126                        # filters = client.describe_metric_filters(2127                        #     logGroupName=group2128                        # )2129                        cloud_kwargs = {2130                            'logGroupName': group,2131                        }2132                        filters = self.connection_manager.call(2133                            service='logs',2134                            command='describe_metric_filters',2135                            kwargs=cloud_kwargs2136                        )2137                        for p in filters['metricFilters']:2138                            patterns = ["\$\.eventName\s*=\s*\"?CreateVpc(\"|\)|\s)",2139                                        "\$\.eventName\s*=\s*\"?DeleteVpc(\"|\)|\s)",2140                                        "\$\.eventName\s*=\s*\"?ModifyVpcAttribute(\"|\)|\s)",2141                                        "\$\.eventName\s*=\s*\"?AcceptVpcPeeringConnection(\"|\)|\s)",2142                                        "\$\.eventName\s*=\s*\"?CreateVpcPeeringConnection(\"|\)|\s)",2143                                        "\$\.eventName\s*=\s*\"?DeleteVpcPeeringConnection(\"|\)|\s)",2144                                        "\$\.eventName\s*=\s*\"?RejectVpcPeeringConnection(\"|\)|\s)",2145                                        "\$\.eventName\s*=\s*\"?AttachClassicLinkVpc(\"|\)|\s)",2146                                        "\$\.eventName\s*=\s*\"?DetachClassicLinkVpc(\"|\)|\s)",2147                                        "\$\.eventName\s*=\s*\"?DisableVpcClassicLink(\"|\)|\s)",2148                                        "\$\.eventName\s*=\s*\"?EnableVpcClassicLink(\"|\)|\s)"]2149                            if find_in_string(patterns, str(p['filterPattern'])):2150                                # cwclient = boto3.client('cloudwatch', region_name=m)2151                                # response = cwclient.describe_alarms_for_metric(2152                                #     MetricName=p['metricTransformations'][0]['metricName'],2153                                #     Namespace=p['metricTransformations'][0]['metricNamespace']2154                                # )2155                                cloud_kwargs = {2156                                    'MetricName': MetricName,2157                                    'Namespace': Namespace,2158                                }2159                                response = self.connection_manager.call(2160                                    service='cloudwatch',2161                                    command='describe_alarms_for_metric',2162                                    kwargs=cloud_kwargs2163                                )2164                                # snsClient = boto3.client('sns', region_name=m)2165                                # subscribers = snsClient.list_subscriptions_by_topic(...

check-celery-queues.py

Source:check-celery-queues.py

...40        return self.client.put_metric_data(*args, **kwargs)41    @backoff.on_exception(backoff.expo,42                          (botocore.exceptions.ClientError),43                          max_tries=max_tries)44    def describe_alarms_for_metric(self, *args, **kwargs):45        return self.client.describe_alarms_for_metric(*args, **kwargs)46    @backoff.on_exception(backoff.expo,47                          (botocore.exceptions.ClientError),48                          max_tries=max_tries)49    def put_metric_alarm(self, *args, **kwargs):50        return self.client.put_metric_alarm(*args, **kwargs)51@click.command()52@click.option('--host', '-h', default='localhost',53              help='Hostname of redis server')54@click.option('--port', '-p', default=6379, help='Port of redis server')55@click.option('--environment', '-e', required=True)56@click.option('--deploy', '-d', required=True,57              help="Deployment (i.e. edx or edge)")58@click.option('--max-metrics', default=30,59              help='Maximum number of CloudWatch metrics to publish')60@click.option('--threshold', default=50,61              help='Default maximum queue length before alarm notification is'62              + ' sent')63@click.option('--queue-threshold', type=(str, int), multiple=True,64              help='Threshold per queue in format --queue-threshold'65              + ' {queue_name} {threshold}. May be used multiple times')66@click.option('--sns-arn', '-s', help='ARN for SNS alert topic', required=True)67def check_queues(host, port, environment, deploy, max_metrics, threshold,68                 queue_threshold, sns_arn):69    thresholds = dict(queue_threshold)70    timeout = 171    namespace = "celery/{}-{}".format(environment, deploy)72    redis_client = RedisWrapper(host=host, port=port, socket_timeout=timeout,73                                socket_connect_timeout=timeout)74    cloudwatch = CwBotoWrapper()75    metric_name = 'queue_length'76    dimension = 'queue'77    response = cloudwatch.list_metrics(Namespace=namespace,78                                       MetricName=metric_name,79                                       Dimensions=[{'Name': dimension}])80    existing_queues = []81    for m in response["Metrics"]:82        existing_queues.extend(83            [d['Value'] for d in m["Dimensions"] if d['Name'] == dimension])84    redis_queues = set([k.decode() for k in redis_client.keys()85                        if redis_client.type(k) == b'list'])86    all_queues = existing_queues + list(87        set(redis_queues).difference(existing_queues)88    )89    if len(all_queues) > max_metrics:90        # TODO: Use proper logging framework91        print("Warning! Too many metrics, refusing to publish more than {}"92              .format(max_metrics))93    # Take first max_metrics number of queues from all_queues and remove94    # queues that aren't in redis95    queues = [q for q in all_queues[:max_metrics] if q in redis_queues]96    metric_data = []97    for queue in queues:98        metric_data.append({99            'MetricName': metric_name,100            'Dimensions': [{101                "Name": dimension,102                "Value": queue103            }],104            'Value': redis_client.llen(queue)105        })106    if len(metric_data) > 0:107        cloudwatch.put_metric_data(Namespace=namespace, MetricData=metric_data)108    for queue in queues:109        dimensions = [{'Name': dimension, 'Value': queue}]110        queue_threshold = threshold111        if queue in thresholds:112            queue_threshold = thresholds[queue]113        # Period is in seconds114        period = 60115        evaluation_periods = 15116        comparison_operator = "GreaterThanThreshold"117        treat_missing_data = "notBreaching"118        statistic = "Maximum"119        actions = [sns_arn]120        alarm_name = "{}-{} {} queue length over threshold".format(environment,121                                                                   deploy,122                                                                   queue)123        if len(cloudwatch.describe_alarms_for_metric(Namespace=namespace,124                                                     MetricName=metric_name,125                                                     Dimensions=dimensions)126                ['MetricAlarms']) < 1:127            print('Creating new alarm "{}"'.format(alarm_name))128            cloudwatch.put_metric_alarm(AlarmName=alarm_name,129                                        AlarmDescription=alarm_name,130                                        Namespace=namespace,131                                        MetricName=metric_name,132                                        Dimensions=dimensions,133                                        Period=period,134                                        EvaluationPeriods=evaluation_periods,135                                        TreatMissingData=treat_missing_data,136                                        Threshold=queue_threshold,137                                        ComparisonOperator=comparison_operator,...

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.