Python Notebook code example for ROI metrics.

.env

RUNDECK_HOST={YOURHOST}
RUNDECK_API_KEY={YOURKEY}

roi.py

# ROI Metrics Example
# Supplied as-is. Please set .env with RUNDECK_HOST and RUNDECK_API_KEY

import requests
import os
import json
import pandas as pd
from dotenv import load_dotenv
from IPython.display import display, HTML

# Load environment variables
load_dotenv(".env")

# Initialize summary DataFrame
summary_df = pd.DataFrame(columns=[
    'Job Tags',
    'Time Frame',
    'Total successful executions',
    'Total savings for Job',
    'Total time saved for Job'
])

# Fetch configuration from environment variables
def get_config():
    return {
        'rundeckhost': os.getenv('RUNDECK_HOST', 'https://rdse.runbook.pagerduty.cloud'),
        'api_key': os.getenv('RUNDECK_API_KEY', 'Your-API-Key-Here'),
        'api_version': '45',
        'project': 'pdt-emea-production',
        'tags': 'roi',
        'execution_history': '1d'
    }
    
# Generate API headers
def get_headers(api_key):
    return {
        'X-Rundeck-Auth-Token': api_key,
        'Content-Type': 'application/json',
        'Accept': 'application/json'
    }

# List jobs by a specific tag
def list_jobs_by_tag(config, headers):
    url = f"{config['rundeckhost']}/api/{config['api_version']}/project/{config['project']}/jobs?tags={config['tags']}"
    response = requests.get(url, headers=headers)
    return "&".join([f"jobIdListFilter={job['id']}" for job in response.json()])

# List executions for specific jobs
def list_executions_from_jobs(config, headers, jobIDList):
    url = f"{config['rundeckhost']}/api/{config['api_version']}/project/{config['project']}/executions?max=1000&statusFilter=succeeded&recentFilter={config['execution_history']}&{jobIDList}"
    response = requests.get(url, headers=headers)
    return response.json().get('executions', [])

# Fetch ROI data for a specific execution
def get_roi_data(config, headers, execution_id, df_list):
    url = f"{config['rundeckhost']}/api/{config['api_version']}/execution/{execution_id}/roimetrics/data"
    response = requests.get(url, headers=headers)
    if 'job_execution_id' in response.json():
        df_list.append(pd.DataFrame([response.json()]))

# Format duration from seconds to HH:MM:SS
def format_duration(seconds):
    hours, remainder = divmod(seconds, 3600)
    minutes, seconds = divmod(remainder, 60)
    return f"{hours:02}:{minutes:02}:{seconds:02}"

def main():
    config = get_config()
    headers = get_headers(config['api_key'])
    print("Starting ROI Job")

    jobIDList = list_jobs_by_tag(config, headers)
    executions_list = list_executions_from_jobs(config, headers, jobIDList)

    df_list = []
    for execution in executions_list:
        get_roi_data(config, headers, execution['id'], df_list)

    final_df = pd.concat(df_list, ignore_index=True)
    if final_df.empty:
        print("No data to display.")
        return

    # Data manipulation and calculations
    final_df['Saving'] = final_df['Saving'].str.replace('$', '').astype(float)
    final_df.fillna(0, inplace=True)
    total_savings = final_df['Saving'].sum()
    total_job_time = final_df['job_duration_secs'].astype(int).sum()
    total_executions = len(final_df)

    # Append summary to DataFrame
    new_row = pd.DataFrame({
        'Job Tags': [config['tags']],
        'Time Frame': [config['execution_history']],
        'Total successful executions': [total_executions],
        'Total savings for Job': [f"${total_savings}"],
        'Total time saved for Job': [format_duration(total_job_time)]
    })
    global summary_df
    summary_df = pd.concat([summary_df, new_row], ignore_index=True)
    print("Summary Statistics:")
    display(HTML(summary_df.to_html()))
    #display(HTML(final_df.to_html()))

if __name__ == "__main__":
    main()