Skip to content

Kentik client

The Kentik service is used for external interactions with Kentik.

NewKentikDevice

Bases: BaseModel

API model for creating a new device in Kentik.

Source code in gso/services/kentik_client.py 
16
17
18
19
20
21
22
23
24
25
26
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

KentikClient

The client for Kentik that interacts with an external instance.

Source code in gso/services/kentik_client.py 
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
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.

        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 `plan` objects that each have the following shape:

        <!-- vale off -->
        ```py
        "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 -->

        Returns:
            A list of plans configured in Kentik.
        """
        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)

__init__()

Instantiate a new Kentik Client.

Source code in gso/services/kentik_client.py 
32
33
34
35
36
37
38
39
40
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",
    })

get_devices()

List all devices in Kentik.

Returns: a list of shape [{**device_1}, {**device_2}, ..., {**device_n}]}.

Source code in gso/services/kentik_client.py 
61
62
63
64
65
66
67
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"]

get_device(device_id)

Get a device by ID.

Source code in gso/services/kentik_client.py 
69
70
71
72
73
74
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

get_device_by_name(device_name)

Fetch a device in Kentik by its FQDN.

If the device is not found, returns an empty dict.

Parameters:

Name Type Description Default
device_name str

The FQDN of the sought device.

 required
Source code in gso/services/kentik_client.py 
76
77
78
79
80
81
82
83
84
85
86
87
88
89
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 {}

get_sites()

Get a list of all available sites in Kentik.

Source code in gso/services/kentik_client.py 
91
92
93
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"]

get_site(site_id)

Get a site by ID.

Source code in gso/services/kentik_client.py 
95
96
97
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}")

get_site_by_name(site_slug)

Get a Kentik site by its name.

If the site is not found, return an empty dict.

Parameters:

Name Type Description Default
site_slug str

The name of the site, should be a three-letter slug like COR or POZ.

 required
Source code in gso/services/kentik_client.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
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.

    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 {}

get_plans()

Get all Kentik plans available.

Returns a list of plan objects that each have the following shape:

"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": {},
}

Returns:

Type Description
list[dict[str, Any]]

A list of plans configured in Kentik.
Source code in gso/services/kentik_client.py 
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def get_plans(self) -> list[dict[str, Any]]:
    """Get all Kentik plans available.

    Returns a list of `plan` objects that each have the following shape:

    <!-- vale off -->
    ```py
    "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 -->

    Returns:
        A list of plans configured in Kentik.
    """
    return self._send_request("GET", "v5/plans")["plans"]

get_plan(plan_id)

Get a Kentik plan by ID.

Source code in gso/services/kentik_client.py 
156
157
158
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}")

get_plan_by_name(plan_name)

Get a Kentik plan by its name.

If the plan is not found, returns an empty dict.

Parameters:

Name Type Description Default
plan_name str

The name of the plan.

 required
Source code in gso/services/kentik_client.py 
160
161
162
163
164
165
166
167
168
169
170
171
172
173
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 {}

create_device(device)

Add a new device to Kentik.

Source code in gso/services/kentik_client.py 
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
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

update_device(device_id, updated_device)

Update an existing device in Kentik.

Source code in gso/services/kentik_client.py 
206
207
208
209
210
211
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

remove_device(device_id, *, archive)

Remove a device from Kentik.

Parameters:

Name Type Description Default
device_id str

The Kentik internal ID of the device that is to be removed.

 required
archive bool

Archive the device instead of completely deleting it.

 required
Source code in gso/services/kentik_client.py 
213
214
215
216
217
218
219
220
221
222
223
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}")

remove_device_by_fqdn(fqdn, *, archive)

Remove a device from Kentik, by its FQDN.

Source code in gso/services/kentik_client.py 
225
226
227
228
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)