From c0265ea79608d35d22ac42e0368a7b029687bf3e Mon Sep 17 00:00:00 2001 From: Konstantinos Kagkelidis Date: Thu, 6 Aug 2026 17:08:40 +0300 Subject: [PATCH] ARGO-5808 automate performance data ingestion --- automation/README.md | 1 + automation/argo_automator | 10 +++-- automation/argo_config.py | 2 + automation/argo_web_api.py | 48 +++-------------------- automation/check_readiness.py | 2 +- automation/config.yml.example | 2 + automation/init_ams.py | 65 ++++++++++++++++++------------- automation/init_compute_engine.py | 6 ++- automation/init_desy.py | 39 ++++++++----------- automation/init_ingest.py | 29 +++++++++++++- automation/init_node_registry.py | 17 +++----- automation/run_desy | 11 +++--- automation/run_ingest | 12 +++++- automation/run_node_registry | 20 +++++----- 14 files changed, 135 insertions(+), 129 deletions(-) diff --git a/automation/README.md b/automation/README.md index 499d4a29..378404d5 100644 --- a/automation/README.md +++ b/automation/README.md @@ -39,6 +39,7 @@ Submit an ingestion job for a tenant: **Options:** - `-c /path/to/config.yml` - Specify config file (default: `.config.yml`) +- `--performance` - Enable ingestion of performance data. By default is not enabled - `--no-verify` - Skip verification of remote endpoints like AMS - `--dry-run` - Preview what would be submitted without executing - `--log-level DEBUG` - Adjust logging verbosity diff --git a/automation/argo_automator b/automation/argo_automator index 7e19dffd..d0a8fa23 100755 --- a/automation/argo_automator +++ b/automation/argo_automator @@ -16,8 +16,8 @@ from argo_config import ArgoConfig from check_readiness import check_readiness from init_ams import init_ams from init_compute_engine import init_compute_engine -from init_mongo import init_mongo from init_desy import init_desy +from init_mongo import init_mongo from init_node_registry import init_node_registry REQUEST_TIMEOUT = 30 @@ -198,8 +198,9 @@ class ArgoAutomator: self.executor.submit(self.job_init_ams, tenant_id, tenant_name) return if job_name == JobName.INIT_COMPUTE_ENGINE.value: + performance=props.get("performance") self.executor.submit( - self.job_init_compute_engine, tenant_id, tenant_name + self.job_init_compute_engine, tenant_id, tenant_name, performance ) return @@ -359,7 +360,7 @@ class ArgoAutomator: except Exception as e: logger.exception(f"job failed for tenant {tenant_name}: {e}") - def job_init_compute_engine(self, tenant_id: str, tenant_name: str): + def job_init_compute_engine(self, tenant_id: str, tenant_name: str, performance: bool): """Job placeholder to init compute engine""" try: @@ -399,7 +400,7 @@ class ArgoAutomator: "Initialising Compute Engine", ) - job_done = init_compute_engine(self.config, tenant_id, tenant_name) + job_done = init_compute_engine(self.config, tenant_id, tenant_name, performance) if job_done: self.mon_api.update_status( @@ -534,3 +535,4 @@ def main(): if __name__ == "__main__": main() + diff --git a/automation/argo_config.py b/automation/argo_config.py index 079b7c5e..6b5d1c0d 100644 --- a/automation/argo_config.py +++ b/automation/argo_config.py @@ -22,6 +22,8 @@ def __init__(self, path: str): self.automation = automation self.tenants = tenants self.run = run + self.performance_db_url = automation.get("performance_db_url") + self.performance_db_token = automation.get("performance_db_token") self.node_registry_url = automation.get("node_registry_url") self.node_registry_token = automation.get("node_registry_token") self.ams_endpoint = automation.get("ams_endpoint") diff --git a/automation/argo_web_api.py b/automation/argo_web_api.py index 1d50fc7c..d02b7add 100644 --- a/automation/argo_web_api.py +++ b/automation/argo_web_api.py @@ -128,7 +128,6 @@ def get_topology_feed( return response.json().get("data") - def get_topology( self, tenant_id: str, @@ -152,9 +151,7 @@ def get_topology( except requests.exceptions.HTTPError as e: if e.response.status_code == 404: - logger.warning( - f"tenant: {tenant_name} ({tenant_id}) - has no topology" - ) + logger.warning(f"tenant: {tenant_name} ({tenant_id}) - has no topology") return [] else: raise @@ -333,43 +330,7 @@ def create_default_report( else: raise - - def create_topology_groups( - self, - tenant_id: str, - tenant_name: str, - tenant_access_token: str, - payload: object, - ): - """Http call to web-api to create topology groups""" - logger.debug( - f"tenant: {tenant_name} ({tenant_id}) - web-api creating topology groups..." - ) - - url = f"https://{self.config.web_api_endpoint}/api/v2/topology/groups" - headers = { - "x-api-key": tenant_access_token, - "Accept": "application/json", - } - try: - response = requests.post( - url, json=payload, headers=headers, timeout=REQUEST_TIMEOUT - ) - response.raise_for_status() - logger.info( - f"tenant: {tenant_name} ({tenant_id}) - web-api topology groups created" - ) - - except requests.exceptions.HTTPError as e: - if e.response.status_code == 409: - logger.warning( - f"tenant: {tenant_name} ({tenant_id}) - web-api topology groups already exist for specific date" - ) - return - else: - raise - - + def create_topology_service_types( self, tenant_id: str, @@ -452,7 +413,9 @@ def create_topology_groups( f"tenant: {tenant_name} ({tenant_id}) - web-api creating groups..." ) - url = f"https://{self.config.web_api_endpoint}/api/v2/topology/groups?force=true" + url = ( + f"https://{self.config.web_api_endpoint}/api/v2/topology/groups?force=true" + ) headers = { "x-api-key": tenant_access_token, "Accept": "application/json", @@ -475,7 +438,6 @@ def create_topology_groups( else: raise - def update_ready_state( self, tenant_id: str, diff --git a/automation/check_readiness.py b/automation/check_readiness.py index a624bd37..f0c6e391 100644 --- a/automation/check_readiness.py +++ b/automation/check_readiness.py @@ -112,5 +112,5 @@ def check_readiness(config: ArgoConfig, tenant_id: str, tenant_name: str) -> boo } # update the payload to web-api - result = web_api.update_ready_state(tenant_id, tenant_name, payload) + web_api.update_ready_state(tenant_id, tenant_name, payload) return True diff --git a/automation/config.yml.example b/automation/config.yml.example index dfa8481b..9b8a5995 100644 --- a/automation/config.yml.example +++ b/automation/config.yml.example @@ -8,6 +8,8 @@ automation: ams_event_project: ARGO-MON-AUTOMATION-PROJECT ams_event_subscription: events-subscription ams_admin_token: admin-s3cr3t + performance_db_url: https://perf.example.foo + perfomrance_db_token: perf-s3cret oidc_token_url: localhost:8080 oidc_client_id: tenant.status.integration.service diff --git a/automation/init_ams.py b/automation/init_ams.py index 6ce001ed..4ac67239 100644 --- a/automation/init_ams.py +++ b/automation/init_ams.py @@ -1,4 +1,5 @@ import logging + import requests from argo_ams_library import (AmsServiceException, AmsUser, AmsUserProject, ArgoMessagingService) @@ -9,17 +10,22 @@ REQUEST_TIMEOUT = 30 + # use http request to create component user because ams library doesn't support it -def create_ams_component_account(ams_endpoint: str, ams_token: str, username: str, email: str, project: str, role: str, component: str, component_project: str): +def create_ams_component_account( + ams_endpoint: str, + ams_token: str, + username: str, + email: str, + project: str, + role: str, + component: str, + component_project: str, +): payload = { "email": email, - "projects": [ - { - "project": project, - "roles": [role] - } - ], + "projects": [{"project": project, "roles": [role]}], } if component and component_project: @@ -32,13 +38,11 @@ def create_ams_component_account(ams_endpoint: str, ams_token: str, username: st "Accept": "application/json", } try: - response = requests.post( - url, json=payload, headers=headers, timeout=REQUEST_TIMEOUT - ) - response.raise_for_status() - logger.info( - f"ams user: {username} created for project: {project}" - ) + response = requests.post( + url, json=payload, headers=headers, timeout=REQUEST_TIMEOUT + ) + response.raise_for_status() + logger.info(f"ams user: {username} created for project: {project}") except requests.exceptions.HTTPError as e: if e.response.status_code == 409: @@ -93,21 +97,30 @@ def init_ams( for username, role, component, component_admin in user_roles: try: + if component and component_admin: + user = create_ams_component_account( + config.ams_endpoint, + config.ams_admin_token, + username, + config.argo_ops_email, + tenant_name, + role, + component, + component_admin, + ) - user = create_ams_component_account(config.ams_endpoint, config.ams_admin_token, username, config.argo_ops_email, tenant_name, role, component, component_admin) - - user = ams.create_user( - AmsUser( - name=username, - projects=[AmsUserProject(project=tenant_name, roles=[role])], - email=config.argo_ops_email, + user = ams.create_user( + AmsUser( + name=username, + projects=[AmsUserProject(project=tenant_name, roles=[role])], + email=config.argo_ops_email, + ) ) - ) - if user: - logger.info(f"ams project {tenant_name} - user created: {username}") - if role == "consumer" and username == consumer_username: - config.set_tenant_ams_access(tenant_id, tenant_name, user.token) + if user: + logger.info(f"ams project {tenant_name} - user created: {username}") + if role == "consumer" and username == consumer_username: + config.set_tenant_ams_access(tenant_id, tenant_name, user.token) except AmsServiceException as e: if e.code == 409: diff --git a/automation/init_compute_engine.py b/automation/init_compute_engine.py index 0425cb44..37b3912b 100644 --- a/automation/init_compute_engine.py +++ b/automation/init_compute_engine.py @@ -1,10 +1,11 @@ import json import logging import os + from jinja2 import Environment, FileSystemLoader + from argo_config import ArgoConfig from argo_web_api import ArgoWebApi - from init_ingest import run_ingest logger = logging.getLogger(__name__) @@ -38,6 +39,7 @@ def init_compute_engine( config: ArgoConfig, tenant_id: str, tenant_name: str, + performance: bool, ) -> bool: """Initialise compute engine users""" @@ -153,7 +155,9 @@ def init_compute_engine( config=config, tenant_name=tenant_name, tenant_ams_token=tenant.get("ams_token"), + tenant_mon_api_token=tenant.get("web_api_token"), dry_run=False, + performance=performance, verify=VERIFY, ) diff --git a/automation/init_desy.py b/automation/init_desy.py index 26956e90..b7a24e08 100644 --- a/automation/init_desy.py +++ b/automation/init_desy.py @@ -1,14 +1,11 @@ -import argparse import logging -import sys -import requests -import uuid -import json import os -from jinja2 import Environment, FileSystemLoader - +import uuid from urllib.parse import urlparse +import requests +from jinja2 import Environment, FileSystemLoader + from argo_config import ArgoConfig from argo_web_api import ArgoWebApi, TopoItem @@ -19,7 +16,6 @@ CRON_DIR = "/etc/cron.d" - def get_desy_topology(url: str): """Retrieve topology from desy marketplace""" logger.debug(f"retrieving topology from desy marketplace: {url}") @@ -57,6 +53,7 @@ def create_desy_cron(config: ArgoConfig, tenant_name: str) -> bool: return False + def remove_desy_cron(tenant_name: str) -> bool: """Remove cron file for desy integration""" logger.info(f"Attempting to remove desy cron file: argo_desy_{tenant_name}") @@ -73,6 +70,7 @@ def remove_desy_cron(tenant_name: str) -> bool: return False + def get_desy_feed(config: ArgoConfig, tenant_name: str) -> str: # get local tenant configuration @@ -92,24 +90,19 @@ def get_desy_feed(config: ArgoConfig, tenant_name: str) -> str: if feed_url: return feed_url - logger.warning( - f"Tenant has no desy-topology feed - aborting..." - ) + logger.warning("Tenant has no desy-topology feed - aborting...") return "" - - + def init_desy(config: ArgoConfig, tenant_name: str) -> bool: url = get_desy_feed(config, tenant_name) if url: create_desy_cron(config, tenant_name) - update_desy_topology(config,tenant_name,url) + update_desy_topology(config, tenant_name, url) return True return False - - def update_desy_topology(config: ArgoConfig, tenant_name: str, url: str): # get local tenant configuration @@ -123,7 +116,6 @@ def update_desy_topology(config: ArgoConfig, tenant_name: str, url: str): # get desy topology from remote endpoint desy = get_desy_topology(url) - old_endpoints = web_api.get_topology( tenant_id, tenant_name, tenant_web_api_token, TopoItem.ENDPOINTS ) @@ -158,7 +150,7 @@ def update_desy_topology(config: ArgoConfig, tenant_name: str, url: str): old_group = index_old_groups.get(item_name) if old_group: - + # Remove date key because it is not relevant for comparison old_group.pop("date", None) new_group = { @@ -248,17 +240,18 @@ def update_desy_topology(config: ArgoConfig, tenant_name: str, url: str): changed_endpoints = True if not changed_endpoints and not changed_groups: - if len(new_endpoints) == len(old_endpoints) and len(new_groups) == len(old_groups): + if len(new_endpoints) == len(old_endpoints) and len(new_groups) == len( + old_groups + ): logger.info( f"Desy-connector: Topology for tenant {tenant_name} remains the same - no upload" ) return - logger.info( - f"Desy-connector: Topology changes found for tenant {tenant_name} - will upload" - ) - + f"Desy-connector: Topology changes found for tenant {tenant_name} - will upload" + ) + web_api.create_topology_groups( tenant_id, tenant_name, tenant_web_api_token, new_groups ) diff --git a/automation/init_ingest.py b/automation/init_ingest.py index 98b1d8e8..9def2936 100644 --- a/automation/init_ingest.py +++ b/automation/init_ingest.py @@ -1,11 +1,10 @@ import logging import subprocess + import requests -import sys from argo_config import ArgoConfig - logger = logging.getLogger(__name__) @@ -21,7 +20,9 @@ def run_ingest( config: ArgoConfig, tenant_name: str, tenant_ams_token: str, + tenant_mon_api_token: str, dry_run: bool, + performance: bool, verify: str, ): """Function that composes the appropriate cli command to submit an ingest job execution in flink""" @@ -64,6 +65,30 @@ def run_ingest( tenant_name, ] + if performance: + cmd.extend( + [ + "--influx.org", + "argo", + "--influx.bucket", + tenant_name, + "--influx.endpoint", + config.performance_db_url, + "--influx.token", + config.performance_db_token, + "--influx.port", + "8086", + "--api.endpoint", + config.mon_api_endpoint, + "--api.token", + tenant_mon_api_token, + "--api.interval", + "1000", + "--api.timeout", + "1000", + ] + ) + if dry_run: print(("\033[92m" + " ".join(str(x) for x in cmd) + "\033[0m")) return 0 diff --git a/automation/init_node_registry.py b/automation/init_node_registry.py index 67b73d44..195a66a9 100644 --- a/automation/init_node_registry.py +++ b/automation/init_node_registry.py @@ -18,7 +18,6 @@ SERVICE_TYPE = "webportal" - def get_node_registry(url: str, token: str): """Retrieve the list of nodes from the node registry""" logger.debug(f"retrieving nodes from node registry: {url}") @@ -129,7 +128,7 @@ def build_group(tenant_name: str, node_name: str) -> dict: def build_endpoint(node_name: str, capability: dict, item_uuid: str) -> dict: item_URL = capability.get("endpoint") - hostname = urlparse(item_URL).hostname + hostname = str(urlparse(item_URL).hostname) tags = { "hostname": hostname, @@ -137,11 +136,11 @@ def build_endpoint(node_name: str, capability: dict, item_uuid: str) -> dict: "info_URL": item_URL, "monitored": "1", } - labels="" + capability_type = capability.get("capability_type") if capability_type: tags["info_capability_type"] = capability_type - tags["labels"]=capability_type + tags["labels"] = capability_type protocol = capability.get("protocol") if protocol: @@ -156,9 +155,7 @@ def build_endpoint(node_name: str, capability: dict, item_uuid: str) -> dict: } -def update_node_registry_topology( - config: ArgoConfig, tenant_name: str -): +def update_node_registry_topology(config: ArgoConfig, tenant_name: str): url = config.node_registry_url token = config.node_registry_token @@ -248,9 +245,7 @@ def update_node_registry_topology( changed_endpoints = True else: item_uuid = str(uuid.uuid4()) - new_endpoints.append( - build_endpoint(node_name, capability, item_uuid) - ) + new_endpoints.append(build_endpoint(node_name, capability, item_uuid)) changed_endpoints = True if not changed_endpoints and not changed_groups: @@ -273,4 +268,4 @@ def update_node_registry_topology( ) web_api.create_topology_endpoints( tenant_id, tenant_name, tenant_web_api_token, new_endpoints - ) \ No newline at end of file + ) diff --git a/automation/run_desy b/automation/run_desy index 536b6e28..e0e13849 100755 --- a/automation/run_desy +++ b/automation/run_desy @@ -1,18 +1,19 @@ #!/usr/bin/env python3 import argparse +import json import logging +import os import sys -import requests import uuid -import json -import os - from urllib.parse import urlparse +import requests + from argo_config import ArgoConfig from argo_web_api import ArgoWebApi, TopoItem -from init_desy import get_desy_feed, update_desy_topology, create_desy_cron, remove_desy_cron +from init_desy import (create_desy_cron, get_desy_feed, remove_desy_cron, + update_desy_topology) logger = logging.getLogger(__name__) diff --git a/automation/run_ingest b/automation/run_ingest index 58d5199f..e69f4946 100755 --- a/automation/run_ingest +++ b/automation/run_ingest @@ -4,8 +4,8 @@ import argparse import logging import sys -from init_ingest import run_ingest from argo_config import ArgoConfig +from init_ingest import run_ingest logger = logging.getLogger(__name__) @@ -45,6 +45,14 @@ def main(): dest="verify", help="Disable verification", ) + parser.add_argument( + "--performance", + action="store_const", + const=True, + default=False, + dest="performance", + help="Enable performance", + ) args = parser.parse_args() @@ -74,7 +82,7 @@ def main(): return 1 return run_ingest( - config, args.tenant, tenant.get("ams_token"), args.dry_run, args.verify + config, args.tenant, tenant.get("ams_token"), tenant.get("web_api_token"), args.dry_run, args.performance, args.verify ) diff --git a/automation/run_node_registry b/automation/run_node_registry index 290cd458..16c58b13 100755 --- a/automation/run_node_registry +++ b/automation/run_node_registry @@ -1,23 +1,21 @@ #!/usr/bin/env python3 import argparse +import json import logging +import os import sys -import requests import uuid -import json -import os - from urllib.parse import urlparse - + +import requests + from argo_config import ArgoConfig from argo_web_api import ArgoWebApi, TopoItem -from init_node_registry import ( - update_node_registry_topology, - create_node_registry_cron, - remove_node_registry_cron, -) - +from init_node_registry import (create_node_registry_cron, + remove_node_registry_cron, + update_node_registry_topology) + logger = logging.getLogger(__name__) REQUEST_TIMEOUT = 30