TriggerConfigAccessBase.py

Go to the documentation of this file.

# Copyright (C) 2002-2025 CERN for the benefit of the ATLAS collaboration
 
from abc import ABC, abstractmethod
import os
import json
import re
from typing import Any
import xml.etree.ElementTree as ET
from TrigConfStorage.TriggerCrestUtil import TriggerCrestUtil
import coral
 
from AthenaCommon.Logging import logging
log = logging.getLogger('TriggerConfigAccessBase.py')
 
def getFileType(filename):
    filetype = "unknown"
    with open(filename, 'r') as fp:
        config = json.load(fp)
        filetype = config['filetype']
    return filetype
 
from enum import Enum
class ConfigType(Enum):
    NONE = ("Config", "None", "None")
    L1MENU = ("L1Menu", "l1menu", "L1M")
    HLTMENU = ("HLTMenu", "hltmenu", "HLTM")
    L1PS = ("L1PrescalesSet", "l1prescale", "L1PS")
    HLTPS = ("HLTPrescalesSet", "hltprescale", "HLTPS")
    BGS = ("L1BunchGroupsSet", "bunchgroupset", "BGS")
    HLTJO = ("HLTJobOptions", "joboptions", "JO")
    HLTMON = ("HLTMonitoring", "hltmonitoringsummary", "MGS")
    def __init__(self, basename, filetype, crestkey):
        self.basename = basename
        self.filetype = filetype
        self.crestkey = crestkey
    def __eq__(self, other):
        if isinstance(other,str):
            return self.filetype == other
        else:
            return self.filetype == other.filetype
    def __ne__(self, other):
        return not self.__eq__(other)
 
class ConfigLoader(ABC):
    """ 
    ConfigLoader derived classes hold the information of the configuration source
    and define the method to load the configuration
    """ 
    def __init__(self,configType: ConfigType):
        self.configType: ConfigType = configType
    def confirmConfigType(self,config):
        """
        checks that the in-file specification of the configuration type matches the expected type
        """
        if config['filetype'] != self.configType:
            raise RuntimeError("Can not load file with filetype '%s' when expecting '%s'" % (config['filetype'], self.configType.filetype))
 
    @abstractmethod
    def load(self) -> dict[str, Any]:
        pass
 
    @abstractmethod
    def getWriteFilename(self) -> str:
        pass
 
    @abstractmethod
    def setQuery(self, query):
        pass
 
class ConfigFileLoader(ConfigLoader):
    def __init__(self, configType, filename ):
        super(ConfigFileLoader,self).__init__(configType) 
        self.filename = filename
    def load(self) -> dict[str, Any]:
        with open(self.filename, 'r') as fp:
            config = json.load(fp)
            self.confirmConfigType(config)
        return config
    def setQuery(self, query):
        pass
    def getWriteFilename(self):
        outfn = os.path.basename(self.filename)
        if outfn.endswith(".json"):
            outfn = outfn.rsplit('.',1)[0]
        return outfn + ".out.json"
 
class ConfigDirectLoader(ConfigLoader):
    """Class to load from json string"""
    def __init__(self, configType,  jsonString):
        super(ConfigDirectLoader,self).__init__(configType) 
        self.jsonString = jsonString
    def load(self) -> dict[str, Any]:
        config = json.loads(self.jsonString)
        self.confirmConfigType(config)
        return config
    def setQuery(self, query):
        pass
    def getWriteFilename(self):
        pass
 
