You may have already heard of  the serverless craze. For those like me who are always six months behind the curve; serverless is the concept of running code without having to provision or manage underlying servers for your code to execute.

Providers such as Heroku, AWS Lambda, Azure Function and GCP's Cloud Functions are extremely popular for hobbyist and professional programmers. It allows them all to focus on getting their code out the door without having to worry about standard sysadmin duties that comes with hosting an application.

In this code-along, I'll write python code to interact with Opsgenie's API to parse it's logs to check if a user has deleted their contact info! What is Opsgenie? Opsgenie is an incident management software similar to VictorOps and PagerDuty. They have an API that can be leveraged with some code to extend it's functions.

The workflow will start with AWS CloudWatch specifically CloudWatch Events. CloudWatch Events will be configured with a cron that executes everyday at a specific time.

CloudWatch will trigger our code in AWS Lambda. The code will interact with Opsgenie's API to check the logs for anyone who deleted their contact info. If it finds an entry, our code will then fire an alert via Opsgenie. Opsgenie will then notify the devops team to investigate why a user deleted their contact info.

In a SaaS environment with strict SLAs being able to reach the designated on call personnel is critical to a healthy operations.


Prerequisites

  • AWS Console access
  • Opsgenie account
  • Python IDE (I'm using PyCharm)

Writing the code

Before you begin writing the code make sure you have logged into Opsgenie and created an app key with with configuration access. Also make sure you have created an API integration key, to do this click on "integration list" and click on "add" for API. Once you have both API key and integration key, you can fire up pycharm!

Start a new project in pycharm and import three modules: requests, datetime, and json.  The requests module will allow us to send POST and GET requests to Opsgenie's API. Datetime and json modules are needed so that we can format the data appropriately for Opsgenie's consumption.

import requests
import datetime
import json

Next we're going to format the date, this is important because Opsgenie names it log files based on the date. Format: year-month-day-hour-minute-second     ex. 2019-06-13-00-00-00

I'm going to create an object called todays_date. In order to get today's date, I will use the datetime module and specifically it's datetime.now() method. Now to format the date in the manner readable by Opsgenie, there is another method in datetime that I'll use. It is call strftime().

I'll create another object called formatted_date and supply it with the following todays_date.strftime("%Y-%m-%d-00-00-00"). This will take today's date and then format it by year, month, and day. Exact time is not necessary here which is why I set it to 00-00-00.

import requests
import datetime
import json
import os
import re



todays_date = datetime.datetime.now()
formatted_date = todays_date.strftime("%Y-%m-%d-00-00-00")  # grabs current date

Variables and Dictionaries

After that create new variables for the url endpoint(url), app_key, dl_url1, alert_app_key, and alert_url. Then create three dictionaries for the headers and data payload. In the alert_data_payload, please review Opsgenie's documentation here.

import requests
import datetime
import json
import os
import re



todays_date = datetime.datetime.now()
formatted_date = todays_date.strftime("%Y-%m-%d-00-00-00")  # grabs current date

url = "https://api.sandbox.opsgenie.com/v2/logs/list/" + formatted_date
app_key = "INSERT_YOUR_OWN_KEY"
dl_url1 = 'https://api.sandbox.opsgenie.com/v2/logs/download/'
alert_app_key = "INSERT_YOUR_OWN_INTEGRATION_KEY"
alert_url = 'https://api.sandbox.opsgenie.com/v2/alerts'

payload = {
    "Content-Type": "application/json",
    "Authorization": app_key
}

alert_header_payload = {
    "Content-Type": "application/json",
    "Authorization": alert_app_key
}

Functions

Now comes the interesting portion of the code – functions! Before creating the functions here is a quick list of procedures I want the script to perform.

  1. List all of the logs for the day.
  2. Count the amount of logs.
  3. Extract the log name from the list in step 1.
  4. Generate download links from the log name.
  5. Download the log files and store the ones that matches my criteria.
  6. If there is a log that matches the criteria, send an alert from Opsgenie.

So for the first function, I'll have to create a list of all the logs for the day. I'll create a function and call it list_logs(), I'll then use the requests.get method and supply it a url and header info to get all of the logs from today only.

The output will be in json or a dictionary in python. In order to grab the data we need, we'll have to slice through the json and specify the data index in the dictionary. This will return the output we need.

def list_logs():

    get_request = requests.get(url, headers=payload)
    output = get_request.json()

    return output['data']

Next is a rather simple function, but is critical for the code to enumerate through a list properly. I'll call this function max_num and all this will do is return the amount of logs in the first function. (list_logs)

def max_num():

    max_list = len(list_logs())
    return max_list

Following that the next step is to create list of log filenames. This step will require the list of logs generated in step 1. list_logs

However we only want filenames to be added to the list if it exists so I'm calling the second function max_num() and checking if it is greater than zero in the if statement. If it is greater than zero, then I want it to add the filename to a list called list_filename.

def get_filename():

    list_filename = []
    i = 0
    if max_num() > 0:
        while i < max_num():
            list_filename.append(list_logs()[i]['filename'])
            i += 1
    return list_filename

As soon as you have a list of filenames, you can download it from Opsgenie...not quite. You will first need to create download links. I accomplished this by creating a function called get_dl_links(). Then I ran a for loop through the output of the previous function (get_filename). In the for loop, dl_url is combined with an entry from get_filename to create a url called dl_url2. That is then ran through a a request.get method and the output is added to another list called list_link which contains a list of download links.

def get_dl_links():

    list_link = []
    for x in get_filename():
        dl_url2 = dl_url1 + x
        get_dl_url = requests.get(dl_url2, headers=payload)
        list_link.append(get_dl_url.text)
    return list_link

Almost there, two more functions left! The next function will download the logs from the links created from get_dl_links() and download the logs. However most of the logs are irrelevant the only entries I care about are the ones that contain UserContact and deleted. So in my if statement, I specifically look for those strings. If they appear then I add then to another list called del_list.

def filter_logs():

    del_list = []
    for raw_log in get_dl_links():
        raw_log = requests.get(raw_log)
        if 'UserContact' and 'deleted' in raw_log.text:
            del_list.append(json.dumps(raw_log.text))
    return del_list

Finally the last function...well sort of!  This function will send an alert to Opsgenie about contact info being deleted. This is similar as before using the request module to do a GET request, but we'll do a POST request instead.

Now this function is quite complex, with for loops and a nested for loop. So the what this function does is take the list returned from filter_logs() and then runs it through a regex (\w.-]+@[\w.-]+) to find an email address.  It will create another list called in_list and if the email is not in_list it will append it to the list.

