From 3993a7f132c255fd4cd0d095d5ff95d2863b0634 Mon Sep 17 00:00:00 2001 From: Benoit Gaudin Date: Tue, 30 Sep 2025 16:40:41 +0100 Subject: [PATCH] add support to collapse JavaStackTrace into a single event --- .github/workflows/test.yml | 27 +++++ .gitignore | 3 + README.md | 4 + log_forwarder/aggregator.py | 85 +++++++++++++ log_forwarder/clients.py | 7 +- log_forwarder/config.py | 2 +- log_forwarder/exporter.py | 32 +++++ log_forwarder/forward.py | 14 +-- log_forwarder/logfile.py | 2 +- test.sh | 16 +++ test_requirements.txt | 2 + tests/test_aggregator.py | 231 ++++++++++++++++++++++++++++++++++++ tests/test_batch.py | 14 ++- tests/test_config.py | 1 - tests/test_exporter.py | 113 ++++++++++++++++++ tests/test_parser.py | 2 +- 16 files changed, 539 insertions(+), 16 deletions(-) create mode 100644 .github/workflows/test.yml create mode 100644 log_forwarder/aggregator.py create mode 100644 log_forwarder/exporter.py create mode 100755 test.sh create mode 100644 test_requirements.txt create mode 100644 tests/test_aggregator.py create mode 100644 tests/test_exporter.py diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml new file mode 100644 index 0000000..f00c69c --- /dev/null +++ b/.github/workflows/test.yml @@ -0,0 +1,27 @@ +name: Test Package + +on: + pull_request: + branches: [ "main" ] + +permissions: + id-token: write + contents: write + packages: write + deployments: write + +jobs: + build: + runs-on: ubuntu-latest + + steps: + - name: Checkout + uses: actions/checkout@v4 + + - name: Set up environment + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Test + run: ./test.sh diff --git a/.gitignore b/.gitignore index c1c7518..026fafc 100644 --- a/.gitignore +++ b/.gitignore @@ -158,3 +158,6 @@ cython_debug/ # and can be added to the global gitignore or merged into this file. For a more nuclear # option (not recommended) you can uncomment the following to ignore the entire idea folder. **/.idea/ + +# test python env +test_env/ diff --git a/README.md b/README.md index b9942af..9141232 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,10 @@ This lambda function can be configured with the following attributes: - `bronto_api_key`: a BrontoBytes account API key - `max_batch_size`: the maximum size of an uncompressed payload sent to BrontoBytes. This lambda function compresses the data with gzip. As a rule of thumb, a compression ratio of about 90% can be expected. +- `aggregator`: the name of an aggregator to use: either `java_stack_trace` or `default`. This property defaults to +`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. - `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 diff --git a/log_forwarder/aggregator.py b/log_forwarder/aggregator.py new file mode 100644 index 0000000..ace8a5e --- /dev/null +++ b/log_forwarder/aggregator.py @@ -0,0 +1,85 @@ +from typing import Union +from queue import Queue + +AGGREGATED_TERMINATION_MARKER = None + + +class Aggregator: + + _NO_AGGREGATED_LINE = None + + def add_line(self, line: Union[str, None]): + raise NotImplementedError() + + def get_complete_aggregated_line(self): + raise NotImplementedError() + + def complete(self): + self.add_line(AGGREGATED_TERMINATION_MARKER) + + def has_complete_aggregated_line(self): + raise NotImplementedError() + + +class NoopAggregator(Aggregator): + + def __init__(self): + self.lines = Queue(maxsize=2) + + def add_line(self, line: Union[str, None]): + if self.has_complete_aggregated_line(): + raise Exception('Complete aggregated lines must be retrieved first') + if line is not None and line == '': + return Aggregator._NO_AGGREGATED_LINE + self.lines.put(line) + return + + def get_complete_aggregated_line(self): + if self.has_complete_aggregated_line(): + return self.lines.get() + return Aggregator._NO_AGGREGATED_LINE + + def has_complete_aggregated_line(self): + return self.lines.full() + + +class JavaStackTraceAggregator(Aggregator): + """ Inspired from https://www.elastic.co/docs/reference/beats/filebeat/multiline-examples#_java_stack_traces """ + + def __init__(self): + self.lines = Queue(maxsize=2) + + def add_line(self, line): + if self.lines.full(): + raise Exception('Aggregator queue is full') + if line is None: + current_line = line + else: + stripped_line = line.lstrip() + if stripped_line == '': + return + if (stripped_line.startswith('at ') or stripped_line.startswith('Caused by: ') or + stripped_line.startswith('...')): + current_line = self.lines.get().rstrip('\n') + current_line += f'\\n{line}' + else: + current_line = line + self.lines.put(current_line) + return + + def get_complete_aggregated_line(self): + if self.has_complete_aggregated_line(): + return self.lines.get() + return Aggregator._NO_AGGREGATED_LINE + + def has_complete_aggregated_line(self): + return self.lines.full() + + +class AggregatorFactory: + + @staticmethod + def get_aggregator(aggregator_name) -> Aggregator: + if aggregator_name == 'java_stack_trace': + return JavaStackTraceAggregator() + return NoopAggregator() diff --git a/log_forwarder/clients.py b/log_forwarder/clients.py index 4b95d95..f0ff2f1 100644 --- a/log_forwarder/clients.py +++ b/log_forwarder/clients.py @@ -23,9 +23,10 @@ def download(self, bucket, object_key): class Batch: - def __init__(self, no_formatting=False): + def __init__(self, max_size: int, no_formatting=False): self.batch = [] self.size = 0 + self.max_size = max_size self.no_formatting = no_formatting def add(self, line): @@ -46,6 +47,10 @@ def get_formatted_data(self, attributes: Dict[str, str]): log_message.update(attributes) return '\n'.join([json.dumps(log_message) for log_message in log_messages]) + def reset(self): + self.batch = [] + self.size = 0 + class BrontoClient: diff --git a/log_forwarder/config.py b/log_forwarder/config.py index bc1b3a5..0e50134 100644 --- a/log_forwarder/config.py +++ b/log_forwarder/config.py @@ -1,7 +1,6 @@ import json import base64 import os -import tempfile import logging from typing import List import boto3 @@ -26,6 +25,7 @@ def __init__(self, event, filepath): self.filepath = filepath self.event = event self.path_regexes = os.environ.get('path_regexes') + self.aggregator = os.environ.get('aggregator', 'default') raw_attributes = os.environ.get('attributes') self.resource_attributes = {} if raw_attributes is not None and raw_attributes != '': diff --git a/log_forwarder/exporter.py b/log_forwarder/exporter.py new file mode 100644 index 0000000..f296a91 --- /dev/null +++ b/log_forwarder/exporter.py @@ -0,0 +1,32 @@ +from typing import Dict, List +from aggregator import Aggregator +from clients import BrontoClient, Batch +from parser import Parser + + +class BrontoExporter: + + def __init__(self, client: BrontoClient, parser: Parser, batch: Batch, aggregator: Aggregator, + attributes: Dict[str, str]): + self.client = client + self.parser = parser + self.batch = batch + self.aggregator = aggregator + self.attributes = attributes + + def export(self): + for line in self.parser.get_parsed_lines(): + self.aggregator.add_line(line) + if not self.aggregator.has_complete_aggregated_line(): + continue + _line = self.aggregator.get_complete_aggregated_line() + self.batch.add(_line) + if self.batch.get_batch_size() > self.batch.max_size: + self.client.send_data(self.batch, self.attributes) + self.batch.reset() + self.aggregator.complete() + if self.aggregator.has_complete_aggregated_line(): + _line = self.aggregator.get_complete_aggregated_line() + self.batch.add(_line) + if self.batch.get_batch_size() > 0: + self.client.send_data(self.batch, self.attributes) diff --git a/log_forwarder/forward.py b/log_forwarder/forward.py index fbd5d8e..5e10ce7 100644 --- a/log_forwarder/forward.py +++ b/log_forwarder/forward.py @@ -2,9 +2,11 @@ import tempfile import shutil +from aggregator import JavaStackTraceAggregator, AggregatorFactory from config import Config, DestinationConfig from data_retriever import DataRetrieverFactory from destination_provider import DestinationProvider +from exporter import BrontoExporter from parser import ParserFactory from clients import BrontoClient, Batch from logfile import LogFileFactory @@ -58,14 +60,10 @@ def process(event): bronto_client = BrontoClient(dest_config.bronto_api_key, dest_config.bronto_endpoint, dataset, collection, client_type) no_formatting = client_type is not None - batch = Batch(no_formatting) - for line in parser.get_parsed_lines(): - batch.add(line) - if batch.get_batch_size() > dest_config.max_batch_size: - bronto_client.send_data(batch, attributes) - batch = Batch() - if batch.get_batch_size() > 0: - bronto_client.send_data(batch, config.get_resource_attributes()) + batch = Batch(dest_config.max_batch_size, no_formatting) + aggregator = AggregatorFactory.get_aggregator(config.aggregator) + exporter = BrontoExporter(bronto_client, parser, batch, aggregator, attributes) + exporter.export() def forward_logs(_event, _): diff --git a/log_forwarder/logfile.py b/log_forwarder/logfile.py index 5f55cf2..b592bcb 100644 --- a/log_forwarder/logfile.py +++ b/log_forwarder/logfile.py @@ -46,7 +46,7 @@ def get_file(self): def get_lines(self): with self.file as f: for line in f.readlines(): - yield line.strip() + yield line.rstrip() # we keep spaces at the start as they may be indicative of a stack trace class LogFileFactory: diff --git a/test.sh b/test.sh new file mode 100755 index 0000000..af4c1b3 --- /dev/null +++ b/test.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash + +set -e + +PROJECT_ROOT_DIR="${PWD}" +TEST_REQUIREMENT_FILE="${PROJECT_ROOT_DIR}/test_requirements.txt" + +rm -rf test_env +python3 -m venv test_env +"${PROJECT_ROOT_DIR}/test_env/bin/python" -m pip install --upgrade pip +test_env/bin/pip install -r test_requirements.txt + +cd "${PROJECT_ROOT_DIR}/tests" || exit 1 + +echo "Running Tests" +PYTHONPATH="${PROJECT_ROOT_DIR}/log_forwarder" "${PROJECT_ROOT_DIR}/test_env/bin/python" -m pytest -v -s diff --git a/test_requirements.txt b/test_requirements.txt new file mode 100644 index 0000000..ee01444 --- /dev/null +++ b/test_requirements.txt @@ -0,0 +1,2 @@ +pytest +boto3 diff --git a/tests/test_aggregator.py b/tests/test_aggregator.py new file mode 100644 index 0000000..7566797 --- /dev/null +++ b/tests/test_aggregator.py @@ -0,0 +1,231 @@ +import pytest +from queue import Queue + +from log_forwarder.aggregator import JavaStackTraceAggregator, NoopAggregator, AggregatorFactory + + +def test_aggregator_factory_stack_trace(): + assert isinstance(AggregatorFactory.get_aggregator('java_stack_trace'), JavaStackTraceAggregator) + + +def test_aggregator_factory_default(): + assert isinstance(AggregatorFactory.get_aggregator('anything_but_java_stack_trace'), NoopAggregator) + + +# +# NoopAggregator +# + +def test_noop_init(): + aggregator = NoopAggregator() + assert isinstance(aggregator.lines, Queue) + assert aggregator.lines.maxsize == 2 + assert aggregator.lines.empty() + + +def test_noop_add_single_line(): + aggregator = NoopAggregator() + line = "Exception in thread main java.lang.NullPointerException" + aggregator.add_line(line) + assert aggregator.get_complete_aggregated_line() is None + + +def test_noop_add_normal_line(): + aggregator = NoopAggregator() + line = "Exception in thread main java.lang.NullPointerException" + aggregator.add_line(line) + aggregator.add_line(None) + assert aggregator.get_complete_aggregated_line() == line + + +def test_noop_add_stack_trace(): + aggregator = NoopAggregator() + input_lines = ["Exception in thread main java.lang.NullPointerException", "at com.example.Main.method(Main.java:42)"] + + aggregator.add_line(input_lines[0]) + aggregator.add_line(input_lines[1]) + result = [aggregator.get_complete_aggregated_line()] + aggregator.add_line(None) + result.append(aggregator.get_complete_aggregated_line()) + assert result == input_lines + +# +# Java Stack Trace Aggregator +# + +def test_init(): + aggregator = JavaStackTraceAggregator() + assert isinstance(aggregator.lines, Queue) + assert aggregator.lines.maxsize == 2 + assert aggregator.lines.empty() + + +def test_add_line_normal_line(): + aggregator = JavaStackTraceAggregator() + line = "Exception in thread main java.lang.NullPointerException" + aggregator.add_line(line) + assert aggregator.lines.qsize() == 1 + assert aggregator.get_complete_aggregated_line() is None + + +def test_add_line_stack_trace_line(): + aggregator = JavaStackTraceAggregator() + main_line = "Exception in thread main java.lang.NullPointerException" + stack_line = "at com.example.Main.method(Main.java:42)" + + aggregator.add_line(main_line) + aggregator.add_line(stack_line) + + assert aggregator.lines.qsize() == 1 + result = aggregator.lines.get() + expected = f"{main_line}\\n{stack_line}" + assert result == expected + + +def test_add_line_stack_trace_line_with_caused_by(): + aggregator = JavaStackTraceAggregator() + main_line = "Exception in thread main java.lang.NullPointerException" + stack_line = "Caused by: java.net.SocketException: some socket error" + + aggregator.add_line(main_line) + aggregator.add_line(stack_line) + + assert aggregator.lines.qsize() == 1 + result = aggregator.lines.get() + expected = f"{main_line}\\n{stack_line}" + assert result == expected + +def test_add_line_stack_trace_line_with_3_dots(): + aggregator = JavaStackTraceAggregator() + main_line = "Exception in thread main java.lang.NullPointerException" + stack_line = "Caused by: java.net.SocketException: some socket error" + + aggregator.add_line(main_line) + aggregator.add_line(stack_line) + + assert aggregator.lines.qsize() == 1 + result = aggregator.lines.get() + expected = f"{main_line}\\n{stack_line}" + assert result == expected + +def test_add_line_multiple_stack_trace_lines(): + aggregator = JavaStackTraceAggregator() + main_line = "Exception in thread main java.lang.NullPointerException" + stack_line1 = "at com.example.Main.method1(Main.java:42)" + stack_line2 = "at com.example.Main.method2(Main.java:35)" + + aggregator.add_line(main_line) + aggregator.add_line(stack_line1) + aggregator.add_line(stack_line2) + + assert aggregator.lines.qsize() == 1 + result = aggregator.lines.get() + expected = f"{main_line}\\n{stack_line1}\\n{stack_line2}" + assert result == expected + + +def test_add_line_none(): + aggregator = JavaStackTraceAggregator() + aggregator.add_line(None) + assert aggregator.lines.qsize() == 1 + assert aggregator.lines.get() is None + + +def test_add_line_queue_full_exception(): + aggregator = JavaStackTraceAggregator() + aggregator.add_line("line1") + aggregator.add_line("line2") + + with pytest.raises(Exception, match="Aggregator queue is full"): + aggregator.add_line("line3") + + +def test_get_line_empty_queue(): + aggregator = JavaStackTraceAggregator() + assert aggregator.get_complete_aggregated_line() is None + + +def test_get_line_partial_queue(): + aggregator = JavaStackTraceAggregator() + aggregator.add_line("single line") + assert aggregator.get_complete_aggregated_line() is None + + +def test_get_line_full_queue(): + aggregator = JavaStackTraceAggregator() + aggregator.add_line("line1") + aggregator.add_line("line2") + + result = aggregator.get_complete_aggregated_line() + assert result == "line1" + assert aggregator.lines.qsize() == 1 + + +def test_stack_trace_aggregation_workflow(): + aggregator = JavaStackTraceAggregator() + + exception_line = "java.lang.RuntimeException: Something went wrong" + stack_line1 = "at com.example.Service.doWork(Service.java:123)" + stack_line2 = "at com.example.Main.main(Main.java:45)" + next_line = "INFO: Processing completed" + + aggregator.add_line(exception_line) + aggregator.add_line(stack_line1) + assert aggregator.get_complete_aggregated_line() is None + + aggregator.add_line(stack_line2) + assert aggregator.get_complete_aggregated_line() is None + + aggregator.add_line(next_line) + + first_result = aggregator.get_complete_aggregated_line() + expected_stack_trace = f"{exception_line}\\n{stack_line1}\\n{stack_line2}" + assert first_result == expected_stack_trace + + # add poison pill + aggregator.add_line(None) + second_result = aggregator.get_complete_aggregated_line() + assert second_result == next_line + + +def test_non_stack_trace_after_stack_trace(): + aggregator = JavaStackTraceAggregator() + + line1 = "Normal log line" + stack_line = "at com.example.Test.method(Test.java:10)" + line2 = "Another normal line" + + aggregator.add_line(line1) + aggregator.add_line(stack_line) + aggregator.add_line(line2) + + first_result = aggregator.get_complete_aggregated_line() + expected = f"{line1}\\n{stack_line}" + assert first_result == expected + + no_result_as_no_poison_pill = aggregator.get_complete_aggregated_line() + assert no_result_as_no_poison_pill is None + aggregator.add_line(None) + second_result = aggregator.get_complete_aggregated_line() + assert second_result == line2 + + +def test_edge_case_stack_trace_prefix(): + aggregator = JavaStackTraceAggregator() + + main_line = "Exception occurred" + not_stack_line = "AT the beginning of line" + actual_stack_line = "at com.example.Class.method(Class.java:1)" + + aggregator.add_line(main_line) + aggregator.add_line(not_stack_line) + + first_result = aggregator.get_complete_aggregated_line() + assert first_result == main_line + + aggregator.add_line(actual_stack_line) + # add poison pill / termination marker + aggregator.add_line(None) + second_result = aggregator.get_complete_aggregated_line() + expected = f"{not_stack_line}\\n{actual_stack_line}" + assert second_result == expected \ No newline at end of file diff --git a/tests/test_batch.py b/tests/test_batch.py index 573e228..fe2b675 100644 --- a/tests/test_batch.py +++ b/tests/test_batch.py @@ -4,15 +4,16 @@ def test_add_to_batch(): - batch = Batch() + batch = Batch(1) entry = "an entry" batch.add(entry) assert batch.get_batch_size() == len(entry) assert batch.get_data() == [entry] + assert batch.max_size == 1 def test_get_formatted_batch(): - batch = Batch() + batch = Batch(1) entry = "an entry" batch.add(entry) attributes = {'key': 'value'} @@ -21,9 +22,16 @@ def test_get_formatted_batch(): assert json.loads(batch.get_formatted_data(attributes)) == expected def test_get_formatted_batch_with_no_formatting(): - batch = Batch(no_formatting=True) + batch = Batch(1, no_formatting=True) entry = "an entry" batch.add(entry) attributes = {'key': 'value'} assert batch.get_formatted_data(attributes) == entry +def test_reset(): + batch = Batch(1) + entry = "an entry" + batch.add(entry) + batch.reset() + assert batch.batch == [] + assert batch.size == 0 diff --git a/tests/test_config.py b/tests/test_config.py index 8674189..0a83b92 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -91,7 +91,6 @@ def test_paths_regex_config(monkeypatch): dest_config = DestinationConfig() assert dest_config.get_paths_regex() == raw_config - def test_file_is_deleted_on_close(): with tempfile.NamedTemporaryFile() as f: filepath = f.name diff --git a/tests/test_exporter.py b/tests/test_exporter.py new file mode 100644 index 0000000..86d8fed --- /dev/null +++ b/tests/test_exporter.py @@ -0,0 +1,113 @@ +import tempfile +import pytest +from typing import List + +from exporter import BrontoExporter +from clients import BrontoClient, Batch +from logfile import LogFileFactory +from parser import ParserFactory +from aggregator import JavaStackTraceAggregator, NoopAggregator + + +class TestBrontoExporter: + + @pytest.fixture() + def client(self, monkeypatch): + return BrontoClient('api_key', 'endpoint', 'my_dataset','my_collection', + 'my_client_type') + + @pytest.fixture() + def log_file(self): + filepath = tempfile.NamedTemporaryFile(delete=True, delete_on_close=True).name + # create the file + open(filepath, 'w') + # cloudwatch log files are plaintext + return LogFileFactory.get_log_file('cloudwatch_log', filepath) + + @pytest.fixture() + def parser(self, log_file): + return ParserFactory.get_parser('cloudwatch_log', log_file) + + @pytest.fixture() + def java_stacktrace_aggregator(self): + return JavaStackTraceAggregator() + + @pytest.fixture() + def noop_aggregator(self): + return NoopAggregator() + + @pytest.fixture() + def batch(self): + return Batch(2) + + @pytest.fixture() + def attributes(self, parser, batch, java_stacktrace_aggregator, client): + return {'key': 'value', 'service': 'test'} + + @staticmethod + def add_lines_to_file(lines: List[str], filename): + with open(filename, 'w') as f: + for line in lines: + f.write(line + '\n') + + def test_export_no_lines(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + sent_batches = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda b: sent_batches.append(b)) + exporter.export() + + assert sent_batches == [] + + def test_export_as_many_lines_as_max_batch_size(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + input_lines = [f'log line {i}' for i in range(0, batch.max_size)] + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [input_line.rstrip('\n') for input_line in input_lines] + + def test_export_less_lines_than_max_batch_size(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + input_lines = [f'log line {i}' for i in range(0, batch.max_size - 1)] + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [input_line.rstrip('\n') for input_line in input_lines] + + def test_export_more_lines_then_max_batch_size(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + input_lines = [f'log line {i}' for i in range(0, batch.max_size + 1)] + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [input_line.rstrip('\n') for input_line in input_lines] + + def test_export_with_stack_trace(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + input_lines = ['line.with.SomeException', 'at some.more.specific.line:123'] + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [f'{input_lines[0].rstrip('\n')}\\n{input_lines[1]}'] + + def test_export_with_stack_trace_with_return(self, parser, batch, java_stacktrace_aggregator, client, attributes, monkeypatch): + input_lines = ['line.with.SomeException\n', 'at some.more.specific.line:123'] + exporter = BrontoExporter(client, parser, batch, java_stacktrace_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [f'{input_lines[0].rstrip('\n')}\\n{input_lines[1]}'] + + def test_export_with_noop_aggregator(self, parser, batch, noop_aggregator, client, attributes, monkeypatch): + input_lines = [f'log line {i}' for i in range(0, batch.max_size)] + exporter = BrontoExporter(client, parser, batch, noop_aggregator, attributes) + TestBrontoExporter.add_lines_to_file(input_lines, parser.input_file.filepath) + sent_lines = [] + monkeypatch.setattr(BrontoClient, 'send_data', lambda _, b, __: sent_lines.extend(b.get_data())) + exporter.export() + assert sent_lines == [input_line.rstrip('\n') for input_line in input_lines] diff --git a/tests/test_parser.py b/tests/test_parser.py index 89fabe9..13b357c 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -3,7 +3,7 @@ from config import (CLOUDFRONT_REALTIME_ACCESS_LOG_TYPE, ALB_ACCESS_LOG_TYPE, NLB_ACCESS_LOG_TYPE, CLOUDFRONT_STANDARD_ACCESS_LOG_TYPE, CLASSIC_LB_ACCESS_LOG_TYPE, CLOUDTRAIL_LOG_TYPE, - S3_ACCESS_LOG_TYPE) + S3_ACCESS_LOG_TYPE, CLOUDWATCH_LOG_TYPE) # https://docs.aws.amazon.com/elasticloadbalancing/latest/application/load-balancer-access-logs.html#access-log-entry-examples ALB_ACCESS_LOG_SAMPLE = 'https 2018-07-02T22:23:00.186641Z app/my-loadbalancer/50dc6c495c0c9188 192.168.131.39:2817 10.0.0.1:80 0.086 0.048 0.037 200 200 0 57 "GET https://www.example.com:443/ HTTP/1.1" "curl/7.46.0" ECDHE-RSA-AES128-GCM-SHA256 TLSv1.2 arn:aws:elasticloadbalancing:us-east-2:123456789012:targetgroup/my-targets/73e2d6bc24d8a067 "Root=1-58337281-1d84f3d73c47ec4e58577259" "www.example.com" "arn:aws:acm:us-east-2:123456789012:certificate/12345678-1234-1234-1234-123456789012" 1 2018-07-02T22:22:48.364000Z "authenticate,forward" "-" "-" "10.0.0.1:80" "200" "-" "-" TID_1234567890'