khushalps/api_internal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20241226) Started working on the server management system.

Browse Source
This commit is contained in:
khushal
2024-12-25 19:48:25 +05:30
parent 7607a427a6
commit 5051f73725
5 changed files with 446 additions and 630 deletions
Download Patch File Download Diff File Expand all files Collapse all files
controllers/servers/server.py
+258 -233
View File
@@ -34,17 +34,16 @@ import sys
sys.path.append(".")
sys.path.append("..")
# For Quart:
from quart import current_app
# My async utils:
from utils_v2.string import json
from utils_v2.date_time import date_time
from utils_v2.database.async_mysql_v2 import AsyncMySQL
from utils_v2.database.async_mongo_v2 import AsyncMongo, AsyncMongoStorage
# Models:
from models.servers.core import CoreServerInfoModel
from models.servers.core import CoreServerInfoModel, CoreServerCheckLogModel
# To make API calls:
import httpx
# To work with MongoDB:
from bson import ObjectId
@@ -61,10 +60,14 @@ import base64
# To work with date and time:
import datetime
import time
# For asynchronous activities:
import asyncio
# For debugging:
from icecream import IceCreamDebugger
# *****************************************************************************************************************
# ***** ****
@@ -111,6 +114,58 @@ class CoreServerController:
# For MongoDB:
SERVERS_COLLECTION = "_servers"
SERVER_CHECK_LOGS_COLLECTION = "_serverCheckLogs"
# ┏┓
# ┃ ┏┓┏┓┏╋┏┓┓┏┏╋┏┓┏┓
# ┗┛┗┛┛┗┛┗┛ ┗┻┗┗┗┛┛
def __init__(
self,
batch_size: int = 50,
batch_timeout: int | float = 60,
http_client: httpx.AsyncClient = None,
debug = True,
debug_prefix = "Srvr. (C) | ",
debug_only_errors = True
):
"""
To initialize the instance of this Server controller.
:param batch_size: When monitoring the servers, how many records will one worker pick at once to poll.
:param batch_timeout: When a particular server has been picked in a batch, for how long must another worker not
touch it before making his own attempt.
:param http_client: An instance of 'httpx' library's AsyncClient. If not given, one will be instantiated
internally. It is recommended that, for multi-bot use cases, you provide a common HTTP client from outside.
:param debug: Whether, or not, you would like to show debugging messages on the terminal.
:param debug_prefix: The prefix string to identify the debugging messages.
:param debug_only_errors: Whether you would like to show all debugging messages or just error messages.
"""
# Prepare the debugging utility:
self._debug_prefix = debug_prefix
self._printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
if not debug: self._printer.disable()
self._debug_only_errors = debug_only_errors
# Accept the configuration:
self._batch_size = batch_size
self._batch_timeout = batch_timeout
# Accept/create an HTTP client to work with:
if http_client: self.__http_client = http_client
else: self.__http_client = httpx.AsyncClient(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 5.0, # ..... Time to wait for sending data.
read = 5.0 # ....... Time to wait for receiving data.
)
)
# ┏┓┳┓┳┳┳┓ ┏┓
# ┃ ┣┫┃┃┃┃ ━━ ┃ ┏┓┏┓┏┓╋┏┓
@@ -118,268 +173,238 @@ class CoreServerController:
async def register(
self,
mongo_conn: AsyncMongo,
mongo_data_conn: AsyncMongo,
server: CoreServerInfoModel
) -> ObjectId:
) -> CoreServerInfoModel | None:
"""
Register one server in the database.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_data_conn: The instance of the database connector to use for the operation.
:param server: The message to save into the database.
:return: The object id of the inserted document.
"""
# Simply insert the document:
return await mongo_conn.insert_one(
collection = self.MESSAGES_COLLECTION,
document = message,
raise_exception = True
# Note down the current time:
event_ts = date_time.get_current_utc_date_time(as_string = False)
# Register/update the server's info:
_server = await mongo_data_conn.find_one_and_update(
collection = self.SERVERS_COLLECTION,
filter = {
"hostname": server.hostname,
"project": server.project,
"service": server.service,
"healthCheckUrl": server.healthCheckUrl
},
update = {
"$setOnInsert": {
"hostname": server.hostname,
"project": server.project,
"service": server.service,
"healthCheckUrl": server.healthCheckUrl,
"firstRegTs": event_ts
},
"$set": {
"os": server.os,
"cpu": server.cpu,
"pid": server.pid,
"ppid": server.ppid,
"ipAddr": server.ipAddr,
"portNo": server.portNo,
"description": server.description,
"healthCheckInterval": server.healthCheckInterval,
"healthAlertUrl": server.healthAlertUrl,
"online": True,
"batchId": server.batchId,
"batchTs": server.batchTs,
"lastRegTs": server.lastRegTs,
"lastCheckTs": server.lastCheckTs,
"checkAfterTs": server.checkAfterTs,
}
},
upsert = True,
return_updated = True
)
async def bulk_write(
self,
mongo_conn: AsyncMongo,
mongo_operations: list
) -> int:
"""
Needed in cases like forcing re-sync of mails where you need to perform actions like bulk replacements of
existing documents. Not recommended to use. Please use very carefully to ensure document integrity.
:param mongo_conn: The instance of the database connector to use for the operation.
:param mongo_operations: The list operations that are supported by MongoDB's Bulk Write system.
:return: The no. of documents affected.
"""
return await mongo_conn.bulk_write(
collection = self.MESSAGES_COLLECTION,
requests = mongo_operations,
raise_exception = True
)
# Done here:
return CoreServerInfoModel(**_server) if _server else None
# ┏┓┳┓┳┳┳┓ ┳┓ •
# ┃ ┣┫┃┃┃┃ ━━ ┣┫┏┓╋┏┓┓┏┓┓┏┏┓
# ┗┛┛┗┗┛┻┛ ┛┗┗ ┗┛ ┗┗ ┗┛┗
async def count_messages(
async def get_batch(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
additional_filter: dict = None
) -> int:
mongo_data_conn: AsyncMongo,
) -> List[CoreServerInfoModel] | None:
"""
Just counts the no. of messages that match a given set of conditions.
:param mongo_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param additional_filter: Any addition filters to use.
:return: The no. of messages that match the given conditions.
"""
# Get a batch id and note down the time:
batch_id = ObjectId()
batch_ts = date_time.get_current_utc_date_time(as_string = False)
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
# Find servers that need to be checked:
servers = await mongo_data_conn.find_many(
collection = self.SERVERS_COLLECTION,
filter = {
"checkAfterTs": {"$lte": batch_ts},
"$or": [
{
"batchId": {"$eq": None}
},
{
"batchTs": {"$lt": batch_ts - datetime.timedelta(seconds = self._batch_timeout)}
}
]
},
sort = {"checkAfterTs": 1},
limit = self._batch_size
)
# Get the count of the documents that match the criteria:
count = await mongo_conn.count(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
raise_exception = True
# If we didn't find any servers:
if not servers: return servers
# Model the servers:
servers = [CoreServerInfoModel(**s) for s in servers]
# Mark these servers as picked for checking:
updated_count = await mongo_data_conn.update_many(
collection = self.SERVERS_COLLECTION,
filter = {"_id": {"$in": [s.serverId for s in servers]}},
update = {
"$set": {
"batchId": batch_id,
"batchTs": batch_ts
}
},
upsert = False
)
# Done here:
return count
async def get_previews(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
) -> List[CoreMessageModel] | None:
"""
Fetches many messages in one call, but leaves out the full payloads.
:param mongo_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param limit: The max. no. of messages to retrieve in this call.
:param skip: The no. of initial messages to skip. Useful for pagination.
:param additional_filter: Any addition filters to use.
:return: The list of messages (as the message model). This list can be empty.
"""
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
# We fetch the messages that are identified by a specific token id,
# with the specified fetching limits, while enforcing the sorting condition:
records = await mongo_conn.find_many(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
limit = limit,
skip = skip,
sort = {"ts": -1},
projection = {
"_id": True,
"ts": True,
"syncTs": True,
"tokenId": True,
"serviceType": True,
"client": True,
"clientMessageId": True,
"clientThreadId": True,
"isSent": True,
"isBroadcast": True,
"sentSuccessfully": True,
"sender": True,
"chat": True,
"snippet": True,
"aiSnippet": True,
"tags": True
},
raise_exception = True
)
# Convert the fetched records to instances of the data model and return:
for record in records: record["message"] = {}
return [CoreMessageModel(**record) for record in records]
async def get_messages(
self,
mongo_conn: AsyncMongo,
token_ids: List[ObjectId | str],
limit: int = 100,
skip: int = 0,
additional_filter: dict = None
) -> List[CoreMessageModel] | None:
"""
Fetches many full messages in one call.
:param mongo_conn: The instance of the database connector to use for the operation.
:param token_ids: The token ids of the accounts from which these messages must be fetched.
:param limit: The max. no. of messages to retrieve in this call.
:param skip: The no. of initial messages to skip. Useful for pagination.
:param additional_filter: Any addition filters to use.
:return: The list of messages (as the message model). This list can be empty.
"""
# Prepare the filter:
if not isinstance(token_ids, list): token_ids = [token_ids]
token_ids = [ObjectId(t) for t in token_ids]
filter_json = {"tokenId": {"$in": token_ids}}
if additional_filter:
for k, v in additional_filter.items():
filter_json[k] = v
# We fetch the messages that are identified by a specific token id,
# with the specified fetching limits, while enforcing the sorting condition:
records = await mongo_conn.find_many(
collection = self.MESSAGES_COLLECTION,
filter = filter_json,
limit = limit,
skip = skip,
sort = {"ts": -1},
raise_exception = True
)
# Convert the fetched records to instances of the data model and return:
return [CoreMessageModel(**record) for record in records]
async def get_message(
self,
mongo_conn: AsyncMongo,
message_id: ObjectId | str,
) -> CoreMessageModel | None:
"""
Gets one message if you know its message id.
:param mongo_conn: The instance of the database connector to use for the operation.
:param message_id: The id of the message that needs to be read.
:return: The contents of that one message in a structured format.
"""
# We fetch the whole payload of that one message:
record = await mongo_conn.find_one(
collection = self.MESSAGES_COLLECTION,
filter = {"_id": ObjectId(message_id)},
raise_exception = True
)
# If no such message was found:
if record is None: return None
# If a record was found,
# we return it as our data model:
return CoreMessageModel(**record)
if updated_count < len(servers):
self._printer("Mismatched count", len(servers), updated_count)
return None
else: return servers
# ┏┓┳┓┳┳┳┓ ┳┳ ┓
# ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┏┫┏┓╋┏┓
# ┗┛┛┗┗┛┻┛ ┗┛┣┛┗┻┗┻┗┗
# ┛
# We don't support updating messages themselves,
# but we will allow updating fields like tags, marking as read or unread, etc.
async def update_tags(
async def check_one_server(
self,
mongo_conn: AsyncMongo,
message_id: ObjectId | str,
unset_tags: List[str] = None,
set_tags: List[str] = None
) -> bool:
mongo_data_conn: AsyncMongo,
server: CoreServerInfoModel,
insert_check_log: bool = True,
release_from_batch: bool = True
) -> CoreServerCheckLogModel:
"""
Updates the tags on one message. The tags to remove are processed first, the ones to add are processed later.
:param mongo_conn: The instance of the database connector to use for the operation.
:param message_id: The id of the message that needs to be read.
:param unset_tags: The tags to remove from the message.
:param set_tags: The tags to add to the message.
:return: True if the update was successful, else False.
"""
# Note down some assumptions and starting parameters:
server_check_log = CoreServerCheckLogModel(serverId = server.serverId)
start_time = time.time()
# Update the tags:
return await mongo_conn.update_one(
collection = self.MESSAGES_COLLECTION,
filter = {"_id": ObjectId(message_id)},
update = [{
"$set": {
"tags": {
"$let": {
"vars": {
"removed_tags": {
"$setDifference": [
"$tags",
unset_tags
]
}
},
"in": {
"$setUnion": [
"$$removed_tags",
set_tags
]
}
}
}
}
}],
raise_exception = True
try:
# Make the API call:
server_response = await self.__http_client.get(url = server.healthCheckUrl)
server_response.raise_for_status()
server_check_log.online = True
server_check_log.message = "The server responded."
# Catch various exceptions:
except httpx.HTTPStatusError as e: server_check_log.message = f"HTTP Error: {e.response.status_code}"
except httpx.ConnectTimeout as e: server_check_log.message = f"Connect Timeout Error: {e}"
except httpx.ReadTimeout as e: server_check_log.message = f"Read Timeout Error: {e}"
except httpx.TimeoutException as e: server_check_log.message = f"Misc. Timeout Error: {e}"
except httpx.NetworkError as e: server_check_log.message = f"Network Error: {e}"
except httpx.RequestError as e: server_check_log.message = f"Request Error: {e}"
# Note down the latency:
server_check_log.latency = time.time() - start_time
# In case of failure, we try to hit the alert URL:
if not server_check_log.online:
try:
alert_url_response = await self.__http_client.get(url = server.healthAlertUrl)
server_check_log.alertRaised = True if alert_url_response.status_code == 200 else False
except:
server_check_log.alertRaised = False
# Store the log in the database:
if insert_check_log: await mongo_data_conn.insert_one(
collection = self.SERVER_CHECK_LOGS_COLLECTION,
document = server_check_log.model_dump()
)
# ┏┓┳┓┳┳┳┓ ┳┓ ┓
# ┃ ┣┫┃┃┃┃ ━━ ┃┃┏┓┃┏┓╋┏┓
# ┗┛┛┗┗┛┻┛ ┻┛┗ ┗┗ ┗┗
# Release the picked server from the batch:
if release_from_batch: await mongo_data_conn.update_one(
collection = self.SERVERS_COLLECTION,
filter = {"_id": server.serverId},
update = {
"$set": {
"batchId": None,
"batchTs": None,
"online": server_check_log.online,
"lastCheckTs": server_check_log.ts,
"checkAfterTs": server_check_log.ts + datetime.timedelta(seconds = server.healthCheckInterval)
}
},
upsert = False
)
# Done here:
return server_check_log
# No support whatsoever for deleting messages.
async def check_server_batch(
self,
mongo_data_conn: AsyncMongo,
servers: List[CoreServerInfoModel]
) -> List[CoreServerCheckLogModel]:
# If the servers list is a blank array:
if not servers: return []
# Create tasks to check individual servers,
# tell the tasks not to update the database individually:
tasks = [
self.check_one_server(
mongo_data_conn = mongo_data_conn,
server = server,
insert_check_log = False,
release_from_batch = False
) for server in servers
]
# Fire the tasks and note down the results:
server_check_logs = await asyncio.gather(*tasks)
# Insert the logs:
await mongo_data_conn.insert_many(
collection = self.SERVER_CHECK_LOGS_COLLECTION,
documents = [log.model_dump() for log in server_check_logs]
)
# Release the batches:
await mongo_data_conn.bulk_write(
collection = self.SERVERS_COLLECTION,
requests = [
UpdateOne(
filter = {"_id": server.serverId},
update = {
"$set": {
"batchId": None,
"batchTs": None,
"online": log.online,
"lastCheckTs": log.ts,
"checkAfterTs": log.ts + datetime.timedelta(seconds = server.healthCheckInterval)
}
},
upsert = False
) for server, log in zip(servers, server_check_logs)
]
)
# Done here:
return server_check_logs
# *****************************************************************************************************************