Following the nested for loop, it will go through the last for loop in this function and plug in each email in the data payload to Opsgenie. Then once it has that data payload it will make a POST request using requests.post to Opsgenie.

def alert_genie():

    a_list_of_logs = filter_logs()

    for output in a_list_of_logs:
        pattern = re.findall(r'[\w\.-]+@[\w\.-]+', output)

        for i in pattern:
            in_list = []
            n = i in in_list
            if n == False:
                in_list.append(i)

    for emails in in_list:
        alert_data_payload = {
            "message": "User Contact Info Deleted",
            "alias": "Username",
            "description": "A user deleted their contact info",
            "responders": [
                {"id": "79202f81-1a5d-4161-8ec6-6798a8738a45",
                 "type": "team"},
                {"name": "Test Team - Wiki Calendar",
                 "type": "team"},
                {"id": "b5be0234-df95-4b17-8669-5619e3c4ace0",
                 "type": "schedule"},
                {"name": "Test Team - Wiki Calendar_schedule",
                 "type": "schedule"}
            ],
            "visibleTo": [
                {"id": "79202f81-1a5d-4161-8ec6-6798a8738a45",
                 "type": "team"},
                {"name": "Test Team - Wiki Calendar",
                 "type": "team"}
            ],
            "tags": ["OverwriteQuietHours",
                     "Critical"],
            "details":
                {"Username": emails},
            "priority": "P2"
        }

        requests.post(alert_url, data=json.dumps(alert_data_payload), headers=alert_header_payload)

Now that most of the heavy functions portion are complete. At this point you can take a break or nap a few hours and then continue again.


Try and Except

Okay I may have jump the gun earlier about being done with any more functions, however I need to create this last one. The last part of the code is creating a function with a try and except statement. This is important because if it encounters an issue I don't want my code to bomb out entirely.

In my try statement I nested an if statement that checks if there are any items in the filter_logs function. If there is, it will print a line indicating that it will alert Opsgenie and then invoke the alert_genie() function.

If it encounters an IndexError, it will print a statement for the operator to check the slicing in get_filename.

def lambda_handler(event, context): 
    try:
        if len(filter_logs()) > 0:
            print('Sending Alert via Opsgenie...\n')
            alert_genie()
        else:
            pass
    except IndexError:
        print('Check the index (i) in get_filename... \n')
    return {
        'statusCode': 200,
        'body': json.dumps("Put back your contact info in Opsgenie or I'll eat your soul!")
    }

Awesome! The whole code put together is below.

import requests
import datetime
import json
import os
import re



todays_date = datetime.datetime.now()
formatted_date = todays_date.strftime("%Y-%m-%d-00-00-00")  # grabs current date

url = "https://api.sandbox.opsgenie.com/v2/logs/list/" + formatted_date
app_key = "INSERT_YOUR_OWN_KEY"
dl_url1 = 'https://api.sandbox.opsgenie.com/v2/logs/download/'
alert_app_key = "INSERT_YOUR_OWN_INTEGRATION_KEY"
alert_url = 'https://api.sandbox.opsgenie.com/v2/alerts'

