ws_workunitsHelpers.cpp 91 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906
  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 "jlib.hpp"
  14. #include "ws_workunitsHelpers.hpp"
  15. #include "exception_util.hpp"
  16. #include "daclient.hpp"
  17. #include "dalienv.hpp"
  18. #include "daaudit.hpp"
  19. #include "portlist.h"
  20. #include "dadfs.hpp"
  21. #include "fileview.hpp"
  22. #include "wuwebview.hpp"
  23. #include "dllserver.hpp"
  24. #include "wujobq.hpp"
  25. #include "hqlexpr.hpp"
  26. #ifdef _USE_ZLIB
  27. #include "zcrypt.hpp"
  28. #endif
  29. namespace ws_workunits {
  30. SecAccessFlags chooseWuAccessFlagsByOwnership(const char *user, const char *owner, SecAccessFlags accessOwn, SecAccessFlags accessOthers)
  31. {
  32. return (isEmpty(owner) || (user && streq(user, owner))) ? accessOwn : accessOthers;
  33. }
  34. SecAccessFlags chooseWuAccessFlagsByOwnership(const char *user, IConstWorkUnit& cw, SecAccessFlags accessOwn, SecAccessFlags accessOthers)
  35. {
  36. SCMStringBuffer owner;
  37. return chooseWuAccessFlagsByOwnership(user, cw.getUser(owner).str(), accessOwn, accessOthers);
  38. }
  39. const char *getWuAccessType(const char *owner, const char *user)
  40. {
  41. return (isEmpty(owner) || (user && streq(user, owner))) ? OWN_WU_ACCESS : OTHERS_WU_ACCESS;
  42. }
  43. const char *getWuAccessType(IConstWorkUnit& cw, const char *user)
  44. {
  45. SCMStringBuffer owner;
  46. return getWuAccessType(cw.getUser(owner).str(), user);
  47. }
  48. void getUserWuAccessFlags(IEspContext& context, SecAccessFlags& accessOwn, SecAccessFlags& accessOthers, bool except)
  49. {
  50. if (!context.authorizeFeature(OWN_WU_ACCESS, accessOwn))
  51. accessOwn = SecAccess_None;
  52. if (!context.authorizeFeature(OTHERS_WU_ACCESS, accessOthers))
  53. accessOthers = SecAccess_None;
  54. if (except && (accessOwn == SecAccess_None) && (accessOthers == SecAccess_None))
  55. {
  56. AuditSystemAccess(context.queryUserId(), false, "Access Denied: User can't view any workunits");
  57. VStringBuffer msg("Access Denied: User %s does not have rights to access workunits.", context.queryUserId());
  58. throw MakeStringException(ECLWATCH_ECL_WU_ACCESS_DENIED, "%s", msg.str());
  59. }
  60. }
  61. SecAccessFlags getWsWorkunitAccess(IEspContext& cxt, IConstWorkUnit& cw)
  62. {
  63. SecAccessFlags accessFlag = SecAccess_None;
  64. cxt.authorizeFeature(getWuAccessType(cw, cxt.queryUserId()), accessFlag);
  65. return accessFlag;
  66. }
  67. void ensureWsWorkunitAccessByOwnerId(IEspContext& cxt, const char* owner, SecAccessFlags minAccess)
  68. {
  69. if (!cxt.validateFeatureAccess(getWuAccessType(owner, cxt.queryUserId()), minAccess, false))
  70. throw MakeStringException(ECLWATCH_ECL_WU_ACCESS_DENIED, "Failed to access workunit. Permission denied.");
  71. }
  72. void ensureWsWorkunitAccess(IEspContext& cxt, IConstWorkUnit& cw, SecAccessFlags minAccess)
  73. {
  74. if (!cxt.validateFeatureAccess(getWuAccessType(cw, cxt.queryUserId()), minAccess, false))
  75. throw MakeStringException(ECLWATCH_ECL_WU_ACCESS_DENIED, "Failed to access workunit. Permission denied.");
  76. }
  77. void ensureWsWorkunitAccess(IEspContext& context, const char* wuid, SecAccessFlags minAccess)
  78. {
  79. Owned<IWorkUnitFactory> wf = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  80. Owned<IConstWorkUnit> cw = wf->openWorkUnit(wuid, false);
  81. if (!cw)
  82. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Failed to open workunit %s when ensuring workunit access", wuid);
  83. ensureWsWorkunitAccess(context, *cw, minAccess);
  84. }
  85. void ensureWsCreateWorkunitAccess(IEspContext& cxt)
  86. {
  87. if (!cxt.validateFeatureAccess(OWN_WU_ACCESS, SecAccess_Write, false))
  88. throw MakeStringException(ECLWATCH_ECL_WU_ACCESS_DENIED, "Failed to create workunit. Permission denied.");
  89. }
  90. StringBuffer &getWuidFromLogicalFileName(IEspContext &context, const char *logicalName, StringBuffer &wuid)
  91. {
  92. Owned<IUserDescriptor> userdesc = createUserDescriptor();
  93. userdesc->set(context.queryUserId(), context.queryPassword());
  94. Owned<IDistributedFile> df = queryDistributedFileDirectory().lookup(logicalName, userdesc);
  95. if (!df)
  96. throw MakeStringException(ECLWATCH_FILE_NOT_EXIST,"Cannot find file %s.",logicalName);
  97. return wuid.append(df->queryAttributes().queryProp("@workunit"));
  98. }
  99. void formatDuration(StringBuffer &s, unsigned ms)
  100. {
  101. unsigned days = ms / (1000*60*60*24);
  102. ms %= (1000*60*60*24);
  103. unsigned hours = ms / (1000*60*60);
  104. ms %= (1000*60*60);
  105. unsigned mins = ms / (1000*60);
  106. ms %= (1000*60);
  107. unsigned secs = ms / 1000;
  108. ms %= 1000;
  109. if (days)
  110. s.appendf("%d days ", days);
  111. if (hours || s.length())
  112. s.appendf("%d:", hours);
  113. if (mins || s.length())
  114. s.appendf("%d:", mins);
  115. if (s.length())
  116. s.appendf("%02d.%03d", secs, ms);
  117. else
  118. s.appendf("%d.%03d", secs, ms);
  119. }
  120. WsWUExceptions::WsWUExceptions(IConstWorkUnit& wu): numerr(0), numwrn(0), numinf(0)
  121. {
  122. Owned<IConstWUExceptionIterator> it = &wu.getExceptions();
  123. ForEach(*it)
  124. {
  125. SCMStringBuffer src, msg, file;
  126. Owned<IEspECLException> e= createECLException("","");
  127. e->setCode(it->query().getExceptionCode());
  128. e->setSource(it->query().getExceptionSource(src).str());
  129. e->setMessage(it->query().getExceptionMessage(msg).str());
  130. e->setFileName(it->query().getExceptionFileName(file).str());
  131. e->setLineNo(it->query().getExceptionLineNo());
  132. e->setColumn(it->query().getExceptionColumn());
  133. const char * label = "";
  134. switch (it->query().getSeverity())
  135. {
  136. default:
  137. case ExceptionSeverityError: label = "Error"; numerr++; break;
  138. case ExceptionSeverityWarning: label = "Warning"; numwrn++; break;
  139. case ExceptionSeverityInformation: label = "Info"; numinf++; break;
  140. }
  141. e->setSeverity(label);
  142. errors.append(*e.getLink());
  143. }
  144. }
  145. #define SDS_LOCK_TIMEOUT 30000
  146. void getSashaNode(SocketEndpoint &ep)
  147. {
  148. Owned<IRemoteConnection> econn = querySDS().connect("/Environment", myProcessSession(), 0, SDS_LOCK_TIMEOUT);
  149. if (!econn)
  150. throw MakeStringException(ECLWATCH_CANNOT_CONNECT_DALI,"Cannot connect to DALI server.");
  151. IPropertyTree *root = econn->queryRoot();
  152. if (!root)
  153. throw MakeStringException(ECLWATCH_CANNOT_GET_ENV_INFO,"Cannot get environment information.");
  154. IPropertyTree *pt = root->queryPropTree("Software/SashaServerProcess[1]/Instance[1]");
  155. if (!pt)
  156. throw MakeStringException(ECLWATCH_ARCHIVE_SERVER_NOT_FOUND, "Archive Server not found.");
  157. ep.set(pt->queryProp("@netAddress"), pt->getPropInt("@port",DEFAULT_SASHA_PORT));
  158. }
  159. void WsWuInfo::getSourceFiles(IEspECLWorkunit &info, unsigned flags)
  160. {
  161. if (!(flags & WUINFO_IncludeSourceFiles))
  162. return;
  163. try
  164. {
  165. Owned<IUserDescriptor> userdesc;
  166. StringBuffer username;
  167. context.getUserID(username);
  168. const char* passwd = context.queryPassword();
  169. userdesc.setown(createUserDescriptor());
  170. userdesc->set(username.str(), passwd);
  171. IArrayOf<IEspECLSourceFile> files;
  172. if (version < 1.27)
  173. {
  174. Owned<IPropertyTreeIterator> f=&cw->getFilesReadIterator();
  175. ForEach(*f)
  176. {
  177. IPropertyTree &query = f->query();
  178. const char *clusterName = query.queryProp("@cluster");
  179. const char *fileName = query.queryProp("@name");
  180. int fileCount = query.getPropInt("@useCount");
  181. Owned<IEspECLSourceFile> file= createECLSourceFile("","");
  182. if(clusterName && *clusterName)
  183. {
  184. file->setFileCluster(clusterName);
  185. }
  186. if (version > 1.11)
  187. {
  188. Owned<IPropertyTreeIterator> filetrees= query.getElements("Subfile");
  189. if (filetrees->first())
  190. file->setIsSuperFile(true);
  191. }
  192. if (fileName && *fileName)
  193. {
  194. file->setName(fileName);
  195. }
  196. file->setCount(fileCount);
  197. files.append(*file.getLink());
  198. }
  199. }
  200. else
  201. {
  202. StringArray fileNames;
  203. Owned<IPropertyTreeIterator> f=&cw->getFilesReadIterator();
  204. ForEach(*f)
  205. {
  206. IPropertyTree &query = f->query();
  207. const char *clusterName = query.queryProp("@cluster");
  208. const char *fileName = query.queryProp("@name");
  209. int fileCount = query.getPropInt("@useCount");
  210. bool bFound = false;
  211. if (fileName && *fileName && (fileNames.length() > 0))
  212. {
  213. for (unsigned i = 0; i < fileNames.length(); i++ )
  214. {
  215. const char *fileName0 = fileNames.item(i);
  216. if (!stricmp(fileName, fileName0))
  217. {
  218. bFound = true;
  219. break;
  220. }
  221. }
  222. }
  223. if (bFound)
  224. continue;
  225. Owned<IEspECLSourceFile> file= createECLSourceFile("","");
  226. if(clusterName && *clusterName)
  227. {
  228. file->setFileCluster(clusterName);
  229. }
  230. if (fileName && *fileName)
  231. {
  232. file->setName(fileName);
  233. }
  234. file->setCount(fileCount);
  235. Owned<IPropertyTreeIterator> filetrees= query.getElements("Subfile");
  236. if (filetrees->first())
  237. {
  238. file->setIsSuperFile(true);
  239. getSubFiles(filetrees, file, fileNames);
  240. }
  241. files.append(*file.getLink());
  242. }
  243. }
  244. info.setSourceFiles(files);
  245. }
  246. catch(IException* e)
  247. {
  248. StringBuffer eMsg;
  249. ERRLOG("%s", e->errorMessage(eMsg).str());
  250. info.setSourceFilesDesc(eMsg.str());
  251. e->Release();
  252. }
  253. }
  254. void WsWuInfo::getExceptions(IEspECLWorkunit &info, unsigned flags)
  255. {
  256. if ((flags & WUINFO_IncludeExceptions) || version > 1.16)
  257. {
  258. WsWUExceptions errors(*cw);
  259. if (version > 1.16)
  260. {
  261. info.setErrorCount(errors.ErrCount());
  262. info.setWarningCount(errors.WrnCount());
  263. info.setInfoCount(errors.InfCount());
  264. }
  265. if ((flags & WUINFO_IncludeExceptions))
  266. info.setExceptions(errors);
  267. }
  268. }
  269. void WsWuInfo::getVariables(IEspECLWorkunit &info, unsigned flags)
  270. {
  271. if (!(flags & WUINFO_IncludeVariables))
  272. return;
  273. try
  274. {
  275. IArrayOf<IEspECLResult> results;
  276. Owned<IConstWUResultIterator> vars = &cw->getVariables();
  277. ForEach(*vars)
  278. getResult(vars->query(), results, flags);
  279. info.setVariables(results);
  280. results.kill();
  281. }
  282. catch(IException* e)
  283. {
  284. StringBuffer eMsg;
  285. ERRLOG("%s", e->errorMessage(eMsg).str());
  286. info.setVariablesDesc(eMsg.str());
  287. e->Release();
  288. }
  289. }
  290. void WsWuInfo::addTimerToList(SCMStringBuffer& name, unsigned count, unsigned duration, unsigned& totalThorTimerCount,
  291. StringBuffer& totalThorTimeValue, IArrayOf<IEspECLTimer>& timers)
  292. {
  293. StringBuffer fd;
  294. formatDuration(fd, duration);
  295. if (strieq(name.str(), TOTALTHORTIME))
  296. {
  297. totalThorTimeValue = fd;
  298. totalThorTimerCount = count;
  299. return;
  300. }
  301. Owned<IEspECLTimer> t= createECLTimer("","");
  302. name.s.replace('_', ' ');
  303. t->setName(name.str());
  304. t->setValue(fd.str());
  305. t->setCount(count);
  306. if (version > 1.19)
  307. {
  308. StringAttr graphName;
  309. unsigned graphNum, subGraphNum, subId;
  310. if (parseGraphTimerLabel(name.str(), graphName, graphNum, subGraphNum, subId))
  311. {
  312. if (graphName.length() > 0)
  313. t->setGraphName(graphName);
  314. if (subId > 0)
  315. t->setSubGraphId((int)subId);
  316. }
  317. }
  318. timers.append(*t.getLink());
  319. }
  320. void WsWuInfo::getTimers(IEspECLWorkunit &info, unsigned flags)
  321. {
  322. if (!(flags & WUINFO_IncludeTimers))
  323. return;
  324. try
  325. {
  326. StringBuffer totalThorTimeValue;
  327. unsigned totalThorTimerCount = 0; //Do we need this?
  328. IArrayOf<IEspECLTimer> timers;
  329. Owned<IConstWUStatisticIterator> it = &cw->getStatistics();
  330. if (it->first())
  331. {
  332. ForEach(*it)
  333. {
  334. IConstWUStatistic & cur = it->query();
  335. //Only interested in timings...
  336. if (cur.getKind() != SMEASURE_TIME_NS)
  337. continue;
  338. SCMStringBuffer name;
  339. cur.getDescription(name);
  340. addTimerToList(name, cur.getCount(), nanoToMilli(cur.getValue()), totalThorTimerCount, totalThorTimeValue, timers);
  341. }
  342. }
  343. else
  344. {//backward compatibility for WU without Statistics
  345. Owned<IConstWUTimerIterator> it = &cw->getTimerIterator();
  346. ForEach(*it)
  347. {
  348. IConstWUTimer& timer = it->query();
  349. SCMStringBuffer name;
  350. timer.getName(name);
  351. addTimerToList(name, timer.getCount(), timer.getDuration(), totalThorTimerCount, totalThorTimeValue, timers);
  352. }
  353. }
  354. if (totalThorTimeValue.length() > 0)
  355. {
  356. Owned<IEspECLTimer> t= createECLTimer("","");
  357. t->setName(TOTALTHORTIME);
  358. t->setValue(totalThorTimeValue.str());
  359. t->setCount(totalThorTimerCount);
  360. timers.append(*t.getLink());
  361. }
  362. info.setTimers(timers);
  363. }
  364. catch(IException* e)
  365. {
  366. StringBuffer eMsg;
  367. ERRLOG("%s", e->errorMessage(eMsg).str());
  368. info.setTimersDesc(eMsg.str());
  369. e->Release();
  370. }
  371. }
  372. struct mapEnums { int val; const char *str; };
  373. mapEnums queryFileTypes[] = {
  374. { FileTypeCpp, "cpp" },
  375. { FileTypeDll, "dll" },
  376. { FileTypeResText, "res" },
  377. { FileTypeHintXml, "hint" },
  378. { FileTypeXml, "xml" },
  379. { FileTypeSize, NULL },
  380. };
  381. const char *getEnumText(int value, mapEnums *map)
  382. {
  383. const char *defval = map->str;
  384. while (map->str)
  385. {
  386. if (value==map->val)
  387. return map->str;
  388. map++;
  389. }
  390. assertex(!"Unexpected value in setEnum");
  391. return defval;
  392. }
  393. void WsWuInfo::getHelpers(IEspECLWorkunit &info, unsigned flags)
  394. {
  395. try
  396. {
  397. IArrayOf<IEspECLHelpFile> helpers;
  398. Owned <IConstWUQuery> query = cw->getQuery();
  399. if(!query)
  400. {
  401. ERRLOG("Cannot get Query for this workunit.");
  402. info.setHelpersDesc("Cannot get Query for this workunit.");
  403. }
  404. else
  405. {
  406. SCMStringBuffer qname;
  407. query->getQueryShortText(qname);
  408. if(qname.length())
  409. {
  410. if((flags & WUINFO_TruncateEclTo64k) && (qname.length() > 64000))
  411. qname.setLen(qname.str(), 64000);
  412. IEspECLQuery* q=&info.updateQuery();
  413. q->setText(qname.str());
  414. }
  415. if (version > 1.34)
  416. {
  417. SCMStringBuffer mainDefinition;
  418. query->getQueryMainDefinition(mainDefinition);
  419. if(mainDefinition.length())
  420. {
  421. IEspECLQuery* q=&info.updateQuery();
  422. q->setQueryMainDefinition(mainDefinition.str());
  423. }
  424. }
  425. if (version > 1.30)
  426. {
  427. SCMStringBuffer qText;
  428. query->getQueryText(qText);
  429. if ((qText.length() > 0) && isArchiveQuery(qText.str()))
  430. info.setHasArchiveQuery(true);
  431. }
  432. for (unsigned i = 0; i < FileTypeSize; i++)
  433. getHelpFiles(query, (WUFileType) i, helpers);
  434. }
  435. getWorkunitThorLogInfo(helpers, info);
  436. if (cw->getWuidVersion() > 0)
  437. {
  438. Owned<IPropertyTreeIterator> eclAgents = cw->getProcesses("EclAgent", NULL);
  439. ForEach (*eclAgents)
  440. {
  441. StringBuffer logName;
  442. IPropertyTree& eclAgent = eclAgents->query();
  443. eclAgent.getProp("@log",logName);
  444. if (!logName.length())
  445. continue;
  446. Owned<IEspECLHelpFile> h= createECLHelpFile("","");
  447. h->setName(logName.str());
  448. h->setType(File_EclAgentLog);
  449. if (version >= 1.43)
  450. {
  451. offset_t fileSize;
  452. if (getFileSize(logName.str(), NULL, fileSize))
  453. h->setFileSize(fileSize);
  454. if (version >= 1.44)
  455. {
  456. if (eclAgent.hasProp("@pid"))
  457. h->setPID(eclAgent.getPropInt("@pid"));
  458. else
  459. h->setPID(cw->getAgentPID());
  460. }
  461. }
  462. helpers.append(*h.getLink());
  463. }
  464. }
  465. else // legacy wuid
  466. {
  467. Owned<IStringIterator> eclAgentLogs = cw->getLogs("EclAgent");
  468. ForEach (*eclAgentLogs)
  469. {
  470. SCMStringBuffer name;
  471. eclAgentLogs->str(name);
  472. if (name.length() < 1)
  473. continue;
  474. Owned<IEspECLHelpFile> h= createECLHelpFile("","");
  475. h->setName(name.str());
  476. h->setType(File_EclAgentLog);
  477. if (version >= 1.43)
  478. {
  479. offset_t fileSize;
  480. if (getFileSize(name.str(), NULL, fileSize))
  481. h->setFileSize(fileSize);
  482. }
  483. helpers.append(*h.getLink());
  484. break;
  485. }
  486. }
  487. info.setHelpers(helpers);
  488. }
  489. catch(IException* e)
  490. {
  491. StringBuffer eMsg;
  492. ERRLOG("%s", e->errorMessage(eMsg).str());
  493. info.setHelpersDesc(eMsg.str());
  494. e->Release();
  495. }
  496. }
  497. void WsWuInfo::getApplicationValues(IEspECLWorkunit &info, unsigned flags)
  498. {
  499. if (!(flags & WUINFO_IncludeApplicationValues))
  500. return;
  501. try
  502. {
  503. IArrayOf<IEspApplicationValue> av;
  504. Owned<IConstWUAppValueIterator> app(&cw->getApplicationValues());
  505. ForEach(*app)
  506. {
  507. IConstWUAppValue& val=app->query();
  508. SCMStringBuffer buf;
  509. Owned<IEspApplicationValue> t= createApplicationValue("","");
  510. t->setApplication(val.getApplication(buf).str());
  511. t->setValue(val.getValue(buf).str());
  512. t->setName(val.getName(buf).str());
  513. t->setValue(val.getValue(buf).str());
  514. av.append(*t.getLink());
  515. }
  516. info.setApplicationValues(av);
  517. }
  518. catch(IException* e)
  519. {
  520. StringBuffer eMsg;
  521. ERRLOG("%s", e->errorMessage(eMsg).str());
  522. info.setApplicationValuesDesc(eMsg.str());
  523. e->Release();
  524. }
  525. }
  526. void WsWuInfo::getDebugValues(IEspECLWorkunit &info, unsigned flags)
  527. {
  528. if (!(flags & WUINFO_IncludeDebugValues))
  529. {
  530. if (version >= 1.50)
  531. {
  532. unsigned debugValueCount = 0;
  533. Owned<IStringIterator> debugs(&cw->getDebugValues());
  534. ForEach(*debugs)
  535. debugValueCount++;
  536. info.setDebugValueCount(debugValueCount);
  537. }
  538. return;
  539. }
  540. try
  541. {
  542. IArrayOf<IEspDebugValue> dv;
  543. Owned<IStringIterator> debugs(&cw->getDebugValues());
  544. ForEach(*debugs)
  545. {
  546. SCMStringBuffer name, val;
  547. debugs->str(name);
  548. cw->getDebugValue(name.str(),val);
  549. Owned<IEspDebugValue> t= createDebugValue("","");
  550. t->setName(name.str());
  551. t->setValue(val.str());
  552. dv.append(*t.getLink());
  553. }
  554. if (version >= 1.50)
  555. info.setDebugValueCount(dv.length());
  556. info.setDebugValues(dv);
  557. }
  558. catch(IException* e)
  559. {
  560. StringBuffer eMsg;
  561. ERRLOG("%s", e->errorMessage(eMsg).str());
  562. info.setDebugValuesDesc(eMsg.str());
  563. e->Release();
  564. }
  565. }
  566. const char *getGraphNum(const char *s,unsigned &num)
  567. {
  568. while (*s && !isdigit(*s))
  569. s++;
  570. num = 0;
  571. while (isdigit(*s))
  572. {
  573. num = num*10+*s-'0';
  574. s++;
  575. }
  576. return s;
  577. }
  578. void WsWuInfo::getGraphInfo(IEspECLWorkunit &info, unsigned flags)
  579. {
  580. if (version > 1.01)
  581. {
  582. info.setHaveSubGraphTimings(false);
  583. Owned<IStringIterator> times = &cw->getTimers();
  584. ForEach(*times)
  585. {
  586. SCMStringBuffer name;
  587. times->str(name);
  588. StringAttr graphName;
  589. unsigned graphNum;
  590. unsigned subGraphNum;
  591. unsigned subId;
  592. if (parseGraphTimerLabel(name.str(), graphName, graphNum, subGraphNum, subId))
  593. {
  594. info.setHaveSubGraphTimings(true);
  595. break;
  596. }
  597. }
  598. }
  599. if (!(flags & WUINFO_IncludeGraphs))
  600. return;
  601. try
  602. {
  603. SCMStringBuffer runningGraph;
  604. WUGraphIDType id;
  605. WUState st = cw->getState();
  606. bool running = (!(st==WUStateFailed || st==WUStateAborted || st==WUStateCompleted) && cw->getRunningGraph(runningGraph,id));
  607. IArrayOf<IEspECLGraph> graphs;
  608. Owned<IConstWUGraphIterator> it = &cw->getGraphs(GraphTypeAny);
  609. ForEach(*it)
  610. {
  611. IConstWUGraph &graph = it->query();
  612. if(!graph.isValid())
  613. continue;
  614. SCMStringBuffer name, label, type;
  615. graph.getName(name);
  616. graph.getLabel(label);
  617. graph.getTypeName(type);
  618. Owned<IEspECLGraph> g= createECLGraph("","");
  619. g->setName(name.str());
  620. g->setLabel(label.str());
  621. g->setType(type.str());
  622. if(running && strcmp(name.str(),runningGraph.str())==0)
  623. {
  624. g->setRunning(true);
  625. g->setRunningId(id);
  626. }
  627. Owned<IConstWUGraphProgress> progress = cw->getGraphProgress(name.str());
  628. if (progress)
  629. {
  630. WUGraphState graphstate= progress->queryGraphState();
  631. if (graphstate == WUGraphComplete)
  632. g->setComplete(true);
  633. if (version > 1.13 && graphstate == WUGraphFailed)
  634. {
  635. g->setFailed(true);
  636. }
  637. }
  638. graphs.append(*g.getLink());
  639. }
  640. info.setGraphs(graphs);
  641. }
  642. catch(IException* e)
  643. {
  644. StringBuffer eMsg;
  645. ERRLOG("%s", e->errorMessage(eMsg).str());
  646. info.setGraphsDesc(eMsg.str());
  647. e->Release();
  648. }
  649. }
  650. void WsWuInfo::getGraphTimingData(IArrayOf<IConstECLTimingData> &timingData, unsigned flags)
  651. {
  652. Owned<IStringIterator> times = &cw->getTimers();
  653. ForEach(*times)
  654. {
  655. SCMStringBuffer name;
  656. times->str(name);
  657. StringAttr graphName;
  658. unsigned graphNum;
  659. unsigned subGraphNum;
  660. unsigned subId;
  661. if (parseGraphTimerLabel(name.str(), graphName, graphNum, subGraphNum, subId))
  662. {
  663. unsigned time = cw->getTimerDuration(name.str());
  664. Owned<IEspECLTimingData> g = createECLTimingData();
  665. g->setName(name.str());
  666. g->setGraphNum(graphNum);
  667. g->setSubGraphNum(subGraphNum);
  668. g->setGID(subId);
  669. g->setMS(time);
  670. g->setMin(time/60000);
  671. timingData.append(*g.getClear());
  672. }
  673. }
  674. }
  675. void WsWuInfo::getRoxieCluster(IEspECLWorkunit &info, unsigned flags)
  676. {
  677. if (version > 1.06)
  678. {
  679. Owned<IConstWURoxieQueryInfo> roxieQueryInfo = cw->getRoxieQueryInfo();
  680. if (roxieQueryInfo)
  681. {
  682. SCMStringBuffer roxieClusterName;
  683. roxieQueryInfo->getRoxieClusterName(roxieClusterName);
  684. info.setRoxieCluster(roxieClusterName.str());
  685. }
  686. }
  687. }
  688. void WsWuInfo::getEventScheduleFlag(IEspECLWorkunit &info)
  689. {
  690. info.setEventSchedule(0);
  691. if (info.getState() && !stricmp(info.getState(), "wait"))
  692. {
  693. info.setEventSchedule(2); //Can deschedule
  694. }
  695. else
  696. {
  697. Owned<IConstWorkflowItemIterator> it = cw->getWorkflowItems();
  698. if (it)
  699. {
  700. ForEach(*it)
  701. {
  702. IConstWorkflowItem *r = it->query();
  703. if (!r)
  704. continue;
  705. IWorkflowEvent *wfevent = r->getScheduleEvent();
  706. if (!wfevent)
  707. continue;
  708. if (!r->hasScheduleCount() || (r->queryScheduleCountRemaining() > 0))
  709. {
  710. info.setEventSchedule(1); //Can reschedule
  711. break;
  712. }
  713. }
  714. }
  715. }
  716. }
  717. void WsWuInfo::getCommon(IEspECLWorkunit &info, unsigned flags)
  718. {
  719. SCMStringBuffer s;
  720. info.setWuid(cw->getWuid(s).str());
  721. info.setProtected(cw->isProtected() ? 1 : 0);
  722. info.setJobname(cw->getJobName(s).str());
  723. info.setOwner(cw->getUser(s).str());
  724. info.setCluster(cw->getClusterName(clusterName).str());
  725. info.setSnapshot(cw->getSnapshot(s).str());
  726. if ((cw->getState() == WUStateScheduled) && cw->aborting())
  727. {
  728. info.setStateID(WUStateAborting);
  729. info.setState("aborting");
  730. }
  731. else
  732. {
  733. info.setStateID(cw->getState());
  734. info.setState(cw->getStateDesc(s).str());
  735. }
  736. if (cw->isPausing())
  737. info.setIsPausing(true);
  738. getEventScheduleFlag(info);
  739. if (version > 1.27)
  740. {
  741. StringBuffer totalThorTimeStr;
  742. unsigned totalThorTimeMS = cw->getTimerDuration(TOTALTHORTIME);
  743. formatDuration(totalThorTimeStr, totalThorTimeMS);
  744. info.setTotalThorTime(totalThorTimeStr.str());
  745. }
  746. WsWuDateTime dt;
  747. cw->getTimeScheduled(dt);
  748. if(dt.isValid())
  749. info.setDateTimeScheduled(dt.getString(s).str());
  750. getRoxieCluster(info, flags);
  751. }
  752. void WsWuInfo::getInfo(IEspECLWorkunit &info, unsigned flags)
  753. {
  754. getCommon(info, flags);
  755. SecAccessFlags accessFlag = getWsWorkunitAccess(context, *cw);
  756. info.setAccessFlag(accessFlag);
  757. SCMStringBuffer s;
  758. info.setStateEx(cw->getStateEx(s).str());
  759. info.setPriorityClass(cw->getPriority());
  760. info.setPriorityLevel(cw->getPriorityLevel());
  761. info.setScope(cw->getWuScope(s).str());
  762. info.setActionEx(cw->getActionEx(s).str());
  763. info.setDescription(cw->getDebugValue("description", s).str());
  764. if (version > 1.21)
  765. info.setXmlParams(cw->getXmlParams(s).str());
  766. if (version >= 1.49)
  767. {
  768. SCMStringBuffer xml;
  769. exportWorkUnitToXML(cw, xml, true, false);
  770. info.setWUXMLSize(xml.length());
  771. }
  772. info.setResultLimit(cw->getResultLimit());
  773. info.setArchived(false);
  774. info.setGraphCount(cw->getGraphCount());
  775. info.setSourceFileCount(cw->getSourceFileCount());
  776. info.setVariableCount(cw->getVariableCount());
  777. info.setTimerCount(cw->getTimerCount());
  778. info.setSourceFileCount(cw->getSourceFileCount());
  779. info.setApplicationValueCount(cw->getApplicationValueCount());
  780. info.setHasDebugValue(cw->hasDebugValue("__calculated__complexity__"));
  781. getClusterInfo(info, flags);
  782. getExceptions(info, flags);
  783. getHelpers(info, flags);
  784. getGraphInfo(info, flags);
  785. getSourceFiles(info, flags);
  786. getResults(info, flags);
  787. getVariables(info, flags);
  788. getTimers(info, flags);
  789. getDebugValues(info, flags);
  790. getApplicationValues(info, flags);
  791. getWorkflow(info, flags);
  792. }
  793. unsigned WsWuInfo::getWorkunitThorLogInfo(IArrayOf<IEspECLHelpFile>& helpers, IEspECLWorkunit &info)
  794. {
  795. unsigned countThorLog = 0;
  796. IArrayOf<IConstThorLogInfo> thorLogList;
  797. if (cw->getWuidVersion() > 0)
  798. {
  799. SCMStringBuffer clusterName;
  800. cw->getClusterName(clusterName);
  801. if (!clusterName.length()) //Cluster name may not be set yet
  802. return countThorLog;
  803. Owned<IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(clusterName.str());
  804. if (!clusterInfo)
  805. {
  806. SCMStringBuffer wuid;
  807. WARNLOG("Cannot find TargetClusterInfo for workunit %s", cw->getWuid(wuid).str());
  808. return countThorLog;
  809. }
  810. unsigned numberOfSlaves = clusterInfo->getSize();
  811. BoolHash uniqueProcesses;
  812. Owned<IStringIterator> thorInstances = cw->getProcesses("Thor");
  813. ForEach (*thorInstances)
  814. {
  815. SCMStringBuffer processName;
  816. thorInstances->str(processName);
  817. if ((processName.length() < 1) || uniqueProcesses.getValue(processName.str()))
  818. continue;
  819. uniqueProcesses.setValue(processName.str(), true);
  820. StringBuffer groupName;
  821. getClusterThorGroupName(groupName, processName.str());
  822. Owned<IStringIterator> thorLogs = cw->getLogs("Thor", processName.str());
  823. ForEach (*thorLogs)
  824. {
  825. SCMStringBuffer logName;
  826. thorLogs->str(logName);
  827. if (logName.length() < 1)
  828. continue;
  829. countThorLog++;
  830. StringBuffer fileType;
  831. if (countThorLog < 2)
  832. fileType.append(File_ThorLog);
  833. else
  834. fileType.appendf("%s%d", File_ThorLog, countThorLog);
  835. Owned<IEspECLHelpFile> h= createECLHelpFile("","");
  836. h->setName(logName.str());
  837. h->setDescription(processName.str());
  838. h->setType(fileType.str());
  839. if (version >= 1.43)
  840. {
  841. offset_t fileSize;
  842. if (getFileSize(logName.str(), NULL, fileSize))
  843. h->setFileSize(fileSize);
  844. }
  845. helpers.append(*h.getLink());
  846. if (version < 1.38)
  847. continue;
  848. const char* pStr = logName.str();
  849. const char* ppStr = strstr(pStr, "/thormaster.");
  850. if (!ppStr)
  851. {
  852. WARNLOG("Invalid thorlog entry in workunit xml: %s", logName.str());
  853. continue;
  854. }
  855. ppStr += 12;
  856. StringBuffer logDate = ppStr;
  857. logDate.setLength(10);
  858. Owned<IEspThorLogInfo> thorLog = createThorLogInfo("","");
  859. thorLog->setProcessName(processName.str());
  860. thorLog->setClusterGroup(groupName.str());
  861. thorLog->setLogDate(logDate.str());
  862. thorLog->setNumberSlaves(numberOfSlaves);
  863. thorLogList.append(*thorLog.getLink());
  864. }
  865. }
  866. }
  867. else //legacy wuid
  868. {
  869. Owned<IStringIterator> thorLogs = cw->getLogs("Thor");
  870. ForEach (*thorLogs)
  871. {
  872. SCMStringBuffer name;
  873. thorLogs->str(name);
  874. if (name.length() < 1)
  875. continue;
  876. countThorLog++;
  877. StringBuffer fileType;
  878. if (countThorLog < 2)
  879. fileType.append(File_ThorLog);
  880. else
  881. fileType.appendf("%s%d", File_ThorLog, countThorLog);
  882. Owned<IEspECLHelpFile> h= createECLHelpFile("","");
  883. h->setName(name.str());
  884. h->setType(fileType.str());
  885. if (version >= 1.43)
  886. {
  887. offset_t fileSize;
  888. if (getFileSize(name.str(), NULL, fileSize))
  889. h->setFileSize(fileSize);
  890. }
  891. helpers.append(*h.getLink());
  892. }
  893. StringBuffer logDir;
  894. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory();
  895. Owned<IConstEnvironment> constEnv = envFactory->openEnvironmentByFile();
  896. Owned<IPropertyTree> logTree = &constEnv->getPTree();
  897. if (logTree)
  898. logTree->getProp("EnvSettings/log", logDir);
  899. if (logDir.length() > 0)
  900. {
  901. Owned<IStringIterator> debugs = cw->getLogs("Thor");
  902. ForEach(*debugs)
  903. {
  904. SCMStringBuffer val;
  905. debugs->str(val);
  906. if (val.length() < 1)
  907. continue;
  908. const char* pStr = val.str();
  909. const char* ppStr = strstr(pStr, logDir.str());
  910. if (!ppStr)
  911. {
  912. WARNLOG("Invalid thorlog entry in workunit xml: %s", val.str());
  913. continue;
  914. }
  915. const char* pProcessName = ppStr + logDir.length();
  916. char sep = pProcessName[0];
  917. StringBuffer processName = pProcessName + 1;
  918. ppStr = strchr(pProcessName + 1, sep);
  919. if (!ppStr)
  920. {
  921. WARNLOG("Invalid thorlog entry in workunit xml: %s", val.str());
  922. continue;
  923. }
  924. processName.setLength(ppStr - pProcessName - 1);
  925. StringBuffer groupName;
  926. getClusterThorGroupName(groupName, processName.str());
  927. StringBuffer logDate = ppStr + 12;
  928. logDate.setLength(10);
  929. Owned<IEspThorLogInfo> thorLog = createThorLogInfo("","");
  930. thorLog->setProcessName(processName.str());
  931. thorLog->setClusterGroup(groupName.str());
  932. thorLog->setLogDate(logDate.str());
  933. //for legacy wuid, the log name does not contain slaveNum. So, a user may not specify
  934. //a slaveNum and we only display the first slave log if > 1 per IP.
  935. thorLog->setNumberSlaves(0);
  936. thorLogList.append(*thorLog.getLink());
  937. }
  938. }
  939. }
  940. if (thorLogList.length() > 0)
  941. info.setThorLogList(thorLogList);
  942. thorLogList.kill();
  943. return countThorLog;
  944. }
  945. bool WsWuInfo::getClusterInfo(IEspECLWorkunit &info, unsigned flags)
  946. {
  947. if (version > 1.04)
  948. {
  949. StringArray allowedClusters;
  950. SCMStringBuffer val;
  951. cw->getAllowedClusters(val);
  952. if (val.length() > 0)
  953. {
  954. const char* ptr = val.str();
  955. while(*ptr != '\0')
  956. {
  957. StringBuffer onesub;
  958. while(*ptr != '\0' && *ptr != ',')
  959. {
  960. onesub.append((char)(*ptr));
  961. ptr++;
  962. }
  963. if(onesub.length() > 0)
  964. allowedClusters.append(onesub.str());
  965. if(*ptr != '\0')
  966. ptr++;
  967. }
  968. }
  969. if (allowedClusters.length() > 0)
  970. info.setAllowedClusters(allowedClusters);
  971. }
  972. if (version > 1.23 && clusterName.length())
  973. {
  974. int clusterTypeFlag = 0;
  975. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory();
  976. Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
  977. Owned<IPropertyTree> root = &constEnv->getPTree();
  978. if (!root)
  979. throw MakeStringException(ECLWATCH_CANNOT_CONNECT_DALI,"Cannot connect to DALI server.");
  980. Owned<IConstWUClusterInfo> clusterInfo = getTargetClusterInfo(clusterName.str());
  981. if (clusterInfo.get())
  982. {//Set thor flag or roxie flag in order to display some options for thor or roxie
  983. ClusterType platform = clusterInfo->getPlatform();
  984. if (isThorCluster(platform))
  985. {
  986. clusterTypeFlag=1;
  987. if (version > 1.29)
  988. info.setThorLCR(ThorLCRCluster == platform);
  989. }
  990. else if (RoxieCluster == platform)
  991. clusterTypeFlag=2;
  992. }
  993. info.setClusterFlag(clusterTypeFlag);
  994. }
  995. return true;
  996. }
  997. void WsWuInfo::getWorkflow(IEspECLWorkunit &info, unsigned flags)
  998. {
  999. bool eventCountRemaining = false;
  1000. bool eventCountUnlimited = false;
  1001. try
  1002. {
  1003. info.setEventSchedule(0);
  1004. unsigned workflowsCount = 0;
  1005. IArrayOf<IConstECLWorkflow> workflows;
  1006. Owned<IConstWorkflowItemIterator> it = cw->getWorkflowItems();
  1007. if (it)
  1008. {
  1009. ForEach(*it)
  1010. {
  1011. IConstWorkflowItem *r = it->query();
  1012. if (r)
  1013. {
  1014. IWorkflowEvent *wfevent = r->getScheduleEvent();
  1015. if (wfevent)
  1016. {
  1017. Owned<IEspECLWorkflow> g;
  1018. if (flags & WUINFO_IncludeWorkflows)
  1019. {
  1020. StringBuffer id;
  1021. g.setown(createECLWorkflow("",""));
  1022. g->setWFID(id.appendf("%d", r->queryWfid()).str());
  1023. g->setEventName(wfevent->queryName());
  1024. g->setEventText(wfevent->queryText());
  1025. }
  1026. if (r->hasScheduleCount())
  1027. {
  1028. if (r->queryScheduleCountRemaining() > 0)
  1029. eventCountRemaining = true;
  1030. if (flags & WUINFO_IncludeWorkflows)
  1031. {
  1032. g->setCount(r->queryScheduleCount());
  1033. g->setCountRemaining(r->queryScheduleCountRemaining());
  1034. }
  1035. }
  1036. else
  1037. {
  1038. eventCountUnlimited = true;
  1039. }
  1040. workflowsCount++;
  1041. if (flags & WUINFO_IncludeWorkflows)
  1042. workflows.append(*g.getLink());
  1043. }
  1044. }
  1045. }
  1046. if (workflows.length() > 0)
  1047. info.setWorkflows(workflows);
  1048. workflows.kill();
  1049. }
  1050. if (version >= 1.50)
  1051. info.setWorkflowCount(workflowsCount);
  1052. }
  1053. catch(IException* e)
  1054. {
  1055. StringBuffer eMsg;
  1056. ERRLOG("%s", e->errorMessage(eMsg).str());
  1057. info.setWorkflowsDesc(eMsg.str());
  1058. e->Release();
  1059. }
  1060. if (info.getState() && !stricmp(info.getState(), "wait"))
  1061. info.setEventSchedule(2); //Can deschedule
  1062. else if (eventCountUnlimited || eventCountRemaining)
  1063. info.setEventSchedule(1); //Can reschedule
  1064. }
  1065. IDistributedFile* WsWuInfo::getLogicalFileData(IEspContext& context, const char* logicalName, bool& showFileContent)
  1066. {
  1067. StringBuffer username;
  1068. context.getUserID(username);
  1069. Owned<IUserDescriptor> userdesc(createUserDescriptor());
  1070. userdesc->set(username.str(), context.queryPassword());
  1071. Owned<IDistributedFile> df = queryDistributedFileDirectory().lookup(logicalName, userdesc);
  1072. if (!df)
  1073. return NULL;
  1074. bool blocked;
  1075. if (df->isCompressed(&blocked) && !blocked)
  1076. return df.getClear();
  1077. IPropertyTree& properties = df->queryAttributes();
  1078. const char * format = properties.queryProp("@format");
  1079. if (format && (stricmp(format,"csv")==0 || memicmp(format, "utf", 3) == 0))
  1080. {
  1081. showFileContent = true;
  1082. return df.getClear();
  1083. }
  1084. const char * recordEcl = properties.queryProp("ECL");
  1085. if (!recordEcl)
  1086. return df.getClear();
  1087. MultiErrorReceiver errs;
  1088. Owned<IHqlExpression> ret = ::parseQuery(recordEcl, &errs);
  1089. showFileContent = errs.errCount() == 0;
  1090. return df.getClear();
  1091. }
  1092. void WsWuInfo::getEclSchemaChildFields(IArrayOf<IEspECLSchemaItem>& schemas, IHqlExpression * expr, bool isConditional)
  1093. {
  1094. if(!expr)
  1095. return;
  1096. ForEachChild(idx, expr)
  1097. getEclSchemaFields(schemas, expr->queryChild(idx), isConditional);
  1098. }
  1099. void WsWuInfo::getEclSchemaFields(IArrayOf<IEspECLSchemaItem>& schemas, IHqlExpression * expr, bool isConditional)
  1100. {
  1101. if(!expr)
  1102. return;
  1103. int ret = expr->getOperator();
  1104. switch (ret)
  1105. {
  1106. case no_record:
  1107. getEclSchemaChildFields(schemas, expr, isConditional);
  1108. break;
  1109. case no_ifblock:
  1110. {
  1111. getEclSchemaChildFields(schemas, expr->queryChild(1), true);
  1112. break;
  1113. }
  1114. case no_field:
  1115. {
  1116. if (expr->hasAttribute(__ifblockAtom))
  1117. break;
  1118. ITypeInfo * type = expr->queryType();
  1119. IAtom * name = expr->queryName();
  1120. IHqlExpression * nameAttr = expr->queryAttribute(namedAtom);
  1121. StringBuffer outname;
  1122. if (nameAttr && nameAttr->queryChild(0) && nameAttr->queryChild(0)->queryValue())
  1123. nameAttr->queryChild(0)->queryValue()->getStringValue(outname);
  1124. else
  1125. outname.append(name).toLowerCase();
  1126. if(type)
  1127. {
  1128. type_t tc = type->getTypeCode();
  1129. if (tc == type_row)
  1130. {
  1131. getEclSchemaChildFields(schemas, expr->queryRecord(), isConditional);
  1132. }
  1133. else
  1134. {
  1135. if (type->getTypeCode() == type_alien)
  1136. {
  1137. IHqlAlienTypeInfo * alien = queryAlienType(type);
  1138. type = alien->queryPhysicalType();
  1139. }
  1140. Owned<IEspECLSchemaItem> schema = createECLSchemaItem("","");
  1141. StringBuffer eclType;
  1142. type->getECLType(eclType);
  1143. schema->setColumnName(outname);
  1144. schema->setColumnType(eclType.str());
  1145. schema->setColumnTypeCode(tc);
  1146. schema->setIsConditional(isConditional);
  1147. schemas.append(*schema.getClear());
  1148. }
  1149. }
  1150. break;
  1151. }
  1152. }
  1153. }
  1154. bool WsWuInfo::getResultEclSchemas(IConstWUResult &r, IArrayOf<IEspECLSchemaItem>& schemas)
  1155. {
  1156. SCMStringBuffer schema;
  1157. r.getResultEclSchema(schema);
  1158. if (!schema.length())
  1159. return false;
  1160. MultiErrorReceiver errs;
  1161. Owned<IHqlExpression> expr = ::parseQuery(schema.str(), &errs);
  1162. if (errs.errCount() != 0)
  1163. return false;
  1164. getEclSchemaFields(schemas, expr, false);
  1165. return true;
  1166. }
  1167. void WsWuInfo::getResult(IConstWUResult &r, IArrayOf<IEspECLResult>& results, unsigned flags)
  1168. {
  1169. SCMStringBuffer name;
  1170. r.getResultName(name);
  1171. SCMStringBuffer filename;
  1172. r.getResultLogicalName(filename);
  1173. bool showFileContent = false;
  1174. Owned<IDistributedFile> df = NULL;
  1175. if (filename.length())
  1176. df.setown(getLogicalFileData(context, filename.str(), showFileContent));
  1177. StringBuffer value, link;
  1178. if (r.getResultStatus() == ResultStatusUndefined)
  1179. value.set("[undefined]");
  1180. else if (r.isResultScalar())
  1181. {
  1182. try
  1183. {
  1184. SCMStringBuffer xml;
  1185. r.getResultXml(xml);
  1186. Owned<IPropertyTree> props = createPTreeFromXMLString(xml.str(), ipt_caseInsensitive);
  1187. IPropertyTree *val = props->queryPropTree("Row/*");
  1188. if(val)
  1189. value.set(val->queryProp(NULL));
  1190. else
  1191. {
  1192. Owned<IResultSetFactory> resultSetFactory = getSecResultSetFactory(context.querySecManager(), context.queryUser(), context.queryUserId(), context.queryPassword());
  1193. Owned<INewResultSet> result;
  1194. result.setown(resultSetFactory->createNewResultSet(&r, wuid.str()));
  1195. Owned<IResultSetCursor> cursor(result->createCursor());
  1196. cursor->first();
  1197. if (cursor->getIsAll(0))
  1198. {
  1199. value.set("<All/>");
  1200. }
  1201. else
  1202. {
  1203. Owned<IResultSetCursor> childCursor = cursor->getChildren(0);
  1204. if (childCursor)
  1205. {
  1206. ForEach(*childCursor)
  1207. {
  1208. StringBuffer out;
  1209. StringBufferAdaptor adaptor(out);
  1210. childCursor->getDisplayText(adaptor, 0);
  1211. if (!value.length())
  1212. value.append('[');
  1213. else
  1214. value.append(", ");
  1215. value.append('\'').append(out.str()).append('\'');
  1216. }
  1217. if (value.length())
  1218. value.append(']');
  1219. }
  1220. }
  1221. }
  1222. }
  1223. catch(...)
  1224. {
  1225. value.append("[value not available]");
  1226. }
  1227. }
  1228. else
  1229. {
  1230. value.append('[').append(r.getResultTotalRowCount()).append(" rows]");
  1231. if((r.getResultSequence()>=0) && (!filename.length() || (df && df->queryAttributes().hasProp("ECL"))))
  1232. link.append(r.getResultSequence());
  1233. }
  1234. Owned<IEspECLResult> result= createECLResult("","");
  1235. if (flags & WUINFO_IncludeEclSchemas)
  1236. {
  1237. IArrayOf<IEspECLSchemaItem> schemas;
  1238. if (getResultEclSchemas(r, schemas))
  1239. result->setECLSchemas(schemas);
  1240. }
  1241. if (flags & WUINFO_IncludeXmlSchema)
  1242. {
  1243. Owned<IResultSetFactory> resultSetFactory = getSecResultSetFactory(context.querySecManager(), context.queryUser(), context.queryUserId(), context.queryPassword());
  1244. Owned<INewResultSet> rs = resultSetFactory->createNewResultSet(&r, wuid.str());
  1245. Owned<IResultSetCursor> cursor(rs->createCursor());
  1246. SCMStringBuffer xsd;
  1247. const IResultSetMetaData & meta = cursor->queryResultSet()->getMetaData();
  1248. meta.getXmlXPathSchema(xsd, false);
  1249. result->setXmlSchema(xsd.str());
  1250. }
  1251. if (filename.length())
  1252. result->setShowFileContent(showFileContent);
  1253. result->setName(name.str());
  1254. result->setLink(link.str());
  1255. result->setSequence(r.getResultSequence());
  1256. result->setValue(value.str());
  1257. result->setFileName(filename.str());
  1258. result->setIsSupplied(r.getResultStatus() == ResultStatusSupplied);
  1259. result->setTotal(r.getResultTotalRowCount());
  1260. results.append(*result.getLink());
  1261. }
  1262. void WsWuInfo::getResults(IEspECLWorkunit &info, unsigned flags)
  1263. {
  1264. try
  1265. {
  1266. unsigned count = 0;
  1267. IArrayOf<IEspECLResult> results;
  1268. Owned<IConstWUResultIterator> it = &(cw->getResults());
  1269. ForEach(*it)
  1270. {
  1271. IConstWUResult &r = it->query();
  1272. if(r.getResultSequence()>=0)
  1273. {
  1274. if (flags & WUINFO_IncludeResults)
  1275. getResult(r, results, flags);
  1276. count++;
  1277. }
  1278. }
  1279. if (version >= 1.17)
  1280. info.setResultCount(count);
  1281. if ((flags & WUINFO_IncludeResults) && results.length() > 0)
  1282. info.setResults(results);
  1283. results.kill();
  1284. }
  1285. catch(IException* e)
  1286. {
  1287. StringBuffer eMsg;
  1288. ERRLOG("%s", e->errorMessage(eMsg).str());
  1289. info.setResultsDesc(eMsg.str());
  1290. e->Release();
  1291. }
  1292. }
  1293. bool WsWuInfo::getFileSize(const char* fileName, const char* IPAddress, offset_t& fileSize)
  1294. {
  1295. if (!fileName || !*fileName)
  1296. return false;
  1297. Owned<IFile> aFile;
  1298. if (!IPAddress || !*IPAddress)
  1299. {
  1300. aFile.setown(createIFile(fileName));
  1301. }
  1302. else
  1303. {
  1304. RemoteFilename rfn;
  1305. rfn.setRemotePath(fileName);
  1306. SocketEndpoint ep(IPAddress);
  1307. rfn.setIp(ep);
  1308. aFile.setown(createIFile(rfn));
  1309. }
  1310. if (!aFile)
  1311. return false;
  1312. bool isDir;
  1313. CDateTime modtime;
  1314. if (!aFile->getInfo(isDir, fileSize, modtime) || isDir)
  1315. return false;
  1316. return true;
  1317. }
  1318. void WsWuInfo::getHelpFiles(IConstWUQuery* query, WUFileType type, IArrayOf<IEspECLHelpFile>& helpers)
  1319. {
  1320. if (!query)
  1321. return;
  1322. Owned<IConstWUAssociatedFileIterator> iter = &query->getAssociatedFiles();
  1323. ForEach(*iter)
  1324. {
  1325. SCMStringBuffer name, Ip, description;
  1326. IConstWUAssociatedFile & cur = iter->query();
  1327. if (cur.getType() != type)
  1328. continue;
  1329. cur.getName(name);
  1330. Owned<IEspECLHelpFile> h= createECLHelpFile("","");
  1331. h->setName(name.str());
  1332. h->setType(getEnumText(type, queryFileTypes));
  1333. if (version > 1.31)
  1334. {
  1335. cur.getIp(Ip);
  1336. h->setIPAddress(Ip.str());
  1337. cur.getDescription(description);
  1338. if ((description.length() < 1) && (name.length() > 0))
  1339. {
  1340. const char* desc = pathTail(name.str());
  1341. if (desc && *desc)
  1342. description.set(desc);
  1343. }
  1344. if (description.length() < 1)
  1345. description.set("Help File");
  1346. h->setDescription(description.str());
  1347. if (version >= 1.43)
  1348. {
  1349. offset_t fileSize;
  1350. if (getFileSize(name.str(), Ip.str(), fileSize))
  1351. h->setFileSize(fileSize);
  1352. }
  1353. }
  1354. helpers.append(*h.getLink());
  1355. }
  1356. }
  1357. void WsWuInfo::getSubFiles(IPropertyTreeIterator* f, IEspECLSourceFile* eclSuperFile, StringArray& fileNames)
  1358. {
  1359. IArrayOf<IEspECLSourceFile> files;
  1360. ForEach(*f)
  1361. {
  1362. IPropertyTree &query = f->query();
  1363. const char *clusterName = query.queryProp("@cluster");
  1364. const char *fileName = query.queryProp("@name");
  1365. int fileCount = query.getPropInt("@useCount");
  1366. bool bFound = false;
  1367. if (fileName && *fileName && (fileNames.length() > 0))
  1368. {
  1369. for (unsigned i = 0; i < fileNames.length(); i++ )
  1370. {
  1371. const char *fileName0 = fileNames.item(i);
  1372. if (!stricmp(fileName, fileName0))
  1373. {
  1374. bFound = true;
  1375. break;
  1376. }
  1377. }
  1378. }
  1379. if (bFound)
  1380. continue;
  1381. Owned<IEspECLSourceFile> file= createECLSourceFile("","");
  1382. if(clusterName && *clusterName)
  1383. {
  1384. file->setFileCluster(clusterName);
  1385. }
  1386. if (fileName && *fileName)
  1387. {
  1388. file->setName(fileName);
  1389. fileNames.append(fileName);
  1390. }
  1391. file->setCount(fileCount);
  1392. Owned<IPropertyTreeIterator> filetrees= query.getElements("Subfile");
  1393. if (filetrees->first())
  1394. {
  1395. file->setIsSuperFile(true);
  1396. getSubFiles(filetrees, file, fileNames);
  1397. }
  1398. files.append(*file.getLink());
  1399. }
  1400. eclSuperFile->setECLSourceFiles(files);
  1401. return;
  1402. }
  1403. bool WsWuInfo::getResourceInfo(StringArray &viewnames, StringArray &urls, unsigned flags)
  1404. {
  1405. if (!(flags & (WUINFO_IncludeResultsViewNames | WUINFO_IncludeResourceURLs)))
  1406. return true;
  1407. try
  1408. {
  1409. Owned<IWuWebView> wv = createWuWebView(*cw, NULL, NULL, false);
  1410. if (wv)
  1411. {
  1412. if (flags & WUINFO_IncludeResultsViewNames)
  1413. wv->getResultViewNames(viewnames);
  1414. if (flags & WUINFO_IncludeResourceURLs)
  1415. wv->getResourceURLs(urls, NULL);
  1416. }
  1417. return true;
  1418. }
  1419. catch(IException* e)
  1420. {
  1421. StringBuffer eMsg;
  1422. ERRLOG("%s", e->errorMessage(eMsg).str());
  1423. e->Release();
  1424. }
  1425. return false;
  1426. }
  1427. unsigned WsWuInfo::getResourceURLCount()
  1428. {
  1429. try
  1430. {
  1431. Owned<IWuWebView> wv = createWuWebView(*cw, NULL, NULL, false);
  1432. if (wv)
  1433. return wv->getResourceURLCount();
  1434. }
  1435. catch(IException* e)
  1436. {
  1437. StringBuffer eMsg;
  1438. ERRLOG("%s", e->errorMessage(eMsg).str());
  1439. e->Release();
  1440. }
  1441. return 0;
  1442. }
  1443. void appendIOStreamContent(MemoryBuffer &mb, IFileIOStream *ios, bool forDownload)
  1444. {
  1445. StringBuffer line;
  1446. bool eof = false;
  1447. while (!eof)
  1448. {
  1449. line.clear();
  1450. loop
  1451. {
  1452. char c;
  1453. size32_t numRead = ios->read(1, &c);
  1454. if (!numRead)
  1455. {
  1456. eof = true;
  1457. break;
  1458. }
  1459. line.append(c);
  1460. if (c=='\n')
  1461. break;
  1462. }
  1463. mb.append(line.length(), line.str());
  1464. if (!forDownload && (mb.length() > 640000))
  1465. break;
  1466. }
  1467. }
  1468. void WsWuInfo::getWorkunitEclAgentLog(const char* fileName, const char* agentPid, MemoryBuffer& buf)
  1469. {
  1470. if(!fileName || !*fileName)
  1471. throw MakeStringException(ECLWATCH_ECLAGENT_LOG_NOT_FOUND,"Log file not specified");
  1472. Owned<IFile> rFile = createIFile(fileName);
  1473. if(!rFile)
  1474. throw MakeStringException(ECLWATCH_CANNOT_OPEN_FILE, "Cannot open file %s.", fileName);
  1475. OwnedIFileIO rIO = rFile->openShared(IFOread,IFSHfull);
  1476. if(!rIO)
  1477. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE, "Cannot read file %s.", fileName);
  1478. OwnedIFileIOStream ios = createBufferedIOStream(rIO);
  1479. StringBuffer line;
  1480. bool eof = false;
  1481. bool wuidFound = false;
  1482. StringBuffer pidstr;
  1483. if (agentPid && *agentPid)
  1484. pidstr.appendf(" %s ", agentPid);
  1485. else
  1486. pidstr.appendf(" %5d ", cw->getAgentPID());
  1487. char const * pidchars = pidstr.str();
  1488. while(!eof)
  1489. {
  1490. line.clear();
  1491. loop
  1492. {
  1493. char c;
  1494. size32_t numRead = ios->read(1, &c);
  1495. if (!numRead)
  1496. {
  1497. eof = true;
  1498. break;
  1499. }
  1500. line.append(c);
  1501. if (c=='\n')
  1502. break;
  1503. }
  1504. //Retain all rows that match a unique program instance - by retaining all rows that match a pid
  1505. if(strstr(line.str(), pidchars))
  1506. {
  1507. //Check if this is a new instance using line sequence number
  1508. if (strncmp(line.str(), "00000000", 8) == 0)
  1509. {
  1510. if (wuidFound) //If the correct instance has been found, return that instance before the next instance.
  1511. break;
  1512. //The last instance is not a correct instance. Clean the buf in order to start a new instance.
  1513. buf.clear();
  1514. }
  1515. //If we spot the workunit id anywhere in the tacing for this pid then assume it is the correct instance.
  1516. if(!wuidFound && strstr(line.str(), wuid.str()))
  1517. wuidFound = true;
  1518. buf.append(line.length(), line.str());
  1519. }
  1520. }
  1521. if (buf.length() < 1)
  1522. buf.append(47, "(Not found a log line related to this workunit)");
  1523. }
  1524. void WsWuInfo::getWorkunitThorLog(const char* fileName, MemoryBuffer& buf)
  1525. {
  1526. if(!fileName || !*fileName)
  1527. throw MakeStringException(ECLWATCH_ECLAGENT_LOG_NOT_FOUND,"Log file not specified");
  1528. Owned<IFile> rFile = createIFile(fileName);
  1529. if (!rFile)
  1530. throw MakeStringException(ECLWATCH_CANNOT_OPEN_FILE,"Cannot open file %s.",fileName);
  1531. OwnedIFileIO rIO = rFile->openShared(IFOread,IFSHfull);
  1532. if (!rIO)
  1533. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read file %s.",fileName);
  1534. OwnedIFileIOStream ios = createBufferedIOStream(rIO);
  1535. StringBuffer line;
  1536. bool eof = false;
  1537. bool include = false;
  1538. VStringBuffer startwuid("Started wuid=%s", wuid.str());
  1539. VStringBuffer endwuid("Finished wuid=%s", wuid.str());
  1540. const char *sw = startwuid.str();
  1541. const char *ew = endwuid.str();
  1542. while (!eof)
  1543. {
  1544. line.clear();
  1545. loop
  1546. {
  1547. char c;
  1548. size32_t numRead = ios->read(1, &c);
  1549. if (!numRead)
  1550. {
  1551. eof = true;
  1552. break;
  1553. }
  1554. line.append(c);
  1555. if (c=='\n')
  1556. break;
  1557. }
  1558. if (strstr(line.str(), sw))
  1559. include = true;
  1560. if (include)
  1561. buf.append(line.length(), line.str());
  1562. if (strstr(line.str(), ew))
  1563. include = false;
  1564. }
  1565. }
  1566. void WsWuInfo::getWorkunitThorSlaveLog(const char *groupName, const char *ipAddress, const char* logDate, const char* logDir, int slaveNum, MemoryBuffer& buf, bool forDownload)
  1567. {
  1568. if (isEmpty(logDir))
  1569. throw MakeStringException(ECLWATCH_INVALID_INPUT,"ThorSlave log path not specified.");
  1570. if (isEmpty(logDate))
  1571. throw MakeStringException(ECLWATCH_INVALID_INPUT,"ThorSlave log date not specified.");
  1572. StringBuffer slaveIPAddress, logName;
  1573. if (slaveNum > 0)
  1574. {
  1575. if (isEmpty(groupName))
  1576. throw MakeStringException(ECLWATCH_INVALID_INPUT,"Thor group not specified.");
  1577. Owned<IGroup> nodeGroup = queryNamedGroupStore().lookup(groupName);
  1578. if (!nodeGroup || (nodeGroup->ordinality() == 0))
  1579. {
  1580. WARNLOG("Node group %s not found", groupName);
  1581. return;
  1582. }
  1583. nodeGroup->queryNode(slaveNum-1).endpoint().getIpText(slaveIPAddress);
  1584. if (slaveIPAddress.length() < 1)
  1585. throw MakeStringException(ECLWATCH_INVALID_INPUT,"ThorSlave log network address not found.");
  1586. logName.appendf("thorslave.%d.%s.log", slaveNum, logDate);
  1587. }
  1588. else
  1589. {//legacy wuid: a user types in an IP address for a thor slave
  1590. if (isEmpty(ipAddress))
  1591. throw MakeStringException(ECLWATCH_INVALID_INPUT,"ThorSlave address not specified.");
  1592. //thorslave.10.239.219.6_20100.2012_05_23.log
  1593. logName.appendf("thorslave.%s*.%s.log", ipAddress, logDate);
  1594. const char* portPtr = strchr(ipAddress, '_');
  1595. if (!portPtr)
  1596. slaveIPAddress.append(ipAddress);
  1597. else
  1598. {
  1599. StringBuffer ipAddressStr = ipAddress;
  1600. ipAddressStr.setLength(portPtr - ipAddress);
  1601. slaveIPAddress.append(ipAddressStr.str());
  1602. }
  1603. }
  1604. RemoteFilename rfn;
  1605. rfn.setRemotePath(logDir);
  1606. SocketEndpoint ep(slaveIPAddress.str());
  1607. rfn.setIp(ep);
  1608. Owned<IFile> dir = createIFile(rfn);
  1609. Owned<IDirectoryIterator> diriter = dir->directoryFiles(logName.str());
  1610. if (!diriter->first())
  1611. throw MakeStringException(ECLWATCH_FILE_NOT_EXIST,"Cannot find Thor slave log file %s.", logName.str());
  1612. Linked<IFile> logfile = &diriter->query();
  1613. diriter.clear();
  1614. dir.clear();
  1615. // logfile is now the file to load
  1616. OwnedIFileIO rIO = logfile->openShared(IFOread,IFSHfull);
  1617. if (!rIO)
  1618. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read file %s.",logName.str());
  1619. OwnedIFileIOStream ios = createBufferedIOStream(rIO);
  1620. if (slaveNum > 0)
  1621. {
  1622. StringBuffer line;
  1623. bool eof = false;
  1624. bool include = false;
  1625. VStringBuffer startwuid("Started wuid=%s", wuid.str());
  1626. VStringBuffer endwuid("Finished wuid=%s", wuid.str());
  1627. const char *sw = startwuid.str();
  1628. const char *ew = endwuid.str();
  1629. while (!eof)
  1630. {
  1631. line.clear();
  1632. loop
  1633. {
  1634. char c;
  1635. size32_t numRead = ios->read(1, &c);
  1636. if (!numRead)
  1637. {
  1638. eof = true;
  1639. break;
  1640. }
  1641. line.append(c);
  1642. if (c=='\n')
  1643. break;
  1644. }
  1645. if (strstr(line.str(), sw))
  1646. include = true;
  1647. if (include)
  1648. buf.append(line.length(), line.str());
  1649. if (strstr(line.str(), ew))
  1650. include = false;
  1651. }
  1652. }
  1653. else
  1654. {//legacy wuid
  1655. appendIOStreamContent(buf, ios.get(), forDownload);
  1656. }
  1657. }
  1658. void WsWuInfo::getWorkunitResTxt(MemoryBuffer& buf)
  1659. {
  1660. Owned<IConstWUQuery> query = cw->getQuery();
  1661. if(!query)
  1662. throw MakeStringException(ECLWATCH_QUERY_NOT_FOUND_FOR_WU,"No query for workunit %s.",wuid.str());
  1663. SCMStringBuffer resname;
  1664. queryDllServer().getDll(query->getQueryResTxtName(resname).str(), buf);
  1665. }
  1666. void WsWuInfo::getWorkunitArchiveQuery(MemoryBuffer& buf)
  1667. {
  1668. Owned<IConstWUQuery> query = cw->getQuery();
  1669. if(!query)
  1670. throw MakeStringException(ECLWATCH_QUERY_NOT_FOUND_FOR_WU,"No query for workunit %s.",wuid.str());
  1671. SCMStringBuffer queryText;
  1672. query->getQueryText(queryText);
  1673. if ((queryText.length() < 1) || !isArchiveQuery(queryText.str()))
  1674. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Archive Query not found for workunit %s.", wuid.str());
  1675. buf.append(queryText.length(), queryText.str());
  1676. }
  1677. void WsWuInfo::getWorkunitDll(StringBuffer &dllname, MemoryBuffer& buf)
  1678. {
  1679. Owned<IConstWUQuery> query = cw->getQuery();
  1680. if(!query)
  1681. throw MakeStringException(ECLWATCH_QUERY_NOT_FOUND_FOR_WU,"No query for workunit %s.",wuid.str());
  1682. StringBufferAdaptor isvName(dllname);
  1683. query->getQueryDllName(isvName);
  1684. queryDllServer().getDll(dllname.str(), buf);
  1685. }
  1686. void WsWuInfo::getWorkunitXml(const char* plainText, MemoryBuffer& buf)
  1687. {
  1688. const char* header;
  1689. if (plainText && (!stricmp(plainText, "yes")))
  1690. header = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>";
  1691. else
  1692. header = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><?xml-stylesheet href=\"../esp/xslt/xmlformatter.xsl\" type=\"text/xsl\"?>";
  1693. SCMStringBuffer xml;
  1694. exportWorkUnitToXML(cw, xml, true, false);
  1695. buf.append(strlen(header), header);
  1696. buf.append(xml.length(), xml.str());
  1697. }
  1698. void WsWuInfo::getWorkunitCpp(const char *cppname, const char* description, const char* ipAddress, MemoryBuffer& buf, bool forDownload)
  1699. {
  1700. if (isEmpty(description))
  1701. throw MakeStringException(ECLWATCH_INVALID_INPUT, "File not specified.");
  1702. if (isEmpty(ipAddress))
  1703. throw MakeStringException(ECLWATCH_INVALID_INPUT, "File location not specified.");
  1704. if (isEmpty(cppname))
  1705. throw MakeStringException(ECLWATCH_INVALID_FILE_NAME, "File path not specified.");
  1706. RemoteFilename rfn;
  1707. rfn.setRemotePath(cppname);
  1708. SocketEndpoint ep(ipAddress);
  1709. rfn.setIp(ep);
  1710. Owned<IFile> cppfile = createIFile(rfn);
  1711. if (!cppfile)
  1712. throw MakeStringException(ECLWATCH_CANNOT_OPEN_FILE, "Cannot open %s.", description);
  1713. OwnedIFileIO rIO = cppfile->openShared(IFOread,IFSHfull);
  1714. if (!rIO)
  1715. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read %s.", description);
  1716. OwnedIFileIOStream ios = createBufferedIOStream(rIO);
  1717. if (!ios)
  1718. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read %s.", description);
  1719. appendIOStreamContent(buf, ios.get(), forDownload);
  1720. }
  1721. void WsWuInfo::getWorkunitAssociatedXml(const char* name, const char* ipAddress, const char* plainText,
  1722. const char* description, bool forDownload, MemoryBuffer& buf)
  1723. {
  1724. if (isEmpty(description)) //'File Name' as shown in WU Details page
  1725. throw MakeStringException(ECLWATCH_INVALID_INPUT, "File not specified.");
  1726. if (isEmpty(ipAddress))
  1727. throw MakeStringException(ECLWATCH_INVALID_INPUT, "File location not specified.");
  1728. if (isEmpty(name)) //file name with full path
  1729. throw MakeStringException(ECLWATCH_INVALID_FILE_NAME, "File path not specified.");
  1730. RemoteFilename rfn;
  1731. rfn.setRemotePath(name);
  1732. SocketEndpoint ep(ipAddress);
  1733. rfn.setIp(ep);
  1734. Owned<IFile> rFile = createIFile(rfn);
  1735. if (!rFile)
  1736. throw MakeStringException(ECLWATCH_CANNOT_OPEN_FILE, "Cannot open %s.", description);
  1737. OwnedIFileIO rIO = rFile->openShared(IFOread,IFSHfull);
  1738. if (!rIO)
  1739. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read %s.", description);
  1740. OwnedIFileIOStream ios = createBufferedIOStream(rIO);
  1741. if (!ios)
  1742. throw MakeStringException(ECLWATCH_CANNOT_READ_FILE,"Cannot read %s.", description);
  1743. const char* header;
  1744. if (plainText && (!stricmp(plainText, "yes")))
  1745. header = "<?xml version=\"1.0\" encoding=\"UTF-8\"?>";
  1746. else
  1747. header = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><?xml-stylesheet href=\"../esp/xslt/xmlformatter.xsl\" type=\"text/xsl\"?>";
  1748. buf.append(strlen(header), header);
  1749. appendIOStreamContent(buf, ios.get(), forDownload);
  1750. }
  1751. WsWuSearch::WsWuSearch(IEspContext& context,const char* owner,const char* state,const char* cluster,const char* startDate,const char* endDate,const char* ecl,const char* jobname,const char* appname,const char* appkey,const char* appvalue)
  1752. {
  1753. SecAccessFlags accessOwn;
  1754. SecAccessFlags accessOthers;
  1755. getUserWuAccessFlags(context, accessOwn, accessOthers, true);
  1756. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  1757. StringBuffer xpath("*");
  1758. if(ecl && *ecl)
  1759. xpath.append("[Query/Text=?~\"*").append(ecl).append("*\"]");
  1760. if(state && *state)
  1761. xpath.append("[@state=\"").append(state).append("\"]");
  1762. if(cluster && *cluster)
  1763. xpath.append("[@clusterName=\"").append(cluster).append("\"]");
  1764. if(owner && *owner)
  1765. xpath.append("[@submitID=?~\"").append(owner).append("\"]");
  1766. if(jobname && *jobname)
  1767. xpath.append("[@jobName=?~\"*").append(jobname).append("*\"]");
  1768. if((appname && *appname) || (appkey && *appkey) || (appvalue && *appvalue))
  1769. {
  1770. xpath.append("[Application/").append(appname && *appname ? appname : "*");
  1771. xpath.append("/").append(appkey && *appkey ? appkey : "*");
  1772. if(appvalue && *appvalue)
  1773. xpath.append("=?~\"").append(appvalue).append("\"");
  1774. xpath.append("]");
  1775. }
  1776. Owned<IConstWorkUnitIterator> it(factory->getWorkUnitsByXPath(xpath.str()));
  1777. StringBuffer wuFrom, wuTo;
  1778. if(startDate && *startDate)
  1779. createWuidFromDate(startDate, wuFrom);
  1780. if(endDate && *endDate)
  1781. createWuidFromDate(endDate, wuTo);
  1782. ForEach(*it)
  1783. {
  1784. IConstWorkUnit &cw = it->query();
  1785. if (chooseWuAccessFlagsByOwnership(context.queryUserId(), cw, accessOwn, accessOthers) < SecAccess_Read)
  1786. continue;
  1787. SCMStringBuffer wuid;
  1788. cw.getWuid(wuid);
  1789. if (wuFrom.length() && strcmp(wuid.str(),wuFrom.str())<0)
  1790. continue;
  1791. if (wuTo.length() && strcmp(wuid.str(),wuTo.str())>0)
  1792. continue;
  1793. if (state && *state)
  1794. {
  1795. SCMStringBuffer descr;
  1796. if(!strieq(cw.getStateDesc(descr).str(),state))
  1797. continue;
  1798. }
  1799. SCMStringBuffer parent;
  1800. if (!cw.getParentWuid(parent).length())
  1801. {
  1802. parent.clear();
  1803. wuids.push_back(cw.getWuid(parent).str());
  1804. }
  1805. }
  1806. std::sort(wuids.begin(),wuids.end(),std::greater<std::string>());
  1807. }
  1808. StringBuffer& WsWuSearch::createWuidFromDate(const char* timestamp,StringBuffer& s)
  1809. {
  1810. CDateTime wuTime;
  1811. wuTime.setString(timestamp,NULL,true);
  1812. unsigned year, month, day, hour, minute, second, nano;
  1813. wuTime.getDate(year, month, day, true);
  1814. wuTime.getTime(hour, minute, second, nano, true);
  1815. s.appendf("W%4d%02d%02d-%02d%02d%02d",year,month,day,hour,minute,second);
  1816. return s;
  1817. }
  1818. struct CompareData
  1819. {
  1820. CompareData(const char* _filter): filter(_filter) {}
  1821. bool operator()(const Linked<DataCacheElement>& e) const
  1822. {
  1823. return stricmp(e->m_filter.c_str(),filter)==0;
  1824. }
  1825. const char* filter;
  1826. };
  1827. DataCacheElement* DataCache::lookup(IEspContext &context, const char* filter, unsigned timeOutMin)
  1828. {
  1829. CriticalBlock block(crit);
  1830. if (cache.size() < 1)
  1831. return NULL;
  1832. //erase data if it should be
  1833. CDateTime timeNow;
  1834. int timeout = timeOutMin;
  1835. timeNow.setNow();
  1836. timeNow.adjustTime(-timeout);
  1837. while (true)
  1838. {
  1839. std::list<Linked<DataCacheElement> >::iterator list_iter = cache.begin();
  1840. if (list_iter == cache.end())
  1841. break;
  1842. DataCacheElement* awu = list_iter->get();
  1843. if (!awu || (awu->m_timeCached > timeNow))
  1844. break;
  1845. cache.pop_front();
  1846. }
  1847. if (cache.size() < 1)
  1848. return NULL;
  1849. //Check whether we have the data cache for this cluster. If yes, get the version
  1850. std::list<Linked<DataCacheElement> >::iterator it = std::find_if(cache.begin(),cache.end(),CompareData(filter));
  1851. if(it!=cache.end())
  1852. {
  1853. return it->getLink();
  1854. }
  1855. return NULL;
  1856. }
  1857. void DataCache::add(const char* filter, const char* data, const char* name, const char* localName, const char* wuid,
  1858. const char* resultName, unsigned seq, __int64 start, unsigned count, __int64 requested, __int64 total)
  1859. {
  1860. CriticalBlock block(crit);
  1861. //Save new data
  1862. Owned<DataCacheElement> e=new DataCacheElement(filter, data, name, localName, wuid, resultName, seq, start, count, requested, total);
  1863. if (cacheSize > 0)
  1864. {
  1865. if (cache.size() >= cacheSize)
  1866. cache.pop_front();
  1867. cache.push_back(e.get());
  1868. }
  1869. return;
  1870. }
  1871. struct CompareArchivedWUs
  1872. {
  1873. CompareArchivedWUs(const char* _filter): filter(_filter) {}
  1874. bool operator()(const Linked<ArchivedWuCacheElement>& e) const
  1875. {
  1876. return stricmp(e->m_filter.c_str(),filter)==0;
  1877. }
  1878. const char* filter;
  1879. };
  1880. ArchivedWuCacheElement* ArchivedWuCache::lookup(IEspContext &context, const char* filter, const char* sashaUpdatedWhen, unsigned timeOutMin)
  1881. {
  1882. CriticalBlock block(crit);
  1883. if (cache.size() < 1)
  1884. return NULL;
  1885. //erase data if it should be
  1886. CDateTime timeNow;
  1887. int timeout = timeOutMin;
  1888. timeNow.setNow();
  1889. timeNow.adjustTime(-timeout);
  1890. while (true)
  1891. {
  1892. std::list<Linked<ArchivedWuCacheElement> >::iterator list_iter = cache.begin();
  1893. if (list_iter == cache.end())
  1894. break;
  1895. ArchivedWuCacheElement* awu = list_iter->get();
  1896. if (awu && !stricmp(sashaUpdatedWhen, awu->m_sashaUpdatedWhen.c_str()) && (awu->m_timeCached > timeNow))
  1897. break;
  1898. cache.pop_front();
  1899. }
  1900. if (cache.size() < 1)
  1901. return NULL;
  1902. //Check whether we have the data cache for this cluster. If yes, get the version
  1903. std::list<Linked<ArchivedWuCacheElement> >::iterator it = std::find_if(cache.begin(),cache.end(),CompareArchivedWUs(filter));
  1904. if(it!=cache.end())
  1905. return it->getLink();
  1906. return NULL;
  1907. }
  1908. void ArchivedWuCache::add(const char* filter, const char* sashaUpdatedWhen, bool hasNextPage, unsigned numWUsReturned, IArrayOf<IEspECLWorkunit>& wus)
  1909. {
  1910. CriticalBlock block(crit);
  1911. //Save new data
  1912. Owned<ArchivedWuCacheElement> e=new ArchivedWuCacheElement(filter, sashaUpdatedWhen, hasNextPage, numWUsReturned, wus);
  1913. if (cacheSize > 0)
  1914. {
  1915. if (cache.size() >= cacheSize)
  1916. cache.pop_front();
  1917. cache.push_back(e.get());
  1918. }
  1919. return;
  1920. }
  1921. WsWuJobQueueAuditInfo::WsWuJobQueueAuditInfo(IEspContext &context, const char *cluster, const char *from , const char *to, CHttpResponse* response, const char *xls)
  1922. {
  1923. if(!response)
  1924. return;
  1925. unsigned maxDisplay = 125;
  1926. IArrayOf<IEspThorQueue> items;
  1927. CDateTime fromTime;
  1928. CDateTime toTime;
  1929. StringBuffer fromstr;
  1930. StringBuffer tostr;
  1931. if(from && *from)
  1932. {
  1933. fromTime.setString(from,NULL,false);
  1934. fromTime.getString(fromstr, false);
  1935. }
  1936. if(to && *to)
  1937. {
  1938. toTime.setString(to,NULL,false);
  1939. toTime.getString(tostr, false);
  1940. }
  1941. StringBuffer filter("ThorQueueMonitor");
  1942. if(notEmpty(cluster))
  1943. filter.appendf(",%s", cluster);
  1944. StringAttrArray lines;
  1945. queryAuditLogs(fromTime, toTime, filter.str(), lines);
  1946. unsigned countLines = 0;
  1947. unsigned maxConnected = 0;
  1948. unsigned longestQueue = 0;
  1949. ForEachItemIn(idx, lines)
  1950. {
  1951. const char* line = lines.item(idx).text;
  1952. if(!line || !*line)
  1953. continue;
  1954. if (idx < (lines.length() - 1))
  1955. getAuditLineInfo(line, longestQueue, maxConnected, maxDisplay, 1, items);
  1956. else
  1957. getAuditLineInfo(line, longestQueue, maxConnected, maxDisplay, 2, items);
  1958. countLines++;
  1959. }
  1960. StringBuffer responsebuf;
  1961. if (items.length() < 1)
  1962. {
  1963. responsebuf.append("<script language=\"javascript\">\r\nparent.displayQEnd(\'No data found\')</script>\r\n");
  1964. response->sendChunk(responsebuf.str());
  1965. return;
  1966. }
  1967. unsigned itemCount = items.length();
  1968. if (itemCount > maxDisplay)
  1969. itemCount = maxDisplay;
  1970. responsebuf.append("<script language=\"javascript\">parent.displayQLegend()</script>\r\n");
  1971. response->sendChunk(responsebuf.str());
  1972. responsebuf.clear();
  1973. responsebuf.append("<script language=\"javascript\">parent.displayQBegin(").append(longestQueue).append(",").append(maxConnected).append(",").append(itemCount).append(")</script>\r\n");
  1974. response->sendChunk(responsebuf.str());
  1975. responsebuf.clear();
  1976. responsebuf.append("<script language=\"javascript\">\r\n");
  1977. //bool displayDT = false;
  1978. unsigned count = 0;
  1979. unsigned jobpending=0;
  1980. ForEachItemIn(i,items)
  1981. {
  1982. IEspThorQueue& tq = items.item(i);
  1983. //displayDT = !displayDT;
  1984. count++;
  1985. if (count > maxDisplay)
  1986. break;
  1987. StringBuffer countStr, dtStr;
  1988. countStr.appendulong(count);
  1989. //if (displayDT)
  1990. dtStr = tq.getDT();
  1991. responsebuf.append("parent.displayQueue(\'").append(count).append("\',\'").append(dtStr.str()).append("\',\'").append(tq.getRunningWUs()).append("\',");
  1992. responsebuf.append("\'").append(tq.getQueuedWUs()).append("\',\'").append(tq.getWaitingThors()).append("\',");
  1993. responsebuf.append("\'").append(tq.getConnectedThors()).append("\',\'").append(tq.getIdledThors()).append("\',");
  1994. responsebuf.append("\'").append(tq.getRunningWU1()).append("\',\'").append(tq.getRunningWU2()).append("\')\r\n");
  1995. if(++jobpending>=50)
  1996. {
  1997. responsebuf.append("</script>\r\n");
  1998. response->sendChunk(responsebuf.str());
  1999. responsebuf.clear();
  2000. responsebuf.append("<script language=\"javascript\">\r\n");
  2001. jobpending=0;
  2002. }
  2003. }
  2004. StringBuffer countStr;
  2005. countStr.appendulong(count);
  2006. StringBuffer msg("<table><tr><td>");
  2007. msg.append("Total Records in the Time Period: ").append(items.length()).append(" (<a href=\"/WsWorkunits/WUClusterJobQueueLOG?").append(xls).append("\">txt</a>...<a href=\"/WsWorkunits/WUClusterJobQueueXLS?").append(xls).append("\">xls</a>).");
  2008. msg.append("</td></tr><tr><td>");
  2009. if (count > maxDisplay)
  2010. msg.append("Displayed: First ").append(maxDisplay).append(". ");
  2011. msg.append("Max. Queue Length: ").append(longestQueue).append(".");
  2012. msg.append("</td></tr></table>");
  2013. responsebuf.append("parent.displayQEnd(\'").append(msg).append("\')</script>\r\n");
  2014. response->sendChunk(responsebuf.str());
  2015. }
  2016. void WsWuJobQueueAuditInfo::getAuditLineInfo(const char* line, unsigned& longestQueue, unsigned& maxConnected, unsigned maxDisplay, unsigned showAll, IArrayOf<IEspThorQueue>& items)
  2017. {
  2018. //2009-08-12 02:44:12 ,ThorQueueMonitor,thor400_88_dev,0,0,1,1,114,---,---
  2019. if(!line || !*line)
  2020. return;
  2021. Owned<IEspThorQueue> tq = createThorQueue();
  2022. StringBuffer dt, runningWUs, queuedWUs, waitingThors, connectedThors, idledThors, runningWU1, runningWU2;
  2023. // date/time
  2024. const char* bptr = line;
  2025. const char* eptr = strchr(bptr, ',');
  2026. if(eptr)
  2027. dt.append(eptr - bptr, bptr);
  2028. else
  2029. dt.append(bptr);
  2030. tq->setDT(dt.str());
  2031. if(!eptr)
  2032. {
  2033. if (checkNewThorQueueItem(tq, showAll, items))
  2034. items.append(*tq.getClear());
  2035. return;
  2036. }
  2037. //skip title
  2038. bptr = eptr + 1;
  2039. eptr = strchr(bptr, ',');
  2040. if(!eptr)
  2041. {
  2042. if (checkNewThorQueueItem(tq, showAll, items))
  2043. items.append(*tq.getClear());
  2044. return;
  2045. }
  2046. //skip queue name
  2047. bptr = eptr + 1;
  2048. eptr = strchr(bptr, ',');
  2049. if(!eptr)
  2050. {
  2051. if (checkNewThorQueueItem(tq, showAll, items))
  2052. items.append(*tq.getClear());
  2053. return;
  2054. }
  2055. //running
  2056. bptr = eptr + 1;
  2057. eptr = strchr(bptr, ',');
  2058. if(eptr)
  2059. runningWUs.append(eptr - bptr, bptr);
  2060. else
  2061. runningWUs.append(bptr);
  2062. tq->setRunningWUs(runningWUs.str());
  2063. if(!eptr)
  2064. {
  2065. if (checkNewThorQueueItem(tq, showAll, items))
  2066. items.append(*tq.getClear());
  2067. return;
  2068. }
  2069. //queued
  2070. bptr = eptr + 1;
  2071. eptr = strchr(bptr, ',');
  2072. if(eptr)
  2073. queuedWUs.append(eptr - bptr, bptr);
  2074. else
  2075. queuedWUs.append(bptr);
  2076. if (maxDisplay > items.length())
  2077. {
  2078. unsigned queueLen = atoi(queuedWUs.str());
  2079. if (queueLen > longestQueue)
  2080. longestQueue = queueLen;
  2081. }
  2082. tq->setQueuedWUs(queuedWUs.str());
  2083. if(!eptr)
  2084. {
  2085. if (checkNewThorQueueItem(tq, showAll, items))
  2086. items.append(*tq.getClear());
  2087. return;
  2088. }
  2089. //waiting
  2090. bptr = eptr + 1;
  2091. eptr = strchr(bptr, ',');
  2092. if(eptr)
  2093. waitingThors.append(eptr - bptr, bptr);
  2094. else
  2095. waitingThors.append(bptr);
  2096. tq->setWaitingThors(waitingThors.str());
  2097. if(!eptr)
  2098. {
  2099. if (checkNewThorQueueItem(tq, showAll, items))
  2100. items.append(*tq.getClear());
  2101. return;
  2102. }
  2103. //connected
  2104. bptr = eptr + 1;
  2105. eptr = strchr(bptr, ',');
  2106. if(eptr)
  2107. connectedThors.append(eptr - bptr, bptr);
  2108. else
  2109. connectedThors.append(bptr);
  2110. if (maxDisplay > items.length())
  2111. {
  2112. unsigned connnectedLen = atoi(connectedThors.str());
  2113. if (connnectedLen > maxConnected)
  2114. maxConnected = connnectedLen;
  2115. }
  2116. tq->setConnectedThors(connectedThors.str());
  2117. if(!eptr)
  2118. {
  2119. if (checkNewThorQueueItem(tq, showAll, items))
  2120. items.append(*tq.getClear());
  2121. return;
  2122. }
  2123. //idled
  2124. bptr = eptr + 1;
  2125. eptr = strchr(bptr, ',');
  2126. if(eptr)
  2127. idledThors.append(eptr - bptr, bptr);
  2128. else
  2129. idledThors.append(bptr);
  2130. tq->setIdledThors(idledThors.str());
  2131. if(!eptr)
  2132. {
  2133. items.append(*tq.getClear());
  2134. return;
  2135. }
  2136. //runningWU1
  2137. bptr = eptr + 1;
  2138. eptr = strchr(bptr, ',');
  2139. if(eptr)
  2140. runningWU1.append(eptr - bptr, bptr);
  2141. else
  2142. {
  2143. runningWU1.append(bptr);
  2144. }
  2145. if (!strcmp(runningWU1.str(), "---"))
  2146. runningWU1.clear();
  2147. if (runningWU1.length() > 0)
  2148. tq->setRunningWU1(runningWU1.str());
  2149. if(!eptr)
  2150. {
  2151. if (checkNewThorQueueItem(tq, showAll, items))
  2152. items.append(*tq.getClear());
  2153. return;
  2154. }
  2155. //runningWU2
  2156. bptr = eptr + 1;
  2157. eptr = strchr(bptr, ',');
  2158. if(eptr)
  2159. runningWU2.append(eptr - bptr, bptr);
  2160. else
  2161. {
  2162. runningWU2.append(bptr);
  2163. }
  2164. if (!strcmp(runningWU2.str(), "---"))
  2165. runningWU2.clear();
  2166. if (runningWU2.length() > 0)
  2167. tq->setRunningWU2(runningWU2.str());
  2168. if (checkNewThorQueueItem(tq, showAll, items))
  2169. items.append(*tq.getClear());
  2170. DBGLOG("Queue log: [%s]", line);
  2171. }
  2172. bool WsWuJobQueueAuditInfo::checkSameStrings(const char* s1, const char* s2)
  2173. {
  2174. if (s1)
  2175. {
  2176. if (!s2)
  2177. return false;
  2178. if (strcmp(s1, s2))
  2179. return false;
  2180. }
  2181. else if (s2)
  2182. {
  2183. if (!s1)
  2184. return false;
  2185. }
  2186. return true;
  2187. }
  2188. bool WsWuJobQueueAuditInfo::checkNewThorQueueItem(IEspThorQueue* tq, unsigned showAll, IArrayOf<IEspThorQueue>& items)
  2189. {
  2190. bool bAdd = false;
  2191. if (showAll < 1) //show every lines
  2192. bAdd = true;
  2193. else if (items.length() < 1)
  2194. bAdd = true;
  2195. else if (showAll > 1) //last line now
  2196. {
  2197. IEspThorQueue& tq0 = items.item(items.length()-1);
  2198. if (!checkSameStrings(tq->getDT(), tq0.getDT()))
  2199. bAdd = true;
  2200. }
  2201. else
  2202. {
  2203. IEspThorQueue& tq0 = items.item(items.length()-1);
  2204. if (!checkSameStrings(tq->getRunningWUs(), tq0.getRunningWUs()))
  2205. bAdd = true;
  2206. if (!checkSameStrings(tq->getQueuedWUs(), tq0.getQueuedWUs()))
  2207. bAdd = true;
  2208. if (!checkSameStrings(tq->getConnectedThors(), tq0.getConnectedThors()))
  2209. bAdd = true;
  2210. if (!checkSameStrings(tq->getConnectedThors(), tq0.getConnectedThors()))
  2211. bAdd = true;
  2212. if (!checkSameStrings(tq->getRunningWU1(), tq0.getRunningWU1()))
  2213. bAdd = true;
  2214. if (!checkSameStrings(tq->getRunningWU2(), tq0.getRunningWU2()))
  2215. bAdd = true;
  2216. }
  2217. return bAdd;
  2218. }
  2219. void xsltTransform(const char* xml, const char* sheet, IProperties *params, StringBuffer& ret)
  2220. {
  2221. StringBuffer xsl;
  2222. if(!checkFileExists(sheet))
  2223. throw MakeStringException(ECLWATCH_FILE_NOT_EXIST, "Could not find stylesheet %s.",sheet);
  2224. Owned<IXslProcessor> proc = getXslProcessor();
  2225. Owned<IXslTransform> trans = proc->createXslTransform();
  2226. trans->setXmlSource(xml, strlen(xml));
  2227. trans->loadXslFromFile(sheet);
  2228. trans->copyParameters(params);
  2229. trans->transform(ret);
  2230. }
  2231. bool addToQueryString(StringBuffer &queryString, const char *name, const char *value, const char delim)
  2232. {
  2233. if (isEmpty(name) || isEmpty(value))
  2234. return false;
  2235. if (queryString.length() > 0)
  2236. queryString.append(delim);
  2237. queryString.append(name).append("=").append(value);
  2238. return true;
  2239. }
  2240. int WUSchedule::run()
  2241. {
  2242. try
  2243. {
  2244. while(!stopping)
  2245. {
  2246. Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
  2247. Owned<IConstWorkUnitIterator> itr = factory->getWorkUnitsByState(WUStateScheduled);
  2248. if (itr)
  2249. {
  2250. ForEach(*itr)
  2251. {
  2252. try
  2253. {
  2254. IConstWorkUnit & cw = itr->query();
  2255. if (cw.aborting())
  2256. {
  2257. WorkunitUpdate wu(&cw.lock());
  2258. wu->setState(WUStateAborted);
  2259. continue;
  2260. }
  2261. WsWuDateTime dt, now;
  2262. now.setNow();
  2263. cw.getTimeScheduled(dt);
  2264. if (now.compare(dt)>=0)
  2265. {
  2266. SCMStringBuffer wuid;
  2267. runWorkUnit(cw.getWuid(wuid).str());
  2268. }
  2269. }
  2270. catch(IException *e)
  2271. {
  2272. StringBuffer msg;
  2273. ERRLOG("Exception %d:%s in WsWorkunits Schedule::run", e->errorCode(), e->errorMessage(msg).str());
  2274. e->Release();
  2275. }
  2276. }
  2277. }
  2278. semSchedule.wait(1000*60);
  2279. }
  2280. }
  2281. catch(IException *e)
  2282. {
  2283. StringBuffer msg;
  2284. ERRLOG("Exception %d:%s in WsWorkunits Schedule::run", e->errorCode(), e->errorMessage(msg).str());
  2285. e->Release();
  2286. }
  2287. catch(...)
  2288. {
  2289. ERRLOG("Unknown exception in WsWorkunits Schedule::run");
  2290. }
  2291. if (m_container)
  2292. m_container->exitESP();
  2293. return 0;
  2294. }
  2295. void WsWuHelpers::setXmlParameters(IWorkUnit *wu, const char *xml, bool setJobname)
  2296. {
  2297. if (!xml || !*xml)
  2298. return;
  2299. Owned<IPropertyTree> tree = createPTreeFromXMLString(xml, ipt_none, (PTreeReaderOptions)(ptr_ignoreWhiteSpace | ptr_ignoreNameSpaces));
  2300. IPropertyTree *root = tree.get();
  2301. if (strieq(root->queryName(), "Envelope"))
  2302. root = root->queryPropTree("Body/*[1]");
  2303. if (!root)
  2304. return;
  2305. if (setJobname)
  2306. {
  2307. SCMStringBuffer name;
  2308. wu->getJobName(name);
  2309. if (!name.length())
  2310. wu->setJobName(root->queryName());
  2311. }
  2312. wu->setXmlParams(LINK(root));
  2313. }
  2314. void WsWuHelpers::setXmlParameters(IWorkUnit *wu, const char *xml, IArrayOf<IConstNamedValue> *variables, bool setJobname)
  2315. {
  2316. StringBuffer extParamXml;
  2317. if (variables && variables->length())
  2318. {
  2319. Owned<IPropertyTree> paramTree = (xml && *xml) ? createPTreeFromXMLString(xml) : createPTree("input");
  2320. ForEachItemIn(i, *variables)
  2321. {
  2322. IConstNamedValue &item = variables->item(i);
  2323. const char *name = item.getName();
  2324. const char *value = item.getValue();
  2325. if (!name || !*name)
  2326. continue;
  2327. if (!value)
  2328. {
  2329. size_t len = strlen(name);
  2330. char last = name[len-1];
  2331. if (last == '-' || last == '+')
  2332. {
  2333. StringAttr s(name, len-1);
  2334. paramTree->setPropInt(s.get(), last == '+' ? 1 : 0);
  2335. }
  2336. else
  2337. paramTree->setPropInt(name, 1);
  2338. continue;
  2339. }
  2340. paramTree->setProp(name, value);
  2341. }
  2342. toXML(paramTree, extParamXml);
  2343. xml=extParamXml.str();
  2344. }
  2345. setXmlParameters(wu, xml, setJobname);
  2346. }
  2347. void WsWuHelpers::submitWsWorkunit(IEspContext& context, IConstWorkUnit* cw, const char* cluster, const char* snapshot, int maxruntime, bool compile, bool resetWorkflow, bool resetVariables,
  2348. const char *paramXml, IArrayOf<IConstNamedValue> *variables, IArrayOf<IConstNamedValue> *debugs)
  2349. {
  2350. ensureWsWorkunitAccess(context, *cw, SecAccess_Write);
  2351. switch(cw->getState())
  2352. {
  2353. case WUStateRunning:
  2354. case WUStateDebugPaused:
  2355. case WUStateDebugRunning:
  2356. case WUStateCompiling:
  2357. case WUStateAborting:
  2358. case WUStateBlocked:
  2359. {
  2360. SCMStringBuffer descr;
  2361. throw MakeStringException(ECLWATCH_CANNOT_SUBMIT_WORKUNIT, "Cannot submit the workunit. Workunit state is '%s'.", cw->getStateDesc(descr).str());
  2362. }
  2363. }
  2364. SCMStringBuffer wuid;
  2365. cw->getWuid(wuid);
  2366. WorkunitUpdate wu(&cw->lock());
  2367. if(!wu.get())
  2368. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Cannot update workunit %s.", wuid.str());
  2369. wu->clearExceptions();
  2370. if(notEmpty(cluster))
  2371. wu->setClusterName(cluster);
  2372. if(notEmpty(snapshot))
  2373. wu->setSnapshot(snapshot);
  2374. wu->setState(WUStateSubmitted);
  2375. if (maxruntime)
  2376. wu->setDebugValueInt("maxRunTime",maxruntime,true);
  2377. if (debugs && debugs->length())
  2378. {
  2379. ForEachItemIn(i, *debugs)
  2380. {
  2381. IConstNamedValue &item = debugs->item(i);
  2382. const char *name = item.getName();
  2383. const char *value = item.getValue();
  2384. if (!name || !*name)
  2385. continue;
  2386. if (!value)
  2387. {
  2388. size_t len = strlen(name);
  2389. char last = name[len-1];
  2390. if (last == '-' || last == '+')
  2391. {
  2392. StringAttr s(name, len-1);
  2393. wu->setDebugValueInt(s.get(), last == '+' ? 1 : 0, true);
  2394. }
  2395. else
  2396. wu->setDebugValueInt(name, 1, true);
  2397. continue;
  2398. }
  2399. wu->setDebugValue(name, value, true);
  2400. }
  2401. }
  2402. if (resetWorkflow)
  2403. wu->resetWorkflow();
  2404. if (!compile)
  2405. wu->schedule();
  2406. if (resetVariables)
  2407. {
  2408. SCMStringBuffer varname;
  2409. Owned<IConstWUResultIterator> vars = &wu->getVariables();
  2410. ForEach (*vars)
  2411. {
  2412. vars->query().getResultName(varname);
  2413. Owned<IWUResult> v = wu->updateVariableByName(varname.str());
  2414. if (v)
  2415. v->setResultStatus(ResultStatusUndefined);
  2416. }
  2417. }
  2418. setXmlParameters(wu, paramXml, variables, (wu->getAction()==WUActionExecuteExisting));
  2419. wu->commit();
  2420. wu.clear();
  2421. if (!compile)
  2422. runWorkUnit(wuid.str());
  2423. else if (context.querySecManager())
  2424. secSubmitWorkUnit(wuid.str(), *context.querySecManager(), *context.queryUser());
  2425. else
  2426. submitWorkUnit(wuid.str(), context.queryUserId(), context.queryPassword());
  2427. AuditSystemAccess(context.queryUserId(), true, "Submitted %s", wuid.str());
  2428. }
  2429. void WsWuHelpers::submitWsWorkunit(IEspContext& context, const char *wuid, const char* cluster, const char* snapshot, int maxruntime, bool compile, bool resetWorkflow, bool resetVariables,
  2430. const char *paramXml, IArrayOf<IConstNamedValue> *variables, IArrayOf<IConstNamedValue> *debugs)
  2431. {
  2432. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  2433. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid, false);
  2434. if(!cw)
  2435. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT,"Cannot open workunit %s.",wuid);
  2436. return submitWsWorkunit(context, cw, cluster, snapshot, maxruntime, compile, resetWorkflow, resetVariables, paramXml, variables, debugs);
  2437. }
  2438. void WsWuHelpers::copyWsWorkunit(IEspContext &context, IWorkUnit &wu, const char *srcWuid)
  2439. {
  2440. Owned<IWorkUnitFactory> factory = getWorkUnitFactory(context.querySecManager(), context.queryUser());
  2441. Owned<IConstWorkUnit> src(factory->openWorkUnit(srcWuid, false));
  2442. SCMStringBuffer wuid;
  2443. wu.getWuid(wuid);
  2444. queryExtendedWU(&wu)->copyWorkUnit(src, false);
  2445. SCMStringBuffer token;
  2446. wu.setSecurityToken(createToken(wuid.str(), context.queryUserId(), context.queryPassword(), token).str());
  2447. wu.commit();
  2448. }
  2449. void WsWuHelpers::runWsWorkunit(IEspContext &context, StringBuffer &wuid, const char *srcWuid, const char *cluster, const char *paramXml,
  2450. IArrayOf<IConstNamedValue> *variables, IArrayOf<IConstNamedValue> *debugs)
  2451. {
  2452. StringBufferAdaptor isvWuid(wuid);
  2453. NewWsWorkunit wu(context);
  2454. wu->getWuid(isvWuid);
  2455. copyWsWorkunit(context, *wu, srcWuid);
  2456. wu.clear();
  2457. submitWsWorkunit(context, wuid.str(), cluster, NULL, 0, false, true, true, paramXml, variables, debugs);
  2458. }
  2459. void WsWuHelpers::runWsWorkunit(IEspContext &context, IConstWorkUnit *cw, const char *srcWuid, const char *cluster, const char *paramXml,
  2460. IArrayOf<IConstNamedValue> *variables, IArrayOf<IConstNamedValue> *debugs)
  2461. {
  2462. WorkunitUpdate wu(&cw->lock());
  2463. copyWsWorkunit(context, *wu, srcWuid);
  2464. wu.clear();
  2465. submitWsWorkunit(context, cw, cluster, NULL, 0, false, true, true, paramXml, variables, debugs);
  2466. }
  2467. IException * WsWuHelpers::noteException(IWorkUnit *wu, IException *e, WUExceptionSeverity level)
  2468. {
  2469. if (wu)
  2470. {
  2471. Owned<IWUException> we = wu->createException();
  2472. StringBuffer s;
  2473. we->setExceptionMessage(e->errorMessage(s).str());
  2474. we->setExceptionSource("WsWorkunits");
  2475. we->setSeverity(level);
  2476. if (level==ExceptionSeverityError)
  2477. wu->setState(WUStateFailed);
  2478. }
  2479. return e;
  2480. }
  2481. StringBuffer & WsWuHelpers::resolveQueryWuid(StringBuffer &wuid, const char *queryset, const char *query, bool notSuspended, IWorkUnit *wu)
  2482. {
  2483. Owned<IPropertyTree> qs = getQueryRegistry(queryset, true);
  2484. if (!qs)
  2485. throw noteException(wu, MakeStringException(ECLWATCH_QUERYSET_NOT_FOUND, "QuerySet '%s' not found", queryset));
  2486. Owned<IPropertyTree> q = resolveQueryAlias(qs, query);
  2487. if (!q)
  2488. throw noteException(wu, MakeStringException(ECLWATCH_QUERYID_NOT_FOUND, "Query '%s/%s' not found", queryset, query));
  2489. if (notSuspended && q->getPropBool("@suspended"))
  2490. throw noteException(wu, MakeStringException(ECLWATCH_QUERY_SUSPENDED, "Query '%s/%s' is suspended", queryset, query));
  2491. return wuid.append(q->queryProp("@wuid"));
  2492. }
  2493. void WsWuHelpers::runWsWuQuery(IEspContext &context, IConstWorkUnit *cw, const char *queryset, const char *query, const char *cluster, const char *paramXml)
  2494. {
  2495. StringBuffer srcWuid;
  2496. WorkunitUpdate wu(&cw->lock());
  2497. resolveQueryWuid(srcWuid, queryset, query, true, wu);
  2498. copyWsWorkunit(context, *wu, srcWuid);
  2499. wu.clear();
  2500. submitWsWorkunit(context, cw, cluster, NULL, 0, false, true, true, paramXml);
  2501. }
  2502. void WsWuHelpers::runWsWuQuery(IEspContext &context, StringBuffer &wuid, const char *queryset, const char *query, const char *cluster, const char *paramXml)
  2503. {
  2504. StringBuffer srcWuid;
  2505. StringBufferAdaptor isvWuid(wuid);
  2506. NewWsWorkunit wu(context);
  2507. wu->getWuid(isvWuid);
  2508. resolveQueryWuid(srcWuid, queryset, query, true, wu);
  2509. copyWsWorkunit(context, *wu, srcWuid);
  2510. wu.clear();
  2511. submitWsWorkunit(context, wuid.str(), cluster, NULL, 0, false, true, true, paramXml);
  2512. }
  2513. void WsWuHelpers::checkAndTrimWorkunit(const char* methodName, StringBuffer& input)
  2514. {
  2515. const char* trimmedInput = input.trim().str();
  2516. if (isEmpty(trimmedInput))
  2517. throw MakeStringException(ECLWATCH_INVALID_INPUT, "%s: Workunit ID not set", methodName);
  2518. if (!looksLikeAWuid(trimmedInput))
  2519. throw MakeStringException(ECLWATCH_INVALID_INPUT, "%s: Invalid Workunit ID: %s", methodName, trimmedInput);
  2520. return;
  2521. }
  2522. }