Source code for awslimitchecker.limit

"""
awslimitchecker/limit.py

The latest version of this package is available at:
<https://github.com/jantman/awslimitchecker>

################################################################################
Copyright 2015-2018 Jason Antman <jason@jasonantman.com>

    This file is part of awslimitchecker, also known as awslimitchecker.

    awslimitchecker is free software: you can redistribute it and/or modify
    it under the terms of the GNU Affero General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.

    awslimitchecker is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Affero General Public License for more details.

    You should have received a copy of the GNU Affero General Public License
    along with awslimitchecker.  If not, see <http://www.gnu.org/licenses/>.

The Copyright and Authors attributions contained herein may not be removed or
otherwise altered, except to add the Author attribution of a contributor to
this work. (Additional Terms pursuant to Section 7b of the AGPL v3)
################################################################################
While not legally required, I sincerely request that anyone who finds
bugs please submit them at <https://github.com/jantman/awslimitchecker> or
to me via email, and that you send any contributions or improvements
either as a pull request on GitHub, or to me via email.
################################################################################

AUTHORS:
Jason Antman <jason@jasonantman.com> <http://www.jasonantman.com>
################################################################################
"""

#: indicates a limit value that came from hard-coded defaults in awslimitchecker
SOURCE_DEFAULT = 0

#: indicates a limit value that came from user-defined limit overrides
SOURCE_OVERRIDE = 1

#: indicates a limit value that came from Trusted Advisor
SOURCE_TA = 2

#: indicates a limit value that came from the service's API
SOURCE_API = 3

#: indicates a limit value that came from the Service Quotas service
SOURCE_QUOTAS = 4


