HPCC-21150 Report Spark-thor in ECLWatch System Server page

1. Add code to retrieve Spark-thor configuration from environment
settings;
2. Add code to display Spark-thor in ECLWatch System Server page,
including a web link to view spark cluster information, a web link
to thor, and a web lonk to Spark-thor log.

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

common/environment/environment.cpp

@@ -40,6 +40,7 @@
 #define DROPZONE_BY_MACHINE_SUFFIX  "-dropzoneByMachine-"
 #define DROPZONE_SUFFIX             "dropzone-"
 #define MACHINE_PREFIX              "machine-"
+#define SPARKTHOR_SUFFIX            "sparkthor-"
 
 static int environmentTraceLevel = 1;
 static Owned <IConstEnvironment> cache;
@@ -100,6 +101,44 @@ protected:
     unsigned maxIndex = 0;
 };
 
+class CConstSparkThorInfoIterator : public CSimpleInterfaceOf<IConstSparkThorInfoIterator>
+{
+public:
+    CConstSparkThorInfoIterator();
+
+    virtual bool first() override;
+    virtual bool next() override;
+    virtual bool isValid() override;
+    virtual IConstSparkThorInfo & query() override;
+    virtual unsigned count() const override;
+
+protected:
+    Owned<IConstSparkThorInfo> curr;
+    Owned<CLocalEnvironment> constEnv;
+    unsigned index = 1;
+    unsigned maxIndex = 0;
+};
+
+class CConstInstanceInfoIterator : public CSimpleInterfaceOf<IConstInstanceInfoIterator>
+{
+public:
+    CConstInstanceInfoIterator(const CLocalEnvironment * env, IPropertyTreeIterator * itr);
+
+    virtual bool first() override;
+    virtual bool next() override;
+    virtual bool isValid() override;
+    virtual IConstInstanceInfo & query() override;
+    virtual unsigned count() const override;
+
+protected:
+    Owned<IPropertyTreeIterator> instanceItr;
+    Owned<IConstInstanceInfo> curr;
+    const CLocalEnvironment* constEnv;
+    unsigned index = 1;
+    unsigned maxIndex = 0;
+};
+
+
 //==========================================================================================
 
 class CConstInstanceInfo;
@@ -114,6 +153,7 @@ private:
     mutable Mutex safeCache;
     mutable bool dropZoneCacheBuilt;
     mutable bool machineCacheBuilt;
+    mutable bool sparkThorCacheBuilt;
     mutable bool clusterKeyNameCache;
     StringBuffer fileAccessUrl;
 
@@ -126,12 +166,14 @@ private:
     StringBuffer xPath;
     mutable unsigned numOfMachines;
     mutable unsigned numOfDropZones;
+    mutable unsigned numOfSparkThors;
 
 
     IConstEnvBase * getCache(const char *path) const;
     void setCache(const char *path, IConstEnvBase *value) const;
     void buildMachineCache() const;
     void buildDropZoneCache() const;
+    void buildSparkThorCache() const;
     void init();
     mutable bool isDropZoneRestrictionLoaded = false;
     mutable bool dropZoneRestrictionEnabled = true;
@@ -298,6 +340,11 @@ public:
         return fileAccessUrl.length() ? fileAccessUrl.str() : nullptr;
     }
     virtual IConstDaFileSrvInfo *getDaFileSrvGroupInfo(const char *name) const override;
+
+    virtual IConstSparkThorInfo *getSparkThor(const char *name) const;
+    virtual IConstSparkThorInfoIterator *getSparkThorIterator() const;
+    unsigned getNumberOfSparkThors() const { buildSparkThorCache(); return numOfSparkThors; }
+    IConstSparkThorInfo *getSparkThorByIndex(unsigned index) const;
 };
 
 class CLockedEnvironment : implements IEnvironment, public CInterface
@@ -420,6 +467,10 @@ public:
             { return c->getFileAccessUrl(); }
     virtual IConstDaFileSrvInfo *getDaFileSrvGroupInfo(const char *name) const override
             { return c->getDaFileSrvGroupInfo(name); }
+    virtual IConstSparkThorInfo *getSparkThor(const char *name) const
+            { return c->getSparkThor(name); }
+    virtual IConstSparkThorInfoIterator *getSparkThorIterator() const
+            { return c->getSparkThorIterator(); }
 };
 
 void CLockedEnvironment::commit()
