Normaliza métricas de uso e organiza índices do OpenSearch

VERSION

@@ -1 +1 @@
-2.3.7
+2.4.0

config/collections.py

@@ -1,3 +1,5 @@
+from log_manager_config.choices import OpenSearchPartitionStrategy
+
 COLLECTION_OPAC_URL_MAP = {
     "dom": "https://scielo.do/api/v1/counter_dict",
     "scl": "https://www.scielo.br/api/v1/counter_dict",
@@ -23,6 +25,7 @@
     },
     {
         "acronym": "chl",
+        "opensearch_partition_strategy": OpenSearchPartitionStrategy.YEARLY,
         "directory_name": "Site clássico",
         "path": "/app/logs/bkp-ratchet/scielo.cl",
         "quantity": 1,
@@ -33,6 +36,7 @@
     },
     {
         "acronym": "col",
+        "opensearch_partition_strategy": OpenSearchPartitionStrategy.YEARLY,
         "directory_name": "Site clássico",
         "path": "/app/logs/bkp-ratchet/scielo.co",
         "quantity": 1,
@@ -100,6 +104,7 @@
     },
     {
         "acronym": "mex",
+        "opensearch_partition_strategy": OpenSearchPartitionStrategy.YEARLY,
         "directory_name": "Site clássico",
         "path": "/app/logs/bkp-ratchet/scielo.mx",
         "quantity": 1,
@@ -166,6 +171,7 @@
     },
     {
         "acronym": "scl",
+        "opensearch_partition_strategy": OpenSearchPartitionStrategy.YEARLY,
         "directory_name": "SciELO Brasil",
         "path": "/app/logs/bkp-bunnynet/scielo-br",
         "quantity": 2,
@@ -176,6 +182,7 @@
     },
     {
         "acronym": "scl",
+        "opensearch_partition_strategy": OpenSearchPartitionStrategy.YEARLY,
         "directory_name": "SciELO Brasil 2",
         "path": "/app/logs/bkp-bunnynet/scielo-br-2",
         "quantity": 2,

config/settings/base.py

@@ -33,7 +33,6 @@
     "wid",
 )
 DEFAULT_PARSING_METADATA_CACHE_RELEASE_COLLECTIONS = ("scl",)
-DEFAULT_YEAR_PARTITIONED_COLLECTIONS = ("chl", "col", "mex", "scl")
 
 ROOT_DIR = Path(__file__).resolve(strict=True).parent.parent.parent
 # core/
@@ -357,6 +356,7 @@
     "metrics.tasks.daily_metric_exports",
     "metrics.tasks.index",
     "metrics.tasks.log_parsing",
+    "metrics.tasks.metadata_sync",
     "metrics.tasks.resume",
 )
 # http://docs.celeryproject.org/en/latest/userguide/configuration.html#std:setting-task_serializer
@@ -467,6 +467,10 @@
     "OPENSEARCH_BULK_CHUNK_SIZE",
     default=500,
 )
+OPENSEARCH_ROLLOVER_MAX_SIZE = env(
+    "OPENSEARCH_ROLLOVER_MAX_SIZE",
+    default="50gb",
+)
 
 # Resources
 # ------------------------------------------------------------------------------
@@ -532,7 +536,3 @@
     "PARSING_METADATA_CACHE_RELEASE_COLLECTIONS",
     default=list(DEFAULT_PARSING_METADATA_CACHE_RELEASE_COLLECTIONS),
 )
-YEAR_PARTITIONED_COLLECTIONS = env.list(
-    "YEAR_PARTITIONED_COLLECTIONS",
-    default=list(DEFAULT_YEAR_PARTITIONED_COLLECTIONS),
-)

log_manager_config/choices.py

@@ -0,0 +1,7 @@
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+
+
+class OpenSearchPartitionStrategy(models.TextChoices):
+    ROLLOVER = "rollover", _("Continuous with size rollover")
+    YEARLY = "yearly", _("One physical index per year")

log_manager_config/migrations/0006_logmanagercollectionconfig_opensearch_index_policy.py

@@ -0,0 +1,38 @@
+# Generated by Django 5.2.12 on 2026-09-11 01:56
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("log_manager_config", "0005_alter_collectionlogdirectory_translator_class"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="logmanagercollectionconfig",
+            name="opensearch_primary_shards",
+            field=models.PositiveSmallIntegerField(
+                default=1,
+                help_text="Applied only when a new physical index is created.",
+                verbose_name="OpenSearch Primary Shards",
+            ),
+        ),
+        migrations.AddField(
+            model_name="logmanagercollectionconfig",
+            name="opensearch_partition_strategy",
+            field=models.CharField(
+                choices=[
+                    ("rollover", "Continuous with size rollover"),
+                    ("yearly", "One physical index per year"),
+                ],
+                default="rollover",
+                help_text=(
+                    "Uses yearly physical indexes or continuous indexes "
+                    "rolled over by size."
+                ),
+                max_length=16,
+                verbose_name="OpenSearch Partition Strategy",
+            ),
+        ),
+    ]

