From b5e9b9b815ceaaba12394b04e10dad55434a5d0e Mon Sep 17 00:00:00 2001 From: PawelPlesniak Date: Thu, 9 Jul 2026 15:17:30 +0200 Subject: [PATCH 1/3] Cleaner URI resolution --- src/connectivityserver/connectionflask.py | 161 ++++++++++++++-------- tests/test_connectivityservice.py | 1 + 2 files changed, 107 insertions(+), 55 deletions(-) diff --git a/src/connectivityserver/connectionflask.py b/src/connectivityserver/connectionflask.py index ff2b50f..212e4c8 100755 --- a/src/connectivityserver/connectionflask.py +++ b/src/connectivityserver/connectionflask.py @@ -6,15 +6,17 @@ # import json -import logging import os import re from collections import namedtuple from datetime import datetime, timedelta from io import StringIO +import socket from threading import Lock +from urllib.parse import urlparse from flask import Flask, abort, make_response, request +from daqpytools.logging import get_daq_logger # Some functions exit with an abort(NNN) instead of return so don't complain! # ruff: noqa RET503 @@ -30,19 +32,52 @@ def convert_log_level(log_level): if log_level == 0: - return logging.WARNING + return 30# logging.WARNING if log_level == 1: - return logging.INFO + return 20# logging.INFO if log_level == 2: - return logging.DEBUG - return logging.INFO + return 10# logging.DEBUG + return 20# logging.INFO -logging.basicConfig( - level=convert_log_level(debug_level), - format="%(asctime)s %(levelname)s %(filename)s:%(funcName)s:%(lineno)d %(message)s", -) -log = logging.getLogger(__name__) +def map_uri_to_hostname(uri: str) -> str: + """ + Map a URI to a hostname. This function is used to extract the hostname from a URI string. + + Args: + uri (str): The URI string to be mapped. + + Returns: + str: The extracted hostname from the URI. + + Raises: + ValueError: If the URI is not in a valid format or does not contain a hostname. + """ + + try: + parsed_url = urlparse(uri) + ip_address = parsed_url.hostname + if not ip_address: + raise ValueError(f"Invalid URI format: {uri}. No hostname found.") + + hostname, aliases, ip_list = socket.gethostbyaddr(ip_address) + return hostname + + except socket.herror: + return f"No reverse DNS record found for IP: {ip_address}" + except Exception as e: + return f"Error parsing URI or resolving host: {e}" + + +# logging.basicConfig( +# level=convert_log_level(debug_level), +# format="%(asctime)s %(levelname)s %(filename)s:%(funcName)s:%(lineno)d %(message)s", +# ) +# log = logging.getLogger(__name__) + + + +log = get_daq_logger(__name__, log_level=convert_log_level(debug_level), stream_handlers=True) if "ENTRY_TTL" in os.environ: ttl = int(os.environ["ENTRY_TTL"]) @@ -105,10 +140,11 @@ def dump(): f'' f'' f'Name' - f'Connection' + f'Connection' # Increased colspan from 5 to 6 f"" f'' f"uri" + f"uri (resolved)" # Added the new header entry f"data_type" f"capacity" f"connection_type" @@ -120,9 +156,12 @@ def dump(): ) for k, v in store.items(): expired = now - v.time >= entry_ttl + resolved_hostname = urlparse(v.uri).scheme + "://" + map_uri_to_hostname(v.uri) # Call resolution function + dstream.write( f"{format_cell(k, expired)}" f"{format_cell(v.uri, expired)}" + f"{format_cell(resolved_hostname, expired)}" # Render resolved uri row f"{format_cell(v.data_type, expired)}" f"{format_cell(v.capacity, expired)}" f"{format_cell(v.connection_type, expired)}" @@ -213,57 +252,69 @@ def publish(): log.debug(f"{js=}") part = js["partition"] - log.info( + log.debug( f"{len(js['connections'])} connections in partition {part} from {request.remote_addr} uri={js['connections'][0]['uri']}..." ) - partlock.acquire() - if part in partitions: - store = partitions[part] - else: - store = {} - partitions[part] = store - global maxpartitions - if len(partitions) > maxpartitions: - maxpartitions = len(partitions) - if part not in maxentries: - maxentries[part] = 0 - - Connection = namedtuple( - "Connection", ["uri", "data_type", "capacity", "connection_type", "time"] - ) + with partlock: + if part in partitions: + store = partitions[part] + else: + store = {} + partitions[part] = store + global maxpartitions + if len(partitions) > maxpartitions: + maxpartitions = len(partitions) + if part not in maxentries: + maxentries[part] = 0 + + registered_partition_applications = list(store.keys()) + + Connection = namedtuple( + "Connection", ["uri", "data_type", "capacity", "connection_type", "time"] + ) + + for connection in js["connections"]: + if "uid" in connection and "uri" in connection: + uid = connection["uid"] + now = datetime.now() + + # current_keys = [str(k) for k in store.keys()] + # keys_str = ", ".join(str(k) for k in store.keys()) + # log.warning(f"store_keys={keys_str}") + if uid in registered_partition_applications: + log.debug(f"Updating address of existing application {uid} with uri {connection['uri']}") + else: + _uri = connection["uri"] + _port = urlparse(_uri).port + _hostname = map_uri_to_hostname(_uri) + log.info(f"Registering new application {uid} with uri {_uri} ({_hostname}:{_port})") + + store[uid] = Connection( + uri=connection["uri"], + connection_type=connection["connection_type"], + data_type=connection["data_type"], + capacity=( + connection["capacity"] if "capacity" in connection else 0 + ), + time=timestamp, + ) - for connection in js["connections"]: - if "uid" in connection and "uri" in connection: - uid = connection["uid"] now = datetime.now() - log.info(f"uid={uid}") - - store[uid] = Connection( - uri=connection["uri"], - connection_type=connection["connection_type"], - data_type=connection["data_type"], - capacity=( - connection["capacity"] if "capacity" in connection else 0 - ), # Backwards compatibility - time=timestamp, + elapsed = now - timestamp + + log.debug( + f"Took {elapsed.microseconds} us to add {len(js['connections'])} connections" ) - now = datetime.now() - elapsed = now - timestamp + global npublishes, publish_time + publish_time += elapsed + npublishes += 1 - log.debug( - f"Took {elapsed.microseconds} us to add {len(js['connections'])} connections" - ) - - global npublishes, publish_time - publish_time += elapsed - npublishes += 1 + if len(store) > maxentries[part]: + maxentries[part] = len(store) - if len(store) > maxentries[part]: - maxentries[part] = len(store) - partlock.release() return "OK" @@ -273,7 +324,7 @@ def retract_partition(): abort(400) js = json.loads(request.data) - log.debug(f"request=[{js}]") + log.warning(f"request=[{js}]") if "partition" not in js: abort(400) @@ -335,7 +386,7 @@ def get_connection(part): log.debug(f"{js=}") if "uid_regex" in js and "data_type" in js: - log.info( + log.debug( f"Searching for connections matching uid_regex<{js['uid_regex']}> and data_type {js['data_type']}" ) @@ -381,7 +432,7 @@ def get_connection(part): return "[" + ",".join(result) + "]" partlock.release() - log.info(f"Partition {part} not found") + log.debug(f"Partition {part} not found") abort(404) else: abort(400) diff --git a/tests/test_connectivityservice.py b/tests/test_connectivityservice.py index 7c8e4a7..fcffe83 100644 --- a/tests/test_connectivityservice.py +++ b/tests/test_connectivityservice.py @@ -1,4 +1,5 @@ import json + import pytest import connectivityserver.connectionflask as cf From 5ac2632d2d2eb59c3e47f44d52428ccf1638ea8d Mon Sep 17 00:00:00 2001 From: Pawel Plesniak <56116795+PawelPlesniak@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:12:29 +0200 Subject: [PATCH 2/3] Update src/connectivityserver/connectionflask.py Co-authored-by: Emir Muhammad <49058518+emmuhamm@users.noreply.github.com> --- src/connectivityserver/connectionflask.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/connectivityserver/connectionflask.py b/src/connectivityserver/connectionflask.py index 212e4c8..e599e70 100755 --- a/src/connectivityserver/connectionflask.py +++ b/src/connectivityserver/connectionflask.py @@ -69,11 +69,6 @@ def map_uri_to_hostname(uri: str) -> str: return f"Error parsing URI or resolving host: {e}" -# logging.basicConfig( -# level=convert_log_level(debug_level), -# format="%(asctime)s %(levelname)s %(filename)s:%(funcName)s:%(lineno)d %(message)s", -# ) -# log = logging.getLogger(__name__) From 9ac397b48a6249e0017175a6681da8c49d2cfd0a Mon Sep 17 00:00:00 2001 From: Pawel Plesniak <56116795+PawelPlesniak@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:13:53 +0200 Subject: [PATCH 3/3] Update src/connectivityserver/connectionflask.py Co-authored-by: Emir Muhammad <49058518+emmuhamm@users.noreply.github.com> --- src/connectivityserver/connectionflask.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/connectivityserver/connectionflask.py b/src/connectivityserver/connectionflask.py index e599e70..6fea27e 100755 --- a/src/connectivityserver/connectionflask.py +++ b/src/connectivityserver/connectionflask.py @@ -55,8 +55,7 @@ def map_uri_to_hostname(uri: str) -> str: """ try: - parsed_url = urlparse(uri) - ip_address = parsed_url.hostname + ip_address = urlparse(uri).hostname if not ip_address: raise ValueError(f"Invalid URI format: {uri}. No hostname found.")