@@ -969,6 +1020,11 @@ public:
         str.set(ep.str());
         return str;
     }
+    virtual IStringVal & getDirectory(IStringVal & str) const
+    {
+        str.set(root->queryProp("@directory"));
+        return str;
+    }
 
     virtual bool doGetRunInfo(IStringVal & progpath, IStringVal & workdir, const char *defprogname, bool useprog) const
     {
@@ -1103,6 +1159,57 @@ private:
     StringBuffer posixPath;
 };
 
+class CConstSparkThorInfo : public CConstEnvBase, implements IConstSparkThorInfo
+{
+public:
+    IMPLEMENT_IINTERFACE;
+    IMPLEMENT_ICONSTENVBASE;
+    CConstSparkThorInfo(CLocalEnvironment *env, IPropertyTree *root) : CConstEnvBase(env, root) {}
+
+    virtual IStringVal &getBuild(IStringVal &str) const
+    {
+        str.set(root->queryProp("@build"));
+        return str;
+    }
+    virtual IStringVal &getThorClusterName(IStringVal &str) const
+    {
+        str.set(root->queryProp("@ThorClusterName"));
+        return str;
+    }
+    virtual unsigned getSparkExecutorCores() const
+    {
+        return root->getPropInt("@SPARK_EXECUTOR_CORES", 0);
+    }
+    virtual unsigned long getSparkExecutorMemory() const
+    {
+        return readSizeSetting(root->queryProp("@SPARK_EXECUTOR_MEMORY"), 0);
+    }
+    virtual unsigned getSparkMasterPort() const
+    {
+        return root->getPropInt("@SPARK_MASTER_PORT", 0);
+    }
+    virtual unsigned getSparkMasterWebUIPort() const
+    {
+        return root->getPropInt("@SPARK_MASTER_WEBUI_PORT", 0);
+    }
+    virtual unsigned getSparkWorkerCores() const
+    {
+        return root->getPropInt("@SPARK_WORKER_CORES", 0);
+    }
+    virtual unsigned long getSparkWorkerMemory() const
+    {
+        return readSizeSetting(root->queryProp("@SPARK_WORKER_MEMORY"), 0);
+    }
+    virtual unsigned getSparkWorkerPort() const
+    {
+        return root->getPropInt("@SPARK_WORKER_PORT", 0);
+    }
+    virtual IConstInstanceInfoIterator *getInstanceIterator() const
+    {
+        return new CConstInstanceInfoIterator(env, root->getElements("Instance"));
+    }
+};
+
 #if 0
 //==========================================================================================
 
@@ -1210,8 +1317,10 @@ void CLocalEnvironment::init()
 {
     machineCacheBuilt = false;
     dropZoneCacheBuilt = false;
+    sparkThorCacheBuilt = false;
     numOfMachines = 0;
     numOfDropZones = 0;
+    numOfSparkThors = 0;
     isDropZoneRestrictionLoaded = false;
     clusterKeyNameCache = false;
     ::getFileAccessUrl(fileAccessUrl);
@@ -1845,6 +1954,64 @@ IConstDaFileSrvInfo *CLocalEnvironment::getDaFileSrvGroupInfo(const char *name)
     return (IConstDaFileSrvInfo *) cached;
 }
 
