Ошибка ведения журнала Python: произошла непредвиденная ошибка: невозможно запланировать новые фьючерсы после завершения ⇐ Python

[Anonymous]

Я вижу ошибки при использовании этого пользовательского HttpHandler в своем регистраторе. Ошибка связана с этим обработчиком, поскольку при использовании других обработчиков ошибок нет.
Я вижу ошибку только в K8S. Проверил, что с ресурсами POD все в порядке.
В деструкторе обработчика __del__ нет исключений
Handler
import logging
import asyncio
import uuid
from weakref import WeakSet
import aiohttp
import threading
import traceback

class HttpHandler(logging.Handler):
def __init__(self, host: str):
super().__init__()
self.host = host

self.loop = asyncio.new_event_loop()
self.thread = threading.Thread(
target=self._run_loop, daemon=True
)
self._futures = WeakSet()

self.thread.start()
self.is_closing = False

def _run_loop(self):
asyncio.set_event_loop(self.loop)
self.loop.run_forever()

async def _send(self, payload):
try:
async with aiohttp.ClientSession() as session:
async with session.post(self.host, json=payload) as resp:
if resp.status != 200:
print(f"Fluent HTTP request failed with payload: {payload} and status: {resp.status}")
resp.raise_for_status()
except aiohttp.ClientError as e:
print(f"Fluent HTTP request failed with payload: {payload} and error: {e}")
except Exception as e:
print(f"Unexpected error occurred: {e}")

def emit(self, record):
if self.is_closing or self.loop.is_closed():
print(f"WARNING: Failed to send {record}. Event loop is closed")
return
payload = self.format(record)
if not self.loop.is_running():
print(f"WARNING: Failed to send {record}. Event loop is not running")
return
fut = asyncio.run_coroutine_threadsafe(
self._send(payload), self.loop
)
self._futures.add(fut)
fut.add_done_callback(self._futures.discard)

def close(self):
self.is_closing = True

# Wait for all pending futures to complete
for fut in list(self._futures):
try:
fut.result(timeout=10)
except Exception as e:
print(f"Force close pending task {e}")

# Stop the event loop if it's running
if self.loop.is_running():
self.loop.call_soon_threadsafe(self.loop.stop)
self.thread.join()

# Ensure the loop is not already closed
if not self.loop.is_closed():
try:
# Shutdown asynchronous generators
self.loop.run_until_complete(self.loop.shutdown_asyncgens())
except Exception as e:
print(f"Error shutting down asynchronous generators: {e}")

try:
# Shutdown default executor
self.loop.run_until_complete(self.loop.shutdown_default_executor())
except Exception as e:
print(f"Error shutting down default executor: {e}")

try:
# Close the loop
self.loop.close()
except Exception as e:
print(f"Error closing the loop: {e}")

super().close()

def __del__(self):
try:
self.close()
except Exception as e:
print(f"WARNING: exception in destructor: {e}. Trace: {traceback.format_exc()}")

Ошибки:
Unexpected error occurred: cannot schedule new futures after shutdownUnexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown
Unexpected error occurred: cannot schedule new futures after shutdown


Подробнее здесь: https://stackoverflow.com/questions/798 ... utures-aft