Merge pull request #13857 from wangkx/h24233

HPCC-24233 Throw exception when WU result is bigger than 10M

Reviewed-By: Jake Smith <jake.smith@lexisnexis.com>
Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

common/fileview2/fileview.hpp

@@ -139,6 +139,7 @@ interface IResultSetFactory : extends IInterface
 {
     virtual INewResultSet * createNewResultSet(IConstWUResult * wuResult, const char * wuid) = 0;
     virtual INewResultSet * createNewFileResultSet(const char * logicalFile, const char * cluster) = 0;
+    virtual INewResultSet * createNewFileResultSet(IDistributedFile * df, const char * cluster) = 0;
     virtual INewResultSet * createNewResultSet(const char * wuid, unsigned sequence, const char * name) = 0;
     virtual INewResultSet * createNewFileResultSet(const char * logicalFile) = 0;
     virtual IResultSetMetaData * createResultSetMeta(IConstWUResult * wuResult) = 0;

common/fileview2/fvresultset.cpp

@@ -2154,7 +2154,12 @@ INewResultSet * CResultSetFactory::createNewResultSet(IConstWUResult * wuResult,
 INewResultSet * CResultSetFactory::createNewFileResultSet(const char * logicalName, const char * cluster)
 {
     Owned<IDistributedFile> df = lookupLogicalName(logicalName);
-    Owned<IFvDataSource> ds = createFileDataSource(df, logicalName, cluster, username, password);
+    return createNewFileResultSet(df, cluster);
+}
+
+INewResultSet * CResultSetFactory::createNewFileResultSet(IDistributedFile * df, const char * cluster)
+{
+    Owned<IFvDataSource> ds = createFileDataSource(df, df->queryLogicalName(), cluster, username, password);
     if (ds)
     {
         Owned<CResultSet> result = createResultSet(ds, false);

common/fileview2/fvresultset.ipp

@@ -492,6 +492,7 @@ public:
 
     virtual INewResultSet * createNewResultSet(IConstWUResult * wuResult, const char * wuid);
     virtual INewResultSet * createNewFileResultSet(const char * logicalFile, const char * cluster);
+    virtual INewResultSet * createNewFileResultSet(IDistributedFile * df, const char * cluster);
     virtual INewResultSet * createNewResultSet(const char * wuid, unsigned sequence, const char * name);
     virtual INewResultSet * createNewFileResultSet(const char * logicalFile);
     virtual IResultSetMetaData * createResultSetMeta(IConstWUResult * wuResult);

esp/services/ws_workunits/ws_workunitsService.cpp

@@ -393,6 +393,11 @@ void CWsWorkunitsEx::init(IPropertyTree *cfg, const char *process, const char *s
     xpath.setf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/ThorSlaveLogThreadPoolSize", process, service);
     thorSlaveLogThreadPoolSize = cfg->getPropInt(xpath, THOR_SLAVE_LOG_THREAD_POOL_SIZE);
 
+    xpath.setf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/WUResultMaxSizeMB", process, service);
+    unsigned wuResultMaxSizeMB = cfg->getPropInt(xpath);
+    if (wuResultMaxSizeMB > 0)
+        wuResultMaxSize = wuResultMaxSizeMB * 1000000;
+
     xpath.setf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/ZAPEmail", process, service);
     IPropertyTree *zapEmail = cfg->queryPropTree(xpath.str());
     if (zapEmail)
@@ -2837,9 +2842,9 @@ INewResultSet* createFilteredResultSet(INewResultSet* result, IArrayOf<IConstNam
     return filter->create();
 }
 
-void getWsWuResult(IEspContext &context, const char* wuid, const char *name, const char *logical, unsigned index, __int64 start,
-    unsigned& count, __int64& total, IStringVal& resname, bool bin, IArrayOf<IConstNamedValue>* filterBy, MemoryBuffer& mb,
-    WUState& wuState, bool xsd=true)
+void CWsWorkunitsEx::getWsWuResult(IEspContext &context, const char *wuid, const char *name, const char *logical, unsigned index, __int64 start,
+    unsigned &count, __int64 &total, IStringVal &resname, bool bin, IArrayOf<IConstNamedValue> *filterBy, MemoryBuffer &mb,
+    WUState &wuState, bool xsd)
 {
     Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
     Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid);
@@ -2868,6 +2873,11 @@ void getWsWuResult(IEspContext &context, const char* wuid, const char *name, con
 
     if (!result)
         throw MakeStringException(ECLWATCH_CANNOT_GET_WU_RESULT,"Cannot open the workunit result.");
+
+    if (result->getResultRawSize(nullptr, nullptr) > wuResultMaxSize)
+        throw makeStringExceptionV(ECLWATCH_INVALID_ACTION, "Failed to get the result for %s. The size is bigger than %lld.",
+            wuid, wuResultMaxSize);
+
     if (!resname.length())
         result->getResultName(resname);
 
@@ -3199,11 +3209,34 @@ void getWorkunitCluster(IEspContext &context, const char *wuid, SCMStringBuffer
     }
 }
 
-void getFileResults(IEspContext &context, const char *logicalName, const char *cluster, __int64 start, unsigned &count, __int64 &total,
+IDistributedFile *CWsWorkunitsEx::lookupLogicalName(IEspContext &context, const char *logicalName)
+{
+    StringBuffer userID;
+    context.getUserID(userID);
+    Owned<IUserDescriptor> userDesc;
+    if (!userID.isEmpty())
+    {
+        userDesc.setown(createUserDescriptor());
+        userDesc->set(userID, context.queryPassword(), context.querySignature());
+    }
+
+    Owned<IDistributedFile> df = queryDistributedFileDirectory().lookup(logicalName, userDesc, false,
+        false, false, nullptr, defaultPrivilegedUser);
+    return df.getClear();
+}
+
+void CWsWorkunitsEx::getFileResults(IEspContext &context, const char *logicalName, const char *cluster, __int64 start, unsigned &count, __int64 &total,
     IStringVal &resname, bool bin, IArrayOf<IConstNamedValue> *filterBy, MemoryBuffer &buf, bool xsd)
 {
+    Owned<IDistributedFile> df = lookupLogicalName(context, logicalName);
+    if (!df)
+        throw makeStringExceptionV(ECLWATCH_FILE_NOT_EXIST, "Cannot find file %s.", logicalName);
+    if (df->getDiskSize(true, false) > wuResultMaxSize)
+        throw makeStringExceptionV(ECLWATCH_INVALID_ACTION, "Failed to get the result from file %s. The size is bigger than %lld.",
+            logicalName, wuResultMaxSize);
+
     Owned<IResultSetFactory> resultSetFactory = getSecResultSetFactory(context.querySecManager(), context.queryUser(), context.queryUserId(), context.queryPassword());
-    Owned<INewResultSet> result(resultSetFactory->createNewFileResultSet(logicalName, cluster));
+    Owned<INewResultSet> result(resultSetFactory->createNewFileResultSet(df, cluster));
     if (!filterBy || !filterBy->length())
         appendResultSet(buf, result, resname.str(), start, count, total, bin, xsd, context.getResponseFormat(), NULL);
     else

esp/services/ws_workunits/ws_workunitsService.hpp

@@ -34,6 +34,8 @@
 #define UFO_RELOAD_MAPPED_QUERIES                0x04
 #define UFO_REMOVE_QUERIES_NOT_IN_QUERYSET       0x08
 
+static const __uint64 defaultWUResultMaxSize = 10000000; //10M
+
 class QueryFilesInUse : public CInterface, implements ISDSSubscription
 {
     mutable CriticalSection crit;
@@ -358,6 +360,12 @@ private:
         IArrayOf<IEspQueryStats> &queryStatsList);
     void readQueryStatsList(IPropertyTree *queryStatsTree, const char *status, const char *ep,
         bool all, IArrayOf<IEspEndpointQueryStats> &endpointQueryStatsList);
+    void getWsWuResult(IEspContext &context, const char *wuid, const char *name, const char *logical, unsigned index, __int64 start,
+        unsigned &count, __int64 &total, IStringVal &resname, bool bin, IArrayOf<IConstNamedValue> *filterBy, MemoryBuffer &mb,
+        WUState &wuState, bool xsd=true);
+    void getFileResults(IEspContext &context, const char *logicalName, const char *cluster, __int64 start, unsigned &count, __int64 &total,
+        IStringVal &resname, bool bin, IArrayOf<IConstNamedValue> *filterBy, MemoryBuffer &buf, bool xsd);
+    IDistributedFile *lookupLogicalName(IEspContext &context, const char *logicalName);
 
     unsigned awusCacheMinutes;
     StringBuffer queryDirectory;
@@ -378,6 +386,7 @@ private:
     unsigned thorSlaveLogThreadPoolSize = THOR_SLAVE_LOG_THREAD_POOL_SIZE;
     Owned<IWorkUnitFactory> wuFactory;
     StringBuffer dataDirectory;
+    __uint64 wuResultMaxSize = defaultWUResultMaxSize;
 
 public:
     QueryFilesInUse filesInUse;

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

@@ -244,6 +244,9 @@ This is required by its binding with ESP service '<xsl:value-of select="$espServ
             <xsl:if test="string(@ThorSlaveLogThreadPoolSize) != ''">
                 <ThorSlaveLogThreadPoolSize><xsl:value-of select="@ThorSlaveLogThreadPoolSize"/></ThorSlaveLogThreadPoolSize>
             </xsl:if>
+            <xsl:if test="string(@WUResultMaxSizeMB) != ''">
+                <WUResultMaxSizeMB><xsl:value-of select="@WUResultMaxSizeMB"/></WUResultMaxSizeMB>
+            </xsl:if>
             <xsl:if test="string(@AWUsCacheTimeout) != ''">
                 <AWUsCacheMinutes><xsl:value-of select="@AWUsCacheTimeout"/></AWUsCacheMinutes>
             </xsl:if>

initfiles/componentfiles/configxml/espsmcservice.xsd.in

@@ -201,6 +201,13 @@
                     </xs:appinfo>
                 </xs:annotation>
             </xs:attribute>
+            <xs:attribute name="WUResultMaxSizeMB" type="xs:nonNegativeInteger" use="optional">
+                <xs:annotation>
+                    <xs:appinfo>
+                        <tooltip>the size limit in megabyte for getting workunit result.</tooltip>
+                    </xs:appinfo>
+                </xs:annotation>
+            </xs:attribute>
             <xs:attribute name="enableLogDaliConnection" type="xs:boolean" use="optional" default="false">
                 <xs:annotation>
                     <xs:appinfo>