Untitled
unknown
plain_text
9 months ago
24 kB
11
Indexable
import argparse
import subprocess
import json
import datetime
import os
import glob
import csv
from concurrent.futures import ThreadPoolExecutor, as_completed
def run_command(command):
"""Run a shell command and return the output."""
try:
output = subprocess.check_output(command, shell=True)
return output.decode('utf-8').strip()
except subprocess.CalledProcessError as e:
print(f"Failed to run command: {command}")
print(f"Error: {e}")
return None
def run_command_list(args):
"""Run a command with a list of arguments to avoid shell parsing issues."""
try:
output = subprocess.check_output(args)
return output.decode('utf-8').strip()
except subprocess.CalledProcessError as e:
print(f"Failed to run command: {' '.join(args)}")
print(f"Error: {e}")
return None
def check_state_history_exists(rack_serial, device_serial):
"""Check if state_history.json exists for a specific rack_serial/device_serial combination."""
bucket_name = "burninator_orchestrator"
namespace = "axi697ybxm4m"
object_name = f"{rack_serial}/{device_serial}/state_history.json"
cmd = [
"oci", "os", "object", "head",
"--bucket-name", bucket_name,
"--namespace", namespace,
"--name", object_name,
"--profile", "BOAT",
"--auth", "security_token"
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return True
except subprocess.CalledProcessError:
return False
def download_state_history(rack_serial, device_serial, save_dir):
"""Download state_history.json file for a specific rack_serial/device_serial combination."""
bucket_name = "burninator_orchestrator"
namespace = "axi697ybxm4m"
object_name = f"{rack_serial}/{device_serial}/state_history.json"
local_path = os.path.join(save_dir, "state_history.json")
os.makedirs(save_dir, exist_ok=True)
cmd = [
"oci", "os", "object", "get",
"--bucket-name", bucket_name,
"--namespace", namespace,
"--name", object_name,
"--file", local_path,
"--profile", "BOAT",
"--auth", "security_token"
]
try:
subprocess.run(cmd, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
print(f"Downloaded state_history.json for {rack_serial}/{device_serial}")
except subprocess.CalledProcessError as e:
print(f"Error downloading {object_name}: {e.stderr}")
def threaded_download_state_histories(download_tasks, max_workers=8):
"""Run downloads in parallel using threads."""
print(f"Starting threaded download of {len(download_tasks)} state history files...")
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [
executor.submit(download_state_history, rack_serial, device_serial, save_dir)
for (rack_serial, device_serial, save_dir) in download_tasks
]
for future in as_completed(futures):
try:
future.result()
except Exception as e:
print(f"Exception during threaded download: {e}")
print("Download completed!")
# Test case categorization
ILOM_BOOTSTRAP_TESTS = [
'test_clear_faults',
'test_ilom_firmware_upgrade',
'test_pldm_firmware_update',
'test_validate_server_assets',
'test_ilom_health_command',
'test_bootstrap_ilom_arts'
]
BOOTSTRAP_TESTS = [
'test_setup_sshfs_bootstrap_configuration',
'test_reboot_command',
'test_validate_server',
'test_cx7_cx8_and_ssd_firmware_upgrade',
'test_max_clock_speed_setting',
'test_osfp_internal_probes'
]
def load_state_history_json(file_path):
"""Load and parse state history JSON file."""
try:
with open(file_path, 'r') as f:
data = json.load(f)
return data
except (json.JSONDecodeError, FileNotFoundError) as e:
print(f"Error loading {file_path}: {e}")
return []
def get_workflow_for_test(test_name):
"""Determine which workflow a test belongs to."""
if test_name in ILOM_BOOTSTRAP_TESTS:
return 'ILOM_BOOTSTRAP'
elif test_name in BOOTSTRAP_TESTS:
return 'BOOTSTRAP'
else:
return 'OTHER'
def calculate_test_runtime(events):
"""Calculate runtime for a specific test across all job runs."""
test_events = [e for e in events if e.get('test_name')]
if not test_events:
return None, None, None
# Group by test_name and job_id
test_jobs = {}
for event in test_events:
test_name = event['test_name']
job_id = event['job_id']
key = f"{test_name}_{job_id}"
if key not in test_jobs:
test_jobs[key] = []
test_jobs[key].append(event)
runtimes = []
failure_count = 0
retry_count = 0
for key, job_events in test_jobs.items():
# Sort by timestamp
job_events.sort(key=lambda x: x['timestamp'])
# Find start and end times
start_time = None
end_time = None
final_status = None
for event in job_events:
if event['status'] in ['IN_PROGRESS', 'PASSED', 'FAILED']:
if start_time is None:
start_time = event['timestamp']
end_time = event['timestamp']
final_status = event['status']
if start_time and end_time:
start_dt = datetime.datetime.fromisoformat(start_time.replace('Z', '+00:00'))
end_dt = datetime.datetime.fromisoformat(end_time.replace('Z', '+00:00'))
runtime_minutes = (end_dt - start_dt).total_seconds() / 60
runtimes.append(runtime_minutes)
if final_status == 'FAILED':
failure_count += 1
elif final_status == 'SKIPPED':
retry_count += 1
if not runtimes:
return None, 0, 0
avg_runtime = sum(runtimes) / len(runtimes)
total_runs = len(runtimes)
return avg_runtime, failure_count, total_runs
def calculate_workflow_runtime(events, workflow_name):
"""Calculate total runtime for a workflow."""
workflow_tests = ILOM_BOOTSTRAP_TESTS if workflow_name == 'ILOM_BOOTSTRAP' else BOOTSTRAP_TESTS
workflow_events = [e for e in events if e.get('test_name') in workflow_tests]
if not workflow_events:
return None
# Group by job_id to get complete workflow runs
job_events = {}
for event in workflow_events:
job_id = event['job_id']
if job_id not in job_events:
job_events[job_id] = []
job_events[job_id].append(event)
workflow_runtimes = []
for job_id, job_events_list in job_events.items():
# Sort by timestamp
job_events_list.sort(key=lambda x: x['timestamp'])
# Find the earliest start and latest end for this job
start_time = None
end_time = None
for event in job_events_list:
if event['status'] in ['IN_PROGRESS', 'PASSED', 'FAILED']:
if start_time is None:
start_time = event['timestamp']
end_time = event['timestamp']
if start_time and end_time:
start_dt = datetime.datetime.fromisoformat(start_time.replace('Z', '+00:00'))
end_dt = datetime.datetime.fromisoformat(end_time.replace('Z', '+00:00'))
runtime_minutes = (end_dt - start_dt).total_seconds() / 60
workflow_runtimes.append(runtime_minutes)
if not workflow_runtimes:
return None
return sum(workflow_runtimes) / len(workflow_runtimes)
def create_combined_csv(csv_file_path, workflow_data, testcase_data, posthandover_ticket_count):
"""Create a single comprehensive CSV file with all data."""
# Calculate summary data
summary_data = []
for test_name in set(tc['Test_Name'] for tc in testcase_data):
test_records = [tc for tc in testcase_data if tc['Test_Name'] == test_name]
avg_runtime = sum(tc['Avg_Runtime_Minutes'] for tc in test_records) / len(test_records)
avg_failure_rate = sum(tc['Failure_Rate_Percent'] for tc in test_records) / len(test_records)
total_runs = sum(tc['Total_Runs'] for tc in test_records)
summary_data.append({
'Test_Name': test_name,
'Workflow': get_workflow_for_test(test_name),
'Avg_Runtime_Minutes': avg_runtime,
'Total_Runs_Across_All_Devices': total_runs,
'Avg_Failure_Rate_Percent': avg_failure_rate,
'Device_Count': len(test_records)
})
# Calculate slowest test case per workflow per rack
slowest_tests_data = []
rack_workflow_tests = {}
# Group test cases by rack and workflow
for data in testcase_data:
rack_serial = data['Rack_Serial']
workflow = data['Workflow']
key = f"{rack_serial}_{workflow}"
if key not in rack_workflow_tests:
rack_workflow_tests[key] = []
rack_workflow_tests[key].append(data)
# Find slowest test case for each rack-workflow combination
for key, tests in rack_workflow_tests.items():
if tests:
# Find the test with maximum runtime
slowest_test = max(tests, key=lambda x: x['Avg_Runtime_Minutes'])
slowest_tests_data.append({
'Rack_Serial': slowest_test['Rack_Serial'],
'Workflow': slowest_test['Workflow'],
'Slowest_Test_Name': slowest_test['Test_Name'],
'Max_Runtime_Minutes': slowest_test['Avg_Runtime_Minutes'],
'Failure_Rate_Percent': slowest_test['Failure_Rate_Percent']
})
with open(csv_file_path, 'w', newline='', encoding='utf-8') as csvfile:
writer = csv.writer(csvfile)
# Write section headers and data
# 0. PostHandover Tickets Count Section
writer.writerow(['=== POSTHANDOVER TICKETS COUNT ==='])
writer.writerow(['Number_of_PostHandover_Tickets'])
writer.writerow([posthandover_ticket_count])
writer.writerow([]) # Empty row separator
# 1. Workflow Runtimes Section
writer.writerow(['=== WORKFLOW RUNTIMES ==='])
writer.writerow(['Rack_Serial', 'Device_Serial', 'ILOM_BOOTSTRAP_Runtime_Minutes', 'BOOTSTRAP_Runtime_Minutes'])
for data in workflow_data:
writer.writerow([
data['Rack_Serial'],
data['Device_Serial'],
data['ILOM_BOOTSTRAP_Runtime_Minutes'],
data['BOOTSTRAP_Runtime_Minutes']
])
writer.writerow([]) # Empty row separator
# 2. Test Case Analysis Section
writer.writerow(['=== TEST CASE ANALYSIS ==='])
writer.writerow(['Rack_Serial', 'Device_Serial', 'Test_Name', 'Workflow', 'Avg_Runtime_Minutes',
'Total_Runs', 'Failure_Count', 'Failure_Rate_Percent'])
for data in testcase_data:
writer.writerow([
data['Rack_Serial'],
data['Device_Serial'],
data['Test_Name'],
data['Workflow'],
data['Avg_Runtime_Minutes'],
data['Total_Runs'],
data['Failure_Count'],
data['Failure_Rate_Percent']
])
writer.writerow([]) # Empty row separator
# 3. Test Case Summary Section
writer.writerow(['=== TEST CASE SUMMARY ==='])
writer.writerow(['Test_Name', 'Workflow', 'Avg_Runtime_Minutes', 'Total_Runs_Across_All_Devices',
'Avg_Failure_Rate_Percent', 'Device_Count'])
for data in summary_data:
writer.writerow([
data['Test_Name'],
data['Workflow'],
data['Avg_Runtime_Minutes'],
data['Total_Runs_Across_All_Devices'],
data['Avg_Failure_Rate_Percent'],
data['Device_Count']
])
writer.writerow([]) # Empty row separator
# 4. Slowest Test Cases per Workflow per Rack Section
writer.writerow(['=== SLOWEST TEST CASES PER WORKFLOW PER RACK ==='])
writer.writerow(['Rack_Serial', 'Workflow', 'Slowest_Test_Name', 'Max_Runtime_Minutes', 'Failure_Rate_Percent'])
for data in slowest_tests_data:
writer.writerow([
data['Rack_Serial'],
data['Workflow'],
data['Slowest_Test_Name'],
data['Max_Runtime_Minutes'],
data['Failure_Rate_Percent']
])
def analyze_state_history_files(csv_file_path, posthandover_ticket_count):
"""Analyze all downloaded state history files and create Excel report."""
print("Analyzing state history files...")
# Find all state history files
state_files = glob.glob("state_history_downloads/**/state_history.json", recursive=True)
if not state_files:
print("No state history files found in state_history_downloads/")
# Still create CSV with ticket count
csv_basename = os.path.basename(csv_file_path)
if csv_basename.endswith('.csv') or csv_basename.endswith('.xlsx'):
csv_basename = os.path.splitext(csv_basename)[0]
output_file = f"{csv_basename}_state_history_analysis.csv"
create_combined_csv(output_file, [], [], posthandover_ticket_count)
print(f"Created CSV with ticket count: {output_file}")
return
print(f"Found {len(state_files)} state history files")
# Extract base name from csv_file_path for sheet naming
csv_basename = os.path.basename(csv_file_path)
if csv_basename.endswith('.csv') or csv_basename.endswith('.xlsx'):
csv_basename = os.path.splitext(csv_basename)[0]
# Data collection
workflow_data = []
testcase_data = []
for file_path in state_files:
# Extract rack_serial and device_serial from path
path_parts = file_path.split('/')
if len(path_parts) >= 3:
rack_serial = path_parts[1]
device_serial = path_parts[2]
else:
continue
print(f"Processing: {rack_serial}/{device_serial}")
# Load state history data
events = load_state_history_json(file_path)
if not events:
continue
# Calculate workflow runtimes
ilom_bootstrap_runtime = calculate_workflow_runtime(events, 'ILOM_BOOTSTRAP')
bootstrap_runtime = calculate_workflow_runtime(events, 'BOOTSTRAP')
workflow_data.append({
'Rack_Serial': rack_serial,
'Device_Serial': device_serial,
'ILOM_BOOTSTRAP_Runtime_Minutes': ilom_bootstrap_runtime,
'BOOTSTRAP_Runtime_Minutes': bootstrap_runtime
})
# Calculate test case runtimes and failure rates
all_tests = set(e.get('test_name') for e in events if e.get('test_name'))
for test_name in all_tests:
if test_name in ILOM_BOOTSTRAP_TESTS + BOOTSTRAP_TESTS:
avg_runtime, failure_count, total_runs = calculate_test_runtime(
[e for e in events if e.get('test_name') == test_name]
)
if avg_runtime is not None:
failure_rate = (failure_count / total_runs * 100) if total_runs > 0 else 0
testcase_data.append({
'Rack_Serial': rack_serial,
'Device_Serial': device_serial,
'Test_Name': test_name,
'Workflow': get_workflow_for_test(test_name),
'Avg_Runtime_Minutes': avg_runtime,
'Total_Runs': total_runs,
'Failure_Count': failure_count,
'Failure_Rate_Percent': failure_rate
})
# Create single comprehensive CSV file
output_file = f"{csv_basename}_state_history_analysis.csv"
# Create combined CSV file (always create, even if no data, to include ticket count)
create_combined_csv(output_file, workflow_data, testcase_data, posthandover_ticket_count)
print(f"Created comprehensive CSV: {output_file}")
print(f"Analysis complete! Results saved to {output_file}")
print(f"Processed {len(workflow_data)} device combinations")
print(f"Analyzed {len(testcase_data)} test case records")
def get_repair_order_details(repair_order_id):
"""Get repair order details using burninatorOps.py command."""
repair_json_command = f"python3 ../../burninator_ops/burninatorOps.py repair get -id '{repair_order_id}'"
repair_json_output = run_command(repair_json_command)
if repair_json_output is None:
return None
try:
repair_json = json.loads(repair_json_output)
except json.JSONDecodeError:
print(f"Failed to parse repair_json output: {repair_json_output}")
return None
return repair_json
def get_device_details(device_serial):
"""Get device details using burninatorOps.py command."""
device_command = f"python3 ../../burninator_ops/burninatorOps.py device get -s '{device_serial}'"
device_output = run_command(device_command)
if device_output is None:
return None
try:
device_json = json.loads(device_output)
except json.JSONDecodeError:
print(f"Failed to parse device_json output: {device_output}")
return None
return device_json
def process_query_results(results):
"""Process query results to extract device serials and rack serials."""
rack_serial_device_serials = {}
tickets_processed = 0
download_tasks = []
scheduled = set()
# Count total number of postHandover tickets
total_tickets = 0
for result in results.split('\n'):
if result.startswith('ROMAOPS'):
total_tickets += 1
for result in results.split('\n'):
try:
if result.startswith('ROMAOPS'):
fields = result.split(',')
if len(fields) >= 4:
key = fields[0]
title = fields[1]
rack_position = fields[2]
repair_order_id = fields[3].strip()
print(f"Processing result: {result}")
print(
f"Key: {key}, Title: {title}, Rack Position: {rack_position}, Repair Order ID: {repair_order_id}")
if repair_order_id:
repair_details = get_repair_order_details(repair_order_id)
if repair_details:
device_serial = repair_details.get('deviceSerial')
if device_serial:
print(f"Device Serial: {device_serial}")
# Get device details to get rack serial
device_details = get_device_details(device_serial)
if device_details:
rack_serial = device_details.get('rackSerial')
print(f"Rack Serial: {rack_serial}")
# Store the mapping
if rack_serial in rack_serial_device_serials:
rack_serial_device_serials[rack_serial].append(device_serial)
else:
rack_serial_device_serials[rack_serial] = [device_serial]
# Check if state history file exists and create download task
if (rack_serial, device_serial) not in scheduled:
if check_state_history_exists(rack_serial, device_serial):
save_dir = os.path.join("state_history_downloads", rack_serial,
device_serial)
download_tasks.append((rack_serial, device_serial, save_dir))
scheduled.add((rack_serial, device_serial))
print(f"Added download task for {rack_serial}/{device_serial}")
else:
print(f"State history file not found for {rack_serial}/{device_serial}")
tickets_processed += 1
print(f"Successfully processed ticket {key}")
else:
print(f"Failed to get device details for {device_serial}")
else:
print("No device serial found in repair order")
else:
print("Failed to get repair order details")
else:
print("No repair order ID found in ticket")
print("-" * 80)
else:
print(f"Skipping malformed result: {result}")
except Exception as e:
print(f"Skipping ticket entry: {result}")
print(f"Failure cause: {e}")
print(f"Total tickets processed: {tickets_processed}")
print(f"Total postHandover tickets: {total_tickets}")
print(f"Total download tasks collected: {len(download_tasks)}")
return rack_serial_device_serials, download_tasks, total_tickets
def query_tickets():
"""Query tickets using burninatorOps.py directly."""
query = 'Project = ROMAOPS and ("Searchable Data" ~ "HandedOver" or Title ~ "ROMA_CUSTOMER") and status in ("Resolved", "Closed","Done","Pending Customer") and (Description ~ "gpu_count_mismatch") AND "component" = "ROMA-Ingestion"'
args = [
'python3',
'../../burninator_ops/burninatorOps.py',
'ots',
'query',
'-q', query,
'-f', 'key,title,rack_position,repair_order_id,component,status,searchable_data,labels'
]
command_str = ' '.join(args)
print(f"Running query with command: \n {command_str}")
results = run_command_list(args)
if results is None:
print("Failed to get query results")
return
print("Query results:")
print(results)
print("=" * 80)
# Process the results to get device serials and rack serials
rack_serial_device_serials, download_tasks, total_tickets = process_query_results(results)
# Print summary
print(f"\nSummary:")
print(f"Found {len(rack_serial_device_serials)} unique rack serials")
for rack_serial, device_serials in rack_serial_device_serials.items():
print(f"Rack Serial: {rack_serial} -> Device Serials: {device_serials}")
# Download state history files for all tickets
if download_tasks:
print(f"\nDownloading state history files for {len(download_tasks)} folders...")
threaded_download_state_histories(download_tasks, max_workers=8)
else:
print("\nNo state history files to download.")
return rack_serial_device_serials, download_tasks, total_tickets
def main(csv_file_path):
print(f"The Script output path: {csv_file_path}")
# Query tickets and get deviceSerial and rackSerial
result = query_tickets()
if result:
rack_serial_device_serials, download_tasks, total_tickets = result
print(f"\nFinal Summary:")
print(f"Processed {len(rack_serial_device_serials)} rack serials")
print(f"Total postHandover tickets: {total_tickets}")
print(f"Downloaded state history files for {len(download_tasks)} folders")
print(f"Files saved in: state_history_downloads/")
# Analyze the downloaded state history files
print(f"\nStarting state history analysis...")
analyze_state_history_files(csv_file_path, total_tickets)
else:
print("No results to process.")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='KPI Analysis Script')
parser.add_argument('--csv_file_path', type=str, help='Path to CSV files', required=True)
args = parser.parse_args()
main(args.csv_file_path)Editor is loading...
Leave a Comment