Merge pull request #3365 from afishbeck/query_settings3312

HPCC-3312 Support setting per-query memory and time limits from CLI

Reviewed-By: Richard Chapman <rchapman@hpccsystems.com>

common/workunit/workunit.cpp

@@ -8948,27 +8948,49 @@ void setQueryAlias(IPropertyTree * queryRegistry, const char * name, const char
     match->setProp("@id", value);
 }
 
-IPropertyTree * resolveQueryAlias(IPropertyTree * queryRegistry, const char * alias)
+extern WORKUNIT_API IPropertyTree * getQueryById(IPropertyTree * queryRegistry, const char *queryid)
 {
+    if (!queryRegistry || !queryid)
+        return NULL;
+    StringBuffer xpath;
+    xpath.append("Query[@id=\"").append(queryid).append("\"]");
+    return queryRegistry->getPropTree(xpath);
+}
+
+extern WORKUNIT_API IPropertyTree * getQueryById(const char *queryset, const char *queryid, bool readonly)
+{
+    Owned<IPropertyTree> queryRegistry = getQueryRegistry(queryset, readonly);
+    return getQueryById(queryRegistry, queryid);
+}
+
+extern WORKUNIT_API IPropertyTree * resolveQueryAlias(IPropertyTree * queryRegistry, const char * alias)
+{
+    if (!queryRegistry || !alias)
+        return NULL;
+
     StringBuffer xpath;
     unsigned cnt = 0;
-    StringBuffer lcAlias(alias);
-    lcAlias.toLowerCase();
-    const char * search = lcAlias.str();
+    StringBuffer lc(alias);
+    const char * search = lc.toLowerCase().str();
     loop
     {
-        xpath.clear().append("Alias[@name=\"").append(search).append("\"]/@id");
+        xpath.set("Alias[@name='").append(search).append("']/@id");
         const char * queryId = queryRegistry->queryProp(xpath);
         if (!queryId)
             break;
         //Check for too many alias indirections.
         if (cnt++ > 10)
             return NULL;
-        search = lcAlias.clear().append(queryId).toLowerCase().str();
+        search = lc.set(queryId).toLowerCase().str();
     }
 
-    xpath.clear().append("Query[@id=\"").append(search).append("\"]");
-    return queryRegistry->getPropTree(xpath);
+    return getQueryById(queryRegistry, search);
+}
+
+extern WORKUNIT_API IPropertyTree * resolveQueryAlias(const char *queryset, const char *alias, bool readonly)
+{
+    Owned<IPropertyTree> queryRegistry = getQueryRegistry(queryset, readonly);
+    return resolveQueryAlias(queryRegistry, alias);
 }
 
 void setQuerySuspendedState(IPropertyTree * queryRegistry, const char *id, bool suspend)
@@ -9273,4 +9295,4 @@ extern WORKUNIT_API void descheduleWorkunit(char const * wuid)
         workunit->deschedule();
     else
         doDescheduleWorkkunit(wuid);
-}
+}

common/workunit/workunit.hpp

@@ -1213,9 +1213,14 @@ extern WORKUNIT_API void removeWuidFromNamedQueries(IPropertyTree * queryRegistr
 extern WORKUNIT_API void removeDllFromNamedQueries(IPropertyTree * queryRegistry, const char * dll);
 extern WORKUNIT_API void removeAliasesFromNamedQuery(IPropertyTree * queryRegistry, const char * id);
 extern WORKUNIT_API void setQueryAlias(IPropertyTree * queryRegistry, const char * name, const char * value);
+
+extern WORKUNIT_API IPropertyTree * getQueryById(IPropertyTree * queryRegistry, const char *queryid);
+extern WORKUNIT_API IPropertyTree * getQueryById(const char *queryset, const char *queryid, bool readonly);
 extern WORKUNIT_API IPropertyTree * resolveQueryAlias(IPropertyTree * queryRegistry, const char * alias);
+extern WORKUNIT_API IPropertyTree * resolveQueryAlias(const char *queryset, const char *alias, bool readonly);
 extern WORKUNIT_API IPropertyTree * getQueryRegistry(const char * wsEclId, bool readonly);
 extern WORKUNIT_API IPropertyTree * getQueryRegistryRoot();
+
 extern WORKUNIT_API void setQueryCommentForNamedQuery(IPropertyTree * queryRegistry, const char *id, const char *queryComment);
 
 extern WORKUNIT_API void setQuerySuspendedState(IPropertyTree * queryRegistry, const char * name, bool suspend);

ecl/eclcmd/eclcmd_common.cpp

@@ -104,6 +104,34 @@ static bool looksLikeOnlyAWuid(const char * wuid)
     return true;
 }
 
