collect_otel_contribution_metrics.py

"""
Collects otel contribution metrics from GitHub and exports them to Grafana Cloud.

This script mirrors the functionality of the otel-contributions repository's main.py,
fetching commit and line-change statistics for user contributions across all repositories
in the open-telemetry GitHub organization, then sending those metrics via OpenTelemetry.
"""

import os
import time
import requests
from datetime import datetime

from opentelemetry import metrics
from opentelemetry.sdk.metrics import MeterProvider, Counter
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, AggregationTemporality
from opentelemetry.exporter.otlp.proto.http.metric_exporter import OTLPMetricExporter
from opentelemetry.sdk.resources import Resource


USERNAME = "jaydeluca"
ORG_NAME = "open-telemetry"

resource = Resource.create({"service.name": "otel-contribution-metrics"})

# OTLP exporter with cumulative temporality for Prometheus/Mimir in Grafana
otlp_exporter = OTLPMetricExporter(
    preferred_temporality={
        Counter: AggregationTemporality.CUMULATIVE,
    }
)
otlp_reader = PeriodicExportingMetricReader(otlp_exporter, export_interval_millis=5000)

provider = MeterProvider(
    resource=resource,
    metric_readers=[otlp_reader]
)
metrics.set_meter_provider(provider)

meter = metrics.get_meter("otel.contribution.metrics.meter")


def github_get(url, headers, **kwargs):
    """Make a GitHub API GET request, waiting automatically when rate limits are hit."""
    while True:
        response = requests.get(url, headers=headers, **kwargs)

        if response.status_code not in (403, 429):
            return response

        retry_after = response.headers.get('retry-after')
        reset_time = response.headers.get('x-ratelimit-reset')
        remaining = response.headers.get('x-ratelimit-remaining')

        # Secondary rate limit: retry-after header is present
        if retry_after:
            wait_seconds = int(retry_after) + 1
            resume_at = datetime.fromtimestamp(time.time() + wait_seconds).strftime('%H:%M:%S')
            print(f"\n[Rate Limited] Secondary rate limit hit. Waiting {wait_seconds}s (resumes ~{resume_at})...")
            time.sleep(wait_seconds)
            continue

        # Primary rate limit: quota exhausted
        if reset_time and remaining == '0':
            wait_seconds = max(1, int(reset_time) - int(time.time()) + 1)
            resume_at = datetime.fromtimestamp(int(reset_time)).strftime('%H:%M:%S')
            print(f"\n[Rate Limited] Primary rate limit hit. Waiting {wait_seconds}s until {resume_at}...")
            time.sleep(wait_seconds)
            continue

        # Non-rate-limit 403 — return as-is so callers can handle it
        return response


def fetch_repos(headers):
    """Fetch all repositories in the open-telemetry organization."""
    repos = []
    page = 1
    while True:
        url = f"https://api.github.com/orgs/{ORG_NAME}/repos?sort=created&direction=desc&page={page}&per_page=100"
        response = github_get(url, headers=headers, timeout=30)
        new_repos = response.json()
        if not isinstance(new_repos, list) or not new_repos:
            break
        repos.extend(new_repos)
        if len(new_repos) < 100:
            break
        page += 1
    return repos


def fetch_commits_for_repo(repo_name, headers):
    """Fetch all commits by USERNAME in a given repository."""
    all_commits = []
    page = 1
    while True:
        url = (
            f"https://api.github.com/repos/{ORG_NAME}/{repo_name}/commits"
            f"?author={USERNAME}&page={page}&per_page=100"
        )
        response = github_get(url, headers=headers, timeout=30)
        commit_data = response.json()
        if not isinstance(commit_data, list) or not commit_data:
            break
        all_commits.extend(commit_data)
        page += 1
    return all_commits


def fetch_commit_stats(commit_url, headers):
    """Fetch additions and deletions for a single commit."""
    response = github_get(commit_url, headers=headers, timeout=30)
    details = response.json()
    stats = details.get('stats', {})
    return stats.get('additions', 0), stats.get('deletions', 0)


def collect_contribution_metrics(headers):
    """
    Collect contribution metrics for USERNAME across all open-telemetry repositories,
    then export them via OpenTelemetry gauges.
    """
    print(f"Fetching repositories for organization: {ORG_NAME}...")
    repos = fetch_repos(headers)
    print(f"  Found {len(repos)} repositories")

    commits_gauge = meter.create_gauge(
        "contribution.commits.total",
        description="Total number of commits by contributor per repository",
        unit="1",
    )
    additions_gauge = meter.create_gauge(
        "contribution.additions.total",
        description="Total lines added by contributor per repository",
        unit="1",
    )
    deletions_gauge = meter.create_gauge(
        "contribution.deletions.total",
        description="Total lines deleted by contributor per repository",
        unit="1",
    )

    total_commits = 0
    total_additions = 0
    total_deletions = 0

    for repo in repos:
        repo_name = repo['name']

        commits = fetch_commits_for_repo(repo_name, headers)
        if not commits:
            continue

        local_commits = 0
        local_additions = 0
        local_deletions = 0

        for commit in commits:
            additions, deletions = fetch_commit_stats(commit['url'], headers)
            local_commits += 1
            local_additions += additions
            local_deletions += deletions

        total_commits += local_commits
        total_additions += local_additions
        total_deletions += local_deletions

        attributes = {"repo": repo_name, "org": ORG_NAME, "contributor": USERNAME}
        commits_gauge.set(local_commits, attributes)
        additions_gauge.set(local_additions, attributes)
        deletions_gauge.set(local_deletions, attributes)

        print(
            f"  {repo_name}: commits={local_commits}, "
            f"additions={local_additions}, deletions={local_deletions}"
        )

    # Export aggregate totals (no repo attribute)
    total_attributes = {"org": ORG_NAME, "contributor": USERNAME}
    commits_gauge.set(total_commits, {**total_attributes, "repo": "_all"})
    additions_gauge.set(total_additions, {**total_attributes, "repo": "_all"})
    deletions_gauge.set(total_deletions, {**total_attributes, "repo": "_all"})

    print(f"\nSummary for {USERNAME} in {ORG_NAME}:")
    print(f"  Total commits:        {total_commits}")
    print(f"  Total lines added:    {total_additions}")
    print(f"  Total lines deleted:  {total_deletions}")


if __name__ == "__main__":
    github_token = os.environ.get("GITHUB_TOKEN")
    if not github_token:
        raise ValueError("GITHUB_TOKEN environment variable not set.")

    auth_headers = {
        "Authorization": f"token {github_token}",
        "Accept": "application/vnd.github.v3+json",
    }

    print("=" * 60)
    print("OTel Contribution Metrics Collection")
    print("=" * 60)

    collect_contribution_metrics(auth_headers)

    print("\nAll contribution metrics collected. Flushing metrics before exit...")
    provider.force_flush()

    print("Metrics flushed. The script will exit after 5 seconds.")
    time.sleep(5)