log_manager_config/models.py

@@ -11,6 +11,7 @@
 
 from collection.models import Collection
 from core.models import CommonControlField
+from log_manager_config.choices import OpenSearchPartitionStrategy
 
 
 class LogManagerCollectionConfig(ClusterableModel, CommonControlField):
@@ -36,12 +37,28 @@ class LogManagerCollectionConfig(ClusterableModel, CommonControlField):
         verbose_name=_("Expected Logs Per Day"),
         default=1,
     )
+    opensearch_primary_shards = models.PositiveSmallIntegerField(
+        verbose_name=_("OpenSearch Primary Shards"),
+        default=1,
+        help_text=_("Applied only when a new physical index is created."),
+    )
+    opensearch_partition_strategy = models.CharField(
+        verbose_name=_("OpenSearch Partition Strategy"),
+        max_length=16,
+        choices=OpenSearchPartitionStrategy.choices,
+        default=OpenSearchPartitionStrategy.ROLLOVER,
+        help_text=_(
+            "Uses yearly physical indexes or continuous indexes rolled over by size."
+        ),
+    )
 
     panels = [
         AutocompletePanel("collection"),
         FieldPanel("sample_size"),
         FieldPanel("buffer_size"),
         FieldPanel("expected_logs_per_day"),
+        FieldPanel("opensearch_primary_shards"),
+        FieldPanel("opensearch_partition_strategy"),
         InlinePanel("directories", label=_("Directories")),
         InlinePanel("emails", label=_("Emails")),
     ]
@@ -68,6 +85,11 @@ def load(cls, data, user):
                 sample_size=item.get("sample_size", 0.1),
                 buffer_size=item.get("buffer_size", 2048),
                 expected_logs_per_day=item.get("quantity", 1),
+                opensearch_primary_shards=item.get("opensearch_primary_shards", 1),
+                opensearch_partition_strategy=item.get(
+                    "opensearch_partition_strategy",
+                    OpenSearchPartitionStrategy.ROLLOVER,
+                ),
             )
 
     @classmethod
@@ -78,6 +100,8 @@ def create_or_update(
         sample_size,
         buffer_size,
         expected_logs_per_day,
+        opensearch_primary_shards=1,
+        opensearch_partition_strategy=OpenSearchPartitionStrategy.ROLLOVER,
     ):
         obj, created = cls.objects.get_or_create(collection=collection)
         if created:
@@ -89,6 +113,8 @@ def create_or_update(
         obj.sample_size = sample_size
         obj.buffer_size = buffer_size
         obj.expected_logs_per_day = expected_logs_per_day
+        obj.opensearch_primary_shards = opensearch_primary_shards
+        obj.opensearch_partition_strategy = opensearch_partition_strategy
         obj.save()
         logging.info(f"Config for {collection.acron3} updated.")
         return obj

log_manager_config/tests/test_models.py

@@ -2,6 +2,7 @@
 
 from collection.models import Collection
 from core.users.tests.factories import UserFactory
+from log_manager_config.choices import OpenSearchPartitionStrategy
 from log_manager_config.models import CollectionLogDirectory, LogManagerCollectionConfig
 
 
@@ -17,12 +18,19 @@ def test_create_or_update_creates_config(self):
             sample_size=0.2,
             buffer_size=4096,
             expected_logs_per_day=3,
+            opensearch_primary_shards=2,
+            opensearch_partition_strategy=OpenSearchPartitionStrategy.YEARLY,
         )
 
         self.assertEqual(config.collection, self.collection)
         self.assertEqual(config.sample_size, 0.2)
         self.assertEqual(config.buffer_size, 4096)
         self.assertEqual(config.expected_logs_per_day, 3)
+        self.assertEqual(config.opensearch_primary_shards, 2)
+        self.assertEqual(
+            config.opensearch_partition_strategy,
+            OpenSearchPartitionStrategy.YEARLY,
+        )
 
     def test_create_or_update_updates_existing(self):
         LogManagerCollectionConfig.create_or_update(
@@ -38,11 +46,17 @@ def test_create_or_update_updates_existing(self):
             sample_size=0.5,
             buffer_size=8192,
             expected_logs_per_day=5,
+            opensearch_primary_shards=3,
         )
 
         self.assertEqual(LogManagerCollectionConfig.objects.count(), 1)
         self.assertEqual(config.sample_size, 0.5)
         self.assertEqual(config.buffer_size, 8192)
