Add support for dataset level tags and add default/auto set tags

README.md

@@ -19,7 +19,14 @@ the data with gzip. As a rule of thumb, a compression ratio of about 90% can be
 `default` if not set. Aggregators aggregate multiline log entries into a single entry. The `default` aggregator is a 
 noop (no entries get aggregated), while the `java_stack_trace` aggregator aggregates Java stack trace entries into a 
 single one.
-- `destination_config`: a base64 encoded map representing the configuration to where each log should be sent to.
+- `destination_config`: a base64 encoded map representing the configuration to where each log should be sent to. 
+`destination_config` takes the following attributes:
+  - `logname` or `dataset`: the Bronto dataset where to send the log data to 
+  - `logset` or `collection`: the Bronto collection that the dataset belongs to 
+  - `log_type`: the type of logs, e.g. cloudwatch, s3_access_log 
+  - `set_individual_subscription`: if set to true, then a log group subscription filter is set. Otherwise, the forwarder relies on an account level subscription filter. 
+  - `subscription_filter_pattern`: the subscription filter pattern to apply to the log group level subscription filter. This overwrites the default level subscription filter pattern.
+  - `tags`: a map whose keys and values are strings, representing tags to be associated to the Bronto dataset, e.g. `{"key1": "value1", "key2": "value2"}`.
 - `paths_regex`: `paths_regex` is a base64-encoded list of objects, each containing a regular expression pattern with a 
 named capture group called `dest_config_id`. This is used for log data delivered to S3 when the S3 object key does not 
 follow standard AWS naming conventions, such as when data is moved between buckets and renamed. In such cases, the 

log_forwarder/clients.py

