diff --git a/api_v2/blueprints/servers/blueprint.py b/api_v2/blueprints/servers/blueprint.py index e8b8b49..4655102 100644 --- a/api_v2/blueprints/servers/blueprint.py +++ b/api_v2/blueprints/servers/blueprint.py @@ -6,12 +6,11 @@ DATE: - Created: Wednesday, 18th Sept., 2024 - Updated: Wednesday, 25th Dec., 2024 + Created: Wednesday, 25th Dec., 2024 OBJECTIVE: - To be able to fetch logs for rapid issue resolution. + To register and enlist servers for various projects. REFERENCES: @@ -60,7 +59,8 @@ from utils_v2.api.async_quart import ( # Models: from utils_v2.api.log import APILogModel -from models.logs.api import LogChainRequestData, LogsByFilterRequestData +from models.servers.api import RegisterServerRequestData +from models.servers.core import CoreServerInfoModel # For asynchronous activities: import asyncio @@ -74,7 +74,7 @@ import asyncio # Related to Quart: -logs_bp = Blueprint("int_logs", __name__) +servers_bp = Blueprint("int_servers", __name__) # ***************************************************************************************************************** @@ -94,7 +94,7 @@ logs_bp = Blueprint("int_logs", __name__) # ***************************************************************************************************************** -@logs_bp.record_once +@servers_bp.record_once def init(blueprint_setup_state): # This gets called when the blueprint is registered. @@ -105,288 +105,57 @@ def init(blueprint_setup_state): # --------------------------------------------------------------------------------------------------------------------- -@logs_bp.route("/get/id/", methods = ["POST", "GET"]) -@set_api_version(api_version = "2.1.0") -@should_not_be_under_maintenance(attr_name = "is_under_maintenance") -@only_whitelisted_ips(attr_name = "whitelisted_ips") -@handle_cancelled_request() -async def get_log( - log_id, - **kwargs -): - - """ - To get the log from its log id. - :param log_id: An identifier (string) for the log to fetch. - """ - - fetched_log = await current_app.mongo.find_one( - collection = "logs", - filter = {"logId": log_id}, - projection = {"_id": False} - ) - - if not fetched_log: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "No such log." - ) - - else: return ResponseModel( - status_code = StatusCodes.OK, - message = f"Log found.", - data = {"total": 1, "fetched": 1, "logs": fetched_log} - ) - - -# --------------------------------------------------------------------------------------------------------------------- - - -@logs_bp.route("/get/exception/id/", methods = ["POST", "GET"]) -@set_api_version(api_version = "2.1.0") -@should_not_be_under_maintenance(attr_name = "is_under_maintenance") -@only_whitelisted_ips(attr_name = "whitelisted_ips") -async def get_exception_from_log( - log_id, - **kwargs -): - - """ - To get the log's exception from its log id. - :param log_id: An identifier (string) for the log to fetch. - """ - - fetched_log = await current_app.mongo.find_one( - collection = "logs", - filter = {"logId": log_id}, - projection = { - "_id": False, - "logId": True, - "log": True, - "operation": True, - "ts": True, - "exception": True - } - ) - - if not fetched_log: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "No such log." - ) - - else: return ResponseModel( - status_code = StatusCodes.OK, - message = f"Log found.", - data = {"total": 1, "fetched": 1, "logs": fetched_log} - ) - - -# --------------------------------------------------------------------------------------------------------------------- - - -@logs_bp.route("/get/chain/", methods = ["POST", "GET"]) +@servers_bp.route("", methods = ["POST", "GET"]) @set_api_version(api_version = "2.1.0") @read_input(sanitize_headers = False, sanitize_data = False) @should_not_be_under_maintenance(attr_name = "is_under_maintenance") @only_whitelisted_ips(attr_name = "whitelisted_ips") -@validate_input(data_validator = lambda x: LogChainRequestData(**x)) -@handle_cancelled_request() -async def get_log_chain( - log_chain, - inbound_headers: dict = None, - inbound_data: dict | LogChainRequestData = None, - inbound_files: dict = None, - **kwargs -): - - """ - To get the series of logs from its chain identifier. - :param log_chain: An identifier (string) for the log chain to fetch. - :param inbound_headers: auto-extracted by the decorators. - :param inbound_data: auto-extracted by the decorators. - :param inbound_files: auto-extracted by the decorators. - :param kwargs: Any number of extra inputs supplied by the decorators. - """ - - total_count = await current_app.mongo.count( - collection = "logs", - filter = {"logChain": log_chain} - ) - - fetched_logs = await current_app.mongo.find_many( - collection = "logs", - filter = {"logChain": log_chain}, - projection = inbound_data.projection, - sort = inbound_data.sort, - limit = inbound_data.limit, - skip = inbound_data.skip - ) - - fetched_count = len(fetched_logs) - - if not fetched_logs: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "No such log chain." - ) - - else: return ResponseModel( - status_code = StatusCodes.OK, - message = f"{fetched_count} log(s) fetched", - data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs} - ) - - -# --------------------------------------------------------------------------------------------------------------------- - - -@logs_bp.route("/get/exception/chain/", methods = ["POST", "GET"]) -@set_api_version(api_version = "2.1.0") -@read_input(sanitize_headers = False, sanitize_data = False) -@should_not_be_under_maintenance(attr_name = "is_under_maintenance") -@only_whitelisted_ips(attr_name = "whitelisted_ips") -@validate_input(data_validator = lambda x: LogChainRequestData(**x)) -@handle_cancelled_request() -async def get_exceptions_from_log_chain( - log_chain, - inbound_headers: dict = None, - inbound_data: dict | LogChainRequestData = None, - inbound_files: dict = None, - **kwargs -): - - """ - To get the series of log exceptions from its chain identifier. - :param log_chain: An identifier (string) for the log chain to fetch. - :param inbound_headers: auto-extracted by the decorators. - :param inbound_data: auto-extracted by the decorators. - :param inbound_files: auto-extracted by the decorators. - :param kwargs: Any number of extra inputs supplied by the decorators. - """ - - total_count = await current_app.mongo.count( - collection = "logs", - filter = {"logChain": log_chain} - ) - - fetched_logs = await current_app.mongo.find_many( - collection = "logs", - filter = {"logChain": log_chain}, - projection = { - "_id": False, - "log": True, - "operation": True, - "ts": True, - "exception": True - }, - sort = inbound_data.sort, - limit = inbound_data.limit, - skip = inbound_data.skip - ) - - fetched_count = len(fetched_logs) - - if not fetched_logs: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "No such log chain." - ) - - else: return ResponseModel( - status_code = StatusCodes.OK, - message = f"{fetched_count} log(s) fetched.", - data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs} - ) - - -# --------------------------------------------------------------------------------------------------------------------- - - -@logs_bp.route("/get/filter", methods = ["POST", "GET"]) -@set_api_version(api_version = "2.1.0") -@read_input(sanitize_headers = False, sanitize_data = False) -@should_not_be_under_maintenance(attr_name = "is_under_maintenance") -@only_whitelisted_ips(attr_name = "whitelisted_ips") -@validate_input(data_validator = lambda x: LogsByFilterRequestData(**x)) +@validate_input(data_validator = lambda x: RegisterServerRequestData(**x)) @handle_cancelled_request() async def get_logs_by_filter( inbound_headers: dict = None, - inbound_data: dict | LogsByFilterRequestData = None, + inbound_data: dict | RegisterServerRequestData = None, inbound_files: dict = None, **kwargs ): """ - To fetch logs by custom filters: + To register a server's details. :param inbound_headers: auto-extracted by the decorators. :param inbound_data: auto-extracted by the decorators. :param inbound_files: auto-extracted by the decorators. :param kwargs: Any number of extra inputs supplied by the decorators. """ - total_count = await current_app.mongo.count( - collection = "logs", - filter = inbound_data.filter + batch = await current_app.server_controller.get_batch(mongo_data_conn = current_app.mongo) + logs = await current_app.server_controller.check_server_batch( + mongo_data_conn = current_app.mongo, + servers = batch ) + for log in logs: print("ONLINE:", log.online) - fetched_logs = await current_app.mongo.find_many( - collection = "logs", - filter = inbound_data.filter, - projection = inbound_data.projection, - sort = inbound_data.sort, - limit = inbound_data.limit, - skip = inbound_data.skip - ) + # for b in batch: + # print("SERVER:", json.to_string(b.model_dump(), default=str)) + # server_log = await current_app.server_controller.check_one_server( + # mongo_data_conn=current_app.mongo, + # server = b + # ) + # print("ONLINE:", server_log.online) + return "ok" - fetched_count = len(fetched_logs) - - if not fetched_logs: return ResponseModel( - status_code = StatusCodes.FAILED, - http_code = HttpCodes.NOT_FOUND, - message = "No matching logs." - ) - - else: return ResponseModel( - status_code = StatusCodes.OK, - message = f"{len(fetched_logs)} / {total_count} log(s) fetched.", - data = {"total": total_count, "fetched": fetched_count, "logs": fetched_logs} - ) - - -# --------------------------------------------------------------------------------------------------------------------- - - -@logs_bp.route("/set/api", methods = ["POST"]) -@set_api_version(api_version = "2.1.0") -@read_input(sanitize_headers = True, sanitize_data = True) -@should_not_be_under_maintenance(attr_name = "is_under_maintenance") -@only_whitelisted_ips(attr_name = "whitelisted_ips") -@validate_input(data_validator = lambda x: APILogModel(**x)) -async def set_api_log( - inbound_headers: dict = None, - inbound_data: dict | APILogModel = None, - inbound_files: dict = None, - **kwargs -): - - """ - To set logs from internal whitelisted IPs. - :param inbound_headers: auto-extracted by the decorators. - :param inbound_data: auto-extracted by the decorators. - :param inbound_files: auto-extracted by the decorators. - :param kwargs: Any number of extra inputs supplied by the decorators. - """ - - inserted_id = await current_app.mongo.insert_one( - collection = "logs", - document = inbound_data.model_dump() - ) - - return ResponseModel( - status_code = StatusCodes.OK if inserted_id else StatusCodes.FAILED, - http_code = HttpCodes.SUCCESS if inserted_id else HttpCodes.BAD_REQUEST - ) + # # register the server to the database: + # registered_server = await current_app.server_controller.register( + # mongo_data_conn = current_app.mongo, + # server = CoreServerInfoModel(**inbound_data.model_dump()) + # ) + # + # # Done here: + # success = True if registered_server else False + # return ResponseModel( + # status_code = StatusCodes.OK if success else StatusCodes.FAILED, + # http_code = HttpCodes.SUCCESS if success else HttpCodes.INTERNAL_SERVER_ERROR, + # message = "Server registered successfully" if success else "Failed to register server." + # ) # ***************************************************************************************************************** diff --git a/api_v2/main.py b/api_v2/main.py index e8429b7..f5004d4 100644 --- a/api_v2/main.py +++ b/api_v2/main.py @@ -67,9 +67,13 @@ from utils_v2.api.async_quart import ( # For debugging: from icecream import IceCreamDebugger +# Controllers: +from controllers.servers.server import CoreServerController + # All the blueprints: from api_v2.blueprints.cred_and_data.blueprint import cred_and_data_bp from api_v2.blueprints.logs.blueprint import logs_bp +from api_v2.blueprints.servers.blueprint import servers_bp # ***************************************************************************************************************** @@ -96,6 +100,7 @@ app = Quart(__name__) app = cors(app) app.register_blueprint(cred_and_data_bp, url_prefix = f"/{MODULE_BASE}") app.register_blueprint(logs_bp, url_prefix = f"/{MODULE_BASE}/logs") +app.register_blueprint(servers_bp, url_prefix = f"/{MODULE_BASE}/servers") # ***************************************************************************************************************** @@ -124,6 +129,10 @@ async def app_startup(**kwargs): :return: None. """ + # ┳ • • ┓• + # ┃┏┓┓╋┓┏┓┃┓┓┏┓ + # ┻┛┗┗┗┗┗┻┗┗┗┗ + # Safe-halt mechanism for upgrades (for a single-worker run): current_app.is_under_maintenance = False @@ -148,6 +157,16 @@ async def app_startup(**kwargs): filter = {"scriptId": script_id} ))["content"] + # ┏┓ ┓┓ + # ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ + # ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ + + current_app.server_controller = CoreServerController() + + # ┳┳┓• + # ┃┃┃┓┏┏ + # ┛ ┗┗┛┗• + # Pick the important stuff: current_app.whitelisted_ips = current_app.script_data["whitelistedIps"] diff --git a/controllers/servers/server.py b/controllers/servers/server.py index ea47e29..9412db7 100644 --- a/controllers/servers/server.py +++ b/controllers/servers/server.py @@ -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 # ***************************************************************************************************************** diff --git a/models/servers/api.py b/models/servers/api.py index 721b950..e97f40c 100644 --- a/models/servers/api.py +++ b/models/servers/api.py @@ -78,15 +78,7 @@ import datetime # ***************************************************************************************************************** -class CoreServerInfoModel(BaseModel): - - serverId: ObjectId | None = Field( - description = "the id of the document in mongodb that holds the info. about this server", - default = None, - frozen = True, - exclude = True, - alias = "_id" - ) +class RegisterServerRequestData(BaseModel): hostname: str = Field( description = "the hostname to identify the server", @@ -118,7 +110,7 @@ class CoreServerInfoModel(BaseModel): frozen = True ) - portNo: int | None = Field( + portNo: int = Field( description = "the port no. that this service is running on", default = None, frozen = True @@ -156,100 +148,19 @@ class CoreServerInfoModel(BaseModel): frozen = True ) - online: bool = Field( - description = "whether this service is online or not", - default = False, - frozen = False - ) - - batchId: str | None = Field( - description = "set some value here when checking the status of this server, set it to null once done", - default = None, - frozen = False - ) - - batchTs: AwareDatetime | None = Field( - description = "set the time (utc) when checking the status of this server, set it to null once done", - default = None, - frozen = False - ) - - firstRegTs: AwareDatetime = Field( - description = "the first time (utc) this server was registered in the database", - default_factory = lambda: date_time.get_current_utc_date_time(as_string = False), - frozen = True - ) - - lastRegTs: AwareDatetime = Field( - description = "the last time (utc) this server was registered in the database", - default_factory = lambda: date_time.get_current_utc_date_time(as_string=False), - frozen = True - ) - - lastCheckTs: AwareDatetime | None = Field( - description = "the last time (utc) this server was checked for being online", - default = None, - frozen = True - ) - - @computed_field - def checkAfterTs(self) -> datetime.datetime: - if self.lastCheckTs is None: return date_time.get_current_utc_date_time(as_string = False) - else: return self.lastCheckTs + datetime.timedelta(seconds = self.healthCheckInterval) - # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ # ┗┛┗┛┛┗┛┗┗┫ # ┛ class Config: - extra = "ignore" - arbitrary_types_allowed = True + extra = "forbid" # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ - @staticmethod - def parse_date_time(value): - - # If a null value was given, - # we can't do anything: - if not value: value = None - - # If the input is a string: - if isinstance(value, str): - value = value.strip() - value = date_time.parse_date_time( - input_value = value, - timezone = date_time.TIMEZONE_UTC, - date_formats = [ - "%Y-%m-%d", - "%Y-%m-%d %M:%H:%S", - "%Y%m%d", - "%Y%m%d %M:%H:%S", - ] - ) - - # If the input is already a date-time object, - # we just normalize the timestamp: - if isinstance(value, datetime.datetime): - value = date_time.to_timezone( - value, - timezone = date_time.TIMEZONE_UTC - ) - - # Done here: - return value - - @field_validator( - "batchTs", - "firstRegTs", "lastRegTs", - "lastCheckTs", - mode = "before" - ) - def parse_given_date_time(cls, value): - return cls.parse_date_time(value) + pass # ***************************************************************************************************************** @@ -261,29 +172,4 @@ class CoreServerInfoModel(BaseModel): if __name__ == "__main__": - test_json = { - "hostname": "kbprod", - "os": "Ubuntu 22.04.5 LTS", - "cpu": "x86_64 (x86_64)", - "pid": 456, - "ppid": 123, - "ipAddr": "127.0.0.1", - "portNo": 5000, - "project": "Bicree", - "service": "user", - "description": "This is Bicree's user module.", - "healthCheckUrl": "https://v2.api.bicree.com/user/metrics/memory", - "healthCheckInterval": 30, - "healthAlertUrl": ( - "https://nexcom.ditscentre.in/wtt/webhook/telegram/out" - "?appKey=9999999" - "&chatId=1275560043" - "&message=%F0%9F%9A%A8%20Bicree%27s%20user%20module%20is%20down%21%20Please%20check%20it%20urgently%21" - ), - "online": True, - "batchId": None, - "batchTs": None - } - - test_model = CoreServerInfoModel(**test_json) - print("TEST MODEL:", json.to_string(test_model.model_dump(), default = str)) + pass diff --git a/models/servers/core.py b/models/servers/core.py index a6efd23..6a81478 100644 --- a/models/servers/core.py +++ b/models/servers/core.py @@ -36,13 +36,17 @@ sys.path.append(".") sys.path.append("..") # For making data behaviour_models: -from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator, AwareDatetime +from pydantic import BaseModel, Field, field_validator, PastDatetime, model_validator, AwareDatetime, computed_field from typing import Optional, Literal, Union # My utils: +from utils_v2.string import json from utils_v2.string import regex from utils_v2.date_time import date_time +# To work with MongoDB: +from bson import ObjectId + # To work with date and time: import datetime @@ -76,6 +80,14 @@ import datetime class CoreServerInfoModel(BaseModel): + serverId: ObjectId | None = Field( + description = "the id of the document in mongodb that holds the info. about this server", + default = None, + frozen = True, + exclude = True, + alias = "_id" + ) + hostname: str = Field( description = "the hostname to identify the server", frozen = True @@ -106,8 +118,9 @@ class CoreServerInfoModel(BaseModel): frozen = True ) - portNo: str | None = Field( + portNo: int = Field( description = "the port no. that this service is running on", + default = None, frozen = True ) @@ -121,6 +134,12 @@ class CoreServerInfoModel(BaseModel): frozen = True ) + description: str = Field( + description = "a brief description about this service", + max_length = 350, + frozen = True + ) + healthCheckUrl: str = Field( description = "a get request will be sent to this url to see if the server is up", frozen = True @@ -143,7 +162,7 @@ class CoreServerInfoModel(BaseModel): frozen = False ) - batchId: str | None = Field( + batchId: ObjectId | str | None = Field( description = "set some value here when checking the status of this server, set it to null once done", default = None, frozen = False @@ -157,23 +176,26 @@ class CoreServerInfoModel(BaseModel): firstRegTs: AwareDatetime = Field( description = "the first time (utc) this server was registered in the database", + default_factory = lambda: date_time.get_current_utc_date_time(as_string = False), frozen = True ) lastRegTs: AwareDatetime = Field( description = "the last time (utc) this server was registered in the database", + default_factory = lambda: date_time.get_current_utc_date_time(as_string=False), frozen = True ) - lastCheckTs: AwareDatetime = Field( + lastCheckTs: AwareDatetime | None = Field( description = "the last time (utc) this server was checked for being online", + default = None, frozen = True ) - checkAfterTs: AwareDatetime = Field( - description = "the time (utc) after which this server needs to be checked again for being online", - frozen = True - ) + @computed_field + def checkAfterTs(self) -> datetime.datetime: + if self.lastCheckTs is None: return date_time.get_current_utc_date_time(as_string = False) + else: return self.lastCheckTs + datetime.timedelta(seconds = self.healthCheckInterval) # ┏┓ ┏• # ┃ ┏┓┏┓╋┓┏┓ @@ -182,6 +204,7 @@ class CoreServerInfoModel(BaseModel): class Config: extra = "ignore" + arbitrary_types_allowed = True # ┓┏ ┓• ┓ • # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ @@ -200,7 +223,12 @@ class CoreServerInfoModel(BaseModel): value = date_time.parse_date_time( input_value = value, timezone = date_time.TIMEZONE_UTC, - date_formats = ["%Y-%m-%d"] + date_formats = [ + "%Y-%m-%d", + "%Y-%m-%d %M:%H:%S", + "%Y%m%d", + "%Y%m%d %M:%H:%S", + ] ) # If the input is already a date-time object, @@ -217,13 +245,77 @@ class CoreServerInfoModel(BaseModel): @field_validator( "batchTs", "firstRegTs", "lastRegTs", - "lastCheckTs", "checkAfterTs", + "lastCheckTs", mode = "before" ) def parse_given_date_time(cls, value): return cls.parse_date_time(value) +# --------------------------------------------------------------------------------------------------------------------- + + +class CoreServerCheckLogModel(BaseModel): + + serverCheckId: ObjectId | None = Field( + description = "the id of the document in mongodb that holds the info. about this check", + default = None, + frozen = True, + exclude = True, + alias = "_id" + ) + + serverId: ObjectId = Field( + description = "the id of the server that we checked", + frozen = True + ) + + ts: AwareDatetime = Field( + description = "the time at which this check was made", + default_factory = lambda: date_time.get_current_utc_date_time(as_string = False), + frozen = True + ) + + latency: float | None = Field( + description = "the time the server took to respond to the health check", + default = None, + frozen = False + ) + + online: bool = Field( + description = "whether, or not, the server was online during this check", + default = False, + frozen = False + ) + + alertRaised: bool | None = Field( + description = "whether, or not, the alert url was successfully called (200 response)", + default = None, + frozen = False + ) + + message: str | None = Field( + description = "any additional message about this check", + default = None, + frozen = False + ) + + # ┏┓ ┏• + # ┃ ┏┓┏┓╋┓┏┓ + # ┗┛┗┛┛┗┛┗┗┫ + # ┛ + + class Config: + extra = "ignore" + arbitrary_types_allowed = True + + # ┓┏ ┓• ┓ • + # ┃┃┏┓┃┓┏┫┏┓╋┓┏┓┏┓ + # ┗┛┗┻┗┗┗┻┗┻┗┗┗┛┛┗ + + pass + + # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** @@ -233,4 +325,29 @@ class CoreServerInfoModel(BaseModel): if __name__ == "__main__": - pass + test_json = { + "hostname": "kbprod", + "os": "Ubuntu 22.04.5 LTS", + "cpu": "x86_64 (x86_64)", + "pid": 456, + "ppid": 123, + "ipAddr": "127.0.0.1", + "portNo": 5000, + "project": "Bicree", + "service": "user", + "description": "This is Bicree's user module.", + "healthCheckUrl": "https://v2.api.bicree.com/user/metrics/memory", + "healthCheckInterval": 30, + "healthAlertUrl": ( + "https://nexcom.ditscentre.in/wtt/webhook/telegram/out" + "?appKey=9999999" + "&chatId=1275560043" + "&message=%F0%9F%9A%A8%20Bicree%27s%20user%20module%20is%20down%21%20Please%20check%20it%20urgently%21" + ), + "online": True, + "batchId": None, + "batchTs": None + } + + test_model = CoreServerInfoModel(**test_json) + print("TEST MODEL:", json.to_string(test_model.model_dump(), default = str))