+bool isValidMemoryValue(const char *value)
+{
+    if (!value || !*value || !isdigit(*value))
+        return false;
+    while (isdigit(*++value));
+
+    if (!*value)
+        return true;
+
+    switch (toupper(*value++))
+    {
+        case 'E':
+        case 'P':
+        case 'T':
+        case 'G':
+        case 'M':
+        case 'K':
+            if (!*value || strieq("B", value))
+                return true;
+            break;
+        case 'B':
+            if (!*value)
+                return true;
+            break;
+    }
+    return false;
+}
+
 //=========================================================================================
 
 #define PE_OFFSET_LOCATION_IN_DOS_SECTION 0x3C

ecl/eclcmd/eclcmd_common.hpp

@@ -75,6 +75,10 @@ typedef IEclCommand *(*EclCommandFactory)(const char *cmdname);
 #define ECLOPT_WAIT_INI "waitTimeout"
 #define ECLOPT_WAIT_ENV "ECL_WAIT_TIMEOUT"
 
+#define ECLOPT_TIME_LIMIT "--timeLimit"
+#define ECLOPT_MEMORY_LIMIT "--memoryLimit"
+#define ECLOPT_WARN_TIME_LIMIT "--warnTimeLimit"
+
 #define ECLOPT_RESULT_LIMIT "--limit"
 #define ECLOPT_RESULT_LIMIT_INI "resultLimit"
 #define ECLOPT_RESULT_LIMIT_ENV "ECL_RESULT_LIMIT"
@@ -112,6 +116,8 @@ typedef IEclCommand *(*EclCommandFactory)(const char *cmdname);
 #define ECLOPT_VERBOSE "--verbose"
 #define ECLOPT_VERBOSE_S "-v"
 
+bool isValidMemoryValue(const char *value);
+
 bool extractEclCmdOption(StringBuffer & option, IProperties * globals, const char * envName, const char * propertyName, const char * defaultPrefix, const char * defaultSuffix);
 bool extractEclCmdOption(StringAttr & option, IProperties * globals, const char * envName, const char * propertyName, const char * defaultPrefix, const char * defaultSuffix);
 bool extractEclCmdOption(bool & option, IProperties * globals, const char * envName, const char * propertyName, bool defval);

ecl/eclcmd/eclcmd_core.cpp

@@ -328,6 +328,8 @@ public:
     EclCmdPublish() : optNoActivate(false), activateSet(false), optNoReload(false), optMsToWait(10000)
     {
         optObj.accept = eclObjWuid | eclObjArchive | eclObjSharedObject;
+        optTimeLimit = (unsigned) -1;
+        optWarnTimeLimit = (unsigned) -1;
     }
     virtual bool parseCommandLineOptions(ArgvIterator &iter)
     {
@@ -349,6 +351,12 @@ public:
                 continue;
             if (iter.matchOption(optMsToWait, ECLOPT_WAIT))
                 continue;
+            if (iter.matchOption(optTimeLimit, ECLOPT_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optWarnTimeLimit, ECLOPT_WARN_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optMemoryLimit, ECLOPT_MEMORY_LIMIT))
+                continue;
             if (iter.matchFlag(optNoActivate, ECLOPT_NO_ACTIVATE))
             {
                 activateSet=true;
@@ -396,6 +404,11 @@ public:
                 return false;
             }
         }
+        if (optMemoryLimit.length() && !isValidMemoryValue(optMemoryLimit))
+        {
+            fprintf(stderr, "invalid --memoryLimit value of %s.\n\n", optMemoryLimit.get());
+            return false;
+        }
         return true;
     }
     virtual int processCMD()
@@ -426,6 +439,13 @@ public:
         req->setWait(optMsToWait);
         req->setNoReload(optNoReload);
 
+        if (optTimeLimit != (unsigned) -1)
+            req->setTimeLimit(optTimeLimit);
+        if (optWarnTimeLimit != (unsigned) -1)
+            req->setWarnTimeLimit(optWarnTimeLimit);
+        if (!optMemoryLimit.isEmpty())
+            req->setMemoryLimit(optMemoryLimit);
+
         Owned<IClientWUPublishWorkunitResponse> resp = client->WUPublishWorkunit(req);
         const char *id = resp->getQueryId();
         if (id && *id)
