Repository: localstack/localstack
Pull request: feat: DynamoDB TTL
<!-- Please refer to the contribution guidelines before raising a PR: https://github.com/localstack/localstack/blob/master/CONTRIBUTING.md -->

<!-- Why am I raising this PR? Add context such as related issues, PRs, or documentation. -->
## Motivation
This PR implements the Time to Live feature in DynamoDB, as reported in #10022.


<!-- What notable changes does this PR make? -->
## Changes
- We added a `ExpiredItemsWorker` class in the DDB provider that schedules a thread removing the stale keys across all the tables that define an enabled TTL specification. It uses a `Scheduler` under the hood and runs every hour. We could make the period interval parameterized.
- We added the `DELETE /_aws/dynamodb/expired` endpoint that programmatically triggers the stale items' deletion.

### Misc
- to implement the deletion, we scan every table with the `Scan` operation, collecting the stale items.
- we use `BatchItemWrite` to delete the collected stale items from the DBs. The call should consume write throughput, while TTL does not.
- an alternative would be to mark the expired element, instead of deleting them, and exclude them from all the relevant operations.

<!-- The following sections are optional, but can be useful! 

## Testing

Description of how to test the changes

## TODO

What's left to do:

- [ ] ...
- [ ] ...

-->



Problem statement:
DynamoDB TTL not implemented
I'm trying to use the recent time-to-live feature in DynamoDB, but I get a `An unknown operation was requested` exception. My guess is that the feature is not yet implemented in Localstack. Are there any plans for including it?

Hints from issue discussion:
Thanks for reporting @justinian336 , this would indeed be a useful feature. Would you be able to work on a pull request? I'd be happy to provide the general direction, and help out where needed. Thanks
Sure, I'd be glad to help. I'll study your code and make a PR when I'm ready. Do you have any guidelines for contributors?
Awesome. Please refer to the README for some basic guidelines: https://github.com/localstack/localstack#contributing
A few notes have been added to the developer docs, but this is currently only a skeleton (work in progress): https://github.com/localstack/localstack/tree/master/doc/developer_guides

Generally, `ProxyListenerDynamoDB.return_response` in [dynamodb_listener.py](https://github.com/localstack/localstack/blob/master/localstack/services/dynamodb/dynamodb_listener.py#L32) would be a good starting point. We can keep a list of TTL timestamps there, and clean up documents when needed.
@whummer Got it, I'll try and let you know if I find some trouble.
Any updates here?
Initial support for TTL added in #599
@whummer could you please clarify, is fully-functional DynamoDB TTL support done or it's just a mock to avoid getting errors on setting TTL settings? Thank you!
Thanks for the follow-up @brainoutsource . We still have this line in the listener code, so as of now the TTL response is simply mocked, without actually removing the items.
https://github.com/localstack/localstack/blob/8729ea45f56c9b61990db187dc0c4c73bb3b5416/localstack/services/dynamodb/dynamodb_listener.py#L119

I guess it shouldn't be too hard to add that functionality, though (check for expiry when a Scan/GetItem operation occurs, and then delete any pending expired items). Do you think you could try to contribute a pull request to add that feature? That would be much appreciated, thanks!
I see that the current master is still lacking an implementation, maybe someone could add it? Please :)
https://github.com/localstack/localstack/blob/master/localstack/services/dynamodb/dynamodb_listener.py#L219
@whummer you closed this issue, but maybe you could re-evaluate it and open it again, as it would be great if the ttl expire actually removed the document.

Commit message: consider max batch size

