src/sentry/sdk_updates.py

from __future__ import annotations

import logging

from django.conf import settings
from django.core.cache import cache
from packaging.version import Version
from typing_extensions import TypedDict

from sentry.tasks.release_registry import SDK_INDEX_CACHE_KEY
from sentry.utils.safe import get_path

logger = logging.getLogger(__name__)


class SdkSetupState:
    def __init__(self, sdk_name, sdk_version, modules, integrations):
        self.sdk_name = sdk_name
        self.sdk_version = sdk_version
        self.modules = dict(modules or ())
        self.integrations = list(integrations or ())

    def copy(self):
        return type(self)(
            sdk_name=self.sdk_name,
            sdk_version=self.sdk_version,
            modules=self.modules,
            integrations=self.integrations,
        )

    @classmethod
    def from_event_json(cls, event_data):
        sdk_name = get_path(event_data, "sdk", "name")
        if sdk_name:
            sdk_name = sdk_name.lower().rsplit(":", 1)[0]

        if sdk_name == "sentry-python":
            sdk_name = "sentry.python"

        return cls(
            sdk_name=sdk_name,
            sdk_version=get_path(event_data, "sdk", "version"),
            modules=get_path(event_data, "modules"),
            integrations=get_path(event_data, "sdk", "integrations"),
        )


class SdkIndexState:
    def __init__(self, sdk_versions=None, deprecated_sdks=None, sdk_supported_modules=None):
        self.sdk_versions = sdk_versions or get_sdk_versions()
        self.deprecated_sdks = deprecated_sdks or settings.DEPRECATED_SDKS
        self.sdk_supported_modules = sdk_supported_modules or SDK_SUPPORTED_MODULES


class Suggestion:
    def to_json(self):
        raise NotImplementedError()

    def __eq__(self, other):
        return self.to_json() == other.to_json()


class EnableIntegrationSuggestion(Suggestion):
    def __init__(self, integration_name, integration_url):
        self.integration_name = integration_name
        self.integration_url = integration_url

    def to_json(self):
        return {
            "type": "enableIntegration",
            "integrationName": self.integration_name,
            "integrationUrl": self.integration_url,
        }

    def get_new_state(self, old_state):
        if self.integration_name in old_state.integrations:
            return old_state

        new_state = old_state.copy()
        new_state.integrations.append(self.integration_name)
        return new_state


class UpdateSDKSuggestion(Suggestion):
    def __init__(self, sdk_name, new_sdk_version, ignore_patch_version):
        self.sdk_name = sdk_name
        self.new_sdk_version = new_sdk_version
        self.ignore_patch_version = ignore_patch_version

    def to_json(self):
        return {
            "type": "updateSdk",
            "sdkName": self.sdk_name,
            "newSdkVersion": self.new_sdk_version,
            "sdkUrl": get_sdk_urls().get(self.sdk_name),
        }

    def get_new_state(self, old_state):
        if self.new_sdk_version is None:
            return old_state

        new_sdk_version = self.new_sdk_version
        if self.ignore_patch_version:
            new_sdk_version = ".".join(v for v in new_sdk_version.split(".")[:2])

        try:
            has_newer_version = Version(old_state.sdk_version) < Version(new_sdk_version)
        except Exception:
            has_newer_version = False

        if not has_newer_version:
            return old_state

        new_state = old_state.copy()
        new_state.sdk_version = self.new_sdk_version
        return new_state


class ChangeSDKSuggestion(Suggestion):
    """
    :param module_names: Hide this suggestion if any of the given modules is
        loaded. This list is used to weed out invalid suggestions when using
        multiple SDKs in e.g. .NET.
    """

    def __init__(self, new_sdk_name, module_names=None):
        self.new_sdk_name = new_sdk_name
        self.module_names = module_names

    def to_json(self):
        return {
            "type": "changeSdk",
            "newSdkName": self.new_sdk_name,
            "sdkUrl": get_sdk_urls().get(self.new_sdk_name),
        }

    def get_new_state(self, old_state):
        if old_state.sdk_name == self.new_sdk_name:
            return old_state

        if any(x in old_state.modules for x in self.module_names or ()):
            return old_state

        new_state = old_state.copy()
        new_state.sdk_name = self.new_sdk_name
        return new_state


