Closed talboren closed 5 months ago
/bounty 50
/attempt #133
with your implementation plan/claim #133
in the PR body to claim the bountyThank 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 |
Can I get this assigned? /attempt #133
/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 |
/attempt
Algora profile | Completed bounties | Tech | Active attempts | Options |
---|---|---|---|---|
@Bhavyajain21 | 8 bounties from 3 projects | TypeScript, Rust, JavaScript & more |
Cancel attempt |
[!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.
@talboren can you kindly assign the issue to me.
/attempt #133 @algora-pbc I have mistakenly cancelled my attempt
/attempt #133
Assigned @ezhil56x
Assigned @ezhil56x
@shahargl I think he has cancelled his attempt.
/attempt #133 @algora-pbc I have mistakenly cancelled my attempt
@35C4n0r I'm still on it. Cancelled my mistake
@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'],
)
💡 @ezhil56x submitted a pull request that claims the bounty. You can visit your bounty board to reward.
🎉🎈 @ezhil56x has been awarded $50! 🎈🎊
As suggested by @nicokruger