ACLP Logs - Extend destination API with new updates

Author: sjer-akamai

linode_api4/groups/monitor.py

@@ -18,6 +18,10 @@
     MonitorService,
     MonitorServiceToken,
 )
+from linode_api4.objects.monitor import (
+    AkamaiObjectStorageLogsDestinationDetails,
+    CustomHTTPSLogsDestinationDetails,
+)
 
 __all__ = [
     "MonitorGroup",
@@ -367,44 +371,60 @@ def destination_create(
         self,
         label: str,
         type: Union[LogsDestinationType, str],
-        access_key_id: str,
-        access_key_secret: str,
-        bucket_name: str,
-        host: str,
-        path: Optional[str] = None,
+        details: Union[
+            AkamaiObjectStorageLogsDestinationDetails,
+            CustomHTTPSLogsDestinationDetails,
+        ],
     ) -> LogsDestination:
         """
-        Creates a new :any:`LogsDestination` for logs on this account with
-        the given label, type, and object storage details. For example::
+        Creates a new :any:`LogsDestination` for logs on this account.
+
+        For an ``akamai_object_storage`` destination::
 
            client = LinodeClient(TOKEN)
 
            new_destination = client.monitor.destination_create(
                label="OBJ_logs_destination",
                type="akamai_object_storage",
-               access_key_id="1ABCD23EFG4HIJKLMNO5",
-               access_key_secret="1aB2CD3e4fgHi5JK6lmnop7qR8STU9VxYzabcdefHh",
-               bucket_name="primary-bucket",
-               host="primary-bucket-1.us-east-12.linodeobjects.com",
-               path="audit-logs"
-            )
+               details=AkamaiObjectStorageLogsDestinationDetails(
+                   access_key_id="1ABCD23EFG4HIJKLMNO5",
+                   access_key_secret="1aB2CD3e4fgHi5JK6lmnop7qR8STU9VxYzabcdefHh",
+                   bucket_name="primary-bucket",
+                   host="primary-bucket-1.us-east-12.linodeobjects.com",
+                   path="audit-logs",
+               )
+           )
+
+        For a ``custom_https`` destination::
+
+           new_destination = client.monitor.destination_create(
+               label="custom_logs_destination",
+               type="custom_https",
+               details=CustomHTTPSLogsDestinationDetails(
+                   endpoint_url="https://my-site.com/log-storage/basicAuth",
+                   authentication=DestinationAuthentication(
+                       type="basic",
+                       details=BasicAuthenticationDetails(
+                           basic_authentication_user="user",
+                           basic_authentication_password="pass",
+                       ),
+                   ),
+                   data_compression="gzip",
+                   content_type="application/json",
+               )
+           )
 
         API Documentation: https://techdocs.akamai.com/linode-api/reference/post-destination
 
-        :param label: The name for this logs destination
+        :param label: The name for this logs destination.
         :type label: str
-        :param type: The type of destination for logs data sync. Currently, only ``akamai_object_storage`` is supported for use.
+        :param type: The type of destination — ``akamai_object_storage`` or ``custom_https``.
         :type type: str or LogsDestinationType
-        :param access_key_id: The unique identifier assigned to the Object Storage key required for authentication to the bucket.
-        :type access_key_id: str
-        :param access_key_secret: The Object Storage key's secret key.
-        :type access_key_secret: str
-        :param bucket_name: The name of the Object Storage bucket
-        :type bucket_name: str
-        :param host: The hostname where the Object Storage bucket can be accessed
-        :type host: str
-        :param path: (Optional) Custom path for log storage in your Object Storage bucket.
-        :type path: Optional[str]
+        :param details: A typed details object matching the destination type.
+                        Use :class:`AkamaiObjectStorageLogsDestinationDetails` for
+                        ``akamai_object_storage`` or :class:`CustomHTTPSLogsDestinationDetails`
+                        for ``custom_https``.
+        :type details: AkamaiObjectStorageLogsDestinationDetails or CustomHTTPSLogsDestinationDetails
 
         :returns: The newly created logs destination.
         :rtype: LogsDestination
@@ -413,17 +433,9 @@ def destination_create(
         params = {
             "label": label,
             "type": type,
-            "details": {
-                "access_key_id": access_key_id,
-                "access_key_secret": access_key_secret,
-                "bucket_name": bucket_name,
-                "host": host,
-            },
+            "details": details.dict,
         }
 
-        if path is not None:
-            params["details"]["path"] = path
-
         result = self.client.post("/monitor/streams/destinations", data=params)
 
         if "id" not in result:

linode_api4/objects/monitor.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass, field
-from typing import List, Optional, Union
+from typing import Any, Dict, List, Optional, Union
 
 from linode_api4.objects import DerivedBase
 from linode_api4.objects.base import Base, Property
@@ -20,8 +20,10 @@
     "MonitorServiceToken",
     "RuleCriteria",
     "TriggerConditions",
+    "AkamaiObjectStorageLogsDestinationDetails",
+    "CustomHTTPSLogsDestinationDetails",
+    "LogsDestinationDetailsBase",
     "LogsDestination",
-    "LogsDestinationDetails",
     "LogsDestinationHistory",
     "LogsDestinationStatus",
     "LogsDestinationType",
@@ -143,10 +145,26 @@ class AlertStatus(StrEnum):
 
 class LogsDestinationType(StrEnum):
     """
-    The type of destination for logs data sync. Currently, only ``akamai_object_storage`` is supported.
+    The type of destination for logs data sync.
     """
 
     akamai_object_storage = "akamai_object_storage"
+    custom_https = "custom_https"
+
+
+class AuthenticationType(StrEnum):
+    none = "none"
+    basic = "basic"
+
+
+class DataCompressionType(StrEnum):
+    gzip = "gzip"
+    none = "none"
+
+
+class ContentType(StrEnum):
+    json = "application/json"
+    json_utf8 = "application/json; charset=utf-8"
 
 
 class LogsDestinationStatus(StrEnum):
@@ -541,9 +559,97 @@ class AlertChannel(Base):
 
 
 @dataclass
-class LogsDestinationDetails(JSONObject):
+class BasicAuthenticationDetails(JSONObject):
+    """
+    Includes additional parameters necessary to define basic authentication.
+    """
+
+    basic_authentication_user: Optional[str] = None
+    basic_authentication_password: Optional[str] = None
+
+
+@dataclass
+class DestinationAuthentication(JSONObject):
+    """
+    Authentication details required to access the endpoint_url.
+    """
+
+    type: Optional[AuthenticationType] = None
+    details: Optional[BasicAuthenticationDetails] = None
+
+
+@dataclass
+class CustomHeader(JSONObject):
+    """
+    Pairs of parameters used to optionally include custom headers in the request.
+    """
+
+    name: str = ""
+    value: str = ""
+
+
+@dataclass
+class ClientCertificateDetails(JSONObject):
+    """
+    Contains TLS client certificate information to additionally secure the connection.
+    """
+
+    client_ca_certificate: Optional[str] = None
+    client_certificate: Optional[str] = None
+    client_private_key: Optional[str] = None
+    tls_hostname: Optional[str] = None
+
+
+@dataclass
+class LogsDestinationDetailsBase(JSONObject):
+    """
+    Base class for Logs Destination details.
+    Use the factory method to instantiate the correct subclass based on destination type.
+    """
+
+    @classmethod
+    def load_by_type(
+        cls, dest_type: str, json_dict: dict
+    ) -> Optional["LogsDestinationDetailsBase"]:
+        """
+        Factory method that instantiates the correct details subclass
+        based on the destination type string.
+
+        :param dest_type: The destination type (e.g. "akamai_object_storage", "custom_https").
+        :param json_dict: The raw JSON dict for the details block.
+        :returns: A populated subclass instance, or None if json_dict is empty/None.
+        """
+        if not json_dict:
+            return None
+
+        if dest_type == LogsDestinationType.akamai_object_storage:
+            return AkamaiObjectStorageLogsDestinationDetails.from_json(
+                json_dict
+            )
+        elif dest_type == LogsDestinationType.custom_https:
+            return CustomHTTPSLogsDestinationDetails.from_json(json_dict)
+
+        return None
+
+
+@dataclass
+class CustomHTTPSLogsDestinationDetails(LogsDestinationDetailsBase):
     """
-    Represents the details block for LogsDestination.
+    Represents the details block for custom_https LogsDestination type.
+    """
+
+    endpoint_url: str = ""
+    authentication: Optional[DestinationAuthentication] = None
+    data_compression: Optional[DataCompressionType] = None
+    content_type: Optional[ContentType] = None
+    custom_headers: Optional[List[CustomHeader]] = None
+    client_certificate_details: Optional[ClientCertificateDetails] = None
+
+
+@dataclass
+class AkamaiObjectStorageLogsDestinationDetails(LogsDestinationDetailsBase):
+    """
+    Represents the details block for Akamai Object Storage LogsDestination type.
     Fields:
       - access_key_id: str - The unique identifier assigned to the Object Storage key required for authentication to the bucket.
       - bucket_name: str - The name of the Object Storage bucket.
@@ -568,7 +674,7 @@ class LogsDestinationHistory(Base):
     properties = {
         "created": Property(is_datetime=True),
         "created_by": Property(),
-        "details": Property(json_object=LogsDestinationDetails),
+        "details": Property(),
         "id": Property(identifier=True),
         "label": Property(),
         "status": Property(),
@@ -578,6 +684,17 @@ class LogsDestinationHistory(Base):
         "version": Property(),
     }
 
+    def _populate(self, json):
+        super()._populate(json)
+
+        if json and "details" in json and "type" in json:
+            self._set(
+                "details",
+                LogsDestinationDetailsBase.load_by_type(
+                    json["type"], json["details"]
+                ),
+            )
+
 
 class LogsDestination(Base):
     """
@@ -591,7 +708,7 @@ class LogsDestination(Base):
     properties = {
         "created": Property(is_datetime=True),
         "created_by": Property(),
-        "details": Property(mutable=True, json_object=LogsDestinationDetails),
+        "details": Property(mutable=True),
         "id": Property(identifier=True),
         "label": Property(mutable=True),
         "status": Property(),
@@ -601,6 +718,17 @@ class LogsDestination(Base):
         "version": Property(),
     }
 
+    def _populate(self, json):
+        super()._populate(json)
+
+        if json and "details" in json and "type" in json:
+            self._set(
+                "details",
+                LogsDestinationDetailsBase.load_by_type(
+                    json["type"], json["details"]
+                ),
+            )
+
     @property
     def history(self):
         """
@@ -636,7 +764,23 @@ class LogsStreamDestination(JSONObject):
     id: int = 0
     label: str = ""
     type: Optional[LogsDestinationType] = None
-    details: Optional[LogsDestinationDetails] = None
+    details: Optional[LogsDestinationDetailsBase] = None
+
+    @classmethod
+    def from_json(
+        cls, json: Dict[str, Any]
+    ) -> Optional["LogsStreamDestination"]:
+        if json is None:
+            return None
+
+        obj = super().from_json(json)
+
+        if obj and json.get("type"):
+            obj.details = LogsDestinationDetailsBase.load_by_type(
+                json["type"], json.get("details")
+            )
+
+        return obj
 
 
 class LogsStreamHistory(Base):

test/fixtures/monitor_streams_2.json

@@ -0,0 +1,43 @@
+{
+  "id": 2,
+  "label": "my-custom-https-stream",
+  "type": "audit_logs",
+  "status": "active",
+  "destinations": [
+    {
+      "id": 2,
+      "label": "my-custom-https-destination",
+      "type": "custom_https",
+      "details": {
+        "endpoint_url": "https://my-site.com/log-storage/basicAuth",
+        "authentication": {
+          "type": "basic",
+          "details": {
+            "basic_authentication_user": "John_Q",
+            "basic_authentication_password": "p@$$w0Rd"
+          }
+        },
+        "data_compression": "gzip",
+        "content_type": "application/json",
+        "custom_headers": [
+          {
+            "name": "Cache-Control",
+            "value": "max-age=0"
+          }
+        ],
+        "client_certificate_details": {
+          "client_ca_certificate": "-----BEGIN CERTIFICATE-----\nMIIBIjANBgkq...\n-----END CERTIFICATE-----",
+          "client_certificate": "-----BEGIN CERTIFICATE-----\nMIIBIjANBgkq...\n-----END CERTIFICATE-----",
+          "client_private_key": "-----BEGIN PRIVATE KEY-----\nMIIBIjANBgkq...\n-----END PRIVATE KEY-----",
+          "tls_hostname": "my-site.com"
+        }
+      }
+    }
+  ],
+  "created": "2024-08-01T12:00:00",
+  "updated": "2024-08-01T12:00:00",
+  "created_by": "tester",
+  "updated_by": "tester",
+  "version": 1
+}
+