chore(octopus): log repeated GraphQL responses and slot lists briefly instead of in full

apps/predbat/const.py

@@ -168,3 +168,7 @@
 # need the same answer to "which of these two states is the more significant one to show".
 CHARGE_STATE_PRECEDENCE = ["Charging", "Freeze charging", "Hold charging"]
 EXPORT_STATE_PRECEDENCE = ["Exporting", "Freeze exporting", "Hold exporting"]
+
+# A log text that repeats (an API response, an Intelligent slot list) is logged in full again after this long, so
+# the full text is never only in a rotated-out log file; in between it is logged as one short line
+REPEAT_FULL_LOG_SECONDS = 3600

apps/predbat/octopus.py

@@ -18,8 +18,8 @@
 import re
 from datetime import datetime, timedelta, timezone
 from predbat_metrics import record_api_call
-from const import TIME_FORMAT, TIME_FORMAT_OCTOPUS, DISPATCH_SOURCE_CHARGER_SCHEDULE
-from utils import str2time, minutes_to_time, dp1, dp2, dp4, minute_data, round_out_to_period, filter_payment_method, is_edge_block_body, token_mint_backoff_seconds, TOKEN_MINT_BACKOFF_LOG_INTERVAL_SECONDS
+from const import TIME_FORMAT, TIME_FORMAT_OCTOPUS, DISPATCH_SOURCE_CHARGER_SCHEDULE, REPEAT_FULL_LOG_SECONDS
+from utils import RepeatLogGate, str2time, minutes_to_time, dp1, dp2, dp4, minute_data, round_out_to_period, filter_payment_method, is_edge_block_body, token_mint_backoff_seconds, TOKEN_MINT_BACKOFF_LOG_INTERVAL_SECONDS
 from component_base import ComponentBase
 from mock_base import MockBase as SharedMockBase
 import aiohttp
@@ -652,6 +652,8 @@ def initialize(self, key, account_id, automatic):
         # API request metrics for monitoring
         self.requests_total = 0
         self.failures_total = 0
+        # Repeated GraphQL requests and responses, and skipped Happy Hours, are logged briefly between full logs
+        self.log_gate = RepeatLogGate(REPEAT_FULL_LOG_SECONDS)
 
         # In-memory cache for product info (keyed by product_code) to avoid repeated API calls
         self._product_info_cache = {}
@@ -1361,7 +1363,8 @@ def get_saving_session_data(self):
             # v19.0.1 for the same reason; match that. The reward/code/type maps are populated above
             # first, because a joined Happy Hour takes its event type from this list.
             if event.get("eventType", None) == "WEEKEND_HAPPY_HOUR":
-                self.log("OctopusAPI: Not offering Weekend Happy Hour event code {} as available - it cannot be joined through the API".format(code))
+                # Logged once an hour per event (by id, so events with no code are told apart): the event list is re-read every poll
+                self.log_gate.log(self.log, "happy-hour", (event_id, code), ["OctopusAPI: Not offering Weekend Happy Hour event code {} as available - it cannot be joined through the API".format(code)], None)
                 continue
             target_regions = [region.get("regionId") for region in (event.get("targetRegion", None) or []) if region]
             if target_regions and account_region_id not in target_regions:
@@ -1646,7 +1649,6 @@ async def async_get_saving_sessions(self, account_id):
         if response_data is None:
             return self.saving_sessions
         else:
-            self.log("OctopusAPI: Fetched saving sessions data from GraphQL API: {}".format(response_data))
             savingSessions = response_data.get("savingSessions", {})
             if savingSessions is None:
                 savingSessions = {}
