forked from keephq/keep
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
258 additions
and
3 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,48 @@ | ||
--- | ||
title: "Pingdom" | ||
sidebarTitle: "Pingdom Provider" | ||
description: "Pingdom provider allows you to pull alerts from Pingdom or install Keep as webhook." | ||
--- | ||
|
||
## Inputs | ||
|
||
Pingdom Provider does not currently support the `notify` function. | ||
|
||
## Outputs | ||
|
||
Pingdom Provider does not currently support the `query` function. | ||
|
||
## Authentication Parameters | ||
|
||
The `api_key` is required for connecting to the Pingdom provider. You can obtain them as described in the "Connecting with the Provider" section. | ||
|
||
## Connecting with the Provider | ||
|
||
### API Key | ||
|
||
To obtain the Pingdom API key, follow these steps: | ||
|
||
1. Log in to your Pingdom account. | ||
2. Navigate to the "Settings" section. | ||
3. Click on the "Pingdom API" tab. | ||
4. Generate a new API Key. | ||
|
||
|
||
## Fingerprinting | ||
|
||
Fingerprints in Pingdom are calculated based on the `check_id` incoming/pulled event. | ||
|
||
## Scopes | ||
|
||
- read (Read) | ||
Required: True | ||
Description: Read data from your Pingdom account. | ||
|
||
## Notes | ||
|
||
_No information yet, feel free to contribute it using the "Edit this page" link at the bottom of the page_ | ||
|
||
## Useful Links | ||
|
||
- [Pingdom Webhook Documentation](https://www.pingdom.com/resources/webhooks) | ||
- [Pingdom Actions API](https://docs.pingdom.com/api/#tag/Actions) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Empty file.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,184 @@ | ||
import dataclasses | ||
import datetime | ||
import json | ||
import random | ||
|
||
import pydantic | ||
import requests | ||
|
||
from keep.api.models.alert import AlertDto | ||
from keep.contextmanager.contextmanager import ContextManager | ||
from keep.providers.base.base_provider import BaseProvider | ||
from keep.providers.models.provider_config import ProviderConfig, ProviderScope | ||
from keep.providers.providers_factory import ProvidersFactory | ||
|
||
|
||
@pydantic.dataclasses.dataclass | ||
class PingdomProviderAuthConfig: | ||
api_key: str = dataclasses.field( | ||
metadata={ | ||
"description": "Pingdom API Key", | ||
"sensitive": True, | ||
"required": True, | ||
}, | ||
) | ||
|
||
|
||
class PingdomProvider(BaseProvider): | ||
"Get alerts from Pingdom." | ||
webhook_description = """Install Keep as Pingdom webhook | ||
1. Go to Settings > Integrations. | ||
2. Click Add Integration. | ||
3. Enter: | ||
Type = Webhook | ||
Name = Keep | ||
URL = {keep_webhook_api_url_with_auth} | ||
4. Click Save Integration. | ||
""" | ||
webhook_template = """""" | ||
|
||
PROVIDER_SCOPES = [ | ||
ProviderScope( | ||
name="read", | ||
description="Read alerts from Pingdom.", | ||
mandatory=True, | ||
), | ||
] | ||
|
||
def __init__(self, *args, **kwargs): | ||
super().__init__(*args, **kwargs) | ||
|
||
def validate_config(self): | ||
""" | ||
Validate provider configuration specific to Pingdom. | ||
""" | ||
self.authentication_config = PingdomProviderAuthConfig( | ||
**self.config.authentication | ||
) | ||
|
||
def dispose(self): | ||
""" | ||
Dispose provider resources. | ||
""" | ||
pass | ||
|
||
def _get_headers(self): | ||
""" | ||
Helper method to get headers for Pingdom API requests. | ||
""" | ||
return { | ||
"Authorization": f"Bearer {self.authentication_config.api_key}", | ||
} | ||
|
||
def validate_scopes(self) -> dict[str, bool | str]: | ||
""" | ||
Validate Pingdom scopes. | ||
""" | ||
# try get alerts from pingdom | ||
try: | ||
self.get_alerts() | ||
return { | ||
"read": True, | ||
} | ||
except Exception as e: | ||
return {"read": str(e)} | ||
|
||
def _get_alerts(self) -> list[AlertDto]: | ||
""" | ||
Retrieve alerts from Pingdom. | ||
""" | ||
# Example API call to Pingdom to retrieve alerts | ||
alerts_response = requests.get( | ||
"https://api.pingdom.com/api/3.1/actions", headers=self._get_headers() | ||
) | ||
alerts_response.raise_for_status() | ||
alerts = alerts_response.json().get("actions", {}).get("alerts") | ||
|
||
checks_response = requests.get( | ||
"https://api.pingdom.com/api/3.1/checks", headers=self._get_headers() | ||
) | ||
checks_response.raise_for_status() | ||
checks = checks_response.json().get("checks", []) | ||
|
||
alerts_dtos = [] | ||
for alert in alerts: | ||
check_name = next( | ||
( | ||
check.get("name") | ||
for check in checks | ||
if check.get("id") == alert.get("checkid") | ||
), | ||
None, | ||
) | ||
alert_dto = AlertDto( | ||
id=alert.get("checkid"), | ||
fingerprint=alert.get("checkid"), | ||
name=check_name, | ||
severity="N/A", | ||
status=alert.get("status"), | ||
lastReceived=datetime.datetime.now().isoformat(), | ||
description=alert.get("messagefull"), | ||
charged=alert.get("charged"), | ||
source=["pingdom"], | ||
username=alert.get("username"), | ||
userid=alert.get("userid"), | ||
via=alert.get("via"), | ||
) | ||
alerts_dtos.append(alert_dto) | ||
|
||
return alerts_dtos | ||
|
||
@staticmethod | ||
def format_alert(event: dict) -> AlertDto: | ||
# https://pingdom.com/resources/webhooks/#Examples-of-webhook-JSON-output-for-uptime-checks | ||
alert = AlertDto( | ||
id=event.get("check_id"), | ||
fingerprint=event.get("check_id"), | ||
name=event.get("check_name"), | ||
status=event.get("current_state"), | ||
severity=event.get("importance_level", None), | ||
lastReceived=datetime.datetime.now().isoformat(), | ||
fatigueMeter=random.randint(0, 100), | ||
description=event.get("long_description"), | ||
source=["pingdom"], | ||
check_params=event.get("check_params", {}), | ||
check_type=event.get("check_type", None), | ||
short_description=event.get("description", None), | ||
previous_status=event.get("previous_state", None), | ||
tags=event.get("tags", []), | ||
version=event.get("version", 1), | ||
state_changed_utc_time=event.get("state_changed_utc_time", None), | ||
state_changed_timestamp=event.get("state_changed_timestamp", None), | ||
custom_message=event.get("custom_message", None), | ||
first_probe=event.get("first_probe", None), | ||
second_probe=event.get("second_probe", None), | ||
) | ||
return alert | ||
|
||
|
||
if __name__ == "__main__": | ||
# Output debug messages | ||
import logging | ||
|
||
logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler()]) | ||
|
||
# Load environment variables | ||
import os | ||
|
||
api_key = os.environ.get("PINGDOM_API_KEY") | ||
if not api_key: | ||
raise Exception("PINGDOM_API_KEY environment variable is not set") | ||
context_manager = ContextManager( | ||
tenant_id="singletenant", | ||
workflow_id="test", | ||
) | ||
config = {"authentication": {"api_key": api_key}} | ||
provider = ProvidersFactory.get_provider( | ||
context_manager, | ||
provider_id="pingdom-keephq", | ||
provider_type="pingdom", | ||
provider_config=config, | ||
) | ||
scopes = provider.validate_scopes() | ||
alerts = provider.get_alerts() | ||
print(alerts) |