diff --git a/estela_scrapy/extensions.py b/estela_scrapy/extensions.py index d95d5d8..38f2a33 100644 --- a/estela_scrapy/extensions.py +++ b/estela_scrapy/extensions.py @@ -10,7 +10,12 @@ from scrapy.exporters import PythonItemExporter from twisted.internet import task -from estela_scrapy.utils import json_serializer, producer, update_job +from estela_scrapy.utils import ( + get_obj_byte_size_id_padding_bytes, + json_serializer, + producer, + update_job, +) import logging @@ -45,7 +50,6 @@ DEFAULT_MAX_DUPLICATE_TRACKING = 100_000 DEFAULT_TIMELINE_BUCKETS = 30 - class BaseExtension: def __init__(self, stats, *args, **kwargs): self.stats = stats @@ -90,6 +94,10 @@ def __init__(self, stats, schema=None, unique_field=None, max_buckets=30): raise NotConfigured("REDIS_URL not found in the settings") self.redis_conn = redis.from_url(redis_url) + self._item_size_exporter = PythonItemExporter( + binary=False, dont_fail=True + ) + self.stats_key = os.getenv("REDIS_STATS_KEY") self.interval = float(os.getenv("REDIS_STATS_INTERVAL")) @@ -101,6 +109,10 @@ def __init__(self, stats, schema=None, unique_field=None, max_buckets=30): "false").lower() == "true" ) + self._item_obj_byte_size_id_padding_bytes = ( + get_obj_byte_size_id_padding_bytes() + ) + # Always initialize all metrics tracking (no conditional) self._init_metrics_tracking(schema, unique_field, max_buckets) @@ -158,6 +170,10 @@ def from_crawler(cls, crawler): max_buckets=max_buckets ) + from estela_scrapy.log import set_scrapy_stats + + set_scrapy_stats(crawler.stats) + crawler.signals.connect( ext.spider_opened, signal=signals.spider_opened) crawler.signals.connect( @@ -169,6 +185,9 @@ def from_crawler(cls, crawler): def spider_opened(self, spider): # Initialize duplicate counter self.stats.set_value("advanced_metrics/items_duplicates", 0) + self.stats.set_value("item_obj_byte_size", 0) + self.stats.set_value("request_obj_byte_size", 0) + self.stats.set_value("log_obj_byte_size", 0) update_job(self.job_url, self.auth_token, status=RUNNING_STATUS) self.task = task.LoopingCall(self.store_stats, spider) @@ -178,9 +197,28 @@ def item_scraped(self, item, spider): # Track metrics for this item self._track_item_metrics(item, spider) + def _item_json_payload_byte_size(self, exported_dict): + """UTF-8 byte length of JSON (same encoding as job payloads).""" + return len( + json.dumps(exported_dict, default=json_serializer).encode("utf-8") + ) + + def _item_stat_byte_size(self, item): + """JSON UTF-8 payload size (job shape) plus stored-id padding bytes. + + Padding is ``ESTELA_ITEM_OBJ_BYTE_SIZE_ID_PADDING_BYTES`` (default 17). + """ + exported = dict(self._item_size_exporter.export_item(item)) + return ( + self._item_json_payload_byte_size(exported) + + self._item_obj_byte_size_id_padding_bytes + ) + def _track_item_metrics(self, item, spider): """Track timeline and duplicate metrics for each scraped item""" + self.stats.inc_value("item_obj_byte_size", self._item_stat_byte_size(item)) + # Timeline tracking with cached time calculation (performance optimization) self.items_since_time_update += 1 if self.items_since_time_update >= TIME_CACHE_UPDATE_INTERVAL: diff --git a/estela_scrapy/log.py b/estela_scrapy/log.py index 5bccfc9..b190dcf 100644 --- a/estela_scrapy/log.py +++ b/estela_scrapy/log.py @@ -1,3 +1,4 @@ +import json import logging import os import sys @@ -7,16 +8,34 @@ from estela_queue_adapter import queue_noisy_libraries from twisted.python import log as txlog -from estela_scrapy.utils import producer, to_standard_str +from estela_scrapy.utils import ( + get_obj_byte_size_id_padding_bytes, + json_serializer, + producer, + to_standard_str, +) _stderr = sys.stderr +# Set from RedisStatsCollector.from_crawler; logging may run before the crawler exists. +_scrapy_stats = None + + +def set_scrapy_stats(stats): + global _scrapy_stats + _scrapy_stats = stats + def _logfn(level, message, parent="none"): data = { "jid": os.getenv("ESTELA_SPIDER_JOB"), "payload": {"log": str(message), "datetime": float(time.time())}, } + if _scrapy_stats is not None: + payload_byte_size = len( + json.dumps(data["payload"], default=json_serializer).encode("utf-8") + ) + get_obj_byte_size_id_padding_bytes() + _scrapy_stats.inc_value("log_obj_byte_size", payload_byte_size) producer.send("job_logs", data) diff --git a/estela_scrapy/middlewares.py b/estela_scrapy/middlewares.py index 46ef141..8f77261 100644 --- a/estela_scrapy/middlewares.py +++ b/estela_scrapy/middlewares.py @@ -1,3 +1,4 @@ +import json import logging import os import re @@ -7,7 +8,12 @@ from scrapy.utils.request import fingerprint as request_fingerprint from twisted.web import http -from estela_scrapy.utils import parse_time, producer +from estela_scrapy.utils import ( + get_obj_byte_size_id_padding_bytes, + json_serializer, + parse_time, + producer, +) proxy_logger = logging.getLogger("proxy_mw") @@ -26,18 +32,31 @@ def get_status_size(response_status): class StorageDownloaderMiddleware: + def __init__(self, stats=None): + self.stats = stats + + @classmethod + def from_crawler(cls, crawler): + return cls(stats=crawler.stats) + def process_response(self, request, response, spider): + payload = { + "url": response.url, + "status": int(response.status), + "method": request.method, + "duration": int(request.meta.get("download_latency", 0) * 1000), + "time": parse_time(), + "response_size": len(response.body), + "fingerprint": request_fingerprint(request), + } + if self.stats is not None: + payload_byte_size = len( + json.dumps(payload, default=json_serializer).encode("utf-8") + ) + get_obj_byte_size_id_padding_bytes() + self.stats.inc_value("request_obj_byte_size", payload_byte_size) data = { "jid": os.getenv("ESTELA_SPIDER_JOB"), - "payload": { - "url": response.url, - "status": int(response.status), - "method": request.method, - "duration": int(request.meta.get("download_latency", 0) * 1000), - "time": parse_time(), - "response_size": len(response.body), - "fingerprint": request_fingerprint(request), - }, + "payload": payload, } producer.send("job_requests", data) return response diff --git a/estela_scrapy/utils.py b/estela_scrapy/utils.py index 8f9a848..5e95dba 100644 --- a/estela_scrapy/utils.py +++ b/estela_scrapy/utils.py @@ -1,9 +1,28 @@ import json +import os from datetime import date, datetime, timedelta import requests from estela_queue_adapter import get_producer_interface +# Default padding for a stored document id (e.g. BSON ObjectId-style ``_id``) in +# ``*_obj_byte_size`` stats. Override with ``ESTELA_ITEM_OBJ_BYTE_SIZE_ID_PADDING_BYTES``. +DEFAULT_OBJ_BYTE_SIZE_ID_PADDING_BYTES = 17 + +_obj_byte_size_id_padding_cache = None + + +def get_obj_byte_size_id_padding_bytes(): + global _obj_byte_size_id_padding_cache + if _obj_byte_size_id_padding_cache is None: + _obj_byte_size_id_padding_cache = int( + os.getenv( + "ESTELA_ITEM_OBJ_BYTE_SIZE_ID_PADDING_BYTES", + str(DEFAULT_OBJ_BYTE_SIZE_ID_PADDING_BYTES), + ) + ) + return _obj_byte_size_id_padding_cache + def parse_time(date=None): if date is None: