546 lines · python
1# Part of the LLVM Project, under the Apache License v2.0 with LLVM Exceptions.2# See https://llvm.org/LICENSE.txt for license information.3# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception4"""Collects Github metrics and uploads them to Grafana.5 6This script contains machinery that will pull metrics periodically from Github7about workflow runs. It will upload the collected metrics to the specified8Grafana instance.9"""10 11import collections12import datetime13import github14import logging15import os16import requests17import time18 19from dataclasses import dataclass20from github import Auth21from github import Github22 23GRAFANA_URL = (24 "https://influx-prod-13-prod-us-east-0.grafana.net/api/v1/push/influx/write"25)26SCRAPE_INTERVAL_SECONDS = 5 * 6027 28# Lists the Github workflows we want to track. Maps the Github job name to29# the metric name prefix in grafana.30# This metric name is also used as a key in the job->name map.31GITHUB_WORKFLOW_TO_TRACK = {32 "CI Checks": "github_llvm_premerge_checks",33 "Build and Test libc++": "github_libcxx_premerge_checks",34}35 36# Lists the Github jobs to track for a given workflow. The key is the stable37# name (metric name) of the workflow (see GITHUB_WORKFLOW_TO_TRACK).38# Each value is a map to link the github job name to the corresponding metric39# name.40GITHUB_JOB_TO_TRACK = {41 "github_llvm_premerge_checks": {42 "Build and Test Linux": "premerge_linux",43 "Build and Test Linux AArch64": "premerge_linux_aarch64",44 "Build and Test Windows": "premerge_windows",45 },46 "github_libcxx_premerge_checks": {47 "stage1": "premerge_libcxx_stage1",48 "stage2": "premerge_libcxx_stage2",49 "stage3": "premerge_libcxx_stage3",50 },51}52 53# The number of workflows to pull when sampling Github workflows.54# - Github API filtering is broken: we cannot apply any filtering:55# - See https://github.com/orgs/community/discussions/8676656# - A workflow can complete before another workflow, even when starting later.57# - We don't want to sample the same workflow twice.58#59# This means we essentially have a list of workflows sorted by creation date,60# and that's all we can deduce from it. So for each iteration, we'll blindly61# process the last N workflows.62GITHUB_WORKFLOWS_MAX_PROCESS_COUNT = 200063# Second reason for the cut: reaching a workflow older than X.64# This means we will miss long-tails (exceptional jobs running for more than65# X hours), but that's also the case with the count cutoff above.66# Only solution to avoid missing any workflow would be to process the complete67# list, which is not possible.68GITHUB_WORKFLOW_MAX_CREATED_AGE_HOURS = 869 70# Grafana will fail to insert any metric older than ~2 hours (value determined71# by trial and error).72GRAFANA_METRIC_MAX_AGE_MN = 12073 74 75@dataclass76class JobMetrics:77 job_name: str78 queue_time: int79 run_time: int80 status: int81 created_at_ns: int82 started_at_ns: int83 completed_at_ns: int84 workflow_id: int85 workflow_name: str86 87 88@dataclass89class GaugeMetric:90 name: str91 value: int92 time_ns: int93 94 95@dataclass96class AggregateMetric:97 aggregate_name: str98 aggregate_queue_time: int99 aggregate_run_time: int100 aggregate_status: int101 completed_at_ns: int102 workflow_id: int103 104 105def _construct_aggregate(ag_name: str, job_list: list[JobMetrics]) -> AggregateMetric:106 """Create a libc++ AggregateMetric from a list of libc++ JobMetrics107 108 How aggregates are computed:109 queue time: Time from when first job in group is created until last job in110 group has started.111 run time: Time from when first job in group starts running until last job112 in group finishes running.113 status: logical 'and' of all the job statuses in the group.114 115 Args:116 ag_name: The name for this particular AggregateMetric117 job_list: This list of JobMetrics to be combined into the AggregateMetric.118 The input list should contain all (and only!) the libc++ JobMetrics119 for a particular stage and a particular workflow_id.120 121 Returns:122 Returns the AggregateMetric constructed from the inputs.123 """124 125 # Initialize the aggregate values126 earliest_create = job_list[0].created_at_ns127 earliest_start = job_list[0].started_at_ns128 earliest_complete = job_list[0].completed_at_ns129 latest_start = job_list[0].started_at_ns130 latest_complete = job_list[0].completed_at_ns131 ag_status = job_list[0].status132 ag_workflow_id = job_list[0].workflow_id133 134 # Go through rest of jobs for this workflow id, if any, updating stats135 for job in job_list[1:]:136 # Update the status137 ag_status = ag_status and job.status138 # Get the earliest & latest times139 if job.created_at_ns < earliest_create:140 earliest_create = job.created_at_ns141 if job.completed_at_ns < earliest_complete:142 earliest_complete = job.completed_at_ns143 if job.started_at_ns > latest_start:144 latest_start = job.started_at_ns145 if job.started_at_ns < earliest_start:146 earliest_start = job.started_at_ns147 if job.completed_at_ns > latest_complete:148 latest_complete = job.completed_at_ns149 150 # Compute aggregate run time (in seconds, not ns)151 ag_run_time = (latest_complete - earliest_start) / 1000000000152 # Compute aggregate queue time (in seconds, not ns)153 ag_queue_time = (latest_start - earliest_create) / 1000000000154 # Append the aggregate metrics to the workflow metrics list.155 return AggregateMetric(156 ag_name, ag_queue_time, ag_run_time, ag_status, latest_complete, ag_workflow_id157 )158 159 160def create_and_append_libcxx_aggregates(workflow_metrics: list[JobMetrics]):161 """Find libc++ JobMetric entries and create aggregate metrics for them.162 163 Sort the libc++ JobMetric entries by workflow id, and for each workflow164 id group them by stages. Call _construct_aggregate to reate an aggregate165 metric for each stage for each unique workflow id. Append each aggregate166 metric to the input workflow_metrics list.167 168 Args:169 workflow_metrics: A list of JobMetrics entries collected so far.170 """171 # Separate the jobs by workflow_id. Only look at JobMetrics entries.172 aggregate_data = dict()173 for job in workflow_metrics:174 # Only want to look at JobMetrics175 if not isinstance(job, JobMetrics):176 continue177 # Only want libc++ jobs.178 if job.workflow_name != "Build and Test libc++":179 continue180 if job.workflow_id not in aggregate_data.keys():181 aggregate_data[job.workflow_id] = [job]182 else:183 aggregate_data[job.workflow_id].append(job)184 185 # Go through each aggregate_data list (workflow id) and find all the186 # needed data187 for ag_workflow_id in aggregate_data:188 job_list = aggregate_data[ag_workflow_id]189 stage1_jobs = list()190 stage2_jobs = list()191 stage3_jobs = list()192 # sort jobs into stage1, stage2, & stage3.193 for job in job_list:194 if job.job_name.find("stage1") > 0:195 stage1_jobs.append(job)196 elif job.job_name.find("stage2") > 0:197 stage2_jobs.append(job)198 elif job.job_name.find("stage3") > 0:199 stage3_jobs.append(job)200 201 if len(stage1_jobs) > 0:202 aggregate = _construct_aggregate(203 "github_libcxx_premerge_checks_stage1_aggregate", stage1_jobs204 )205 workflow_metrics.append(aggregate)206 if len(stage2_jobs) > 0:207 aggregate = _construct_aggregate(208 "github_libcxx_premerge_checks_stage2_aggregate", stage2_jobs209 )210 workflow_metrics.append(aggregate)211 if len(stage3_jobs) > 0:212 aggregate = _construct_aggregate(213 "github_libcxx_premerge_checks_stage3_aggregate", stage3_jobs214 )215 workflow_metrics.append(aggregate)216 217 218def clean_up_libcxx_job_name(old_name: str) -> str:219 """Convert libcxx job names to generically legal strings.220 221 Args:222 old_name: A string with the full name of the libc++ test that was run.223 224 Returns:225 Returns the input string with characters that might not be acceptable226 in some indentifier strings replaced with safer characters.227 228 Take a name like 'stage1 (generic-cxx03, clang-22, clang++-22)'229 and convert it to 'stage1_generic_cxx03__clang_22__clangxx_22'.230 (Remove parentheses; replace commas, hyphens and spaces with231 underscores; replace '+' with 'x'.)232 """233 # Names should have exactly one set of parentheses, so break on that. If234 # they don't have any parentheses, then don't update them at all.235 if old_name.find("(") == -1:236 return old_name237 stage, remainder = old_name.split("(")238 stage = stage.strip()239 if remainder[-1] == ")":240 remainder = remainder[:-1]241 remainder = remainder.replace("-", "_")242 remainder = remainder.replace(",", "_")243 remainder = remainder.replace(" ", "_")244 remainder = remainder.replace("+", "x")245 new_name = stage + "_" + remainder246 return new_name247 248 249def github_get_metrics(250 github_repo: github.Repository, last_workflows_seen_as_completed: set[int]251) -> tuple[list[JobMetrics], int]:252 """Gets the metrics for specified Github workflows.253 254 This function takes in a list of workflows to track, and optionally the255 workflow ID of the last tracked invocation. It grabs the relevant data256 from Github, returning it to the caller.257 If the last_seen_workflow parameter is None, this returns no metrics, but258 returns the id of the most recent workflow.259 260 Args:261 github_repo: A github repo object to use to query the relevant information.262 last_seen_workflow: the last workflow this function processed.263 264 Returns:265 Returns a tuple with 2 elements:266 - a list of JobMetrics objects, one per processed job.267 - the ID of the most recent processed workflow run.268 """269 workflow_metrics = []270 queued_count = collections.Counter()271 running_count = collections.Counter()272 273 # Initialize all the counters to 0 so we report 0 when no job is queued274 # or running.275 for wf_name, wf_metric_name in GITHUB_WORKFLOW_TO_TRACK.items():276 for job_name, job_metric_name in GITHUB_JOB_TO_TRACK[wf_metric_name].items():277 queued_count[wf_metric_name + "_" + job_metric_name] = 0278 running_count[wf_metric_name + "_" + job_metric_name] = 0279 280 # The list of workflows this iteration will process.281 # MaxSize = GITHUB_WORKFLOWS_MAX_PROCESS_COUNT282 workflow_seen_as_completed = set()283 284 # Since we process a fixed count of workflows, we want to know when285 # the depth is too small and if we miss workflows.286 # E.g.: is there was more than N workflows int last 2 hours.287 # To monitor this, we'll log the age of the oldest workflow processed,288 # and setup alterting in Grafana to help us adjust this depth.289 oldest_seen_workflow_age_mn = None290 291 # Do not apply any filters to this query.292 # See https://github.com/orgs/community/discussions/86766293 # Applying filters like `status=completed` will break pagination, and294 # return a non-sorted and incomplete list of workflows.295 i = 0296 for task in iter(github_repo.get_workflow_runs()):297 # Max depth reached, stopping.298 if i >= GITHUB_WORKFLOWS_MAX_PROCESS_COUNT:299 break300 i += 1301 302 workflow_age_mn = (303 datetime.datetime.now(datetime.timezone.utc) - task.created_at304 ).total_seconds() / 60305 oldest_seen_workflow_age_mn = workflow_age_mn306 # If we reach a workflow older than X, stop.307 if workflow_age_mn > GITHUB_WORKFLOW_MAX_CREATED_AGE_HOURS * 60:308 break309 310 # This workflow is not interesting to us.311 if task.name not in GITHUB_WORKFLOW_TO_TRACK:312 continue313 314 libcxx_testing = False315 if task.name == "Build and Test libc++":316 libcxx_testing = True317 318 if task.status == "completed":319 workflow_seen_as_completed.add(task.id)320 321 # This workflow has already been seen completed in the previous run.322 if task.id in last_workflows_seen_as_completed:323 continue324 325 name_prefix = GITHUB_WORKFLOW_TO_TRACK[task.name]326 for job in task.jobs():327 if libcxx_testing:328 # We're not running macos or windows libc++ tests on our329 # infrastructure.330 if job.name.find("macos") != -1 or job.name.find("windows") != -1:331 continue332 # This job is not interesting to us.333 elif job.name not in GITHUB_JOB_TO_TRACK[name_prefix]:334 continue335 336 if libcxx_testing:337 name_suffix = clean_up_libcxx_job_name(job.name)338 else:339 name_suffix = GITHUB_JOB_TO_TRACK[name_prefix][job.name]340 metric_name = name_prefix + "_" + name_suffix341 342 ag_metric_name = None343 if libcxx_testing:344 job_key = None345 if job.name.find("stage1") != -1:346 job_key = "stage1"347 elif job.name.find("stage2") != -1:348 job_key = "stage2"349 elif job.name.find("stage3") != -1:350 job_key = "stage3"351 if job_key:352 ag_name = (353 name_prefix + "_" + GITHUB_JOB_TO_TRACK[name_prefix][job_key]354 )355 356 if task.status != "completed":357 if job.status == "queued":358 queued_count[metric_name] += 1359 if libcxx_testing:360 queued_count[ag_name] += 1361 elif job.status == "in_progress":362 running_count[metric_name] += 1363 if libcxx_testing:364 running_count[ag_name] += 1365 continue366 367 job_result = int(job.conclusion == "success" or job.conclusion == "skipped")368 369 created_at = job.created_at370 started_at = job.started_at371 completed_at = job.completed_at372 373 if completed_at is None:374 logging.info(375 f"Workflow {task.id} is marked completed but has a job without a "376 "completion timestamp."377 )378 continue379 380 # GitHub API can return results where the started_at is slightly381 # later then the created_at (or completed earlier than started).382 # This would cause a -23h59mn delta, which will show up as +24h383 # queue/run time on grafana.384 if started_at < created_at:385 logging.info(386 "Workflow {} started before being created.".format(task.id)387 )388 queue_time = datetime.timedelta(seconds=0)389 else:390 queue_time = started_at - created_at391 if completed_at < started_at:392 logging.info("Workflow {} finished before starting.".format(task.id))393 run_time = datetime.timedelta(seconds=0)394 else:395 run_time = completed_at - started_at396 397 if run_time.seconds == 0:398 continue399 400 # Grafana will refuse to ingest metrics older than ~2 hours, so we401 # should avoid sending historical data.402 metric_age_mn = (403 datetime.datetime.now(datetime.timezone.utc) - completed_at404 ).total_seconds() / 60405 if metric_age_mn > GRAFANA_METRIC_MAX_AGE_MN:406 logging.warning(407 f"Job {job.id} from workflow {task.id} dropped due"408 + f" to staleness: {metric_age_mn}mn old."409 )410 continue411 412 logging.info(f"Adding a job metric for job {job.id} in workflow {task.id}")413 # The completed_at_ns timestamp associated with the event is414 # expected by Grafana to be in nanoseconds. Because we do math using415 # all three times (when creating libc++ aggregates), we need them416 # all to be in nanoseconds, even though created_at and started_at417 # are not returned to Grafana.418 created_at_ns = int(created_at.timestamp()) * 10**9419 started_at_ns = int(started_at.timestamp()) * 10**9420 completed_at_ns = int(completed_at.timestamp()) * 10**9421 workflow_metrics.append(422 JobMetrics(423 metric_name,424 queue_time.seconds,425 run_time.seconds,426 job_result,427 created_at_ns,428 started_at_ns,429 completed_at_ns,430 task.id,431 task.name,432 )433 )434 435 # Finished collecting the JobMetrics for all jobs; now create the436 # aggregates for any libc++ jobs.437 create_and_append_libcxx_aggregates(workflow_metrics)438 439 for name, value in queued_count.items():440 workflow_metrics.append(441 GaugeMetric(f"workflow_queue_size_{name}", value, time.time_ns())442 )443 for name, value in running_count.items():444 workflow_metrics.append(445 GaugeMetric(f"running_workflow_count_{name}", value, time.time_ns())446 )447 448 # Always send a hearbeat metric so we can monitor is this container is still able to log to Grafana.449 workflow_metrics.append(450 GaugeMetric("metrics_container_heartbeat", 1, time.time_ns())451 )452 453 # Log the oldest workflow we saw, allowing us to monitor if the processing454 # depth is correctly set-up.455 if oldest_seen_workflow_age_mn is not None:456 workflow_metrics.append(457 GaugeMetric(458 "github_oldest_processed_workflow_mn",459 oldest_seen_workflow_age_mn,460 time.time_ns(),461 )462 )463 return workflow_metrics, workflow_seen_as_completed464 465 466def upload_metrics(workflow_metrics, metrics_userid, api_key):467 """Upload metrics to Grafana.468 469 Takes in a list of workflow metrics and then uploads them to Grafana470 through a REST request.471 472 Args:473 workflow_metrics: A list of metrics to upload to Grafana.474 metrics_userid: The userid to use for the upload.475 api_key: The API key to use for the upload.476 """477 478 if len(workflow_metrics) == 0:479 logging.info("No metrics found to upload.")480 return481 482 metrics_batch = []483 for workflow_metric in workflow_metrics:484 if isinstance(workflow_metric, GaugeMetric):485 name = workflow_metric.name.lower().replace(" ", "_")486 metrics_batch.append(487 f"{name} value={workflow_metric.value} {workflow_metric.time_ns}"488 )489 elif isinstance(workflow_metric, JobMetrics):490 name = workflow_metric.job_name.lower().replace(" ", "_")491 metrics_batch.append(492 f"{name} queue_time={workflow_metric.queue_time},run_time={workflow_metric.run_time},status={workflow_metric.status} {workflow_metric.completed_at_ns}"493 )494 elif isinstance(workflow_metric, AggregateMetric):495 name = workflow_metric.aggregate_name.lower().replace(" ", "_")496 metrics_batch.append(497 f"{name} queue_time={workflow_metric.aggregate_queue_time},run_time={workflow_metric.aggregate_run_time},status={workflow_metric.aggregate_status} {workflow_metric.completed_at_ns}"498 )499 else:500 raise ValueError(501 f"Unsupported object type {type(workflow_metric)}: {str(workflow_metric)}"502 )503 504 request_data = "\n".join(metrics_batch)505 response = requests.post(506 GRAFANA_URL,507 headers={"Content-Type": "text/plain"},508 data=request_data,509 auth=(metrics_userid, api_key),510 )511 512 if response.status_code < 200 or response.status_code >= 300:513 logging.info(f"Failed to submit data to Grafana: {response.status_code}")514 515 516def main():517 # Authenticate with Github518 github_auth = Auth.Token(os.environ["GITHUB_TOKEN"])519 grafana_api_key = os.environ["GRAFANA_API_KEY"]520 grafana_metrics_userid = os.environ["GRAFANA_METRICS_USERID"]521 522 # The last workflow this script processed.523 # Because the Github queries are broken, we'll simply log a 'processed'524 # bit for the last COUNT_TO_PROCESS workflows.525 gh_last_workflows_seen_as_completed = set()526 527 # Enter the main loop. Every five minutes we wake up and dump metrics for528 # the relevant jobs.529 while True:530 github_object = Github(auth=github_auth)531 github_repo = github_object.get_repo("llvm/llvm-project")532 533 gh_metrics, gh_last_workflows_seen_as_completed = github_get_metrics(534 github_repo, gh_last_workflows_seen_as_completed535 )536 537 upload_metrics(gh_metrics, grafana_metrics_userid, grafana_api_key)538 logging.info(f"Uploaded {len(gh_metrics)} metrics")539 540 time.sleep(SCRAPE_INTERVAL_SECONDS)541 542 543if __name__ == "__main__":544 logging.basicConfig(level=logging.INFO)545 main()546