@@ -60,12 +60,13 @@ def __init__(self, api_key, ingestion_endpoint, dataset, collection, client_type
         self.collection = collection
         self.client_type = client_type
         self.ingestion_endpoint = ingestion_endpoint
+        self.formatted_tags = ','.join([f'{key}={value}' for key, value in tags.items()])
         self.headers = {
             'Content-Encoding': 'gzip',
             'Content-Type': 'application/json',
             'User-Agent': 'bronto-aws-integration',
             'x-bronto-api-key': self.api_key,
-            'x-bronto-tags': ','.join([f'{key}={value}' for key, value in tags.items()])
+            'x-bronto-tags': self.formatted_tags
         }
         if self.dataset is not None:
             self.headers.update({'x-bronto-service-name': self.dataset})
@@ -87,8 +88,8 @@ def _send_batch(self, compressed_batch):
                 time.sleep(delay_sec)
                 self._send_batch(compressed_batch)
             elif resp.status == 200:
-                logger.info('data sent successfully. collection=%s, dataset=%s', self.collection,
-                            self.dataset)
+                logger.info('data sent successfully. collection=%s, dataset=%s, tags=%s', self.collection,
+                            self.dataset, self.formatted_tags)
             else:
                 logger.error('max attempts reached. attempt=%s, max_attempts=%s', attempt, max_attempts)
                 raise Exception('BrontoClientMaxAttemptReached')

log_forwarder/config.py

@@ -124,6 +124,10 @@ def get_collection(self, key):
             return collection_value
         return self._get_attribute_value(key, 'collection')
 
+    def get_dataset_tags(self, key):
+        dataset_tags_value = self._get_attribute_value(key, 'tags')
+        return dataset_tags_value if dataset_tags_value is not None and isinstance(dataset_tags_value, dict) else {}
+
     def get_log_type(self, key):
         return self._get_attribute_value(key, 'log_type')
 

log_forwarder/destination_provider.py

@@ -33,3 +33,9 @@ def get_collection(self, data_id):
         if self._data_retriever.get_collection_type() == 'cloudwatch':
             return self._config.get_cloudwatch_default_collection()
         return None
+
+    def get_dataset_tags(self, data_id):
+        log_type = self.get_type(data_id)
+        tags = {'aws_log_type': log_type if log_type is not None else 'unknown'}
+        tags.update(self._config.get_dataset_tags(data_id))
+        return tags

log_forwarder/forward.py

@@ -44,6 +44,10 @@ def process(event):
             destination_provider = DestinationProvider(dest_config, data_retriever)
             dataset = destination_provider.get_dataset(data_id)
             collection = destination_provider.get_collection(data_id)
+            all_tags = {}
+            all_tags.update(config.get_tags())
+            dataset_tags = destination_provider.get_dataset_tags(data_id)
+            all_tags.update(dataset_tags)
             log_type = dest_config.get_log_type(data_id)
             client_type = dest_config.get_client_type(data_id)
             logger.info('Destination information retrieved. dataset=%s, collection=%s, log_type=%s', dataset,
@@ -58,7 +62,7 @@ def process(event):
             attributes = config.get_resource_attributes()
             attributes.update(data_retriever.get_log_attributes_from_payload())
             bronto_client = BrontoClient(dest_config.bronto_api_key, dest_config.bronto_endpoint, dataset, collection,
-                client_type, config.tags)
+                client_type, all_tags)
             no_formatting = client_type is not None
             batch = Batch(dest_config.max_batch_size, no_formatting)
             aggregator = AggregatorFactory.get_aggregator(config.aggregator)

tests/test_config.py

@@ -8,12 +8,14 @@
 
 def test_destination_config(monkeypatch):
     key = 'some_id'
-    raw_config = {key: {'dataset': 'my_dataset', 'collection': 'my_collection', 'log_type': 'my_log_type'}}
+    raw_config = {key: {'dataset': 'my_dataset', 'collection': 'my_collection', 'log_type': 'my_log_type',
+                        'tags': {'key1': 'val1', 'key2': 'val2'}}}
     monkeypatch.setenv('destination_config', base64.b64encode(json.dumps(raw_config).encode()).decode())
     dest_config = DestinationConfig()
     assert dest_config.get_log_type(key) == raw_config[key]['log_type']
     assert dest_config.get_dataset(key) == raw_config[key]['dataset']
     assert dest_config.get_collection(key) == raw_config[key]['collection']
+    assert dest_config.get_dataset_tags(key) == raw_config[key]['tags']
     assert dest_config.get_keys() == [key]
     assert dest_config.get_client_type(key) is None
 
@@ -33,6 +35,7 @@ def test_destination_config_is_optional():
     assert dest_config.get_log_type(key) is None
     assert dest_config.get_dataset(key) is None
     assert dest_config.get_collection(key) is None
+    assert dest_config.get_dataset_tags(key) == {}
 
 def test_destination_config_dataset_not_defined(monkeypatch):
     key = 'some_id'

tests/test_destination_provider.py

@@ -41,7 +41,6 @@ def test_s3_custom_path():
         assert type(data_retriever) == CustomS3Retriever
         assert data_retriever.get_data_id() == expected_data_id
 
-
 def test_cloudwatch_no_config(monkeypatch):
     log_group_name = 'whatever'
     with tempfile.NamedTemporaryFile() as f:
@@ -53,6 +52,7 @@ def test_cloudwatch_no_config(monkeypatch):
         assert destination_provider.get_type(log_group_name) == CLOUDWATCH_LOG_TYPE
         assert destination_provider.get_dataset(log_group_name) == log_group_name
         assert destination_provider.get_collection(log_group_name) is None
+        assert destination_provider.get_dataset_tags(log_group_name) == {'aws_log_type': 'cloudwatch_log'}
 
 
 def test_cloudwatch_default_collection(monkeypatch):
@@ -77,8 +77,55 @@ def test_cloudwatch_config_takes_precedence(monkeypatch):
         raw_dest_config = {log_group_name: {'dataset': log_group_name_from_config, 'collection': log_set_from_config}}
         monkeypatch.setenv('destination_config', base64.b64encode(json.dumps(raw_dest_config).encode()).decode())
         dest_config = DestinationConfig()
+        # mock _get_data in order to associate log group name to the data retriever
         monkeypatch.setattr(DataRetriever, 'get_data', lambda: _get_data(data_retriever, log_group_name_from_config))
         destination_provider = DestinationProvider(dest_config, data_retriever)
         assert destination_provider.get_type(log_group_name) == CLOUDWATCH_LOG_TYPE
         assert destination_provider.get_dataset(log_group_name) == log_group_name_from_config
         assert destination_provider.get_collection(log_group_name) == log_set_from_config
+
+
+def test_custom_dataset_tags(monkeypatch):
+    log_group_name = 'whatever'
+    with tempfile.NamedTemporaryFile() as f:
+        config = Config({}, f.name)
+        data_retriever = CloudwatchDataRetriever(config)
+        auto_set_tags = {'aws_log_type': 'cloudwatch_log'}
+        dataset_custom_tags = {'key1': 'val1', 'key2': 'val2'}
+        raw_dest_config = {log_group_name: {'tags': dataset_custom_tags}}
+        monkeypatch.setenv('destination_config', base64.b64encode(json.dumps(raw_dest_config).encode()).decode())
+        dest_config = DestinationConfig()
+        destination_provider = DestinationProvider(dest_config, data_retriever)
+        tags = auto_set_tags
+        tags.update(dataset_custom_tags)
+        assert destination_provider.get_dataset_tags(log_group_name) == tags
+
+
+def test_custom_dataset_tags_not_set(monkeypatch):
+    log_group_name = 'whatever'
+    with tempfile.NamedTemporaryFile() as f:
+        config = Config({}, f.name)
+        data_retriever = CloudwatchDataRetriever(config)
+        auto_set_tags = {'aws_log_type': 'cloudwatch_log'}
+        raw_dest_config = {log_group_name: {}}
+        monkeypatch.setenv('destination_config', base64.b64encode(json.dumps(raw_dest_config).encode()).decode())
+        dest_config = DestinationConfig()
+        destination_provider = DestinationProvider(dest_config, data_retriever)
+        assert destination_provider.get_dataset_tags(log_group_name) == auto_set_tags
+
+
+def test_custom_dataset_tags_overwrites_auto_set_tags(monkeypatch):
+    log_group_name = 'whatever'
+    with tempfile.NamedTemporaryFile() as f:
+        auto_set_tag_key = 'aws_log_type'
+        config = Config({}, f.name)
+        data_retriever = CloudwatchDataRetriever(config)
+        auto_set_tags = {auto_set_tag_key: 'cloudwatch_log'}
+        dataset_custom_tags = {'key1': 'val1', auto_set_tag_key: 'val2'}
+        raw_dest_config = {log_group_name: {'tags': dataset_custom_tags}}
+        monkeypatch.setenv('destination_config', base64.b64encode(json.dumps(raw_dest_config).encode()).decode())
+        dest_config = DestinationConfig()
+        destination_provider = DestinationProvider(dest_config, data_retriever)
+        tags = auto_set_tags
+        tags.update(dataset_custom_tags)
+        assert destination_provider.get_dataset_tags(log_group_name) == tags

tests/test_forward.py

@@ -0,0 +1,38 @@
+import base64
+import json
+
+import pytest
+
+import log_forwarder.clients
+from log_forwarder.forward import process
+
+
+EVENT = {'awslogs': {'data': 'H4sIAAAAAAAA/7VWXW/bVhL9K8LFPkryfN6Zyzc3dYMAdlLU7hbdKDAo6cohIJNakrLrBv7vC1IO0GAtNAJ238iZuXMuzxnOzJdwn7uuvMs3T7scivDj+c357dXF9fX524swDc1jndtQBGWmaJAc0cM0bJu7t22z34UinJWP3dm2vF+uy7OqvstdXzX1VVNXfdP+tK9Xw+vhxHXf5vI+FIGA9AzhDPjs4z8uz28urm8+Uam43JSicYOiVC6TlymhS1nKRgTDNHT7Zbdqq92Q8adq2+e2C8XHb8yzzWgPn0bAi4dc90PMl1CtQxE4ETsnTWIRk6VEkQlZ2IUlEjmJkKgkZFSV6MLsaDFMQ1/d564v73ehQNOkCEombtOv9IUifFmMYYtQLMZPnCHMgG8wFqoF2Vzc/rUI00Xon3aHqN227DdNez/v+rLtR1+bV027XoTiy/D8733u+nfrMdhj3HCJebZmX81kDetZWTLPCMiWOUsE1DHF5oX187YeD5ZtXZSPXXEQqcj72WPu+hkWfxW1+HqqOCbimPsht934XCzCi3SL8Pwcnqf/zbEpR2BjERjYRWMxAnZjQVcgNGIjIIJofJTjxK9wPIYtQjF5lemB5ckibPND3h6Cfjv/5f27928P9pdsB8/FH7um7XM7KbdtLtdPk+7zvl83j/V0Ut3VTVvVd5Nl2a8+v+Rs7u5yezja7HLd522+z337NM8vieZNv93Nd23TN/PPfb+bD+5qdfvVf8jzV2kni5DWsFxGW84iLnEmsIFZEsCZs6xWKfoy+mYRnhf1a0Sn5BzNSYXUoou4RdKIoBFFkqu5CaNTJDI5RrT+n4n+8JDbtloPfDabyWrftrnuJ1e5z+3PbfNQrXM7qbpJ3fSTcrttHvP67xg/ENvNb6u6z21dbl+l9vt+m9epVSJkUzITRDCJDmiW0IndDSUmAZKEkYey1mPUxm+p/SV3u6bucjEhgMnrwKyKlkw8ApG6YAIyZFJxBWNNajiIhuTORxuU6cnA4p5caGiPLEI4FE9kia6OwiTJBDQyJ2VLdgzY8WTgoTjdY4LkEkk9RSPViBhtoJxTgqQpkZmb+lFgOxnYNGpijISmDimyqg3NSwbVFRjAPLLi0MswHQNOfDKwJ+BkkZljTInZRIiTJ3MDhURAYtEQhDw6HAU+ubgikBAQgqg6miEyGRMIpBg9OoJgNIiJlOM4eV8Ddji5uCKqA0Q1Hdp6BFdCIkEXiExkAOpqQj4wDnQMGE8urkgJHRHGH1ZFNEEyjEyGgw6aYJR7qD7WeGwWOZ5cXFGGRqyKjoxGzpScwMWAVJOIwxCJokaudKw3O51cXFGMRAycQTwKRXZBh2H+cooGQuKUzMZrxaMak8r3bzg+J5VjG06bh/H3v1pxXlr/Icd635bjvjK8W4xzAZ0uwrLabvP6x2+dNh6+b9qn6+rPfPXDIhRIPhjLP65G+69dXo920+fpInR92e+78XbdfrXKXXfYdz5Nw67ZVquny2H4hSKcv3nz4df3N7eXF/+8uLz9+cPluze/h+f/AK22qw1fCwAA'}}
+MESSAGES = [
+    "{\"time\":\"2025-10-03T16:55:27.487Z\",\"type\":\"platform.start\",\"record\":{\"requestId\":\"866f3a1e-d38c-4d0d-aa33-2027bee46015\",\"functionArn\":\"arn:aws:lambda:eu-west-1:533267098118:function:ingestionMonitorFunction\",\"version\":\"$LATEST\"}}",
+    "{\"timestamp\": \"2025-10-03T16:55:27Z\", \"level\": \"WARNING\", \"message\": \"Exporter already shutdown, ignoring batch\", \"logger\": \"opentelemetry.exporter.otlp.proto.http.metric_exporter\", \"requestId\": \"9d0bb67b-61b1-40f0-9401-834cc968b68f\"}",
+    "{\"timestamp\": \"2025-10-03T16:55:27Z\", \"level\": \"WARNING\", \"message\": \"Overriding of current MeterProvider is not allowed\", \"logger\": \"opentelemetry.metrics._internal\", \"requestId\": \"866f3a1e-d38c-4d0d-aa33-2027bee46015\"}",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "Response: 200",
+    "{\"time\":\"2025-10-03T16:55:28.254Z\",\"type\":\"platform.report\",\"record\":{\"requestId\":\"866f3a1e-d38c-4d0d-aa33-2027bee46015\",\"metrics\":{\"durationMs\":766.405,\"billedDurationMs\":767,\"memorySizeMB\":128,\"maxMemoryUsedMB\":75},\"status\":\"success\"}}"
+]
+
+
+def test_forward(monkeypatch):
+    log_data = []
+    destination_config = {'/aws/lambda/ingestionMonitorFunction': {'tags': 'my_tag_key,my_tag_value'}}
+    destination_config_str = json.dumps(destination_config)
+    monkeypatch.setenv('destination_config', base64.b64encode(destination_config_str.encode()).decode())
+    monkeypatch.setattr(log_forwarder.forward.BrontoClient, 'send_data',
+                        lambda _, batch, attributes=None: log_data.extend(batch.batch))
+    process(EVENT)
+    assert log_data == MESSAGES
+