Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 73 additions & 14 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,12 @@

import jwt
import requests
import sentry_sdk
from google.api_core import exceptions as api_exceptions
from google.api_core.client_options import ClientOptions
from google.auth.credentials import AnonymousCredentials
from google.cloud import bigquery
from sentry_sdk.integrations.logging import LoggingIntegration

BUG_RE = re.compile(r"\b(?:bug|b=)\s*#?(\d+)\b", re.I)

Expand Down Expand Up @@ -362,20 +364,21 @@ def get_installation_access_token(
expected_status=201,
)

# Never include resp.text in these errors: the body may carry the
# token, and error messages are reported to Sentry.
try:
data = resp.json()
except requests.exceptions.JSONDecodeError as e:
raise RuntimeError(
f"Failed to parse access token response: {e}: {resp.text}"
)
raise RuntimeError(f"Failed to parse access token response: {e}")
try:
access_token = AccessToken(
token=data["token"],
expires_at=datetime.fromisoformat(data["expires_at"]),
)
except KeyError as e:
raise RuntimeError(
f"Unexpected access token response structure, missing key {e}: {resp.text}"
f"Unexpected access token response structure, missing key {e}; "
f"keys present: {sorted(data)}"
)
except ValueError as e:
raise RuntimeError(
Expand All @@ -398,6 +401,31 @@ def setup_logging() -> None:
)


def setup_sentry() -> bool:
"""
Report errors to Sentry when SENTRY_DSN is set; no-op otherwise.

Log records at INFO and above are attached as breadcrumbs only; events are
sent explicitly with sentry_sdk.capture_exception/capture_message so each
failure is reported exactly once (several paths log an error and then raise).
SENTRY_ENVIRONMENT and SENTRY_RELEASE are read from the environment by the SDK.

Returns:
True if Sentry was initialized, False if SENTRY_DSN is unset.
"""
dsn = os.environ.get("SENTRY_DSN")
if not dsn:
return False
sentry_sdk.init(
dsn=dsn,
integrations=[LoggingIntegration(level=logging.INFO, event_level=None)],
# Tracebacks pass through frames holding the GitHub private key, app JWT
# and installation tokens; never ship frame locals to Sentry.
include_local_variables=False,
Comment thread
dklawren marked this conversation as resolved.
)
return True


def _parse_github_timestamp(value: str | None) -> datetime | None:
"""
Parse a GitHub ISO-8601 timestamp (e.g. ``2026-06-17T14:53:58Z``) into an
Expand Down Expand Up @@ -1711,6 +1739,26 @@ def _flush_pending() -> None:
return processed


def _process_repo_in_sentry_scope(repo: str, *args, **kwargs) -> int:
"""
Run process_repo for *repo* inside its own Sentry isolation scope.