class ConfigDBLoader(ConfigLoader):
    def __init__(self, configType, dbalias, dbkey):
        super().__init__(configType)
        self.dbalias = dbalias
        self.dbkey = dbkey
        self.query = {}
        self.schema = None
 
    def setQuery(self, query):
        """
        query template is a dictionary of queries, identified by schema version, 
        similar to TrigConf::TrigDBMenuLoader::m_hltQueries and TrigConf::TrigDBMenuLoader::m_l1Queries
        """
        self.query = query
 
    @staticmethod
    def getResolvedFileName(filename, pathenv=""):
        """ looks for file, first absolute, then by resolving envvar pathenv"""
        if os.access(filename,os.R_OK):
            return filename
        pathlist = os.getenv(pathenv,'').split(os.pathsep)
        for path in pathlist:
            f = os.path.join( path, filename )
            if os.access( f, os.R_OK ):
                return f
        raise RuntimeError("Can't read file %s, neither locally nor in %s" % (filename, pathenv) )
 
    @staticmethod
    def getConnectionParameters(dbalias):
        dblookupFile = ConfigDBLoader.getResolvedFileName("dblookup.xml", "CORAL_DBLOOKUP_PATH")
        dbp = ET.parse(dblookupFile)
        listOfServices = []
        foundAlias = False
        for logSvc in dbp.iter("logicalservice"):
            if logSvc.attrib["name"] != dbalias:
                continue
            foundAlias = True
            listOfServices = [ serv.attrib["name"] for serv in logSvc.iter("service") ]
            if len(listOfServices) == 0:
                raise RuntimeError("DB %s has no services listed in %s" % (dbalias, dblookupFile))
            break
        if not foundAlias:
            raise RuntimeError("DB %s not available in %s" % (dbalias, dblookupFile))
        
        if "FRONTIER_SERVER" not in os.environ:
            # remove all frontier connnections in the list if the environment FRONTIER_SERVER variable does not exist
            # this speeds up the resolution of the connection specification (dbalias)
            listOfServices: list[str] = [svc for svc in listOfServices if not svc.startswith("frontier:")]
 
        # now get the account and pw for oracle connections
        credentials: dict[str, Any] = dict.fromkeys(listOfServices)
 
        for svc in filter(lambda s : s.startswith("frontier:"), listOfServices):
            credentials[svc] = {}
            credentials[svc]["user"] = svc
            credentials[svc]["password"] = ""
 
        try:
            authFile = ConfigDBLoader.getResolvedFileName("authentication.xml", "CORAL_AUTH_PATH")
        except Exception as e:
            log.warning("File authentication.xml is not available! Oracle connection cannot be established. Exception message is: %s",e)
        else:
            for svc in filter(lambda s : s.startswith("oracle:"), listOfServices):
                ap = ET.parse(authFile)
                count = 0
                for con in filter( lambda c: c.attrib["name"]==svc, ap.iter("connection")):
                    credentials[svc] = dict([(par.attrib["name"],par.attrib["value"]) for par in con])
                    count += 1
                if count==0:
                    raise RuntimeError("No credentials found for connection %s from service %s for db %s" % (con,svc,dbalias))
                if count>1:
                    raise RuntimeError("More than 1 connection found in %s for service %s" % (authFile, svc))
        
        return credentials
 
    @staticmethod
    def getSchema(connStr):
        ''' Read schema from connection string '''
        if connStr.startswith("oracle:"):
            [_, schema] = connStr.split("/")[-2:]
            return schema
 
        if connStr.startswith("frontier:"):
            import re
            pattern = r"frontier://ATLF/\‍(\‍)/(.*)"
            m = re.match(pattern, connStr)
            if not m:
                raise RuntimeError("connection string '%s' doesn't match the pattern '%s'?" % (connStr, pattern))
            (schema, ) = m.groups()
            return schema
 
        if connStr.startswith("sqlite_file:"):
            raise NotImplementedError("Python-loading of trigger configuration from sqlite has not yet been implemented")
 
    @staticmethod
    def readSchemaVersion(qdict, session):
        ''' Read schema version form database, based on TrigConf::TrigDBLoader::schemaVersion '''
        try:
            q = "SELECT TS_TAG FROM {schema}.TRIGGER_SCHEMA TS"
            query = ConfigDBLoader.getCoralQuery(session, q.format(**qdict))
            cursor = query.execute()
            cursor.next()
 
            versionTag = cursor.currentRow()[0].data()
 
            versionTagPrefix = "Trigger-Run3-Schema-v"
            if not versionTag.startswith(versionTagPrefix):
                raise RuntimeError( "Tag format error: Trigger schema version tag %s does not start with %s", versionTag, versionTagPrefix) 
 
            vstr = versionTag[len(versionTagPrefix):]
 
            if not vstr.isdigit():
                raise RuntimeError( "Invalid argument when interpreting the version part %s of schema tag %s is %s", vstr, versionTag, type(vstr))
 
            log.debug("Found schema version %s", vstr)
            return int(vstr)
 
        except Exception as e:
            log.warning("Failed to read schema version: %r", e)
 
    @staticmethod
    def getCoralQuery(session, queryStr, qdict = None):
        ''' Parse output, tables and condition from the query string into coral query object'''
        query = session.nominalSchema().newQuery()
 
        if qdict is not None:
            queryStr = queryStr.format(**qdict)
 
        # bind vars
        bindVars = coral.AttributeList()  # type: ignore
        bindVarsInQuery = re.findall(r":(\w*)", queryStr)
        if len(bindVarsInQuery) > 0 and qdict is None:
            log.error("Query has bound-variable syntax but no value dictionary is provided. Query: %s", queryStr)
        for k in bindVarsInQuery:
            bindVars.extend(k, "int")
            bindVars[k].setData(qdict[k])
 
        output = queryStr.split("SELECT")[1].split("FROM")[0]
        for field in output.split(','):
            query.addToOutputList(field)
 
        log.debug("Conversion for Coral of query: %s", queryStr)
 
        for table in queryStr.split("FROM")[1].split("WHERE")[0].split(","):
            tableSplit = list(filter(None, table.split(" ")))
            # Schema name is stripped from TableList in Coral query
            query.addToTableList(tableSplit[0].split(".")[1], tableSplit[1])
 
        if "WHERE" in queryStr:
            cond = queryStr.split("WHERE")[1]
            m = re.match("(.*)(?i: ORDER *BY )(.*)", cond) # check for "order by" clause
            if m:
                where, order = m.groups()
                query.setCondition(where, bindVars)
                query.addToOrderList(order)
            else:
                query.setCondition(cond, bindVars)
 
        return query
 
    def getQueryDefinition(self, schemaVersion):
        '''Choose query based on schema version, based on TrigConf::TrigDBLoader::getQueryDefinition '''
        maxDefVersion = 0
        for vkey in self.query.keys():
            if vkey>maxDefVersion and vkey<=schemaVersion:
                maxDefVersion = vkey
 
        if maxDefVersion == 0:
            raise RuntimeError("No query available for schema version {0}".format(schemaVersion))
 
        return self.query[maxDefVersion]
 
    def load(self) -> dict[str, Any]:
        credentials: dict[str,Any] = ConfigDBLoader.getConnectionParameters(self.dbalias)
 
        if not credentials:
            log.error("No TriggerDB connections found for %s", self.dbalias)
            raise RuntimeError(f"No TriggerDB connections found for {self.dbalias}")
 
        svc = coral.ConnectionService() # type: ignore
        svcconfig = svc.configuration()
        svcconfig.disablePoolAutomaticCleanUp()
        svcconfig.setConnectionTimeOut(0)
 
        failureMode = 0
        for credential in credentials:
            log.debug("Trying credentials %s",credential)
 
            try: 
                session = svc.connect(credential, coral.access_ReadOnly) # type: ignore
            except Exception as e:
                log.warning("Failed to establish connection: %s",e)
                failureMode = max(1, failureMode)
                continue
 
            # Check that the FRONTIER_SERVER is set properly, if not reduce the retrial period and time out values
            if 'FRONTIER_SERVER' in os.environ and os.environ['FRONTIER_SERVER']:
                svcconfig.setConnectionRetrialPeriod(300)
                svcconfig.setConnectionRetrialTimeOut(3600)
            else:
                svcconfig.setConnectionRetrialPeriod(1)
                svcconfig.setConnectionRetrialTimeOut(1)
 
            try:
                session.transaction().start(True) # readOnly
                self.schema = ConfigDBLoader.getSchema(credential)
                qdict = { "schema" : self.schema, "dbkey" : self.dbkey }
                
                # Choose query based on schema
                schemaVersion = ConfigDBLoader.readSchemaVersion(qdict, session)
                qstr = self.getQueryDefinition(schemaVersion)
                # execute coral query
                query = ConfigDBLoader.getCoralQuery(session, qstr, qdict)
                cursor = query.execute()
 
            except Exception as e:
                log.warning(f"DB query on {credential} failed to execute.")
                log.warning("Exception message: %r", e)
                failureMode = max(2, failureMode)
                continue # to next source
 
            # Read query result
            if not cursor.next():
                # empty result
                log.warning(f"DB query on {credential} returned empty result, likely due to non-existing key {self.dbkey}")
                failureMode = 3
                continue # to next source
 
            configblob = cursor.currentRow()[0].data()
            if type(configblob) is not str:
                configblob = configblob.readline()
            config = json.loads(configblob)
            session.transaction().commit()
            
            self.confirmConfigType(config)
            return config
 
        if failureMode == 1:
            log.error("TriggerDB query: could not connect to any source for %s", self.configType.basename)
            log.error("Considered sources: %s", ", ".join(credentials))
            raise RuntimeError("TriggerDB query: could not connect to any source", self.configType.basename)
        if failureMode == 2:
            log.error("Query failed due to wrong definition for %s", self.configType.basename)
            log.error("DB query was: %s", qstr.format(**qdict))
            raise RuntimeError("Query failed due to wrong definition", self.configType.basename)
        elif failureMode == 3:
            log.error("DB key %s does not exist for %s", self.dbkey, self.configType.basename)
            raise KeyError("DB key does not exist", self.dbkey, self.configType.basename)
        else:
            raise RuntimeError("Query failed for unknown reason")
 
    # proposed filename when writing config to file
    def getWriteFilename(self) -> str:
        return "{basename}_{schema}_{dbkey}.json".format(basename = self.configType.basename, schema = self.schema, dbkey = self.dbkey)
 