@@ -467,6 +487,10 @@ public:
             "   -A, --activate         Activate query when published (default)\n"
             "   -A-, --no-activate     Do not activate query when published\n"
             "   --no-reload            Do not request a reload of the (roxie) cluster\n"
+            "   --timeLimit=<ms>       Value to set for query timeLimit configuration\n"
+            "   --warnTimeLimit=<ms>   Value to set for query warnTimeLimit configuration\n"
+            "   --memoryLimit=<mem>    Value to set for query memoryLimit configuration\n"
+            "                          format <mem> as 500000B, 550K, 100M, 10G, 1T etc.\n"
             "   --wait=<ms>            Max time to wait in milliseconds\n",
             stdout);
         EclCmdWithEclTarget::usage();
@@ -474,7 +498,10 @@ public:
 private:
     StringAttr optTargetCluster;
     StringAttr optName;
+    StringAttr optMemoryLimit;
     unsigned optMsToWait;
+    unsigned optTimeLimit;
+    unsigned optWarnTimeLimit;
     bool optNoActivate;
     bool activateSet;
     bool optNoReload;

ecl/eclcmd/queries/ecl-queries.cpp

@@ -180,20 +180,23 @@ public:
         line.append(query.getSuspended() ? 'S' : ' ');
         line.append(isActive ? 'A' : ' ');
         line.append(' ').append(queryid);
-        if (isActive)
+        if (!query.getTimeLimit_isNull())
         {
-            Owned<IPropertyTreeIterator> activeNames = queryMap.getActiveNames(queryid);
-            if (line.length() < 35)
-                line.appendN(35 - line.length(), ' ');
-            line.append("[");
-            activeNames->first();
-            while (activeNames->isValid())
-            {
-                line.append(activeNames->query().queryProp(NULL));
-                if (activeNames->next())
-                    line.append(',');
-            }
-            line.append("]");
+            if (line.length() < 34)
+                line.appendN(34 - line.length(), ' ');
+            line.append(' ').append(query.getTimeLimit());
+        }
+        if (!query.getWarnTimeLimit_isNull())
+        {
+            if (line.length() < 41)
+                line.appendN(41 - line.length(), ' ');
+            line.append(' ').append(query.getWarnTimeLimit());
+        }
+        if (query.getMemoryLimit())
+        {
+            if (line.length() < 48)
+                line.appendN(48 - line.length(), ' ');
+            line.append(' ').append(query.getMemoryLimit());
         }
         fputs(line.append('\n').str(), stdout);
     }
@@ -203,8 +206,10 @@ public:
         ActiveQueryMap queryMap(qs);
         if (qs.getQuerySetName())
             fprintf(stdout, "\nQuerySet: %s\n", qs.getQuerySetName());
-        fputs("\nFlags Query Id                     [Active Name(s)]\n", stdout);
-        fputs("----- ---------------------------- ----------------\n", stdout);
+        fputs("\n", stdout);
+        fputs("                                   Time   Warn   Memory\n", stdout);
+        fputs("Flags Query Id                     Limit  Limit  Limit\n", stdout);
+        fputs("----- ---------------------------- ------ ------ ----------\n", stdout);
 
         IArrayOf<IConstQuerySetQuery> &queries = qs.getQueries();
         ForEachItemIn(id, queries)
@@ -269,6 +274,8 @@ class EclCmdQueriesCopy : public EclCmdCommon
 public:
     EclCmdQueriesCopy() : optActivate(false), optNoReload(false), optMsToWait(10000), optDontCopyFiles(false), optOverwrite(false)
     {
+        optTimeLimit = (unsigned) -1;
+        optWarnTimeLimit = (unsigned) -1;
     }
     virtual bool parseCommandLineOptions(ArgvIterator &iter)
     {
@@ -308,6 +315,12 @@ public:
                 continue;
             if (iter.matchOption(optMsToWait, ECLOPT_WAIT))
                 continue;
+            if (iter.matchOption(optTimeLimit, ECLOPT_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optWarnTimeLimit, ECLOPT_WARN_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optMemoryLimit, ECLOPT_MEMORY_LIMIT))
+                continue;
             if (EclCmdCommon::matchCommandLineOption(iter, true)!=EclCmdOptionMatch)
                 return false;
         }
