volume-status-check

#!/usr/bin/env python3
#
# Copyright 2012-2022, 42Lines, Inc.
# Original Author: Jim Browne
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from argparse import ArgumentParser
import datetime

import boto3
import pynagios
from pynagios import Plugin, Response


def ts_in_past(timestamp):
    '''Return true if an AWS format timestamp is in the past.
    Return false if not or it timestamp is unparseable.'''

    try:
        tstamp = datetime.datetime.strptime(timestamp,
                                            '%Y-%m-%dT%H:%M:%S.%fZ')
    except ValueError:
        return False

    tstamp_delta = datetime.datetime.utcnow() - tstamp
    return tstamp_delta.total_seconds() > 0


class VolumeEventsCheck(Plugin):
    parser = ArgumentParser(add_help=False)

    parser.add_argument("--region",
                        help="Region to check (default: us-east-1)",
                        default=[],
                        action="append")
    parser.add_argument("--profile",
                        help="AWS profile to use (default: none)",
                        action="store")
    parser.add_argument("--all",
                        help="Check all regions",
                        action="store_true")

    warns = []
    criticals = []
    infos = []
    session = None
    ec2_client = None

    def check(self):
        self.session = boto3.Session(profile_name=self.options.profile)

        if not self.options.region:
            self.options.region = ["us-east-1"]

        if self.options.all:
            check_regions = self.session.get_available_regions('ec2')
        else:
            check_regions = []
            for option_region in self.options.region:
                for aws_region in self.session.get_available_regions('ec2'):
                    if aws_region == option_region:
                        check_regions.append(aws_region)
                        break
                else:
                    message = "Region %s not found." % option_region
                    return Response(pynagios.UNKNOWN, message)

        for check_region in check_regions:
            # Create client here to facilitate testing
            self.ec2_client = self.session.client('ec2', region_name=check_region)
            self.regioncheck(check_region)

        wmessages = ", ".join(self.warns)
        cmessages = ", ".join(self.criticals)
        infos = ", ".join(self.infos)

        if cmessages:
            if wmessages:
                cmessages += " WARNING: " + wmessages
            result = Response(pynagios.CRITICAL, cmessages)
        elif wmessages:
            result = Response(pynagios.WARNING, wmessages)
        else:
            result = Response(pynagios.OK, "Checked regions: " +
                              ", ".join(check_regions) + " OK: " + infos)
        return result

    def regioncheck(self, region):

        #vfilter = [{ 'Name': 'volume-status.status',  'Values' : 'impaired' }]
        paginator = self.ec2_client.get_paginator('describe_volume_status')

        for statuses in paginator.paginate():
            for stat in statuses['VolumeStatuses']:
                self.process_volume_status(region, stat)

    def process_volume_status(self, region, stat):
        if self.options.verbosity:
            print("id %s status %s" % (stat['VolumeId'], stat.volume_status))

        attachinfo = None

        if stat.get('Events', False) or stat.get('Actions', False):
            vstates = self.ec2_client.describe_volumes(VolumeIds=[stat['VolumeId']])
            for vol in vstates['Volumes']:
                if vol['VolumeId'] == stat['VolumeId']:
                    for attachment in  vol['Attachments']:
                        if attachment['State'] == 'attached':
                            attachinfo = "(%s %s)" % (attachment['InstanceId'],
                                                      attachment['Device'])
                        break

                if attachinfo:  # vstates loop
                    break

            if self.options.verbosity:
                print("id %s attached %s" % (stat['VolumeId'], attachinfo))

        for event in stat.get('Events', []):
            if self.options.verbosity:
                print(event)
            message = "%s:%s" % (region, stat['VolumeId'])
            message += ":%s %s" % (event['EventType'], event['Description'])
            message += " nb %s" % (event['NotBefore'])
            message += " na %s" % (event['NotAfter'])

            if ts_in_past(event['NotAfter']):
                message += " %s" % (attachinfo)
                self.infos.append(message)
            elif attachinfo:
                message += " %s" % (attachinfo)
                self.criticals.append(message)
            else:
                self.warns.append(message)

        for action in stat.get('Actions', []):
            if self.options.verbosity:
                print(action)
            message = "%s:%s" % (region, stat['VolumeId'])
            message += ":%s %s" % (action['EventType'], action['Description'])
            message += " code %s" % (action['Code'])
            if attachinfo:
                message += " %s" % (attachinfo)
                self.criticals.append(message)
            else:
                self.warns.append(message)


if __name__ == "__main__":
    # Instantiate the plugin, check it, and then exit
    VolumeEventsCheck().check().exit()