Automate Error Reports from GCP to Jira along with attaching log file

automate_error_jira.py

#!/usr/bin/env python3
"""
This is a prototype for automating prod error tickets from GCP to Jira.
"""
import os
import json
import argparse
import subprocess
import logging as pylog
import datetime as d
import requests
from requests.auth import HTTPBasicAuth
from google.cloud import logging

fmt_str = '%(message)s'
client = logging.Client()
client.setup_logging()

dry_run = False


def define_parameters(json_object) -> dict:
    """
		Initial definition of all variables
		"""
    project_id = json_object['group_info']['project_id']
    error_link = json_object['group_info']['detail_link'].split('&')[0]
    log_file_name = d.datetime.now().strftime(
        "/tmp/downloaded-logs-%Y%m%d-%H%M%S.json")
    impact_service = json_object['event_info']['service']
    impact_version = json_object['event_info']['version']
    stack_trace = json_object['event_info']['log_message']
    summary = f"{json_object['exception_info']['type']}: {json_object['exception_info']['message']}"
    error_group = error_link.split('errors/')[1].split('?project')[0]
    params = {
        'pid': project_id,
        'url': error_link,
        'i_s': impact_service,
        'i_v': impact_version,
        'stack': stack_trace,
        's': summary,
        'grp': error_group,
        'file': log_file_name
    }
    return params


def fix_vars(params) -> list:
    """
		Queries are different in case there is a service version specified or not.
		"""
    if not params['i_v']:
        log_query = f"error_group({params['grp']}) \
AND resource.labels.project_id={params['pid']} \
AND resource.labels.function_name={params['i_s']} \
AND severity=ERROR \
AND timestamp<2025"

        impact = params['i_s']
    else:
        log_query = f"error_group({params['grp']}) \
AND resource.labels.project_id={params['pid']} \
AND resource.labels.module_id={params['i_s']} \
AND resource.labels.version_id={params['i_v']} \
AND severity=ERROR \
AND timestamp<2025"

        impact = f"{params['i_s']}:{params['i_v']}"
    return [log_query, impact]


def jira_ticket_create(user, api_token, params, impact) -> (str, bool):
    """
		This is where jira ticket gets generated and we get a response with the ticket id
		"""
    data = json.dumps({
        "update": {},
        "fields": {
            "summary": f"Production - {params['s']}",
            "issuetype": {
                "id": "10000"
            },
            "project": {
                "id": "10000"
            },
            "description": {
                "type":
                "doc",
                "version":
                1,
                "content": [{
                    "type": "heading",
                    "attrs": {
                        "level": 2
                    },
                    "content": [{
                        "type": "text",
                        "text": "Summary"
                    }]
                }, {
                    "type":
                    "heading",
                    "attrs": {
                        "level": 3
                    },
                    "content": [{
                        "type": "text",
                        "text": f"{params['s']}",
                        "marks": [{
                            "type": "code"
                        }]
                    }]
                }, {
                    "type": "heading",
                    "attrs": {
                        "level": 2
                    },
                    "content": [{
                        "type": "text",
                        "text": "Error Link"
                    }]
                }, {
                    "type":
                    "paragraph",
                    "content": [{
                        "type":
                        "text",
                        "text":
                        f"{params['url']}",
                        "marks": [{
                            "type": "link",
                            "attrs": {
                                "href": f"{params['url']}"
                            }
                        }]
                    }]
                }, {
                    "type": "heading",
                    "attrs": {
                        "level": 2
                    },
                    "content": [{
                        "type": "text",
                        "text": "Impact"
                    }]
                }, {
                    "type":
                    "orderedList",
                    "content": [{
                        "type":
                        "listItem",
                        "content": [{
                            "type":
                            "paragraph",
                            "content": [{
                                "type": "text",
                                "text": f"Found in {impact}"
                            }]
                        }]
                    }]
                }, {
                    "type":
                    "heading",
                    "attrs": {
                        "level": 2
                    },
                    "content": [{
                        "type": "text",
                        "text": "Stack Trace"
                    }]
                }, {
                    "type":
                    "codeBlock",
                    "attrs": {},
                    "content": [{
                        "type": "text",
                        "text": f"{params['stack']}"
                    }]
                }]
            },
            "reporter": {
                "id": "fffffffffffffffffffff"
            },
            "labels": ["bugs-1337", "prod-tickets"],
            "assignee": {
                "id": "-1"
            }
        }
    })
    auth = HTTPBasicAuth(user, api_token)
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    try:
        create_ticket = requests.request(
            "POST",
            url="https://<org_url>.atlassian.net/rest/api/3/issue",
            headers=headers,
            data=data,
            auth=auth)
    except Exception as e:
        pylog.error(f"Error occured during attaching file: {e}")
        return False

    pylog.info(create_ticket.json())
    return create_ticket.json()['key']


def attach_log_file(user, api_token, key, query, log_file_name) -> bool:
    """
		the ticket id is taken from the creation function and then passed here to attach the queried log file (max log downloaded is 2)
		"""
    entries = client.list_entries(filter_=query)

    with open(log_file_name, mode="w") as DATA:
        for each in entries:
            log_write = {
                "httpRequest": each.http_request,
                "insertId": each.insert_id,
                "labels": each.labels,
                "logName": each.log_name,
                "operation": each.operation,
                "textPayload": each.payload,
                "receiveTimestamp": str(each.received_timestamp),
                "resource": each.resource,
                "severity": each.severity,
                "spanId": each.span_id,
                "timestamp": str(each.timestamp.isoformat()),
                "sourceLocation": each.source_location,
                "trace": each.trace,
                "traceSampled": each.trace_sampled
            }
            DATA.write(json.dumps(log_write, indent=4))

    auth = HTTPBasicAuth(user, api_token)
    headers = {"Accept": "application/json", "X-Atlassian-Token": "no-check"}
    try:
        attach_log_file = requests.request(
            "POST",
            url=
            f"https://<org_url>.atlassian.net/rest/api/3/issue/{key}/attachments",
            headers=headers,
            auth=auth,
            files={
                "file": (log_file_name, open(log_file_name,
                                             "r"), "application-type")
            })
    except Exception as e:
        pylog.error(f"Error occured during attaching file: {e}")
        return False

    pylog.info(attach_log_file.json())
    os.remove(log_file_name)
    return True


def main(request):
    user = os.environ.get('user')
    token = os.environ.get('token')

    json_output = request.get_json()
    params = define_parameters(json_output)
    query_and_impact = fix_vars(params)
    ticket_key = jira_ticket_create(user, token, params, query_and_impact[1])
    if ticket_key:
        attach_log_file(user, token, ticket_key, query_and_impact[0],
                        params['file'])
        return {'status': 'SUCCESS'}
    else:
        return {'status': 'FAILURE'}