@@ -327,6 +340,12 @@ public:
             fputs("cluster must be specified for remote copies.\n\n", stderr);
             return false;
         }
+        if (optMemoryLimit.length() && !isValidMemoryValue(optMemoryLimit))
+        {
+            fprintf(stderr, "invalid --memoryLimit value of %s.\n\n", optMemoryLimit.get());
+            return false;
+        }
+
         return true;
     }
 
@@ -349,6 +368,13 @@ public:
         req->setWait(optMsToWait);
         req->setNoReload(optNoReload);
 
+        if (optTimeLimit != (unsigned) -1)
+            req->setTimeLimit(optTimeLimit);
+        if (optWarnTimeLimit != (unsigned) -1)
+            req->setWarnTimeLimit(optWarnTimeLimit);
+        if (!optMemoryLimit.isEmpty())
+            req->setMemoryLimit(optMemoryLimit);
+
         Owned<IClientWUQuerySetCopyQueryResponse> resp = client->WUQuerysetCopyQuery(req);
         if (resp->getExceptions().ordinality())
             outputMultiExceptions(resp->getExceptions());
@@ -383,6 +409,10 @@ public:
             "   --no-reload            Do not request a reload of the (roxie) cluster\n"
             "   -O, --overwrite        Overwrite existing files\n"
             "   --wait=<ms>            Max time to wait in milliseconds\n"
+            "   --timeLimit=<sec>      Value to set for query timeLimit configuration\n"
+            "   --warnTimeLimit=<sec>  Value to set for query warnTimeLimit configuration\n"
+            "   --memoryLimit=<mem>    Value to set for query memoryLimit configuration\n"
+            "                          format <mem> as 500000B, 550K, 100M, 10G, 1T etc.\n"
             " Common Options:\n",
             stdout);
         EclCmdCommon::usage();
@@ -392,19 +422,152 @@ private:
     StringAttr optTargetQuerySet;
     StringAttr optTargetCluster;
     StringAttr optDaliIP;
+    StringAttr optMemoryLimit;
     unsigned optMsToWait;
+    unsigned optTimeLimit;
+    unsigned optWarnTimeLimit;
     bool optActivate;
     bool optNoReload;
     bool optOverwrite;
     bool optDontCopyFiles;
 };
 