+IConstSparkThorInfo *CLocalEnvironment::getSparkThor(const char *name) const
+{
+    if (isEmptyString(name))
+        return nullptr;
+    buildSparkThorCache();
+    VStringBuffer xpath("Software/SparkThor[@name=\"%s\"]", name);
+    synchronized procedure(safeCache);
+    return (CConstSparkThorInfo *) getCache(xpath);
+}
+
+IConstSparkThorInfo *CLocalEnvironment::getSparkThorByIndex(unsigned index) const
+{
+    if (index == 0)
+        return nullptr;
+
+    buildSparkThorCache();
+    if (index > numOfSparkThors)
+        return nullptr;
+
+    StringBuffer xpath("Software/SparkThor[@id=\"");
+    xpath.append(SPARKTHOR_SUFFIX).append(index).append("\"]");
+    synchronized procedure(safeCache);
+    return (CConstSparkThorInfo *) getCache(xpath);
+}
+
+IConstSparkThorInfoIterator *CLocalEnvironment::getSparkThorIterator() const
+{
+    return new CConstSparkThorInfoIterator();
+}
+
+void CLocalEnvironment::buildSparkThorCache() const
+{
+    synchronized procedure(safeCache);
+    if (sparkThorCacheBuilt)
+        return;
+
+    Owned<IPropertyTreeIterator> it = p->getElements("Software/SparkThorProcess");
+    ForEach(*it)
+    {
+        const char *name = it->query().queryProp("@name");
+        if (!isEmptyString(name))
+        {
+            StringBuffer x("Software/SparkThor[@name=\"");
+            x.append(name).append("\"]");
+            Owned<IConstEnvBase> cached = new CConstSparkThorInfo((CLocalEnvironment *) this, &it->query());
+            cache.setValue(x, cached);
+        }
+
+        numOfSparkThors++;
+        StringBuffer x("Software/SparkThor[@id=\"");
+        x.append(SPARKTHOR_SUFFIX).append(numOfSparkThors).append("\"]");
+        Owned<IConstEnvBase> cached = new CConstSparkThorInfo((CLocalEnvironment *) this, &it->query());
+        cache.setValue(x, cached);
+    }
+    sparkThorCacheBuilt = true;
+}
+
+
 //==========================================================================================
 // Iterators implementation
 
@@ -2012,6 +2179,98 @@ unsigned CConstDropZoneInfoIterator::count() const
     return maxIndex;
 }
 
+//--------------------------------------------------
+
+CConstSparkThorInfoIterator::CConstSparkThorInfoIterator()
+{
+    Owned<IEnvironmentFactory> factory = getEnvironmentFactory(true);
+    constEnv.setown((CLocalEnvironment *)factory->openEnvironment());
+    maxIndex = constEnv->getNumberOfSparkThors();
+}
+
+bool CConstSparkThorInfoIterator::first()
+{
+    index = 1;
+    curr.setown(constEnv->getSparkThorByIndex(index));
+    return curr != nullptr;
+}
+
+bool CConstSparkThorInfoIterator::next()
+{
+    if (index < maxIndex)
+    {
+        index++;
+        curr.setown(constEnv->getSparkThorByIndex(index));
+    }
+    else
+        curr.clear();
+
+    return curr != nullptr;
+}
+
+bool CConstSparkThorInfoIterator::isValid()
+{
+    return curr != nullptr;
+}
+
+IConstSparkThorInfo &CConstSparkThorInfoIterator::query()
+{
+    return *curr;
+}
+
+unsigned CConstSparkThorInfoIterator::count() const
+{
+    return maxIndex;
+}
+
+//--------------------------------------------------
+
+CConstInstanceInfoIterator::CConstInstanceInfoIterator(const CLocalEnvironment *env, IPropertyTreeIterator *itr)
+    : constEnv(env)
+{
+    instanceItr.setown(itr);
+    maxIndex = 0;
+    ForEach(*instanceItr)
+        maxIndex++;
+}
+
+bool CConstInstanceInfoIterator::first()
+{
+    index = 1;
+    instanceItr->first();
+    curr.setown(new CConstInstanceInfo(constEnv, &instanceItr->query()));
+    return curr != nullptr;
+}
+
+bool CConstInstanceInfoIterator::next()
+{
+    if (index < maxIndex)
+    {
+        index++;
+        instanceItr->next();
+        curr.setown(new CConstInstanceInfo(constEnv, &instanceItr->query()));
+    }
+    else
+        curr.clear();
+
+    return curr != nullptr;
+}
+
+bool CConstInstanceInfoIterator::isValid()
+{
+    return curr != nullptr;
+}
+
+IConstInstanceInfo &CConstInstanceInfoIterator::query()
+{
+    return *curr;
+}
+
+unsigned CConstInstanceInfoIterator::count() const
+{
+    return maxIndex;
+}
+
 //==========================================================================================
 
 
@@ -2056,3 +2315,50 @@ extern ENVIRONMENT_API void closeEnvironment()
         EXCLOG(e);
     }
 }
