slow_log_checker.py

# -*- encoding: utf-8 -*-
import os
import boto3
import botocore
import datetime
import difflib
from urllib2 import Request, urlopen, URLError, HTTPError
import json


def handler(event, context):
    firstRun = False
    logFileData = ""

    RDSclient = boto3.client('rds')
    S3client = boto3.client('s3')

    S3BucketName = os.environ['BucketName']
    S3BucketPrefix = 'slow-query-log'
    logNamePrefix = 'error/postgresql.log.' + today #defined in parameters group of RDS instance
    lastRecievedFile = S3BucketPrefix + 'lastMarker'
    distinguisher='duration'
    today = datetime.datetime.utcnow().strftime("%Y-%m-%d")
    RDSInstanceNames = list_rds_instances(RDSclient)

    for RDSInstanceName in RDSInstanceNames:
        dbLogs = RDSclient.describe_db_log_files( DBInstanceIdentifier=RDSInstanceName, FilenameContains=logNamePrefix)
        lastWrittenTime = 0
        lastWrittenThisRun = 0
        try:
            S3response = S3client.head_bucket(Bucket=S3BucketName)
        except botocore.exceptions.ClientError as e:
            error_code = int(e.response['ResponseMetadata']['HTTPStatusCode'])
            if error_code == 404:
                print("Error: Bucket name provided not found")
            else:
                print("Error: Unable to access bucket name, error: " + e.response['Error']['Message'])
        try:
            S3response = S3client.get_object(Bucket=S3BucketName, Key=lastRecievedFile + RDSInstanceName)
        except botocore.exceptions.ClientError as e:
            error_code = int(e.response['ResponseMetadata']['HTTPStatusCode'])
            if error_code == 404:
                print("It appears this is the first log import, all files will be retrieved from RDS")
                firstRun = True
            else:
                print("Error: Unable to access lastRecievedFile name, error: " + e.response['Error']['Message'])

        if not firstRun:
            lastWrittenTime = int(S3response['Body'].read(S3response['ContentLength']))
            print("Found marker from last log download, retrieving log files with lastWritten time after %s" % str(lastWrittenTime))
        for dbLog in dbLogs['DescribeDBLogFiles']:
            if ( dbLog['LogFileName'] == logNamePrefix and int(dbLog['LastWritten']) > lastWrittenTime ) or firstRun:
                print("Downloading log file: %s found and with LastWritten value of: %s " % (dbLog['LogFileName'],dbLog['LastWritten']))
                if int(dbLog['LastWritten']) > lastWrittenThisRun:
                    lastWrittenThisRun = int(dbLog['LastWritten'])
                logFile = RDSclient.download_db_log_file_portion(DBInstanceIdentifier=RDSInstanceName, LogFileName=dbLog['LogFileName'],Marker='0')
                logFileData = logFile['LogFileData']
                objectName = S3BucketPrefix + dbLog['LogFileName'] + RDSInstanceName
                try:
                    S3response = S3client.get_object(Bucket=S3BucketName, Key=objectName)
                    oldData = S3response['Body'].read()
                    oldData = oldData.splitlines(1)
                    newData = logFileData.splitlines(1)
                    diff = difflib.unified_diff(oldData, newData)
                    lines = list(diff)
                    added = [line[1:] for line in lines if line[0] == '+']
                    added = ''.join(added)
                    for msg in added.split(today):
                        if distinguisher in msg:
                            send_slack_notification(msg, RDSInstanceName)
                except botocore.exceptions.ClientError as e:
                    print("%s", e.response['Error']['Message'])
                logFileData += logFile['LogFileData']
                byteData = str.encode(logFileData)
                try:
                    objectName = S3BucketPrefix + dbLog['LogFileName'] + RDSInstanceName
                    S3response = S3client.put_object(Bucket=S3BucketName, Key=objectName,Body=byteData)
                except botocore.exceptions.ClientError as e:
                    return "Error writting object to S3 bucket or SES, ClientError: " + e.response['Error']['Message']
                print("Writting log file %s to S3 bucket %s" % (objectName,S3BucketName))
        try:
            S3response = S3client.put_object(Bucket=S3BucketName, Key=lastRecievedFile + '-' + RDSInstanceName, Body=str.encode(str(lastWrittenThisRun)))
        except botocore.exceptions.ClientError as e:
            return "Error writting object to S3 bucket, S3 ClientError: " + e.response['Error']['Message']
    print("Wrote new Last Written Marker to %s in Bucket %s" % (lastRecievedFile ,S3BucketName))
    return "Log file export complete"


def send_slack_notification(message, rds_instance_name):
    slack_channel = os.environ['SlackChannel']
    hook_url = os.environ['HookUrl']
    slack_message = {
        'channel': slack_channel,
        'text': 'Slow log query detected on %s datadase\n %s' % (rds_instance_name.upper(), message)
    }

    req = Request(hook_url, json.dumps(slack_message))
    try:
        response = urlopen(req)
        response.read()
        logger.info("Message posted to %s", slack_message['channel'])
    except HTTPError as e:
        logger.error("Request failed: %d %s", e.code, e.reason)
    except URLError as e:
        logger.error("Server connection failed: %s", e.reason)


def list_rds_instances(client):
    db_list = []
    dbs = client.describe_db_instances()
    for db in dbs['DBInstances']:
        tags = client.list_tags_for_resource(ResourceName=db['DBInstanceArn'])
        for tag in tags['TagList']:
            if "SlowLog" in tag['Key']:
                db_list.append(db['DBInstanceIdentifier'])
    return db_list