+class EclCmdQueriesConfig : public EclCmdCommon
+{
+public:
+    EclCmdQueriesConfig() : optNoReload(false), optMsToWait(10000)
+    {
+        optTimeLimit = (unsigned) -1;
+        optWarnTimeLimit = (unsigned) -1;
+    }
+    virtual bool parseCommandLineOptions(ArgvIterator &iter)
+    {
+        if (iter.done())
+        {
+            usage();
+            return false;
+        }
+
+        for (; !iter.done(); iter.next())
+        {
+            const char *arg = iter.query();
+            if (*arg!='-')
+            {
+                if (optTargetCluster.isEmpty())
+                    optTargetCluster.set(arg);
+                else if (optQueryId.isEmpty())
+                    optQueryId.set(arg);
+                else
+                {
+                    fprintf(stderr, "\nunrecognized argument %s\n", arg);
+                    return false;
+                }
+                continue;
+            }
+            if (iter.matchFlag(optNoReload, ECLOPT_NORELOAD))
+                continue;
+            if (iter.matchOption(optMsToWait, ECLOPT_WAIT))
+                continue;
+            if (iter.matchOption(optTimeLimit, ECLOPT_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optWarnTimeLimit, ECLOPT_WARN_TIME_LIMIT))
+                continue;
+            if (iter.matchOption(optMemoryLimit, ECLOPT_MEMORY_LIMIT))
+                continue;
+            if (EclCmdCommon::matchCommandLineOption(iter, true)!=EclCmdOptionMatch)
+                return false;
+        }
+        return true;
+    }
+    virtual bool finalizeOptions(IProperties *globals)
+    {
+        if (!EclCmdCommon::finalizeOptions(globals))
+            return false;
+        if (optTargetCluster.isEmpty() || optQueryId.isEmpty())
+        {
+            fputs("Target and QueryId must both be specified.\n\n", stderr);
+            return false;
+        }
+        if (optMemoryLimit.length() && !isValidMemoryValue(optMemoryLimit))
+        {
+            fprintf(stderr, "invalid --memoryLimit value of %s.\n\n", optMemoryLimit.get());
+            return false;
+        }
+        return true;
+    }
+
+    virtual int processCMD()
+    {
+        Owned<IClientWsWorkunits> client = createWsWorkunitsClient();
+        VStringBuffer url("http://%s:%s/WsWorkunits", optServer.sget(), optPort.sget());
+        client->addServiceUrl(url.str());
+        if (optUsername.length())
+            client->setUsernameToken(optUsername.get(), optPassword.sget(), NULL);
+
+        Owned<IClientWUQueryConfigRequest> req = client->createWUQueryConfigRequest();
+        req->setTarget(optTargetCluster.get());
+        req->setQueryId(optQueryId.get());
+        req->setWait(optMsToWait);
+        req->setNoReload(optNoReload);
+
+        if (optTimeLimit != (unsigned) -1)
+            req->setTimeLimit(optTimeLimit);
+        if (optWarnTimeLimit != (unsigned) -1)
+            req->setWarnTimeLimit(optWarnTimeLimit);
+        if (!optMemoryLimit.isEmpty())
+            req->setMemoryLimit(optMemoryLimit);
+
+        Owned<IClientWUQueryConfigResponse> resp = client->WUQueryConfig(req);
+        if (resp->getExceptions().ordinality())
+            outputMultiExceptions(resp->getExceptions());
+        IArrayOf<IConstWUQueryConfigResult> &results = resp->getResults();
+        if (results.length())
+        {
+            fputs("configured:\n", stdout);
+            ForEachItemIn(i, results)
+                fprintf(stdout, "   %s\n", results.item(i).getQueryId());
+        }
+        return 0;
+    }
+    virtual void usage()
+    {
+        fputs("\nUsage:\n"
+            "\n"
+            "The 'queries config' command updates query configuration values.\n"
+            "\n"
+            "ecl queries config <target> <queryid> [options]\n"
+            "\n"
+            " Options:\n"
+            "   <target>               Name of target queryset containing query\n"
+            "   <queryid>              Id of the query to configure\n"
+            "   --no-reload            Do not request a reload of the (roxie) cluster\n"
+            "   --wait=<ms>            Max time to wait in milliseconds\n"
+            "   --timeLimit=<sec>      Value to set for query timeLimit configuration\n"
+            "   --warnTimeLimit=<sec>  Value to set for query warnTimeLimit configuration\n"
+            "   --memoryLimit=<mem>    Value to set for query memoryLimit configuration\n"
+            "                          format <mem> as 500000B, 550K, 100M, 10G, 1T etc.\n"
+            " Common Options:\n",
+            stdout);
+        EclCmdCommon::usage();
+    }
+private:
+    StringAttr optTargetCluster;
+    StringAttr optQueryId;
+    StringAttr optMemoryLimit;
+    unsigned optMsToWait;
+    unsigned optTimeLimit;
+    unsigned optWarnTimeLimit;
+    bool optNoReload;
+};
+
 IEclCommand *createEclQueriesCommand(const char *cmdname)
 {
     if (!cmdname || !*cmdname)
         return NULL;
     if (strieq(cmdname, "list"))
         return new EclCmdQueriesList();
+    if (strieq(cmdname, "config"))
+        return new EclCmdQueriesConfig();
     if (strieq(cmdname, "copy"))
         return new EclCmdQueriesCopy();
     return NULL;
@@ -426,6 +589,7 @@ public:
             "ecl queries <command> [command options]\n\n"
             "   Queries Commands:\n"
             "      list         list queries in queryset(s)\n"
+            "      config       update query settings\n"
             "      copy         copy a query from one queryset to another\n"
         );
     }

esp/scm/ws_workunits.ecm

@@ -1053,7 +1053,7 @@ ESPresponse [exceptions_inline] WUCopyLogicalFilesResponse
 };
 
 
-ESPrequest WUPublishWorkunitRequest
+ESPrequest [nil_remove] WUPublishWorkunitRequest
 {
     string Wuid;
     string Cluster;
@@ -1063,6 +1063,9 @@ ESPrequest WUPublishWorkunitRequest
     int Wait(10000);
     bool NoReload(0);
     bool UpdateWorkUnitName(0);
+    string memoryLimit;
+    nonNegativeInteger TimeLimit(0);
+    nonNegativeInteger WarnTimeLimit(0);
 };
 
 ESPresponse [exceptions_inline] WUPublishWorkunitResponse