payload = {
    "Content-Type": "application/json",
    "Authorization": app_key
}

alert_header_payload = {
    "Content-Type": "application/json",
    "Authorization": alert_app_key
}



####### Functions #######

# Interact with opsgenie api to get a list of logs for the day
def list_logs():

    get_request = requests.get(url, headers=payload)
    output = get_request.json()

    return output['data']


# Calculate the max number of logs for the day
def max_num():

    max_list = len(list_logs())
    return max_list


# Create a list of all log filenames
def get_filename():

    list_filename = []
    i = 0
    if max_num() > 0:
        while i < max_num():
            list_filename.append(list_logs()[i]['filename'])
            i += 1
    return list_filename


# Create a list of download urls
def get_dl_links():

    list_link = []
    for x in get_filename():
        dl_url2 = dl_url1 + x
        get_dl_url = requests.get(dl_url2, headers=payload)
        list_link.append(get_dl_url.text)
    return list_link


# Filters out raw log output
def filter_logs():

    del_list = []
    for raw_log in get_dl_links():
        raw_log = requests.get(raw_log)
        if 'UserContact' and 'deleted' in raw_log.text:
            del_list.append(json.dumps(raw_log.text))
    return del_list


# Send alert through Opsgenie
def alert_genie():

    a_list_of_logs = filter_logs()

    for output in a_list_of_logs:
        pattern = re.findall(r'[\w\.-]+@[\w\.-]+', output)

        for i in pattern:
            in_list = []
            n = i in in_list
            if n == False:
                in_list.append(i)

    for emails in in_list:
        alert_data_payload = {
            "message": "User Contact Info Deleted",
            "alias": "Username",
            "description": "A user deleted their contact info",
            "responders": [
                {"id": "79202f81-1a5d-4161-8ec6-6798a8738a45",
                 "type": "team"},
                {"name": "Test Team - Wiki Calendar",
                 "type": "team"},
                {"id": "b5be0234-df95-4b17-8669-5619e3c4ace0",
                 "type": "schedule"},
                {"name": "Test Team - Wiki Calendar_schedule",
                 "type": "schedule"}
            ],
            "visibleTo": [
                {"id": "79202f81-1a5d-4161-8ec6-6798a8738a45",
                 "type": "team"},
                {"name": "Test Team - Wiki Calendar",
                 "type": "team"}
            ],
            "tags": ["OverwriteQuietHours",
                     "Critical"],
            "details":
                {"Username": emails},
            "priority": "P2"
        }

        requests.post(alert_url, data=json.dumps(alert_data_payload), headers=alert_header_payload)



####### MAIN #######
def lambda_handler(event, context): 
    try:
        if len(filter_logs()) > 0:
            print('Sending Alert via Opsgenie...\n')
            alert_genie()
        else:
            pass
    except IndexError:
        print('Check the index (i) in get_filename... \n')
    return {
        'statusCode': 200,
        'body': json.dumps("Put back your contact info in Opsgenie or I'll eat your soul!")
    }

Configuring AWS Lambda

Log into AWS Console and then go the services drop down and select "Lambda" from the list. Click on "Create function".

Select "Author from scratch" and give it a name, I called mine "genie1". Select python 3.7 for the runtime and leave the permissions as default. Then click on "Create function".

Now on your local machine create a new folder and copy and paste your code and all of it's dependencies. Then zip them all up in one zip file.

On the AWS console go to the function code section in your Lambda function and select "Upload a .zip file" for code entry type. Then click on the "Upload" button and select your zip file.

You may want to play with the basic settings and edit the timeout and memory usage for your function. I set mine to 256 MB and a 3 minute timeout.

Once you have uploaded your function and it's dependencies you will see the code displayed within Lambda's code editor.

Scroll back up to designer and click on "Add trigger".

It will open a new window for trigger configuration. Select CloudWatch events as your trigger.

Give the rule a name and make sure to select "Schedule expression" and then fill in the cron for schedule expression. I went with executing the lambda function at 7AM every day. 0 7 * * ? *

Now we're all primed and set to go. CloudWatch will execute the Lambda function everyday at 7AM. Let's go ahead and test the function! Create a test using the default values. Then click on "Test".

You should see an output similar to mine below.

Success! AWS recognizes it's working now check if you got an email, text, message from Opsgenie. The email from Opsgenie will look something like this.

Sweet now, AWS Lambda is running our code.  You don't have to use Opsgenie as an API endpoint. You can use Zabbix's API or Yahoo weather's API. With the power of AWS Lambda behind you, all you need to do is focus on your code and let AWS handle the system administration aspect of code delivery.