feat: add metrics endpoint for load balancers and servers (#331)

Adds the missing metrics endpoint for the load balancers and servers
resources.

-
https://docs.hetzner.cloud/#load-balancers-get-metrics-for-a-loadbalancer
- https://docs.hetzner.cloud/#servers-get-metrics-for-a-server

Author: jooola

examples/get_server_metrics.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta
+from os import environ
+
+from hcloud import Client
+from hcloud.images import Image
+from hcloud.server_types import ServerType
+
+assert (
+    "HCLOUD_TOKEN" in environ
+), "Please export your API token in the HCLOUD_TOKEN environment variable"
+token = environ["HCLOUD_TOKEN"]
+
+client = Client(token=token)
+
+server = client.servers.get_by_name("my-server")
+if server is None:
+    response = client.servers.create(
+        name="my-server",
+        server_type=ServerType("cx11"),
+        image=Image(name="ubuntu-22.04"),
+    )
+    server = response.server
+
+end = datetime.now()
+start = end - timedelta(hours=1)
+
+response = server.get_metrics(
+    type=["cpu", "network"],
+    start=start,
+    end=end,
+)
+
+print(json.dumps(response.metrics))

hcloud/load_balancers/__init__.py

@@ -7,6 +7,7 @@
 )
 from .domain import (  # noqa: F401
     CreateLoadBalancerResponse,
+    GetMetricsResponse,
     IPv4Address,
     IPv6Network,
     LoadBalancer,

hcloud/load_balancers/client.py

@@ -1,16 +1,21 @@
 from __future__ import annotations
 
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, NamedTuple
 
+from dateutil.parser import isoparse
+
 from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
 from ..certificates import BoundCertificate
 from ..core import BoundModelBase, ClientEntityBase, Meta
 from ..load_balancer_types import BoundLoadBalancerType
 from ..locations import BoundLocation
+from ..metrics import Metrics
 from ..networks import BoundNetwork
 from ..servers import BoundServer
 from .domain import (
     CreateLoadBalancerResponse,
+    GetMetricsResponse,
     IPv4Address,
     IPv6Network,
     LoadBalancer,
@@ -23,6 +28,7 @@
     LoadBalancerTargetHealthStatus,
     LoadBalancerTargetIP,
     LoadBalancerTargetLabelSelector,
+    MetricsType,
     PrivateNet,
     PublicNetwork,
 )
@@ -177,6 +183,28 @@ def delete(self) -> bool:
         """
         return self._client.delete(self)
 
+    def get_metrics(
+        self,
+        type: MetricsType,
+        start: datetime | str,
+        end: datetime | str,
+        step: float | None = None,
+    ) -> GetMetricsResponse:
+        """Get Metrics for a LoadBalancer.
+
+        :param type: Type of metrics to get.
+        :param start: Start of period to get Metrics for (in ISO-8601 format).
+        :param end: End of period to get Metrics for (in ISO-8601 format).
+        :param step: Resolution of results in seconds.
+        """
+        return self._client.get_metrics(
+            self,
+            type=type,
+            start=start,
+            end=end,
+            step=step,
+        )
+
     def get_actions_list(
         self,
         status: list[str] | None = None,
@@ -533,6 +561,46 @@ def delete(self, load_balancer: LoadBalancer | BoundLoadBalancer) -> bool:
         )
         return True
 
+    def get_metrics(
+        self,
+        load_balancer: LoadBalancer | BoundLoadBalancer,
+        type: MetricsType | list[MetricsType],
+        start: datetime | str,
+        end: datetime | str,
+        step: float | None = None,
+    ) -> GetMetricsResponse:
+        """Get Metrics for a LoadBalancer.
+
+        :param load_balancer: The Load Balancer to get the metrics for.
+        :param type: Type of metrics to get.
+        :param start: Start of period to get Metrics for (in ISO-8601 format).
+        :param end: End of period to get Metrics for (in ISO-8601 format).
+        :param step: Resolution of results in seconds.
+        """
+        if not isinstance(type, list):
+            type = [type]
+        if isinstance(start, str):
+            start = isoparse(start)
+        if isinstance(end, str):
+            end = isoparse(end)
+
+        params: dict[str, Any] = {
+            "type": ",".join(type),
+            "start": start.isoformat(),
+            "end": end.isoformat(),
+        }
+        if step is not None:
+            params["step"] = step
+
+        response = self._client.request(
+            url=f"/load_balancers/{load_balancer.id}/metrics",
+            method="GET",
+            params=params,
+        )
+        return GetMetricsResponse(
+            metrics=Metrics(**response["metrics"]),
+        )
+
     def get_actions_list(
         self,
         load_balancer: LoadBalancer | BoundLoadBalancer,

hcloud/load_balancers/domain.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 from dateutil.parser import isoparse
 
@@ -11,6 +11,7 @@
     from ..certificates import BoundCertificate
     from ..load_balancer_types import BoundLoadBalancerType
     from ..locations import BoundLocation
+    from ..metrics import Metrics
     from ..networks import BoundNetwork
     from ..servers import BoundServer
     from .client import BoundLoadBalancer
@@ -508,3 +509,26 @@ def __init__(
     ):
         self.load_balancer = load_balancer
         self.action = action
+
+
+MetricsType = Literal[
+    "open_connections",
+    "connections_per_second",
+    "requests_per_second",
+    "bandwidth",
+]
+
+
+class GetMetricsResponse(BaseDomain):
+    """Get a Load Balancer Metrics Response Domain
+
+    :param metrics: The Load Balancer metrics
+    """
+
+    __slots__ = ("metrics",)
+
+    def __init__(
+        self,
+        metrics: Metrics,
+    ):
+        self.metrics = metrics

hcloud/metrics/__init__.py

@@ -0,0 +1,3 @@
+from __future__ import annotations
+
+from .domain import Metrics, TimeSeries  # noqa: F401

hcloud/metrics/domain.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Dict, List, Literal, Tuple
+
+from dateutil.parser import isoparse
+
+from ..core import BaseDomain
+
+TimeSeries = Dict[str, Dict[Literal["values"], List[Tuple[float, str]]]]
+
+
+class Metrics(BaseDomain):
+    """Metrics Domain
+
+    :param start: Start of period of metrics reported.
+    :param end: End of period of metrics reported.
+    :param step: Resolution of results in seconds.
+    :param time_series: Dict with time series data, using the name of the time series as
+        key. The metrics timestamps and values are stored in a list of tuples
+        ``[(timestamp, value), ...]``.
+    """
+
+    start: datetime
+    end: datetime
+    step: float
+    time_series: TimeSeries
+
+    __slots__ = (
+        "start",
+        "end",
+        "step",
+        "time_series",
+    )
+
+    def __init__(
+        self,
+        start: str,
+        end: str,
+        step: float,
+        time_series: TimeSeries,
+    ):
+        self.start = isoparse(start)
+        self.end = isoparse(end)
+        self.step = step
+        self.time_series = time_series

hcloud/servers/__init__.py

@@ -4,6 +4,7 @@
 from .domain import (  # noqa: F401
     CreateServerResponse,
     EnableRescueResponse,
+    GetMetricsResponse,
     IPv4Address,
     IPv6Network,
     PrivateNet,

hcloud/servers/client.py

@@ -1,24 +1,30 @@
 from __future__ import annotations
 
 import warnings
+from datetime import datetime
 from typing import TYPE_CHECKING, Any, NamedTuple
 
+from dateutil.parser import isoparse
+
 from ..actions import ActionsPageResult, BoundAction, ResourceActionsClient
 from ..core import BoundModelBase, ClientEntityBase, Meta
 from ..datacenters import BoundDatacenter
 from ..firewalls import BoundFirewall
 from ..floating_ips import BoundFloatingIP
 from ..images import BoundImage, CreateImageResponse
 from ..isos import BoundIso
+from ..metrics import Metrics
 from ..placement_groups import BoundPlacementGroup
 from ..primary_ips import BoundPrimaryIP
 from ..server_types import BoundServerType
 from ..volumes import BoundVolume
 from .domain import (
     CreateServerResponse,
     EnableRescueResponse,
+    GetMetricsResponse,
     IPv4Address,
     IPv6Network,
+    MetricsType,
     PrivateNet,
     PublicNetwork,
     PublicNetworkFirewall,
@@ -210,6 +216,29 @@ def update(
         """
         return self._client.update(self, name, labels)
 
+    def get_metrics(
+        self,
+        type: MetricsType | list[MetricsType],
+        start: datetime | str,
+        end: datetime | str,
+        step: float | None = None,
+    ) -> GetMetricsResponse:
+        """Get Metrics for a Server.
+
+        :param server: The Server to get the metrics for.
+        :param type: Type of metrics to get.
+        :param start: Start of period to get Metrics for (in ISO-8601 format).
+        :param end: End of period to get Metrics for (in ISO-8601 format).
+        :param step: Resolution of results in seconds.
+        """
+        return self._client.get_metrics(
+            self,
+            type=type,
+            start=start,
+            end=end,
+            step=step,
+        )
+
     def delete(self) -> BoundAction:
         """Deletes a server. This immediately removes the server from your account, and it is no longer accessible.
 
@@ -742,6 +771,46 @@ def update(
         )
         return BoundServer(self, response["server"])
 
+    def get_metrics(
+        self,
+        server: Server | BoundServer,
+        type: MetricsType | list[MetricsType],
+        start: datetime | str,
+        end: datetime | str,
+        step: float | None = None,
+    ) -> GetMetricsResponse:
+        """Get Metrics for a Server.
+
+        :param server: The Server to get the metrics for.
+        :param type: Type of metrics to get.
+        :param start: Start of period to get Metrics for (in ISO-8601 format).
+        :param end: End of period to get Metrics for (in ISO-8601 format).
+        :param step: Resolution of results in seconds.
+        """
+        if not isinstance(type, list):
+            type = [type]
+        if isinstance(start, str):
+            start = isoparse(start)
+        if isinstance(end, str):
+            end = isoparse(end)
+
+        params: dict[str, Any] = {
+            "type": ",".join(type),
+            "start": start.isoformat(),
+            "end": end.isoformat(),
+        }
+        if step is not None:
+            params["step"] = step
+
+        response = self._client.request(
+            url=f"/servers/{server.id}/metrics",
+            method="GET",
+            params=params,
+        )
+        return GetMetricsResponse(
+            metrics=Metrics(**response["metrics"]),
+        )
+
     def delete(self, server: Server | BoundServer) -> BoundAction:
         """Deletes a server. This immediately removes the server from your account, and it is no longer accessible.