[docs]class AwsLimit(object):
[docs] def __init__(self, name, service, default_limit, def_warning_threshold, def_critical_threshold, limit_type=None, limit_subtype=None, ta_service_name=None, ta_limit_name=None, quotas_service_code=None, quotas_name=None, quotas_unit='None', quotas_unit_converter=None): """ Describes one specific AWS service limit, as well as its current utilization, default limit, thresholds, and any Trusted Advisor information about this limit. :param name: the name of this limit (may contain spaces); if possible, this should be the name used by AWS, i.e. TrustedAdvisor :type name: str :param service: the :py:class:`~._AwsService` class that this limit is for :type service: :py:class:`~._AwsService` :param default_limit: the default value of this limit for new accounts :type default_limit: :py:obj:`int`, or ``None`` if unlimited :param def_warning_threshold: the default warning threshold, as an integer percentage. :type def_warning_threshold: int :param def_critical_threshold: the default critical threshold, as an integer percentage. :type def_critical_threshold: int :param limit_type: the type of resource this limit describes, specified as one of the type names used in `CloudFormation <http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-template-resource-type-ref.html>`_ # noqa such as "AWS::EC2::Instance" or "AWS::RDS::DBSubnetGroup". :param limit_subtype: resource sub-type for this limit, if applicable, such as "t2.micro" or "SecurityGroup" :type limit_subtype: str :param ta_service_name: The service name returned by Trusted Advisor for this limit, if different from the name of ``service`` :type ta_service_name: str :param ta_limit_name: The limit name returned by Trusted Advisor for this limit, if different from ``name``. :type ta_limit_name: str :param quotas_service_code: the Service Quotas service code to retrieve this limit from, if different from the :py:attr:`~._AwsService.quotas_service_code` attribute of ``service``. :type quotas_service_code: str or None :param quotas_name: the Service Quotas quota name to use for this limit, if different from the limit ``name``. :type quotas_name: str or None :param quotas_unit: the Service Quotas quota unit that we need our limit value to be, for quotas that use units. This must be one of the units supported by :py:class:`~.ServiceQuotasClient`. It defaults to the string "None", which (for some strange reason) is what's returned by the Service Quotas API. :type quotas_unit: str :param quotas_unit_converter: A callable to be passed to :py:meth:`~.ServiceQuotasClient.get_quota_value` for unit conversion. Must take three positional arguments: the Service Quotas value for this quota (float), the quota ``Unit`` str, and the return value of :py:meth:`~.quotas_unit`. This callable is responsible for converting the quota value from the quota Unit to this class's expected unit. If they cannot be converted, it should log an error and return None. :type quotas_unit_converter: ``callable`` :raises: ValueError """ if def_warning_threshold >= def_critical_threshold: raise ValueError("critical threshold must be greater than warning " "threshold") self.name = name self.service = service self.default_limit = default_limit self.limit_type = limit_type self.limit_subtype = limit_subtype self.limit_override = None self.override_ta = True self.ta_limit = None self.ta_unlimited = False self.api_limit = None self._current_usage = [] self.def_warning_threshold = def_warning_threshold self.def_critical_threshold = def_critical_threshold self.warn_percent = None self.warn_count = None self.crit_percent = None self.crit_count = None self._warnings = [] self._criticals = [] self._ta_service_name = ta_service_name self._ta_limit_name = ta_limit_name self._quotas_service_code = quotas_service_code self._quotas_name = quotas_name self._quotas_unit = quotas_unit self.quotas_limit = None self.quotas_unit_converter = quotas_unit_converter
[docs] def set_limit_override(self, limit_value, override_ta=True): """ Set a new value for this limit, to override the default (such as when AWS Support has increased a limit of yours). If ``override_ta`` is True, this value will also supersede any found through Trusted Advisor. :param limit_value: the new limit value :type limit_value: int :param override_ta: whether or not to also override Trusted Advisor information :type override_ta: bool """ self.limit_override = limit_value self.override_ta = override_ta
[docs] def _set_ta_limit(self, limit_value): """ Set the value for the limit as reported by Trusted Advisor. This method should only be called by :py:class:`~.TrustedAdvisor`. :param limit_value: the Trusted Advisor limit value :type limit_value: int """ self.ta_limit = limit_value
[docs] def _set_ta_unlimited(self): """ Set state to indicate that TrustedAdvisor reports this limit as having no maximum (unlimited). This method should only be called by :py:class:`~.TrustedAdvisor`. """ self.ta_unlimited = True
[docs] def _set_api_limit(self, limit_value): """ Set the value for the limit as reported by the service's API. This method should only be called from the Service class. :param limit_value: the API limit value :type limit_value: int """ self.api_limit = limit_value
[docs] def _set_quotas_limit(self, limit_value): """ Set the value for the limit as reported by the Service Quotas service. This method should only be called from the Service class. :param limit_value: the Service Quotas limit value :type limit_value: float """ self.quotas_limit = limit_value
[docs] def get_limit_source(self): """ Return :py:data:`~awslimitchecker.limit.SOURCE_DEFAULT` if :py:meth:`~.get_limit` returns the default limit, :py:data:`~awslimitchecker.limit.SOURCE_OVERRIDE` if it returns a manually-overridden limit, :py:data:`~awslimitchecker.limit.SOURCE_TA` if it returns a limit from Trusted Advisor, :py:data:`~awslimitchecker.limit.SOURCE_API` if it returns a limit retrieved from the service's API, or :py:data:`~.SOURCE_QUOTAS` if it returns a limit from the Service Quotas service. :returns: one of :py:data:`~awslimitchecker.limit.SOURCE_DEFAULT`, :py:data:`~awslimitchecker.limit.SOURCE_OVERRIDE`, or :py:data:`~awslimitchecker.limit.SOURCE_TA`, or :py:data:`~awslimitchecker.limit.SOURCE_API`, or :py:data:`~.awslimitchecker.limit.SOURCE_QUOTAS` :rtype: int """ if self.limit_override is not None and ( self.override_ta is True or (self.ta_limit is None and self.ta_unlimited is False) ): return SOURCE_OVERRIDE if self.api_limit is not None: return SOURCE_API if self.quotas_limit is not None: return SOURCE_QUOTAS if self.ta_limit is not None or self.ta_unlimited is True: return SOURCE_TA return SOURCE_DEFAULT
[docs] def get_limit(self): """ Returns the effective limit value for this Limit, taking into account limit overrides and Trusted Advisor data. None is returned for limits that are explicitly unlimited. :returns: effective limit value, ``int`` or ``None`` """ limit_type = self.get_limit_source() if limit_type == SOURCE_OVERRIDE: return self.limit_override elif limit_type == SOURCE_API: return self.api_limit elif limit_type == SOURCE_QUOTAS: return self.quotas_limit elif limit_type == SOURCE_TA: if self.ta_unlimited is True: return None return self.ta_limit return self.default_limit
[docs] def has_resource_limits(self): """ Determines if this limit contains usages with a specified maximum. Some AWS resources have a limit that is a different for each item. :returns: whether of not some resources have a defined maximum :rtype: bool """ return any(usage for usage in self._current_usage if usage.get_maximum())
[docs] def get_current_usage(self): """ Get the current usage for this limit, as a list of :py:class:`~.AwsLimitUsage` instances. :returns: list of current usage values :rtype: :py:obj:`list` of :py:class:`~.AwsLimitUsage` """ return self._current_usage
[docs] def get_current_usage_str(self): """ Get the a string describing the current usage for this limit. If no usage has been added for this limit, the result will be "<unknown>". If the limit has only one current usage instance, this will be that instance's ``AwsLimitUsage.__str__`` value. If the limit has more than one current usage instance, this will be the a string of the form ``max: X (Y)`` where ``X`` is the ``AwsLimitUsage.__str__`` value of the instance with the maximum value, and ``Y`` is a comma-separated list of the ``AwsLimitUsage.__str__`` values of all usage instances in ascending order. :returns: representation of current usage :rtype: str """ if len(self._current_usage) == 0: return '<unknown>' if len(self._current_usage) == 1: return str(self._current_usage[0]) lim_str = ', '.join([str(x) for x in sorted(self._current_usage)]) s = 'max: {m} ({l})'.format( m=str(max(self._current_usage)), l=lim_str ) return s
[docs] def _add_current_usage(self, value, maximum=None, resource_id=None, aws_type=None): """ Add a new current usage value for this limit. Creates a new :py:class:`~.AwsLimitUsage` instance and appends it to the internal list. If more than one usage value is given to this service, they should have ``id`` and ``aws_type`` set. This method should only be called from the :py:class:`~._AwsService` instance that created and manages this Limit. :param value: the numeric usage value :type value: :py:obj:`int` or :py:obj:`float` :param resource_id: If there can be multiple usage values for one limit, an AWS ID for the resource this instance describes :type resource_id: str :param aws_type: if ``id`` is not None, the AWS resource type that ID represents. As a convention, we use the AWS Resource Type names used by `CloudFormation <http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-template-resource-type-ref.html>`_ # noqa :type aws_type: str """ self._current_usage.append( AwsLimitUsage( self, value, maximum=maximum, resource_id=resource_id, aws_type=aws_type ) )
[docs] def _reset_usage(self): """Discard all current usage data.""" self._current_usage = []
[docs] def _get_thresholds(self): """ Get the warning and critical thresholds for this Limit. Return type is a 4-tuple of: 1. warning integer (usage) threshold, or None 2. warning percent threshold 3. critical integer (usage) threshold, or None 4. critical percent threshold :rtype: tuple """ t = ( self.warn_count, self.warn_percent or self.def_warning_threshold, self.crit_count, self.crit_percent or self.def_critical_threshold, ) return t
[docs] def set_threshold_override(self, warn_percent=None, warn_count=None, crit_percent=None, crit_count=None): """ Override the default warning and critical thresholds used to evaluate this limit's usage. Theresholds can be specified as a percentage of the limit, or as a usage count, or both. **Note:** The percent thresholds (``warn_percent`` and ``crit_percent``) have default values that are set globally for awslimitchecker, unlike the count thresholds. When setting threshold overrides to quiet or suppress alerts for a limit, you **must** set the percent thresholds. If you only set overrides for the ``count`` thresholds, the percent thresholds will continue to be evaluated at their awslimitchecker-wide default, and likely prevent alerts from being suppressed. see :py:meth:`~.check_thresholds` for further information on threshold evaluation. :param warn_percent: new warning threshold, percentage used :type warn_percent: int :param warn_count: new warning threshold, actual count/number :type warn_count: int :param crit_percent: new critical threshold, percentage used :type crit_percent: int :param crit_count: new critical threshold, actual count/number :type crit_count: int """ self.warn_percent = warn_percent self.warn_count = warn_count self.crit_percent = crit_percent self.crit_count = crit_count
[docs] def check_thresholds(self): """ Check this limit's current usage against the specified default thresholds, and any custom theresholds that have been set on the class instance. Return True if usage is within thresholds, or false if warning or critical thresholds have been surpassed. This method sets internal variables in this instance which can be queried via :py:meth:`~.get_warnings` and :py:meth:`~.get_criticals` to obtain further details about the thresholds that were crossed. **Note** This function returns False if *any* thresholds were crossed. Please be aware of this when setting threshold overrides to suppress alerts. Each threshold (``warn_percent``, ``warn_count``, ``crit_percent``, ``crit_count``) that has been set is evaluated individually and the result appended to a list of warnings or criticals, respectively. If *any* of these evaluations failed, the method returns False. :returns: False if any thresholds were crossed, True otherwise :rtype: bool """ (warn_int, warn_pct, crit_int, crit_pct) = self._get_thresholds() all_ok = True for u in self._current_usage: usage = u.get_value() limit = u.get_maximum() or self.get_limit() if limit is None or limit == 0: continue pct = (usage / (limit * 1.0)) * 100 if crit_int is not None and usage >= crit_int: self._criticals.append(u) all_ok = False elif pct >= crit_pct: self._criticals.append(u) all_ok = False elif warn_int is not None and usage >= warn_int: self._warnings.append(u) all_ok = False elif pct >= warn_pct: self._warnings.append(u) all_ok = False return all_ok
[docs] def get_warnings(self): """ Return a list of :py:class:`~.AwsLimitUsage` instances that crossed the warning threshold. These objects are comparable and can be sorted. :rtype: list """ return self._warnings
[docs] def get_criticals(self): """ Return a list of :py:class:`~.AwsLimitUsage` instances that crossed the critical threshold. These objects are comparable and can be sorted. :rtype: list """ return self._criticals
@property def ta_service_name(self): """ Return the effective Trusted Advisor service name that this limit's data will have. This should be ``self._ta_service_name`` if set, otherwise the name of ``self.service``. :return: Trusted Advisor service data name :rtype: str """ if self._ta_service_name is not None: return self._ta_service_name return self.service.service_name @property def ta_limit_name(self): """ Return the effective Trusted Advisor limit name that this limit's data will have. This should be ``self._ta_limit_name`` if set, otherwise ``self.name``. :return: Trusted Advisor limit data name :rtype: str """ if self._ta_limit_name is not None: return self._ta_limit_name return self.name @property def quotas_service_code(self): """ Return the Service Quotas service code to use for this limit. :return: Service Quotas service code :rtype: str """ if self._quotas_service_code is not None: return self._quotas_service_code return self.service.quotas_service_code @property def quota_name(self): """ Return the Service Quotas quota name to use for this limit. :return: Service Quotas quota name :rtype: str """ if self._quotas_name is not None: return self._quotas_name return self.name @property def quotas_unit(self): """ Return the Service Quotas unit to use for this limit. :return: Service Quotas unit :rtype: str """ return self._quotas_unit
[docs]class AwsLimitUsage(object):
[docs] def __init__(self, limit, value, maximum=None, resource_id=None, aws_type=None): """ This object describes the usage of an AWS resource, with the capability of containing information about the resource beyond an integer usage. The simplest case is an account- / region-wide count, such as the number of running EC2 Instances, in which case a simple integer value is sufficient. In this case, the :py:class:`~.AwsLimit` would have one instance of this class for the single value. In more complex cases, such as the "Subnets per VPC", the limit is applied by AWS on multiple resources (once per VPC). In this case, the :py:class:`~.AwsLimit` should have one instance of this class per VPC, so we can determine *which* VPCs have crossed thresholds. AwsLimitUsage objects are comparable based on their numeric ``value``. :param limit: the AwsLimit that this instance describes :type limit: :py:class:`~.AwsLimit` :param value: the numeric usage value :type value: :py:obj:`int` or :py:obj:`float` :param maximum: the numeric maximum value :type maximum: :py:obj:`int` or :py:obj:`float` :param resource_id: If there can be multiple usage values for one limit, an AWS ID for the resource this instance describes :type resource_id: str :param aws_type: if ``id`` is not None, the AWS resource type that ID represents. As a convention, we use the AWS Resource Type names used by `CloudFormation <http://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-template-resource-type-ref.html>`_ # noqa :type aws_type: str """ self.limit = limit self.value = value self.maximum = maximum self.resource_id = resource_id self.aws_type = aws_type
[docs] def get_value(self): """ Get the current usage value :returns: current usage value :rtype: :py:obj:`int` or :py:obj:`float` """ return self.value
[docs] def get_maximum(self): """ Get the current maximum value :returns: current maximum value :rtype: :py:obj:`int` or :py:obj:`float` """ return self.maximum
[docs] def __str__(self): """ Return a string representation of this object. If ``id`` is not set, return ``value`` formatted as a string; otherwise, return a string of the format ``id=value``. :rtype: str """ s = '{v}'.format(v=self.value) if self.resource_id is not None: s = self.resource_id + '=' + s return s
[docs] def __eq__(self, other): return self.value == other.value
[docs] def __ne__(self, other): return self.value != other.value
[docs] def __gt__(self, other): return self.value > other.value
[docs] def __lt__(self, other): return self.value < other.value
[docs] def __ge__(self, other): return self.value >= other.value