Patch to review:
diff --git a/localstack/services/dynamodb/models.py b/localstack/services/dynamodb/models.py
index e128540690891..ec9e1cc433a1b 100644
--- a/localstack/services/dynamodb/models.py
+++ b/localstack/services/dynamodb/models.py
@@ -1,4 +1,9 @@
-from localstack.aws.api.dynamodb import RegionName, ReplicaDescription, TableName
+from localstack.aws.api.dynamodb import (
+    RegionName,
+    ReplicaDescription,
+    TableName,
+    TimeToLiveSpecification,
+)
 from localstack.services.stores import (
     AccountRegionBundle,
     BaseStore,
@@ -29,7 +34,7 @@ class DynamoDBStore(BaseStore):
     table_properties: dict[str, dict] = LocalAttribute(default=dict)
 
     # maps table names to TTL specifications
-    ttl_specifications: dict[str, dict] = LocalAttribute(default=dict)
+    ttl_specifications: dict[str, TimeToLiveSpecification] = LocalAttribute(default=dict)
 
     # maps backups
     backups: dict[str, dict] = LocalAttribute(default=dict)
diff --git a/localstack/services/dynamodb/provider.py b/localstack/services/dynamodb/provider.py
index e928ed02d6c49..6bdbd992f94bc 100644
--- a/localstack/services/dynamodb/provider.py
+++ b/localstack/services/dynamodb/provider.py
@@ -4,12 +4,14 @@
 import os
 import random
 import re
+import threading
 import time
 import traceback
 from binascii import crc32
 from contextlib import contextmanager
+from datetime import datetime
 from operator import itemgetter
-from typing import Dict, List
+from typing import Dict, List, Optional
 
 import requests
 import werkzeug
@@ -105,7 +107,7 @@
     AWS_REGION_US_EAST_1,
     TEST_AWS_SECRET_ACCESS_KEY,
 )
