From 2b0dbb4e8e9a8a260aebb9347ca32ea5000051c9 Mon Sep 17 00:00:00 2001 From: DevTekVE Date: Tue, 11 Jun 2024 07:19:54 +0000 Subject: [PATCH] [sunnylink] Swaglogs to cloudwatch Enable log_handler in sunnylinkd.py --- selfdrive/athena/athenad.py | 87 +++++++++++++++++++++++++++------- selfdrive/athena/sunnylinkd.py | 31 ++++++++++-- system/loggerd/xattr_cache.py | 15 ++++-- 3 files changed, 109 insertions(+), 24 deletions(-) diff --git a/selfdrive/athena/athenad.py b/selfdrive/athena/athenad.py index 443173e36d..e8ef682026 100755 --- a/selfdrive/athena/athenad.py +++ b/selfdrive/athena/athenad.py @@ -16,6 +16,7 @@ import sys import tempfile import threading import time +import gzip from dataclasses import asdict, dataclass, replace from datetime import datetime from functools import partial @@ -551,7 +552,7 @@ def takeSnapshot() -> str | dict[str, str] | None: raise Exception("not available while camerad is started") -def get_logs_to_send_sorted() -> list[str]: +def get_logs_to_send_sorted(log_attr_name=LOG_ATTR_NAME) -> list[str]: # TODO: scan once then use inotify to detect file creation/deletion curr_time = int(time.time()) logs = [] @@ -559,7 +560,7 @@ def get_logs_to_send_sorted() -> list[str]: log_path = os.path.join(Paths.swaglog_root(), log_entry) time_sent = 0 try: - value = getxattr(log_path, LOG_ATTR_NAME) + value = getxattr(log_path, log_attr_name) if value is not None: time_sent = int.from_bytes(value, sys.byteorder) except (ValueError, TypeError): @@ -571,8 +572,68 @@ def get_logs_to_send_sorted() -> list[str]: return sorted(logs)[:-1] -def log_handler(end_event: threading.Event) -> None: +def add_log_to_queue(log_path, log_id, is_sunnylink = False): + MAX_SIZE_KB = 32 + MAX_SIZE_BYTES = MAX_SIZE_KB * 1024 + + with open(log_path, 'r') as f: + data = f.read() + + # Check if the file is empty + if not data: + cloudlog.warning(f"Log file {log_path} is empty.") + return + + # Initialize variables for encoding + payload = data + is_compressed = False + + # Log the current size of the file + current_size = len(json.dumps(payload).encode("utf-8")) + len(log_id.encode("utf-8")) + 100 # Add 100 bytes to account for encoding overhead + cloudlog.debug(f"Current size of log file {log_path}: {current_size} bytes") + + if is_sunnylink and current_size > MAX_SIZE_BYTES: + # Compress and encode the data if it exceeds the maximum size + compressed_data = gzip.compress(data.encode()) + payload = base64.b64encode(compressed_data).decode() + is_compressed = True + + # Log the size after compression and encoding + compressed_size = len(compressed_data) + encoded_size = len(payload) + cloudlog.debug(f"Size of log file {log_path} " + f"after compression: {compressed_size} bytes, " + f"after encoding: {encoded_size} bytes") + + jsonrpc = { + "method": "forwardLogs", + "params": { + "logs": payload + }, + "jsonrpc": "2.0", + "id": log_id + } + + if is_sunnylink and is_compressed: + jsonrpc["params"]["compressed"] = is_compressed + + jsonrpc_str = json.dumps(jsonrpc) + size_in_bytes = len(jsonrpc_str.encode('utf-8')) + + if is_sunnylink and size_in_bytes <= MAX_SIZE_BYTES: + cloudlog.debug(f"Target is sunnylink and log file {log_path} is small enough to send in one request ({size_in_bytes} bytes).") + low_priority_send_queue.put_nowait(jsonrpc_str) + elif is_sunnylink: + cloudlog.warning(f"Target is sunnylink and log file {log_path} is too large to send in one request.") + else: + cloudlog.debug(f"Target is not sunnylink, proceeding to send log file {log_path} in one request ({size_in_bytes} bytes).") + low_priority_send_queue.put_nowait(jsonrpc_str) + + +def log_handler(end_event: threading.Event, log_attr_name=LOG_ATTR_NAME) -> None: + is_sunnylink = log_attr_name != LOG_ATTR_NAME if PC: + cloudlog.debug("athena.log_handler: Not supported on PC") return log_files = [] @@ -581,7 +642,7 @@ def log_handler(end_event: threading.Event) -> None: try: curr_scan = time.monotonic() if curr_scan - last_scan > 10: - log_files = get_logs_to_send_sorted() + log_files = get_logs_to_send_sorted(log_attr_name) last_scan = curr_scan # send one log @@ -592,18 +653,10 @@ def log_handler(end_event: threading.Event) -> None: try: curr_time = int(time.time()) log_path = os.path.join(Paths.swaglog_root(), log_entry) - setxattr(log_path, LOG_ATTR_NAME, int.to_bytes(curr_time, 4, sys.byteorder)) - with open(log_path) as f: - jsonrpc = { - "method": "forwardLogs", - "params": { - "logs": f.read() - }, - "jsonrpc": "2.0", - "id": log_entry - } - low_priority_send_queue.put_nowait(json.dumps(jsonrpc)) - curr_log = log_entry + setxattr(log_path, log_attr_name, int.to_bytes(curr_time, 4, sys.byteorder)) + + add_log_to_queue(log_path, log_entry, is_sunnylink) + curr_log = log_entry except OSError: pass # file could be deleted by log rotation @@ -620,7 +673,7 @@ def log_handler(end_event: threading.Event) -> None: if log_entry and log_success: log_path = os.path.join(Paths.swaglog_root(), log_entry) try: - setxattr(log_path, LOG_ATTR_NAME, LOG_ATTR_VALUE_MAX_UNIX_TIME) + setxattr(log_path, log_attr_name, LOG_ATTR_VALUE_MAX_UNIX_TIME) except OSError: pass # file could be deleted by log rotation if curr_log == log_entry: diff --git a/selfdrive/athena/sunnylinkd.py b/selfdrive/athena/sunnylinkd.py index b0c6e8d4dc..0b784d0371 100755 --- a/selfdrive/athena/sunnylinkd.py +++ b/selfdrive/athena/sunnylinkd.py @@ -11,7 +11,7 @@ import threading import time from openpilot.selfdrive.athena.athenad import ws_send, jsonrpc_handler, \ - recv_queue, RECONNECT_TIMEOUT_S, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage + recv_queue, RECONNECT_TIMEOUT_S, UploadQueueCache, upload_queue, cur_upload_items, backoff, ws_manage, log_handler from jsonrpc import dispatcher from websocket import (ABNF, WebSocket, WebSocketException, WebSocketTimeoutException, create_connection) @@ -20,16 +20,20 @@ from openpilot.common.api import SunnylinkApi from openpilot.common.params import Params from openpilot.common.realtime import set_core_affinity from openpilot.common.swaglog import cloudlog +import cereal.messaging as messaging SUNNYLINK_ATHENA_HOST = os.getenv('SUNNYLINK_ATHENA_HOST', 'wss://ws.stg.api.sunnypilot.ai') HANDLER_THREADS = int(os.getenv('HANDLER_THREADS', "4")) LOCAL_PORT_WHITELIST = {8022} +SUNNYLINK_LOG_ATTR_NAME = "user.sunny.upload" params = Params() sunnylink_api = SunnylinkApi(params.get("SunnylinkDongleId", encoding='utf-8')) def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None: cloudlog.info("sunnylinkd.handle_long_poll started") + sm = messaging.SubMaster(['deviceState']) end_event = threading.Event() + comma_prime_cellular_end_event = threading.Event() threads = [ threading.Thread(target=ws_manage, args=(ws, end_event), name='ws_manage'), @@ -38,7 +42,7 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None: threading.Thread(target=ws_ping, args=(ws, end_event), name='ws_ping'), threading.Thread(target=ws_queue, args=(end_event,), name='ws_queue'), # threading.Thread(target=upload_handler, args=(end_event,), name='upload_handler'), - # threading.Thread(target=log_handler, args=(end_event,), name='log_handler'), + threading.Thread(target=sunny_log_handler, args=(end_event, comma_prime_cellular_end_event), name='log_handler'), # threading.Thread(target=stat_handler, args=(end_event,), name='stat_handler'), ] + [ threading.Thread(target=jsonrpc_handler, args=(end_event,), name=f'worker_{x}') @@ -49,14 +53,28 @@ def handle_long_poll(ws: WebSocket, exit_event: threading.Event | None) -> None: thread.start() try: while not end_event.wait(0.1): + sm.update(0) if exit_event is not None and exit_event.is_set(): end_event.set() + comma_prime_cellular_end_event.set() + + prime_type = params.get("PrimeType", encoding='utf-8') + metered = sm['deviceState'].networkMetered + + if int(prime_type) > 2 and metered: + cloudlog.debug(f"sunnylinkd.handle_long_poll: PrimeType({prime_type}) > 2 and networkMetered({metered})") + comma_prime_cellular_end_event.set() + elif comma_prime_cellular_end_event.is_set(): + cloudlog.debug(f"sunnylinkd.handle_long_poll: comma_prime_cellular_end_event is set and not PrimeType({prime_type}) > 2 or not networkMetered({metered})") + comma_prime_cellular_end_event.clear() + except (KeyboardInterrupt, SystemExit): end_event.set() + comma_prime_cellular_end_event.set() raise finally: for thread in threads: - cloudlog.info(f"sunnylinkd athena.joining {thread.name}") + cloudlog.debug(f"sunnylinkd athena.joining {thread.name}") thread.join() @@ -114,6 +132,13 @@ def ws_queue(end_event: threading.Event) -> None: cloudlog.debug("Resume requested or end_event is set, exiting ws_queue thread") +def sunny_log_handler(end_event: threading.Event, comma_prime_cellular_end_event: threading.Event) -> None: + while not end_event.wait(0.1): + if not comma_prime_cellular_end_event.is_set(): + log_handler(comma_prime_cellular_end_event, SUNNYLINK_LOG_ATTR_NAME) + comma_prime_cellular_end_event.set() + + @dispatcher.add_method def getParamsAllKeys() -> list[str]: keys: list[str] = [k.decode('utf-8') for k in Params().all_keys()] diff --git a/system/loggerd/xattr_cache.py b/system/loggerd/xattr_cache.py index d3220118ac..88aa4f7916 100644 --- a/system/loggerd/xattr_cache.py +++ b/system/loggerd/xattr_cache.py @@ -1,5 +1,12 @@ -import os import errno +import platform + +if platform.system() == 'Darwin': # macOS + from xattr import getxattr as _getxattr + from xattr import setxattr as _setxattr +else: + from os import getxattr as _getxattr + from os import setxattr as _setxattr _cached_attributes: dict[tuple, bytes | None] = {} @@ -7,10 +14,10 @@ def getxattr(path: str, attr_name: str) -> bytes | None: key = (path, attr_name) if key not in _cached_attributes: try: - response = os.getxattr(path, attr_name) + response = _getxattr(path, attr_name) except OSError as e: # ENODATA means attribute hasn't been set - if e.errno == errno.ENODATA: + if e.errno == errno.ENODATA or e.errno == errno.ENOATTR: response = None else: raise @@ -19,4 +26,4 @@ def getxattr(path: str, attr_name: str) -> bytes | None: def setxattr(path: str, attr_name: str, attr_value: bytes) -> None: _cached_attributes.pop((path, attr_name), None) - return os.setxattr(path, attr_name, attr_value) + return _setxattr(path, attr_name, attr_value) \ No newline at end of file