+
+extern ENVIRONMENT_API unsigned long readSizeSetting(const char * sizeStr, const unsigned long defaultSize)
+{
+    StringBuffer buf = sizeStr;
+    buf.trim();
+
+    if (buf.isEmpty())
+        return defaultSize;
+
+    const char* ptrStart = buf;
+    const char* ptrAfterDigit = ptrStart;
+    while (*ptrAfterDigit && isdigit(*ptrAfterDigit))
+        ptrAfterDigit++;
+
+    if (!*ptrAfterDigit)
+        return atol(buf);
+
+    const char* ptr = ptrAfterDigit;
+    while (*ptr && (ptr[0] == ' '))
+        ptr++;
+
+    char c = ptr[0];
+    buf.setLength(ptrAfterDigit - ptrStart);
+    unsigned long size = atol(buf);
+    switch (c)
+    {
+    case 'k':
+    case 'K':
+        size *= 1000;
+        break;
+    case 'm':
+    case 'M':
+        size *= 1000000;
+        break;
+    case 'g':
+    case 'G':
+        size *= 1000000000;
+        break;
+    case 't':
+    case 'T':
+        size *= 1000000000000;
+        break;
+    default:
+        break;
+    }
+    return size;
+}

common/environment/environment.hpp

@@ -97,6 +97,7 @@ interface IConstInstanceInfo : extends IConstEnvBase
     virtual IConstMachineInfo * getMachine() const = 0;
     virtual IStringVal & getEndPoint(IStringVal & str) const = 0;
     virtual unsigned getPort() const = 0;
+    virtual IStringVal & getDirectory(IStringVal & str) const = 0;
     virtual IStringVal & getExecutableDirectory(IStringVal & str) const = 0;
     virtual bool getRunInfo(IStringVal & progpath, IStringVal & workdir, const char * defaultprogname) const = 0;
 };
@@ -134,6 +135,31 @@ interface IConstDaFileSrvInfo : extends IConstEnvBase
     virtual bool getSecure() const = 0;
 };
 
+interface IConstInstanceInfoIterator : extends IIteratorOf<IConstInstanceInfo>
+{
+    virtual unsigned count() const = 0;
+};
+
+interface IConstSparkThorInfo : extends IConstEnvBase
+{
+    virtual IStringVal & getName(IStringVal & str) const = 0;
+    virtual IStringVal & getBuild(IStringVal & str) const = 0;
+    virtual IStringVal & getThorClusterName(IStringVal & str) const = 0;
+    virtual unsigned getSparkExecutorCores() const = 0;
+    virtual unsigned long getSparkExecutorMemory() const = 0;
+    virtual unsigned getSparkMasterPort() const = 0;
+    virtual unsigned getSparkMasterWebUIPort() const = 0;
+    virtual unsigned getSparkWorkerCores() const = 0;
+    virtual unsigned long getSparkWorkerMemory() const = 0;
+    virtual unsigned getSparkWorkerPort() const = 0;
+    virtual IConstInstanceInfoIterator * getInstanceIterator() const = 0;
+};
+
+interface IConstSparkThorInfoIterator : extends IIteratorOf<IConstSparkThorInfo>
+{
+    virtual unsigned count() const = 0;
+};
+
 interface IConstEnvironment : extends IConstEnvBase
 {
     virtual IConstDomainInfo * getDomain(const char * name) const = 0;
@@ -159,6 +185,8 @@ interface IConstEnvironment : extends IConstEnvBase
     virtual const char *getPrivateKeyPath(const char *keyPairName) const = 0;
     virtual const char *getFileAccessUrl() const = 0;
     virtual IConstDaFileSrvInfo *getDaFileSrvGroupInfo(const char *name) const = 0;
+    virtual IConstSparkThorInfo *getSparkThor(const char *name) const = 0;
+    virtual IConstSparkThorInfoIterator *getSparkThorIterator() const = 0;
 };
 
 
@@ -186,7 +214,7 @@ class StringBuffer;
 extern "C" ENVIRONMENT_API IEnvironmentFactory * getEnvironmentFactory(bool update);
 extern "C" ENVIRONMENT_API void closeEnvironment();
 
-
+extern ENVIRONMENT_API unsigned long readSizeSetting(const char * sizeStr, const unsigned long defaultSize);
 
 
 #endif // _ENVIRONMENT_INCL

esp/eclwatch/ws_XSLT/nls/en/hpcc.xml

@@ -127,6 +127,7 @@
 <st id='ShowProcessesUsingFilter'>Show processes using filter</st>
 <st id='Size'>Size</st>
 <st id='SlaveNumber'>Slave Number</st>
+<st id='SparkThor'>SparkThor</st>
 <st id='Starting'>Starting</st>
 <st id='State'>State</st>
 <st id='Stopping'>Stopping</st>