Runs in the worker thread so the retry/backoff warnings logged while
processing the repo are attached as breadcrumbs to the failure event (Sentry
scopes are per-thread, and pool threads are reused across repos). The event
is tagged with the repo for filtering. Exceptions are logged, reported, then
re-raised so the caller can record the repo as failed.
"""
with sentry_sdk.isolation_scope() as scope:
scope.set_tag("repo", repo)
try:
return process_repo(repo, *args, **kwargs)
except Exception as exc:
logger.exception(f"Failed to process repo {repo}: {exc}")
sentry_sdk.capture_exception(exc)
raise


def main() -> int:
"""
Main ETL process with chunked processing.
Expand All @@ -1722,14 +1770,23 @@ def main() -> int:
4. Repeat until no more data
"""
setup_logging()
sentry_enabled = setup_sentry()
try:
return _main()
return _main(sentry_enabled)
except RuntimeError as e:
logger.error(str(e))
logger.exception(str(e))
sentry_sdk.capture_exception(e)
return 1
except SystemExit as e:
# Missing configuration: report it, then still fail the run.
sentry_sdk.capture_message(str(e), level="fatal")
raise
finally:
# The process exits right after this; send queued events first.
sentry_sdk.flush()


def _main() -> int:
def _main(sentry_enabled: bool = False) -> int:
logger.info("Starting GitHub ETL process with chunked processing")

github_app_id = os.environ.get("GITHUB_APP_ID") or None
Expand Down Expand Up @@ -1810,7 +1867,7 @@ def _main() -> int:
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_repo = {
executor.submit(
process_repo,
_process_repo_in_sentry_scope,
repo,
github_app_id,
github_private_key,
Expand All @@ -1829,13 +1886,12 @@ def _main() -> int:
repo = future_to_repo[future]
try:
processed = future.result()
except Exception as exc:
except Exception:
# Catch broadly so one repo's failure (a TooManyRetriesError, a
# RuntimeError, or a bare Exception from load_data) is recorded as a
# failed repo rather than propagating out of the executor and
# discarding the results of other in-flight repos. logger.exception
# records the worker thread's traceback for debugging in CI/prod.
logger.exception(f"Failed to process repo {repo}: {exc}")
# discarding the results of other in-flight repos. The worker has
# already logged/reported it (see _process_repo_in_sentry_scope).
failed_repos.append(repo)
continue
total_processed += processed
Expand All @@ -1844,10 +1900,13 @@ def _main() -> int:
)

if failed_repos:
logger.error(
logger.warning(
f"ETL completed with failures. Failed repos: {', '.join(failed_repos)}"
)
return 1
# With Sentry configured, per-repo failures were reported there, so exit 0
# to avoid marking the cron job as failed. Without it, the exit code is the
# only signal, so fail the run.
return 0 if sentry_enabled else 1

logger.info(
f"GitHub ETL process completed successfully. Total PRs processed: {total_processed}"
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@ dependencies = [
"requests>=2.25.0",
"google-cloud-bigquery==3.25.0",
"PyJWT[crypto]>=2.0.0",
"sentry-sdk>=2.0.0",
]

[project.optional-dependencies]
Expand Down
12 changes: 10 additions & 2 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,9 @@
certifi==2026.1.4 \
--hash=sha256:9943707519e4add1115f44c2bc244f782c0249876bf51b6599fee1ffbedd685c \
--hash=sha256:ac726dd470482006e014ad384921ed6438c457018f4b3d204aea4281258b2120
# via requests
# via
# requests
# sentry-sdk
cffi==2.0.0 \
--hash=sha256:00bdf7acc5f795150faa6957054fbbca2439db2f775ce831222b66f192f03beb \
--hash=sha256:07b271772c100085dd28b74fa0cd81c8fb1a3ba18b21e03d7c27f3436a10606b \
Expand Down Expand Up @@ -458,6 +460,10 @@ rsa==4.9.1 \
--hash=sha256:68635866661c6836b8d39430f97a996acbd61bfa49406748ea243539fe239762 \
--hash=sha256:e7bdbfdb5497da4c07dfd35530e1a902659db6ff241e39d9953cad06ebd0ae75
# via google-auth
sentry-sdk==2.71.0 \
--hash=sha256:7beb27a22f06396f3a05510c3da6ea10e1c79e02e859993c27ce4ff074e296d6 \
--hash=sha256:9ed36b4c206048aa67b5ddba0a6e28d59fc55cf10d60f5f7aea2c3917184bf4a
# via github-etl (pyproject.toml)
six==1.17.0 \
--hash=sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274 \
--hash=sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81
Expand All @@ -469,4 +475,6 @@ typing-extensions==4.15.0 \
urllib3==2.6.3 \
--hash=sha256:1b62b6884944a57dbe321509ab94fd4d3b307075e0c2eae991ac71ee15ad38ed \
--hash=sha256:bf272323e553dfb2e87d9bfd225ca7b0f467b919d7bbd355436d3fd37cb0acd4
# via requests
# via
# requests
# sentry-sdk
138 changes: 136 additions & 2 deletions tests/test_main.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,11 @@
import logging
import os
import threading
import uuid
from unittest.mock import MagicMock, Mock, patch

import pytest
import sentry_sdk

import main

Expand Down Expand Up @@ -540,7 +543,7 @@ def extract_side_effect(*args, **kwargs):
):
result = main.main()

assert result == 1 # partial failure
assert result == 1 # no SENTRY_DSN, so the exit code must signal the failure
assert mock_extract.call_count == 2 # both repos were attempted
mock_load.assert_called_once() # only the successful repo loaded data

Expand All @@ -562,7 +565,7 @@ def test_bare_exception_on_one_repo_is_isolated(
"""A bare Exception (e.g. from load_data) on one repo must not abort the others.

