Ingester: Updating storage store_latest, tests and conftest.

Author: ABPLMC

ingester/datalake_ingester/storage.py

@@ -29,7 +29,7 @@ def __init__(self, table_name=None, latest_table=None, connection=None):
         self.table_name = table_name
         self.latest_table_name = os.environ.get("DATALAKE_DNAMODB_LATEST_TABLE",
                                                 f"{latest_table}")
-        self.use_latest = os.environ.get("DATALAKE_LATEST_FLAG", False)
+        self.use_latest = os.environ.get("DATALAKE_USE_LATEST_TABLE", False)
         self._prepare_connection(connection)
         self.logger = logging.getLogger('storage')
 
@@ -73,7 +73,14 @@ def store_latest(self, record):
         note: Record must utilize AttributeValue syntax
               for the conditional put.
         """
+
+        condition_expression = " attribute_not_exists(what_where_key) OR metadata.start < :new_start"
+        expression_attribute_values = {
+            ':what_where_key': {'S': record['what_where_key']},
+            ':new_start': {'N': str(record['metadata']['start'])}
+        }
         record = {
+            'what_where_key': {"S": record['metadata']['what']+':'+record['metadata']['where']},
             'time_index_key': {"S": record['time_index_key']},
             'range_key': {"S": record['range_key']},
             'metadata': {
@@ -110,12 +117,13 @@ def store_latest(self, record):
             'url': {"S": record['url']},
             'create_time': {'N': str(record['create_time'])}
         }
+
         try:
             self._connection.put_item(
                 table_name=self.latest_table_name,
                 item=record,
-                condition_expression=\
-                    f"attribute_not_exists(metadata.M.start.N) OR metadata.M.start.N < {record['metadata']['M']['start']['N']}",
+                condition_expression=condition_expression,
+                expression_attribute_values=expression_attribute_values
             )
             self.logger.info("Record stored successfully.")
         except ConditionalCheckFailedException:

ingester/tests/conftest.py

@@ -70,7 +70,7 @@ def dynamodb_records_table(dynamodb_table_maker):
 
 @pytest.fixture
 def dynamodb_latest_table(dynamodb_table_maker):
-    schema = [HashKey('time_index_key'), RangeKey('range_key')]
+    schema = [HashKey('what_where_key')]
     return dynamodb_table_maker('latest', schema)
 
 

ingester/tests/test_storage.py

@@ -21,6 +21,7 @@ def test_insert_new_record(dynamodb_latest_table, dynamodb_connection):
     storage = DynamoDBStorage(latest_table='latest', connection=dynamodb_connection)
 
     new_record = {
+        'what_where_key': 'syslog:ground_server2',
         'time_index_key': '15225:newlog',
         'range_key': 'new_server:12345abcde',
         'metadata': {
@@ -44,8 +45,7 @@ def test_insert_new_record(dynamodb_latest_table, dynamodb_connection):
         print(f"Failed to store record: {str(e)}")
 
     stored_record = dynamodb_latest_table.get_item(
-        time_index_key='15225:newlog',
-        range_key='new_server:12345abcde'
+        what_where_key=new_record['what_where_key']
     )
     assert stored_record['metadata']['start'] == new_record['metadata']['start']
 
@@ -54,6 +54,7 @@ def test_store_conditional_put_latest_multiple_files(dynamodb_latest_table, dyna
     storage = DynamoDBStorage(latest_table='latest', connection=dynamodb_connection)
 
     file1 = {
+        'what_where_key': 'syslog:ground_server2',
         'time_index_key': '15219:zlcdzvawsp',
         'range_key': 'lawvuunyws:447a4a801cabc6089f04922abdfa8aad099824e9',
         'metadata': {
@@ -72,6 +73,7 @@ def test_store_conditional_put_latest_multiple_files(dynamodb_latest_table, dyna
     }
 
     file2 = {
+        'what_where_key': 'syslog:ground_server2',
         'time_index_key': '15219:zlcdzvawsp',
         'range_key': 'lawvuunyws:447a4a801cabc6089f04922abdfa8aad099824e9',
         'metadata': {
@@ -91,6 +93,7 @@ def test_store_conditional_put_latest_multiple_files(dynamodb_latest_table, dyna
     }
 
     file3 = {
+        'what_where_key': 'syslog:s114',
         'time_index_key': '15220:syslog',
         'range_key': 'ground_server2:34fb2d1ec54245c7a57e29ed5a6ea9b2',
         'metadata': {
@@ -99,7 +102,7 @@ def test_store_conditional_put_latest_multiple_files(dynamodb_latest_table, dyna
             'end': 1415128740728,
             'path': '/var/log/syslog.2',
             'work_id': 'foo-bizz',
-            'where': 'ground_server2',
+            'where': 's114',
             'what': 'syslog',
             'id': '34fb2d1ec54245c7a57e29ed5a6ea9b2',
             'hash': 'b4f2d8de24af342643d5b78a8f2b9b88'
@@ -113,52 +116,28 @@ def test_store_conditional_put_latest_multiple_files(dynamodb_latest_table, dyna
     storage.store_latest(file1)
     storage.store_latest(file2) # same what:where, but should replace file1 b/c newer
 
+    query_what_where = 'syslog:ground_server2'
+
     records = [dict(i) for i in dynamodb_latest_table.scan()]
 
-    res = dict(dynamodb_latest_table.get_item(time_index_key='15219:zlcdzvawsp',
-                                              range_key='lawvuunyws:447a4a801cabc6089f04922abdfa8aad099824e9'))
+    res = dict(dynamodb_latest_table.get_item(what_where_key=query_what_where))
     assert res['metadata']['start'] == Decimal('1314877177413')
     assert len(records) == 2
     assert file2 == res
 
+    storage.store_latest(file3)
+    storage.store_latest(file2)
+    storage.store_latest(file1)
 
-def test_concurrent_updates(dynamodb_latest_table, dynamodb_connection):
-    storage = DynamoDBStorage(latest_table='latest', connection=dynamodb_connection)
-
-    base_record = {
-        'time_index_key': '15219:zlcdzvawsp',
-        'range_key': 'lawvuunyws:447a4a801cabc6089f04922abdfa8aad099824e9',
-        'metadata': {
-            'version': 1,
-            'start': 1314877177402,
-            'end': 1314877177412, # ends ten seconds later
-            'path': '/var/log/syslog.2',
-            'work_id': 'abc-123',
-            'where': 'ground_server2',
-            'what': 'syslog',
-            'id': '34fb2d1ec54245c7a57e29ed5a6ea9b2',
-            'hash': 'b4f2d8de24af342643d5b78a8f2b9b88'
-        },
-        'url': 's3://existingfile/url',
-        'create_time': 1314877177402
-    }
-    storage.store_latest(base_record)
-
+    records = [dict(i) for i in dynamodb_latest_table.scan()]
+    res = dict(dynamodb_latest_table.get_item(what_where_key=query_what_where))
+    assert res['metadata']['id'] != file1['metadata']['id']
+    assert res['metadata']['id'] == file2['metadata']['id']
 
-    updated_record1 = base_record.copy()
-    updated_record1['metadata']['start'] += 10
-
-    updated_record2 = base_record.copy()
-    updated_record2['metadata']['start'] += 5
-
-
-    storage.store_latest(updated_record1)
-    storage.store_latest(updated_record2)
-
-    stored_record = dynamodb_latest_table.get_item(
-        time_index_key='15219:zlcdzvawsp',
-        range_key='lawvuunyws:447a4a801cabc6089f04922abdfa8aad099824e9'
-    )
-
-    assert stored_record['metadata']['start'] == updated_record1['metadata']['start']
+    storage.store_latest(file1)
+    storage.store_latest(file1)
+    storage.store_latest(file2)
+    storage.store_latest(file3)
+    res = dict(dynamodb_latest_table.get_item(what_where_key=query_what_where))
+    assert res['metadata']['start'] == file2['metadata']['start']