Add a Kentik service to GSO

Add and remove devices using Kentik, the router creation and termination workflows have been updated

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."""