class SupportedModule(TypedDict):
    sdk_name: str
    sdk_version_added: str
    module_name: str
    module_version_min: str
    suggestion: Suggestion


SDK_SUPPORTED_MODULES: list[SupportedModule] = [
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.3.2",
        "module_name": "django",
        "module_version_min": "1.6.0",
        "suggestion": EnableIntegrationSuggestion(
            "django", "https://docs.sentry.io/platforms/python/django/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.3.2",
        "module_name": "flask",
        "module_version_min": "0.11.0",
        "suggestion": EnableIntegrationSuggestion(
            "flask", "https://docs.sentry.io/platforms/python/flask/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "1.8.0",
        "module_name": "starlette",
        "module_version_min": "0.19.1",
        "suggestion": EnableIntegrationSuggestion(
            "starlette", "https://docs.sentry.io/platforms/python/starlette/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "1.8.0",
        "module_name": "fastapi",
        "module_version_min": "0.79.0",
        "suggestion": EnableIntegrationSuggestion(
            "fastapi", "https://docs.sentry.io/platforms/python/fastapi/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.7.9",
        "module_name": "bottle",
        "module_version_min": "0.12.0",
        "suggestion": EnableIntegrationSuggestion(
            "bottle", "https://docs.sentry.io/platforms/python/bottle/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.7.11",
        "module_name": "falcon",
        "module_version_min": "1.4.0",
        "suggestion": EnableIntegrationSuggestion(
            "falcon", "https://docs.sentry.io/platforms/python/falcon/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.3.6",
        "module_name": "sanic",
        "module_version_min": "0.8.0",
        "suggestion": EnableIntegrationSuggestion(
            "sanic", "https://docs.sentry.io/platforms/python/sanic/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.3.2",
        "module_name": "celery",
        "module_version_min": "3.0.0",
        "suggestion": EnableIntegrationSuggestion(
            "celery", "https://docs.sentry.io/platforms/python/celery/"
        ),
    },
    # TODO: Detect AWS Lambda for Python
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.5.0",
        "module_name": "pyramid",
        "module_version_min": "1.3.0",
        "suggestion": EnableIntegrationSuggestion(
            "pyramid", "https://docs.sentry.io/platforms/python/pyramid/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.5.1",
        "module_name": "rq",
        "module_version_min": "0.6",
        "suggestion": EnableIntegrationSuggestion(
            "rq", "https://docs.sentry.io/platforms/python/rq/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.6.1",
        "module_name": "aiohttp",
        "module_version_min": "3.4.0",
        "suggestion": EnableIntegrationSuggestion(
            "aiohttp", "https://docs.sentry.io/platforms/python/aiohttp/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.6.3",
        "module_name": "tornado",
        "module_version_min": "5.0.0",
        "suggestion": EnableIntegrationSuggestion(
            "tornado", "https://docs.sentry.io/platforms/python/tornado/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.10.0",
        "module_name": "redis",
        "module_version_min": "0.0.0",
        "suggestion": EnableIntegrationSuggestion(
            "redis", "https://docs.sentry.io/platforms/python/redis/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.11.0",
        "module_name": "sqlalchemy",
        "module_version_min": "1.2.0",
        "suggestion": EnableIntegrationSuggestion(
            "sqlalchemy", "https://docs.sentry.io/platforms/python/sqlalchemy/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.11.0",
        "module_name": "apache_beam",
        "module_version_min": "2.12.0",
        "suggestion": EnableIntegrationSuggestion(
            "beam", "https://docs.sentry.io/platforms/python/beam/"
        ),
    },
    {
        "sdk_name": "sentry.python",
        "sdk_version_added": "0.13.0",
        "module_name": "pyspark",
        "module_version_min": "2.0.0",
        "suggestion": EnableIntegrationSuggestion(
            "spark", "https://docs.sentry.io/platforms/python/pyspark/"
        ),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "Microsoft.AspNetCore.Hosting",
        "module_version_min": "2.1.0",
        "suggestion": ChangeSDKSuggestion("sentry.dotnet.aspnetcore", ["Sentry.AspNetCore"]),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "EntityFramework",
        "module_version_min": "6.0.0",
        "suggestion": ChangeSDKSuggestion(
            "sentry.dotnet.entityframework", ["Sentry.EntityFramework"]
        ),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "log4net",
        "module_version_min": "2.0.8",
        "suggestion": ChangeSDKSuggestion("sentry.dotnet.log4net", ["Sentry.Log4Net"]),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "Microsoft.Extensions.Logging.Configuration",
        "module_version_min": "2.1.0",
        "suggestion": ChangeSDKSuggestion(
            "sentry.dotnet.extensions.logging",
            [
                "Sentry.Extensions.Logging",
                # If AspNetCore is used, do not show this suggestion at all,
                # because the (hopefully visible) suggestion to use the
                # AspNetCore SDK is more specific.
                "Microsoft.AspNetCore.Hosting",
            ],
        ),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "Serilog",
        "module_version_min": "2.7.1",
        "suggestion": ChangeSDKSuggestion("sentry.dotnet.serilog", ["Sentry.Serilog"]),
    },
    {
        "sdk_name": "sentry.dotnet",
        "sdk_version_added": "0.0.0",
        "module_name": "NLog",
        "module_version_min": "4.6.0",
        "suggestion": ChangeSDKSuggestion("sentry.dotnet.nlog", ["Sentry.NLog"]),
    },
]


def get_sdk_index():
    """
    Get the SDK index from cache, if available.

    The cache is filled by a regular background task (see sentry/tasks/release_registry)
    """
    if not settings.SENTRY_RELEASE_REGISTRY_BASEURL:
        return {}

    return cache.get(SDK_INDEX_CACHE_KEY) or {}


def get_sdk_versions():
    try:
        rv = settings.SDK_VERSIONS
        rv.update((key, info["version"]) for (key, info) in get_sdk_index().items())
        return rv
    except Exception:
        logger.exception("sentry-release-registry.sdk-versions")
        return {}


def get_sdk_urls():
    try:
        rv = dict(settings.SDK_URLS)
        rv.update((key, info["main_docs_url"]) for (key, info) in get_sdk_index().items())
        return rv
    except Exception:
        logger.exception("sentry-release-registry.sdk-urls")
        return {}


def _get_suggested_updates_step(setup_state, index_state, ignore_patch_version):
    if not setup_state.sdk_name or not setup_state.sdk_version:
        return

    yield UpdateSDKSuggestion(
        setup_state.sdk_name,
        index_state.sdk_versions.get(setup_state.sdk_name),
        ignore_patch_version,
    )

    # If an SDK is both outdated and entirely deprecated, we want to inform
    # the user of both. It's unclear if they would want to upgrade the SDK
    # or migrate to the new one.
    newest_name = settings.DEPRECATED_SDKS.get(setup_state.sdk_name, setup_state.sdk_name)
    yield ChangeSDKSuggestion(newest_name)

    for support_info in SDK_SUPPORTED_MODULES:
        if support_info["sdk_name"] != setup_state.sdk_name and not setup_state.sdk_name.startswith(
            support_info["sdk_name"] + "."
        ):
            continue

        if support_info["module_name"] not in setup_state.modules:
            continue

        try:
            if Version(support_info["sdk_version_added"]) > Version(setup_state.sdk_version):
                continue
        except Exception:
            continue

        try:
            if Version(support_info["module_version_min"]) > Version(
                setup_state.modules[support_info["module_name"]]
            ):
                # TODO(markus): Maybe we want to suggest people to upgrade their module?
                #
                # E.g. "please upgrade Django so you can get the Django
                # integration"
                continue
        except Exception:
            continue

        yield support_info["suggestion"]


def get_suggested_updates(
    setup_state, index_state=None, parent_suggestions=None, ignore_patch_version=True
):
    if index_state is None:
        index_state = SdkIndexState()

    if parent_suggestions is None:
        parent_suggestions = []

    suggestions = list(_get_suggested_updates_step(setup_state, index_state, ignore_patch_version))

    rv = []
    new_setup_states = []

    for suggestion in suggestions:
        if suggestion in parent_suggestions:
            continue

        new_setup_state = suggestion.get_new_state(setup_state)
        if new_setup_state == setup_state:
            continue

        rv.append(suggestion)
        new_setup_states.append(new_setup_state)

    for new_setup_state, suggestion in zip(new_setup_states, rv):
        json = suggestion.to_json()
        json["enables"] = list(
            get_suggested_updates(
                new_setup_state, parent_suggestions=parent_suggestions + rv, index_state=index_state
            )
        )

        yield json