Add Kentik service

gso/oss-params-example.json

@@ -102,5 +102,23 @@
       "p_router": "UUID"
     },
     "scopes": ["https://graph.microsoft.com/.default"]
+  },
+  "KENTIK": {
+    "api_base": "https://api.kentik.com/api/",
+    "user_email": "robot-user@geant.org",
+    "api_key": "kentik_api_key",
+    "device_type": "router",
+    "billing_plans": {
+      "1": "XL license",
+      "2": "L license",
+      "3": "M license",
+      "4": "S license"
+    },
+    "sample_rate": 100,
+    "bgp_type": "device",
+    "ASN": 137,
+    "route_selection": "both directions",
+    "snmp_community": "secret community string",
+    "md5_password": "snmp password"
   }
 }

gso/services/kentik_client.py

+"""The Kentik service is used for external interactions with Kentik."""
+
+import logging
+from typing import Any
+
+import requests
+
+from gso.products.product_types.router import Router
+from gso.settings import load_oss_params
+
+logger = logging.getLogger(__name__)
+
+
+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.headers = {
+            "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, endpoint: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
+        url = self.config.api_base + endpoint
+        result = requests.request(method, url, json=data, headers=self.headers)
+
+        return result.json()
+
+    def get_devices(self) -> list[dict[str, Any]]:
+        """List all devices in Kentik."""
+        return [self._send_request("GET", "v5/devices")]
+
+    def get_device(self, device_id: str) -> dict[str, Any]:
+        """Get a device by ID."""
+        return self._send_request("GET", f"v5/device/{device_id}")
+
+    def get_device_by_name(self, device_name: str) -> dict[str, Any]:
+        """Fetch a device in Kentik by its :term:`FQDN`.
+
+        If the device is not found, returns an empty dict.
+
+        :param str device_name: The :term:`FQDN` of the sought device.
+        """
+        devices = self.get_devices()
+        for device in devices:
+            if 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.
+
+        :param str 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."""
+        return self._send_request("GET", "v5/plans")["plans"]
+
+    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.
+
+        :param str 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, router: Router) -> dict[str, Any]:
+        """Add a new device to Kentik."""
+        site_id = self.get_site_by_name(router.router.router_site.site_name)["id"]
+        plan_id = self.get_plan_by_name(self.config.billing_plans[router.router.router_site.site_tier])
+        request_body = {
+            "device": {
+                # Route selection missing
+                "deviceName": router.router.router_fqdn,
+                "deviceSubtype": self.config.device_type,
+                "sendingIps": [str(router.router.router_lo_ipv4_address)],
+                "deviceSampleRate": self.config.sample_rate,
+                "planId": plan_id,
+                "siteId": site_id,
+                "deviceSnmpIp": str(router.router.router_lo_ipv4_address),
+                "deviceSnmpCommunity": self.config.snmp_community,
+                "deviceBgpType": self.config.bgp_type,
+                "deviceBgpFlowspec": False,
+                "deviceBgpNeighborIp": str(router.router.router_lo_ipv4_address),
+                "deviceBgpNeighborIp6": str(router.router.router_lo_ipv6_address),
+                "deviceBgpNeighborAsn": str(self.config.ASN),
+                "deviceBgpPassword": self.config.md5_password,
+            }
+        }
+
+        return self._send_request("POST", "v5/device", request_body)
+
+    def remove_device(self, hostname: str) -> None:
+        """Remove a device from Kentik."""

gso/settings.py

@@ -16,6 +16,8 @@ from pydantic_forms.types import UUIDstr
 from pydantic_settings import BaseSettings
 from typing_extensions import Doc
 
+from gso.products.product_blocks.site import SiteTier
+
 logger = logging.getLogger(__name__)
 
 
@@ -171,6 +173,22 @@ class SharepointParams(BaseSettings):
     scopes: list[str]
 
 
+class KentikParams(BaseSettings):
+    """Settings for accessing Kentik's API."""
+
+    api_base: str
+    user_email: str
+    api_key: str
+    device_type: str
+    billing_plans: dict[SiteTier, str]
+    sample_rate: int
+    bgp_type: str
+    ASN: int
+    route_selection: str
+    snmp_community: str
+    md5_password: str
+
+
 class OSSParams(BaseSettings):
     """The set of parameters required for running :term:`GSO`."""
 
@@ -183,6 +201,7 @@ class OSSParams(BaseSettings):
     THIRD_PARTY_API_KEYS: dict[str, str]
     EMAIL: EmailParams
     SHAREPOINT: SharepointParams
+    KENTIK: KentikParams
 
 
 def load_oss_params() -> OSSParams: