Rollbar forward to

README.md

Code snippet based on a Django project for routing to a different Rollbar project based on a file path.

The squad ownership is based on a CODEOWNERS file and is parsed using this Python library https://github.com/sbdchd/codeowners

CODEOWNERS

/basedir/documents/ @myorg/squad1
/basedir/documents/otherplace @myorg/squad1
/basedir/ui/ @myorg/squad3
/basedir/preferences/ @myorg/squad2

settings.py

def initialize_codeowners() -> CodeOwners:
    with open(os.path.join(BASE_DIR, os.pardir, "CODEOWNERS")) as codeowners_file:
        return CodeOwners(codeowners_file.read())


CODEOWNERS = initialize_codeowners()


SQUAD_ROLLBAR_ACCESS_TOKENS = {
    "@myorg/squad1": env.str("SQUAD1_ROLLBAR_ACCESS_TOKEN", None),
    "@myorg/squad2": env.str("SQUAD2_ROLLBAR_ACCESS_TOKEN", None),
    "@myorg/squad3": env.str("SQUAD3_ROLLBAR_ACCESS_TOKEN", None),
}

import rollbar

rollbar_orig_send_payload = rollbar.send_payload


def wrapped_rollbar_send_payload(payload: Dict[str, Any], access_token: str) -> None:
    """
    Wraps the Rollbar payload send to modify the access token away from the default, if applicable.
    If the given 'payload' being sent to Rollbar is an exception that has a file name, we can be
    more specific in the Rollbar project that we send this to

    :param payload: the entire payload being sent to Rollbar
    :param access_token: the default Rollbar access token when Rollbar was initially configured
    """

    if env.bool("ROLLBAR_SQUAD_ROUTING_DISABLED", False):
        return rollbar_orig_send_payload(payload, access_token)

    # Frames are the individual stack frames in reverse order from where an exception occurred. These
    # are already reverse order meaning the first item in the list is the top-level of the stack trace (so
    # somewhere in Django) and the last item in the list is where the exception was raised
    exception_stackframes: Optional[List] = (
        payload.get("data", {}).get("body", {}).get("trace", {}).get("frames")
    )

    # This pathname is used for logger.error records and identifies exactly where the logger.error
    # occurred
    logger_call_pathname: Optional[str] = (
        payload.get("data", {}).get("body", {}).get("message", {}).get("record", {}).get("pathname")
    )

    filepaths: Optional[List[str]] = None
    if exception_stackframes:
        filepaths = [frame["filename"] for frame in exception_stackframes]
    elif logger_call_pathname:
        filepaths = [logger_call_pathname]

    if filepaths:
        from django.conf import settings

        specific_owners = None
        for path in filepaths:
            if settings.BASE_DIR not in path:
                # only care about files in this repo
                continue
            # ends up transforming /path/to/basedir -> basedir
            code_start_path = pathlib.PurePath(settings.BASE_DIR).name
            relative_repo_path = f"{code_start_path}{path.replace(settings.BASE_DIR, '')}"
            specific_owners = [
                defined_owner[1] for defined_owner in settings.CODEOWNERS.of(relative_repo_path)
            ]

        sent_squad_specific_rollbar = False
        if specific_owners:

            logger.debug(f"Sending Rollbar to {specific_owners}")
            owner_access_tokens = [
                settings.SQUAD_ROLLBAR_ACCESS_TOKENS.get(owner)
                for owner in specific_owners
                if settings.SQUAD_ROLLBAR_ACCESS_TOKENS.get(owner)
            ]
            if owner_access_tokens:
                for token in owner_access_tokens:
                    rollbar_orig_send_payload(payload, token)
                    sent_squad_specific_rollbar = True

        if not sent_squad_specific_rollbar:
            rollbar_orig_send_payload(payload, access_token)

    else:
        rollbar_orig_send_payload(payload, access_token)


rollbar.send_payload = wrapped_rollbar_send_payload