@@ -156,6 +157,7 @@
 <st id='ViewingPage'>Viewing page</st>
 <st id='ViewLogFile'>View Log File</st>
 <st id='ViewAuditLogFile'>View Audit Log File</st>
+<st id='ViewSparkClusterInfo'>View Integrated Spark Cluster Information</st>
 <st id='WarnIfAvailableDiskSpaceIsUnder'>Warn if available disk space is under</st>
 <st id='WarnIfAvailableMemoryIsUnder'>Warn if available memory is under</st>
 <st id='WarnIfCPUUsageIsOver'>Warn if CPU usage is over</st>

esp/eclwatch/ws_XSLT/services.xslt

@@ -234,6 +234,18 @@
                 var url = protocol + "://" + window.location.hostname + ":" + port;
                 parent.document.location.href = url;
             }
+            function showPopup(protocol, ip, port)
+            {
+                if (!protocol || protocol.length === 0)
+                    protocol = "http";
+                var url = protocol + "://" + ip + ":" + port;
+                popupWin = window.open (url,
+                    "popupWin", "location=0,status=1,scrollbars=1,resizable=1,width=500,height=600");
+                if (popupWin.opener == null)
+                    popupWin.opener = window;
+                popupWin.focus();
+                return false;
+            }
             allowReloadPage = false;
             ]]>
         </xsl:text>
@@ -319,11 +331,17 @@
                 <xsl:with-param name="nodes" select="TpEspServers/TpEspServer"/>
                 <xsl:with-param name="compType" select="'EspProcess'"/>
             </xsl:call-template>
-            
+
             <xsl:call-template name="showMachines">
                 <xsl:with-param name="caption" select="$hpccStrings/st[@id='FTSlaves']"/>
                 <xsl:with-param name="nodes" select="TpFTSlaves/TpFTSlave"/>
                 <xsl:with-param name="checked" select="0"/>
+            </xsl:call-template>
+            
+            <xsl:call-template name="showMachines">
+                <xsl:with-param name="caption" select="$hpccStrings/st[@id='SparkThor']"/>
+                <xsl:with-param name="nodes" select="TpSparkThors/TpSparkThor"/>
+                <xsl:with-param name="compType" select="'SparkThorProcess'"/>
             </xsl:call-template>                    
             
             <xsl:call-template name="showMachines">
@@ -386,7 +404,7 @@
     </form>
     </body>
 </xsl:template>
-        
+
 <xsl:template name="showMachines">
     <xsl:param name="caption"/>
     <xsl:param name="showCheckbox" select="1"/>
@@ -426,6 +444,12 @@
             <xsl:variable name="nodeCount" select="position()" />
             <xsl:for-each select="TpMachines/*">
                 <xsl:variable name="machineCount" select="position()" />
+                <xsl:variable name="absolutePath">
+                    <xsl:call-template name="makeAbsolutePath">
+                        <xsl:with-param name="path" select="Directory"/>
+                        <xsl:with-param name="isLinuxInstance" select="OS!=0"/>
+                    </xsl:call-template>
+                </xsl:variable>
                 <tr>
                     <xsl:variable name="compName" select="../../Name"/>
                     <xsl:if test="$showBindings">
@@ -484,6 +508,30 @@
                             </xsl:if>
                           </a>
                         </xsl:when>
+                        <xsl:when test="$compType!='' and $compType='SparkThorProcess'">
+                            <table class="C">
+                                <tbody>
+                                    <tr>
+                                        <td><xsl:value-of select="../../Name"/></td>
+                                        <td>
+                                            (<a>
+                                                <xsl:attribute name="href">
+                                                    <xsl:text>/WsTopology/TpMachineQuery?Type=THORMACHINES&amp;Cluster=</xsl:text>
+                                                    <xsl:value-of select="../../ThorClusterName"/>
+                                                    <xsl:text>&amp;Directory=</xsl:text>
+                                                    <xsl:value-of select="$absolutePath"/>
+                                                    <xsl:text>&amp;LogDirectory=</xsl:text>
+                                                    <xsl:value-of select="../../LogDirectory"/>
+                                                    <xsl:text>&amp;Path=</xsl:text>
+                                                    <xsl:value-of select="../../ThorPath"/>
+                                                </xsl:attribute>
+                                                <xsl:value-of select="../../ThorClusterName"/>
+                                            </a>)
+                                        </td>
+                                    </tr>
+                                </tbody>
+                            </table>
+                        </xsl:when>
                         <xsl:otherwise>
                                                     <xsl:value-of select="../../Name"/>
                                                     <xsl:if test="count(../TpMachine) &gt; 1"> (Instance <xsl:value-of select="position()"/>)</xsl:if>
