From 71ad84e70449daf2fa7cdcdb8a7ca4b4bac5acad Mon Sep 17 00:00:00 2001 From: fusion44 Date: Fri, 21 Apr 2023 22:58:29 +0200 Subject: [PATCH] feat: improve logging and error handling --- app/api/utils.py | 10 +- app/bitcoind/service.py | 62 ++++++++---- app/bitcoind/utils.py | 94 ++++++++++++------- app/lightning/impl/cln_jrpc.py | 16 +++- app/lightning/impl/lnd_grpc.py | 13 ++- .../impl/specializations/cln_grpc_blitz.py | 2 +- .../impl/specializations/cln_jrpc_blitz.py | 8 +- app/system/impl/raspiblitz.py | 10 +- 8 files changed, 144 insertions(+), 71 deletions(-) diff --git a/app/api/utils.py b/app/api/utils.py index af99f75..f01bd8a 100644 --- a/app/api/utils.py +++ b/app/api/utils.py @@ -1,7 +1,6 @@ import array import asyncio import json -import logging import os import random import re @@ -11,6 +10,7 @@ from typing import Dict, Optional from fastapi.encoders import jsonable_encoder from fastapi_plugins import redis_plugin +from loguru import logger from app.api.sse_manager import SSEManager from app.external.sse_starlette import ServerSentEvent @@ -60,9 +60,9 @@ async def redis_get(key: str) -> str: if not v: logstr = f"Key '{key}' not found in Redis DB." if "tor_web_addr" in key: - logging.info(logstr) + logger.info(logstr) else: - logging.warning(logstr) + logger.warning(logstr) return "" try: @@ -215,7 +215,7 @@ async def call_script(scriptPath) -> str: if stdout: return stdout.decode() if stderr: - logging.error(stderr.decode()) + logger.error(stderr.decode()) return "" @@ -254,7 +254,7 @@ async def call_sudo_script(scriptPath) -> str: if stdout: return stdout.decode() if stderr: - logging.error(stderr.decode()) + logger.error(stderr.decode()) return "" diff --git a/app/bitcoind/service.py b/app/bitcoind/service.py index bf74704..7510019 100644 --- a/app/bitcoind/service.py +++ b/app/bitcoind/service.py @@ -1,13 +1,13 @@ import asyncio import binascii import json -import logging import zmq import zmq.asyncio from aiohttp import client_exceptions from fastapi import Request from fastapi.exceptions import HTTPException +from loguru import logger from starlette import status from app.api.utils import SSE, broadcast_sse_msg @@ -24,51 +24,59 @@ from app.bitcoind.utils import bitcoin_config, bitcoin_rpc_async _initialized = False +@logger.catch(exclude=(HTTPException,)) async def initialize_bitcoin_repo() -> bool: global _initialized if _initialized: return True - logging.info("Initializing bitcoin repository") + logger.info("Initializing bitcoin repository") # Wait until the bitcoin node is ready to accept RPC calls while not _initialized: try: await get_blockchain_info() _initialized = True - logging.info("Bitcoin repository initialized") + logger.success("Bitcoin repository initialized") return True except client_exceptions.ClientConnectorError: - logging.debug("Unable to connect to Bitcoin Core, waiting...") - await asyncio.sleep(2) + logger.debug("Unable to connect to Bitcoin Core, waiting 5 seconds...") + await asyncio.sleep(5) except HTTPException as e: + if e.status_code == status.HTTP_425_TOO_EARLY: + logger.info("Bitcoin Core initializing, waiting 10 seconds...") + await asyncio.sleep(10) + continue + if e.status_code == status.HTTP_500_INTERNAL_SERVER_ERROR: - logging.error(e.detail) - else: - logging.debug( - f"Connected to Bitcoin Core but it seems to be initializing, waiting... \n{e.detail}" - ) + logger.error(e.detail) + + logger.debug( + f"Connected to Bitcoin Core but it seems to be initializing, waiting 2 seconds... \n{e.detail}" + ) await asyncio.sleep(2) +@logger.catch(exclude=(HTTPException,)) async def get_blockchain_info() -> BlockchainInfo: result = await bitcoin_rpc_async("getblockchaininfo") + if result["error"] != None: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"] - ) + raise HTTPException(result["status"], detail=result["error"]) + return BlockchainInfo.from_rpc(result["result"]) +@logger.catch(exclude=(HTTPException,)) async def estimate_fee( target_conf: int = 6, mode: FeeEstimationMode = FeeEstimationMode.CONSERVATIVE, ) -> int: result = await bitcoin_rpc_async("estimatesmartfee", [target_conf, mode]) + if result["error"] != None: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"] - ) + raise HTTPException(result["status"], detail=result["error"]) + if "errors" in result["result"]: errors = "Bitcoin Core returned error(s):\n" for e in result["result"]["errors"]: @@ -80,18 +88,21 @@ async def estimate_fee( # returned in BTC by Bitcoin Core => convert to msat rate_btc = result["result"]["feerate"] + return rate_btc * 100000000 +@logger.catch(exclude=(HTTPException,)) async def get_network_info() -> NetworkInfo: result = await bitcoin_rpc_async("getnetworkinfo") + if result["error"] != None: - raise HTTPException( - status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"] - ) + raise HTTPException(result["status"], detail=result["error"]) + return NetworkInfo.from_rpc(result["result"]) +@logger.catch(exclude=(HTTPException,)) async def get_raw_transaction(txid: str) -> RawTransaction: result = await bitcoin_rpc_async("getrawtransaction", [txid, 1]) @@ -107,12 +118,15 @@ async def get_raw_transaction(txid: str) -> RawTransaction: raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail=result["error"]) +@logger.catch(exclude=(HTTPException,)) async def get_btc_info() -> BtcInfo: binfo = await get_blockchain_info() ninfo = await get_network_info() + return BtcInfo.from_rpc(binfo, ninfo) +@logger.catch(exclude=(HTTPException,)) async def handle_block_sub(request: Request, verbosity: int = 1) -> str: ctx = zmq.asyncio.Context() zmq_socket = ctx.socket(zmq.SUB) @@ -128,9 +142,11 @@ async def handle_block_sub(request: Request, verbosity: int = 1) -> str: _, body, _ = await zmq_socket.recv_multipart() hash = binascii.hexlify(body).decode("utf-8") r = await bitcoin_rpc_async("getblock", [hash, verbosity]) + yield json.dumps(r["result"]) +@logger.catch(exclude=(HTTPException,)) async def handle_block_sub_redis(verbosity: int = 1) -> str: ctx = zmq.asyncio.Context() zmq_socket = ctx.socket(zmq.SUB) @@ -155,19 +171,24 @@ async def handle_block_sub_redis(verbosity: int = 1) -> str: await broadcast_sse_msg(SSE.BTC_NEW_BLOC, r["result"]) +@logger.catch(exclude=(HTTPException,)) async def register_bitcoin_zmq_sub(): loop = asyncio.get_event_loop() loop.create_task(handle_block_sub_redis()) +@logger.catch(exclude=(HTTPException,)) async def _handle_gather_bitcoin_status(): last_info = {} while True: try: info = await get_btc_info() + if info == None: + continue + info.verification_progress = round(info.verification_progress, 2) except HTTPException as e: - logging.error(e.detail) + logger.error(e.detail) await asyncio.sleep(2) continue @@ -179,6 +200,7 @@ async def _handle_gather_bitcoin_status(): await asyncio.sleep(2) +@logger.catch(exclude=(HTTPException,)) async def register_bitcoin_status_gatherer(): loop = asyncio.get_event_loop() loop.create_task(_handle_gather_bitcoin_status()) diff --git a/app/bitcoind/utils.py b/app/bitcoind/utils.py index 9d00921..2a9024d 100644 --- a/app/bitcoind/utils.py +++ b/app/bitcoind/utils.py @@ -5,6 +5,7 @@ from types import coroutine import aiohttp import requests from decouple import config +from loguru import logger from starlette import status from app.bitcoind.models import BlockRpcFunc @@ -76,38 +77,65 @@ async def bitcoin_rpc_async(method: str, params: list = []) -> coroutine: + json.dumps(params) + "}" ) + try: + async with aiohttp.ClientSession(auth=auth, headers=headers) as session: + async with session.post(bitcoin_config.rpc_url, data=data) as resp: + return await _process_response(resp) - # TODO: Refactor this to use Exceptions - async with aiohttp.ClientSession(auth=auth, headers=headers) as session: - async with session.post(bitcoin_config.rpc_url, data=data) as resp: - if resp.status == status.HTTP_200_OK: - return await resp.json() - elif resp.status == status.HTTP_401_UNAUTHORIZED: - return { - "error": "Access denied to Bitcoin Core RPC. Check if username and password is correct", - "status": status.HTTP_403_FORBIDDEN, - } - elif resp.status == status.HTTP_403_FORBIDDEN: - return { - "error": "Access denied to Bitcoin Core RPC. If this is a remote node, check if 'network.rpcallowip=0.0.0.0/0' is set.", - "status": status.HTTP_403_FORBIDDEN, - } - else: - e = await resp.json() - m = e["error"]["message"] - if e["error"]: - if "No such mempool or blockchain transaction." in m: - return { - "error": "No such mempool or blockchain transaction.", - "status": status.HTTP_404_NOT_FOUND, - } - if "parameter 1 must be of length 64" in m: - return { - "error": m, - "status": status.HTTP_400_BAD_REQUEST, - } + except aiohttp.client_exceptions.ClientConnectionError as e: + return { + "error": f"Aiohttp client connection error: {str(e)}", + "status": status.HTTP_503_SERVICE_UNAVAILABLE, + } - return { - "error": f"Unknown answer from Bitcoin Core. Reason: {resp.reason}", - "status": resp.status, - } + except aiohttp.client_exceptions.ClientError as e: + return { + "error": f"Aiohttp client error: {str(e)}", + "status": status.HTTP_503_SERVICE_UNAVAILABLE, + } + + +async def _process_response(resp: aiohttp.ClientResponse): + if resp.status == status.HTTP_200_OK: + return await resp.json() + + if resp.status == status.HTTP_401_UNAUTHORIZED: + return { + "error": "Access denied to Bitcoin Core RPC. Check if username and password is correct", + "status": status.HTTP_403_FORBIDDEN, + } + + if resp.status == status.HTTP_403_FORBIDDEN: + return { + "error": "Access denied to Bitcoin Core RPC. If this is a remote node, check if 'network.rpcallowip=0.0.0.0/0' is set.", + "status": status.HTTP_403_FORBIDDEN, + } + + e = await resp.json() + m = e["error"]["message"] + + if e["error"]: + if ( + "Loading block index" in m + or "Verifying blocks" in m + or "Starting network threads" in m + ): + return { + "error": "Initializing Bitcoin Core (loading, verifying blocks or starting network threads etc)", + "status": status.HTTP_425_TOO_EARLY, + } + if "No such mempool or blockchain transaction." in m: + return { + "error": "No such mempool or blockchain transaction.", + "status": status.HTTP_404_NOT_FOUND, + } + if "parameter 1 must be of length 64" in m: + return { + "error": m, + "status": status.HTTP_400_BAD_REQUEST, + } + + return { + "error": f"Unknown answer from Bitcoin Core. Reason: {resp.reason}", + "status": resp.status, + } diff --git a/app/lightning/impl/cln_jrpc.py b/app/lightning/impl/cln_jrpc.py index 01b705b..7b89768 100644 --- a/app/lightning/impl/cln_jrpc.py +++ b/app/lightning/impl/cln_jrpc.py @@ -98,10 +98,18 @@ class LnNodeCLNjRPC(LightningNodeBase): ) self._loop = asyncio.get_running_loop() - self._reader, self._writer = await asyncio.open_unix_connection( - path=self._socket_path, - limit=_SOCKET_BUFFER_SIZE_LIMIT, - ) + + while True: + try: + self._reader, self._writer = await asyncio.open_unix_connection( + path=self._socket_path, + limit=_SOCKET_BUFFER_SIZE_LIMIT, + ) + + break + except ConnectionRefusedError: + logger.info("CLN ConnectionRefusedError. Retrying in 10 seconds.") + await asyncio.sleep(10) asyncio.create_task(self._read_loop()) diff --git a/app/lightning/impl/lnd_grpc.py b/app/lightning/impl/lnd_grpc.py index ceac9f1..b4eae87 100644 --- a/app/lightning/impl/lnd_grpc.py +++ b/app/lightning/impl/lnd_grpc.py @@ -113,9 +113,11 @@ This will show more debug information. # Reason is that gRPC seems to only try and connect every 5 seconds to # the node if it is not running. To avoid the delay we create a new # channel each iteration. - temp_channel = None temp_stub = None + + # We want to log the wallet locked error only once to avoid spamming the log + wallet_locked_sent = False while True: try: if temp_channel is None: @@ -162,6 +164,12 @@ This will show more debug information. await temp_channel.close() temp_channel = None elif "wallet locked, unlock it to enable full RPC access" in details: + if not wallet_locked_sent: + logger.info( + "Wallet is locked. Unlock by calling /lightning/unlock-wallet" + ) + + wallet_locked_sent = True await self._init_queue.put( InitLnRepoUpdate( state=LnInitState.LOCKED, @@ -251,6 +259,7 @@ This will show more debug information. elif ( res.state == LnInitState.OFFLINE or res.state == LnInitState.LOCKED + or res.state == LnInitState.BOOTSTRAPPING or res.state == LnInitState.BOOTSTRAPPING_AFTER_UNLOCK ): pass # do nothing here @@ -259,7 +268,7 @@ This will show more debug information. yield res - logger.info("Initialization complete.") + logger.success("Initialization complete.") @logger.catch(exclude=(HTTPException,)) async def get_wallet_balance(self) -> WalletBalance: diff --git a/app/lightning/impl/specializations/cln_grpc_blitz.py b/app/lightning/impl/specializations/cln_grpc_blitz.py index c3cb13d..5301dfe 100644 --- a/app/lightning/impl/specializations/cln_grpc_blitz.py +++ b/app/lightning/impl/specializations/cln_grpc_blitz.py @@ -46,7 +46,7 @@ class LnNodeCLNgRPCBlitz(LnNodeCLNgRPC): key = f"ln_cl_{self._NETWORK}_locked" res = await redis_get(key) if res == "0": - logger.debug( + logger.success( f"Redis key {key} indicates that RaspiBlitz has been unlocked" ) diff --git a/app/lightning/impl/specializations/cln_jrpc_blitz.py b/app/lightning/impl/specializations/cln_jrpc_blitz.py index 9f44015..4b3b056 100644 --- a/app/lightning/impl/specializations/cln_jrpc_blitz.py +++ b/app/lightning/impl/specializations/cln_jrpc_blitz.py @@ -45,8 +45,10 @@ class LnNodeCLNjRPCBlitz(LnNodeCLNjRPC): while not self._unlocked: key = f"ln_cl_{self._NETWORK}_locked" res = await redis_get(key) - if res == "0": - logger.debug( + if res == "": + logger.info(f"Redis key {key} not yet found, waiting...") + elif res == "0": + logger.success( f"Redis key {key} indicates that RaspiBlitz has been unlocked" ) @@ -76,7 +78,7 @@ class LnNodeCLNjRPCBlitz(LnNodeCLNjRPC): if u.state == LnInitState.DONE: break - logger.info("Initialization complete.") + logger.success("Initialization complete.") async def get_wallet_balance(self) -> WalletBalance: self._check_if_locked() diff --git a/app/system/impl/raspiblitz.py b/app/system/impl/raspiblitz.py index 8a5900e..9749757 100644 --- a/app/system/impl/raspiblitz.py +++ b/app/system/impl/raspiblitz.py @@ -338,8 +338,10 @@ class RaspiBlitzSystem(SystemBase): setup_phase = await redis_get("setupPhase") info["disks"] = [] if setup_phase == "done": - total = int(await redis_get("hdd_capacity_bytes")) - free = int(await redis_get("hdd_free_bytes")) + res = await redis_get("hdd_capacity_bytes") + total = int(res if len(res) > 0 else 0) + res = await redis_get("hdd_free_bytes") + free = int(res if len(res) > 0 else 0) info["disks"] = [ { "device": "/", @@ -348,7 +350,9 @@ class RaspiBlitzSystem(SystemBase): "partition_total_bytes": total, "partition_used_bytes": total - free, "partition_free_bytes": free, - "partition_percent": round((100 / total) * free, 2), + "partition_percent": 0 + if total == 0 + else round((100 / total) * free, 2), } ]