@@ -2328,6 +2330,15 @@ async def async_graphql_query(self, query, request_context, returns_data=True, i
             if returns_data:
                 self.log(f"Warn: OctopusAPI: Failed to retrieve data from graphql query {request_context} - token refresh failed")
             return None
+        # Whether the request was logged in full, and its full line, so a failure can show the request beside it
+        request_logged = [True, None]
+
+        def log_request_beside_failure():
+            """Log the full request beside a failure when it was only logged briefly ("as logged in full at ...")."""
+            if not request_logged[0] and request_logged[1]:
+                self.log(request_logged[1])
+                request_logged[0] = True
+
         try:
             self.requests_total += 1
             client = await self.api.async_create_client_session()
@@ -2338,13 +2349,17 @@ async def async_graphql_query(self, query, request_context, returns_data=True, i
             headers = {"Authorization": f"{auth_prefix}{self.graphql_token}", integration_context_header: request_context}
             # Redact the Authorization header so the JWT token is never written to the log
             log_headers = {**headers, "Authorization": f"{auth_prefix}<redacted>"}
-            self.log("OctopusAPI: Making GraphQL request to {} payload {} headers {}".format(url, payload, log_headers))
+            request_line = "OctopusAPI: Making GraphQL request to {} payload {} headers {}".format(url, payload, log_headers)
+            request_logged[0] = self.log_gate.log(self.log, ("request", request_context), query, [request_line], lambda when: "OctopusAPI: Making GraphQL request {} to {} (query as logged in full at {})".format(request_context, url, when))
+            request_logged[1] = request_line
             async with client.post(url, json=payload, headers=headers) as response:
                 # Check for HTTP-level 401/403 (transport-level auth failure) and retry once.
                 # This handles cases where the JWT has been revoked server-side and the server
                 # returns a bare 401/403 status rather than a GraphQL error body — which would
                 # otherwise loop forever without ever refreshing the token. The one exception is
                 # a 403 carrying a CDN/WAF block page, handled immediately below.
+                if response.status in [401, 403]:
+                    log_request_beside_failure()
                 if response.status in [401, 403] and _retry_count == 0:
                     # A CDN/WAF block is rate limiting, not an auth failure. Re-minting a token
                     # here would be rejected by the same block, leaving the component permanently
@@ -2378,7 +2393,20 @@ async def async_graphql_query(self, query, request_context, returns_data=True, i
 
                 # Process response (which reads the text)
                 response_body = await self.async_read_response_retry(response, url, ignore_errors=ignore_errors)
-                self.log("OctopusAPI: GraphQL response for {} (status {}): {}".format(request_context, response.status, response_body))
+                body_text = str(response_body)
+                # Logged in full with its request: a bad status, a body that is missing (unreadable JSON, a rate limit)
+                # or has no data, or any errors key - even a partial error returned with data when ignore_errors is set
+                failed = response.status != 200 or not isinstance(response_body, dict) or "errors" in response_body or response_body.get("data") is None
+                if failed:
+                    log_request_beside_failure()
+                self.log_gate.log(
+                    self.log,
+                    ("response", request_context),
+                    "{} {}".format(response.status, body_text),
+                    ["OctopusAPI: GraphQL response for {} (status {}): {}".format(request_context, response.status, body_text)],
+                    lambda when: "OctopusAPI: GraphQL response for {} (status {}) same as logged in full at {}, {} characters".format(request_context, response.status, when, len(body_text)),
+                    always_full=failed,
+                )
 
                 # Check for auth errors and retry once
                 if response_body and "errors" in response_body and _retry_count == 0:
@@ -2419,6 +2447,7 @@ async def async_graphql_query(self, query, request_context, returns_data=True, i
                     return None
         except TimeoutError:
             self.failures_total += 1
+            log_request_beside_failure()
             self.log(f"Warn: OctopusAPI: Failed to connect, timeout exceeded.")
             record_api_call("octopus", False, "connection_error")
 
@@ -3819,6 +3848,7 @@ def rate_add_io_slots(self, car_n, rates, octopus_slots):
         slots_added_set = set()
         plan_interval_minutes = self.plan_interval_minutes
         saved_slots = set()  # For logging purposes, track which slots we actually applied as low rate
+        slot_lines = []  # One log line per half hour of each slot, logged after the loop only when the list changes
         # Dynamic load has seen this car in its slot but not charging: none of its dispatches from now
         # on get the cheap rate (see dynamic_load_car_check()). Elapsed minutes keep theirs - they record
         # what the tariff charged, and today's cost is built from them. None when the car is not cancelled.
@@ -3913,12 +3943,16 @@ def rate_add_io_slots(self, car_n, rates, octopus_slots):
                                 rates[minute] = assumed_price
 
                         if minute % 30 == 0 and start_minutes > -24 * 60:
-                            self.log(
+                            slot_lines.append(
                                 "Octopus: Intelligent slot at {}-{}, assumed price {}, amount {}, kWh location {}, source {}, octopus_slot_low_rate {}".format(
                                     self.time_abs_str(start_minutes), self.time_abs_str(end_minutes), dp2(assumed_price), dp2(kwh), location, source, octopus_slot_low_rate
                                 )
                             )
 
+        # The slot list is re-derived every plan but rarely changes, so log it only when it does, and at least hourly
+        if slot_lines:
+            self.io_slot_log_gate.log(self.log, ("slots", car_n), "\n".join(slot_lines), slot_lines, lambda when: "Octopus: Intelligent slots for car {} same as logged in full at {}, {} half-hour slot lines".format(car_n, when, len(slot_lines)))
+
         # Log daily slot counts for debugging
         for day_offset in sorted(slots_per_day.keys()):
             if slots_per_day[day_offset] > 0:

apps/predbat/predbat.py

@@ -65,6 +65,7 @@
     print("Running in compiled mode; skipping local file checks and auto-update.")
 
 from const import (
+    REPEAT_FULL_LOG_SECONDS,
     TIME_FORMAT,
     PREDICT_STEP,
     RUN_EVERY,
@@ -77,6 +78,7 @@
 from config import APPS_SCHEMA, CONFIG_ITEMS
 import debug_history
 from utils import (
+    RepeatLogGate,
     minutes_since_yesterday,
     minutes_since_midnight,
     dp1,
@@ -491,6 +493,8 @@ def reset(self):
         self.car_charging_power = 0
         self.car_charging_power_configured = False
         self.io_adjusted = {}
+        # Repeated Intelligent slot lists are logged briefly between full logs (rate_add_io_slots)
+        self.io_slot_log_gate = RepeatLogGate(REPEAT_FULL_LOG_SECONDS)
         self.current_charge_limit = 0.0
         self.current_charge_limit_kwh = 0.0
         self.inverter_limit = 0.0

apps/predbat/tests/test_octopus_logging.py

@@ -7,7 +7,9 @@
 
 import asyncio
 from unittest.mock import AsyncMock, MagicMock
+from const import REPEAT_FULL_LOG_SECONDS
 from octopus import OctopusAPI
+from utils import RepeatLogGate
 
 
 def test_octopus_logging_wrapper(my_predbat):
@@ -82,4 +84,112 @@ async def test_octopus_logging(my_predbat):
     else:
         print("PASS: GraphQL response body logged")
 
+    # A repeat of the same query and response is logged briefly, without the query or body
+    log_lines.clear()
+    await api.async_graphql_query("query { account }", "test-logging", returns_data=True)
+    repeat = "\n".join(log_lines)
+    brief_request = "Making GraphQL request test-logging to " in repeat and "(query as logged in full at" in repeat
+    brief_response = "GraphQL response for test-logging (status 200) same as logged in full" in repeat
+    if "A-LOG-12345" in repeat or "query { account }" in repeat or not brief_request or not brief_response:
+        print("ERROR: a repeated query and response should be logged briefly, got {}".format(log_lines))
+        failed = True
+    else:
+        print("PASS: repeated query and response logged briefly")
+
+    # A changed response is logged in full again
+    log_lines.clear()
+    api.async_read_response_retry = AsyncMock(return_value={"data": {"account": {"accountNumber": "A-LOG-67890"}}})
+    await api.async_graphql_query("query { account }", "test-logging", returns_data=True)
+    if "A-LOG-67890" not in "\n".join(log_lines):
+        print("ERROR: a changed response should be logged in full, got {}".format(log_lines))
+        failed = True
+
+    # An unchanged response is logged in full again once REPEAT_FULL_LOG_SECONDS have passed
+    for entry, (logged_at, wall) in list(api.log_gate.logged.items()):
+        api.log_gate.logged[entry] = (logged_at - REPEAT_FULL_LOG_SECONDS - 1, wall)
+    log_lines.clear()
+    await api.async_graphql_query("query { account }", "test-logging", returns_data=True)
+    logged_again = "\n".join(log_lines)
+    if "A-LOG-67890" not in logged_again or "query { account }" not in logged_again:
+        print("ERROR: an unchanged response should be logged in full again after an hour, got {}".format(log_lines))
+        failed = True
+
+    # Two queries sharing one request context (one per device) are each logged briefly when they repeat
+    api.log_gate = RepeatLogGate(REPEAT_FULL_LOG_SECONDS)
+    api.async_read_response_retry = AsyncMock(return_value=response_body)
+    await api.async_graphql_query("query { device(id: 1) }", "test-device", returns_data=True)
+    await api.async_graphql_query("query { device(id: 2) }", "test-device", returns_data=True)
+    log_lines.clear()
+    await api.async_graphql_query("query { device(id: 1) }", "test-device", returns_data=True)
+    if not any("(query as logged in full at" in line for line in log_lines) or not any("same as logged in full" in line for line in log_lines):
+        print("ERROR: a repeated query should be logged briefly even when another query shares its context, got {}".format(log_lines))
+        failed = True
+
+    # An error response is always logged in full, even when it repeats
+    api.async_read_response_retry = AsyncMock(return_value={"errors": [{"message": "Something failed", "extensions": {"errorCode": "KT-CT-9999"}}]})
+    await api.async_graphql_query("query { broken }", "test-error", returns_data=True, ignore_errors=True)
+    log_lines.clear()
+    await api.async_graphql_query("query { broken }", "test-error", returns_data=True, ignore_errors=True)
+    if not any("Something failed" in line for line in log_lines):
+        print("ERROR: a repeated error response should still be logged in full, got {}".format(log_lines))
+        failed = True
+    if not any("query { broken }" in line for line in log_lines):
+        print("ERROR: the request behind a repeated error should be logged in full beside it, got {}".format(log_lines))
+        failed = True
+
+    # A status 200 with no usable body (unreadable JSON, a rate limit) or no data is a failure too: the request and
+    # the response are logged in full beside it, even when they repeat
+    for bad_body, context in [(None, "test-no-body"), ({"other": 1}, "test-no-data"), ({"data": None}, "test-null-data")]:
+        api.async_read_response_retry = AsyncMock(return_value=bad_body)
+        await api.async_graphql_query("query { empty }", context, returns_data=True)
+        log_lines.clear()
+        await api.async_graphql_query("query { empty }", context, returns_data=True)
+        if not any("query { empty }" in line for line in log_lines) or not any("GraphQL response for {} (status 200): {}".format(context, bad_body) in line for line in log_lines):
+            print("ERROR: a repeated response with no data should be logged in full with its request, got {}".format(log_lines))
+            failed = True
+
+    # The token must not leak through any of the later lines either
+    if secret_token in logged_again or secret_token in repeat:
+        print("ERROR: JWT token leaked on a repeat or re-log")
+        failed = True
+
+    failed |= test_repeat_log_gate()
+    return failed
+
+
+def test_repeat_log_gate():
+    """RepeatLogGate logs a text in full when new, briefly while it repeats, alternating texts each keep their own
+    entry, always_full forces a full log, and entries older than the interval are dropped."""
+    failed = False
+    gate = RepeatLogGate(3600)
+    if gate.last_full("key", "a") is not None:
+        print("ERROR: a new text should be logged in full")
+        failed = True
+    if gate.last_full("key", "a") is None:
+        print("ERROR: a repeated text should be logged briefly")
+        failed = True
+    if gate.last_full("key", "b") is not None or gate.last_full("key", "a") is None:
+        print("ERROR: a second text under the same key should not make the first one log in full again")
+        failed = True
+    if gate.last_full("other", "a") is not None:
+        print("ERROR: the same text under another key is new")
+        failed = True
+    if gate.last_full("key", "a", always_full=True) is not None:
+        print("ERROR: always_full should log in full")
+        failed = True
+    for entry, (logged_at, wall) in list(gate.logged.items()):
+        gate.logged[entry] = (logged_at - 3601, wall)
+    if gate.last_full("key", "a") is not None or len(gate.logged) != 1:
+        print("ERROR: entries older than the interval should be dropped, leaving {}".format(gate.logged))
+        failed = True
+    if gate.last_full("key", None) is not None or gate.last_full("key", None) is None:
+        print("ERROR: a None text should be gated like any other")
+        failed = True
+    lines = []
+    if not gate.log(lines.append, "lines", "x", ["one", "two"], lambda when: "brief " + when) or gate.log(lines.append, "lines", "x", ["one"], lambda when: "brief " + when):
+        print("ERROR: log() should report a full log, then a brief one")
+        failed = True
+    if lines[:2] != ["one", "two"] or not lines[2].startswith("brief "):
+        print("ERROR: log() should write every full line, then the brief line, got {}".format(lines))
+        failed = True
     return failed