The executor catches broadly, so the failing repo is recorded in failed_repos
(overall exit code 1) while the healthy repo still completes its load.
(exit code 1 without SENTRY_DSN) while the healthy repo still completes its load.
"""
# Fresh iterator per repo (a shared return_value iterator would be exhausted
# by whichever repo consumes it first).
Expand Down Expand Up @@ -602,6 +605,137 @@ def transform_side_effect(chunk, repo):
assert mock_load.call_count == 2 # both repos reached the load step


@patch("main.sentry_sdk")
@patch("main.setup_logging")
@patch("main.bigquery.Client")
@patch("requests.Session")
@patch("main.extract_pull_requests")
def test_repo_failure_is_tagged_for_sentry(
mock_extract,
mock_session_class,
mock_bq_client,
mock_setup_logging,
mock_sentry,
):
"""A failed repo is reported under a Sentry scope tagged with the repo name,
and the run exits 0 because Sentry is configured."""
mock_extract.side_effect = main.TooManyRetriesError("GitHub API error 502")
scope = mock_sentry.isolation_scope.return_value.__enter__.return_value

with patch.dict(
os.environ,
{
"SENTRY_DSN": "https://key@o0.ingest.sentry.io/0",
"GITHUB_REPOS": "mozilla/firefox",
"BIGQUERY_PROJECT": "test",
"BIGQUERY_DATASET": "test",
},
clear=True,
):
result = main.main()

assert result == 0
scope.set_tag.assert_called_once_with("repo", "mozilla/firefox")
# Reported exactly once, even though the failure is also logged.
mock_sentry.capture_exception.assert_called_once()
mock_sentry.flush.assert_called_once()


@patch("main.sentry_sdk")
@patch("main.setup_logging")
def test_missing_config_is_reported_and_still_fails(mock_setup_logging, mock_sentry):
"""Missing required env vars are sent to Sentry and still exit non-zero."""
with patch.dict(os.environ, {}, clear=True):
with pytest.raises(SystemExit):
main.main()

mock_sentry.capture_message.assert_called_once()
assert "BIGQUERY_PROJECT" in mock_sentry.capture_message.call_args.args[0]
mock_sentry.flush.assert_called_once()


@patch("main.sentry_sdk")
@patch("main.setup_logging")
@patch("main._main", side_effect=RuntimeError("token exchange failed"))
def test_runtime_error_still_fails(mock_main, mock_setup_logging, mock_sentry):
"""A top-level RuntimeError still returns a non-zero exit code."""
assert main.main() == 1
mock_sentry.capture_exception.assert_called_once()
mock_sentry.flush.assert_called_once()


class TestSetupSentry:
"""Tests for setup_sentry initialization."""

@patch("main.sentry_sdk.init")
def test_noop_without_dsn(self, mock_init):
with patch.dict(os.environ, {}, clear=True):
assert main.setup_sentry() is False
mock_init.assert_not_called()

@patch("main.sentry_sdk.init")
def test_initializes_with_dsn(self, mock_init):
dsn = "https://key@o0.ingest.sentry.io/0"
with patch.dict(os.environ, {"SENTRY_DSN": dsn}, clear=True):
assert main.setup_sentry() is True
mock_init.assert_called_once()
assert mock_init.call_args.kwargs["dsn"] == dsn

def test_does_not_send_local_variables(self):
"""Frame locals hold the private key, JWT and tokens. Never send them."""
# Generated at runtime so it can't leak via source context lines,
# only via frame locals.
secret = uuid.uuid4().hex
events = []

class CapturingTransport(sentry_sdk.transport.Transport):
def capture_envelope(self, envelope):
events.append(envelope.get_event())

real_init = sentry_sdk.init

def init_with_capturing_transport(*args, **kwargs):
return real_init(*args, transport=CapturingTransport(), **kwargs)

def fail_with_secret_in_scope():
private_key = secret # noqa: F841
raise RuntimeError("token exchange failed")

dsn = "https://key@o0.ingest.sentry.io/0"
try:
with (
patch.dict(os.environ, {"SENTRY_DSN": dsn}, clear=True),
patch(
"main.sentry_sdk.init", side_effect=init_with_capturing_transport
),
):
assert main.setup_sentry() is True
try:
fail_with_secret_in_scope()
except RuntimeError:
sentry_sdk.capture_exception()
finally:
sentry_sdk.get_client().close()
sentry_sdk.get_global_scope().set_client(None)

assert len(events) == 1
frames = events[0]["exception"]["values"][-1]["stacktrace"]["frames"]
assert any(f["function"] == "fail_with_secret_in_scope" for f in frames)
assert all("vars" not in f for f in frames)
assert secret not in repr(events[0])

@patch("main.LoggingIntegration")
@patch("main.sentry_sdk.init")
def test_logs_are_breadcrumbs_only(self, mock_init, mock_logging_integration):
"""Log records must not become events; failures are captured explicitly."""
dsn = "https://key@o0.ingest.sentry.io/0"
with patch.dict(os.environ, {"SENTRY_DSN": dsn}, clear=True):
main.setup_sentry()
mock_logging_integration.assert_called_once_with(
level=logging.INFO, event_level=None
)


class TestResolveMaxWorkers:
"""Tests for _resolve_max_workers worker-count resolution."""

Expand Down
Loading
Loading