HPCC-21168 Setup ESP logging agent to only run specified services

The changes are for all ESP logging agents (MySQLLoggingAgent,
CassandraLoggingAgent, ESPServerLoggingAgent, and
WSLogServiceLoggingAgent).

1. Modify setting scripts to allow users to set which logging services are
provided by a logging agent;
2. Create CLogAgentBase for basic functions shared by all log agents;
3. Move agent name and agent services into CLogAgentBase;
4. Set agent name and agent services in all log agent;
5. Call transaction seed/ID functions only the transaction seed/ID services
is set for the log agent;
6. Call UpdateLog functions only the UpdateLog service is set for the log
agent.

Signed-off-by: wangkx <kevin.wang@lexisnexis.com>

esp/logging/loggingagent/cassandraloggingagent/cassandralogagent.cpp

@@ -43,24 +43,35 @@ bool CCassandraLogAgent::init(const char* name, const char* type, IPropertyTree*
     if (!cfg)
         throw MakeStringException(-1, "Unable to find configuration for log agent %s:%s", name, type);
 
+    agentName.set(name);
+    const char* servicesConfig = cfg->queryProp("@services");
+    if (isEmptyString(servicesConfig))
+        throw MakeStringException(-1,"No Logging Service defined for %s", agentName.get());
+    setServices(servicesConfig);
+
     IPropertyTree* cassandra = cfg->queryBranch("Cassandra");
     if(!cassandra)
         throw MakeStringException(-1, "Unable to find Cassandra settings for log agent %s:%s", name, type);
 
     readDBCfg(cassandra, dbServer, dbUserID, dbPassword);
 
-    //Read information about data mapping for every log groups
-    readLogGroupCfg(cfg, defaultLogGroup, logGroups);
-    if (defaultLogGroup.isEmpty())
-        throw MakeStringException(-1,"LogGroup not defined");
+    if (hasService(LGSTUpdateLOG))
+    {
+        //Read information about data mapping for every log groups
+        readLogGroupCfg(cfg, defaultLogGroup, logGroups);
+        if (defaultLogGroup.isEmpty())
+            throw MakeStringException(-1,"LogGroup not defined");
 
-    //Read mapping between log sources and log groups
-    readLogSourceCfg(cfg, logSourceCount, logSourcePath, logSources);
+        //Read mapping between log sources and log groups
+        readLogSourceCfg(cfg, logSourceCount, logSourcePath, logSources);
+    }
 
     //Read transactions settings
-    readTransactionCfg(cfg);
-
-    maxTriesGTS = cfg->getPropInt("MaxTriesGTS", defaultMaxTriesGTS);
+    if (hasService(LGSTGetTransactionSeed))
+    {
+        readTransactionCfg(cfg);
+        maxTriesGTS = cfg->getPropInt("MaxTriesGTS", defaultMaxTriesGTS);
+    }
 
     //Setup Cassandra
     initKeySpace();

esp/logging/loggingagent/espserverloggingagent/loggingagent.cpp

@@ -43,6 +43,12 @@ bool CESPServerLoggingAgent::init(const char * name, const char * type, IPropert
     if (!cfg)
         return false;
 
+    agentName.set(name);
+    const char* servicesConfig = cfg->queryProp("@services");
+    if (isEmptyString(servicesConfig))
+        throw MakeStringException(-1,"No Logging Service defined for %s", agentName.get());
+    setServices(servicesConfig);
+
     IPropertyTree* espServer = cfg->queryBranch(PropESPServer);
     if(!espServer)
         throw MakeStringException(-1,"Unable to find ESPServer settings for log agent %s:%s", name, type);
@@ -59,63 +65,73 @@ bool CESPServerLoggingAgent::init(const char * name, const char * type, IPropert
         decrypt(serverPassword, password);
     }
     maxServerWaitingSeconds = cfg->getPropInt(PropServerWaitingSeconds);
-    maxGTSRetries = cfg->getPropInt(MaxTriesGTS, DefaultMaxTriesGTS);
-
-    transactionSeedType.set(cfg->hasProp(PropTransactionSeedType) ? cfg->queryProp(PropTransactionSeedType) :
-        DefaultTransactionSeedType);
-    alternativeTransactionSeedType.set(cfg->hasProp(PropAlternativeTransactionSeedType) ?
-        cfg->queryProp(PropAlternativeTransactionSeedType) : DefaultAlternativeTransactionSeedType);
 
-    BoolHash uniqueGroupNames;
-    StringBuffer sourceName, groupName, dbName, localTransactionSeed;
-    Owned<IPropertyTreeIterator> iter = cfg->getElements("LogSourceMap/LogSource");
-    ForEach(*iter)
+    if (hasService(LGSTUpdateLOG))
     {
-        ensureInputString(iter->query().queryProp("@name"), false, sourceName, -1, "LogSource @name required");
-        ensureInputString(iter->query().queryProp("@maptologgroup"), true, groupName, -1, "LogSource @maptologgroup required");
-        ensureInputString(iter->query().queryProp("@maptodb"), true, dbName, -1, "LogSource @maptodb required");
-        Owned<CLogSource> logSource = new CLogSource(sourceName.str(), groupName.str(), dbName.str());
-        logSources.setValue(sourceName.str(), logSource);
-
-        bool* found = uniqueGroupNames.getValue(groupName.str());
-        if (!found || !*found)
+        BoolHash uniqueGroupNames;
+        StringBuffer sourceName, groupName, dbName;
+        Owned<IPropertyTreeIterator> iter = cfg->getElements("LogSourceMap/LogSource");
+        ForEach(*iter)
         {
-            uniqueGroupNames.setValue(groupName.str(), true);
-
-            unsigned maxSeq = 0;
-            unsigned maxLength = 0;
-            unsigned seedExpiredSeconds = 0;
-            VStringBuffer xpath("LogGroup/[@name='%s']", groupName.str());
-            IPropertyTree* logGroup = cfg->queryBranch(xpath.str());
-            if (logGroup)
-            {
-                maxLength = logGroup->getPropInt(PropMaxTransIDLength, 0);
-                maxSeq = logGroup->getPropInt(PropMaxTransIDSequenceNumber, 0),
-                seedExpiredSeconds = 60 * logGroup->getPropInt(PropMaxTransSeedTimeoutMinutes, 0);
-            }
-
-            StringBuffer transactionSeed, statusMessage;
-            getTransactionSeed(groupName.str(), transactionSeed, statusMessage);
-            if (transactionSeed.length() > 0)
+            ensureInputString(iter->query().queryProp("@name"), false, sourceName, -1, "LogSource @name required");
+            ensureInputString(iter->query().queryProp("@maptologgroup"), true, groupName, -1, "LogSource @maptologgroup required");
+            ensureInputString(iter->query().queryProp("@maptodb"), true, dbName, -1, "LogSource @maptodb required");
+            Owned<CLogSource> logSource = new CLogSource(sourceName.str(), groupName.str(), dbName.str());
+            logSources.setValue(sourceName.str(), logSource);
+
+            bool* found = uniqueGroupNames.getValue(groupName.str());
+            if (!found || !*found)
             {
-                Owned<CTransIDBuilder> entry = new CTransIDBuilder(transactionSeed.str(), false, transactionSeedType.get(),
-                    maxLength, maxSeq, seedExpiredSeconds);
-                transIDMap.setValue(groupName.str(), entry);
-                if (iter->query().getPropBool("@default", false))
-                    defaultGroup.set(groupName.str());
+                uniqueGroupNames.setValue(groupName.str(), true);
+
+                unsigned maxSeq = 0;
+                unsigned maxLength = 0;
+                unsigned seedExpiredSeconds = 0;
+                VStringBuffer xpath("LogGroup/[@name='%s']", groupName.str());
+                IPropertyTree* logGroup = cfg->queryBranch(xpath.str());
+                if (logGroup)
+                {
+                    maxLength = logGroup->getPropInt(PropMaxTransIDLength, 0);
+                    maxSeq = logGroup->getPropInt(PropMaxTransIDSequenceNumber, 0),
+                    seedExpiredSeconds = 60 * logGroup->getPropInt(PropMaxTransSeedTimeoutMinutes, 0);
+                }
+
+                if (!hasService(LGSTGetTransactionSeed) && !hasService(LGSTGetTransactionID))
+                    continue;
+
+                StringBuffer transactionSeed, statusMessage;
+                getTransactionSeed(groupName.str(), transactionSeed, statusMessage);
+                if (transactionSeed.length() > 0)
+                {
+                    Owned<CTransIDBuilder> entry = new CTransIDBuilder(transactionSeed.str(), false, transactionSeedType.get(),
+                        maxLength, maxSeq, seedExpiredSeconds);
+                    transIDMap.setValue(groupName.str(), entry);
+                    if (iter->query().getPropBool("@default", false))
+                        defaultGroup.set(groupName.str());
+                }
+                else
+                    PROGLOG("Failed to get TransactionSeed for <%s>", groupName.str());
             }
-            else
-                PROGLOG("Failed to get TransactionSeed for <%s>", groupName.str());
         }
+        logContentFilter.readAllLogFilters(cfg);
     }
 
+    if (!hasService(LGSTGetTransactionSeed) && !hasService(LGSTGetTransactionID))
+        return true;
+
+    maxGTSRetries = cfg->getPropInt(MaxTriesGTS, DefaultMaxTriesGTS);
+    transactionSeedType.set(cfg->hasProp(PropTransactionSeedType) ? cfg->queryProp(PropTransactionSeedType) :
+        DefaultTransactionSeedType);
+    alternativeTransactionSeedType.set(cfg->hasProp(PropAlternativeTransactionSeedType) ?
+        cfg->queryProp(PropAlternativeTransactionSeedType) : DefaultAlternativeTransactionSeedType);
+
+    StringBuffer localTransactionSeed;
     createLocalTransactionSeed(localTransactionSeed);
     Owned<CTransIDBuilder> localTransactionEntry = new CTransIDBuilder(localTransactionSeed.str(), true, alternativeTransactionSeedType.get(),
         cfg->getPropInt(PropMaxTransIDLength, 0), cfg->getPropInt(PropMaxTransIDSequenceNumber, 0),
         60 * cfg->getPropInt(PropMaxTransSeedTimeoutMinutes, 0));
     transIDMap.setValue(appESPServerLoggingAgent, localTransactionEntry);
 
-    logContentFilter.readAllLogFilters(cfg);
     return true;
 }
 
@@ -132,6 +148,9 @@ void CESPServerLoggingAgent::createLocalTransactionSeed(StringBuffer& transactio
 
 bool CESPServerLoggingAgent::getTransactionSeed(IEspGetTransactionSeedRequest& req, IEspGetTransactionSeedResponse& resp)
 {
+    if (!hasService(LGSTGetTransactionSeed))
+        throw MakeStringException(EspLoggingErrors::GetTransactionSeedFailed, "%s: no getTransactionSeed service configured", agentName.get());
+
     StringBuffer statusMessage, transactionSeed;
     int statusCode = getTransactionSeed(req.getApplication(), transactionSeed, statusMessage);
     resp.setStatusCode(statusCode);
@@ -229,6 +248,9 @@ void CESPServerLoggingAgent::resetTransSeed(CTransIDBuilder *builder, const char
 
 void CESPServerLoggingAgent::getTransactionID(StringAttrMapping* transFields, StringBuffer& transactionID)
 {
+    if (!hasService(LGSTGetTransactionID))
+        throw MakeStringException(EspLoggingErrors::GetTransactionSeedFailed, "%s: no getTransactionID service configured", agentName.get());
+
     const char* groupName = nullptr;
     CTransIDBuilder* transIDBuilder = nullptr;
     StringAttr* source = nullptr;
@@ -272,6 +294,9 @@ bool CESPServerLoggingAgent::updateLog(IEspUpdateLogRequestWrap& req, IEspUpdate
 {
     try
     {
+        if (!hasService(LGSTUpdateLOG))
+            throw MakeStringException(EspLoggingErrors::UpdateLogFailed, "%s: no updateLog service configured", agentName.get());
+
         StringBuffer soapreq(
             "<?xml version=\"1.0\" encoding=\"UTF-8\"?>"
             "<soap:Envelope xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\""

esp/logging/loggingagent/espserverloggingagent/loggingagent.hpp

@@ -27,7 +27,7 @@
     #define ESPSERVERLOGGINGAGENT_API DECL_IMPORT
 #endif
 
-class CESPServerLoggingAgent : public CInterface, implements IEspLogAgent
+class CESPServerLoggingAgent : public CLogAgentBase
 {
     StringBuffer serviceName, loggingAgentName, defaultGroup;
     StringBuffer serverUrl, serverUserID, serverPassword;

esp/logging/logginglib/loggingagentbase.cpp

@@ -321,6 +321,9 @@ void CDBLogAgentBase::readTransactionCfg(IPropertyTree* cfg)
 
 bool CDBLogAgentBase::getTransactionSeed(IEspGetTransactionSeedRequest& req, IEspGetTransactionSeedResponse& resp)
 {
+    if (!hasService(LGSTGetTransactionSeed))
+        throw MakeStringException(EspLoggingErrors::GetTransactionSeedFailed, "%s: no getTransactionSeed service configured", agentName.get());
+
     bool bRet = false;
     StringBuffer appName = req.getApplication();
     appName.trim();
@@ -366,6 +369,9 @@ bool CDBLogAgentBase::getTransactionSeed(IEspGetTransactionSeedRequest& req, IEs
 
 bool CDBLogAgentBase::updateLog(IEspUpdateLogRequestWrap& req, IEspUpdateLogResponse& resp)
 {
+    if (!hasService(LGSTUpdateLOG))
+        throw MakeStringException(EspLoggingErrors::UpdateLogFailed, "%s: no updateLog service configured", agentName.get());
+
     unsigned startTime = (getEspLogLevel()>=LogNormal) ? msTick() : 0;
     bool ret = false;
     try

esp/logging/logginglib/loggingagentbase.hpp

@@ -241,6 +241,7 @@ interface IEspLogAgent : extends IInterface
     virtual bool getTransactionSeed(IEspGetTransactionSeedRequest& req, IEspGetTransactionSeedResponse& resp) = 0;
     virtual void getTransactionID(StringAttrMapping* transFields, StringBuffer& transactionID) = 0;
     virtual bool updateLog(IEspUpdateLogRequestWrap& req, IEspUpdateLogResponse& resp) = 0;
+    virtual bool hasService(LOGServiceType service) = 0;
     virtual IEspUpdateLogRequestWrap* filterLogContent(IEspUpdateLogRequestWrap* req) = 0;
 };
 
@@ -284,7 +285,50 @@ public:
     IEspUpdateLogRequestWrap* filterLogContent(IEspUpdateLogRequestWrap* req);
 };
 
-class LOGGINGCOMMON_API CDBLogAgentBase : public CInterface, implements IEspLogAgent
+class LOGGINGCOMMON_API CLogAgentBase : public CInterface, implements IEspLogAgent
+{
+protected:
+    StringAttr agentName;
+    LOGServiceType services[MAXLOGSERVICES];
+
+    bool hasService(LOGServiceType service)
+    {
+        unsigned int i = 0;
+        while (services[i] != LGSTterm)
+        {
+            if (services[i] == service)
+                return true;
+            i++;
+        }
+        return false;
+    }
+    void setServices(const char* servicesConfig)
+    {
+        StringArray serviceArray;
+        serviceArray.appendListUniq(servicesConfig, ",");
+
+        unsigned i=0;
+        ForEachItemIn(s, serviceArray)
+        {
+            const char* service = serviceArray.item(s);
+            if (service && strieq(service, "UpdateLOG"))
+                services[i++] = LGSTUpdateLOG;
+            else if (service && strieq(service, "GetTransactionSeed"))
+                services[i++] = LGSTGetTransactionSeed;
+            else if (service && strieq(service, "GetTransactionID"))
+                services[i++] = LGSTGetTransactionID;
+        }
+        services[i] = LGSTterm;
+    };
+public:
+    IMPLEMENT_IINTERFACE;
+
+    CLogAgentBase() { services[0] = LGSTterm; };
+    virtual ~CLogAgentBase() {};
+};
+
+
+class LOGGINGCOMMON_API CDBLogAgentBase : public CLogAgentBase
 {
 protected:
     StringBuffer defaultDB, transactionTable, loggingTransactionSeed;

esp/logging/logginglib/logthread.cpp

@@ -58,26 +58,6 @@ CLogThread::CLogThread(IPropertyTree* _cfg , const char* _service, const char* _
     if(!_logAgent)
         throw MakeStringException(-1,"No Logging agent interface for %s", _agentName);
 
-    const char* servicesConfig = _cfg->queryProp("@services");
-    if (!servicesConfig || !*servicesConfig)
-        throw MakeStringException(-1,"No Logging Service defined for %s", _agentName);
-
-    StringArray serviceArray;
-    serviceArray.appendListUniq(servicesConfig, ",");
-
-    unsigned i=0;
-    ForEachItemIn(s, serviceArray)
-    {
-        const char* service = serviceArray.item(s);
-        if (service && strieq(service, "UpdateLOG"))
-            services[i++] = LGSTUpdateLOG;
-        else if (service && strieq(service, "GetTransactionSeed"))
-            services[i++] = LGSTGetTransactionSeed;
-        else if (service && strieq(service, "GetTransactionID"))
-            services[i++] = LGSTGetTransactionID;
-    }
-    services[i] = LGSTterm;
-
     logAgent.setown(_logAgent);
 
     maxLogQueueLength = _cfg->getPropInt(PropMaxLogQueueLength, MaxLogQueueLength);

esp/logging/logginglib/logthread.hpp

@@ -46,7 +46,6 @@ class CLogThread : public Thread , implements IUpdateLogThread
     unsigned maxLogRetries;   // Max. # of attempts to send log message
 
     Owned<IEspLogAgent> logAgent;
-    LOGServiceType services[MAXLOGSERVICES];
     QueueOf<IInterface, false> logQueue;
     CriticalSection logQueueCrit;
     Semaphore       m_sem;
@@ -72,21 +71,8 @@ public:
 
     bool hasService(LOGServiceType service)
     {
-        unsigned int i = 0;
-        while (services[i] != LGSTterm)
-        {
-            if (services[i] == service)
-                return true;
-            i++;
-        }
-        return false;
+        return logAgent->hasService(service);
     }
-    void addService(LOGServiceType service)
-    {
-        unsigned i=0;
-        while (services[i] != LGSTterm) i++;
-        services[i] = service;
-    };
 
     int run();
     void start();

initfiles/componentfiles/configxml/@temp/logging_agent.xsl

@@ -118,7 +118,13 @@ xmlns:set="http://exslt.org/sets">
         <xsl:if test="string($agentNode/@serverIP) = ''">
             <xsl:message terminate="yes">Cassandra server network address is undefined for <xsl:value-of select="$agentName"/>!</xsl:message>
         </xsl:if>
-        <LogAgent name="{$agentName}" type="LogAgent" services="GetTransactionSeed,UpdateLog,GetTransactionID" plugin="cassandralogagent">
+        <xsl:variable name="Services">
+            <xsl:choose>
+                <xsl:when test="string($agentNode/@Services) != ''"><xsl:value-of select="$agentNode/@Services"/></xsl:when>
+                <xsl:otherwise>GetTransactionSeed,UpdateLog,GetTransactionID</xsl:otherwise>
+            </xsl:choose>
+        </xsl:variable>
+        <LogAgent name="{$agentName}" type="LogAgent" services="{$Services}" plugin="cassandralogagent">
             <Cassandra server="{$agentNode/@serverIP}" dbUser="{$agentNode/@userName}" dbPassWord="{$agentNode/@userPassword}" dbName="{$agentNode/@ksName}"/>
 
             <xsl:call-template name="LogBasic">
@@ -165,7 +171,13 @@ xmlns:set="http://exslt.org/sets">
         </xsl:if>
 
         <xsl:variable name="wsloggingUrl"><xsl:text>http://</xsl:text><xsl:value-of select="$espNetAddress"/><xsl:text>:</xsl:text><xsl:value-of select="$espPort"/></xsl:variable>
-        <LogAgent name="{$agentName}" type="LogAgent" services="GetTransactionSeed,UpdateLog,GetTransactionID" plugin="espserverloggingagent">
+        <xsl:variable name="Services">
+            <xsl:choose>
+                <xsl:when test="string($agentNode/@Services) != ''"><xsl:value-of select="$agentNode/@Services"/></xsl:when>
+                <xsl:otherwise>GetTransactionSeed,UpdateLog,GetTransactionID</xsl:otherwise>
+            </xsl:choose>
+        </xsl:variable>
+        <LogAgent name="{$agentName}" type="LogAgent" services="{$Services}" plugin="espserverloggingagent">
             <ESPServer url="{$wsloggingUrl}" user="{$agentNode/@User}" password="{$agentNode/@Password}"/>
             <xsl:if test="string($agentNode/@MaxServerWaitingSeconds) != ''">
                 <MaxServerWaitingSeconds><xsl:value-of select="$agentNode/@MaxServerWaitingSeconds"/></MaxServerWaitingSeconds>

initfiles/componentfiles/configxml/@temp/wslogserviceespagent.xsl

@@ -40,7 +40,13 @@ xmlns:set="http://exslt.org/sets">
             <xsl:message terminate="yes">Log Data XPath is undefined for <xsl:value-of select="$agentName"/> </xsl:message>
         </xsl:if>
 
-        <LogAgent name="{$agentName}" type="LogAgent" services="GetTransactionSeed,UpdateLog,GetTransactionID" plugin="wslogserviceespagent">
+        <xsl:variable name="Services">
+            <xsl:choose>
+                <xsl:when test="string($agentNode/@Services) != ''"><xsl:value-of select="$agentNode/@Services"/></xsl:when>
+                <xsl:otherwise>GetTransactionSeed,UpdateLog,GetTransactionID</xsl:otherwise>
+            </xsl:choose>
+        </xsl:variable>
+        <LogAgent name="{$agentName}" type="LogAgent" services="{$Services}" plugin="wslogserviceespagent">
             <LoggingServer url="{$loggingServerUrl}" user="{$loggingServer/@User}" password="{$loggingServer/@Password}"/>
             <xsl:if test="string($agentNode/@FailSafe) != ''">
                 <FailSafe><xsl:value-of select="$agentNode/@FailSafe"/></FailSafe>

initfiles/componentfiles/configxml/cassandraloggingagent.xsd

@@ -63,6 +63,13 @@
           </xs:appinfo>
         </xs:annotation>
       </xs:attribute>
+      <xs:attribute name="Services" type="xs:string" use="optional" default="GetTransactionSeed,UpdateLog">
+        <xs:annotation>
+          <xs:appinfo>
+            <tooltip>the names of logging services to be supported in this logging agent.</tooltip>
+          </xs:appinfo>
+        </xs:annotation>
+      </xs:attribute>
       <xs:attribute name="MaxTriesGTS" type="xs:nonNegativeInteger" use="optional" default="3">
         <xs:annotation>
           <xs:appinfo>
@@ -238,6 +245,7 @@
         </xs:appinfo>
       </xs:annotation>
     </xs:attribute>
+    
     <xs:attribute name="ksName" type="xs:string" use="optional" >
       <xs:annotation>
          <xs:appinfo>

initfiles/componentfiles/configxml/esploggingagent.xsd

@@ -44,6 +44,13 @@
           </xs:appinfo>
         </xs:annotation>
       </xs:attribute>
+      <xs:attribute name="Services" type="xs:string" use="optional" default="GetTransactionSeed,UpdateLog,GetTransactionID">
+        <xs:annotation>
+          <xs:appinfo>
+            <tooltip>the names of logging services to be supported in this logging agent.</tooltip>
+          </xs:appinfo>
+        </xs:annotation>
+      </xs:attribute>
       <xs:attribute name="description" type="xs:string" use="optional" default="ESP Logging Agent">
         <xs:annotation>
           <xs:appinfo>