From 64c8d84a6b2d52011b9f6050d2c4534f83677af2 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:13:52 -0300 Subject: [PATCH 01/14] =?UTF-8?q?Mede=20mem=C3=B3ria=20do=20processo=20e?= =?UTF-8?q?=20do=20cont=C3=AAiner?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/services/memory.py | 55 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 55 insertions(+) create mode 100644 metrics/services/memory.py diff --git a/metrics/services/memory.py b/metrics/services/memory.py new file mode 100644 index 0000000..7e06071 --- /dev/null +++ b/metrics/services/memory.py @@ -0,0 +1,55 @@ +import resource +from pathlib import Path + +_CGROUP_CURRENT_PATHS = ( + Path("/sys/fs/cgroup/memory.current"), + Path("/sys/fs/cgroup/memory/memory.usage_in_bytes"), +) +_CGROUP_PEAK_PATHS = ( + Path("/sys/fs/cgroup/memory.peak"), + Path("/sys/fs/cgroup/memory/memory.max_usage_in_bytes"), +) +_MIB = 1024 * 1024 + + +def snapshot(): + return { + "rss_mib": _current_rss_mib(), + "peak_rss_mib": resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024, + "cgroup_current_mib": _read_cgroup_mib(_CGROUP_CURRENT_PATHS), + "cgroup_peak_mib": _read_cgroup_mib(_CGROUP_PEAK_PATHS), + } + + +def format_snapshot(values=None): + values = values or snapshot() + parts = [ + f"RSS {values['rss_mib']:.1f} MiB", + f"peak RSS {values['peak_rss_mib']:.1f} MiB", + ] + if values["cgroup_current_mib"] is not None: + parts.append(f"cgroup current {values['cgroup_current_mib']:.1f} MiB") + if values["cgroup_peak_mib"] is not None: + parts.append(f"cgroup peak {values['cgroup_peak_mib']:.1f} MiB") + return "; ".join(parts) + + +def _current_rss_mib(): + try: + for line in Path("/proc/self/status").read_text().splitlines(): + if line.startswith("VmRSS:"): + return int(line.split()[1]) / 1024 + except (OSError, ValueError, IndexError): + pass + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 + + +def _read_cgroup_mib(paths): + for path in paths: + try: + value = path.read_text().strip() + if value != "max": + return int(value) / _MIB + except (OSError, ValueError): + continue + return None From d4bb90cc4addb68a8a2bf689391fa6e8bee152fb Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:00 -0300 Subject: [PATCH 02/14] =?UTF-8?q?Registra=20mem=C3=B3ria=20durante=20a=20e?= =?UTF-8?q?xporta=C3=A7=C3=A3o=20di=C3=A1ria?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/services/export.py | 11 +++-------- 1 file changed, 3 insertions(+), 8 deletions(-) diff --git a/metrics/services/export.py b/metrics/services/export.py index 3fa5b90..dce465f 100644 --- a/metrics/services/export.py +++ b/metrics/services/export.py @@ -1,5 +1,4 @@ import logging -import resource from itertools import chain from time import monotonic @@ -7,7 +6,7 @@ from metrics.opensearch.mappings import get_index_mappings from metrics.opensearch.names import generate_month_index_name, generate_year_index_name -from metrics.services import daily_payloads +from metrics.services import daily_payloads, memory def daily_metric_payload_exists(job): @@ -40,12 +39,12 @@ def export_daily_metric_payload(search_client, job): ) logging.info( "Daily metric job %s %s OpenSearch export completed in %.3f " - "seconds; %s documents; peak RSS %.1f MiB.", + "seconds; %s documents; %s.", job.pk, granularity, monotonic() - started, exported, - _peak_rss_mib(), + memory.format_snapshot(), ) @@ -86,7 +85,3 @@ def _sync_documents_group( document_items=chain((first_item,), document_items), job_id=job_id, ) - - -def _peak_rss_mib(): - return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 From c266339a8e6d95f44a79487c31b728c72ed44ed0 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:08 -0300 Subject: [PATCH 03/14] =?UTF-8?q?Exp=C3=B5e=20refer=C3=AAncias=20dos=20reg?= =?UTF-8?q?istros=20compactos?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/counter/access/daily_accumulator.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/metrics/counter/access/daily_accumulator.py b/metrics/counter/access/daily_accumulator.py index d376765..56e1793 100644 --- a/metrics/counter/access/daily_accumulator.py +++ b/metrics/counter/access/daily_accumulator.py @@ -167,6 +167,18 @@ def iter_materialized_values(self, consume=False): finally: self.clear() + def iter_materialized_record_items(self): + for record_key, record in self._records.items(): + yield record_key, record.as_dict(self) + + def iter_materialized_record_keys(self, record_keys, consume=False): + for record_key in record_keys: + if consume: + record = self._records.pop(record_key) + else: + record = self._records[record_key] + yield record.as_dict(self) + def clear(self): self._records.clear() self._documents.clear() From e532982a81c7f23d1422fffbf3be7708de7b242f Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:17 -0300 Subject: [PATCH 04/14] =?UTF-8?q?Define=20agrupamento=20de=20documentos=20?= =?UTF-8?q?por=20parti=C3=A7=C3=A3o?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/counter/indexing/engines/base.py | 3 +++ metrics/counter/indexing/engines/book.py | 11 +++++++++++ 2 files changed, 14 insertions(+) diff --git a/metrics/counter/indexing/engines/base.py b/metrics/counter/indexing/engines/base.py index 902cf83..e0ca076 100644 --- a/metrics/counter/indexing/engines/base.py +++ b/metrics/counter/indexing/engines/base.py @@ -29,6 +29,9 @@ def accumulate(self, data, unique_state, value, granularity): is_request_event=is_request(value.get("content_type")), ) + def partition_key(self, value, granularity): + return self._generate_document_id(value, granularity) + def _generate_document_id( self, value, granularity, metric_scope=None, pid_generic=None ): diff --git a/metrics/counter/indexing/engines/book.py b/metrics/counter/indexing/engines/book.py index 0ec3bd2..e52b570 100644 --- a/metrics/counter/indexing/engines/book.py +++ b/metrics/counter/indexing/engines/book.py @@ -8,6 +8,17 @@ class BookPipeline(DocumentPipeline): + def partition_key(self, value, granularity): + title_pid_generic = _extract_title_pid_generic(value) + if title_pid_generic: + return self._generate_document_id( + value, + granularity, + metric_scope="title", + pid_generic=title_pid_generic, + ) + return self._generate_document_id(value, granularity) + def accumulate(self, data, unique_state, value, granularity): if not isinstance(value, dict): return From 1eb0b510e10b0b03749a76dedf0917e61a099647 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:27 -0300 Subject: [PATCH 05/14] =?UTF-8?q?Converte=20m=C3=A9tricas=20em=20parti?= =?UTF-8?q?=C3=A7=C3=B5es=20determin=C3=ADsticas?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/counter/indexing/converter.py | 94 +++++++++++++++++++++++---- 1 file changed, 83 insertions(+), 11 deletions(-) diff --git a/metrics/counter/indexing/converter.py b/metrics/counter/indexing/converter.py index d6a1653..c4c73a6 100644 --- a/metrics/counter/indexing/converter.py +++ b/metrics/counter/indexing/converter.py @@ -1,3 +1,5 @@ +import hashlib + from metrics.counter.indexing.engines.article import ArticlePipeline from metrics.counter.indexing.engines.base import DocumentPipeline from metrics.counter.indexing.engines.book import BookPipeline @@ -12,22 +14,92 @@ "chapter": BookPipeline(), } _DEFAULT = DocumentPipeline() +_DEFAULT_PARTITION_COUNT = 64 + + +def iter_partitioned_documents( + accumulator, + granularity, + partition_count=_DEFAULT_PARTITION_COUNT, +): + if partition_count <= 0: + raise ValueError("Partition count must be greater than zero.") + + consume = granularity == "year" + partitions = [[] for _ in range(partition_count)] + for record_key, value in accumulator.iter_materialized_record_items(): + partition = _partition_for_value(value, granularity, partition_count) + partitions[partition].append(record_key) + + try: + for record_keys in partitions: + values = accumulator.iter_materialized_record_keys( + record_keys, + consume=consume, + ) + yield from _convert_partition(values, granularity) + record_keys.clear() + finally: + for record_keys in partitions: + record_keys.clear() + partitions.clear() + if consume: + accumulator.clear() + + +def iter_partitioned_values( + values, + granularity, + partition_count=_DEFAULT_PARTITION_COUNT, +): + if partition_count <= 0: + raise ValueError("Partition count must be greater than zero.") + partitions = [[] for _ in range(partition_count)] + for value in values: + partition = _partition_for_value(value, granularity, partition_count) + partitions[partition].append(value) + + try: + for partition_values in partitions: + yield from _convert_partition(partition_values, granularity) + partition_values.clear() + finally: + for partition_values in partitions: + partition_values.clear() + partitions.clear() + + +def _partition_for_value(value, granularity, partition_count): + pipeline = _get_pipeline(value) + partition_key = pipeline.partition_key(value, granularity) + digest = hashlib.blake2b( + partition_key.encode("utf-8"), + digest_size=8, + ).digest() + return int.from_bytes(digest, "big") % partition_count -def convert_granularity(values, granularity): + +def _convert_partition(values, granularity): converted_data = {} unique_state = _initialize_unique_state() - for value in values: - pipeline = _get_pipeline(value) - pipeline.accumulate( - data=converted_data, - unique_state=unique_state, - value=value, - granularity=granularity, - ) - - return converted_data + try: + for value in values: + pipeline = _get_pipeline(value) + pipeline.accumulate( + data=converted_data, + unique_state=unique_state, + value=value, + granularity=granularity, + ) + + for document_id in sorted(converted_data): + yield document_id, converted_data[document_id] + finally: + converted_data.clear() + for bucket in unique_state.values(): + bucket.clear() def _get_pipeline(value): From acad21d2c7c5ea491dbf8671ef1b65da767fd301 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:35 -0300 Subject: [PATCH 06/14] =?UTF-8?q?Testa=20convers=C3=A3o=20e=20libera=C3=A7?= =?UTF-8?q?=C3=A3o=20das=20parti=C3=A7=C3=B5es?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tests/counter/access/test_accumulation.py | 4 ++-- .../counter/access/test_daily_accumulator.py | 19 +++++++++++++++++++ .../tests/counter/indexing/test_converter.py | 10 +++++----- metrics/tests/helpers.py | 4 ++-- 4 files changed, 28 insertions(+), 9 deletions(-) diff --git a/metrics/tests/counter/access/test_accumulation.py b/metrics/tests/counter/access/test_accumulation.py index 8f5366c..6c4c45e 100644 --- a/metrics/tests/counter/access/test_accumulation.py +++ b/metrics/tests/counter/access/test_accumulation.py @@ -223,8 +223,8 @@ def test_compact_accumulator_preserves_metrics_after_repeated_events(self): accumulation.accumulate(compact, self._book_counter_access(), event) values = list(compact.iter_materialized_values()) - month = index_docs.convert_granularity(iter(values), "month") - year = index_docs.convert_granularity(iter(values), "year") + month = dict(index_docs.iter_partitioned_values(values, "month")) + year = dict(index_docs.iter_partitioned_values(values, "year")) self.assertEqual(len(month), 2) self.assertEqual(len(year), 2) diff --git a/metrics/tests/counter/access/test_daily_accumulator.py b/metrics/tests/counter/access/test_daily_accumulator.py index 4505b5f..819dcbf 100644 --- a/metrics/tests/counter/access/test_daily_accumulator.py +++ b/metrics/tests/counter/access/test_daily_accumulator.py @@ -62,3 +62,22 @@ def test_consuming_materialization_releases_structures_after_consumer_error(): assert len(accumulator) == 0 assert accumulator._documents == [] + + +def test_partitioned_year_conversion_releases_accumulator_after_consumer_error(): + from metrics.counter.indexing import converter + + accumulator = DailyAccessAccumulator() + _accumulate(accumulator, "first", "127.0.0.1") + _accumulate(accumulator, "second", "127.0.0.2") + documents = converter.iter_partitioned_documents( + accumulator, + "year", + partition_count=2, + ) + + next(documents) + documents.close() + + assert len(accumulator) == 0 + assert accumulator._documents == [] diff --git a/metrics/tests/counter/indexing/test_converter.py b/metrics/tests/counter/indexing/test_converter.py index 1bc3888..45cd93e 100644 --- a/metrics/tests/counter/indexing/test_converter.py +++ b/metrics/tests/counter/indexing/test_converter.py @@ -13,8 +13,8 @@ def _convert(data): values = list(data.values()) return { - "month": index_docs.convert_granularity(iter(values), "month"), - "year": index_docs.convert_granularity(iter(values), "year"), + "month": dict(index_docs.iter_partitioned_values(values, "month")), + "year": dict(index_docs.iter_partitioned_values(values, "year")), } @@ -434,8 +434,8 @@ def test_double_click_collapses_same_url_within_30_seconds(self): values = list(results.iter_materialized_values()) metrics_data = { - "month": index_docs.convert_granularity(iter(values), "month"), - "year": index_docs.convert_granularity(iter(values), "year"), + "month": dict(index_docs.iter_partitioned_values(values, "month")), + "year": dict(index_docs.iter_partitioned_values(values, "year")), } month_item = metrics_data["month"][ "books|c2248|||BOOK:C2248/CHAPTER:03|2024-01|Open|Regular|2018" @@ -481,4 +481,4 @@ def test_article_pipeline_sets_journal_parent(self): self.assertEqual(month_doc["total_investigations"], 1) def test_empty_iterable_returns_empty(self): - self.assertEqual(index_docs.convert_granularity(iter(()), "month"), {}) + self.assertEqual(dict(index_docs.iter_partitioned_values((), "month")), {}) diff --git a/metrics/tests/helpers.py b/metrics/tests/helpers.py index 0d6a47c..c4d0de6 100644 --- a/metrics/tests/helpers.py +++ b/metrics/tests/helpers.py @@ -4,6 +4,6 @@ def convert_accumulator(accumulator): values = list(accumulator.iter_materialized_values()) return { - "month": converter.convert_granularity(iter(values), "month"), - "year": converter.convert_granularity(iter(values), "year"), + "month": dict(converter.iter_partitioned_values(values, "month")), + "year": dict(converter.iter_partitioned_values(values, "year")), } From d9f6121b7ce14bcf2f6010692aa13e84d37621a7 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:45 -0300 Subject: [PATCH 07/14] Escreve documentos do payload de forma incremental --- metrics/services/daily_payloads.py | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/metrics/services/daily_payloads.py b/metrics/services/daily_payloads.py index 132bb81..2580447 100644 --- a/metrics/services/daily_payloads.py +++ b/metrics/services/daily_payloads.py @@ -57,18 +57,28 @@ def __enter__(self): self._write_text(',"documents":{"month":') return self - def write_documents(self, granularity, documents): + def write_document_items(self, granularity, document_items): if granularity != self.next_granularity: raise RuntimeError( f"Expected {self.next_granularity} documents, got {granularity}." ) - self._write_json(documents) + document_count = 0 + self._write_text("{") + for document_id, document in document_items: + if document_count: + self._write_text(",") + self._write_json(document_id) + self._write_text(":") + self._write_json(document) + document_count += 1 + self._write_text("}") if granularity == "month": self._write_text(',"year":') self.next_granularity = "year" else: self.next_granularity = None + return document_count def finalize(self, input_log_hashes, summary): if self.next_granularity is not None: From cfe7a0deef5933c721f49a407506935858588ab5 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:14:53 -0300 Subject: [PATCH 08/14] =?UTF-8?q?Testa=20escrita=20determin=C3=ADstica=20d?= =?UTF-8?q?o=20payload=20di=C3=A1rio?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/tests/services/test_daily_payloads.py | 40 ++++++++++++++++--- 1 file changed, 34 insertions(+), 6 deletions(-) diff --git a/metrics/tests/services/test_daily_payloads.py b/metrics/tests/services/test_daily_payloads.py index d3f169b..826fa9d 100644 --- a/metrics/tests/services/test_daily_payloads.py +++ b/metrics/tests/services/test_daily_payloads.py @@ -44,8 +44,12 @@ def test_incremental_writer_preserves_canonical_bytes_and_hash(self): payload["collection"], payload["access_date"], ) as writer: - writer.write_documents("month", payload["documents"]["month"]) - writer.write_documents("year", payload["documents"]["year"]) + writer.write_document_items( + "month", sorted(payload["documents"]["month"].items()) + ) + writer.write_document_items( + "year", sorted(payload["documents"]["year"].items()) + ) payload_hash = writer.finalize( payload["input_log_hashes"], payload["summary"], @@ -72,8 +76,8 @@ def test_iter_document_items_reads_each_granularity_incrementally(self): payload["collection"], payload["access_date"], ) as writer: - writer.write_documents("month", payload["documents"]["month"]) - writer.write_documents("year", payload["documents"]["year"]) + writer.write_document_items("month", payload["documents"]["month"].items()) + writer.write_document_items("year", payload["documents"]["year"].items()) writer.finalize(payload["input_log_hashes"], payload["summary"]) self.assertEqual( @@ -97,8 +101,32 @@ def test_incremental_writer_removes_temporary_file_after_error(self): "scl", "2026-08-25", ) as writer: - writer.write_documents("month", {}) - writer.write_documents("year", {"invalid": object()}) + writer.write_document_items("month", iter(())) + writer.write_document_items("year", [("invalid", object())]) self.assertEqual(resolved_path.read_bytes(), b"previous canonical payload") self.assertFalse(resolved_path.with_suffix(".json.tmp").exists()) + + def test_document_item_order_produces_deterministic_payload(self): + hashes = [] + contents = [] + + for filename in ("first.json", "second.json"): + storage_path = Path("scl/2026/08") / filename + with daily_payloads.DailyPayloadWriter( + storage_path, + "scl", + "2026-08-25", + ) as writer: + writer.write_document_items( + "month", + [("b", {"total": 2}), ("a", {"total": 1})], + ) + writer.write_document_items("year", [("c", {"total": 3})]) + hashes.append(writer.finalize(["abc"], {"valid_lines": 1})) + contents.append( + daily_payloads.resolve_storage_path(storage_path).read_bytes() + ) + + self.assertEqual(hashes[0], hashes[1]) + self.assertEqual(contents[0], contents[1]) From 16779c9a53594933cb081bffaedd4b81e8b06ec1 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:15:02 -0300 Subject: [PATCH 09/14] =?UTF-8?q?Integra=20parti=C3=A7=C3=B5es=20=C3=A0=20?= =?UTF-8?q?gera=C3=A7=C3=A3o=20do=20payload=20di=C3=A1rio?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/services/parsing/job_payloads.py | 67 +++++++++++------------- 1 file changed, 31 insertions(+), 36 deletions(-) diff --git a/metrics/services/parsing/job_payloads.py b/metrics/services/parsing/job_payloads.py index 7fb1ebf..94991c5 100644 --- a/metrics/services/parsing/job_payloads.py +++ b/metrics/services/parsing/job_payloads.py @@ -1,14 +1,15 @@ import gc import logging -import resource from time import monotonic from django.conf import settings +from config.collections import get_collection_size from log_manager.models import LogFile from metrics.counter.access.daily_accumulator import DailyAccessAccumulator from metrics.counter.indexing import converter as index_docs -from metrics.services import daily_payloads +from metrics.services import daily_payloads, memory +from metrics.services.parsing import metadata_cache from metrics.services.parsing.environment import setup_parsing_environment from metrics.services.parsing.lines import process_line from metrics.services.parsing.log_files import ( @@ -41,13 +42,22 @@ def build_daily_metric_job_payload(job, robots_list, mmdb, track_errors=False): _merge_log_summary(summary, log_summary) logging.info( "Daily metric job %s parsing completed in %.3f seconds; " - "%s compact records; peak RSS %.1f MiB.", + "%s compact records; %s.", job.pk, monotonic() - parsing_started, len(results), - _peak_rss_mib(), + memory.format_snapshot(), ) + if get_collection_size(job.collection.acron3) == "xlarge": + metadata_cache.clear() + gc.collect() + logging.info( + "Daily metric job %s released parsing metadata cache; %s.", + job.pk, + memory.format_snapshot(), + ) + return _write_job_payload(job, results, summary) @@ -142,7 +152,7 @@ def _write_job_payload(job, results, summary): ) month_document_count = 0 year_document_count = 0 - serialization_seconds = 0 + payload_started = monotonic() with daily_payloads.DailyPayloadWriter( storage_path=storage_path, @@ -150,57 +160,46 @@ def _write_job_payload(job, results, summary): access_date=job.access_date.isoformat(), ) as writer: month_started = monotonic() - month_documents = index_docs.convert_granularity( - results.iter_materialized_values(), + month_documents = index_docs.iter_partitioned_documents( + results, "month", ) - month_document_count = len(month_documents) + month_document_count = writer.write_document_items("month", month_documents) month_conversion_seconds = monotonic() - month_started - - serialization_started = monotonic() - writer.write_documents("month", month_documents) - serialization_seconds += monotonic() - serialization_started - del month_documents gc.collect() logging.info( - "Daily metric job %s monthly conversion completed in %.3f seconds; " - "%s documents; peak RSS %.1f MiB.", + "Daily metric job %s monthly conversion and serialization completed " + "in %.3f seconds; %s documents; %s.", job.pk, month_conversion_seconds, month_document_count, - _peak_rss_mib(), + memory.format_snapshot(), ) year_started = monotonic() - year_documents = index_docs.convert_granularity( - results.iter_materialized_values(consume=True), + year_documents = index_docs.iter_partitioned_documents( + results, "year", ) - year_document_count = len(year_documents) - year_conversion_seconds = monotonic() - year_started + year_document_count = writer.write_document_items("year", year_documents) del results - - serialization_started = monotonic() - writer.write_documents("year", year_documents) - del year_documents payload_hash = writer.finalize(summary["input_log_hashes"], summary) - serialization_seconds += monotonic() - serialization_started + year_conversion_seconds = monotonic() - year_started gc.collect() logging.info( - "Daily metric job %s yearly conversion completed in %.3f seconds; " - "%s documents; peak RSS %.1f MiB.", + "Daily metric job %s yearly conversion and serialization completed " + "in %.3f seconds; %s documents; %s.", job.pk, year_conversion_seconds, year_document_count, - _peak_rss_mib(), + memory.format_snapshot(), ) logging.info( - "Daily metric job %s serialization completed in %.3f seconds; " - "peak RSS %.1f MiB.", + "Daily metric job %s payload generation completed in %.3f seconds; " "%s.", job.pk, - serialization_seconds, - _peak_rss_mib(), + monotonic() - payload_started, + memory.format_snapshot(), ) job.input_log_hashes = summary["input_log_hashes"] @@ -221,7 +220,3 @@ def _write_job_payload(job, results, summary): ] ) return storage_path.as_posix(), payload_hash - - -def _peak_rss_mib(): - return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024 From 7a0e97a62040e0092e5c3f58d9960dd7bf17f285 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:15:11 -0300 Subject: [PATCH 10/14] =?UTF-8?q?Testa=20libera=C3=A7=C3=A3o=20do=20acumul?= =?UTF-8?q?ador=20no=20job=20di=C3=A1rio?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/tests/services/test_daily_jobs.py | 27 +---------------------- 1 file changed, 1 insertion(+), 26 deletions(-) diff --git a/metrics/tests/services/test_daily_jobs.py b/metrics/tests/services/test_daily_jobs.py index 0db79c4..57ef4d6 100644 --- a/metrics/tests/services/test_daily_jobs.py +++ b/metrics/tests/services/test_daily_jobs.py @@ -1,5 +1,4 @@ import tempfile -import weakref from datetime import date, timedelta from types import SimpleNamespace from unittest.mock import Mock, patch @@ -23,10 +22,6 @@ ) -class TrackableDict(dict): - pass - - class DailyMetricJobServiceTests(TestCase): def setUp(self): self.collection = Collection.objects.create(acron3="books", acron2="bk") @@ -142,10 +137,6 @@ def test_mark_daily_metric_job_exported_sets_status_and_timestamp(self): self.assertEqual(job.status, DailyMetricJob.STATUS_EXPORTED) self.assertIsNotNone(job.exported_at) - @patch( - "metrics.services.parsing.job_payloads.index_docs.convert_granularity", - side_effect=({}, {}), - ) @patch( "metrics.services.parsing.job_payloads.process_line", return_value=(True, None) ) @@ -154,7 +145,6 @@ def test_build_daily_metric_job_payload_uses_only_input_log_hashes( self, mock_setup_parsing_environment, mock_process_line, - mock_convert_documents, ): selected = self._log_file("1" * 32) extra = self._log_file("2" * 32) @@ -217,11 +207,7 @@ def test_build_daily_metric_job_payload_rejects_missing_input_hashes(self): job, robots_list=["robot"], mmdb=Mock(data={}) ) - @patch("metrics.services.parsing.job_payloads.index_docs.convert_granularity") - def test_payload_generation_releases_each_granularity_and_accumulator( - self, - mock_convert, - ): + def test_payload_generation_releases_accumulator(self): job = DailyMetricJob.objects.create( collection=self.collection, access_date=date(2012, 3, 10), @@ -235,15 +221,6 @@ def test_payload_generation_releases_each_granularity_and_accumulator( url="/book", second=5, ) - references = {} - - def convert(values, granularity): - list(values) - documents = TrackableDict() - references[granularity] = weakref.ref(documents) - return documents - - mock_convert.side_effect = convert summary = { "log_files": 1, "input_log_hashes": ["1" * 32], @@ -256,7 +233,5 @@ def convert(values, granularity): with self.settings(MEDIA_ROOT=media_root): _write_job_payload(job, accumulator, summary) - self.assertIsNone(references["month"]()) - self.assertIsNone(references["year"]()) self.assertEqual(len(accumulator), 0) self.assertEqual(accumulator._documents, []) From c01ac2242fe01ab495fe4d5c04ccb4f61a33cd00 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:15:23 -0300 Subject: [PATCH 11/14] =?UTF-8?q?Adapta=20o=20teste=20da=20exporta=C3=A7?= =?UTF-8?q?=C3=A3o=20incremental?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/tests/services/test_daily_metric_exports.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/metrics/tests/services/test_daily_metric_exports.py b/metrics/tests/services/test_daily_metric_exports.py index c24fd88..e6850b3 100644 --- a/metrics/tests/services/test_daily_metric_exports.py +++ b/metrics/tests/services/test_daily_metric_exports.py @@ -39,13 +39,13 @@ def _write_payload(self): "scl", "2026-08-25", ) as writer: - writer.write_documents( + writer.write_document_items( "month", - {"month-doc": {"access": {"month": "2026-08"}}}, + [("month-doc", {"access": {"month": "2026-08"}})], ) - writer.write_documents( + writer.write_document_items( "year", - {"year-doc": {"access": {"year": "2026"}}}, + [("year-doc", {"access": {"year": "2026"}})], ) writer.finalize(["abc"], {"valid_lines": 1}) From 7dcfc42937291858bc7e1fd8bf95fccc589082f9 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:15:31 -0300 Subject: [PATCH 12/14] =?UTF-8?q?Atualiza=20a=20vers=C3=A3o=20para=202.3.4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/VERSION b/VERSION index 0bee604..3f684d2 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.3.3 +2.3.4 From d6bc63dccd5963aabf8b2f0aff6ee8cb1dd5bbb7 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:25:57 -0300 Subject: [PATCH 13/14] =?UTF-8?q?Configura=20lote=20e=20compress=C3=A3o=20?= =?UTF-8?q?do=20OpenSearch=20por=20ambiente?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- config/settings/base.py | 8 ++++++++ metrics/opensearch/client.py | 27 ++++++++++++++++++++------- 2 files changed, 28 insertions(+), 7 deletions(-) diff --git a/config/settings/base.py b/config/settings/base.py index f6780e1..fb0fea5 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -433,6 +433,14 @@ "OPENSEARCH_VERIFY_CERTS", default=False, ) +OPENSEARCH_HTTP_COMPRESS = env.bool( + "OPENSEARCH_HTTP_COMPRESS", + default=True, +) +OPENSEARCH_BULK_CHUNK_SIZE = env.int( + "OPENSEARCH_BULK_CHUNK_SIZE", + default=500, +) # Resources # ------------------------------------------------------------------------------ diff --git a/metrics/opensearch/client.py b/metrics/opensearch/client.py index 9dd2fb3..ffb34d9 100644 --- a/metrics/opensearch/client.py +++ b/metrics/opensearch/client.py @@ -10,13 +10,25 @@ merge_metric_document, ) -_BULK_CHUNK_SIZE = 500 - class OpenSearchUsageClient: def __init__(self, url=None, basic_auth=None, api_key=None, verify_certs=None): + self.bulk_chunk_size = getattr( + settings, + "OPENSEARCH_BULK_CHUNK_SIZE", + 500, + ) + if self.bulk_chunk_size <= 0: + raise ValueError("OpenSearch bulk chunk size must be greater than zero.") + self.client = self.get_opensearch_client(url, basic_auth, api_key, verify_certs) - logging.info("OpenSearch HTTP request compression is enabled.") + logging.info( + "OpenSearch HTTP request compression is %s; bulk chunk size is %s.", + "enabled" + if getattr(settings, "OPENSEARCH_HTTP_COMPRESS", True) + else "disabled", + self.bulk_chunk_size, + ) def get_opensearch_client( self, @@ -30,25 +42,26 @@ def get_opensearch_client( api_key = api_key or getattr(settings, "OPENSEARCH_API_KEY", None) if verify_certs is None: verify_certs = getattr(settings, "OPENSEARCH_VERIFY_CERTS", False) + http_compress = getattr(settings, "OPENSEARCH_HTTP_COMPRESS", True) if basic_auth: return OpenSearch( url, http_auth=tuple(basic_auth), verify_certs=verify_certs, - http_compress=True, + http_compress=http_compress, ) if api_key: return OpenSearch( url, api_key=api_key, verify_certs=verify_certs, - http_compress=True, + http_compress=http_compress, ) return OpenSearch( url, verify_certs=verify_certs, - http_compress=True, + http_compress=http_compress, ) def ping(self): @@ -138,7 +151,7 @@ def increment_document_items_for_daily_job( ) for doc_id, document in document_items ), - chunk_size=_BULK_CHUNK_SIZE, + chunk_size=self.bulk_chunk_size, ) return succeeded From c9b66761d9fc63e11111c6f2d23fd960aa21ec93 Mon Sep 17 00:00:00 2001 From: Rafael Damaceno Date: Fri, 4 Sep 2026 13:26:07 -0300 Subject: [PATCH 14/14] =?UTF-8?q?Testa=20configura=C3=A7=C3=A3o=20do=20env?= =?UTF-8?q?io=20ao=20OpenSearch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- metrics/tests/opensearch/test_client.py | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/metrics/tests/opensearch/test_client.py b/metrics/tests/opensearch/test_client.py index d3ec0db..227202b 100644 --- a/metrics/tests/opensearch/test_client.py +++ b/metrics/tests/opensearch/test_client.py @@ -37,6 +37,7 @@ def test_create_index_sends_mappings_in_request_body(self, mock_get_client): OPENSEARCH_VERIFY_CERTS=True, OPENSEARCH_BASIC_AUTH=None, OPENSEARCH_API_KEY=None, + OPENSEARCH_HTTP_COMPRESS=True, ) @patch("metrics.opensearch.client.OpenSearch") def test_verify_certs_false_explicitly_overrides_settings(self, mock_opensearch): @@ -51,6 +52,21 @@ def test_verify_certs_false_explicitly_overrides_settings(self, mock_opensearch) http_compress=True, ) + @override_settings( + OPENSEARCH_BASIC_AUTH=None, + OPENSEARCH_API_KEY=None, + OPENSEARCH_HTTP_COMPRESS=False, + ) + @patch("metrics.opensearch.client.OpenSearch") + def test_http_compression_can_be_disabled(self, mock_opensearch): + OpenSearchUsageClient(url="https://example.org:9200") + + mock_opensearch.assert_called_once_with( + "https://example.org:9200", + verify_certs=False, + http_compress=False, + ) + def test_get_index_mappings_returns_books_specific_mappings(self): self.assertIs( get_index_mappings("books", "month"), @@ -95,6 +111,7 @@ def test_get_index_mappings_returns_books_specific_mappings(self): ) self.assertFalse(source_mapping["properties"]["publisher_name"]["index"]) + @override_settings(OPENSEARCH_BULK_CHUNK_SIZE=2000) @patch("metrics.opensearch.client.helpers.bulk") @patch.object(OpenSearchUsageClient, "get_opensearch_client") def test_increment_documents_for_daily_job_uses_applied_jobs( @@ -138,4 +155,9 @@ def test_increment_documents_for_daily_job_uses_applied_jobs( ) self.assertEqual(action["upsert"], {"applied_jobs": []}) self.assertEqual(succeeded, 1) - self.assertEqual(mock_bulk.call_args.kwargs["chunk_size"], 500) + self.assertEqual(mock_bulk.call_args.kwargs["chunk_size"], 2000) + + @override_settings(OPENSEARCH_BULK_CHUNK_SIZE=0) + def test_bulk_chunk_size_must_be_positive(self): + with self.assertRaisesRegex(ValueError, "greater than zero"): + OpenSearchUsageClient(url="https://example.org:9200")