@@ -507,12 +555,6 @@
                     </td>
                     <td>
                         <xsl:if test="$showDirectory">
-                            <xsl:variable name="absolutePath">
-                                <xsl:call-template name="makeAbsolutePath">
-                                    <xsl:with-param name="path" select="Directory"/>
-                                    <xsl:with-param name="isLinuxInstance" select="OS!=0"/>
-                                </xsl:call-template>
-                            </xsl:variable>
                             <table width="100%" cellpadding="0" cellspacing="0" class="C">
                                 <tbody>
                                     <tr>
@@ -638,6 +680,9 @@
                                         </xsl:if>
                                         <td width="14">
                                             <xsl:choose>
+                                                <xsl:when test="$compType!='' and $compType='SparkThorProcess'">
+                                                    <img onclick="return showPopup('http', '{Netaddress}', '{../../SparkMasterWebUIPort}')" border="0" src="/esp/files_/img/information.png" alt="{$hpccStrings/st[@id='ViewSparkClusterInfo']}" title="{$hpccStrings/st[@id='ViewSparkClusterInfo']}" width="14" height="14"/>
+                                                </xsl:when>
                                                 <xsl:when test="$compType!='' and ($compType!='EclAgentProcess' or OS=0)">
                           <xsl:variable name="captionLen" select="string-length($caption)-1"/>
                           <xsl:variable name="href0">
@@ -726,12 +771,6 @@
                     </tr>
                 </xsl:if>
         <xsl:if test="$showAgentExec">
-          <xsl:variable name="absolutePath">
-            <xsl:call-template name="makeAbsolutePath">
-              <xsl:with-param name="path" select="string(Directory)"/>
-              <xsl:with-param name="isLinuxInstance" select="OS!=0"/>
-            </xsl:call-template>
-          </xsl:variable>
           <xsl:variable name="logDir" select="string(../../LogDir)"/>
           <xsl:variable name="logPath">
             <xsl:choose>

esp/scm/ws_topology.ecm

@@ -18,6 +18,7 @@
 ////////////////////////////////////////////////////////////
 
 //  ===========================================================================
+
 ESPStruct TpMachine
 {
     string Name;
@@ -32,6 +33,7 @@ ESPStruct TpMachine
     int    Port;
     [min_ver("1.18")] int    ProcessNumber;
 };
+
 //  ===========================================================================
 ESPStruct TpCluster
 {
@@ -252,6 +254,25 @@ ESPStruct TpGenesisServer
 };
 
 //  ===========================================================================
+ESPStruct TpSparkThor
+{
+    string Name;
+    string Build;
+    string ThorClusterName;
+    string ThorPath;
+    unsigned SparkExecutorCores;
+    int64 SparkExecutorMemory;
+    unsigned SparkMasterPort;
+    unsigned SparkMasterWebUIPort;
+    unsigned SparkWorkerCores;
+    int64 SparkWorkerMemory;
+    unsigned SparkWorkerPort;
+    string LogDirectory;
+    string Path;
+    ESParray<ESPstruct TpMachine> TpMachines;
+};
+
+//  ===========================================================================
 ESPStruct TpQueue
 {
     string Name;
@@ -276,6 +297,7 @@ ESPStruct TpServices
     ESParray<ESPstruct TpLdapServer>  TpLdapServers;
     ESParray<ESPstruct TpMySqlServer> TpMySqlServers;
     ESParray<ESPstruct TpSashaServer> TpSashaServers;
+    [min_ver("1.28")] ESParray<ESPstruct TpSparkThor>   TpSparkThors;
 };
 
 ESPStruct TpClusterNameType
@@ -612,7 +634,7 @@ ESPresponse [nil_remove, exceptions_inline] TpDropZoneQueryResponse
     ESParray<ESPstruct TpDropZone>    TpDropZones;
 };
 
