diff --git a/.github/workflows/e2e-test-pr.yml b/.github/workflows/e2e-test-pr.yml
index f765b0a0d..b9464315f 100644
--- a/.github/workflows/e2e-test-pr.yml
+++ b/.github/workflows/e2e-test-pr.yml
@@ -2,6 +2,14 @@ on:
   pull_request:
   workflow_dispatch:
     inputs:
+      run_aclp_logs_stream_tests:
+        description: 'Set this parameter to "true" to run ACLP logs stream related test cases'
+        required: false
+        default: 'false'
+        type: choice
+        options:
+          - 'true'
+          - 'false'
       run_db_fork_tests:
         description: 'Set this parameter to "true" to run fork database related test cases'
         required: false
@@ -104,7 +112,7 @@ jobs:
         run: |
           timestamp=$(date +'%Y%m%d%H%M')
           report_filename="${timestamp}_sdk_test_report.xml"
-          make test-int RUN_DB_FORK_TESTS=${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS=${{ github.event.inputs.run_db_tests }} TEST_ARGS="--junitxml=${report_filename}" TEST_SUITE="${{ github.event.inputs.test_suite }}"
+          make test-int RUN_DB_FORK_TESTS=${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS=${{ github.event.inputs.run_db_tests }} RUN_ACLP_LOGS_STREAM_TESTS=${{ github.event.inputs.run_aclp_logs_stream_tests }} TEST_ARGS="--junitxml=${report_filename}" TEST_SUITE="${{ github.event.inputs.test_suite }}"
         env:
           LINODE_TOKEN: ${{ secrets.LINODE_TOKEN }}
 
diff --git a/.github/workflows/e2e-test.yml b/.github/workflows/e2e-test.yml
index 8a02599cc..a0350f2c3 100644
--- a/.github/workflows/e2e-test.yml
+++ b/.github/workflows/e2e-test.yml
@@ -3,6 +3,14 @@ name: Integration Tests
 on:
   workflow_dispatch:
     inputs:
+      run_aclp_logs_stream_tests:
+        description: 'Set this parameter to "true" to run ACLP logs stream related test cases'
+        required: false
+        default: 'false'
+        type: choice
+        options:
+          - 'true'
+          - 'false'
       run_db_fork_tests:
         description: 'Set this parameter to "true" to run fork database related test cases'
         required: false
@@ -99,7 +107,7 @@ jobs:
         run: |
           timestamp=$(date +'%Y%m%d%H%M')
           report_filename="${timestamp}_sdk_test_report.xml"
-          make test-int RUN_DB_FORK_TESTS=${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS=${{ github.event.inputs.run_db_tests }} TEST_SUITE="${{ github.event.inputs.test_suite }}" TEST_ARGS="--junitxml=${report_filename}"
+          make test-int RUN_DB_FORK_TESTS=${{ github.event.inputs.run_db_fork_tests }} RUN_DB_TESTS=${{ github.event.inputs.run_db_tests }} RUN_ACLP_LOGS_STREAM_TESTS=${{ github.event.inputs.run_aclp_logs_stream_tests }} TEST_SUITE="${{ github.event.inputs.test_suite }}" TEST_ARGS="--junitxml=${report_filename}"
         env:
           LINODE_TOKEN: ${{ env.LINODE_TOKEN }}
 
diff --git a/linode_api4/groups/monitor.py b/linode_api4/groups/monitor.py
index 0d7f19ce8..e00cb0fab 100644
--- a/linode_api4/groups/monitor.py
+++ b/linode_api4/groups/monitor.py
@@ -12,6 +12,11 @@
     MonitorMetricsDefinition,
     MonitorService,
     MonitorServiceToken,
+    LogsDestination,
+    LogsDestinationType,
+    LogsStream,
+    LogsStreamStatus,
+    LogsStreamType,
 )
 
 __all__ = [
@@ -332,3 +337,180 @@ def alert_definition_entities(
             *filters,
             endpoint=endpoint,
         )
+
+    def destinations(self, *filters) -> PaginatedList:
+        """
+        List available logs destinations.
+
+        Returns a paginated collection of :class:`LogsDestination` objects which
+        describe logs destinations. By default, this method returns all available
+        destinations; you can supply optional filter expressions to restrict
+        the results, for example::
+
+            # Get destinations created by username and with id 111
+            destinations = client.monitor.destinations(LogsDestination.created_by == "username",
+                                                        LogsDestination.id == 111)
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/get-destinations
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+
+        :returns: A list of :class:`LogsDestination` objects matching the query.
+        :rtype: PaginatedList of LogsDestination
+        """
+
+        return self.client._get_and_filter(LogsDestination, *filters)
+
+    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,
+    ) -> LogsDestination:
+        """
+        Creates a new :any:`LogsDestination` for logs on this account with
+        the given label, type, and object storage details. For example::
+
+           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"
+            )
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/post-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.
+        :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]
+
+        :returns: The newly created logs destination.
+        :rtype: LogsDestination
+        """
+
+        params = {
+            "label": label,
+            "type": type,
+            "details": {
+                "access_key_id": access_key_id,
+                "access_key_secret": access_key_secret,
+                "bucket_name": bucket_name,
+                "host": host,
+            }
+        }
+
+        if path is not None:
+            params["details"]["path"] = path
+
+        result = self.client.post("/monitor/streams/destinations", data=params)
+
+        if "id" not in result:
+            raise UnexpectedResponseError(
+                "Unexpected response when creating destination!",
+                json=result,
+            )
+
+        return LogsDestination(self.client, result["id"], result)
+
+    def streams(self, *filters) -> PaginatedList:
+        """
+        List available logs streams.
+
+        Returns a paginated collection of :class:`LogsStream` objects which
+        describe logs streams. By default, this method returns all available
+        streams; you can supply optional filter expressions to restrict
+        the results, for example::
+
+            # Get all streams with status ``provisioning``
+            provisioning_streams = client.monitor.streams(LogsStream.status == "provisioning")
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/get-streams
+
+        :param filters: Any number of filters to apply to this query.
+                        See :doc:`Filtering Collections</linode_api4/objects/filtering>`
+                        for more details on filtering.
+        :returns: A list of :class:`LogsStream` objects matching the query.
+        :rtype: PaginatedList of LogsStream
+        """
+
+        return self.client._get_and_filter(LogsStream, *filters)
+
+    def stream_create(
+            self,
+            destinations: list[int],
+            label: str,
+            type: Union[LogsStreamType, str],
+            status: Optional[Union[LogsStreamStatus, str]] = None
+    ) -> LogsStream:
+        """
+        Creates a new :any:`LogsStream` for logs on this account with
+        the given label, type, and object storage details. For example::
+
+           client = LinodeClient(TOKEN)
+
+           new_stream = client.monitor.stream_create(
+               destinations= [1234],
+               label="Linode_services",
+               status="active",
+               type="audit_logs"
+            )
+
+        API Documentation: https://techdocs.akamai.com/linode-api/reference/post-stream
+
+        :param destinations: List of unique identifiers for the sync points that will receive logs data.
+                            Run the List destinations operation and store the id values for each applicable destination.
+                            At the moment only single destination is supported.
+        :type destinations: list[int]
+        :param label: The name of the stream. This is used for display purposes in Akamai Cloud Manager.
+        :type label: str
+        :param type: The type of stream. Set this to ``audit_logs`` for logs consisting of all the control plane
+                    operations for the services in your Linodes.
+        :type type: str
+        :param status: (Optional) The availability status of the stream. Possible values are: ``active``, ``inactive``.
+                        Defaults to ``active``.
+        :type status: str
+
+        :returns: The newly created logs stream.
+        :rtype: LogsStream
+        """
+
+        params = {
+            "label": label,
+            "type": type,
+            "destinations": destinations,
+        }
+
+        if status is not None:
+            params["status"] = status
+
+        result = self.client.post("/monitor/streams", data=params)
+
+        if "id" not in result:
+            raise UnexpectedResponseError(
+                "Unexpected response when creating logs stream!",
+                json=result,
+            )
+
+        return LogsStream(self.client, result["id"], result)
diff --git a/linode_api4/objects/monitor.py b/linode_api4/objects/monitor.py
index 1a83b59d6..49daa6d40 100644
--- a/linode_api4/objects/monitor.py
+++ b/linode_api4/objects/monitor.py
@@ -20,6 +20,16 @@
     "MonitorServiceToken",
     "RuleCriteria",
     "TriggerConditions",
+    "LogsDestination",
+    "LogsDestinationDetails",
+    "LogsDestinationHistory",
+    "LogsDestinationStatus",
+    "LogsDestinationType",
+    "LogsStream",
+    "LogsStreamHistory",
+    "LogsStreamType",
+    "LogsStreamStatus",
+    "LogsStreamDestination",
 ]
 
 
@@ -131,6 +141,18 @@ class AlertStatus(StrEnum):
     AlertDefinitionStatusFailed = "failed"
 
 
+class LogsDestinationType(StrEnum):
+    """
+    The type of destination for logs data sync. Currently, only ``akamai_object_storage`` is supported.
+    """
+    akamai_object_storage = "akamai_object_storage"
+
+
+class LogsDestinationStatus(StrEnum):
+    active = "active"
+    inactive = "inactive"
+
+
 @dataclass
 class Filter(JSONObject):
     """
@@ -515,3 +537,185 @@ class AlertChannel(Base):
         "created_by": Property(),
         "updated_by": Property(),
     }
+
+
+@dataclass
+class LogsDestinationDetails(JSONObject):
+    """
+    Represents the details block for LogsDestination.
+    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.
+      - host: str - The hostname where the Object Storage bucket can be accessed.
+      - path: Optional[str] - The specific path in an Object Storage bucket where audit logs files are uploaded. May be absent or None in API responses.
+    """
+
+    access_key_id: str = ""
+    access_key_secret: Optional[str] = None
+    bucket_name: str = ""
+    host: str = ""
+    path: Optional[str] = None
+
+
+class LogsDestinationHistory(Base):
+    """
+    Represents a read-only historical snapshot of a Logs Destination.
+
+    API documentation: https://techdocs.akamai.com/linode-api/reference/get-destination-history
+    """
+
+    properties = {
+        "created": Property(is_datetime=True),
+        "created_by": Property(),
+        "details": Property(json_object=LogsDestinationDetails),
+        "id": Property(identifier=True),
+        "label": Property(),
+        "status": Property(),
+        "type": Property(),
+        "updated": Property(is_datetime=True),
+        "updated_by": Property(),
+        "version": Property(),
+    }
+
+
+class LogsDestination(Base):
+    """
+    Represents a logs destination object.
+
+    API documentation: https://techdocs.akamai.com/linode-api/reference/get-destination
+    """
+
+    api_endpoint = "/monitor/streams/destinations/{id}"
+
+    properties = {
+        "created": Property(is_datetime=True),
+        "created_by": Property(),
+        "details": Property(mutable=True, json_object=LogsDestinationDetails),
+        "id": Property(identifier=True),
+        "label": Property(mutable=True),
+        "status": Property(),
+        "type": Property(),
+        "updated": Property(is_datetime=True),
+        "updated_by": Property(),
+        "version": Property(),
+    }
+
+    @property
+    def history(self):
+        """
+        Retrieves the version history for this LogsDestination.
+
+        API documentation: https://techdocs.akamai.com/linode-api/reference/get-destination-history
+        """
+
+        return self._client._get_objects(
+            "{}/history".format(LogsDestination.api_endpoint.format(id=self.id)),
+            LogsDestinationHistory
+        )
+
+
+class LogsStreamStatus(StrEnum):
+    active = "active"
+    inactive = "inactive"
+    provisioning = "provisioning"
+
+
+class LogsStreamType(StrEnum):
+    audit_logs = "audit_logs"
+
+
+@dataclass
+class LogsStreamDestination(JSONObject):
+    """
+    Represents a destination attached to a LogsStream.
+    """
+
+    id: int = 0
+    label: str = ""
+    type: Optional[LogsDestinationType] = None
+    details: Optional[LogsDestinationDetails] = None
+
+
+class LogsStreamHistory(Base):
+    """
+    Represents a read-only historical snapshot of a logs stream.
+
+    API documentation: https://techdocs.akamai.com/linode-api/reference/get-stream-history
+    """
+
+    properties = {
+        "created": Property(is_datetime=True),
+        "created_by": Property(),
+        "destinations": Property(json_object=LogsStreamDestination),
+        "id": Property(identifier=True),
+        "label": Property(),
+        "status": Property(),
+        "type": Property(),
+        "updated": Property(is_datetime=True),
+        "updated_by": Property(),
+        "version": Property(),
+    }
+
+
+class LogsStream(Base):
+    """
+    Represents a logs stream object.
+
+    API documentation: https://techdocs.akamai.com/linode-api/reference/get-stream
+    """
+
+    api_endpoint = "/monitor/streams/{id}"
+
+    properties = {
+        "created": Property(is_datetime=True),
+        "created_by": Property(),
+        "destinations": Property(json_object=LogsStreamDestination),
+        "id": Property(identifier=True),
+        "label": Property(mutable=True),
+        "status": Property(mutable=True),
+        "type": Property(),
+        "updated": Property(is_datetime=True),
+        "updated_by": Property(),
+        "version": Property(),
+    }
+
+    def update_destinations(self, destinations: List[int]):
+        """
+        Updates the sync points that receive logs data for this stream.
+        Replaces existing destinations with the provided list.
+
+        :param destinations: A list of destination IDs.
+                            At the moment only single destination per stream is supported.
+                            Passing more than one element in the list will result in an error from the API.
+        :type destinations: list[int]
+
+        :returns: True if the update was successful.
+        :rtype: bool
+        """
+        if not destinations:
+            raise ValueError("A destination id must be provided.")
+        payload = {
+            "destinations": destinations
+        }
+
+        # The Linode API PUT request expects the flat list of IDs
+        result = self._client.put(
+            self.api_endpoint.format(id=self.id),
+            data=payload
+        )
+        self._populate(result)
+
+        return True
+
+    @property
+    def history(self):
+        """
+        Retrieves the version history for this LogsStream.
+
+        API documentation: https://techdocs.akamai.com/linode-api/reference/get-stream-history
+        """
+
+        return self._client._get_objects(
+            "{}/history".format(LogsStream.api_endpoint.format(id=self.id)),
+            LogsStreamHistory
+        )
diff --git a/test/fixtures/monitor_streams.json b/test/fixtures/monitor_streams.json
new file mode 100644
index 000000000..def47b365
--- /dev/null
+++ b/test/fixtures/monitor_streams.json
@@ -0,0 +1,31 @@
+{
+  "data": [
+    {
+      "id": 1,
+      "label": "my-logs-stream",
+      "type": "audit_logs",
+      "status": "active",
+      "destinations": [
+        {
+          "id": 1,
+          "label": "my-logs-destination",
+          "type": "akamai_object_storage",
+          "details": {
+            "access_key_id": "1ABCD23EFG4HIJKLMNO5",
+            "bucket_name": "primary-bucket",
+            "host": "primary-bucket.us-east-1.linodeobjects.com",
+            "path": "audit-logs"
+          }
+        }
+      ],
+      "created": "2024-06-01T12:00:00",
+      "updated": "2024-06-01T12:00:00",
+      "created_by": "tester",
+      "updated_by": "tester",
+      "version": 1
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 1
+}
diff --git a/test/fixtures/monitor_streams_1_history.json b/test/fixtures/monitor_streams_1_history.json
new file mode 100644
index 000000000..8f536303e
--- /dev/null
+++ b/test/fixtures/monitor_streams_1_history.json
@@ -0,0 +1,31 @@
+{
+  "data": [
+    {
+      "id": 1,
+      "label": "my-logs-stream",
+      "type": "audit_logs",
+      "status": "active",
+      "destinations": [
+        {
+          "id": 1,
+          "label": "my-logs-destination",
+          "type": "akamai_object_storage",
+          "details": {
+            "access_key_id": "1ABCD23EFG4HIJKLMNO5",
+            "bucket_name": "primary-bucket",
+            "host": "primary-bucket.us-east-1.linodeobjects.com",
+            "path": "audit-logs"
+          }
+        }
+      ],
+      "created": "2024-06-01T12:00:00",
+      "updated": "2024-06-02T09:00:00",
+      "created_by": "tester",
+      "updated_by": "tester",
+      "version": 2
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 1
+}
diff --git a/test/fixtures/monitor_streams_destinations.json b/test/fixtures/monitor_streams_destinations.json
new file mode 100644
index 000000000..0e1365e26
--- /dev/null
+++ b/test/fixtures/monitor_streams_destinations.json
@@ -0,0 +1,24 @@
+{
+  "data": [
+    {
+      "id": 1,
+      "label": "my-logs-destination",
+      "type": "akamai_object_storage",
+      "status": "active",
+      "details": {
+        "access_key_id": "1ABCD23EFG4HIJKLMNO5",
+        "bucket_name": "primary-bucket",
+        "host": "primary-bucket.us-east-1.linodeobjects.com",
+        "path": "audit-logs"
+      },
+      "created": "2024-06-01T12:00:00",
+      "updated": "2024-06-01T12:00:00",
+      "created_by": "tester",
+      "updated_by": "tester",
+      "version": 1
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 1
+}
diff --git a/test/fixtures/monitor_streams_destinations_1_history.json b/test/fixtures/monitor_streams_destinations_1_history.json
new file mode 100644
index 000000000..11f262c81
--- /dev/null
+++ b/test/fixtures/monitor_streams_destinations_1_history.json
@@ -0,0 +1,24 @@
+{
+  "data": [
+    {
+      "id": 1,
+      "label": "my-logs-destination",
+      "type": "akamai_object_storage",
+      "status": "active",
+      "details": {
+        "access_key_id": "1ABCD23EFG4HIJKLMNO5",
+        "bucket_name": "primary-bucket",
+        "host": "primary-bucket.us-east-1.linodeobjects.com",
+        "path": "audit-logs"
+      },
+      "created": "2024-06-01T12:00:00",
+      "updated": "2024-06-02T09:00:00",
+      "created_by": "tester",
+      "updated_by": "tester",
+      "version": 2
+    }
+  ],
+  "page": 1,
+  "pages": 1,
+  "results": 1
+}
diff --git a/test/integration/models/monitor/test_monitor_logs.py b/test/integration/models/monitor/test_monitor_logs.py
new file mode 100644
index 000000000..5f57d0219
--- /dev/null
+++ b/test/integration/models/monitor/test_monitor_logs.py
@@ -0,0 +1,417 @@
+import os
+import urllib.request
+
+import pytest
+
+from linode_api4 import LinodeClient, PaginatedList, LogsStreamType
+from linode_api4.objects import (ObjectStorageACL,
+                                 ObjectStorageKeys,
+                                 ObjectStorageBucket,
+                                 Capability)
+from linode_api4.objects.monitor import (
+    LogsDestination,
+    LogsStream,
+    LogsStreamStatus,
+)
+
+from test.integration.helpers import (
+    get_test_label,
+    send_request_when_resource_available,
+    wait_for_condition,
+)
+
+_RUN_ACLP_LOGS_STREAM_TESTS = "RUN_ACLP_LOGS_STREAM_TESTS"
+_SKIP_STREAM_TESTS = pytest.mark.skipif(
+    os.getenv(_RUN_ACLP_LOGS_STREAM_TESTS, "").strip().lower() not in {"yes", "true"},
+    reason=f"{_RUN_ACLP_LOGS_STREAM_TESTS} environment variable must be set to 'yes' or 'true'",
+)
+
+
+@pytest.fixture(scope="session", autouse=True)
+def require_aclp_logs(test_linode_client: LinodeClient):
+    """Skip all tests in this module if the aclp_logs feature is not enabled for the account."""
+    account = test_linode_client.account()
+    if Capability.aclp_logs not in account.capabilities:
+        pytest.skip("aclp_logs feature is not enabled for this account")
+
+
+@pytest.fixture(scope="session")
+def create_object_storage_key(test_linode_client: LinodeClient):
+    key = test_linode_client.object_storage.keys_create(
+        label=get_test_label(),
+    )
+    yield key
+    key.delete()
+
+
+@pytest.fixture(scope="session")
+def test_destination(
+        test_linode_client: LinodeClient,
+        create_object_storage_key: ObjectStorageKeys,
+):
+    dest, bucket = _create_destination_with_bucket(test_linode_client, create_object_storage_key)
+    yield dest
+    _delete_destination_with_bucket(test_linode_client, dest, bucket)
+
+
+def _create_destination_with_bucket(client: LinodeClient, key: ObjectStorageKeys):
+    """Helper that creates an OBJ bucket and a logs destination backed by it."""
+    bucket = client.object_storage.bucket_create(
+        cluster_or_region="us-southeast",
+        label=get_test_label(),
+        acl=ObjectStorageACL.PRIVATE,
+        cors_enabled=False,
+    )
+    dest = client.monitor.destination_create(
+        label=get_test_label(),
+        type="akamai_object_storage",
+        access_key_id=key.access_key,
+        access_key_secret=key.secret_key,
+        bucket_name=bucket.label,
+        host=f"{bucket.label}.us-southeast-1.linodeobjects.com",
+    )
+    return dest, bucket
+
+
+def _delete_destination_with_bucket(client: LinodeClient, dest: LogsDestination, bucket: ObjectStorageBucket):
+    """Helper that deletes a logs destination and its backing OBJ bucket."""
+    send_request_when_resource_available(timeout=100, func=dest.delete)
+    _empty_bucket(client, bucket)
+    send_request_when_resource_available(timeout=100, func=bucket.delete)
+
+
+def _skip_if_streams_exist(client: LinodeClient):
+    """Skip the current test if any streams already exist on the account.
+    Only one stream can be present per account at a time."""
+    existing_streams = client.monitor.streams()
+    if len(existing_streams) > 0:
+        stream_labels = [s.label for s in existing_streams]
+        pytest.skip(
+            f"Skipping: existing stream(s) found on this account "
+            f"(labels: {stream_labels}). Only one stream can be present per account."
+        )
+
+
+def _empty_bucket(client: LinodeClient, bucket: ObjectStorageBucket):
+    """
+    Helper function clearing objects in the test bucket so it can be deleted.
+    """
+    for obj in bucket.contents():
+        signed = client.object_storage.object_url_create(
+            cluster_or_region_id=bucket.region,
+            bucket=bucket.label,
+            method="DELETE",
+            name=obj.name,
+        )
+        urllib.request.urlopen(
+            urllib.request.Request(signed.url, method="DELETE")
+        )
+
+
+def test_list_destinations(test_linode_client: LinodeClient, test_destination: LogsDestination):
+    """
+    Test that listing destinations returns a PaginatedList containing the previously created destination.
+    """
+    destinations = test_linode_client.monitor.destinations()
+
+    assert isinstance(destinations, PaginatedList)
+    assert len(destinations) > 0
+    assert all(isinstance(d, LogsDestination) for d in destinations)
+
+    ids = [d.id for d in destinations]
+    assert test_destination.id in ids
+
+
+def test_get_destination_by_id(test_linode_client: LinodeClient, test_destination: LogsDestination):
+    """
+    Test that fetching destination with id filter returns correct destination.
+    """
+    destination_by_id = test_linode_client.load(LogsDestination, test_destination.id)
+
+    assert isinstance(destination_by_id, LogsDestination)
+    assert destination_by_id.id == test_destination.id
+    assert destination_by_id.label == test_destination.label
+    assert destination_by_id.type == test_destination.type
+
+
+def test_update_destination_label_and_version_history(
+        test_linode_client: LinodeClient,
+        test_destination: LogsDestination,
+        create_object_storage_key: ObjectStorageKeys,
+):
+    """
+    Test that a LogsDestination label can be updated via save(),
+    and that history reflects both states.
+    """
+    new_label = test_destination.label + "-upd"
+    new_path = "updated/logs/path/"
+
+    dest = test_linode_client.load(LogsDestination, test_destination.id)
+    original_version = dest.version
+    dest.label = new_label
+    dest.details.path = new_path
+    dest.details.access_key_secret = create_object_storage_key.secret_key
+    dest.save()
+
+    updated = test_linode_client.load(LogsDestination, test_destination.id)
+    assert updated.label == new_label
+    assert updated.details.path == new_path
+
+    history = updated.history
+    assert history is not None
+    assert len(history) >= 2
+
+    snapshot_original = next(snap for snap in history if snap.version == original_version)
+    snapshot_updated = next(snap for snap in history if snap.version == updated.version)
+
+    assert snapshot_updated.label == new_label
+    assert snapshot_updated.details.path == new_path
+    assert snapshot_updated.id == test_destination.id
+
+    assert snapshot_original.label == test_destination.label
+    assert snapshot_original.details.path is None
+    assert snapshot_original.id == test_destination.id
+
+
+def test_fails_to_create_destination_invalid_secret(test_linode_client: LinodeClient):
+    """
+    Test that a destination create request with invalid access key results in a 400 ApiError.
+    """
+    from linode_api4.errors import ApiError
+
+    with pytest.raises(ApiError) as excinfo:
+        test_linode_client.monitor.destination_create(
+            label=get_test_label(),
+            type="akamai_object_storage",
+            access_key_id="1",
+            access_key_secret="1",
+            bucket_name="some-bucket",
+            host="some-bucket.us-southeast-1.linodeobjects.com",
+        )
+    assert excinfo.value.status == 400
+    assert excinfo.value.errors == ['Invalid access key id or secret key']
+
+
+def test_fails_to_create_destination_invalid_type(test_linode_client: LinodeClient):
+    """
+    Test that a destination create request with an unsupported type
+    results in a 400 ApiError.
+    """
+    from linode_api4.errors import ApiError
+
+    with pytest.raises(ApiError) as excinfo:
+        test_linode_client.monitor.destination_create(
+            label=get_test_label(),
+            type="invalid_type",
+            access_key_id="SOMEACCESSKEY",
+            access_key_secret="SOMESECRETKEY",
+            bucket_name="some-bucket",
+            host="some-bucket.us-southeast-1.linodeobjects.com",
+        )
+    assert excinfo.value.status == 400
+    assert excinfo.value.errors == ['Must be one of akamai_object_storage, custom_https']
+
+
+def test_fails_to_create_destination_empty_required_fields(test_linode_client: LinodeClient):
+    """
+    Test that a destination create request with missing required fields
+    results in a 400 ApiError.
+    """
+    from linode_api4.errors import ApiError
+
+    with pytest.raises(ApiError) as excinfo:
+        test_linode_client.monitor.destination_create(
+            label=get_test_label(),
+            type="akamai_object_storage",
+            access_key_id="",
+            access_key_secret="",
+            bucket_name="",
+            host="",
+        )
+    assert excinfo.value.status == 400
+    assert len(excinfo.value.errors) == 4
+    assert all(
+        error == "Length must be 1-255 characters"
+        for error in excinfo.value.errors
+    )
+
+
+@pytest.fixture(scope="session")
+def invalid_destination_error(test_linode_client: LinodeClient):
+    """
+    Session-scoped fixture to attempt invalid stream creation deterministically
+    before any valid streams are created. Yields the resulting exception so
+    assertions can be handled safely within the test case.
+    """
+    from linode_api4.errors import ApiError
+
+    _skip_if_streams_exist(test_linode_client)
+
+    try:
+        test_linode_client.monitor.stream_create(
+            label=get_test_label(),
+            type=LogsStreamType.audit_logs,
+            destinations=[999999999],
+        )
+        yield None
+    except ApiError as excinfo:
+        yield excinfo
+
+@_SKIP_STREAM_TESTS
+def test_fails_to_create_stream_invalid_destination(invalid_destination_error):
+    """
+    Test that creating a stream with a non-existent destination ID results in a 400 ApiError.
+    Requires no other streams to be present on account.
+    """
+    assert invalid_destination_error is not None, "Expected an ApiError but none was raised"
+
+    assert invalid_destination_error.status == 400
+    assert invalid_destination_error.errors == ['Destination not found']
+
+
+@pytest.fixture(scope="session")
+def create_secondary_destination(
+        test_linode_client: LinodeClient,
+        create_object_storage_key: ObjectStorageKeys,
+):
+    dest, bucket = _create_destination_with_bucket(test_linode_client, create_object_storage_key)
+    yield dest
+    _delete_destination_with_bucket(test_linode_client, dest, bucket)
+
+
+@pytest.fixture(scope="session")
+def create_stream(test_linode_client: LinodeClient,
+                  test_destination: LogsDestination,
+                  invalid_destination_error #This ensures run order to keep negative test case deterministic
+):
+    _skip_if_streams_exist(test_linode_client)
+
+    stream = test_linode_client.monitor.stream_create(
+        label=get_test_label(),
+        destinations=[test_destination.id],
+        type=LogsStreamType.audit_logs
+    )
+    assert stream.id is not None
+    assert stream.status == LogsStreamStatus.provisioning
+    yield stream
+    send_request_when_resource_available(timeout=100, func=stream.delete)
+
+
+@pytest.fixture(scope="session")
+def provisioned_stream(test_linode_client: LinodeClient, create_stream: LogsStream):
+    """
+    Waits until the stream transitions out of provisioning state.
+    NOTE: Stream provisioning can take up to 60 minutes to finish.
+    """
+
+    def is_stream_provisioned():
+        stream = test_linode_client.load(LogsStream, create_stream.id)
+        return stream.status in (LogsStreamStatus.active, LogsStreamStatus.inactive)
+
+    wait_for_condition(60, 3600, is_stream_provisioned)
+
+    yield test_linode_client.load(LogsStream, create_stream.id)
+
+
+@_SKIP_STREAM_TESTS
+def test_list_streams(test_linode_client: LinodeClient, provisioned_stream: LogsStream):
+    """
+    Test that listing streams returns a PaginatedList containing the previously created stream.
+    """
+    streams = test_linode_client.monitor.streams()
+
+    assert isinstance(streams, PaginatedList)
+    assert len(streams) > 0
+    assert all(isinstance(s, LogsStream) for s in streams)
+
+    ids = [s.id for s in streams]
+    assert provisioned_stream.id in ids
+
+
+@_SKIP_STREAM_TESTS
+def test_get_stream_by_id(test_linode_client: LinodeClient, provisioned_stream: LogsStream):
+    """
+    Test that loading a stream by ID returns the correct stream with expected fields.
+    """
+    stream = test_linode_client.load(LogsStream, provisioned_stream.id)
+
+    assert isinstance(stream, LogsStream)
+    assert stream.id == provisioned_stream.id
+    assert stream.label == provisioned_stream.label
+    assert stream.status == provisioned_stream.status
+    assert len(stream.destinations) == 1
+
+
+@_SKIP_STREAM_TESTS
+def test_update_stream_label_and_status(test_linode_client: LinodeClient, provisioned_stream: LogsStream):
+    """
+    Test that a LogsStream label and status can both be updated via save(), and that
+    the version history reflects label changes across versions.
+    """
+    stream = test_linode_client.load(LogsStream, provisioned_stream.id)
+    original_label = stream.label
+    original_status = stream.status
+    version_before = stream.version
+
+    new_label = original_label + "-upd"
+    new_status = (
+        LogsStreamStatus.inactive
+        if original_status == LogsStreamStatus.active
+        else LogsStreamStatus.active
+    )
+
+    stream.label = new_label
+    stream.status = new_status
+    result = stream.save()
+    assert result is True
+
+    try:
+        updated = test_linode_client.load(LogsStream, provisioned_stream.id)
+        assert updated.label == new_label
+        assert updated.status == new_status
+
+        history = updated.history
+        snapshot_original = next(h for h in history if h.version == version_before)
+        snapshot_updated = next(h for h in history if h.version == updated.version)
+
+        assert snapshot_original.label == original_label
+        assert snapshot_updated.label == new_label
+        assert snapshot_updated.id == provisioned_stream.id
+    finally:
+        # Revert to original label and status
+        stream.label = original_label
+        stream.status = original_status
+        stream.save()
+
+
+@_SKIP_STREAM_TESTS
+def test_update_stream_destinations(
+        test_linode_client: LinodeClient,
+        provisioned_stream: LogsStream,
+        create_secondary_destination: LogsDestination,
+):
+    """
+    Test that a stream destination can be replaced via update_destinations(),
+    and that history reflects the change. The API allows exactly one destination per stream.
+    """
+    stream = test_linode_client.load(LogsStream, provisioned_stream.id)
+    original_destinations = [stream.destinations[0].id]
+    version_before = stream.version
+
+    result = stream.update_destinations([create_secondary_destination.id])
+    assert result is True
+
+    try:
+        updated = test_linode_client.load(LogsStream, provisioned_stream.id)
+        assert len(updated.destinations) == 1
+        assert updated.destinations[0].id == create_secondary_destination.id
+
+        history = updated.history
+        snapshot_original = next(h for h in history if h.version == version_before)
+        snapshot_updated = next(h for h in history if h.version == updated.version)
+
+        assert snapshot_original.destinations[0].id == original_destinations[0]
+        assert snapshot_updated.destinations[0].id == create_secondary_destination.id
+    finally:
+        # Revert to original destination
+        stream.update_destinations(original_destinations)
diff --git a/test/unit/objects/monitor_test.py b/test/unit/objects/monitor_test.py
index 5913b3b28..79e98a471 100644
--- a/test/unit/objects/monitor_test.py
+++ b/test/unit/objects/monitor_test.py
@@ -1,7 +1,15 @@
 import datetime
 from test.unit.base import ClientBaseCase
 
-from linode_api4.objects import AlertChannel, MonitorDashboard, MonitorService
+from linode_api4.objects import (
+    AlertChannel,
+    MonitorDashboard,
+    MonitorService,
+    LogsDestination,
+    LogsDestinationHistory,
+    LogsStream,
+    LogsStreamDestination,
+)
 
 
 class MonitorTest(ClientBaseCase):
@@ -169,3 +177,327 @@ def test_alert_channels(self):
             "/monitor/alert-channels/123/alerts",
         )
         self.assertEqual(channels[0].alerts.alert_count, 0)
+
+
+class LogsDestinationTest(ClientBaseCase):
+    """
+    Tests methods for LogsDestination class
+    """
+
+    def test_list_destinations(self):
+        """
+        Test that listing destinations returns LogsDestination objects with all fields populated.
+        """
+        destinations = self.client.monitor.destinations()
+
+        self.assertEqual(len(destinations), 1)
+        dest = destinations[0]
+        self.assertIsInstance(dest, LogsDestination)
+        self.assertEqual(dest.id, 1)
+        self.assertEqual(dest.label, "my-logs-destination")
+        self.assertEqual(dest.type, "akamai_object_storage")
+        self.assertEqual(dest.status, "active")
+        self.assertEqual(dest.version, 1)
+        self.assertEqual(
+            dest.created, datetime.datetime(2024, 6, 1, 12, 0, 0)
+        )
+        self.assertEqual(
+            dest.updated, datetime.datetime(2024, 6, 1, 12, 0, 0)
+        )
+        self.assertEqual(dest.created_by, "tester")
+        self.assertEqual(dest.updated_by, "tester")
+
+    def test_list_destinations_details(self):
+        """
+        Test that the nested LogsDestinationDetails are deserialized correctly.
+        """
+        dest = self.client.load(LogsDestination, 1)
+
+        self.assertIsNotNone(dest.details)
+        self.assertEqual(dest.details.access_key_id, "1ABCD23EFG4HIJKLMNO5")
+        self.assertEqual(dest.details.bucket_name, "primary-bucket")
+        self.assertEqual(
+            dest.details.host, "primary-bucket.us-east-1.linodeobjects.com"
+        )
+        self.assertEqual(dest.details.path, "audit-logs")
+
+        self.assertIsNone(dest.details.access_key_secret)
+
+    def test_destination_history(self):
+        """
+        Test that the history property returns LogsDestinationHistory objects.
+        """
+        dest = self.client.load(LogsDestination, 1)
+        history = dest.history
+
+        self.assertEqual(len(history), 1)
+        snapshot = history[0]
+        self.assertIsInstance(snapshot, LogsDestinationHistory)
+        self.assertEqual(snapshot.id, 1)
+        self.assertEqual(snapshot.label, "my-logs-destination")
+        self.assertEqual(snapshot.type, "akamai_object_storage")
+        self.assertEqual(snapshot.status, "active")
+        self.assertEqual(snapshot.version, 2)
+        self.assertEqual(
+            snapshot.updated, datetime.datetime(2024, 6, 2, 9, 0, 0)
+        )
+        self.assertIsNotNone(snapshot.details)
+        self.assertEqual(snapshot.details.bucket_name, "primary-bucket")
+
+    def test_create_destination(self):
+        """
+        Test that destination_create sends the right payload and returns
+        a LogsDestination object.
+        """
+        create_response = {
+            "id": 2,
+            "label": "new-dest",
+            "type": "akamai_object_storage",
+            "status": "active",
+            "details": {
+                "access_key_id": "KEYID999",
+                "bucket_name": "new-bucket",
+                "host": "new-bucket.us-east-1.linodeobjects.com",
+                "path": "logs/audit",
+            },
+            "created": "2024-07-01T00:00:00",
+            "updated": "2024-07-01T00:00:00",
+            "created_by": "tester",
+            "updated_by": "tester",
+            "version": 1,
+        }
+
+        with self.mock_post(create_response) as m:
+            result = self.client.monitor.destination_create(
+                label="new-dest",
+                type="akamai_object_storage",
+                access_key_id="KEYID999",
+                access_key_secret="SUPERSECRET",
+                bucket_name="new-bucket",
+                host="new-bucket.us-east-1.linodeobjects.com",
+                path="logs/audit",
+            )
+
+        self.assertEqual(m.call_url, "/monitor/streams/destinations")
+        self.assertEqual(m.call_data["label"], "new-dest")
+        self.assertEqual(m.call_data["type"], "akamai_object_storage")
+        self.assertEqual(m.call_data["details"]["access_key_id"], "KEYID999")
+        self.assertEqual(
+            m.call_data["details"]["access_key_secret"], "SUPERSECRET"
+        )
+        self.assertEqual(m.call_data["details"]["bucket_name"], "new-bucket")
+        self.assertEqual(
+            m.call_data["details"]["host"],
+            "new-bucket.us-east-1.linodeobjects.com",
+        )
+        self.assertEqual(m.call_data["details"]["path"], "logs/audit")
+
+        self.assertIsInstance(result, LogsDestination)
+        self.assertEqual(result.id, 2)
+        self.assertEqual(result.label, "new-dest")
+
+    def test_update_destination(self):
+        """
+        Test that mutating a LogsDestination's mutable fields and calling save()
+        sends a PUT to the correct endpoint with the updated values.
+        """
+        dest = self.client.load(LogsDestination, 1)
+
+        updated_response = {
+            "id": 1,
+            "label": "renamed-destination",
+            "type": "akamai_object_storage",
+            "status": "active",
+            "details": {
+                "access_key_id": "1ABCD23EFG4HIJKLMNO5",
+                "bucket_name": "primary-bucket",
+                "host": "primary-bucket.us-east-1.linodeobjects.com",
+                "path": "audit-logs",
+            },
+            "created": "2024-06-01T12:00:00",
+            "updated": "2024-06-03T08:00:00",
+            "created_by": "tester",
+            "updated_by": "tester",
+            "version": 2,
+        }
+
+        with self.mock_put(updated_response) as m:
+            dest.label = "renamed-destination"
+            dest.save()
+
+        self.assertEqual(m.call_url, "/monitor/streams/destinations/1")
+        self.assertEqual(m.call_data["label"], "renamed-destination")
+
+    def test_delete_destination(self):
+        """
+        Test that deleting a LogsDestination issues a DELETE to the correct URL.
+        """
+        dest = self.client.load(LogsDestination, 1)
+
+        with self.mock_delete() as m:
+            dest.delete()
+
+        self.assertEqual(
+            m.call_url, "/monitor/streams/destinations/1"
+        )
+
+
+class LogsStreamTest(ClientBaseCase):
+    """
+    Tests methods for LogsStream class.
+    """
+
+    def test_list_streams(self):
+        """
+        Test that listing streams returns LogsStream objects with all fields populated.
+        """
+        streams = self.client.monitor.streams()
+
+        self.assertEqual(len(streams), 1)
+        stream = streams[0]
+        self.assertIsInstance(stream, LogsStream)
+        self.assertEqual(stream.id, 1)
+        self.assertEqual(stream.label, "my-logs-stream")
+        self.assertEqual(stream.type, "audit_logs")
+        self.assertEqual(stream.status, "active")
+        self.assertEqual(stream.version, 1)
+        self.assertEqual(stream.created, datetime.datetime(2024, 6, 1, 12, 0, 0))
+        self.assertEqual(stream.updated, datetime.datetime(2024, 6, 1, 12, 0, 0))
+        self.assertEqual(stream.created_by, "tester")
+        self.assertEqual(stream.updated_by, "tester")
+
+    def test_list_streams_destinations(self):
+        """
+        Test that the nested destinations are deserialized as LogsStreamDestination objects.
+        """
+        stream = self.client.load(LogsStream, 1)
+
+        self.assertIsNotNone(stream.destinations)
+        self.assertEqual(len(stream.destinations), 1)
+        dest = stream.destinations[0]
+        self.assertIsInstance(dest, LogsStreamDestination)
+        self.assertEqual(dest.id, 1)
+        self.assertEqual(dest.label, "my-logs-destination")
+        self.assertEqual(dest.type, "akamai_object_storage")
+        self.assertIsNotNone(dest.details)
+        self.assertEqual(dest.details.bucket_name, "primary-bucket")
+        self.assertEqual(dest.details.access_key_id, "1ABCD23EFG4HIJKLMNO5")
+        self.assertEqual(dest.details.host, "primary-bucket.us-east-1.linodeobjects.com")
+        self.assertEqual(dest.details.path, "audit-logs")
+
+    def test_stream_history(self):
+        """
+        Test that the history property returns LogsStreamHistory objects.
+        """
+        stream = self.client.load(LogsStream, 1)
+        history = stream.history
+
+        self.assertEqual(len(history), 1)
+        snapshot = history[0]
+        self.assertEqual(snapshot.id, 1)
+        self.assertEqual(snapshot.label, "my-logs-stream")
+        self.assertEqual(snapshot.type, "audit_logs")
+        self.assertEqual(snapshot.status, "active")
+        self.assertEqual(snapshot.version, 2)
+        self.assertEqual(snapshot.updated, datetime.datetime(2024, 6, 2, 9, 0, 0))
+        self.assertIsNotNone(snapshot.destinations)
+
+    def test_create_stream(self):
+        """
+        Test that stream_create sends the correct payload and returns a LogsStream object.
+        """
+        create_response = {
+            "id": 2,
+            "label": "new-stream",
+            "type": "audit_logs",
+            "status": "active",
+            "destinations": [{"id": 1, "label": "my-logs-destination", "type": "akamai_object_storage", "details": {}}],
+            "created": "2024-07-01T00:00:00",
+            "updated": "2024-07-01T00:00:00",
+            "created_by": "tester",
+            "updated_by": "tester",
+            "version": 1,
+        }
+
+        with self.mock_post(create_response) as m:
+            result = self.client.monitor.stream_create(
+                destinations=[1],
+                label="new-stream",
+                status="active",
+                type="audit_logs",
+            )
+
+        self.assertEqual(m.call_url, "/monitor/streams")
+        self.assertEqual(m.call_data["label"], "new-stream")
+        self.assertEqual(m.call_data["type"], "audit_logs")
+        self.assertEqual(m.call_data["status"], "active")
+        self.assertEqual(m.call_data["destinations"], [1])
+
+        self.assertIsInstance(result, LogsStream)
+        self.assertEqual(result.id, 2)
+        self.assertEqual(result.label, "new-stream")
+
+    def test_update_stream_save(self):
+        """
+        Test that mutating a LogsStream's mutable fields and calling save()
+        sends a PUT with correct payload.
+        """
+        stream = self.client.load(LogsStream, 1)
+
+        updated_response = {
+            "id": 1,
+            "label": "renamed-stream",
+            "type": "audit_logs",
+            "status": "inactive",
+            "destinations": [{"id": 1, "label": "my-logs-destination", "type": "akamai_object_storage", "details": {}}],
+            "created": "2024-06-01T12:00:00",
+            "updated": "2024-06-03T08:00:00",
+            "created_by": "tester",
+            "updated_by": "tester",
+            "version": 2,
+        }
+
+        with self.mock_put(updated_response) as m:
+            stream.label = "renamed-stream"
+            stream.status = "inactive"
+            stream.save()
+
+        self.assertEqual(m.call_url, "/monitor/streams/1")
+        self.assertEqual(m.call_data["label"], "renamed-stream")
+        self.assertEqual(m.call_data["status"], "inactive")
+
+    def test_update_stream_destinations(self):
+        """
+        Test that update_destinations sends PUT request with flat destination ids list.
+        """
+        stream = self.client.load(LogsStream, 1)
+
+        with self.mock_put({}) as m:
+            result = stream.update_destinations([1])
+
+        self.assertEqual(m.call_url, "/monitor/streams/1")
+        self.assertEqual(m.call_data["destinations"], [1])
+        self.assertTrue(result)
+
+    def test_fail_update_stream_destinations_when_no_destination_ids_passed(self):
+        """
+        Test that update_destinations raises exception and doesn't send PUT request when id list is empty.
+        """
+        stream = self.client.load(LogsStream, 1)
+        with self.mock_put({}) as m:
+            with self.assertRaises(ValueError) as context:
+                stream.update_destinations([])
+
+        self.assertFalse(m.called)
+        self.assertIn("A destination id must be provided.", str(context.exception))
+
+    def test_delete_stream(self):
+        """
+        Test that deleting a LogsStream issues a DELETE to the correct URL.
+        """
+        stream = self.client.load(LogsStream, 1)
+
+        with self.mock_delete() as m:
+            stream.delete()
+
+        self.assertEqual(m.call_url, "/monitor/streams/1")