GlcoinCore/blitz_api
1
0
Fork 0
mirror of https://github.com/fusion44/blitz_api.git synced 2026-08-13 11:52:45 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

feat: improve logging and error handling

Browse source
This commit is contained in:
fusion44 2023-04-21 22:58:29 +02:00
parent 72862f8d6a
commit 71ad84e704
No known key found for this signature in database
8 changed files with 144 additions and 71 deletions
Download patch file Download diff file Expand all files Collapse all files

10
app/api/utils.py
View file

@ -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 ""

62
app/bitcoind/service.py
View file

@ -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())

94
app/bitcoind/utils.py
View file

@ -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,
}

16
app/lightning/impl/cln_jrpc.py
View file

@ -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())

13
app/lightning/impl/lnd_grpc.py
View file

@ -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:

2
app/lightning/impl/specializations/cln_grpc_blitz.py
View file

@ -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"
)

8
app/lightning/impl/specializations/cln_jrpc_blitz.py
View file

@ -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()

10
app/system/impl/raspiblitz.py
View file

@ -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),
}
]