-ESPservice [auth_feature("DEFERRED"), noforms, version("1.27"), cache_group("ESPWsTP"), exceptions_inline("./smc_xslt/exceptions.xslt")] WsTopology
+ESPservice [auth_feature("DEFERRED"), noforms, version("1.28"), cache_group("ESPWsTP"), exceptions_inline("./smc_xslt/exceptions.xslt")] WsTopology
 {
     ESPmethod [cache_seconds(180), cache_global(1), resp_xsl_default("/esp/xslt/targetclusters.xslt")] TpTargetClusterQuery(TpTargetClusterQueryRequest, TpTargetClusterQueryResponse);
     ESPmethod [cache_seconds(180), cache_global(1), resp_xsl_default("/esp/xslt/topology.xslt")] TpClusterQuery(TpClusterQueryRequest, TpClusterQueryResponse);

esp/services/ws_topology/ws_topologyService.cpp

@@ -1264,6 +1264,10 @@ bool CWsTopologyEx::onTpServiceQuery(IEspContext &context, IEspTpServiceQueryReq
             {       
                 m_TpWrapper.getTpEclSchedulers( ServiceList.getTpEclSchedulers() );
             }
+            if (version >= 1.28)
+            {       
+                m_TpWrapper.getTpSparkThors(version, nullptr, ServiceList.getTpSparkThors() );
+            }
         }
 
         resp.setMemThreshold( m_memThreshold );

esp/smc/SMCLib/TpWrapper.cpp

@@ -1737,6 +1737,84 @@ void CTpWrapper::appendTpDropZone(double clientVersion, IConstEnvironment* const
     list.append(*dropZone.getLink());
 }
 
+void CTpWrapper::getTpSparkThors(double clientVersion, const char* name, IArrayOf<IConstTpSparkThor>& list)
+{
+    Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory(true);
+    Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
+    if (!isEmptyString(name))
+    {
+        Owned<IConstSparkThorInfo> sparkThorInfo = constEnv->getSparkThor(name);
+        if (sparkThorInfo)
+            appendTpSparkThor(clientVersion, constEnv, *sparkThorInfo, list);
+    }
+    else
+    {
+        Owned<IConstSparkThorInfoIterator> it = constEnv->getSparkThorIterator();
+        ForEach(*it)
+            appendTpSparkThor(clientVersion, constEnv, it->query(), list);
+    }
+}
+
+void CTpWrapper::appendTpSparkThor(double clientVersion, IConstEnvironment* constEnv, IConstSparkThorInfo& sparkThorInfo, IArrayOf<IConstTpSparkThor>& list)
+{
+    SCMStringBuffer name, build, thorClusterName;
+    sparkThorInfo.getName(name);
+    sparkThorInfo.getBuild(build);
+    sparkThorInfo.getThorClusterName(thorClusterName);
+
+    Owned<IEspTpSparkThor> sparkThor = createTpSparkThor();
+    if (name.length() > 0)
+        sparkThor->setName(name.str());
+    if (build.length() > 0)
+        sparkThor->setBuild(build.str());
+    if (thorClusterName.length() > 0)
+        sparkThor->setThorClusterName(thorClusterName.str());
+    sparkThor->setSparkExecutorCores(sparkThorInfo.getSparkExecutorCores());
+    sparkThor->setSparkExecutorMemory(sparkThorInfo.getSparkExecutorMemory());
+    sparkThor->setSparkMasterPort(sparkThorInfo.getSparkMasterPort());
+    sparkThor->setSparkMasterWebUIPort(sparkThorInfo.getSparkMasterWebUIPort());
+    sparkThor->setSparkWorkerCores(sparkThorInfo.getSparkWorkerCores());
+    sparkThor->setSparkWorkerMemory(sparkThorInfo.getSparkWorkerMemory());
+    sparkThor->setSparkWorkerPort(sparkThorInfo.getSparkWorkerPort());
+
+    //Create the Path used by the thor cluster.
+    StringBuffer tmpPath;
+    StringBuffer path("/Environment/Software");
+    setAttPath(path, eqThorCluster, "name", thorClusterName.str(), tmpPath);
+    sparkThor->setThorPath(tmpPath.str());
+
+    StringBuffer dirBuf;
+    Owned<IPropertyTree> root = &constEnv->getPTree();
+    if (getConfigurationDirectory(root->queryPropTree("Directories"), "log", "sparkthor", name.str(), dirBuf))
+        sparkThor->setLogDirectory(dirBuf.str());
+
+    IArrayOf<IConstTpMachine> machines;
+    Owned<IConstInstanceInfoIterator> instanceInfoItr = sparkThorInfo.getInstanceIterator();
+    ForEach(*instanceInfoItr)
+        appendTpMachine(clientVersion, constEnv, instanceInfoItr->query(), machines);
+    sparkThor->setTpMachines(machines);
+
+    list.append(*sparkThor.getLink());
+}
+
+void CTpWrapper::appendTpMachine(double clientVersion, IConstEnvironment* constEnv, IConstInstanceInfo& instanceInfo, IArrayOf<IConstTpMachine>& machines)
+{
+    SCMStringBuffer name, networkAddress, description, directory;
+    Owned<IConstMachineInfo> machineInfo = instanceInfo.getMachine();
+    machineInfo->getName(name);
+    machineInfo->getNetAddress(networkAddress);
+    instanceInfo.getDirectory(directory);
+
+    Owned<IEspTpMachine> machine = createTpMachine();
+    machine->setName(name.str());
+    machine->setNetaddress(networkAddress.str());
+    machine->setPort(instanceInfo.getPort());
+    machine->setOS(machineInfo->getOS());
+    machine->setDirectory(directory.str());
+    machine->setType(eqSparkThorProcess);
+    machines.append(*machine.getLink());
+}
+
 IEspTpMachine* CTpWrapper::createTpMachineEx(const char* name, const char* type, IConstMachineInfo* machineInfo)
 {
     if (!machineInfo)

esp/smc/SMCLib/TpWrapper.hpp

@@ -120,6 +120,7 @@ using std::string;
 #define eqHoleControlProcess    "HoleControlProcess"
 #define eqHoleCollatorProcess   "HoleCollatorProcess"
 #define eqHoleStandbyProcess    "HoleStandbyProcess"
+#define eqSparkThorProcess      "SparkThorProcess"
 
 #define SDS_LOCK_TIMEOUT 30000
 
@@ -135,6 +136,8 @@ private:
     void fetchInstances(const char* ServiceType, IPropertyTree& service, IArrayOf<IEspTpMachine>& tpMachines);
     bool checkGroupReplicateOutputs(const char* groupName, const char* kind);
     void appendTpDropZone(double clientVersion, IConstEnvironment* constEnv, IConstDropZoneInfo& dropZoneInfo, IArrayOf<IConstTpDropZone>& list);
+    void appendTpSparkThor(double clientVersion, IConstEnvironment* constEnv, IConstSparkThorInfo& sparkThorInfo, IArrayOf<IConstTpSparkThor>& list);
+    void appendTpMachine(double clientVersion, IConstEnvironment* constEnv, IConstInstanceInfo& instanceInfo, IArrayOf<IConstTpMachine>& machines);
 
 public:
     IMPLEMENT_IINTERFACE;
@@ -181,6 +184,7 @@ public:
     void getTpFTSlaves(IArrayOf<IConstTpFTSlave>& list);
     void getTpDkcSlaves(IArrayOf<IConstTpDkcSlave>& list);
     void getTpGenesisServers(IArrayOf<IConstTpGenesisServer>& list);
+    void getTpSparkThors(double clientVersion, const char* name, IArrayOf<IConstTpSparkThor>& list);
 
     void queryTargetClusters(double version, const char* clusterType, const char* clusterName, IArrayOf<IEspTpTargetCluster>& clusterList);
     void getTargetClusterList(IArrayOf<IEspTpLogicalCluster>& clusters, const char* clusterType = NULL, const char* clusterName = NULL);

esp/src/eclwatch/nls/hpcc.js

@@ -650,6 +650,7 @@ define({root:
     SourceLogicalFile: "Source Logical Name",
     SourcePath: "Source Path (Wildcard Enabled)",
     SourceProcess: "Source Process",
+    SparkThor: "SparkThor",
     Spill: "Spill",
     SplitPrefix: "Split Prefix",
     Spray: "Spray",
@@ -827,6 +828,7 @@ define({root:
     Version: "Version",
     ViewByScope: "View By Scope",
     Views: "Views",
+    ViewSparkClusterInfo: "View Spark Cluster Information",
     Visualize: "Visualize",
     Warning: "Warning",
     Warnings: "Warning(s)",
@@ -880,4 +882,4 @@ define({root:
 "pt-br": true,
 "sr": true,
 "zh": true
-});
+});