Provider: Rollbar

[talboren]

trafficstars

As suggested by @nicokruger

[talboren]

/bounty 50

[algora-pbc[bot]]

💎 $50 bounty • Keep (YC W23)

Steps to solve:

1. Start working: Comment /attempt #133 with your implementation plan
2. Submit work: Create a pull request including /claim #133 in the PR body to claim the bounty
3. Receive payment: 100% of the bounty is received 2-5 days post-reward. Make sure you are eligible for payouts

Thank you for contributing to keephq/keep!

Add a bounty • Share on socials

Attempt	Started (GMT+0)	Solution
🟢 @35C4n0r	May 13, 2024, 5:42:36 PM	WIP
🟢 @Bhavyajain21	May 13, 2024, 5:42:43 PM	WIP
🔴 @ezhil56x	May 13, 2024, 5:43:10 PM	#1177
🔴 @osmandemiroz	May 14, 2024, 1:33:46 PM	WIP

[ezhil56x]

Can I get this assigned? /attempt #133

[35C4n0r]

/attempt #133

Algora profile	Completed bounties	Tech	Active attempts	Options
@35C4n0r	10 keephq bounties + 7 bounties from 3 projects	Python, HTML, JavaScript & more		Cancel attempt

[Bhavyajain21]

/attempt

Algora profile	Completed bounties	Tech	Active attempts	Options
@Bhavyajain21	8 bounties from 3 projects	TypeScript, Rust, JavaScript & more		Cancel attempt

[algora-pbc[bot]]

[!NOTE] The user @35C4n0r is already attempting to complete issue #133 and claim the bounty. We recommend checking in on @35C4n0r's progress, and potentially collaborating, before starting a new solution.

[35C4n0r]

@talboren can you kindly assign the issue to me.

[ezhil56x]

/attempt #133 @algora-pbc I have mistakenly cancelled my attempt

[osmandemiroz]

/attempt #133

Options

• Cancel my attempt

[shahargl]

Assigned @ezhil56x

[35C4n0r]

Assigned @ezhil56x

@shahargl I think he has cancelled his attempt.

[ezhil56x]

/attempt #133 @algora-pbc I have mistakenly cancelled my attempt

@35C4n0r I'm still on it. Cancelled my mistake

[35C4n0r]

@ezhil56x Gotit. Feel free to use this, scope validation is left and maybe a bit of matching b/w payload and AlertDTO

"""
RollbarProvider is a class that allows to install webhooks and get alerts in Rollbar.
"""

import dataclasses
import datetime
from typing import List, Optional
from urllib.parse import urlencode, urljoin

import pydantic
import requests
from requests import HTTPError

from keep.api.models.alert import AlertDto, AlertSeverity
from keep.contextmanager.contextmanager import ContextManager
from keep.providers.base.base_provider import BaseProvider
from keep.providers.models.provider_config import ProviderConfig, ProviderScope


class ResourceAlreadyExists(Exception):
    def __init__(self, *args):
        super().__init__(*args)


@pydantic.dataclasses.dataclass
class RollbarProviderAuthConfig:
    """
    Rollbar authentication configuration.
    """
    rollbarAccessToken: str = dataclasses.field(
        metadata={
            "required": True,
            "description": "rollbarAccessToken",
            "hint": "project access token for rollbar API auth",
            "sensitive": True,
        },
    )


class RollbarProvider(BaseProvider):
    """Install Webhooks and receive alerts from Rollbar."""

    PROVIDER_SCOPES = [
        ProviderScope(
            name="authenticated",
            description="User is Authenticated",
            mandatory=True,
            mandatory_for_webhook=True,
            alias="Rules Reader",
        ),
    ]

    SEVERITIES_MAP = {
        "warning": AlertSeverity.WARNING,
        "error": AlertSeverity.HIGH,
        "info": AlertSeverity.INFO,
        "critical": AlertSeverity.CRITICAL,
        "debug": AlertSeverity.LOW
    }

    def __init__(
            self, context_manager: ContextManager, provider_id: str, config: ProviderConfig
    ):
        super().__init__(context_manager, provider_id, config)

    def dispose(self):
        """
        Dispose the provider.
        """
        pass

    def validate_config(self):
        """
        Validates required configuration for Rollbar provider.
        """
        self.authentication_config = RollbarProviderAuthConfig(
            **self.config.authentication
        )

    def __get_url(self, paths: List[str] = [], query_params: dict = None, **kwargs):
        """
        Helper method to build the url for Rollbar api requests.

        Example:

        paths = ["issue", "createmeta"]
        query_params = {"projectKeys": "key1"}
        url = __get_url("test", paths, query_params)

        # url = https://api.rollbar.com/api/1/issues/creatmeta?
        """

        url = urljoin(
            f"https://api.rollbar.com/api/1/",
            "/".join(str(path) for path in paths),
        )

        # add query params
        if query_params:
            url = f"{url}?{urlencode(query_params)}"

        return url

    def __get_headers(self):
        """
        Preparing the headers for Rollbar API requests
        """
        return {
            "X-Rollbar-Access-Token": self.authentication_config.rollbarAccessToken,
            "accept": "application/json; charset=utf-8",
            "content-type": "application/json"
        }

    def validate_scopes(self) -> dict[str, bool | str]:
        try:
            response = requests.get(self.__get_url(path="notifications/webhook/rules"), headers=self.__get_headers())

        except HTTPError as err:
            ...

    def setup_webhook(
            self, tenant_id: str, keep_api_url: str, api_key: str, setup_alerts: bool = True
    ):
        webhook_data = [
            {
                "trigger": "occurrence",
                "config": {
                    "format": "JSON",
                    "url": "	https://webhook.site/d65846b8-96d0-4c88-b047-846658e9eda0"
                }
            }
        ]
        response = requests.post(self.__get_url(paths=["notifications/webhook/rules"]), json=webhook_data,
                                 headers=self.__get_headers())
        if not response.ok:
            response_json = response.json()
            self.logger.error("Error while creating webhook", extra=response_json)
            raise Exception(response_json['message'])
        else:
            self.logger.info("Webhook created successfully")

    @staticmethod
    def _format_alert(
            event: dict,
            provider_instance: Optional["RollbarProvider"] = None,
    ) -> AlertDto:
        item_data = event['data']['item']
        occurrence_data = event['data']['occurrence']
        return AlertDto(
            id=str(item_data['id']),
            name=event['event_name'],
            severity=RollbarProvider.SEVERITIES_MAP[occurrence_data["level"]],
            lastReceived=datetime.datetime.fromtimestamp(item_data['last_occurrence_timestamp']).isoformat(),
            environment=item_data['environment'],
            service='Rollbar',
            source=[occurrence_data['framework']],
            url=event['data']['url'],
            message=occurrence_data['body']['message']['body'],
            description=item_data['title'],
            event_id=str(occurrence_data['uuid']),
            labels={'level': item_data['level']},
            fingerprint=item_data['hash'],
        )

[algora-pbc[bot]]

💡 @ezhil56x submitted a pull request that claims the bounty. You can visit your bounty board to reward.

[algora-pbc[bot]]

🎉🎈 @ezhil56x has been awarded $50! 🎈🎊

• Share on socials
• Give feedback