@@ -1076,6 +1079,28 @@ ESPresponse [exceptions_inline] WUPublishWorkunitResponse
     ESParray<ESPStruct WUCopyLogicalClusterFileSections, Cluster> ClusterFiles;
 };
 
+ESPrequest [nil_remove] WUQueryConfigRequest
+{
+    string Target;
+    string QueryId;
+    int Wait(10000);
+    bool NoReload(0);
+    string memoryLimit;
+    nonNegativeInteger TimeLimit(0);
+    nonNegativeInteger WarnTimeLimit(0);
+};
+
+ESPStruct WUQueryConfigResult
+{
+    string  QueryId;
+};
+
+ESPresponse [exceptions_inline] WUQueryConfigResponse
+{
+    bool ReloadFailed;
+    ESParray<ESPStruct WUQueryConfigResult, Result> Results;
+};
+
 
 ESPStruct QuerySet
 {
@@ -1098,7 +1123,7 @@ ESPStruct ClusterQueryState
     string State;
 };
 
-ESPStruct QuerySetQuery
+ESPStruct [nil_remove] QuerySetQuery
 {
     string Id;
     string Name;
@@ -1106,6 +1131,9 @@ ESPStruct QuerySetQuery
     string Dll;
     bool Suspended;
     ESParray<ESPstruct ClusterQueryState> Clusters;
+    string memoryLimit;
+    nonNegativeInteger timeLimit;
+    nonNegativeInteger warnTimeLimit;
 };
 
 ESPStruct QuerySetAlias
@@ -1241,7 +1269,7 @@ ESPresponse [exceptions_inline] WUQuerySetAliasActionResponse
     ESParray<ESPstruct QuerySetAliasActionResult, Result> Results;
 };
 
-ESPrequest WUQuerySetCopyQueryRequest
+ESPrequest [nil_remove] WUQuerySetCopyQueryRequest
 {
     string Source;
     string Target;
@@ -1252,6 +1280,9 @@ ESPrequest WUQuerySetCopyQueryRequest
     bool DontCopyFiles(false);
     int Wait(10000);
     bool NoReload(0);
+    string memoryLimit;
+    nonNegativeInteger TimeLimit(0);
+    nonNegativeInteger WarnTimeLimit(0);
 };
 
 ESPresponse [exceptions_inline] WUQuerySetCopyQueryResponse
@@ -1325,6 +1356,7 @@ ESPservice [
     ESPmethod WUQuerysetAliasAction(WUQuerySetAliasActionRequest, WUQuerySetAliasActionResponse);
     ESPmethod WUQuerysetCopyQuery(WUQuerySetCopyQueryRequest, WUQuerySetCopyQueryResponse);
     ESPmethod [resp_xsl_default("/esp/xslt/WUCopyLogicalFiles.xslt")] WUCopyLogicalFiles(WUCopyLogicalFilesRequest, WUCopyLogicalFilesResponse);
+    ESPmethod WUQueryConfig(WUQueryConfigRequest, WUQueryConfigResponse);
 };
 
 

esp/services/ws_workunits/ws_workunitsQuerySets.cpp

@@ -298,6 +298,88 @@ bool reloadCluster(const char *cluster, unsigned wait)
     return (clusterInfo) ? reloadCluster(clusterInfo, wait) : true;
 }
 
