diff --git a/README.md b/README.md index e985429..0d6e718 100644 --- a/README.md +++ b/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 diff --git a/log_forwarder/clients.py b/log_forwarder/clients.py index 7e36cb6..0c8b774 100644 --- a/log_forwarder/clients.py +++ b/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') diff --git a/log_forwarder/config.py b/log_forwarder/config.py index 8ce7286..77c140a 100644 --- a/log_forwarder/config.py +++ b/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') diff --git a/log_forwarder/destination_provider.py b/log_forwarder/destination_provider.py index 17a585e..41b5f5e 100644 --- a/log_forwarder/destination_provider.py +++ b/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 diff --git a/log_forwarder/forward.py b/log_forwarder/forward.py index 8110576..e44788b 100644 --- a/log_forwarder/forward.py +++ b/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) diff --git a/tests/test_config.py b/tests/test_config.py index 3cbe0d4..72c43d0 100644 --- a/tests/test_config.py +++ b/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' diff --git a/tests/test_destination_provider.py b/tests/test_destination_provider.py index b4b6d7c..15f5a6c 100644 --- a/tests/test_destination_provider.py +++ b/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 diff --git a/tests/test_forward.py b/tests/test_forward.py new file mode 100644 index 0000000..4079ca0 --- /dev/null +++ b/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 +