+        self.assertEqual(config.opensearch_primary_shards, 3)
+        self.assertEqual(
+            config.opensearch_partition_strategy,
+            OpenSearchPartitionStrategy.ROLLOVER,
+        )
 
 
 class CollectionLogDirectoryTests(TestCase):

log_manager_config/tests/test_tasks.py

@@ -53,6 +53,16 @@ def test_default_seed_matches_log_directories():
     assert {
         item["quantity"] for item in LOG_MANAGER_SEED_DATA if item["acronym"] == "scl"
     } == {2}
+    assert {
+        item.get("opensearch_primary_shards", 1)
+        for item in LOG_MANAGER_SEED_DATA
+        if item["acronym"] == "scl"
+    } == {1}
+    assert {
+        item["acronym"]
+        for item in LOG_MANAGER_SEED_DATA
+        if item.get("opensearch_partition_strategy", "rollover") == "yearly"
+    } == {"chl", "col", "mex", "scl"}
     assert not {
         "dom",
         "rvt",

metrics/apps.py

@@ -4,3 +4,6 @@
 class MetricsConfig(AppConfig):
     default_auto_field = "django.db.models.BigAutoField"
     name = "metrics"
+
+    def ready(self):
+        from metrics import signals  # noqa: F401

metrics/counter/access/accumulation.py

@@ -4,7 +4,7 @@
 from core.utils.date_utils import coerce_datetime
 
 
-def accumulate(results, counter_access, line):
+def accumulate(results, counter_access, line, reporting_date=None):
     access_url = counter_access.get("access_url") or _normalized_access_path(
         line.get("url")
     )
@@ -18,20 +18,21 @@ def accumulate(results, counter_access, line):
     if local_datetime is None:
         raise ValueError("Invalid local_datetime in parsed log line.")
 
+    reporting_date = reporting_date or local_datetime.date()
     access_datetime = local_datetime.replace(minute=0, second=0, microsecond=0)
     second_of_hour = local_datetime.minute * 60 + local_datetime.second
 
     session_key = (
         client_name,
         client_version,
         ip_address,
-        access_datetime.date().toordinal(),
+        reporting_date.toordinal(),
         access_datetime.hour,
     )
     raw_record = _build_record(
         counter_access=counter_access,
         line=line,
-        access_datetime=access_datetime,
+        reporting_date=reporting_date,
     )
     access_url_key = access_url or "|".join(
         [
@@ -52,7 +53,7 @@ def accumulate(results, counter_access, line):
 def _build_record(
     counter_access,
     line,
-    access_datetime,
+    reporting_date,
 ):
     collection = counter_access.get("collection")
     source_key = _source_key(counter_access, collection)
@@ -63,7 +64,7 @@ def _build_record(
     content_language = counter_access.get("media_language")
     content_type = counter_access.get("content_type")
     access_country_code = line.get("country_code")
-    access_date = access_datetime.strftime("%Y-%m-%d")
+    access_date = reporting_date.strftime("%Y-%m-%d")
 
     return {
         "collection": collection,
@@ -72,15 +73,12 @@ def _build_record(
         "pid_v2": pid_v2,
         "pid_v3": pid_v3,
         "pid_generic": pid_generic,
-        "document": _document_metadata(counter_access),
         "title_pid_generic": counter_access.get("title_pid_generic") or pid_generic,
         "media_format": media_format,
         "content_language": content_language,
         "content_type": content_type,
         "access_country_code": access_country_code,
         "access_date": access_date,
-        "access_year": access_date[:4],
-        "access_month": access_date[:7].replace("-", ""),
         "publication_year": counter_access.get("publication_year"),
         "counter_access_type": counter_access.get("counter_access_type") or "Open",
         "access_method": counter_access.get("access_method") or "Regular",
@@ -102,25 +100,10 @@ def _normalized_access_path(url):
     return path or None
 
 
-def _document_metadata(counter_access):
-    document_title = counter_access.get("document_title")
-    return {"title": document_title} if document_title else {}
-
-
 def _source_metadata(counter_access):
     return {
         "source_type": counter_access.get("source_type"),
         "source_id": counter_access.get("source_id"),
-        "scielo_issn": counter_access.get("scielo_issn"),
-        "main_title": counter_access.get("source_main_title"),
-        "identifiers": counter_access.get("source_identifiers"),
-        "access_type": counter_access.get("source_access_type"),
-        "city": counter_access.get("source_city"),
-        "country": counter_access.get("source_country"),
-        "subject_area_capes": counter_access.get("source_subject_area_capes"),
-        "subject_area_wos": counter_access.get("source_subject_area_wos"),
-        "acronym": counter_access.get("source_acronym"),
-        "publisher_name": counter_access.get("source_publisher_name"),
     }