+static inline void updateQuerySetting(bool ignore, IPropertyTree *queryTree, const char *xpath, int value)
+{
+    if (ignore || !queryTree)
+        return;
+    if (value!=0)
+        queryTree->setPropInt(xpath, value);
+    else
+        queryTree->removeProp(xpath);
+}
+
+static inline unsigned __int64 memoryLimitUInt64FromString(const char *value)
+{
+    if (!value || !*value || !isdigit(*value))
+        return 0;
+    unsigned __int64 result = (*value - '0');
+    const char *s = value+1;
+    while (isdigit(*s))
+    {
+        result = 10 * result + ((*s) - '0');
+        s++;
+    }
+    if (*s)
+    {
+        const char unit = toupper(*s++);
+        if (*s && !strieq("B", s)) //more?
+            return 0;
+        switch (unit)
+        {
+            case 'E':
+                result <<=60;
+                break;
+            case 'P':
+                result <<=50;
+                break;
+            case 'T':
+                result <<=40;
+                break;
+            case 'G':
+                result <<=30;
+                break;
+            case 'M':
+                result <<=20;
+                break;
+            case 'K':
+                result <<=10;
+                break;
+            case 'B':
+                break;
+            default:
+                return 0;
+        }
+    }
+    return result;
+}
+
+const char memUnitAbbrev[] = {'B', 'K', 'M', 'G', 'T', 'P', 'E'};
+#define MAX_MEMUNIT_ABBREV 6
+
+static inline StringBuffer &memoryLimitStringFromUInt64(StringBuffer &s, unsigned __int64 in)
+{
+    if (!in)
+        return s;
+    unsigned __int64 value = in;
+    unsigned char unit = 0;
+    while (!(value & 0x3FF) && unit < MAX_MEMUNIT_ABBREV)
+    {
+        value >>= 10;
+        unit++;
+    }
+    return s.append(value).append(memUnitAbbrev[unit]);
+}
+
+static inline void updateMemoryLimitSetting(IPropertyTree *queryTree, const char *value)
+{
+    if (!value || !queryTree)
+        return;
+    unsigned __int64 limit = memoryLimitUInt64FromString(value);
+    if (0==limit)
+        queryTree->removeProp("@memoryLimit");
+    else
+        queryTree->setPropInt64("@memoryLimit", limit);
+}
 
 bool CWsWorkunitsEx::onWUPublishWorkunit(IEspContext &context, IEspWUPublishWorkunitRequest & req, IEspWUPublishWorkunitResponse & resp)
 {
@@ -343,7 +425,15 @@ bool CWsWorkunitsEx::onWUPublishWorkunit(IEspContext &context, IEspWUPublishWork
         wu->setJobName(req.getJobName());
 
     StringBuffer queryId;
-    addQueryToQuerySet(wu, target.str(), queryName.str(), NULL, (WUQueryActivationOptions)req.getActivate(), queryId);
+    WUQueryActivationOptions activate = (WUQueryActivationOptions)req.getActivate();
+    addQueryToQuerySet(wu, target.str(), queryName.str(), NULL, activate, queryId);
+    if (req.getMemoryLimit() || !req.getTimeLimit_isNull() || ! req.getWarnTimeLimit_isNull())
+    {
+        Owned<IPropertyTree> queryTree = getQueryById(target.str(), queryId, false);
+        updateMemoryLimitSetting(queryTree, req.getMemoryLimit());
+        updateQuerySetting(req.getTimeLimit_isNull(), queryTree, "@timeLimit", req.getTimeLimit());
+        updateQuerySetting(req.getWarnTimeLimit_isNull(), queryTree, "@warnTimeLimit", req.getWarnTimeLimit());
+    }
     wu->commit();
     wu.clear();
 
@@ -383,6 +473,16 @@ void gatherQuerySetQueryDetails(IPropertyTree *query, IEspQuerySetQuery *queryIn
     queryInfo->setDll(query->queryProp("@dll"));
     queryInfo->setWuid(query->queryProp("@wuid"));
     queryInfo->setSuspended(query->getPropBool("@suspended", false));
+    if (query->hasProp("@memoryLimit"))
+    {
+        StringBuffer s;
+        memoryLimitStringFromUInt64(s, query->getPropInt64("@memoryLimit"));
+        queryInfo->setMemoryLimit(s);
+    }
+    if (query->hasProp("@timeLimit"))
+        queryInfo->setTimeLimit(query->getPropInt("@timeLimit"));
+    if (query->hasProp("@warnTimeLimit"))
+        queryInfo->setWarnTimeLimit(query->getPropInt("@warnTimeLimit"));
     if (queriesOnCluster)
     {
         IArrayOf<IEspClusterQueryState> clusters;
@@ -675,6 +775,58 @@ void expandQueryActionTargetList(IProperties *queryIds, IPropertyTree *queryset,
     }
 }
 
+void expandQueryActionTargetList(IProperties *queryIds, IPropertyTree *queryset, const char *id, CQuerySetQueryActionTypes action)
+{
+    IArrayOf<IConstQuerySetQueryActionItem> items;
+    Owned<IEspQuerySetQueryActionItem> item = createQuerySetQueryActionItem();
+    item->setQueryId(id);
+    items.append(*(IConstQuerySetQueryActionItem*)item.getClear());
+    expandQueryActionTargetList(queryIds, queryset, items, action);
+}
+
+bool CWsWorkunitsEx::onWUQueryConfig(IEspContext &context, IEspWUQueryConfigRequest & req, IEspWUQueryConfigResponse & resp)
+{
+    StringAttr target(req.getTarget());
+    if (target.isEmpty())
+        throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Target name required");
+    if (!isValidCluster(target))
+        throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid target name: %s", target.get());
+
+    Owned<IPropertyTree> queryset = getQueryRegistry(target.get(), false);
+    if (!queryset)
+        throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Target Queryset %s not found", req.getTarget());
+
+    Owned<IProperties> queryIds = createProperties();
+    expandQueryActionTargetList(queryIds, queryset, req.getQueryId(), QuerySetQueryActionTypes_Undefined);
+
+    IArrayOf<IEspWUQueryConfigResult> results;
+    Owned<IPropertyIterator> it = queryIds->getIterator();
+    ForEach(*it)
+    {
+        Owned<IEspWUQueryConfigResult> result = createWUQueryConfigResult();
+        result->setQueryId(it->getPropKey());
+
+        VStringBuffer xpath("Query[@id='%s']", it->getPropKey());
+        IPropertyTree *queryTree = queryset->queryPropTree(xpath);
+        if (queryTree)
+        {
+            updateMemoryLimitSetting(queryTree, req.getMemoryLimit());
+            updateQuerySetting(req.getTimeLimit_isNull(), queryTree, "@timeLimit", req.getTimeLimit());
+            updateQuerySetting(req.getWarnTimeLimit_isNull(), queryTree, "@warnTimeLimit", req.getWarnTimeLimit());
+        }
+
+        results.append(*result.getClear());
+    }
+    resp.setResults(results);
+
+    bool reloadFailed = false;
+    if (0!=req.getWait() && !req.getNoReload())
+        reloadFailed = !reloadCluster(target.get(), (unsigned)req.getWait());
+    resp.setReloadFailed(reloadFailed);
+
+    return true;
+}
+
 bool CWsWorkunitsEx::onWUQuerysetQueryAction(IEspContext &context, IEspWUQuerySetQueryActionRequest & req, IEspWUQuerySetQueryActionResponse & resp)
 {
     resp.setQuerySetName(req.getQuerySetName());
@@ -898,7 +1050,15 @@ bool CWsWorkunitsEx::onWUQuerysetCopyQuery(IEspContext &context, IEspWUQuerySetC
         throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Error opening wuid %s for query %s", wuid.str(), source);
 
     StringBuffer targetQueryId;
-    addQueryToQuerySet(wu, target, queryName.str(), NULL, (WUQueryActivationOptions)req.getActivate(), targetQueryId);
+    WUQueryActivationOptions activate = (WUQueryActivationOptions)req.getActivate();
+    addQueryToQuerySet(wu, target, queryName.str(), NULL, activate, targetQueryId);
+    if (req.getMemoryLimit() || !req.getTimeLimit_isNull() || ! req.getWarnTimeLimit_isNull())
+    {
+        Owned<IPropertyTree> queryTree = getQueryById(target, targetQueryId, false);
+        updateMemoryLimitSetting(queryTree, req.getMemoryLimit());
+        updateQuerySetting(req.getTimeLimit_isNull(), queryTree, "@timeLimit", req.getTimeLimit());
+        updateQuerySetting(req.getWarnTimeLimit_isNull(), queryTree, "@warnTimeLimit", req.getWarnTimeLimit());
+    }
     wu.clear();
 
     resp.setQueryId(targetQueryId.str());

esp/services/ws_workunits/ws_workunitsService.hpp

@@ -49,6 +49,7 @@ public:
     bool onWUMultiQuerysetDetails(IEspContext &context, IEspWUMultiQuerySetDetailsRequest &req, IEspWUMultiQuerySetDetailsResponse &resp);
     bool onWUQuerysetQueryAction(IEspContext &context, IEspWUQuerySetQueryActionRequest & req, IEspWUQuerySetQueryActionResponse & resp);
     bool onWUQuerysetAliasAction(IEspContext &context, IEspWUQuerySetAliasActionRequest &req, IEspWUQuerySetAliasActionResponse &resp);
+    bool onWUQueryConfig(IEspContext &context, IEspWUQueryConfigRequest &req, IEspWUQueryConfigResponse &resp);
     bool onWUQuerysetCopyQuery(IEspContext &context, IEspWUQuerySetCopyQueryRequest &req, IEspWUQuerySetCopyQueryResponse &resp);
     bool onWUCopyLogicalFiles(IEspContext &context, IEspWUCopyLogicalFilesRequest &req, IEspWUCopyLogicalFilesResponse &resp);