"""The Kentik service is used for external interactions with Kentik."""
import logging
from typing import Any, Literal
import requests
from orchestrator.utils.errors import ProcessFailureError
from pydantic import BaseModel
from requests import Response
from gso.settings import load_oss_params
logger = logging.getLogger(__name__)
class NewKentikDevice(BaseModel):
"""API model for creating a new device in Kentik."""
device_name: str
device_description: str
sending_ips: list[str]
site_id: int
device_snmp_ip: str
device_bgp_flowspec: bool
device_bgp_neighbor_ip: str
device_bgp_neighbor_ip6: str
class KentikClient:
"""The client for Kentik that interacts with an external instance."""
def __init__(self) -> None:
"""Instantiate a new Kentik Client."""
self.config = load_oss_params().KENTIK
self.session = requests.Session()
self.session.headers.update({
"X-CH-Auth-Email": self.config.user_email,
"X-CH-Auth-API-Token": self.config.api_key,
"Content-Type": "application/json",
})
def _send_request(
self, method: Literal["GET", "POST", "PUT"], endpoint: str, data: dict[str, Any] | None = None
) -> dict[str, Any]:
url = self.config.api_base + endpoint
logger.debug("Kentik - Sending %s request to %s with headers %s", method, url, data)
result = self.session.request(method, url, json=data).json()
if "error" in result or "kentik_error" in result:
msg = "Failed to process request in Kentik"
raise ProcessFailureError(msg, details=result)
logger.debug("Kentik - Received response %s", result)
return result
def _send_delete(self, endpoint: str, data: dict[str, Any] | None = None) -> Response:
url = self.config.api_base + endpoint
logger.debug("Kentik - Sending delete request to %s with headers %s", url, data)
return self.session.delete(url, json=data)
def get_devices(self) -> list[dict[str, Any]]:
"""List all devices in Kentik.
Returns:
a list of shape ``[{**device_1}, {**device_2}, ..., {**device_n}]}``.
"""
return self._send_request("GET", "v5/devices")["devices"]
def get_device(self, device_id: str) -> dict[str, Any]:
"""Get a device by ID."""
device = self._send_request("GET", f"v5/device/{device_id}")
device.pop("custom_column_data", None)
device.pop("custom_columns", None)
return device
def get_device_by_name(self, device_name: str) -> dict[str, Any]:
"""Fetch a device in Kentik by its FQDN.
If the device is not found, returns an empty dict.
Args:
device_name: The FQDN of the sought device.
"""
devices = self.get_devices()
for device in devices:
if device["device_name"] == device_name:
return device
return {}
def get_sites(self) -> list[dict[str, Any]]:
"""Get a list of all available sites in Kentik."""
return self._send_request("GET", "v5/sites")["sites"]
def get_site(self, site_id: str) -> dict[str, Any]:
"""Get a site by ID."""
return self._send_request("GET", f"v5/site/{site_id}")
def get_site_by_name(self, site_slug: str) -> dict[str, Any]:
"""Get a Kentik site by its name.
If the site is not found, return an empty dict.
.. vale off
Args:
site_slug: The name of the site, should be a three-letter slug like COR or POZ.
"""
sites = self.get_sites()
for site in sites:
if site["site_name"] == site_slug:
return site
return {}
def get_plans(self) -> list[dict[str, Any]]:
"""Get all Kentik plans available.
Returns:
a list of ``plans`` that each have the following shape:
.. vale off
.. code-block:: json
"plan": {
"active": true,
"bgp_enabled": true,
"cdate" "1970-01-01T01:01:01.000Z",
"company_id": 111111,
"description": "A description of this plan",
"deviceTypes": [
{"device_type": "router"},
{"device_type": "host-nprobe-dns-www"}
],
"devices": [
{
"id": "111111",
"device_name": "rt0.city.tld.internal",
"device_type": "router"
},
],
"edate": "2999-01-01T09:09:09.000Z",
"fast_retention": 10,
"full_retention": 5,
"id": 11111,
"max_bigdata_fps": 100,
"max_devices": 9001,
"max_fps": 200,
"name": "KENTIK-PLAN-01",
"metadata": {},
}
.. vale on
"""
return self._send_request("GET", "v5/plans")["plans"]
def get_plan(self, plan_id: int) -> dict[str, Any]:
"""Get a Kentik plan by ID."""
return self._send_request("GET", f"v5/plan/{plan_id}")
def get_plan_by_name(self, plan_name: str) -> dict[str, Any]:
"""Get a Kentik plan by its name.
If the plan is not found, returns an empty dict.
Args:
plan_name: The name of the plan.
"""
plans = self.get_plans()
for plan in plans:
if plan["name"] == plan_name:
return plan
return {}
def create_device(self, device: NewKentikDevice) -> dict[str, Any]:
"""Add a new device to Kentik."""
plan_id = self.get_plan_by_name(self.config.placeholder_license_key)["id"]
request_body = {
"device": {
**device.model_dump(exclude=set("device_name")),
"device_name": device.device_description,
"device_type": self.config.device_type,
"device_subtype": self.config.device_type,
"minimize_snmp": self.config.minimize_snmp,
"device_sample_rate": self.config.sample_rate,
"plan_id": plan_id,
"device_snmp_community": self.config.snmp_community,
"device_bgp_type": self.config.bgp_type,
"bgp_lookup_strategy": self.config.bgp_lookup_strategy,
"device_bgp_neighbor_asn": str(self.config.ASN),
"device_bgp_password": self.config.md5_password,
}
}
new_device = self._send_request("POST", "v5/device", request_body)["device"]
# The name of the device has to be updated from the subscription ID to its FQDN.
# This is a limitation of the Kentik API that disallows settings device names containing a . symbol.
self.update_device(new_device["id"], {"device": {"device_name": device.device_name}})
new_device["device_name"] = device.device_name
new_device.pop("custom_column_data", None)
new_device.pop("custom_columns", None)
return new_device
def update_device(self, device_id: str, updated_device: dict[str, Any]) -> dict[str, Any]:
"""Update an existing device in Kentik."""
device = self._send_request("PUT", f"v5/device/{device_id}", updated_device)["device"]
device.pop("custom_column_data", None)
device.pop("custom_columns", None)
return device
def remove_device(self, device_id: str, *, archive: bool) -> None:
"""Remove a device from Kentik.
Args:
device_id: The Kentik internal ID of the device that is to be removed.
archive: Archive the device instead of completely deleting it.
"""
if not archive:
self._send_delete(f"v5/device/{device_id}")
self._send_delete(f"v5/device/{device_id}")
def remove_device_by_fqdn(self, fqdn: str, *, archive: bool) -> None:
"""Remove a device from Kentik, by its FQDN."""
device_id = self.get_device_by_name(fqdn)["id"]
self.remove_device(device_id, archive=archive)