services/jenkins-pipeline-monitor/handler.py

import os
import boto3
import json
import logging
import secret_manager

from datetime import datetime, timezone
from jenkinsapi.jenkins import Jenkins

logging.getLogger().setLevel(logging.INFO)
logging.getLogger('boto3').setLevel(logging.CRITICAL)
logging.getLogger('botocore').setLevel(logging.CRITICAL)

release_job_type = ['mxnet_lib', 'python/pypi', 'python/docker']


def get_jenkins_obj(secret):
    """
    This method returns an object of Jenkins instantiated using username, password
    """
    jenkins_url, jenkins_username, jenkins_password = os.environ["JENKINS_URL"], secret["jenkins_username"], secret["jenkins_password"]
    return Jenkins(jenkins_url, username=jenkins_username, password=jenkins_password)


def get_secret():
    """
    This method is to get secret value from Secrets Manager
    """
    secret = json.loads(secret_manager.get_secret())
    return secret


def get_pipeline_job(jenkinsObj):
    job = jenkinsObj["restricted-mxnet-cd/mxnet-cd-release-job"]
    return job


def get_latest_build_number(job):
    return job.get_last_build().get_number()


def get_build_from_build_number(job, build_number):
    return job.get_build(build_number)


def get_build_timestamp(build):
    return build.get_timestamp()


def get_build_date(timestamp):
    return timestamp.date()


def is_latest_day_build(current_build):
    current_build_timestamp = get_build_timestamp(current_build)
    current_time_stamp = datetime.now().replace(tzinfo=timezone.utc)
    # if current build is within 24 hours of the current time
    seconds_difference = (current_time_stamp - current_build_timestamp).total_seconds()
    hour_difference = divmod(seconds_difference, 3600)[0]
    if(hour_difference < 24):
        return True
    else:
        return False


def get_latest_day_builds(job, latest_build_number):
    """
    Get all the builds that were triggered in the past 24 hours from the current time
    i.e. the time when the Lambda function is triggered
    :param job: Jenkins Job object
    :param latest_build_number: latest build number from which to start checking
    :result: List[builds]
    """
    builds = []
    current_build_number = latest_build_number
    while True:
        current_build = get_build_from_build_number(job, current_build_number)
        if is_latest_day_build(current_build):
            builds.append(current_build)
            current_build_number -= 1
        else:
            break
    return builds


def get_release_job_type(build):
    return build.get_params()['RELEASE_JOB_TYPE']


def filter_by_release_job_type(latest_day_builds):
    filtered_builds = []
    for build in latest_day_builds:
        if get_release_job_type(build) in release_job_type:
            filtered_builds.append(build)
    return filtered_builds


def status_check(builds):
    """
    Check the status of the filtered builds
    i.e. Check if all the required release job types are present in the pipeline
    If a build from the list of desired release job types doesn't exist, log the failure
    else check the status via Jenkins API and report accordingly
    :param builds
    """
    # dictionary of the type release_job_type: count
    # e.g. {'mxnet_lib/static':0, 'python/pypi':0}
    global release_job_type
    success_count = 0
    release_job_type_dict = {el: 0 for el in release_job_type}

    # iterate over the builds to count number of the desired release job types
    for build in builds:
        build_release_job_type = get_release_job_type(build)
        if build.get_status() == 'SUCCESS':
            logging.info(f'Successful build {build_release_job_type} {build.get_number()}')
        else:
            logging.info(f'Failure build {build_release_job_type} {build.get_number()}')
        release_job_type_dict[build_release_job_type] += 1

    # iterate over the map of release_job_type: count
    # if 'mxnet_lib/static':1 indicates static jobtype job ran in the pipeline
    # else 'mxnet_lib/static':0 indicates static jobtype never ran -> log as failed
    for release_job_type_name, release_job_type_count in release_job_type_dict.items():
        if release_job_type_count == 0:
            logging.info(f'Failure build {release_job_type_name}')
        elif release_job_type_count == 1:
            success_count += 1
        else:
            logging.info(f'{release_job_type} ran {release_job_type_count} times')
    # if success_count = 2 [i.e. len of release_job_type], it means both static & pypi jobs have run
    if success_count == len(release_job_type):
        logging.info(f'All the required jobs ran')
    else:
        logging.info(f'1/more of the required jobs did not run')


def get_cause(build):
    return build.get_causes()[0]['_class']


def filter_by_upstream_cause(builds, desired_cause):
    filtered_builds = []
    for build in builds:
        if get_cause(build) == desired_cause:
            filtered_builds.append(build)
    return filtered_builds


def jenkins_pipeline_monitor():
    # retrieve secret from secert manager
    secret = get_secret()
    logging.info(f'Secrets retrieved')
    # get jenkins object
    jenkinsObj = get_jenkins_obj(secret)
    logging.info(f'Jenkins obj created')
    # get relevant pipeline job
    job = get_pipeline_job(jenkinsObj)
    logging.info(f'Job fetch {job}')
    # get the latest build on the pipeline
    latest_build_number = get_latest_build_number(job)

    # get builds scheduled for the latest day
    latest_day_builds = get_latest_day_builds(job, latest_build_number)
    logging.info(f'latest builds {latest_day_builds}')

    # exit if no builds found
    if not latest_day_builds:
        logging.error('No builds for the latest day')
        return

    # filter latest day builds by desired build type a.k.a release job type
    filtered_builds = filter_by_release_job_type(latest_day_builds)
    logging.info(f'Builds filtered by desired release job type : {filtered_builds}')

    # exit if no builds of desired build type
    if not filtered_builds:
        logging.error('No builds of desired type')
        return

    desired_cause = 'hudson.model.Cause$UpstreamCause'
    filtered_builds = filter_by_upstream_cause(filtered_builds, desired_cause)

    # exit if builds not triggered by UpstreamCause
    if not filtered_builds:
        logging.error(f'Builds dont belong to desired cause:{desired_cause}')
    else:
        logging.info(f'Filtered builds by {desired_cause} : {filtered_builds}')
        status_check(filtered_builds)


def lambda_handler(event, context):
    try:
        logging.info(f'Lambda handler invoked')
        jenkins_pipeline_monitor()
    except Exception as e:
        logging.error("Lambda raised an exception! %s", exc_info=e)


if __name__ == "__main__":
    logging.basicConfig(level=logging.DEBUG)
    lambda_handler(None, None)