class ConfigCrestLoader(ConfigLoader):
    def __init__(self, *, configType: ConfigType, dbname: str, dbkey: int, crestServer: str):
        super().__init__(configType)
        self.crestServer = crestServer
        self.dbname = dbname
        self.dbkey = dbkey
        self.schema = ""
 
    def setQuery(self, query):
        """
        With CREST all queries are defined in the CREST server
        """
        pass
 
    def _get_payload(self, hash: str) -> dict:
        """get payload from crest server using request library
 
        Args:
            hash (str): the query part of the url as required by the REST api
 
        Raises:
            RuntimeError: when connection or query failed
 
        Returns:
            dict: the json content
        """
        import requests
 
        url = f"{self.crestServer}/payloads/data"
        params = {
            "format": "BLOB", 
            "hash": hash
            }
        preq = requests.Request(method='GET', url=url, params=params).prepare()
        with requests.Session() as session:
            try:
                resp = session.send(preq)
            except requests.ConnectionError as exc:
                log.error(f"Could not connect to crest server {self.crestServer} ({exc})")
                raise RuntimeError(f"Could not connect to CREST server {self.crestServer}")
        
        if resp.status_code != 200:
            log.error(f"Error: HTTP GET request '{preq.url}' failed")
            raise RuntimeError(f"Query {hash} to crest failed with status code {resp.status_code}")
        
        config = json.loads(resp.content)
        self.confirmConfigType(config)
        return config
 
    def load(self) -> dict[str, Any]:
        # see SCHEMA_MAP in https://gitlab.cern.ch/crest-db/crest/-/blob/master/src/main/java/hep/crest/server/repositories/triggerdb/TriggerDb.java
        crest_conn = TriggerCrestUtil.getCrestConnection(self.dbname)
        if crest_conn is None:
            raise RuntimeError(f"Cannot resolve CREST connection for dbname {self.dbname}")
        hash = f"triggerdb://{crest_conn}/{self.configType.crestkey}/{self.dbkey}"
        config = self._get_payload(hash=hash)
        return config
 
    # proposed filename when writing config to file
    def getWriteFilename(self):
        schema = TriggerCrestUtil.crestconn_dbname_mapping.get(self.dbname, self.dbname)
        return "{basename}_{schema}_{dbkey}.json".format(basename = self.configType.basename, schema = schema, dbkey = self.dbkey)
 
