ws_workunitsQuerySets.cpp 105 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647
  1. /*##############################################################################
  2. HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. ############################################################################## */
  13. #include "ws_workunitsService.hpp"
  14. #include "ws_fs.hpp"
  15. #include "jlib.hpp"
  16. #include "daclient.hpp"
  17. #include "dalienv.hpp"
  18. #include "dadfs.hpp"
  19. #include "dfuwu.hpp"
  20. #include "eclhelper.hpp"
  21. #include "roxiecontrol.hpp"
  22. #include "dfuutil.hpp"
  23. #include "dautils.hpp"
  24. #include "httpclient.hpp"
  25. #define DALI_FILE_LOOKUP_TIMEOUT (1000*15*1) // 15 seconds
  26. const unsigned ROXIECONNECTIONTIMEOUT = 1000; //1 second
  27. const unsigned ROXIECONTROLQUERYTIMEOUT = 3000; //3 second
  28. const unsigned ROXIECONTROLQUERIESTIMEOUT = 30000; //30 second
  29. const unsigned ROXIELOCKCONNECTIONTIMEOUT = 60000; //60 second
  30. #define SDS_LOCK_TIMEOUT (5*60*1000) // 5mins, 30s a bit short
  31. bool isRoxieProcess(const char *process)
  32. {
  33. if (!process)
  34. return false;
  35. Owned<IEnvironmentFactory> factory = getEnvironmentFactory();
  36. Owned<IConstEnvironment> env = factory->openEnvironment();
  37. if (!env)
  38. return false;
  39. Owned<IPropertyTree> root = &env->getPTree();
  40. VStringBuffer xpath("Software/RoxieCluster[@name=\"%s\"]", process);
  41. return root->hasProp(xpath.str());
  42. }
  43. void checkUseEspOrDaliIP(SocketEndpoint &ep, const char *ip, const char *esp)
  44. {
  45. if (!ip || !*ip)
  46. return;
  47. ep.set(ip, 7070);
  48. if (ep.isLoopBack() || *ip=='.' || (ip[0]=='0' && ip[1]=='.'))
  49. ep.ipset(esp);
  50. }
  51. static IClientWsWorkunits *ensureWsWorkunitsClient(IClientWsWorkunits *ws, IEspContext *ctx, const char *netAddress)
  52. {
  53. if (ws)
  54. return LINK(ws);
  55. StringBuffer url;
  56. if (netAddress && *netAddress)
  57. url.appendf("http://%s%s/WsWorkunits", netAddress, (!strchr(netAddress, ':')) ? ":8010" : "");
  58. else
  59. {
  60. if (!ctx)
  61. throw MakeStringException(ECLWATCH_INVALID_IP, "Missing WsWorkunits service address");
  62. StringBuffer ip;
  63. short port = 0;
  64. ctx->getServAddress(ip, port);
  65. url.appendf("http://%s:%d/WsWorkunits", ip.str(), port);
  66. }
  67. Owned<IClientWsWorkunits> cws = createWsWorkunitsClient();
  68. cws->addServiceUrl(url);
  69. if (ctx && ctx->queryUserId() && *ctx->queryUserId())
  70. cws->setUsernameToken(ctx->queryUserId(), ctx->queryPassword(), NULL);
  71. return cws.getClear();
  72. }
  73. IClientWUQuerySetDetailsResponse *fetchQueryDetails(IClientWsWorkunits *_ws, IEspContext *ctx, const char *netAddress, const char *target, const char *queryid)
  74. {
  75. Owned<IClientWsWorkunits> ws = ensureWsWorkunitsClient(_ws, ctx, netAddress);
  76. //using existing WUQuerysetDetails rather than extending WUQueryDetails, to support copying query meta data from prior releases
  77. Owned<IClientWUQuerySetDetailsRequest> reqQueryInfo = ws->createWUQuerysetDetailsRequest();
  78. reqQueryInfo->setClusterName(target);
  79. reqQueryInfo->setQuerySetName(target);
  80. reqQueryInfo->setFilter(queryid);
  81. reqQueryInfo->setFilterType("Id");
  82. return ws->WUQuerysetDetails(reqQueryInfo);
  83. }
  84. void fetchRemoteWorkunit(IClientWsWorkunits *_ws, IEspContext *ctx, const char *netAddress, const char *queryset, const char *query, const char *wuid, StringBuffer &name, StringBuffer &xml, StringBuffer &dllname, MemoryBuffer &dll, StringBuffer &daliServer)
  85. {
  86. Owned<IClientWsWorkunits> ws = ensureWsWorkunitsClient(_ws, ctx, netAddress);
  87. Owned<IClientWULogFileRequest> req = ws->createWUFileRequest();
  88. if (queryset && *queryset)
  89. req->setQuerySet(queryset);
  90. if (query && *query)
  91. req->setQuery(query);
  92. if (wuid && *wuid)
  93. req->setWuid(wuid);
  94. req->setType("xml");
  95. Owned<IClientWULogFileResponse> resp = ws->WUFile(req);
  96. if (!resp || resp->getExceptions().ordinality() || !resp->getThefile().length())
  97. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Cannot retrieve remote workunit");
  98. xml.append(resp->getThefile().length(), resp->getThefile().toByteArray());
  99. req->setType("dll");
  100. resp.setown(ws->WUFile(req));
  101. if (!resp || resp->getExceptions().ordinality() || !resp->getThefile().length())
  102. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Cannot retrieve remote workunit shared object");
  103. dll.append(resp->getThefile());
  104. dllname.append(resp->getFileName());
  105. name.append(resp->getQueryName());
  106. SocketEndpoint ep;
  107. checkUseEspOrDaliIP(ep, resp->getDaliServer(), netAddress);
  108. if (!ep.isNull())
  109. ep.getUrlStr(daliServer);
  110. }
  111. void fetchRemoteWorkunitAndQueryDetails(IClientWsWorkunits *_ws, IEspContext *ctx, const char *netAddress, const char *queryset, const char *query, const char *wuid, StringBuffer &name, StringBuffer &xml, StringBuffer &dllname, MemoryBuffer &dll, StringBuffer &daliServer, Owned<IClientWUQuerySetDetailsResponse> &respQueryInfo)
  112. {
  113. Owned<IClientWsWorkunits> ws = ensureWsWorkunitsClient(_ws, ctx, netAddress);
  114. fetchRemoteWorkunit(ws, ctx, netAddress, queryset, query, wuid, name, xml, dllname, dll, daliServer);
  115. respQueryInfo.setown(fetchQueryDetails(ws, ctx, netAddress, queryset, query));
  116. }
  117. void doWuFileCopy(IClientFileSpray &fs, IEspWULogicalFileCopyInfo &info, const char *logicalname, const char *cluster, bool isRoxie, bool supercopy)
  118. {
  119. try
  120. {
  121. Owned<IClientCopy> req = fs.createCopyRequest();
  122. req->setSourceLogicalName(logicalname);
  123. req->setDestLogicalName(logicalname);
  124. req->setDestGroup(cluster);
  125. req->setSuperCopy(supercopy);
  126. if (isRoxie)
  127. req->setDestGroupRoxie("Yes");
  128. Owned<IClientCopyResponse> resp = fs.Copy(req);
  129. info.setDfuCopyWuid(resp->getResult());
  130. }
  131. catch (IException *e)
  132. {
  133. StringBuffer msg;
  134. info.setDfuCopyError(e->errorMessage(msg).str());
  135. e->Release();
  136. }
  137. }
  138. bool copyWULogicalFiles(IEspContext &context, IConstWorkUnit &cw, const char *cluster, bool copyLocal, IEspWUCopyLogicalClusterFileSections &lfinfo)
  139. {
  140. if (isEmpty(cluster))
  141. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "copyWULogicalFiles Cluster parameter not set.");
  142. Owned<IUserDescriptor> udesc = createUserDescriptor();
  143. udesc->set(context.queryUserId(), context.queryPassword());
  144. IArrayOf<IEspWULogicalFileCopyInfo> foreign;
  145. IArrayOf<IEspWULogicalFileCopyInfo> onCluster;
  146. IArrayOf<IEspWULogicalFileCopyInfo> notOnCluster;
  147. IArrayOf<IEspWULogicalFileCopyInfo> notFound;
  148. Owned<IClientFileSpray> fs;
  149. if (copyLocal)
  150. {
  151. fs.setown(createFileSprayClient());
  152. VStringBuffer url("http://.:%d/FileSpray", 8010);
  153. fs->addServiceUrl(url.str());
  154. }
  155. bool isRoxie = isRoxieProcess(cluster);
  156. Owned<IConstWUGraphIterator> graphs = &cw.getGraphs(GraphTypeActivities);
  157. ForEach(*graphs)
  158. {
  159. Owned <IPropertyTree> xgmml = graphs->query().getXGMMLTree(false);
  160. Owned<IPropertyTreeIterator> iter = xgmml->getElements(".//node");
  161. ForEach(*iter)
  162. {
  163. try
  164. {
  165. IPropertyTree &node = iter->query();
  166. ThorActivityKind kind = (ThorActivityKind) node.getPropInt("att[@name='_kind']/@value", TAKnone);
  167. if(kind==TAKdiskwrite || kind==TAKindexwrite || kind==TAKcsvwrite || kind==TAKxmlwrite || kind==TAKjsonwrite)
  168. continue;
  169. if (node.getPropBool("att[@name='_isSpill']/@value") || node.getPropBool("att[@name='_isTransformSpill']/@value"))
  170. continue;
  171. Owned<IEspWULogicalFileCopyInfo> info = createWULogicalFileCopyInfo();
  172. const char *logicalname = node.queryProp("att[@name='_indexFileName']/@value");
  173. if (logicalname)
  174. info->setIsIndex(true);
  175. else
  176. logicalname = node.queryProp("att[@name='_fileName']/@value");
  177. info->setLogicalName(logicalname);
  178. if (logicalname)
  179. {
  180. if (!strnicmp("~foreign::", logicalname, 10))
  181. foreign.append(*info.getClear());
  182. else
  183. {
  184. Owned<IDistributedFile> df = queryDistributedFileDirectory().lookup(logicalname, udesc);
  185. if(!df)
  186. notFound.append(*info.getClear());
  187. else if (df->findCluster(cluster)!=NotFound)
  188. {
  189. onCluster.append(*info.getClear());
  190. }
  191. else
  192. {
  193. StringArray clusters;
  194. df->getClusterNames(clusters);
  195. info->setClusters(clusters);
  196. if (copyLocal)
  197. {
  198. StringBuffer wuid;
  199. bool supercopy = queryDistributedFileDirectory().isSuperFile(logicalname, udesc, NULL);
  200. doWuFileCopy(*fs, *info, logicalname, cluster, isRoxie, supercopy);
  201. }
  202. notOnCluster.append(*info.getClear());
  203. }
  204. }
  205. }
  206. }
  207. catch(IException *e)
  208. {
  209. e->Release();
  210. }
  211. }
  212. lfinfo.setClusterName(cluster);
  213. lfinfo.setNotOnCluster(notOnCluster);
  214. lfinfo.setOnCluster(onCluster);
  215. lfinfo.setForeign(foreign);
  216. lfinfo.setNotFound(notFound);
  217. }
  218. return true;
  219. }
  220. void copyWULogicalFilesToTarget(IEspContext &context, IConstWUClusterInfo &clusterInfo, IConstWorkUnit &cw, IArrayOf<IConstWUCopyLogicalClusterFileSections> &clusterfiles, bool doLocalCopy)
  221. {
  222. const StringArray &thors = clusterInfo.getThorProcesses();
  223. ForEachItemIn(i, thors)
  224. {
  225. Owned<IEspWUCopyLogicalClusterFileSections> files = createWUCopyLogicalClusterFileSections();
  226. copyWULogicalFiles(context, cw, thors.item(i), doLocalCopy, *files);
  227. clusterfiles.append(*files.getClear());
  228. }
  229. SCMStringBuffer roxie;
  230. clusterInfo.getRoxieProcess(roxie);
  231. if (roxie.length())
  232. {
  233. Owned<IEspWUCopyLogicalClusterFileSections> files = createWUCopyLogicalClusterFileSections();
  234. copyWULogicalFiles(context, cw, roxie.str(), doLocalCopy, *files);
  235. clusterfiles.append(*files.getClear());
  236. }
  237. }
  238. void QueryFilesInUse::loadTarget(IPropertyTree *t, const char *target, unsigned flags)
  239. {
  240. if (!target || !*target)
  241. return;
  242. Owned<IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(target);
  243. if (!clusterInfo || !(clusterInfo->getPlatform() == RoxieCluster))
  244. return;
  245. Owned<IPropertyTree> queryRegistry = getQueryRegistry(target, true);
  246. if (!queryRegistry)
  247. return;
  248. SCMStringBuffer process;
  249. clusterInfo->getRoxieProcess(process);
  250. if (!process.length())
  251. return;
  252. Owned<IHpccPackageSet> ps = createPackageSet(process.str());
  253. const IHpccPackageMap *pm = (ps) ? ps->queryActiveMap(target) : NULL;
  254. const char *pmid = (pm) ? pm->queryPackageId() : NULL;
  255. VStringBuffer xpath("%s/@pmid", target);
  256. const char *pmidPrev = t->queryProp(xpath);
  257. if ((flags & UFO_RELOAD_TARGETS_CHANGED_PMID) && (pmid || pmidPrev))
  258. {
  259. if (!(pmid && pmidPrev) || !streq(pmid, pmidPrev))
  260. t->removeProp(target);
  261. }
  262. IPropertyTree *targetTree = ensurePTree(t, target);
  263. if (pm)
  264. targetTree->setProp("@pmid", pmid);
  265. if (flags & UFO_REMOVE_QUERIES_NOT_IN_QUERYSET)
  266. {
  267. Owned<IPropertyTreeIterator> cachedQueries = targetTree->getElements("Query");
  268. ForEach(*cachedQueries)
  269. {
  270. IPropertyTree &cachedQuery = cachedQueries->query();
  271. VStringBuffer xpath("Query[@id='%s']", cachedQuery.queryProp("@id"));
  272. if (!queryRegistry->hasProp(xpath))
  273. targetTree->removeTree(&cachedQuery);
  274. }
  275. }
  276. Owned<IPropertyTreeIterator> queries = queryRegistry->getElements("Query");
  277. ForEach(*queries)
  278. {
  279. if (aborting)
  280. return;
  281. IPropertyTree &query = queries->query();
  282. const char *queryid = query.queryProp("@id");
  283. if (!queryid || !*queryid)
  284. continue;
  285. const char *wuid = query.queryProp("@wuid");
  286. if (!wuid || !*wuid)
  287. continue;
  288. const char *pkgid=NULL;
  289. if (pm)
  290. {
  291. const IHpccPackage *pkg = pm->matchPackage(queryid);
  292. if (pkg)
  293. pkgid = pkg->queryId();
  294. }
  295. VStringBuffer xpath("Query[@id='%s']", queryid);
  296. IPropertyTree *queryTree = targetTree->queryPropTree(xpath);
  297. if (queryTree)
  298. {
  299. const char *cachedPkgid = queryTree->queryProp("@pkgid");
  300. if (pkgid && *pkgid)
  301. {
  302. if (!(flags & UFO_RELOAD_MAPPED_QUERIES) && (cachedPkgid && streq(pkgid, cachedPkgid)))
  303. continue;
  304. }
  305. else if (!cachedPkgid || !*cachedPkgid)
  306. continue;
  307. targetTree->removeTree(queryTree);
  308. queryTree = NULL;
  309. }
  310. Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
  311. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid);
  312. if (!cw)
  313. continue;
  314. queryTree = targetTree->addPropTree("Query", createPTree("Query"));
  315. queryTree->setProp("@target", target); //for reference when searching across targets
  316. queryTree->setProp("@id", queryid);
  317. if (pkgid && *pkgid)
  318. queryTree->setProp("@pkgid", pkgid);
  319. IUserDescriptor **roxieUser = roxieUserMap.getValue(target);
  320. Owned<IReferencedFileList> wufiles = createReferencedFileList(roxieUser ? *roxieUser : NULL, true, true);
  321. wufiles->addFilesFromQuery(cw, pm, queryid);
  322. if (aborting)
  323. return;
  324. wufiles->resolveFiles(process.str(), NULL, NULL, NULL, true, true, false, false);
  325. Owned<IReferencedFileIterator> files = wufiles->getFiles();
  326. ForEach(*files)
  327. {
  328. if (aborting)
  329. return;
  330. IReferencedFile &rf = files->query();
  331. //if (!(rf.getFlags() & RefSubFile))
  332. // continue;
  333. const char *lfn = rf.getLogicalName();
  334. if (!lfn || !*lfn)
  335. continue;
  336. if (!queryTree->hasProp(xpath.setf("File[@lfn='%s']", lfn)))
  337. {
  338. IPropertyTree *fileTree = queryTree->addPropTree("File", createPTree("File"));
  339. fileTree->setProp("@lfn", lfn);
  340. if (rf.getFlags() & RefFileSuper)
  341. fileTree->setPropBool("@super", true);
  342. if (rf.getFlags() & RefFileNotFound)
  343. fileTree->setPropBool("@notFound", true);
  344. const char *fpkgid = rf.queryPackageId();
  345. if (fpkgid && *fpkgid)
  346. fileTree->setProp("@pkgid", fpkgid);
  347. if (rf.getFileSize())
  348. fileTree->setPropInt64("@size", rf.getFileSize());
  349. if (rf.getNumParts())
  350. fileTree->setPropInt("@numparts", rf.getNumParts());
  351. }
  352. }
  353. }
  354. }
  355. void QueryFilesInUse::loadTargets(IPropertyTree *t, unsigned flags)
  356. {
  357. Owned<IStringIterator> targets = getTargetClusters("RoxieCluster", NULL);
  358. SCMStringBuffer s;
  359. ForEach(*targets)
  360. {
  361. if (aborting)
  362. return;
  363. loadTarget(t, targets->str(s).str(), flags);
  364. }
  365. }
  366. IPropertyTreeIterator *QueryFilesInUse::findAllQueriesUsingFile(const char *lfn)
  367. {
  368. if (!lfn || !*lfn)
  369. return NULL;
  370. Owned<IPropertyTree> t = getTree();
  371. VStringBuffer xpath("*/Query[File/@lfn='%s']", lfn);
  372. return t->getElements(xpath);
  373. }
  374. IPropertyTreeIterator *QueryFilesInUse::findQueriesUsingFile(const char *target, const char *lfn, StringAttr &pmid)
  375. {
  376. if (!lfn || !*lfn)
  377. return NULL;
  378. if (!target || !*target)
  379. return findAllQueriesUsingFile(lfn);
  380. Owned<IPropertyTree> t = getTree();
  381. IPropertyTree *targetTree = t->queryPropTree(target);
  382. if (!targetTree)
  383. return NULL;
  384. pmid.set(targetTree->queryProp("@pmid"));
  385. VStringBuffer xpath("Query[File/@lfn='%s']", lfn);
  386. return targetTree->getElements(xpath);
  387. }
  388. bool CWsWorkunitsEx::onWUCopyLogicalFiles(IEspContext &context, IEspWUCopyLogicalFilesRequest &req, IEspWUCopyLogicalFilesResponse &resp)
  389. {
  390. StringBuffer wuid = req.getWuid();
  391. WsWuHelpers::checkAndTrimWorkunit("WUCopyLogicalFiles", wuid);
  392. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  393. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid.str());
  394. if (!cw)
  395. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT,"Cannot open workunit %s", wuid.str());
  396. resp.setWuid(wuid.str());
  397. StringAttr cluster;
  398. if (notEmpty(req.getCluster()))
  399. cluster.set(req.getCluster());
  400. else
  401. cluster.set(cw->queryClusterName());
  402. if (!isValidCluster(cluster))
  403. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid cluster name: %s", cluster.str());
  404. Owned <IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(cluster.str());
  405. IArrayOf<IConstWUCopyLogicalClusterFileSections> clusterfiles;
  406. copyWULogicalFilesToTarget(context, *clusterInfo, *cw, clusterfiles, req.getCopyLocal());
  407. resp.setClusterFiles(clusterfiles);
  408. return true;
  409. }
  410. static inline unsigned remainingMsWait(unsigned wait, unsigned start)
  411. {
  412. if (wait==0 || wait==(unsigned)-1)
  413. return wait;
  414. unsigned waited = msTick()-start;
  415. return (wait>waited) ? wait-waited : 0;
  416. }
  417. bool reloadCluster(IConstWUClusterInfo *clusterInfo, unsigned wait)
  418. {
  419. if (0==wait || !clusterInfo || clusterInfo->getPlatform()!=RoxieCluster)
  420. return true;
  421. const SocketEndpointArray &addrs = clusterInfo->getRoxieServers();
  422. if (addrs.length())
  423. {
  424. try
  425. {
  426. Owned<IPropertyTree> result = sendRoxieControlAllNodes(addrs.item(0), "<control:reload/>", false, wait);
  427. const char *status = result->queryProp("Endpoint[1]/Status");
  428. if (!status || !strieq(status, "ok"))
  429. return false;
  430. }
  431. catch(IMultiException *me)
  432. {
  433. StringBuffer err;
  434. DBGLOG("ERROR control:reloading roxie query info %s", me->errorMessage(err.append(me->errorCode()).append(' ')).str());
  435. me->Release();
  436. return false;
  437. }
  438. catch(IException *e)
  439. {
  440. StringBuffer err;
  441. DBGLOG("ERROR control:reloading roxie query info %s", e->errorMessage(err.append(e->errorCode()).append(' ')).str());
  442. e->Release();
  443. return false;
  444. }
  445. }
  446. return true;
  447. }
  448. bool reloadCluster(const char *cluster, unsigned wait)
  449. {
  450. Owned <IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(cluster);
  451. return (clusterInfo) ? reloadCluster(clusterInfo, wait) : true;
  452. }
  453. static inline void updateQuerySetting(bool ignore, IPropertyTree *queryTree, const char *xpath, int value)
  454. {
  455. if (ignore || !queryTree)
  456. return;
  457. if (value!=0)
  458. queryTree->setPropInt(xpath, value);
  459. else
  460. queryTree->removeProp(xpath);
  461. }
  462. static inline void updateTimeLimitSetting(IPropertyTree *queryTree, bool valueNotSet, int value, IConstQuerySetQuery *srcInfo=NULL)
  463. {
  464. if (valueNotSet && srcInfo && !srcInfo->getTimeLimit_isNull())
  465. {
  466. value = srcInfo->getTimeLimit();
  467. valueNotSet=false;
  468. }
  469. updateQuerySetting(valueNotSet, queryTree, "@timeLimit", value);
  470. }
  471. static inline void updateWarnTimeLimitSetting(IPropertyTree *queryTree, bool valueNotSet, int value, IConstQuerySetQuery *srcInfo=NULL)
  472. {
  473. if (valueNotSet && srcInfo && !srcInfo->getWarnTimeLimit_isNull())
  474. {
  475. value = srcInfo->getWarnTimeLimit();
  476. valueNotSet=false;
  477. }
  478. updateQuerySetting(valueNotSet, queryTree, "@warnTimeLimit", value);
  479. }
  480. static inline unsigned __int64 memoryLimitUInt64FromString(const char *value)
  481. {
  482. if (!value || !*value || !isdigit(*value))
  483. return 0;
  484. unsigned __int64 result = (*value - '0');
  485. const char *s = value+1;
  486. while (isdigit(*s))
  487. {
  488. result = 10 * result + ((*s) - '0');
  489. s++;
  490. }
  491. if (*s)
  492. {
  493. const char unit = toupper(*s++);
  494. if (*s && !strieq("B", s)) //more?
  495. return 0;
  496. switch (unit)
  497. {
  498. case 'E':
  499. result <<=60;
  500. break;
  501. case 'P':
  502. result <<=50;
  503. break;
  504. case 'T':
  505. result <<=40;
  506. break;
  507. case 'G':
  508. result <<=30;
  509. break;
  510. case 'M':
  511. result <<=20;
  512. break;
  513. case 'K':
  514. result <<=10;
  515. break;
  516. case 'B':
  517. break;
  518. default:
  519. return 0;
  520. }
  521. }
  522. return result;
  523. }
  524. const char memUnitAbbrev[] = {'B', 'K', 'M', 'G', 'T', 'P', 'E'};
  525. #define MAX_MEMUNIT_ABBREV 6
  526. static inline StringBuffer &memoryLimitStringFromUInt64(StringBuffer &s, unsigned __int64 in)
  527. {
  528. if (!in)
  529. return s;
  530. unsigned __int64 value = in;
  531. unsigned char unit = 0;
  532. while (!(value & 0x3FF) && unit < MAX_MEMUNIT_ABBREV)
  533. {
  534. value >>= 10;
  535. unit++;
  536. }
  537. return s.append(value).append(memUnitAbbrev[unit]);
  538. }
  539. static inline void updateMemoryLimitSetting(IPropertyTree *queryTree, const char *value, IConstQuerySetQuery *srcInfo=NULL)
  540. {
  541. if (!queryTree)
  542. return;
  543. if (!value && srcInfo)
  544. value = srcInfo->getMemoryLimit();
  545. if (!value)
  546. return;
  547. unsigned __int64 limit = memoryLimitUInt64FromString(value);
  548. if (0==limit)
  549. queryTree->removeProp("@memoryLimit");
  550. else
  551. queryTree->setPropInt64("@memoryLimit", limit);
  552. }
  553. enum QueryPriority {
  554. QueryPriorityNone = -1,
  555. QueryPriorityLow = 0,
  556. QueryPriorityHigh = 1,
  557. QueryPrioritySLA = 2,
  558. QueryPriorityInvalid = 3
  559. };
  560. static inline const char *getQueryPriorityName(int value)
  561. {
  562. switch (value)
  563. {
  564. case QueryPriorityLow:
  565. return "LOW";
  566. case QueryPriorityHigh:
  567. return "HIGH";
  568. case QueryPrioritySLA:
  569. return "SLA";
  570. case QueryPriorityNone:
  571. return "NONE";
  572. }
  573. return "INVALID";
  574. }
  575. static inline void updateQueryPriority(IPropertyTree *queryTree, const char *value, IConstQuerySetQuery *srcInfo=NULL)
  576. {
  577. if (!queryTree)
  578. return;
  579. if ((!value || !*value) && srcInfo)
  580. value = srcInfo->getPriority();
  581. if (!value || !*value)
  582. return;
  583. int priority = QueryPriorityInvalid;
  584. if (strieq("LOW", value))
  585. priority=QueryPriorityLow;
  586. else if (strieq("HIGH", value))
  587. priority=QueryPriorityHigh;
  588. else if (strieq("SLA", value))
  589. priority=QueryPrioritySLA;
  590. else if (strieq("NONE", value))
  591. priority=QueryPriorityNone;
  592. switch (priority)
  593. {
  594. case QueryPriorityInvalid:
  595. break;
  596. case QueryPriorityNone:
  597. queryTree->removeProp("@priority");
  598. break;
  599. default:
  600. queryTree->setPropInt("@priority", priority);
  601. break;
  602. }
  603. }
  604. void copyQueryFilesToCluster(unsigned updateFlags, IEspContext &context, IConstWorkUnit *cw, const char *remoteIP, const char *remotePrefix, const char *target, const char *srcCluster, const char *queryname, bool allowForeignFiles)
  605. {
  606. if (!target || !*target)
  607. return;
  608. SCMStringBuffer process;
  609. Owned <IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(target);
  610. if (clusterInfo && clusterInfo->getPlatform()==RoxieCluster)
  611. {
  612. clusterInfo->getRoxieProcess(process);
  613. if (!process.length())
  614. return;
  615. Owned<IReferencedFileList> wufiles = createReferencedFileList(context.queryUserId(), context.queryPassword(), allowForeignFiles, false);
  616. Owned<IHpccPackageSet> ps = createPackageSet(process.str());
  617. StringBuffer queryid;
  618. if (queryname && *queryname)
  619. queryname = queryid.append(queryname).append(".0").str(); //prepublish dummy version number to support fuzzy match like queries="myquery.*" in package
  620. wufiles->addFilesFromQuery(cw, (ps) ? ps->queryActiveMap(target) : NULL, queryname);
  621. wufiles->resolveFiles(process.str(), remoteIP, remotePrefix, srcCluster, !(updateFlags & (DALI_UPDATEF_REPLACE_FILE | DALI_UPDATEF_CLONE_FROM | DALI_UPDATEF_SUPERFILES)), true, false, true);
  622. StringBuffer defReplicateFolder;
  623. getConfigurationDirectory(NULL, "data2", "roxie", process.str(), defReplicateFolder);
  624. Owned<IDFUhelper> helper = createIDFUhelper();
  625. wufiles->cloneAllInfo(updateFlags, helper, true, true, clusterInfo->getRoxieRedundancy(), clusterInfo->getChannelsPerNode(), clusterInfo->getRoxieReplicateOffset(), defReplicateFolder);
  626. }
  627. }
  628. bool CWsWorkunitsEx::isQuerySuspended(const char* query, IConstWUClusterInfo *clusterInfo, unsigned wait, StringBuffer& errorMessage)
  629. {
  630. try
  631. {
  632. if (0==wait || !clusterInfo || clusterInfo->getPlatform()!=RoxieCluster)
  633. return false;
  634. const SocketEndpointArray &addrs = clusterInfo->getRoxieServers();
  635. if (addrs.length() < 1)
  636. return false;
  637. StringBuffer control;
  638. control.appendf("<control:queries><Query id='%s'/></control:queries>", query);
  639. Owned<IPropertyTree> result = sendRoxieControlAllNodes(addrs.item(0), control.str(), false, wait);
  640. if (!result)
  641. return false;
  642. Owned<IPropertyTreeIterator> suspendedQueries = result->getElements("Endpoint/Queries/Query[@suspended='1']");
  643. if (!suspendedQueries->first())
  644. return false;
  645. errorMessage.set(suspendedQueries->query().queryProp("@error"));
  646. return true;
  647. }
  648. catch(IMultiException *me)
  649. {
  650. StringBuffer err;
  651. DBGLOG("ERROR control:queries roxie query info %s", me->errorMessage(err.append(me->errorCode()).append(' ')).str());
  652. me->Release();
  653. return false;
  654. }
  655. catch(IException *e)
  656. {
  657. StringBuffer err;
  658. DBGLOG("ERROR control:queries roxie query info %s", e->errorMessage(err.append(e->errorCode()).append(' ')).str());
  659. e->Release();
  660. return false;
  661. }
  662. }
  663. bool CWsWorkunitsEx::onWUPublishWorkunit(IEspContext &context, IEspWUPublishWorkunitRequest & req, IEspWUPublishWorkunitResponse & resp)
  664. {
  665. StringBuffer wuid = req.getWuid();
  666. WsWuHelpers::checkAndTrimWorkunit("WUPublishWorkunit", wuid);
  667. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  668. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid.str());
  669. if (!cw)
  670. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT,"Cannot find the workunit %s", wuid.str());
  671. resp.setWuid(wuid.str());
  672. StringAttr queryName;
  673. if (notEmpty(req.getJobName()))
  674. queryName.set(req.getJobName());
  675. else
  676. queryName.set(cw->queryJobName());
  677. if (!queryName.length())
  678. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Query/Job name not defined for publishing workunit %s", wuid.str());
  679. StringAttr target;
  680. if (notEmpty(req.getCluster()))
  681. target.set(req.getCluster());
  682. else
  683. target.set(cw->queryClusterName());
  684. if (!target.length())
  685. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Cluster name not defined for publishing workunit %s", wuid.str());
  686. if (!isValidCluster(target.str()))
  687. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid cluster name: %s", target.str());
  688. DBGLOG("%s publishing wuid %s to target %s as query %s", context.queryUserId(), wuid.str(), target.str(), queryName.str());
  689. StringBuffer daliIP;
  690. StringBuffer srcCluster;
  691. StringBuffer srcPrefix;
  692. splitDerivedDfsLocation(req.getRemoteDali(), srcCluster, daliIP, srcPrefix, req.getSourceProcess(),req.getSourceProcess(), NULL, NULL);
  693. if (srcCluster.length())
  694. {
  695. if (!isProcessCluster(daliIP, srcCluster))
  696. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Process cluster %s not found on %s DALI", srcCluster.str(), daliIP.length() ? daliIP.str() : "local");
  697. }
  698. unsigned updateFlags = 0;
  699. if (req.getUpdateDfs())
  700. updateFlags |= (DALI_UPDATEF_SUPERFILES | DALI_UPDATEF_REPLACE_FILE | DALI_UPDATEF_CLONE_FROM);
  701. if (req.getUpdateCloneFrom())
  702. updateFlags |= DALI_UPDATEF_CLONE_FROM;
  703. if (req.getUpdateSuperFiles())
  704. updateFlags |= DALI_UPDATEF_SUPERFILES;
  705. if (req.getAppendCluster())
  706. updateFlags |= DALI_UPDATEF_APPEND_CLUSTER;
  707. if (!req.getDontCopyFiles())
  708. copyQueryFilesToCluster(updateFlags, context, cw, daliIP, srcPrefix, target.str(), srcCluster, queryName.str(), req.getAllowForeignFiles());
  709. WorkunitUpdate wu(&cw->lock());
  710. if (req.getUpdateWorkUnitName() && notEmpty(req.getJobName()))
  711. wu->setJobName(req.getJobName());
  712. StringBuffer queryId;
  713. WUQueryActivationOptions activate = (WUQueryActivationOptions)req.getActivate();
  714. addQueryToQuerySet(wu, target.str(), queryName.str(), activate, queryId, context.queryUserId());
  715. if (req.getMemoryLimit() || !req.getTimeLimit_isNull() || !req.getWarnTimeLimit_isNull() || req.getPriority() || req.getComment())
  716. {
  717. Owned<IPropertyTree> queryTree = getQueryById(target.str(), queryId, false);
  718. updateMemoryLimitSetting(queryTree, req.getMemoryLimit());
  719. updateQuerySetting(req.getTimeLimit_isNull(), queryTree, "@timeLimit", req.getTimeLimit());
  720. updateQuerySetting(req.getWarnTimeLimit_isNull(), queryTree, "@warnTimeLimit", req.getWarnTimeLimit());
  721. updateQueryPriority(queryTree, req.getPriority());
  722. if (req.getComment())
  723. queryTree->setProp("@comment", req.getComment());
  724. }
  725. wu->commit();
  726. wu.clear();
  727. if (queryId.length())
  728. resp.setQueryId(queryId.str());
  729. resp.setQueryName(queryName.str());
  730. resp.setQuerySet(target.str());
  731. Owned <IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(target.str());
  732. bool reloadFailed = false;
  733. if (0!=req.getWait() && !req.getNoReload())
  734. reloadFailed = !reloadCluster(clusterInfo, (unsigned)req.getWait());
  735. resp.setReloadFailed(reloadFailed);
  736. double version = context.getClientVersion();
  737. if (version > 1.38)
  738. {
  739. StringBuffer errorMessage;
  740. if (!reloadFailed && !req.getNoReload() && isQuerySuspended(queryName.str(), clusterInfo, (unsigned)req.getWait(), errorMessage))
  741. {
  742. resp.setSuspended(true);
  743. resp.setErrorMessage(errorMessage);
  744. }
  745. }
  746. return true;
  747. }
  748. bool CWsWorkunitsEx::onWUQuerysets(IEspContext &context, IEspWUQuerysetsRequest & req, IEspWUQuerysetsResponse & resp)
  749. {
  750. IArrayOf<IEspQuerySet> querySets;
  751. Owned<IStringIterator> targets = getTargetClusters(NULL, NULL);
  752. SCMStringBuffer target;
  753. ForEach(*targets)
  754. {
  755. Owned<IEspQuerySet> qs = createQuerySet();
  756. qs->setQuerySetName(targets->str(target).str());
  757. querySets.append(*qs.getClear());
  758. }
  759. resp.setQuerysets(querySets);
  760. return true;
  761. }
  762. void addClusterQueryStates(IPropertyTree* queriesOnCluster, const char *target, const char *id, IArrayOf<IEspClusterQueryState>& clusterStates, double version)
  763. {
  764. if (queriesOnCluster)
  765. queriesOnCluster = queriesOnCluster->queryPropTree("Endpoint[1]/Queries[1]");
  766. if (!queriesOnCluster)
  767. return;
  768. int reporting = queriesOnCluster->getPropInt("@reporting");
  769. Owned<IEspClusterQueryState> clusterState = createClusterQueryState();
  770. clusterState->setCluster(target);
  771. VStringBuffer xpath("Query[@id='%s']", id);
  772. IPropertyTree *query = queriesOnCluster->queryPropTree(xpath.str());
  773. if (!query)
  774. clusterState->setState("Not Found");
  775. else
  776. {
  777. int suspended = query->getPropInt("@suspended");
  778. const char* error = query->queryProp("@error");
  779. if (0==suspended)
  780. clusterState->setState("Available");
  781. else
  782. {
  783. clusterState->setState("Suspended");
  784. if (suspended<reporting)
  785. clusterState->setMixedNodeStates(true);
  786. }
  787. if (error && *error)
  788. clusterState->setErrors(error);
  789. }
  790. clusterStates.append(*clusterState.getClear());
  791. }
  792. void gatherQuerySetQueryDetails(IEspContext &context, IPropertyTree *query, IEspQuerySetQuery *queryInfo, const char *cluster, IPropertyTree *queriesOnCluster)
  793. {
  794. queryInfo->setId(query->queryProp("@id"));
  795. queryInfo->setName(query->queryProp("@name"));
  796. queryInfo->setDll(query->queryProp("@dll"));
  797. queryInfo->setWuid(query->queryProp("@wuid"));
  798. queryInfo->setSuspended(query->getPropBool("@suspended", false));
  799. if (query->hasProp("@memoryLimit"))
  800. {
  801. StringBuffer s;
  802. memoryLimitStringFromUInt64(s, query->getPropInt64("@memoryLimit"));
  803. queryInfo->setMemoryLimit(s);
  804. }
  805. if (query->hasProp("@timeLimit"))
  806. queryInfo->setTimeLimit(query->getPropInt("@timeLimit"));
  807. if (query->hasProp("@warnTimeLimit"))
  808. queryInfo->setWarnTimeLimit(query->getPropInt("@warnTimeLimit"));
  809. if (query->hasProp("@priority"))
  810. queryInfo->setPriority(getQueryPriorityName(query->getPropInt("@priority")));
  811. if (query->hasProp("@comment"))
  812. queryInfo->setComment(query->queryProp("@comment"));
  813. if (query->hasProp("@snapshot"))
  814. queryInfo->setSnapshot(query->queryProp("@snapshot"));
  815. double version = context.getClientVersion();
  816. if (version >= 1.46)
  817. {
  818. queryInfo->setPublishedBy(query->queryProp("@publishedBy"));
  819. queryInfo->setIsLibrary(query->getPropBool("@isLibrary"));
  820. }
  821. if (queriesOnCluster)
  822. {
  823. IArrayOf<IEspClusterQueryState> clusters;
  824. addClusterQueryStates(queriesOnCluster, cluster, query->queryProp("@id"), clusters, version);
  825. queryInfo->setClusters(clusters);
  826. }
  827. }
  828. void gatherQuerySetAliasDetails(IPropertyTree *alias, IEspQuerySetAlias *aliasInfo)
  829. {
  830. aliasInfo->setName(alias->queryProp("@name"));
  831. aliasInfo->setId(alias->queryProp("@id"));
  832. }
  833. void retrieveAllQuerysetDetails(IEspContext &context, IPropertyTree *registry, IArrayOf<IEspQuerySetQuery> &queries, IArrayOf<IEspQuerySetAlias> &aliases, const char *cluster=NULL, IPropertyTree *queriesOnCluster=NULL, const char *type=NULL, const char *value=NULL)
  834. {
  835. Owned<IPropertyTreeIterator> regQueries = registry->getElements("Query");
  836. ForEach(*regQueries)
  837. {
  838. IPropertyTree &query = regQueries->query();
  839. Owned<IEspQuerySetQuery> q = createQuerySetQuery();
  840. gatherQuerySetQueryDetails(context, &query, q, cluster, queriesOnCluster);
  841. if (isEmpty(cluster) || isEmpty(type) || isEmpty(value) || !strieq(type, "Status"))
  842. queries.append(*q.getClear());
  843. else
  844. {
  845. IArrayOf<IConstClusterQueryState>& cs = q->getClusters();
  846. ForEachItemIn(i, cs)
  847. {
  848. IConstClusterQueryState& c = cs.item(i);
  849. if (strieq(c.getCluster(), cluster) && (strieq(value, "All") || strieq(c.getState(), value)))
  850. {
  851. queries.append(*q.getClear());
  852. break;
  853. }
  854. }
  855. }
  856. }
  857. Owned<IPropertyTreeIterator> regAliases = registry->getElements("Alias");
  858. ForEach(*regAliases)
  859. {
  860. IPropertyTree &alias = regAliases->query();
  861. Owned<IEspQuerySetAlias> a = createQuerySetAlias();
  862. gatherQuerySetAliasDetails(&alias, a);
  863. aliases.append(*a.getClear());
  864. }
  865. }
  866. void retrieveQuerysetDetailsFromAlias(IEspContext &context, IPropertyTree *registry, const char *name, IArrayOf<IEspQuerySetQuery> &queries, IArrayOf<IEspQuerySetAlias> &aliases, const char *cluster, IPropertyTree *queriesOnCluster)
  867. {
  868. StringBuffer xpath;
  869. xpath.append("Alias[@name='").append(name).append("']");
  870. IPropertyTree *alias = registry->queryPropTree(xpath);
  871. if (!alias)
  872. {
  873. DBGLOG("Alias %s not found", name);
  874. return;
  875. }
  876. Owned<IEspQuerySetAlias> a = createQuerySetAlias();
  877. gatherQuerySetAliasDetails(alias, a);
  878. xpath.clear().append("Query[@id='").append(a->getId()).append("']");
  879. aliases.append(*a.getClear());
  880. IPropertyTree *query = registry->queryPropTree(xpath);
  881. if (!query)
  882. {
  883. DBGLOG("No matching Query %s found for Alias %s", a->getId(), name);
  884. return;
  885. }
  886. Owned<IEspQuerySetQuery> q = createQuerySetQuery();
  887. gatherQuerySetQueryDetails(context, query, q, cluster, queriesOnCluster);
  888. queries.append(*q.getClear());
  889. }
  890. void retrieveQuerysetDetailsFromQuery(IEspContext &context, IPropertyTree *registry, const char *value, const char *type, IArrayOf<IEspQuerySetQuery> &queries, IArrayOf<IEspQuerySetAlias> &aliases, const char *cluster=NULL, IPropertyTree *queriesOnCluster=NULL)
  891. {
  892. if (!strieq(type, "Id") && !strieq(type, "Name"))
  893. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Unrecognized queryset filter type %s", type);
  894. StringBuffer attributeName(type);
  895. StringBuffer xpath;
  896. xpath.clear().append("Query[@").append(attributeName.toLowerCase()).append("='").append(value).append("']");
  897. IPropertyTree *query = registry->queryPropTree(xpath);
  898. if (!query)
  899. {
  900. DBGLOG("No matching Query %s found for %s", value, type);
  901. return;
  902. }
  903. Owned<IEspQuerySetQuery> q = createQuerySetQuery();
  904. gatherQuerySetQueryDetails(context, query, q, cluster, queriesOnCluster);
  905. xpath.clear().append("Alias[@id='").append(q->getId()).append("']");
  906. queries.append(*q.getClear());
  907. Owned<IPropertyTreeIterator> regAliases = registry->getElements(xpath.str());
  908. ForEach(*regAliases)
  909. {
  910. IPropertyTree &alias = regAliases->query();
  911. Owned<IEspQuerySetAlias> a = createQuerySetAlias();
  912. gatherQuerySetAliasDetails(&alias, a);
  913. aliases.append(*a.getClear());
  914. }
  915. }
  916. void retrieveQuerysetDetails(IEspContext &context, IPropertyTree *registry, const char *type, const char *value, IArrayOf<IEspQuerySetQuery> &queries, IArrayOf<IEspQuerySetAlias> &aliases, const char *cluster=NULL, IPropertyTree *queriesOnCluster=NULL)
  917. {
  918. if (strieq(type, "All"))
  919. return retrieveAllQuerysetDetails(context, registry, queries, aliases, cluster, queriesOnCluster);
  920. if (!value || !*value)
  921. return;
  922. if (strieq(type, "Alias"))
  923. return retrieveQuerysetDetailsFromAlias(context, registry, value, queries, aliases, cluster, queriesOnCluster);
  924. if (strieq(type, "Status") && !isEmpty(cluster))
  925. return retrieveAllQuerysetDetails(context, registry, queries, aliases, cluster, queriesOnCluster, type, value);
  926. return retrieveQuerysetDetailsFromQuery(context, registry, value, type, queries, aliases, cluster, queriesOnCluster);
  927. }
  928. void retrieveQuerysetDetails(IEspContext &context, IArrayOf<IEspWUQuerySetDetail> &details, IPropertyTree *registry, const char *type, const char *value, const char *cluster=NULL, IPropertyTree *queriesOnCluster=NULL)
  929. {
  930. if (!registry)
  931. return;
  932. IArrayOf<IEspQuerySetQuery> queries;
  933. IArrayOf<IEspQuerySetAlias> aliases;
  934. retrieveQuerysetDetails(context, registry, type, value, queries, aliases, cluster, queriesOnCluster);
  935. Owned<IEspWUQuerySetDetail> detail = createWUQuerySetDetail();
  936. detail->setQuerySetName(registry->queryProp("@id"));
  937. detail->setQueries(queries);
  938. detail->setAliases(aliases);
  939. details.append(*detail.getClear());
  940. }
  941. void retrieveQuerysetDetails(IEspContext &context, IArrayOf<IEspWUQuerySetDetail> &details, const char *queryset, const char *type, const char *value, const char *cluster=NULL, IPropertyTree *queriesOnCluster=NULL)
  942. {
  943. if (!queryset || !*queryset)
  944. return;
  945. Owned<IPropertyTree> registry = getQueryRegistry(queryset, true);
  946. if (!registry)
  947. return;
  948. retrieveQuerysetDetails(context, details, registry, type, value, cluster, queriesOnCluster);
  949. }
  950. IPropertyTree* getQueriesOnCluster(const char *target, const char *queryset, bool checkAllNodes)
  951. {
  952. if (isEmpty(target))
  953. target = queryset;
  954. Owned<IConstWUClusterInfo> info = getTargetClusterInfo(target);
  955. if (!info)
  956. throw MakeStringException(ECLWATCH_CANNOT_RESOLVE_CLUSTER_NAME, "Cluster %s not found", target);
  957. if (queryset && *queryset && !strieq(target, queryset))
  958. throw MakeStringException(ECLWATCH_QUERYSET_NOT_ON_CLUSTER, "Target %s and QuerySet %s should match", target, queryset);
  959. if (info->getPlatform()!=RoxieCluster)
  960. return NULL;
  961. const SocketEndpointArray &eps = info->getRoxieServers();
  962. if (!eps.length())
  963. return NULL;
  964. try
  965. {
  966. Owned<ISocket> sock = ISocket::connect_timeout(eps.item(0), ROXIECONNECTIONTIMEOUT);
  967. if (checkAllNodes)
  968. return sendRoxieControlAllNodes(sock, "<control:queries/>", false, ROXIECONTROLQUERIESTIMEOUT);
  969. else
  970. return sendRoxieControlQuery(sock, "<control:queries/>", ROXIECONTROLQUERIESTIMEOUT);
  971. }
  972. catch(IException* e)
  973. {
  974. StringBuffer err;
  975. DBGLOG("Get exception in control:queries: %s", e->errorMessage(err.append(e->errorCode()).append(' ')).str());
  976. e->Release();
  977. return NULL;
  978. }
  979. }
  980. void retrieveQuerysetDetailsByCluster(IEspContext &context, IArrayOf<IEspWUQuerySetDetail> &details, const char *target, const char *queryset, const char *type, const char *value, bool checkAllNodes)
  981. {
  982. Owned<IPropertyTree> queriesOnCluster = getQueriesOnCluster(target, queryset, checkAllNodes);
  983. retrieveQuerysetDetails(context, details, target, type, value, target, queriesOnCluster);
  984. }
  985. void retrieveAllQuerysetDetails(IEspContext &context, IArrayOf<IEspWUQuerySetDetail> &details, const char *type, const char *value)
  986. {
  987. Owned<IPropertyTree> root = getQueryRegistryRoot();
  988. if (!root)
  989. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "QuerySet Registry not found");
  990. Owned<IPropertyTreeIterator> querysets = root->getElements("QuerySet");
  991. ForEach(*querysets)
  992. retrieveQuerysetDetails(context, details, &querysets->query(), type, value);
  993. }
  994. bool CWsWorkunitsEx::onWUQuerysetDetails(IEspContext &context, IEspWUQuerySetDetailsRequest & req, IEspWUQuerySetDetailsResponse & resp)
  995. {
  996. resp.setQuerySetName(req.getQuerySetName());
  997. double version = context.getClientVersion();
  998. if (version > 1.36)
  999. {
  1000. Owned<IPropertyTree> queryRegistry = getQueryRegistry(req.getQuerySetName(), false);
  1001. resp.setClusterName(req.getClusterName());
  1002. resp.setFilter(req.getFilter());
  1003. resp.setFilterType(req.getFilterType());
  1004. }
  1005. Owned<IPropertyTree> registry = getQueryRegistry(req.getQuerySetName(), true);
  1006. if (!registry)
  1007. return false;
  1008. IArrayOf<IEspQuerySetQuery> respQueries;
  1009. IArrayOf<IEspQuerySetAlias> respAliases;
  1010. if (isEmpty(req.getClusterName()) || isEmpty(req.getFilterTypeAsString()) || !strieq(req.getFilterTypeAsString(), "Status") || isEmpty(req.getFilter()))
  1011. {
  1012. const char* cluster = req.getClusterName();
  1013. if (isEmpty(cluster))
  1014. cluster = req.getQuerySetName();
  1015. Owned<IPropertyTree> queriesOnCluster = getQueriesOnCluster(cluster, req.getQuerySetName(), req.getCheckAllNodes());
  1016. retrieveQuerysetDetails(context, registry, req.getFilterTypeAsString(), req.getFilter(), respQueries, respAliases, cluster, queriesOnCluster);
  1017. resp.setQuerysetQueries(respQueries);
  1018. resp.setQuerysetAliases(respAliases);
  1019. }
  1020. else
  1021. {
  1022. IArrayOf<IEspWUQuerySetDetail> respDetails;
  1023. retrieveQuerysetDetailsByCluster(context, respDetails, req.getClusterName(), req.getQuerySetName(), req.getFilterTypeAsString(), req.getFilter(), false);
  1024. if (respDetails.ordinality())
  1025. {
  1026. IEspWUQuerySetDetail& detail = respDetails.item(0);
  1027. resp.setQuerysetQueries(detail.getQueries());
  1028. resp.setQuerysetAliases(detail.getAliases());
  1029. }
  1030. }
  1031. return true;
  1032. }
  1033. bool CWsWorkunitsEx::onWUMultiQuerysetDetails(IEspContext &context, IEspWUMultiQuerySetDetailsRequest & req, IEspWUMultiQuerySetDetailsResponse & resp)
  1034. {
  1035. IArrayOf<IEspWUQuerySetDetail> respDetails;
  1036. if (notEmpty(req.getClusterName()))
  1037. retrieveQuerysetDetailsByCluster(context, respDetails, req.getClusterName(), req.getQuerySetName(), req.getFilterTypeAsString(), req.getFilter(), req.getCheckAllNodes());
  1038. else if (notEmpty(req.getQuerySetName()))
  1039. retrieveQuerysetDetails(context, respDetails, req.getQuerySetName(), req.getFilterTypeAsString(), req.getFilter());
  1040. else
  1041. retrieveAllQuerysetDetails(context, respDetails, req.getFilterTypeAsString(), req.getFilter());
  1042. resp.setQuerysets(respDetails);
  1043. return true;
  1044. }
  1045. bool addWUQSQueryFilter(WUQuerySortField *filters, unsigned short &count, MemoryBuffer &buff, const char* value, WUQuerySortField name)
  1046. {
  1047. if (isEmpty(value))
  1048. return false;
  1049. filters[count++] = name;
  1050. buff.append(value);
  1051. return true;
  1052. }
  1053. bool addWUQSQueryFilterInt(WUQuerySortField *filters, unsigned short &count, MemoryBuffer &buff, int value, WUQuerySortField name)
  1054. {
  1055. VStringBuffer vBuf("%d", value);
  1056. filters[count++] = name;
  1057. buff.append(vBuf.str());
  1058. return true;
  1059. }
  1060. bool addWUQSQueryFilterInt64(WUQuerySortField *filters, unsigned short &count, MemoryBuffer &buff, __int64 value, WUQuerySortField name)
  1061. {
  1062. VStringBuffer vBuf("%" I64F "d", value);
  1063. filters[count++] = name;
  1064. buff.append(vBuf.str());
  1065. return true;
  1066. }
  1067. unsigned CWsWorkunitsEx::getGraphIdsByQueryId(const char *target, const char *queryId, StringArray& graphIds)
  1068. {
  1069. if (!target || !*target)
  1070. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Target name required");
  1071. if (!queryId || !*queryId)
  1072. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Query Id required");
  1073. Owned<IConstWUClusterInfo> info = getTargetClusterInfo(target);
  1074. if (!info || (info->getPlatform()!=RoxieCluster)) //Only roxie query has query graph.
  1075. return 0;
  1076. const SocketEndpointArray &eps = info->getRoxieServers();
  1077. if (eps.empty())
  1078. return 0;
  1079. VStringBuffer xpath("<control:querystats><Query id='%s'/></control:querystats>", queryId);
  1080. Owned<ISocket> sock = ISocket::connect_timeout(eps.item(0), ROXIECONNECTIONTIMEOUT);
  1081. Owned<IPropertyTree> querystats = sendRoxieControlQuery(sock, xpath.str(), ROXIECONTROLQUERYTIMEOUT);
  1082. if (!querystats)
  1083. return 0;
  1084. Owned<IPropertyTreeIterator> graphs = querystats->getElements("Endpoint/Query/Graph");
  1085. ForEach(*graphs)
  1086. {
  1087. IPropertyTree &graph = graphs->query();
  1088. const char* graphId = graph.queryProp("@id");
  1089. if (graphId && *graphId)
  1090. graphIds.append(graphId);
  1091. }
  1092. return graphIds.length();
  1093. }
  1094. //This method is thread safe because a query belongs to a single queryset. The method may be called by different threads.
  1095. //Since one thread is for one queryset and a query only belongs to a single queryset, it is impossible for different threads
  1096. //to update the same query object.
  1097. void CWsWorkunitsEx::checkAndSetClusterQueryState(IEspContext &context, const char* cluster, const char* querySetId, IArrayOf<IEspQuerySetQuery>& queries, bool checkAllNodes)
  1098. {
  1099. try
  1100. {
  1101. double version = context.getClientVersion();
  1102. if (isEmpty(cluster))
  1103. cluster = querySetId;
  1104. Owned<IPropertyTree> queriesOnCluster = getQueriesOnCluster(cluster, querySetId, checkAllNodes);
  1105. if (!queriesOnCluster)
  1106. {
  1107. DBGLOG("getQueriesOnCluster() returns NULL for cluster<%s> and querySetId<%s>", cluster, querySetId);
  1108. return;
  1109. }
  1110. ForEachItemIn(i, queries)
  1111. {
  1112. IEspQuerySetQuery& query = queries.item(i);
  1113. const char* queryId = query.getId();
  1114. const char* querySetId0 = query.getQuerySetId();
  1115. if (!queryId || !querySetId0 || !strieq(querySetId0, querySetId))
  1116. continue;
  1117. IArrayOf<IEspClusterQueryState> clusters;
  1118. addClusterQueryStates(queriesOnCluster, cluster, queryId, clusters, version);
  1119. query.setClusters(clusters);
  1120. }
  1121. }
  1122. catch(IException *e)
  1123. {
  1124. EXCLOG(e, "CWsWorkunitsEx::checkAndSetClusterQueryState: Failed to read Query State On Cluster");
  1125. e->Release();
  1126. }
  1127. }
  1128. void CWsWorkunitsEx::checkAndSetClusterQueryState(IEspContext &context, const char* cluster, StringArray& querySetIds, IArrayOf<IEspQuerySetQuery>& queries, bool checkAllNodes)
  1129. {
  1130. UnsignedArray threadHandles;
  1131. ForEachItemIn(i, querySetIds)
  1132. {
  1133. const char* querySetId = querySetIds.item(i);
  1134. if(!querySetId || !*querySetId)
  1135. continue;
  1136. Owned<CClusterQueryStateParam> threadReq = new CClusterQueryStateParam(this, context, cluster, querySetId, queries, checkAllNodes);
  1137. PooledThreadHandle handle = clusterQueryStatePool->start( threadReq.getClear() );
  1138. threadHandles.append(handle);
  1139. }
  1140. //block for worker threads to finish, if necessary and then collect results
  1141. //Not use joinAll() because multiple threads may call this method. Each call uses the pool to create
  1142. //its own threads of checking query state. Each call should only join the ones created by that call.
  1143. ForEachItemIn(ii, threadHandles)
  1144. clusterQueryStatePool->join(threadHandles.item(ii));
  1145. }
  1146. bool CWsWorkunitsEx::onWUListQueries(IEspContext &context, IEspWUListQueriesRequest & req, IEspWUListQueriesResponse & resp)
  1147. {
  1148. bool descending = req.getDescending();
  1149. const char *sortBy = req.getSortby();
  1150. WUQuerySortField sortOrder[2] = {WUQSFId, WUQSFterm};
  1151. if(notEmpty(sortBy))
  1152. {
  1153. if (strieq(sortBy, "Name"))
  1154. sortOrder[0] = WUQSFname;
  1155. else if (strieq(sortBy, "WUID"))
  1156. sortOrder[0] = WUQSFwuid;
  1157. else if (strieq(sortBy, "DLL"))
  1158. sortOrder[0] = WUQSFdll;
  1159. else if (strieq(sortBy, "Activated"))
  1160. sortOrder[0] = WUQSFActivited;
  1161. else if (strieq(sortBy, "MemoryLimit"))
  1162. sortOrder[0] = (WUQuerySortField) (WUQSFmemoryLimit | WUQSFnumeric);
  1163. else if (strieq(sortBy, "TimeLimit"))
  1164. sortOrder[0] = (WUQuerySortField) (WUQSFtimeLimit | WUQSFnumeric);
  1165. else if (strieq(sortBy, "WarnTimeLimit"))
  1166. sortOrder[0] = (WUQuerySortField) (WUQSFwarnTimeLimit | WUQSFnumeric);
  1167. else if (strieq(sortBy, "Priority"))
  1168. sortOrder[0] = (WUQuerySortField) (WUQSFpriority | WUQSFnumeric);
  1169. else if (strieq(sortBy, "PublishedBy"))
  1170. sortOrder[0] = WUQSFPublishedBy;
  1171. else if (strieq(sortBy, "QuerySetId"))
  1172. sortOrder[0] = WUQSFQuerySet;
  1173. else
  1174. sortOrder[0] = WUQSFId;
  1175. sortOrder[0] = (WUQuerySortField) (sortOrder[0] | WUQSFnocase);
  1176. if (descending)
  1177. sortOrder[0] = (WUQuerySortField) (sortOrder[0] | WUQSFreverse);
  1178. }
  1179. WUQuerySortField filters[16];
  1180. unsigned short filterCount = 0;
  1181. MemoryBuffer filterBuf;
  1182. const char* clusterReq = req.getClusterName();
  1183. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getQuerySetName(), WUQSFQuerySet);
  1184. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getQueryID(), (WUQuerySortField) (WUQSFId | WUQSFwild | WUSFnocase));
  1185. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getQueryName(), (WUQuerySortField) (WUQSFname | WUQSFwild | WUSFnocase));
  1186. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getWUID(), (WUQuerySortField) (WUQSFwuid | WUSFnocase));
  1187. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getLibraryName(), (WUQuerySortField) (WUQSFLibrary | WUQSFnocase));
  1188. addWUQSQueryFilter(filters, filterCount, filterBuf, req.getPublishedBy(), (WUQuerySortField) (WUQSFPublishedBy | WUQSFwild | WUSFnocase));
  1189. if (!req.getMemoryLimitLow_isNull())
  1190. addWUQSQueryFilterInt64(filters, filterCount, filterBuf, req.getMemoryLimitLow(), (WUQuerySortField) (WUQSFmemoryLimit | WUQSFnumeric));
  1191. if (!req.getMemoryLimitHigh_isNull())
  1192. addWUQSQueryFilterInt64(filters, filterCount, filterBuf, req.getMemoryLimitHigh(), (WUQuerySortField) (WUQSFmemoryLimitHi | WUQSFnumeric));
  1193. if (!req.getTimeLimitLow_isNull())
  1194. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getTimeLimitLow(), (WUQuerySortField) (WUQSFtimeLimit | WUQSFnumeric));
  1195. if (!req.getTimeLimitHigh_isNull())
  1196. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getTimeLimitHigh(), (WUQuerySortField) (WUQSFtimeLimitHi | WUQSFnumeric));
  1197. if (!req.getWarnTimeLimitLow_isNull())
  1198. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getWarnTimeLimitLow(), (WUQuerySortField) (WUQSFwarnTimeLimit | WUQSFnumeric));
  1199. if (!req.getWarnTimeLimitHigh_isNull())
  1200. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getWarnTimeLimitHigh(), (WUQuerySortField) (WUQSFwarnTimeLimitHi | WUQSFnumeric));
  1201. if (!req.getPriorityLow_isNull())
  1202. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getPriorityLow(), (WUQuerySortField) (WUQSFpriority | WUQSFnumeric));
  1203. if (!req.getPriorityHigh_isNull())
  1204. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getPriorityHigh(), (WUQuerySortField) (WUQSFpriorityHi | WUQSFnumeric));
  1205. if (!req.getActivated_isNull())
  1206. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getActivated(), (WUQuerySortField) (WUQSFActivited | WUQSFnumeric));
  1207. if (!req.getSuspendedByUser_isNull())
  1208. addWUQSQueryFilterInt(filters, filterCount, filterBuf, req.getSuspendedByUser(), (WUQuerySortField) (WUQSFSuspendedByUser | WUQSFnumeric));
  1209. filters[filterCount] = WUQSFterm;
  1210. unsigned numberOfQueries = 0;
  1211. unsigned pageSize = req.getPageSize();
  1212. unsigned pageStartFrom = req.getPageStartFrom();
  1213. if(pageSize < 1)
  1214. pageSize = 100;
  1215. __int64 cacheHint = 0;
  1216. if (!req.getCacheHint_isNull())
  1217. cacheHint = req.getCacheHint();
  1218. Owned<MapStringTo<bool> > queriesUsingFileMap;
  1219. const char *lfn = req.getFileName();
  1220. if (lfn && *lfn)
  1221. {
  1222. queriesUsingFileMap.setown(new MapStringTo<bool>());
  1223. StringAttr dummy;
  1224. Owned<IPropertyTreeIterator> queriesUsingFile = filesInUse.findQueriesUsingFile(clusterReq, lfn, dummy);
  1225. ForEach (*queriesUsingFile)
  1226. {
  1227. IPropertyTree &queryUsingFile = queriesUsingFile->query();
  1228. const char *queryTarget = queryUsingFile.queryProp("@target");
  1229. const char *queryId = queryUsingFile.queryProp("@id");
  1230. if (queryTarget && *queryTarget && queryId && *queryId)
  1231. {
  1232. VStringBuffer targetQuery("%s/%s", queryTarget, queryId);
  1233. queriesUsingFileMap->setValue(targetQuery, true);
  1234. }
  1235. }
  1236. }
  1237. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  1238. Owned<IConstQuerySetQueryIterator> it = factory->getQuerySetQueriesSorted(sortOrder, filters, filterBuf.bufferBase(), pageStartFrom, pageSize, &cacheHint, &numberOfQueries, queriesUsingFileMap);
  1239. resp.setCacheHint(cacheHint);
  1240. StringArray querySetIds;
  1241. IArrayOf<IEspQuerySetQuery> queries;
  1242. double version = context.getClientVersion();
  1243. ForEach(*it)
  1244. {
  1245. IPropertyTree &query=it->query();
  1246. const char *queryId = query.queryProp("@id");
  1247. const char *queryTarget = query.queryProp("@querySetId");
  1248. Owned<IEspQuerySetQuery> q = createQuerySetQuery();
  1249. q->setId(queryId);
  1250. q->setQuerySetId(queryTarget);
  1251. q->setName(query.queryProp("@name"));
  1252. q->setDll(query.queryProp("@dll"));
  1253. q->setWuid(query.queryProp("@wuid"));
  1254. q->setActivated(query.getPropBool("@activated", false));
  1255. q->setSuspended(query.getPropBool("@suspended", false));
  1256. if (query.hasProp("@memoryLimit"))
  1257. {
  1258. StringBuffer s;
  1259. memoryLimitStringFromUInt64(s, query.getPropInt64("@memoryLimit"));
  1260. q->setMemoryLimit(s);
  1261. }
  1262. if (query.hasProp("@timeLimit"))
  1263. q->setTimeLimit(query.getPropInt("@timeLimit"));
  1264. if (query.hasProp("@warnTimeLimit"))
  1265. q->setWarnTimeLimit(query.getPropInt("@warnTimeLimit"));
  1266. if (query.hasProp("@priority"))
  1267. q->setPriority(getQueryPriorityName(query.getPropInt("@priority")));
  1268. if (query.hasProp("@comment"))
  1269. q->setComment(query.queryProp("@comment"));
  1270. if (version >= 1.46)
  1271. {
  1272. q->setPublishedBy(query.queryProp("@publishedBy"));
  1273. q->setIsLibrary(query.getPropBool("@isLibrary"));
  1274. }
  1275. if (!querySetIds.contains(queryTarget))
  1276. querySetIds.append(queryTarget);
  1277. queries.append(*q.getClear());
  1278. }
  1279. checkAndSetClusterQueryState(context, clusterReq, querySetIds, queries, req.getCheckAllNodes());
  1280. resp.setQuerysetQueries(queries);
  1281. resp.setNumberOfQueries(numberOfQueries);
  1282. return true;
  1283. }
  1284. bool CWsWorkunitsEx::onWUListQueriesUsingFile(IEspContext &context, IEspWUListQueriesUsingFileRequest &req, IEspWUListQueriesUsingFileResponse &resp)
  1285. {
  1286. const char *target = req.getTarget();
  1287. const char *process = req.getProcess();
  1288. StringBuffer lfn(req.getFileName());
  1289. resp.setFileName(lfn.toLowerCase());
  1290. resp.setProcess(process);
  1291. StringArray targets;
  1292. if (target && *target)
  1293. targets.append(target);
  1294. else // if (process && *process)
  1295. {
  1296. SCMStringBuffer targetStr;
  1297. Owned<IStringIterator> targetClusters = getTargetClusters("RoxieCluster", process);
  1298. ForEach(*targetClusters)
  1299. targets.append(targetClusters->str(targetStr).str());
  1300. }
  1301. IArrayOf<IEspTargetQueriesUsingFile> respTargets;
  1302. ForEachItemIn(i, targets)
  1303. {
  1304. target = targets.item(i);
  1305. Owned<IEspTargetQueriesUsingFile> respTarget = createTargetQueriesUsingFile();
  1306. respTarget->setTarget(target);
  1307. StringAttr pmid;
  1308. Owned<IPropertyTreeIterator> queries = filesInUse.findQueriesUsingFile(target, lfn, pmid);
  1309. if (!pmid.isEmpty())
  1310. respTarget->setPackageMap(pmid);
  1311. if (queries)
  1312. {
  1313. IArrayOf<IEspQueryUsingFile> respQueries;
  1314. ForEach(*queries)
  1315. {
  1316. IPropertyTree &query = queries->query();
  1317. Owned<IEspQueryUsingFile> q = createQueryUsingFile();
  1318. q->setId(query.queryProp("@id"));
  1319. VStringBuffer xpath("File[@lfn='%s']/@pkgid", lfn.str());
  1320. if (query.hasProp(xpath))
  1321. q->setPackage(query.queryProp(xpath));
  1322. respQueries.append(*q.getClear());
  1323. }
  1324. respTarget->setQueries(respQueries);
  1325. }
  1326. respTargets.append(*respTarget.getClear());
  1327. }
  1328. resp.setTargets(respTargets);
  1329. return true;
  1330. }
  1331. bool CWsWorkunitsEx::onWUQueryFiles(IEspContext &context, IEspWUQueryFilesRequest &req, IEspWUQueryFilesResponse &resp)
  1332. {
  1333. const char *target = req.getTarget();
  1334. const char *query = req.getQueryId();
  1335. if (!target || !*target)
  1336. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Target not specified");
  1337. if (!isValidCluster(target))
  1338. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid target name: %s", target);
  1339. if (!query || !*query)
  1340. throw MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "Query not specified");
  1341. Owned<IPropertyTree> registeredQuery = resolveQueryAlias(target, query, true);
  1342. if (!registeredQuery)
  1343. throw MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "Query not found");
  1344. StringAttr queryid(registeredQuery->queryProp("@id"));
  1345. registeredQuery.clear();
  1346. Owned<IPropertyTree> tree = filesInUse.getTree();
  1347. VStringBuffer xpath("%s/Query[@id='%s']", target, queryid.get());
  1348. IPropertyTree *queryTree = tree->queryPropTree(xpath);
  1349. if (!queryTree)
  1350. throw MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "Query not found in file cache (%s)", xpath.str());
  1351. IArrayOf<IEspFileUsedByQuery> referencedFiles;
  1352. Owned<IPropertyTreeIterator> files = queryTree->getElements("File");
  1353. ForEach(*files)
  1354. {
  1355. IPropertyTree &file = files->query();
  1356. if (file.getPropBool("@super", 0))
  1357. continue;
  1358. Owned<IEspFileUsedByQuery> respFile = createFileUsedByQuery();
  1359. respFile->setFileName(file.queryProp("@lfn"));
  1360. respFile->setFileSize(file.getPropInt64("@size"));
  1361. respFile->setNumberOfParts(file.getPropInt("@numparts"));
  1362. referencedFiles.append(*respFile.getClear());
  1363. }
  1364. resp.setFiles(referencedFiles);
  1365. return true;
  1366. }
  1367. bool CWsWorkunitsEx::onWUQueryDetails(IEspContext &context, IEspWUQueryDetailsRequest & req, IEspWUQueryDetailsResponse & resp)
  1368. {
  1369. const char* querySet = req.getQuerySet();
  1370. const char* queryId = req.getQueryId();
  1371. bool includeStateOnClusters = req.getIncludeStateOnClusters();
  1372. if (!querySet || !*querySet)
  1373. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "QuerySet not specified");
  1374. if (!queryId || !*queryId)
  1375. throw MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "QueryId not specified");
  1376. resp.setQueryId(queryId);
  1377. resp.setQuerySet(querySet);
  1378. Owned<IPropertyTree> queryRegistry = getQueryRegistry(querySet, false);
  1379. StringBuffer xpath;
  1380. xpath.clear().append("Query[@id='").append(queryId).append("']");
  1381. IPropertyTree *query = queryRegistry->queryPropTree(xpath.str());
  1382. if (!query)
  1383. {
  1384. DBGLOG("No matching Query");
  1385. return false;
  1386. }
  1387. const char* queryName = query->queryProp("@name");
  1388. const char* wuid = query->queryProp("@wuid");
  1389. resp.setQueryName(queryName);
  1390. resp.setWuid(wuid);
  1391. resp.setDll(query->queryProp("@dll"));
  1392. resp.setPublishedBy(query->queryProp("@publishedBy"));
  1393. resp.setSuspended(query->getPropBool("@suspended", false));
  1394. resp.setSuspendedBy(query->queryProp("@suspendedBy"));
  1395. resp.setComment(query->queryProp("@comment"));
  1396. double version = context.getClientVersion();
  1397. if (version >= 1.46)
  1398. {
  1399. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  1400. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid);
  1401. if(!cw)
  1402. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT,"Cannot open workunit %s.",wuid);
  1403. if (query->hasProp("@priority"))
  1404. resp.setPriority(getQueryPriorityName(query->getPropInt("@priority")));
  1405. resp.setIsLibrary(query->getPropBool("@isLibrary"));
  1406. SCMStringBuffer s;
  1407. resp.setWUSnapShot(cw->getSnapshot(s).str()); //Label
  1408. Owned<IConstWUStatistic> whenCompiled = cw->getStatistic(NULL, NULL, StWhenCompiled);
  1409. if (whenCompiled)
  1410. {
  1411. whenCompiled->getFormattedValue(s);
  1412. resp.setCompileTime(s.str());
  1413. }
  1414. StringArray libUsed, graphIds;
  1415. Owned<IConstWULibraryIterator> libs = &cw->getLibraries();
  1416. ForEach(*libs)
  1417. libUsed.append(libs->query().getName(s).str());
  1418. if (libUsed.length())
  1419. resp.setLibrariesUsed(libUsed);
  1420. unsigned numGraphIds = getGraphIdsByQueryId(querySet, queryId, graphIds);
  1421. resp.setCountGraphs(numGraphIds);
  1422. if (numGraphIds > 0)
  1423. resp.setGraphIds(graphIds);
  1424. }
  1425. StringArray logicalFiles;
  1426. IArrayOf<IEspQuerySuperFile> superFiles;
  1427. getQueryFiles(context, wuid, queryId, querySet, logicalFiles, req.getIncludeSuperFiles() ? &superFiles : NULL);
  1428. if (logicalFiles.length())
  1429. resp.setLogicalFiles(logicalFiles);
  1430. if (superFiles.length())
  1431. resp.setSuperFiles(superFiles);
  1432. if (version >= 1.42)
  1433. {
  1434. xpath.clear().appendf("Alias[@id='%s']", queryId);
  1435. IPropertyTree *alias = queryRegistry->queryPropTree(xpath.str());
  1436. if (!alias)
  1437. resp.setActivated(false);
  1438. else
  1439. resp.setActivated(true);
  1440. }
  1441. if (includeStateOnClusters && (version >= 1.43))
  1442. {
  1443. Owned<IPropertyTree> queriesOnCluster = getQueriesOnCluster(querySet, querySet, req.getCheckAllNodes());
  1444. if (queriesOnCluster)
  1445. {
  1446. IArrayOf<IEspClusterQueryState> clusterStates;
  1447. addClusterQueryStates(queriesOnCluster, querySet, queryId, clusterStates, version);
  1448. resp.setClusters(clusterStates);
  1449. }
  1450. }
  1451. if (version >= 1.50)
  1452. {
  1453. WsWuInfo winfo(context, wuid);
  1454. resp.setResourceURLCount(winfo.getResourceURLCount());
  1455. }
  1456. if (req.getIncludeWsEclAddresses())
  1457. {
  1458. StringArray wseclAddresses;
  1459. Owned<IEnvironmentFactory> factory = getEnvironmentFactory();
  1460. Owned<IConstEnvironment> env = factory->openEnvironment();
  1461. if (env)
  1462. {
  1463. Owned<IPropertyTree> root = &env->getPTree();
  1464. Owned<IPropertyTreeIterator> services = root->getElements("Software/EspService[Properties/@type='ws_ecl']");
  1465. StringArray serviceNames;
  1466. VStringBuffer xpath("Target[@name='%s']", querySet);
  1467. ForEach(*services)
  1468. {
  1469. IPropertyTree &service = services->query();
  1470. if (!service.hasProp("Target") || service.hasProp(xpath))
  1471. serviceNames.append(service.queryProp("@name"));
  1472. }
  1473. Owned<IPropertyTreeIterator> processes = root->getElements("Software/EspProcess");
  1474. ForEach(*processes)
  1475. {
  1476. StringArray netAddrs;
  1477. IPropertyTree &process = processes->query();
  1478. Owned<IPropertyTreeIterator> instances = process.getElements("Instance");
  1479. ForEach(*instances)
  1480. {
  1481. IPropertyTree &instance = instances->query();
  1482. const char *netAddr = instance.queryProp("@netAddress");
  1483. if (!netAddr || !*netAddr)
  1484. continue;
  1485. if (streq(netAddr, "."))
  1486. netAddrs.appendUniq(envLocalAddress); //not necessarily local to this server
  1487. else
  1488. netAddrs.appendUniq(netAddr);
  1489. }
  1490. Owned<IPropertyTreeIterator> bindings = process.getElements("EspBinding");
  1491. ForEach(*bindings)
  1492. {
  1493. IPropertyTree &binding = bindings->query();
  1494. const char *srvName = binding.queryProp("@service");
  1495. if (!serviceNames.contains(srvName))
  1496. continue;
  1497. const char *port = binding.queryProp("@port"); //should always be an integer, but we're just concatenating strings
  1498. if (!port || !*port)
  1499. continue;
  1500. ForEachItemIn(i, netAddrs)
  1501. {
  1502. VStringBuffer wseclAddr("%s:%s", netAddrs.item(i), port);
  1503. wseclAddresses.append(wseclAddr);
  1504. }
  1505. }
  1506. }
  1507. }
  1508. resp.setWsEclAddresses(wseclAddresses);
  1509. }
  1510. return true;
  1511. }
  1512. int EspQuerySuperFileCompareFunc(IInterface * const *i1, IInterface * const *i2)
  1513. {
  1514. if (!i1 || !*i1 || !i2 || !*i2)
  1515. return 0;
  1516. IEspQuerySuperFile *sf1 = QUERYINTERFACE(*i1, IEspQuerySuperFile);
  1517. IEspQuerySuperFile *sf2 = QUERYINTERFACE(*i2, IEspQuerySuperFile);
  1518. if (!sf1 || !sf2)
  1519. return 0;
  1520. const char *name1 = sf1->getName();
  1521. const char *name2 = sf2->getName();
  1522. if (!name1 || !name2)
  1523. return 0;
  1524. return strcmp(name1, name2);
  1525. }
  1526. IReferencedFile* CWsWorkunitsEx::getReferencedFileByName(const char* name, IReferencedFileList* wufiles)
  1527. {
  1528. Owned<IReferencedFileIterator> refFileItr = wufiles->getFiles();
  1529. ForEach(*refFileItr)
  1530. {
  1531. IReferencedFile& rf = refFileItr->query();
  1532. const char* lfn = rf.getLogicalName();
  1533. if (lfn && strieq(lfn, name))
  1534. return &rf;
  1535. }
  1536. return NULL;
  1537. }
  1538. void CWsWorkunitsEx::readSuperFiles(IEspContext &context, IReferencedFile* rf, const char* fileName, IReferencedFileList* wufiles, IArrayOf<IEspQuerySuperFile>* files)
  1539. {
  1540. double version = context.getClientVersion();
  1541. StringArray subFiles;
  1542. IArrayOf<IEspQuerySuperFile> superFiles;
  1543. const StringArray& subFileNames = rf->getSubFileNames();
  1544. ForEachItemIn(i, subFileNames)
  1545. {
  1546. const char* name = subFileNames.item(i);
  1547. if (!name || !*name)
  1548. continue;
  1549. IReferencedFile* pRF = getReferencedFileByName(name, wufiles);
  1550. if (!pRF)
  1551. continue;
  1552. if (!(pRF->getFlags() & RefFileSuper))
  1553. {
  1554. subFiles.append(name);
  1555. }
  1556. else if (version >= 1.57)
  1557. {
  1558. readSuperFiles(context, pRF, name, wufiles, &superFiles);
  1559. }
  1560. }
  1561. Owned<IEspQuerySuperFile> newSuperFile = createQuerySuperFile();
  1562. newSuperFile->setName(fileName);
  1563. if (subFiles.length())
  1564. {
  1565. subFiles.sortAscii();
  1566. newSuperFile->setSubFiles(subFiles);
  1567. }
  1568. if ((version >= 1.57) && superFiles.length())
  1569. {
  1570. superFiles.sort(EspQuerySuperFileCompareFunc);
  1571. newSuperFile->setSuperFiles(superFiles);
  1572. }
  1573. files->append(*newSuperFile.getClear());
  1574. }
  1575. bool CWsWorkunitsEx::getQueryFiles(IEspContext &context, const char* wuid, const char* query, const char* target, StringArray& logicalFiles, IArrayOf<IEspQuerySuperFile> *respSuperFiles)
  1576. {
  1577. try
  1578. {
  1579. Owned<IConstWUClusterInfo> info = getTargetClusterInfo(target);
  1580. if (!info || (info->getPlatform()!=RoxieCluster))
  1581. return false;
  1582. SCMStringBuffer process;
  1583. info->getRoxieProcess(process);
  1584. if (!process.length())
  1585. return false;
  1586. Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
  1587. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid);
  1588. if (!cw)
  1589. return false;
  1590. StringArray superFileNames;
  1591. Owned<IHpccPackageSet> ps = createPackageSet(process.str());
  1592. Owned<IReferencedFileList> wufiles = createReferencedFileList(context.queryUserId(),
  1593. context.queryPassword(), true, true);
  1594. wufiles->addFilesFromQuery(cw, (ps) ? ps->queryActiveMap(target) : NULL, query);
  1595. wufiles->resolveFiles(process.str(), NULL, NULL, NULL, true, true, true, true);
  1596. Owned<IReferencedFileIterator> refFileItr = wufiles->getFiles();
  1597. ForEach(*refFileItr)
  1598. {
  1599. IReferencedFile &rf = refFileItr->query();
  1600. const char *lfn = rf.getLogicalName();
  1601. if (lfn && *lfn)
  1602. {
  1603. logicalFiles.append(lfn);
  1604. if (respSuperFiles && (rf.getFlags() & RefFileSuper))
  1605. readSuperFiles(context, &rf, lfn, wufiles, respSuperFiles);
  1606. }
  1607. }
  1608. logicalFiles.sortAscii();
  1609. if (respSuperFiles)
  1610. respSuperFiles->sort(EspQuerySuperFileCompareFunc);
  1611. return true;
  1612. }
  1613. catch(IMultiException *me)
  1614. {
  1615. StringBuffer err;
  1616. DBGLOG("ERROR control:getQueryXrefInfo roxie query info %s", me->errorMessage(err.append(me->errorCode()).append(' ')).str());
  1617. me->Release();
  1618. return false;
  1619. }
  1620. catch(IException *e)
  1621. {
  1622. StringBuffer err;
  1623. DBGLOG("ERROR control:getQueryXrefInfo roxie query info %s", e->errorMessage(err.append(e->errorCode()).append(' ')).str());
  1624. e->Release();
  1625. return false;
  1626. }
  1627. }
  1628. inline void verifyQueryActionAllowsWild(bool &allowWildChecked, CQuerySetQueryActionTypes action)
  1629. {
  1630. if (allowWildChecked)
  1631. return;
  1632. switch (action)
  1633. {
  1634. case CQuerySetQueryActionTypes_ToggleSuspend:
  1635. throw MakeStringException(ECLWATCH_INVALID_ACTION, "Wildcards not supported for toggling suspended state");
  1636. case CQuerySetQueryActionTypes_Activate:
  1637. throw MakeStringException(ECLWATCH_INVALID_ACTION, "Wildcards not supported for Activating queries");
  1638. }
  1639. allowWildChecked=true;
  1640. }
  1641. inline bool verifyQueryActionAllowsAlias(CQuerySetQueryActionTypes action)
  1642. {
  1643. return (action!=CQuerySetQueryActionTypes_Activate && action!=CQuerySetQueryActionTypes_RemoveAllAliases);
  1644. }
  1645. void expandQueryActionTargetList(IProperties *queryIds, IPropertyTree *queryset, IArrayOf<IConstQuerySetQueryActionItem> &items, CQuerySetQueryActionTypes action)
  1646. {
  1647. bool allowWildChecked=false;
  1648. ForEachItemIn(i, items)
  1649. {
  1650. const char *itemId = items.item(i).getQueryId();
  1651. if (!isWildString(itemId))
  1652. {
  1653. bool suspendedByUser = false;
  1654. const char *itemSuspendState = items.item(i).getClientState().getSuspended();
  1655. if (itemSuspendState && (strieq(itemSuspendState, "By User") || strieq(itemSuspendState, "1")))
  1656. suspendedByUser = true;
  1657. if (!verifyQueryActionAllowsAlias(action))
  1658. queryIds->setProp(itemId, suspendedByUser);
  1659. else
  1660. {
  1661. Owned<IPropertyTree> query = resolveQueryAlias(queryset, itemId);
  1662. if (query)
  1663. {
  1664. const char *id = query->queryProp("@id");
  1665. if (id && *id)
  1666. queryIds->setProp(id, suspendedByUser);
  1667. }
  1668. }
  1669. }
  1670. else
  1671. {
  1672. verifyQueryActionAllowsWild(allowWildChecked, action);
  1673. if (verifyQueryActionAllowsAlias(action))
  1674. {
  1675. Owned<IPropertyTreeIterator> active = queryset->getElements("Alias");
  1676. ForEach(*active)
  1677. {
  1678. const char *name = active->query().queryProp("@name");
  1679. const char *id = active->query().queryProp("@id");
  1680. if (name && id && WildMatch(name, itemId))
  1681. queryIds->setProp(id, 0);
  1682. }
  1683. }
  1684. Owned<IPropertyTreeIterator> queries = queryset->getElements("Query");
  1685. ForEach(*queries)
  1686. {
  1687. const char *id = queries->query().queryProp("@id");
  1688. if (id && WildMatch(id, itemId))
  1689. queryIds->setProp(id, 0);
  1690. }
  1691. }
  1692. }
  1693. }
  1694. void expandQueryActionTargetList(IProperties *queryIds, IPropertyTree *queryset, const char *id, CQuerySetQueryActionTypes action)
  1695. {
  1696. IArrayOf<IConstQuerySetQueryActionItem> items;
  1697. Owned<IEspQuerySetQueryActionItem> item = createQuerySetQueryActionItem();
  1698. item->setQueryId(id);
  1699. items.append(*(IConstQuerySetQueryActionItem*)item.getClear());
  1700. expandQueryActionTargetList(queryIds, queryset, items, action);
  1701. }
  1702. bool CWsWorkunitsEx::onWUQueryConfig(IEspContext &context, IEspWUQueryConfigRequest & req, IEspWUQueryConfigResponse & resp)
  1703. {
  1704. StringAttr target(req.getTarget());
  1705. if (target.isEmpty())
  1706. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Target name required");
  1707. if (!isValidCluster(target))
  1708. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid target name: %s", target.get());
  1709. Owned<IPropertyTree> queryset = getQueryRegistry(target.get(), false);
  1710. if (!queryset)
  1711. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Target Queryset %s not found", req.getTarget());
  1712. Owned<IProperties> queryIds = createProperties();
  1713. expandQueryActionTargetList(queryIds, queryset, req.getQueryId(), QuerySetQueryActionTypes_Undefined);
  1714. IArrayOf<IEspWUQueryConfigResult> results;
  1715. Owned<IPropertyIterator> it = queryIds->getIterator();
  1716. ForEach(*it)
  1717. {
  1718. Owned<IEspWUQueryConfigResult> result = createWUQueryConfigResult();
  1719. result->setQueryId(it->getPropKey());
  1720. VStringBuffer xpath("Query[@id='%s']", it->getPropKey());
  1721. IPropertyTree *queryTree = queryset->queryPropTree(xpath);
  1722. if (queryTree)
  1723. {
  1724. updateMemoryLimitSetting(queryTree, req.getMemoryLimit());
  1725. updateQueryPriority(queryTree, req.getPriority());
  1726. updateQuerySetting(req.getTimeLimit_isNull(), queryTree, "@timeLimit", req.getTimeLimit());
  1727. updateQuerySetting(req.getWarnTimeLimit_isNull(), queryTree, "@warnTimeLimit", req.getWarnTimeLimit());
  1728. if (req.getComment())
  1729. queryTree->setProp("@comment", req.getComment());
  1730. }
  1731. results.append(*result.getClear());
  1732. }
  1733. resp.setResults(results);
  1734. bool reloadFailed = false;
  1735. if (0!=req.getWait() && !req.getNoReload())
  1736. reloadFailed = !reloadCluster(target.get(), (unsigned)req.getWait());
  1737. resp.setReloadFailed(reloadFailed);
  1738. return true;
  1739. }
  1740. bool CWsWorkunitsEx::onWUQuerysetQueryAction(IEspContext &context, IEspWUQuerySetQueryActionRequest & req, IEspWUQuerySetQueryActionResponse & resp)
  1741. {
  1742. resp.setQuerySetName(req.getQuerySetName());
  1743. resp.setAction(req.getAction());
  1744. if (isEmpty(req.getQuerySetName()))
  1745. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Queryset name required");
  1746. Owned<IPropertyTree> queryset = getQueryRegistry(req.getQuerySetName(), true);
  1747. if (!queryset)
  1748. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Queryset %s not found", req.getQuerySetName());
  1749. Owned<IProperties> queryIds = createProperties();
  1750. expandQueryActionTargetList(queryIds, queryset, req.getQueries(), req.getAction());
  1751. if (req.getAction() == CQuerySetQueryActionTypes_ResetQueryStats)
  1752. return resetQueryStats(context, req.getQuerySetName(), queryIds, resp);
  1753. IArrayOf<IEspQuerySetQueryActionResult> results;
  1754. Owned<IPropertyIterator> it = queryIds->getIterator();
  1755. ForEach(*it)
  1756. {
  1757. const char *id = it->getPropKey();
  1758. Owned<IEspQuerySetQueryActionResult> result = createQuerySetQueryActionResult();
  1759. result->setQueryId(id);
  1760. try
  1761. {
  1762. Owned<IPropertyTree> query = getQueryById(queryset, id);
  1763. if (!query)
  1764. throw MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "Query %s/%s not found.", req.getQuerySetName(), id);
  1765. switch (req.getAction())
  1766. {
  1767. case CQuerySetQueryActionTypes_ToggleSuspend:
  1768. setQuerySuspendedState(queryset, id, !queryIds->getPropBool(id), context.queryUserId());
  1769. break;
  1770. case CQuerySetQueryActionTypes_Suspend:
  1771. setQuerySuspendedState(queryset, id, true, context.queryUserId());
  1772. break;
  1773. case CQuerySetQueryActionTypes_Unsuspend:
  1774. setQuerySuspendedState(queryset, id, false, NULL);
  1775. break;
  1776. case CQuerySetQueryActionTypes_Activate:
  1777. setQueryAlias(queryset, query->queryProp("@name"), id);
  1778. break;
  1779. case CQuerySetQueryActionTypes_Delete:
  1780. removeNamedQuery(queryset, id);
  1781. query.clear();
  1782. break;
  1783. case CQuerySetQueryActionTypes_RemoveAllAliases:
  1784. removeAliasesFromNamedQuery(queryset, id);
  1785. break;
  1786. }
  1787. result->setSuccess(true);
  1788. if (query)
  1789. result->setSuspended(query->getPropBool("@suspended"));
  1790. }
  1791. catch(IException *e)
  1792. {
  1793. StringBuffer msg;
  1794. result->setMessage(e->errorMessage(msg).str());
  1795. result->setCode(e->errorCode());
  1796. result->setSuccess(false);
  1797. e->Release();
  1798. }
  1799. results.append(*result.getClear());
  1800. }
  1801. resp.setResults(results);
  1802. return true;
  1803. }
  1804. bool CWsWorkunitsEx::onWUQuerysetAliasAction(IEspContext &context, IEspWUQuerySetAliasActionRequest &req, IEspWUQuerySetAliasActionResponse &resp)
  1805. {
  1806. resp.setQuerySetName(req.getQuerySetName());
  1807. resp.setAction(req.getAction());
  1808. if (isEmpty(req.getQuerySetName()))
  1809. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Queryset name required");
  1810. Owned<IPropertyTree> queryset = getQueryRegistry(req.getQuerySetName(), true);
  1811. if (!queryset)
  1812. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Queryset %s not found", req.getQuerySetName());
  1813. IArrayOf<IEspQuerySetAliasActionResult> results;
  1814. ForEachItemIn(i, req.getAliases())
  1815. {
  1816. IConstQuerySetAliasActionItem& item=req.getAliases().item(i);
  1817. Owned<IEspQuerySetAliasActionResult> result = createQuerySetAliasActionResult();
  1818. try
  1819. {
  1820. VStringBuffer xpath("Alias[@name='%s']", item.getName());
  1821. IPropertyTree *alias = queryset->queryPropTree(xpath.str());
  1822. if (!alias)
  1823. throw MakeStringException(ECLWATCH_ALIAS_NOT_FOUND, "Alias %s/%s not found.", req.getQuerySetName(), item.getName());
  1824. switch (req.getAction())
  1825. {
  1826. case CQuerySetAliasActionTypes_Deactivate:
  1827. removeQuerySetAlias(req.getQuerySetName(), item.getName());
  1828. break;
  1829. }
  1830. result->setSuccess(true);
  1831. }
  1832. catch(IException *e)
  1833. {
  1834. StringBuffer msg;
  1835. result->setMessage(e->errorMessage(msg).str());
  1836. result->setCode(e->errorCode());
  1837. result->setSuccess(false);
  1838. e->Release();
  1839. }
  1840. results.append(*result.getClear());
  1841. }
  1842. resp.setResults(results);
  1843. return true;
  1844. }
  1845. #define QUERYPATH_SEP_CHAR '/'
  1846. bool nextQueryPathNode(const char *&path, StringBuffer &node)
  1847. {
  1848. if (*path==QUERYPATH_SEP_CHAR)
  1849. path++;
  1850. while (*path && *path!=QUERYPATH_SEP_CHAR)
  1851. node.append(*path++);
  1852. return (*path && *++path);
  1853. }
  1854. bool splitQueryPath(const char *path, StringBuffer &netAddress, StringBuffer &queryset, StringBuffer *query)
  1855. {
  1856. if (!path || !*path)
  1857. return false;
  1858. if (*path==QUERYPATH_SEP_CHAR && path[1]==QUERYPATH_SEP_CHAR)
  1859. {
  1860. path+=2;
  1861. if (!nextQueryPathNode(path, netAddress))
  1862. return false;
  1863. }
  1864. if (!nextQueryPathNode(path, queryset))
  1865. return (query==NULL);
  1866. if (!query)
  1867. return false;
  1868. if (nextQueryPathNode(path, *query))
  1869. return false; //query path too deep
  1870. return true;
  1871. }
  1872. IPropertyTree *fetchRemoteQuerySetInfo(IEspContext *context, const char *srcAddress, const char *srcTarget)
  1873. {
  1874. if (!srcAddress || !*srcAddress || !srcTarget || !*srcTarget)
  1875. return NULL;
  1876. VStringBuffer url("http://%s%s/WsWorkunits/WUQuerysetDetails.xml?ver_=1.51&QuerySetName=%s&FilterType=All", srcAddress, (!strchr(srcAddress, ':')) ? ":8010" : "", srcTarget);
  1877. Owned<IHttpClientContext> httpCtx = getHttpClientContext();
  1878. Owned<IHttpClient> httpclient = httpCtx->createHttpClient(NULL, url);
  1879. const char *user = context->queryUserId();
  1880. if (user && *user)
  1881. httpclient->setUserID(user);
  1882. const char *pw = context->queryPassword();
  1883. if (pw && *pw)
  1884. httpclient->setPassword(pw);
  1885. StringBuffer request; //empty
  1886. StringBuffer response;
  1887. StringBuffer status;
  1888. if (0 > httpclient->sendRequest("GET", NULL, request, response, status) || !response.length() || strncmp("200", status, 3))
  1889. throw MakeStringException(-1, "Error fetching remote queryset information: %s %s %s", srcAddress, srcTarget, status.str());
  1890. return createPTreeFromXMLString(response);
  1891. }
  1892. class QueryCloner
  1893. {
  1894. public:
  1895. QueryCloner(IEspContext *_context, const char *address, const char *source, const char *_target) :
  1896. context(_context), cloneFilesEnabled(false), target(_target), updateFlags(0), srcAddress(address)
  1897. {
  1898. if (srcAddress.length())
  1899. srcQuerySet.setown(fetchRemoteQuerySetInfo(context, srcAddress, source));
  1900. else
  1901. srcQuerySet.setown(getQueryRegistry(source, true));
  1902. if (!srcQuerySet)
  1903. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Source Queryset %s %s not found", srcAddress.str(), source);
  1904. destQuerySet.setown(getQueryRegistry(target, false));
  1905. if (!destQuerySet) // getQueryRegistry should have created if not found
  1906. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Destination Queryset %s could not be created, or found", target.str());
  1907. factory.setown(getWorkUnitFactory(context->querySecManager(), context->queryUser()));
  1908. }
  1909. void cloneQueryRemote(IPropertyTree *query, bool makeActive)
  1910. {
  1911. StringBuffer wuid = query->queryProp("Wuid");
  1912. if (!wuid.length())
  1913. return;
  1914. const char *queryName = query->queryProp("Name");
  1915. if (!queryName || !*queryName)
  1916. return;
  1917. StringBuffer xml;
  1918. MemoryBuffer dll;
  1919. StringBuffer dllname;
  1920. StringBuffer fetchedName;
  1921. StringBuffer remoteDfs;
  1922. fetchRemoteWorkunit(NULL, context, srcAddress.str(), NULL, NULL, wuid, fetchedName, xml, dllname, dll, remoteDfs);
  1923. deploySharedObject(*context, wuid, dllname, target, queryName, dll, NULL, xml.str());
  1924. SCMStringBuffer existingQueryId;
  1925. queryIdFromQuerySetWuid(destQuerySet, wuid, queryName, existingQueryId);
  1926. if (existingQueryId.length())
  1927. {
  1928. existingQueryIds.append(existingQueryId.str());
  1929. if (makeActive)
  1930. activateQuery(destQuerySet, ACTIVATE_SUSPEND_PREVIOUS, queryName, existingQueryId.str(), context->queryUserId());
  1931. return;
  1932. }
  1933. StringBuffer newQueryId;
  1934. Owned<IWorkUnit> workunit = factory->updateWorkUnit(wuid);
  1935. addQueryToQuerySet(workunit, destQuerySet, queryName, makeActive ? ACTIVATE_SUSPEND_PREVIOUS : DO_NOT_ACTIVATE, newQueryId, context->queryUserId());
  1936. copiedQueryIds.append(newQueryId);
  1937. Owned<IPropertyTree> destQuery = getQueryById(destQuerySet, newQueryId);
  1938. if (destQuery)
  1939. {
  1940. Owned<IAttributeIterator> aiter = query->getAttributes();
  1941. ForEach(*aiter)
  1942. {
  1943. const char *atname = aiter->queryName();
  1944. if (!destQuery->hasProp(atname))
  1945. destQuery->setProp(atname, aiter->queryValue());
  1946. }
  1947. if (cloneFilesEnabled && wufiles)
  1948. wufiles->addFilesFromQuery(workunit, pm, newQueryId);
  1949. }
  1950. }
  1951. void cloneQueryLocal(IPropertyTree *query, bool makeActive)
  1952. {
  1953. const char *wuid = query->queryProp("@wuid");
  1954. if (!wuid || !*wuid)
  1955. return;
  1956. const char *queryName = query->queryProp("@name");
  1957. if (!queryName || !*queryName)
  1958. return;
  1959. SCMStringBuffer existingQueryId;
  1960. queryIdFromQuerySetWuid(destQuerySet, wuid, queryName, existingQueryId);
  1961. if (existingQueryId.length())
  1962. {
  1963. existingQueryIds.append(existingQueryId.str());
  1964. if (makeActive)
  1965. activateQuery(destQuerySet, ACTIVATE_SUSPEND_PREVIOUS, queryName, existingQueryId.str(), context->queryUserId());
  1966. return;
  1967. }
  1968. StringBuffer newQueryId;
  1969. Owned<IWorkUnit> workunit = factory->updateWorkUnit(wuid);
  1970. addQueryToQuerySet(workunit, destQuerySet, queryName, makeActive ? ACTIVATE_SUSPEND_PREVIOUS : DO_NOT_ACTIVATE, newQueryId, context->queryUserId());
  1971. copiedQueryIds.append(newQueryId);
  1972. Owned<IPropertyTree> destQuery = getQueryById(destQuerySet, newQueryId);
  1973. if (destQuery)
  1974. {
  1975. Owned<IAttributeIterator> aiter = query->getAttributes();
  1976. ForEach(*aiter)
  1977. {
  1978. const char *atname = aiter->queryName();
  1979. if (!destQuery->hasProp(atname))
  1980. destQuery->setProp(atname, aiter->queryValue());
  1981. }
  1982. Owned<IPropertyTreeIterator> children = query->getElements("*");
  1983. ForEach(*children)
  1984. {
  1985. IPropertyTree &child = children->query();
  1986. destQuery->addPropTree(child.queryName(), createPTreeFromIPT(&child));
  1987. }
  1988. if (cloneFilesEnabled && wufiles)
  1989. wufiles->addFilesFromQuery(workunit, pm, newQueryId);
  1990. }
  1991. }
  1992. void cloneActiveRemote(bool makeActive)
  1993. {
  1994. Owned<IPropertyTreeIterator> activeQueries = srcQuerySet->getElements("QuerysetAliases/QuerySetAlias");
  1995. ForEach(*activeQueries)
  1996. {
  1997. IPropertyTree &alias = activeQueries->query();
  1998. VStringBuffer xpath("QuerysetQueries/QuerySetQuery[Id='%s'][1]", alias.queryProp("Id"));
  1999. IPropertyTree *query = srcQuerySet->queryPropTree(xpath);
  2000. if (!query)
  2001. continue;
  2002. cloneQueryRemote(query, makeActive);
  2003. }
  2004. }
  2005. void cloneActiveLocal(bool makeActive)
  2006. {
  2007. Owned<IPropertyTreeIterator> activeQueries = srcQuerySet->getElements("Alias");
  2008. ForEach(*activeQueries)
  2009. {
  2010. IPropertyTree &alias = activeQueries->query();
  2011. Owned<IPropertyTree> query = getQueryById(srcQuerySet, alias.queryProp("@id"));
  2012. if (!query)
  2013. return;
  2014. cloneQueryLocal(query, makeActive);
  2015. }
  2016. }
  2017. void cloneActive(bool makeActive)
  2018. {
  2019. if (srcAddress.length())
  2020. cloneActiveRemote(makeActive);
  2021. else
  2022. cloneActiveLocal(makeActive);
  2023. }
  2024. void cloneAllRemote(bool cloneActiveState)
  2025. {
  2026. Owned<IPropertyTreeIterator> allQueries = srcQuerySet->getElements("QuerysetQueries/QuerySetQuery");
  2027. ForEach(*allQueries)
  2028. {
  2029. IPropertyTree &query = allQueries->query();
  2030. bool makeActive = false;
  2031. if (cloneActiveState)
  2032. {
  2033. VStringBuffer xpath("QuerysetAliases/QuerySetAlias[Id='%s']", query.queryProp("Id"));
  2034. makeActive = srcQuerySet->hasProp(xpath);
  2035. }
  2036. cloneQueryRemote(&query, makeActive);
  2037. }
  2038. }
  2039. void cloneAllLocal(bool cloneActiveState)
  2040. {
  2041. Owned<IPropertyTreeIterator> allQueries = srcQuerySet->getElements("Query");
  2042. ForEach(*allQueries)
  2043. {
  2044. IPropertyTree &query = allQueries->query();
  2045. bool makeActive = false;
  2046. if (cloneActiveState)
  2047. {
  2048. VStringBuffer xpath("Alias[@id='%s']", query.queryProp("@id"));
  2049. makeActive = srcQuerySet->hasProp(xpath);
  2050. }
  2051. cloneQueryLocal(&query, makeActive);
  2052. }
  2053. }
  2054. void cloneAll(bool cloneActiveState)
  2055. {
  2056. if (srcAddress.length())
  2057. cloneAllRemote(cloneActiveState);
  2058. else
  2059. cloneAllLocal(cloneActiveState);
  2060. }
  2061. void enableFileCloning(unsigned _updateFlags, const char *dfsServer, const char *destProcess, const char *sourceProcess, bool allowForeign)
  2062. {
  2063. cloneFilesEnabled = true;
  2064. updateFlags = _updateFlags;
  2065. splitDerivedDfsLocation(dfsServer, srcCluster, dfsIP, srcPrefix, sourceProcess, sourceProcess, NULL, NULL);
  2066. wufiles.setown(createReferencedFileList(context->queryUserId(), context->queryPassword(), allowForeign, false));
  2067. Owned<IHpccPackageSet> ps = createPackageSet(destProcess);
  2068. pm.set(ps->queryActiveMap(target));
  2069. process.set(destProcess);
  2070. }
  2071. void cloneFiles()
  2072. {
  2073. if (cloneFilesEnabled)
  2074. {
  2075. wufiles->resolveFiles(process, dfsIP, srcPrefix, srcCluster, !(updateFlags & (DALI_UPDATEF_REPLACE_FILE | DALI_UPDATEF_CLONE_FROM)), true, false, true);
  2076. Owned<IDFUhelper> helper = createIDFUhelper();
  2077. Owned <IConstWUClusterInfo> cl = getTargetClusterInfo(target);
  2078. if (cl)
  2079. {
  2080. SCMStringBuffer process;
  2081. StringBuffer defReplicateFolder;
  2082. getConfigurationDirectory(NULL, "data2", "roxie", cl->getRoxieProcess(process).str(), defReplicateFolder);
  2083. wufiles->cloneAllInfo(updateFlags, helper, true, true, cl->getRoxieRedundancy(), cl->getChannelsPerNode(), cl->getRoxieReplicateOffset(), defReplicateFolder);
  2084. }
  2085. }
  2086. }
  2087. private:
  2088. Linked<IEspContext> context;
  2089. Linked<IWorkUnitFactory> factory;
  2090. Owned<IPropertyTree> destQuerySet;
  2091. Owned<IPropertyTree> srcQuerySet;
  2092. Owned<IReferencedFileList> wufiles;
  2093. Owned<const IHpccPackageMap> pm;
  2094. StringBuffer dfsIP;
  2095. StringBuffer srcAddress;
  2096. StringBuffer srcCluster;
  2097. StringBuffer srcPrefix;
  2098. StringAttr target;
  2099. StringAttr process;
  2100. bool cloneFilesEnabled;
  2101. unsigned updateFlags;
  2102. public:
  2103. StringArray existingQueryIds;
  2104. StringArray copiedQueryIds;
  2105. };
  2106. bool CWsWorkunitsEx::onWUCopyQuerySet(IEspContext &context, IEspWUCopyQuerySetRequest &req, IEspWUCopyQuerySetResponse &resp)
  2107. {
  2108. const char *source = req.getSource();
  2109. if (!source || !*source)
  2110. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "No source target specified");
  2111. StringBuffer srcAddress;
  2112. StringBuffer srcTarget;
  2113. if (!splitQueryPath(source, srcAddress, srcTarget, NULL))
  2114. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid source target");
  2115. if (!srcAddress.length() && !isValidCluster(srcTarget))
  2116. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid source target name: %s", source);
  2117. const char *target = req.getTarget();
  2118. if (!target || !*target)
  2119. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "No destination target specified");
  2120. if (!isValidCluster(target))
  2121. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid destination target name: %s", target);
  2122. DBGLOG("%s copying queryset %s from %s target %s", context.queryUserId(), target, srcAddress.str(), srcTarget.str());
  2123. QueryCloner cloner(&context, srcAddress, srcTarget, target);
  2124. SCMStringBuffer process;
  2125. if (req.getCopyFiles())
  2126. {
  2127. Owned <IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(target);
  2128. if (clusterInfo && clusterInfo->getPlatform()==RoxieCluster)
  2129. {
  2130. clusterInfo->getRoxieProcess(process);
  2131. if (!process.length())
  2132. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_INFO, "DFS process cluster not found for destination target %s", target);
  2133. unsigned updateFlags = 0;
  2134. if (req.getOverwriteDfs())
  2135. updateFlags |= (DALI_UPDATEF_REPLACE_FILE | DALI_UPDATEF_CLONE_FROM | DALI_UPDATEF_SUPERFILES);
  2136. if (req.getUpdateCloneFrom())
  2137. updateFlags |= DALI_UPDATEF_CLONE_FROM;
  2138. if (req.getUpdateSuperFiles())
  2139. updateFlags |= DALI_UPDATEF_SUPERFILES;
  2140. if (req.getAppendCluster())
  2141. updateFlags |= DALI_UPDATEF_APPEND_CLUSTER;
  2142. cloner.enableFileCloning(updateFlags, req.getDfsServer(), process.str(), req.getSourceProcess(), req.getAllowForeignFiles());
  2143. }
  2144. }
  2145. if (req.getActiveOnly())
  2146. cloner.cloneActive(req.getCloneActiveState());
  2147. else
  2148. cloner.cloneAll(req.getCloneActiveState());
  2149. cloner.cloneFiles();
  2150. resp.setCopiedQueries(cloner.copiedQueryIds);
  2151. resp.setExistingQueries(cloner.existingQueryIds);
  2152. return true;
  2153. }
  2154. bool CWsWorkunitsEx::onWUQuerysetCopyQuery(IEspContext &context, IEspWUQuerySetCopyQueryRequest &req, IEspWUQuerySetCopyQueryResponse &resp)
  2155. {
  2156. unsigned start = msTick();
  2157. const char *source = req.getSource();
  2158. if (!source || !*source)
  2159. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "No source query specified");
  2160. const char *target = req.getTarget();
  2161. if (!target || !*target)
  2162. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "No destination specified");
  2163. if (strchr(target, '/')) //for future use
  2164. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid target queryset name");
  2165. if (req.getCluster() && *req.getCluster() && !strieq(req.getCluster(), target)) //backward compatability check
  2166. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid target cluster and queryset must match");
  2167. if (!isValidCluster(target))
  2168. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid target name: %s", target);
  2169. StringBuffer srcAddress, srcQuerySet, srcQuery;
  2170. if (!splitQueryPath(source, srcAddress, srcQuerySet, &srcQuery))
  2171. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid source query path");
  2172. StringAttr targetQueryName(req.getDestName());
  2173. Owned<IClientWUQuerySetDetailsResponse> sourceQueryInfoResp;
  2174. IConstQuerySetQuery *srcInfo=NULL;
  2175. DBGLOG("%s copying query %s to target %s from %s target %s", context.queryUserId(), srcQuery.str(), target, srcAddress.str(), srcQuerySet.str());
  2176. StringBuffer remoteIP;
  2177. StringBuffer wuid;
  2178. if (srcAddress.length())
  2179. {
  2180. StringBuffer xml;
  2181. MemoryBuffer dll;
  2182. StringBuffer dllname;
  2183. StringBuffer queryName;
  2184. fetchRemoteWorkunitAndQueryDetails(NULL, &context, srcAddress.str(), srcQuerySet.str(), srcQuery.str(), NULL, queryName, xml, dllname, dll, remoteIP, sourceQueryInfoResp);
  2185. if (sourceQueryInfoResp && sourceQueryInfoResp->getQuerysetQueries().ordinality())
  2186. srcInfo = &sourceQueryInfoResp->getQuerysetQueries().item(0);
  2187. if (srcInfo)
  2188. wuid.set(srcInfo->getWuid());
  2189. if (targetQueryName.isEmpty())
  2190. targetQueryName.set(queryName);
  2191. deploySharedObject(context, wuid, dllname.str(), target, targetQueryName.get(), dll, queryDirectory.str(), xml.str());
  2192. }
  2193. else
  2194. {
  2195. //Could get the atributes without soap call, but this creates a common data structure shared with fetching remote query info
  2196. //Get query attributes before resolveQueryAlias, to avoid deadlock
  2197. sourceQueryInfoResp.setown(fetchQueryDetails(NULL, &context, NULL, srcQuerySet, srcQuery));
  2198. if (sourceQueryInfoResp && sourceQueryInfoResp->getQuerysetQueries().ordinality())
  2199. srcInfo = &sourceQueryInfoResp->getQuerysetQueries().item(0);
  2200. Owned<IPropertyTree> queryset = getQueryRegistry(srcQuerySet.str(), true);
  2201. if (!queryset)
  2202. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Source Queryset %s not found", srcQuery.str());
  2203. Owned<IPropertyTree> query = resolveQueryAlias(queryset, srcQuery.str());
  2204. if (!query)
  2205. throw MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "Source query %s not found", source);
  2206. wuid.set(query->queryProp("@wuid"));
  2207. if (targetQueryName.isEmpty())
  2208. targetQueryName.set(query->queryProp("@name"));
  2209. }
  2210. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  2211. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid.str());
  2212. if (!req.getDontCopyFiles())
  2213. {
  2214. StringBuffer daliIP;
  2215. StringBuffer srcCluster;
  2216. StringBuffer srcPrefix;
  2217. splitDerivedDfsLocation(req.getDaliServer(), srcCluster, daliIP, srcPrefix, req.getSourceProcess(), req.getSourceProcess(), remoteIP.str(), NULL);
  2218. unsigned updateFlags = 0;
  2219. if (req.getOverwrite())
  2220. updateFlags |= (DALI_UPDATEF_REPLACE_FILE | DALI_UPDATEF_CLONE_FROM | DALI_UPDATEF_SUPERFILES);
  2221. if (req.getUpdateCloneFrom())
  2222. updateFlags |= DALI_UPDATEF_CLONE_FROM;
  2223. if (req.getUpdateSuperFiles())
  2224. updateFlags |= DALI_UPDATEF_SUPERFILES;
  2225. if (req.getAppendCluster())
  2226. updateFlags |= DALI_UPDATEF_APPEND_CLUSTER;
  2227. copyQueryFilesToCluster(updateFlags, context, cw, daliIP.str(), srcPrefix, target, srcCluster, targetQueryName.get(), req.getAllowForeignFiles());
  2228. }
  2229. WorkunitUpdate wu(&cw->lock());
  2230. if (!wu)
  2231. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Error opening wuid %s for query %s", wuid.str(), source);
  2232. StringBuffer targetQueryId;
  2233. WUQueryActivationOptions activate = (WUQueryActivationOptions)req.getActivate();
  2234. addQueryToQuerySet(wu, target, targetQueryName.get(), activate, targetQueryId, context.queryUserId());
  2235. Owned<IPropertyTree> queryTree = getQueryById(target, targetQueryId, false);
  2236. if (queryTree)
  2237. {
  2238. updateMemoryLimitSetting(queryTree, req.getMemoryLimit(), srcInfo);
  2239. updateQueryPriority(queryTree, req.getPriority(), srcInfo);
  2240. updateTimeLimitSetting(queryTree, req.getTimeLimit_isNull(), req.getTimeLimit(), srcInfo);
  2241. updateWarnTimeLimitSetting(queryTree, req.getWarnTimeLimit_isNull(), req.getWarnTimeLimit(), srcInfo);
  2242. if (req.getComment())
  2243. queryTree->setProp("@comment", req.getComment());
  2244. else if (srcInfo && srcInfo->getComment())
  2245. queryTree->setProp("@comment", srcInfo->getComment());
  2246. if (srcInfo && srcInfo->getSnapshot())
  2247. queryTree->setProp("@snapshot", srcInfo->getSnapshot());
  2248. }
  2249. wu.clear();
  2250. resp.setQueryId(targetQueryId.str());
  2251. if (0!=req.getWait() && !req.getNoReload())
  2252. reloadCluster(target, remainingMsWait(req.getWait(), start));
  2253. return true;
  2254. }
  2255. void CWsWorkunitsEx::getGraphsByQueryId(const char *target, const char *queryId, const char *graphId, const char *subGraphId, IArrayOf<IEspECLGraphEx>& ECLGraphs)
  2256. {
  2257. if (!target || !*target)
  2258. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Target name required");
  2259. if (!queryId || !*queryId)
  2260. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Query Id required");
  2261. Owned<IConstWUClusterInfo> info = getTargetClusterInfo(target);
  2262. if (!info || (info->getPlatform()!=RoxieCluster)) //Only support roxie for now
  2263. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "Invalid Roxie name");
  2264. const SocketEndpointArray &eps = info->getRoxieServers();
  2265. if (eps.empty())
  2266. return;
  2267. VStringBuffer control("<control:querystats><Query id='%s'/></control:querystats>", queryId);
  2268. Owned<IPropertyTree> querystats = sendRoxieControlAllNodes(eps.item(0), control.str(), false, ROXIELOCKCONNECTIONTIMEOUT);
  2269. if (!querystats)
  2270. return;
  2271. Owned<IPropertyTreeIterator> graphs = querystats->getElements("Endpoint/Query/Graph");
  2272. ForEach(*graphs)
  2273. {
  2274. IPropertyTree &graph = graphs->query();
  2275. const char* aGraphId = graph.queryProp("@id");
  2276. if (graphId && *graphId && !strieq(graphId, aGraphId))
  2277. continue;
  2278. IPropertyTree* xgmml = graph.getBranch("xgmml/graph");
  2279. if (!xgmml)
  2280. continue;
  2281. Owned<IEspECLGraphEx> g = createECLGraphEx("","");
  2282. g->setName(aGraphId);
  2283. StringBuffer xml;
  2284. if (!subGraphId || !*subGraphId)
  2285. toXML(xgmml, xml);
  2286. else
  2287. {
  2288. VStringBuffer xpath("//node[@id='%s']", subGraphId);
  2289. toXML(xgmml->queryPropTree(xpath.str()), xml);
  2290. }
  2291. g->setGraph(xml.str());
  2292. ECLGraphs.append(*g.getClear());
  2293. }
  2294. return;
  2295. }
  2296. bool CWsWorkunitsEx::onWUQueryGetGraph(IEspContext& context, IEspWUQueryGetGraphRequest& req, IEspWUQueryGetGraphResponse& resp)
  2297. {
  2298. try
  2299. {
  2300. IArrayOf<IEspECLGraphEx> graphs;
  2301. getGraphsByQueryId(req.getTarget(), req.getQueryId(), req.getGraphName(), req.getSubGraphId(), graphs);
  2302. resp.setGraphs(graphs);
  2303. }
  2304. catch(IException* e)
  2305. {
  2306. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2307. }
  2308. return true;
  2309. }
  2310. bool CWsWorkunitsEx::resetQueryStats(IEspContext& context, const char* target, IProperties* queryIds, IEspWUQuerySetQueryActionResponse& resp)
  2311. {
  2312. IArrayOf<IEspQuerySetQueryActionResult> results;
  2313. Owned<IEspQuerySetQueryActionResult> result = createQuerySetQueryActionResult();
  2314. try
  2315. {
  2316. StringBuffer control;
  2317. Owned<IPropertyIterator> it = queryIds->getIterator();
  2318. ForEach(*it)
  2319. {
  2320. const char *queryId = it->getPropKey();
  2321. if (queryId && *queryId)
  2322. {
  2323. appendXMLOpenTag(control, "Query", NULL, false);
  2324. appendXMLAttr(control, "id", queryId);
  2325. if (target && *target)
  2326. appendXMLAttr(control, "target", target);
  2327. control.append("/>");
  2328. }
  2329. }
  2330. if (!control.length())
  2331. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "CWsWorkunitsEx::resetQueryStats: Query ID not specified");
  2332. control.insert(0, "<control:resetquerystats>");
  2333. control.append("</control:resetquerystats>");
  2334. if (!sendControlQuery(context, target, control.str(), ROXIECONNECTIONTIMEOUT))
  2335. throw MakeStringException(ECLWATCH_INTERNAL_ERROR, "CWsWorkunitsEx::resetQueryStats: Failed to send roxie control query");
  2336. result->setMessage("Query stats reset succeeded");
  2337. result->setSuccess(true);;
  2338. }
  2339. catch(IMultiException *me)
  2340. {
  2341. StringBuffer msg;
  2342. result->setMessage(me->errorMessage(msg).str());
  2343. result->setCode(me->errorCode());
  2344. result->setSuccess(false);
  2345. me->Release();
  2346. }
  2347. catch(IException *e)
  2348. {
  2349. StringBuffer msg;
  2350. result->setMessage(e->errorMessage(msg).str());
  2351. result->setCode(e->errorCode());
  2352. result->setSuccess(false);
  2353. e->Release();
  2354. }
  2355. results.append(*result.getClear());
  2356. resp.setResults(results);
  2357. return true;
  2358. }
  2359. IPropertyTree* CWsWorkunitsEx::sendControlQuery(IEspContext& context, const char* target, const char* query, unsigned timeout)
  2360. {
  2361. if (!target || !*target)
  2362. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "CWsWorkunitsEx::sendControlQuery: target not specified");
  2363. if (!query || !*query)
  2364. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "CWsWorkunitsEx::sendControlQuery: Control query not specified");
  2365. Owned<IConstWUClusterInfo> info = getTargetClusterInfo(target);
  2366. if (!info || (info->getPlatform()!=RoxieCluster)) //Only support roxie for now
  2367. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "CWsWorkunitsEx::sendControlQuery: Invalid target name %s", target);
  2368. const SocketEndpointArray &eps = info->getRoxieServers();
  2369. if (eps.empty())
  2370. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_NAME, "CWsWorkunitsEx::sendControlQuery: Server not found for %s", target);
  2371. Owned<ISocket> sock = ISocket::connect_timeout(eps.item(0), timeout);
  2372. return sendRoxieControlQuery(sock, query, timeout);
  2373. }