-from localstack.http import Response
+from localstack.http import Request, Response, route
 from localstack.services.dynamodb.models import DynamoDBStore, dynamodb_stores
 from localstack.services.dynamodb.server import DynamodbServer
 from localstack.services.dynamodb.utils import (
@@ -132,8 +134,9 @@
 from localstack.utils.collections import select_attributes, select_from_typed_dict
 from localstack.utils.common import short_uid, to_bytes
 from localstack.utils.json import BytesEncoder, canonical_json
+from localstack.utils.scheduler import Scheduler
 from localstack.utils.strings import long_uid, to_str
-from localstack.utils.threads import start_worker_thread
+from localstack.utils.threads import FuncThread, start_thread, start_worker_thread
 
 # set up logger
 LOG = logging.getLogger(__name__)
@@ -356,15 +359,123 @@ def modify_context_region(context: RequestContext, region: str):
         context.request.headers["Authorization"] = original_authorization
 
 
+class DynamoDBDeveloperEndpoints:
+    """
+    Developer endpoints for DynamoDB
+    DELETE /_aws/dynamodb/expired - delete expired items from tables with TTL enabled; return the number of expired
+        items deleted
+    """
+
+    @route("/_aws/dynamodb/expired", methods=["DELETE"])
+    def delete_expired_messages(self, _: Request):
+        no_expired_items = delete_expired_items()
+        return {"ExpiredItems": no_expired_items}
+
+
+def delete_expired_items() -> int:
+    """
+    This utility function iterates over all stores, looks for tables with TTL enabled,
+    scan such tables and delete expired items.
+    """
+    no_expired_items = 0
+
+    for account_id, account_bundle in dynamodb_stores.items():
+        for region_name, state in account_bundle.items():
+            ttl_specs = state.ttl_specifications
+            for table_name, ttl_spec in ttl_specs.items():
+                if ttl_spec.get("Enabled", False):
+                    attribute_name = ttl_spec.get("AttributeName")
+                    client = connect_to(region_name=region_name).dynamodb
+                    current_time = int(datetime.now().timestamp())
+                    try:
+                        result = client.scan(
+                            TableName=table_name,
+                            FilterExpression=f"{attribute_name} <= :threshold",
+                            ExpressionAttributeValues={":threshold": {"N": str(current_time)}},
+                        )
+                        items_to_delete = result.get("Items", [])
+                        no_expired_items += len(items_to_delete)
+                        table_description = client.describe_table(TableName=table_name)
+                        partition_key = _get_partition_key(table_description)
+                        keys_to_delete = [
+                            {partition_key: item.get(partition_key)} for item in items_to_delete
+                        ]
+                        delete_requests = [
+                            {"DeleteRequest": {"Key": key}} for key in keys_to_delete
+                        ]
+                        for i in range(0, len(delete_requests), 25):
+                            batch = delete_requests[i : i + 25]
+                            client.batch_write_item(RequestItems={table_name: batch})
+                    except Exception as e:
+                        LOG.warning(
+                            "An error occurred when deleting expired items from table %s: %s",
+                            table_name,
+                            e,
+                        )
+    return no_expired_items
+
+
+def _get_partition_key(table_description: DescribeTableOutput) -> str:
+    key_schema = table_description.get("Table", {}).get("KeySchema", [])
+    for key in key_schema:
+        if key["KeyType"] == "HASH":
+            return key["AttributeName"]
+
+
+class ExpiredItemsWorker:
+    """A worker that periodically computes and deletes expired items from DynamoDB tables"""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.scheduler = Scheduler()
+        self.thread: Optional[FuncThread] = None
+        self.mutex = threading.RLock()
+
+    def start(self):
+        with self.mutex:
+            if self.thread:
+                return
+
+            self.scheduler = Scheduler()
+            self.scheduler.schedule(
+                delete_expired_items, period=60 * 60
+            )  # the background process seems slow on AWS
+
+            def _run(*_args):
+                self.scheduler.run()
+
+            self.thread = start_thread(_run, name="ddb-remove-expired-items")
+
+    def stop(self):
+        with self.mutex:
+            if self.scheduler:
+                self.scheduler.close()
+
+            if self.thread:
+                self.thread.stop()
+
+            self.thread = None
+            self.scheduler = None
+
+
 class DynamoDBProvider(DynamodbApi, ServiceLifecycleHook):
     server: DynamodbServer
     """The instance of the server managing the instance of DynamoDB local"""
 
     def __init__(self):
         self.server = self._new_dynamodb_server()
+        self._expired_items_worker = ExpiredItemsWorker()
+        self._router_rules = []
 
     def on_before_start(self):
         self.server.start_dynamodb()
+        self._expired_items_worker.start()
+        self._router_rules = ROUTER.add(DynamoDBDeveloperEndpoints())
+
+    def on_before_stop(self):
+        self._expired_items_worker.stop()
+        for rule in self._router_rules:
+            ROUTER.remove_rule(rule)
 
     def accept_state_visitor(self, visitor: StateVisitor):
         visitor.visit(dynamodb_stores)
diff --git a/tests/aws/services/dynamodb/test_dynamodb.py b/tests/aws/services/dynamodb/test_dynamodb.py
index 3f984cc2dc700..ceb147da508a5 100644
--- a/tests/aws/services/dynamodb/test_dynamodb.py
+++ b/tests/aws/services/dynamodb/test_dynamodb.py
@@ -1,14 +1,17 @@
 import json
 import re
+import time
 from datetime import datetime
 from time import sleep
 from typing import Dict
 
 import botocore.exceptions
 import pytest
+import requests
 from boto3.dynamodb.types import STRING
 from botocore.exceptions import ClientError
 
+from localstack import config
 from localstack.aws.api.dynamodb import PointInTimeRecoverySpecification
 from localstack.constants import AWS_REGION_US_EAST_1, TEST_AWS_ACCOUNT_ID, TEST_AWS_REGION_NAME
 from localstack.services.dynamodbstreams.dynamodbstreams_api import get_kinesis_stream_name
@@ -102,9 +105,40 @@ def test_large_data_download(self, aws_client, ddb_test_table):
         result = aws_client.dynamodb.scan(TableName=ddb_test_table)
         assert len(result["Items"]) == num_items
 
+    @markers.aws.only_localstack
+    def test_time_to_live_deletion(self, aws_client, ddb_test_table):
+        table_name = ddb_test_table
+        aws_client.dynamodb.update_time_to_live(
+            TableName=table_name,
+            TimeToLiveSpecification={"Enabled": True, "AttributeName": "expiringAt"},
+        )
+        aws_client.dynamodb.describe_time_to_live(TableName=table_name)
+
+        exp = int(time.time()) - 10  # expired
+        items = [
+            {PARTITION_KEY: {"S": "expired"}, "expiringAt": {"N": str(exp)}},
+            {PARTITION_KEY: {"S": "not-expired"}, "expiringAt": {"N": str(exp + 120)}},
+        ]
+        for item in items:
+            aws_client.dynamodb.put_item(TableName=table_name, Item=item)
+
+        url = f"{config.internal_service_url()}/_aws/dynamodb/expired"
+        response = requests.delete(url)
+        assert response.status_code == 200
+        assert response.json() == {"ExpiredItems": 1}
+
+        result = aws_client.dynamodb.get_item(
+            TableName=table_name, Key={PARTITION_KEY: {"S": "not-expired"}}
+        )
+        assert result.get("Item")
+        result = aws_client.dynamodb.get_item(
+            TableName=table_name, Key={PARTITION_KEY: {"S": "expired"}}
+        )
+        assert not result.get("Item")
+
     @markers.aws.only_localstack
     def test_time_to_live(self, aws_client, ddb_test_table):
-        # check response for inexistent table
+        # check response for nonexistent table
         response = testutil.send_describe_dynamodb_ttl_request("test")
         assert json.loads(response._content)["__type"] == "ResourceNotFoundException"
         assert response.status_code == 400


Code review comments:
diff_hunk: @@ -356,15 +359,123 @@ def modify_context_region(context: RequestContext, region: str):
         context.request.headers["Authorization"] = original_authorization
 
 
+class DynamoDBDeveloperEndpoints:
+    """
+    Developer endpoints for DynamoDB
+    DELETE /_aws/dynamodb/expired - delete expired items from tables with TTL enabled; return the number of expired
+        items deleted
+    """
+
+    @route("/_aws/dynamodb/expired", methods=["DELETE"])
+    def delete_expired_messages(self, _: Request):
+        no_expired_items = delete_expired_items()
+        return {"ExpiredItems": no_expired_items}
+
+
+def delete_expired_items() -> int:
+    """
+    This utility function iterates over all stores, looks for tables with TTL enabled,
+    scan such tables and delete expired items.
+    """
+    no_expired_items = 0
+
+    for account_id, account_bundle in dynamodb_stores.items():
+        for region_name, state in account_bundle.items():
path: localstack/services/dynamodb/provider.py
text: @user1:
```suggestion
    for account_id, region_name, state in dynamodb_stores.iter_stores():
```

diff_hunk: @@ -356,15 +359,123 @@ def modify_context_region(context: RequestContext, region: str):
         context.request.headers["Authorization"] = original_authorization
 
 
+class DynamoDBDeveloperEndpoints:
+    """
+    Developer endpoints for DynamoDB
+    DELETE /_aws/dynamodb/expired - delete expired items from tables with TTL enabled; return the number of expired
+        items deleted
+    """
+
+    @route("/_aws/dynamodb/expired", methods=["DELETE"])
+    def delete_expired_messages(self, _: Request):
+        no_expired_items = delete_expired_items()
+        return {"ExpiredItems": no_expired_items}
+
+
+def delete_expired_items() -> int:
+    """
+    This utility function iterates over all stores, looks for tables with TTL enabled,
+    scan such tables and delete expired items.
+    """
+    no_expired_items = 0
+
+    for account_id, account_bundle in dynamodb_stores.items():
+        for region_name, state in account_bundle.items():
+            ttl_specs = state.ttl_specifications
+            for table_name, ttl_spec in ttl_specs.items():
+                if ttl_spec.get("Enabled", False):
+                    attribute_name = ttl_spec.get("AttributeName")
+                    client = connect_to(region_name=region_name).dynamodb
path: localstack/services/dynamodb/provider.py
text: @user1:
can we move the client connection to the outer loop?