class TriggerConfigAccess:
    """ 
    base class to hold the configuration (dict)
    and provides basic functions to access and print
    """
    def __init__(self, configType: ConfigType, mainkey: str, filename: str = "", jsonString: str = "", 
                 dbalias: str = "", dbkey: int = 0, useCrest: bool = False, crestServer: str = ""):
        self.loader: ConfigLoader = TriggerConfigAccess._getLoader(configType = configType, filename = filename, jsonString = jsonString, dbalias = dbalias, dbkey = dbkey, 
                        useCrest=useCrest, crestServer=crestServer)
        self._mainkey = mainkey
        self._config = None
 
    @staticmethod
    def _isCrestConnection(dbalias: str, useCrest: bool, crestServer: str) -> dict[str, str]:
        """ Determine whether the connection is a CREST connection
        """
        connParams: dict[str, str] = {}
        if dbalias.startswith("http://") or dbalias.startswith("https://"):
            # expect a complete crest connection string, consisting of server and dbname
            if (match := re.match(r'^(http[s]?://[^/]+/api-v\d+\.\d+)/(.+)$', dbalias)):
                connParams["crest_server"] = match.group(1)
                connParams["dbname"] = match.group(2)
            else:
                raise RuntimeError(f"Invalid CREST connection string: {dbalias}")
        elif dbalias in TriggerCrestUtil.allCrestConnections():
            connParams["crest_server"] = crestServer
            connParams["dbname"] = dbalias
        elif useCrest:
            connParams["crest_server"] = crestServer
            if (dbname := TriggerCrestUtil.getCrestConnection(dbalias)) is not None:
                connParams["dbname"] = dbname
            else:
                raise RuntimeError(f"Cannot resolve CREST connection '{dbalias}'")
        return connParams
 
    @staticmethod
    def _getLoader(*, configType, filename: str = "", jsonString: str = "", dbalias: str = "", dbkey: int = 0,
                   useCrest: bool = False, crestServer: str = "") -> ConfigLoader:
 
        if filename:
            return ConfigFileLoader(configType, filename )
        elif dbalias and dbkey>0:
            if (connParams := TriggerConfigAccess._isCrestConnection(dbalias, useCrest, crestServer)):
                return ConfigCrestLoader(configType=configType, dbname=connParams["dbname"], dbkey=dbkey, crestServer=connParams["crest_server"])
            else:
                return ConfigDBLoader(configType, dbalias, dbkey)
        elif jsonString:
            return ConfigDirectLoader(configType, jsonString )
        else:
            raise RuntimeError("Neither input file, nor JSON nor db alias and key provided")
 
    def load(self) -> None:
        self._config = self.loader.load()
 
    def __str__(self):
        return str(self._config)
 
    def __iter__(self):
        return iter(self[self._mainkey])
 
    def __getitem__(self, item):
        return self._config[item]
 
    def __len__(self):
        return len(self[self._mainkey])
 
    def config(self):
        """ returns the configuration """
        return self._config
 
    def prettyPrint(self):
        if self._config:
            print(json.dumps(self._config, indent = 4, separators=(',', ': ')))
    
    def name(self):
        return self["name"]
 
    def filetype(self):
        return self["filetype"]
 
    def printSummary(self):
        """ print summary info, should be overwritten by derived classes """
        log.info("Configuration name: {0}".format(self.name()))
        log.info("Configuration size: {0}".format(len(self)))
 
    def writeFile(self, filename: str | None = None):
        if filename is None:
            filename = self.loader.getWriteFilename()
        with open(filename, 'w') as fh:
            json.dump(self.config(), fh, indent = 4, separators=(',', ': '))
            log.info("Wrote file %s", filename)