ws_smcService.cpp 84 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221
  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. #pragma warning (disable : 4786)
  14. #include "build-config.h"
  15. #ifdef _USE_OPENLDAP
  16. #include "ldapsecurity.ipp"
  17. #endif
  18. #include "ws_smcService.hpp"
  19. #include "wshelpers.hpp"
  20. #include "dalienv.hpp"
  21. #include "WUWrapper.hpp"
  22. #include "wujobq.hpp"
  23. #include "dfuwu.hpp"
  24. #include "exception_util.hpp"
  25. #include "roxiecontrol.hpp"
  26. #include "workunit.hpp"
  27. #define STATUS_SERVER_THOR "ThorMaster"
  28. #define STATUS_SERVER_HTHOR "HThorServer"
  29. #define STATUS_SERVER_ROXIE "RoxieServer"
  30. #define STATUS_SERVER_DFUSERVER "DFUserver"
  31. #define STATUS_SERVER_ECLSERVER "ECLserver"
  32. #define STATUS_SERVER_ECLCCSERVER "ECLCCserver"
  33. #define STATUS_SERVER_ECLAGENT "ECLagent"
  34. #define CLUSTER_TYPE_THOR "Thor"
  35. #define CLUSTER_TYPE_HTHOR "HThor"
  36. #define CLUSTER_TYPE_ROXIE "Roxie"
  37. static const char* FEATURE_URL = "SmcAccess";
  38. const char* THORQUEUE_FEATURE = "ThorQueueAccess";
  39. static const char* ROXIE_CONTROL_URL = "RoxieControlAccess";
  40. static const char* OWN_WU_ACCESS = "OwnWorkunitsAccess";
  41. static const char* OTHERS_WU_ACCESS = "OthersWorkunitsAccess";
  42. const char* PERMISSIONS_FILENAME = "espsmc_permissions.xml";
  43. const unsigned DEFAULTACTIVITYINFOCACHETIMEOUTSECOND = 10;
  44. void AccessSuccess(IEspContext& context, char const * msg,...) __attribute__((format(printf, 2, 3)));
  45. void AccessSuccess(IEspContext& context, char const * msg,...)
  46. {
  47. StringBuffer buf;
  48. buf.appendf("User %s: ",context.queryUserId());
  49. va_list args;
  50. va_start(args, msg);
  51. buf.valist_appendf(msg, args);
  52. va_end(args);
  53. AUDIT(AUDIT_TYPE_ACCESS_SUCCESS,buf.str());
  54. }
  55. void AccessFailure(IEspContext& context, char const * msg,...) __attribute__((format(printf, 2, 3)));
  56. void AccessFailure(IEspContext& context, char const * msg,...)
  57. {
  58. StringBuffer buf;
  59. buf.appendf("User %s: ",context.queryUserId());
  60. va_list args;
  61. va_start(args, msg);
  62. buf.valist_appendf(msg, args);
  63. va_end(args);
  64. AUDIT(AUDIT_TYPE_ACCESS_FAILURE,buf.str());
  65. }
  66. struct QueueLock
  67. {
  68. QueueLock(IJobQueue* q): queue(q) { queue->lock(); }
  69. ~QueueLock()
  70. {
  71. queue->unlock();
  72. }
  73. Linked<IJobQueue> queue;
  74. };
  75. static int sortTargetClustersByNameDescending(IInterface **L, IInterface **R)
  76. {
  77. IEspTargetCluster *left = (IEspTargetCluster *) *L;
  78. IEspTargetCluster *right = (IEspTargetCluster *) *R;
  79. return strcmp(right->getClusterName(), left->getClusterName());
  80. }
  81. static int sortTargetClustersByNameAscending(IInterface **L, IInterface **R)
  82. {
  83. IEspTargetCluster *left = (IEspTargetCluster *) *L;
  84. IEspTargetCluster *right = (IEspTargetCluster *) *R;
  85. return strcmp(left->getClusterName(), right->getClusterName());
  86. }
  87. static int sortTargetClustersBySizeDescending(IInterface **L, IInterface **R)
  88. {
  89. IEspTargetCluster *left = (IEspTargetCluster *) *L;
  90. IEspTargetCluster *right = (IEspTargetCluster *) *R;
  91. return right->getClusterSize() - left->getClusterSize();
  92. }
  93. static int sortTargetClustersBySizeAscending(IInterface **L, IInterface **R)
  94. {
  95. IEspTargetCluster *left = (IEspTargetCluster *) *L;
  96. IEspTargetCluster *right = (IEspTargetCluster *) *R;
  97. return left->getClusterSize() - right->getClusterSize();
  98. }
  99. void CWsSMCEx::init(IPropertyTree *cfg, const char *process, const char *service)
  100. {
  101. if (!daliClientActive())
  102. {
  103. ERRLOG("No Dali Connection Active.");
  104. throw MakeStringException(-1, "No Dali Connection Active. Please Specify a Dali to connect to in you configuration file");
  105. }
  106. m_BannerAction = 0;
  107. m_EnableChatURL = false;
  108. m_BannerSize = "4";
  109. m_BannerColor = "red";
  110. m_BannerScroll = "2";
  111. StringBuffer xpath;
  112. xpath.appendf("Software/EspProcess[@name='%s']/@portalurl", process);
  113. const char* portalURL = cfg->queryProp(xpath.str());
  114. if (portalURL && *portalURL)
  115. m_PortalURL.append(portalURL);
  116. xpath.setf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/ActivityInfoCacheSeconds", process, service);
  117. activityInfoCacheSeconds = cfg->getPropInt(xpath.str(), DEFAULTACTIVITYINFOCACHETIMEOUTSECOND);
  118. }
  119. static void countProgress(IPropertyTree *t,unsigned &done,unsigned &total)
  120. {
  121. total = 0;
  122. done = 0;
  123. Owned<IPropertyTreeIterator> it = t->getElements("DFT/progress");
  124. ForEach(*it) {
  125. IPropertyTree &e=it->query();
  126. if (e.getPropInt("@done",0))
  127. done++;
  128. total++;
  129. }
  130. }
  131. struct CActiveWorkunitWrapper: public CActiveWorkunit
  132. {
  133. CActiveWorkunitWrapper(IEspContext &context, const char* wuid,const char* location = NULL,unsigned index=0): CActiveWorkunit("","")
  134. {
  135. CWUWrapper wu(wuid, context);
  136. setActiveWorkunit(wu, wuid, location, index, context.getClientVersion(), false);
  137. }
  138. CActiveWorkunitWrapper(const char* wuid, const char* location = NULL, unsigned index=0): CActiveWorkunit("","")
  139. {
  140. CWUWrapper wu(wuid);
  141. setActiveWorkunit(wu, wuid, location, index, 0.0, true);
  142. }
  143. CActiveWorkunitWrapper(const char* wuid,const char* owner, const char* jobname, const char* state, const char* priority): CActiveWorkunit("","")
  144. {
  145. setWuid(wuid);
  146. setState(state);
  147. setOwner(owner);
  148. setJobname(jobname);
  149. setPriority(priority);
  150. }
  151. void setActiveWorkunit(CWUWrapper& wu, const char* wuid, const char* location, unsigned index, double version, bool notCheckVersion)
  152. {
  153. StringBuffer stateStr;
  154. SCMStringBuffer state, stateEx, owner, jobname;
  155. setWuid(wuid);
  156. wu->getStateDesc(state);
  157. setStateID(wu->getState());
  158. if (wu->getState() == WUStateBlocked)
  159. {
  160. wu->getStateEx(stateEx);
  161. if (notCheckVersion || (version > 1.00))
  162. setExtra(stateEx.str());
  163. }
  164. buildAndSetState(state.str(), stateEx.str(), location, index);
  165. if ((notCheckVersion || (version > 1.09)) && (wu->getState() == WUStateFailed))
  166. setWarning("The job will ultimately not complete. Please check ECLAgent.");
  167. setOwner(wu->getUser(owner).str());
  168. setJobname(wu->getJobName(jobname).str());
  169. setPriorityStr(wu->getPriority());
  170. if ((notCheckVersion || (version > 1.08)) && wu->isPausing())
  171. {
  172. setIsPausing(true);
  173. }
  174. if (notCheckVersion || (version > 1.14))
  175. {
  176. SCMStringBuffer clusterName;
  177. setClusterName(wu->getClusterName(clusterName).str());
  178. }
  179. }
  180. void buildAndSetState(const char* state, const char* stateEx, const char* location, unsigned index)
  181. {
  182. if (!state || !*state)
  183. return;
  184. StringBuffer stateStr;
  185. if(index && location && *location)
  186. stateStr.setf("queued(%d) [%s on %s]", index, state, location);
  187. else if(index)
  188. stateStr.setf("queued(%d) [%s]", index, state);
  189. else if(location && *location)
  190. stateStr.setf("%s [%s]", state, location);
  191. else
  192. stateStr.set(state);
  193. if (stateEx && *stateEx)
  194. stateStr.appendf(" %s", stateEx);
  195. setState(stateStr.str());
  196. }
  197. void setPriorityStr(unsigned priorityType)
  198. {
  199. switch(priorityType)
  200. {
  201. case PriorityClassHigh: setPriority("high"); break;
  202. default:
  203. case PriorityClassNormal: setPriority("normal"); break;
  204. case PriorityClassLow: setPriority("low"); break;
  205. }
  206. return;
  207. }
  208. };
  209. bool CWsSMCEx::onIndex(IEspContext &context, IEspSMCIndexRequest &req, IEspSMCIndexResponse &resp)
  210. {
  211. resp.setRedirectUrl("/");
  212. return true;
  213. }
  214. static int stringcmp(const char **a, const char **b)
  215. {
  216. return strcmp(*a, *b);
  217. }
  218. bool isInWuList(IArrayOf<IEspActiveWorkunit>& aws, const char* wuid)
  219. {
  220. bool bFound = false;
  221. if (wuid && *wuid && (aws.length() > 0))
  222. {
  223. ForEachItemIn(k, aws)
  224. {
  225. IEspActiveWorkunit& wu = aws.item(k);
  226. const char* wuid0 = wu.getWuid();
  227. const char* server0 = wu.getServer();
  228. if (wuid0 && !strcmp(wuid0, wuid) && (!server0 || strcmp(server0, "ECLagent")))
  229. {
  230. bFound = true;
  231. break;
  232. }
  233. }
  234. }
  235. return bFound;
  236. }
  237. void CWsSMCEx::readBannerAndChatRequest(IEspContext& context, IEspActivityRequest &req, IEspActivityResponse& resp)
  238. {
  239. StringBuffer chatURLStr, bannerStr;
  240. const char* chatURL = req.getChatURL();
  241. const char* banner = req.getBannerContent();
  242. //Filter out invalid chars
  243. if (chatURL && *chatURL)
  244. {
  245. const char* pStr = chatURL;
  246. unsigned len = strlen(chatURL);
  247. for (unsigned i = 0; i < len; i++)
  248. {
  249. if (isprint(*pStr))
  250. chatURLStr.append(*pStr);
  251. pStr++;
  252. }
  253. }
  254. if (banner && *banner)
  255. {
  256. const char* pStr = banner;
  257. unsigned len = strlen(banner);
  258. for (unsigned i = 0; i < len; i++)
  259. {
  260. bannerStr.append(isprint(*pStr) ? *pStr : '.');
  261. pStr++;
  262. }
  263. }
  264. chatURLStr.trim();
  265. bannerStr.trim();
  266. if (!req.getBannerAction_isNull() && req.getBannerAction() && (bannerStr.length() < 1))
  267. throw MakeStringException(ECLWATCH_MISSING_BANNER_CONTENT, "If a Banner is enabled, the Banner content must be specified.");
  268. if (!req.getEnableChatURL_isNull() && req.getEnableChatURL() && (chatURLStr.length() < 1))
  269. throw MakeStringException(ECLWATCH_MISSING_CHAT_URL, "If a Chat is enabled, the Chat URL must be specified.");
  270. //Now, store the strings since they are valid.
  271. m_ChatURL = chatURLStr;
  272. m_Banner = bannerStr;
  273. const char* bannerSize = req.getBannerSize();
  274. if (bannerSize && *bannerSize)
  275. m_BannerSize.set(bannerSize);
  276. const char* bannerColor = req.getBannerColor();
  277. if (bannerColor && *bannerColor)
  278. m_BannerColor.set(bannerColor);
  279. const char* bannerScroll = req.getBannerScroll();
  280. if (bannerScroll && *bannerScroll)
  281. m_BannerScroll.set(bannerScroll);
  282. m_BannerAction = req.getBannerAction();
  283. if(!req.getEnableChatURL_isNull())
  284. m_EnableChatURL = req.getEnableChatURL();
  285. }
  286. void CWsSMCEx::setBannerAndChatData(double version, IEspActivityResponse& resp)
  287. {
  288. resp.setShowBanner(m_BannerAction);
  289. resp.setShowChatURL(m_EnableChatURL);
  290. resp.setBannerContent(m_Banner.str());
  291. resp.setBannerSize(m_BannerSize.str());
  292. resp.setBannerColor(m_BannerColor.str());
  293. resp.setChatURL(m_ChatURL.str());
  294. if (version >= 1.08)
  295. resp.setBannerScroll(m_BannerScroll.str());
  296. }
  297. void CWsSMCEx::createActiveWorkUnit(Owned<IEspActiveWorkunit>& ownedWU, IEspContext &context, const char* wuid, const char* location,
  298. unsigned index, const char* serverName, const char* queueName, const char* instanceName, const char* targetClusterName, bool useContext)
  299. {
  300. try
  301. {
  302. if (useContext)
  303. ownedWU.setown(new CActiveWorkunitWrapper(context, wuid, location, index));
  304. else
  305. ownedWU.setown(new CActiveWorkunitWrapper(wuid, location, index));
  306. }
  307. catch (IException *e)
  308. { //if the wu cannot be opened for some reason, the openWorkUnit() inside the CActiveWorkunitWrapper() may throw an exception.
  309. //We do not want the exception stops this process of retrieving/showing all active WUs. And that WU should still be displayed
  310. //with the exception.
  311. StringBuffer msg;
  312. ownedWU.setown(new CActiveWorkunitWrapper(wuid, "", "", e->errorMessage(msg).str(), "normal"));
  313. ownedWU->setStateID(WUStateUnknown);
  314. e->Release();
  315. }
  316. ownedWU->setServer(serverName);
  317. ownedWU->setQueueName(queueName);
  318. if (instanceName && *instanceName)
  319. ownedWU->setInstance(instanceName); // JCSMORE In thor case at least, if queued it is unknown which instance it will run on..
  320. if (targetClusterName && *targetClusterName)
  321. ownedWU->setTargetClusterName(targetClusterName);
  322. }
  323. void CWsSMCEx::readWUsAndStateFromJobQueue(IEspContext& context, CWsSMCTargetCluster& targetCluster,
  324. CWsSMCQueue& jobQueue, const char* queueName, BoolHash& uniqueWUIDs, IArrayOf<IEspActiveWorkunit>& aws)
  325. {
  326. CJobQueueContents contents;
  327. Owned<IJobQueue> queue = createJobQueue(jobQueue.queueName.str());
  328. queue->copyItemsAndState(contents, jobQueue.queueState, jobQueue.queueStateDetails);
  329. Owned<IJobQueueIterator> iter = contents.getIterator();
  330. jobQueue.countQueuedJobs=0;
  331. ForEach(*iter)
  332. {
  333. const char* wuid = iter->query().queryWUID();
  334. if (!wuid || !*wuid || uniqueWUIDs.getValue(wuid))
  335. continue;
  336. uniqueWUIDs.setValue(wuid, true);
  337. const char* queue = NULL;
  338. if (queueName && *queueName)
  339. queue = queueName;
  340. else
  341. queue = targetCluster.clusterName.get();
  342. Owned<IEspActiveWorkunit> wu;
  343. createActiveWorkUnit(wu, context, wuid, jobQueue.queueName.str(), ++jobQueue.countQueuedJobs, targetCluster.statusServerName.str(),
  344. queue, NULL, targetCluster.clusterName.get(), false);
  345. aws.append(*wu.getLink());
  346. }
  347. }
  348. bool CWsSMCEx::findQueueInStatusServer(IEspContext& context, IPropertyTree* serverStatusRoot, const char* serverName, const char* queueName)
  349. {
  350. bool foundServer = false;
  351. VStringBuffer path("Server[@name=\"%s\"]", serverName);
  352. Owned<IPropertyTreeIterator> it(serverStatusRoot->getElements(path.str()));
  353. ForEach(*it)
  354. {
  355. IPropertyTree& serverStatusNode = it->query();
  356. const char* queue = serverStatusNode.queryProp("@queue");
  357. if (!queue || !*queue)
  358. continue;
  359. StringArray qlist;
  360. qlist.appendListUniq(queue, ",");
  361. ForEachItemIn(q, qlist)
  362. {
  363. if (strieq(qlist.item(q), queueName))
  364. {
  365. foundServer = true;
  366. break;
  367. }
  368. }
  369. if (foundServer)
  370. break;
  371. }
  372. return foundServer;
  373. }
  374. void CWsSMCEx::sortTargetClusters(IArrayOf<IEspTargetCluster>& clusters, const char* sortBy, bool descending)
  375. {
  376. if (!sortBy || !*sortBy || strieq(sortBy, "name"))
  377. clusters.sort(descending ? sortTargetClustersByNameDescending : sortTargetClustersByNameAscending);
  378. else
  379. clusters.sort(descending ? sortTargetClustersBySizeDescending : sortTargetClustersBySizeAscending);
  380. }
  381. void CWsSMCEx::getClusterQueueStatus(const CWsSMCTargetCluster& targetCluster, ClusterStatusType& queueStatusType, StringBuffer& queueStatusDetails)
  382. {
  383. const CWsSMCQueue* jobQueue = &targetCluster.clusterQueue;
  384. if (targetCluster.clusterType != ThorLCRCluster)
  385. jobQueue = &targetCluster.agentQueue;
  386. if (!jobQueue->queueName.length())
  387. return;
  388. bool queuePausedOrStopped = false;
  389. //get queueStatusDetails
  390. if (targetCluster.clusterStatusDetails.length())
  391. queueStatusDetails.set(targetCluster.clusterStatusDetails.str());
  392. if (jobQueue->queueState.length())
  393. {
  394. const char* queueState = jobQueue->queueState.str();
  395. queueStatusDetails.appendf("%s: queue %s; ", jobQueue->queueName.str(), queueState);
  396. if (jobQueue->queueStateDetails.length())
  397. queueStatusDetails.appendf(" %s;", jobQueue->queueStateDetails.str());
  398. if (strieq(queueState,"stopped") || strieq(queueState,"paused"))
  399. queuePausedOrStopped = true;
  400. }
  401. //get queueStatusType
  402. if (!jobQueue->foundQueueInStatusServer)
  403. {
  404. if (queuePausedOrStopped)
  405. queueStatusType = QueuePausedOrStoppedNotFound;
  406. else
  407. queueStatusType = QueueRunningNotFound;
  408. }
  409. else if (!queuePausedOrStopped)
  410. queueStatusType = RunningNormal;
  411. else if (jobQueue->countRunningJobs > 0)
  412. queueStatusType = QueuePausedOrStoppedWithJobs;
  413. else
  414. queueStatusType = QueuePausedOrStoppedWithNoJob;
  415. return;
  416. }
  417. void CWsSMCEx::setClusterStatus(IEspContext& context, const CWsSMCTargetCluster& targetCluster, IEspTargetCluster* returnCluster)
  418. {
  419. ClusterStatusType queueStatusType = RunningNormal;
  420. StringBuffer queueStatusDetails;
  421. getClusterQueueStatus(targetCluster, queueStatusType, queueStatusDetails);
  422. returnCluster->setClusterStatus(queueStatusType);
  423. //Set 'Warning' which may be displayed beside cluster name
  424. if (queueStatusType == QueueRunningNotFound)
  425. returnCluster->setWarning("Cluster not attached");
  426. else if (queueStatusType == QueuePausedOrStoppedNotFound)
  427. returnCluster->setWarning("Queue paused or stopped - Cluster not attached");
  428. else if (queueStatusType != RunningNormal)
  429. returnCluster->setWarning("Queue paused or stopped");
  430. //Set 'StatusDetails' which may be displayed when a mouse is moved over cluster icon
  431. if (queueStatusDetails.length())
  432. returnCluster->setStatusDetails(queueStatusDetails.str());
  433. }
  434. void CWsSMCEx::getWUsNotOnTargetCluster(IEspContext &context, IPropertyTree* serverStatusRoot, IArrayOf<IEspServerJobQueue>& serverJobQueues,
  435. IArrayOf<IEspActiveWorkunit>& aws)
  436. {
  437. BoolHash uniqueServers;
  438. Owned<IPropertyTreeIterator> it(serverStatusRoot->getElements("Server"));
  439. ForEach(*it)
  440. {
  441. IPropertyTree& serverNode = it->query();
  442. const char* serverName = serverNode.queryProp("@name");
  443. const char* instance = serverNode.queryProp("@node");
  444. const char* queueName = serverNode.queryProp("@queue");
  445. unsigned port = serverNode.getPropInt("@mpport", 0);
  446. if (!serverName || !*serverName || !instance || !*instance || strieq(serverName, STATUS_SERVER_DFUSERVER) ||//DFUServer already handled separately
  447. strieq(serverName, "ThorMaster") || strieq(serverName, "RoxieServer") || strieq(serverName, "HThorServer"))//target clusters already handled separately
  448. continue;
  449. VStringBuffer instanceName("%s_on_%s:%d", serverName, instance, port);
  450. Owned<IPropertyTreeIterator> wuids(serverNode.getElements("WorkUnit"));
  451. ForEach(*wuids)
  452. {
  453. const char* wuid=wuids->query().queryProp(NULL);
  454. if (!wuid || !*wuid)
  455. continue;
  456. if (isInWuList(aws, wuid))
  457. continue;
  458. Owned<IEspActiveWorkunit> wu;
  459. createActiveWorkUnit(wu, context, wuid, NULL, 0, serverName, queueName, instanceName.str(), NULL, false);
  460. aws.append(*wu.getLink());
  461. }
  462. if (!uniqueServers.getValue(instanceName))
  463. {
  464. uniqueServers.setValue(instanceName, true);
  465. addServerJobQueue(serverJobQueues, queueName, instanceName, serverName, instance, port);
  466. }
  467. }
  468. return;
  469. }
  470. void CWsSMCEx::readDFUWUs(IEspContext &context, const char* queueName, const char* serverName, IArrayOf<IEspActiveWorkunit>& aws)
  471. {
  472. StringAttrArray wulist;
  473. unsigned running = queuedJobs(queueName, wulist);
  474. ForEachItemIn(i, wulist)
  475. {
  476. StringBuffer jname, uname, state, error;
  477. const char *wuid = wulist.item(i).text.get();
  478. if (i<running)
  479. state.set("running");
  480. else
  481. state.set("queued");
  482. try
  483. {
  484. Owned<IConstDFUWorkUnit> dfuwu = getDFUWorkUnitFactory()->openWorkUnit(wuid, false);
  485. dfuwu->getUser(uname);
  486. dfuwu->getJobName(jname);
  487. }
  488. catch (IException *e)
  489. {
  490. e->errorMessage(error);
  491. state.appendf(" (%s)", error.str());
  492. e->Release();
  493. }
  494. Owned<IEspActiveWorkunit> wu(new CActiveWorkunitWrapper(wuid, uname.str(), jname.str(), state.str(), "normal"));
  495. wu->setServer(STATUS_SERVER_DFUSERVER);
  496. wu->setInstance(serverName);
  497. wu->setQueueName(queueName);
  498. aws.append(*wu.getLink());
  499. }
  500. }
  501. void CWsSMCEx::getDFUServersAndWUs(IEspContext &context, IPropertyTree* envRoot, IArrayOf<IEspServerJobQueue>& serverJobQueues, IArrayOf<IEspActiveWorkunit>& aws)
  502. {
  503. if (!envRoot)
  504. return;
  505. VStringBuffer path("Software/%s", eqDfu);
  506. Owned<IPropertyTreeIterator> services = envRoot->getElements(path);
  507. ForEach(*services)
  508. {
  509. IPropertyTree &serviceTree = services->query();
  510. const char *qname = serviceTree.queryProp("@queue");
  511. const char *serverName = serviceTree.queryProp("@name");
  512. if (!qname || !*qname)
  513. continue;
  514. StringArray queues;
  515. queues.appendListUniq(qname, ",");
  516. ForEachItemIn(q, queues)
  517. {
  518. const char *queueName = queues.item(q);
  519. readDFUWUs(context, queueName, serverName, aws);
  520. addServerJobQueue(serverJobQueues, queueName, serverName, STATUS_SERVER_DFUSERVER, NULL, 0);
  521. }
  522. }
  523. }
  524. void CWsSMCEx::getDFURecoveryJobs(IEspContext &context, const IPropertyTree* dfuRecoveryRoot, IArrayOf<IEspDFUJob>& jobs)
  525. {
  526. if (!dfuRecoveryRoot)
  527. return;
  528. Owned<IPropertyTreeIterator> it(dfuRecoveryRoot->getElements("job"));
  529. ForEach(*it)
  530. {
  531. IPropertyTree &e=it->query();
  532. if (!e.getPropBool("Running",false))
  533. continue;
  534. StringBuffer cmd;
  535. unsigned done, total;
  536. countProgress(&e,done,total);
  537. cmd.append(e.queryProp("@command")).append(" ").append(e.queryProp("@command_parameters"));
  538. Owned<IEspDFUJob> job = new CDFUJob("","");
  539. job->setTimeStarted(e.queryProp("@time_started"));
  540. job->setDone(done);
  541. job->setTotal(total);
  542. job->setCommand(cmd.str());
  543. jobs.append(*job.getLink());
  544. }
  545. }
  546. bool ActivityInfo::isCachedActivityInfoValid(unsigned timeOutSeconds)
  547. {
  548. CDateTime timeNow;
  549. timeNow.setNow();
  550. return timeNow.getSimple() <= timeCached.getSimple() + timeOutSeconds;;
  551. }
  552. void CWsSMCEx::clearActivityInfoCache()
  553. {
  554. CriticalBlock b(getActivityCrit);
  555. activityInfoCache.clear();
  556. }
  557. ActivityInfo* CWsSMCEx::getActivityInfo(IEspContext &context)
  558. {
  559. CriticalBlock b(getActivityCrit);
  560. if (activityInfoCache && activityInfoCache->isCachedActivityInfoValid(activityInfoCacheSeconds))
  561. return activityInfoCache.getLink();
  562. DBGLOG("CWsSMCEx::getActivityInfo - rebuild cached information");
  563. {
  564. EspTimeSection timer("createActivityInfo");
  565. activityInfoCache.setown(createActivityInfo(context));
  566. }
  567. return activityInfoCache.getLink();
  568. }
  569. ActivityInfo* CWsSMCEx::createActivityInfo(IEspContext &context)
  570. {
  571. Owned<IEnvironmentFactory> factory = getEnvironmentFactory();
  572. Owned<IConstEnvironment> env = factory->openEnvironment();
  573. if (!env)
  574. throw MakeStringException(ECLWATCH_CANNOT_GET_ENV_INFO,"Failed to get environment information.");
  575. CConstWUClusterInfoArray clusters;
  576. Owned<IPropertyTree> envRoot= &env->getPTree();
  577. getEnvironmentClusterInfo(envRoot, clusters);
  578. Owned<IRemoteConnection> connStatusServers = querySDS().connect("/Status/Servers",myProcessSession(),RTM_LOCK_READ,30000);
  579. IPropertyTree* serverStatusRoot = connStatusServers->queryRoot();
  580. if (!serverStatusRoot)
  581. throw MakeStringException(ECLWATCH_CANNOT_GET_STATUS_INFO, "Failed to get status server information.");
  582. IPropertyTree* dfuRecoveryRoot = NULL;
  583. Owned<IRemoteConnection> connDFURecovery = querySDS().connect("DFU/RECOVERY",myProcessSession(), RTM_LOCK_READ, 30000);
  584. if (connDFURecovery)
  585. dfuRecoveryRoot = connDFURecovery->queryRoot();
  586. Owned<ActivityInfo> activityInfo = new ActivityInfo();
  587. readTargetClusterInfo(context, clusters, serverStatusRoot, activityInfo);
  588. readRunningWUsAndQueuedWUs(context, envRoot, serverStatusRoot, dfuRecoveryRoot, activityInfo);
  589. activityInfo->timeCached.setNow();
  590. return activityInfo.getClear();
  591. }
  592. // This method reads job information from both /Status/Servers and IJobQueue.
  593. //
  594. // Each server component (a thor cluster, a dfuserver, or an eclagent) is one 'Server' branch under
  595. // /Status/Servers. A 'Server' branch has a @queue which indicates the queue name of the server.
  596. // A 'Server' branch also contains the information about running WUs on that 'Server'. This
  597. // method reads the information. Those WUs are displays under that server (identified by its queue name)
  598. // on Activity page.
  599. //
  600. // For the WUs list inside /Status/Servers/Server[@name=ECLagent] but not list under other 'Server', the
  601. // existing code has to find out WUID and @clusterName of the WU. Then, uses @clusterName to find out the
  602. // queue name in IConstWUClusterInfo. Those WUs list under that server (identified by its queue name) with
  603. // a note 'on ECLagent'. TBD: the logic here will be simpler if the /Status/Servers/Server is named the
  604. // instance and/or cluster.
  605. //
  606. // In order to get information about queued WUs, this method gets queue names from both IConstWUClusterInfo
  607. // and other environment functions. Each of those queue names is linked to one IJobQueues. From the
  608. // IJobQueues, this method reads queued jobs for each server component and list them under the server
  609. // component (identified by its queue name).
  610. bool CWsSMCEx::onActivity(IEspContext &context, IEspActivityRequest &req, IEspActivityResponse& resp)
  611. {
  612. context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, true);
  613. try
  614. {
  615. const char* build_ver = getBuildVersion();
  616. resp.setBuild(build_ver);
  617. double version = context.getClientVersion();
  618. bool isSuperUser = true;
  619. #ifdef _USE_OPENLDAP
  620. CLdapSecManager* secmgr = dynamic_cast<CLdapSecManager*>(context.querySecManager());
  621. if(secmgr && !secmgr->isSuperUser(context.queryUser()))
  622. isSuperUser = false;
  623. #endif
  624. if(isSuperUser && req.getFromSubmitBtn())
  625. readBannerAndChatRequest(context, req, resp);
  626. if (version >= 1.12)
  627. resp.setSuperUser(isSuperUser);
  628. if (version >= 1.06)
  629. setBannerAndChatData(version, resp);
  630. Owned<ActivityInfo> activityInfo = getActivityInfo(context);
  631. setActivityResponse(context, activityInfo, req, resp);
  632. }
  633. catch(IException* e)
  634. {
  635. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  636. }
  637. return true;
  638. }
  639. const char *CWsSMCEx::getStatusServerTypeName(WsSMCStatusServerType type)
  640. {
  641. return (type < WsSMCSSTterm) ? WsSMCStatusServerTypeNames[type] : NULL;
  642. }
  643. void CWsSMCEx::readTargetClusterInfo(IEspContext &context, CConstWUClusterInfoArray& clusters, IPropertyTree* serverStatusRoot,
  644. ActivityInfo* activityInfo)
  645. {
  646. ForEachItemIn(c, clusters)
  647. {
  648. IConstWUClusterInfo &cluster = clusters.item(c);
  649. Owned<CWsSMCTargetCluster> targetCluster = new CWsSMCTargetCluster();
  650. readTargetClusterInfo(context, cluster, serverStatusRoot, targetCluster);
  651. if (cluster.getPlatform() == ThorLCRCluster)
  652. activityInfo->thorTargetClusters.append(*targetCluster.getClear());
  653. else if (cluster.getPlatform() == RoxieCluster)
  654. activityInfo->roxieTargetClusters.append(*targetCluster.getClear());
  655. else
  656. activityInfo->hthorTargetClusters.append(*targetCluster.getClear());
  657. }
  658. }
  659. void CWsSMCEx::readTargetClusterInfo(IEspContext& context, IConstWUClusterInfo& cluster, IPropertyTree* serverStatusRoot, CWsSMCTargetCluster* targetCluster)
  660. {
  661. SCMStringBuffer clusterName;
  662. cluster.getName(clusterName);
  663. targetCluster->clusterName.set(clusterName.str());
  664. targetCluster->clusterType = cluster.getPlatform();
  665. targetCluster->clusterSize = cluster.getSize();
  666. cluster.getServerQueue(targetCluster->serverQueue.queueName);
  667. cluster.getAgentQueue(targetCluster->agentQueue.queueName);
  668. StringBuffer statusServerName;
  669. CWsSMCQueue* jobQueue = NULL;
  670. if (targetCluster->clusterType == ThorLCRCluster)
  671. {
  672. statusServerName.set(getStatusServerTypeName(WsSMCSSTThorLCRCluster));
  673. jobQueue = &targetCluster->clusterQueue;
  674. cluster.getThorQueue(jobQueue->queueName);
  675. }
  676. else if (targetCluster->clusterType == RoxieCluster)
  677. {
  678. statusServerName.set(getStatusServerTypeName(WsSMCSSTRoxieCluster));
  679. jobQueue = &targetCluster->agentQueue;
  680. }
  681. else
  682. {
  683. statusServerName.set(getStatusServerTypeName(WsSMCSSTHThorCluster));
  684. jobQueue = &targetCluster->agentQueue;
  685. }
  686. targetCluster->statusServerName.set(statusServerName.str());
  687. targetCluster->queueName.set(jobQueue->queueName.str());
  688. if (serverStatusRoot)
  689. {
  690. jobQueue->foundQueueInStatusServer = findQueueInStatusServer(context, serverStatusRoot, statusServerName.str(), targetCluster->queueName.get());
  691. if (!jobQueue->foundQueueInStatusServer)
  692. targetCluster->clusterStatusDetails.appendf("Cluster %s not attached; ", clusterName.str());
  693. }
  694. return;
  695. }
  696. void CWsSMCEx::readRunningWUsAndQueuedWUs(IEspContext &context, IPropertyTree* envRoot, IPropertyTree* serverStatusRoot,
  697. IPropertyTree* dfuRecoveryRoot, ActivityInfo* activityInfo)
  698. {
  699. BoolHash uniqueWUIDs;
  700. readRunningWUsOnStatusServer(context, serverStatusRoot, WsSMCSSTThorLCRCluster, activityInfo->thorTargetClusters, activityInfo->roxieTargetClusters, activityInfo->hthorTargetClusters, uniqueWUIDs, activityInfo->aws);
  701. readWUsAndStateFromJobQueue(context, activityInfo->thorTargetClusters, uniqueWUIDs, activityInfo->aws);
  702. readRunningWUsOnStatusServer(context, serverStatusRoot, WsSMCSSTRoxieCluster, activityInfo->roxieTargetClusters, activityInfo->thorTargetClusters, activityInfo->hthorTargetClusters, uniqueWUIDs, activityInfo->aws);
  703. readWUsAndStateFromJobQueue(context, activityInfo->roxieTargetClusters, uniqueWUIDs, activityInfo->aws);
  704. readRunningWUsOnStatusServer(context, serverStatusRoot, WsSMCSSTHThorCluster, activityInfo->hthorTargetClusters, activityInfo->thorTargetClusters, activityInfo->roxieTargetClusters, uniqueWUIDs, activityInfo->aws);
  705. readWUsAndStateFromJobQueue(context, activityInfo->hthorTargetClusters, uniqueWUIDs, activityInfo->aws);
  706. readRunningWUsOnStatusServer(context, serverStatusRoot, WsSMCSSTECLagent, activityInfo->thorTargetClusters, activityInfo->roxieTargetClusters, activityInfo->hthorTargetClusters, uniqueWUIDs, activityInfo->aws);
  707. getWUsNotOnTargetCluster(context, serverStatusRoot, activityInfo->serverJobQueues, activityInfo->aws);
  708. getDFUServersAndWUs(context, envRoot, activityInfo->serverJobQueues, activityInfo->aws);
  709. getDFURecoveryJobs(context, dfuRecoveryRoot, activityInfo->DFURecoveryJobs);
  710. }
  711. void CWsSMCEx::readRunningWUsOnStatusServer(IEspContext& context, IPropertyTree* serverStatusRoot, WsSMCStatusServerType statusServerType,
  712. CIArrayOf<CWsSMCTargetCluster>& targetClusters, CIArrayOf<CWsSMCTargetCluster>& targetClusters1, CIArrayOf<CWsSMCTargetCluster>& targetClusters2,
  713. BoolHash& uniqueWUIDs, IArrayOf<IEspActiveWorkunit>& aws)
  714. {
  715. const char* serverName = getStatusServerTypeName(statusServerType);
  716. if (!serverName || !*serverName)
  717. return;
  718. bool isECLAgent = (statusServerType == WsSMCSSTECLagent);
  719. VStringBuffer path("Server[@name=\"%s\"]", serverName);
  720. Owned<IPropertyTreeIterator> itrStatusServer(serverStatusRoot->getElements(path.str()));
  721. ForEach(*itrStatusServer)
  722. {
  723. IPropertyTree& serverStatusNode = itrStatusServer->query();
  724. StringBuffer instance;
  725. if ((statusServerType == WsSMCSSTThorLCRCluster) || (statusServerType == WsSMCSSTRoxieCluster))
  726. serverStatusNode.getProp("@cluster", instance);
  727. else
  728. instance.appendf("%s on %s", serverName, serverStatusNode.queryProp("@node"));
  729. const char* graph = NULL;
  730. int sgDuration = -1;
  731. int subgraph = -1;
  732. StringBuffer durationStr, subgraphStr;
  733. if (!isECLAgent)
  734. {
  735. sgDuration = serverStatusNode.getPropInt("@sg_duration", -1);
  736. subgraph = serverStatusNode.getPropInt("@subgraph", -1);
  737. graph = serverStatusNode.queryProp("@graph");
  738. durationStr.appendf("%d min", sgDuration);
  739. subgraphStr.appendf("%d", subgraph);
  740. }
  741. Owned<IPropertyTreeIterator> wuids(serverStatusNode.getElements("WorkUnit"));
  742. ForEach(*wuids)
  743. {
  744. const char* wuid=wuids->query().queryProp(NULL);
  745. if (!wuid || !*wuid || (isECLAgent && uniqueWUIDs.getValue(wuid)))
  746. continue;
  747. CWsSMCTargetCluster* targetCluster = findWUClusterInfo(context, wuid, isECLAgent, targetClusters, targetClusters1, targetClusters2);
  748. if (!targetCluster)
  749. continue;
  750. const char* targetClusterName = targetCluster->clusterName.get();
  751. CWsSMCQueue* jobQueue;
  752. if (statusServerType == WsSMCSSTThorLCRCluster)
  753. jobQueue = &targetCluster->clusterQueue;
  754. else
  755. jobQueue = &targetCluster->agentQueue;
  756. Owned<IEspActiveWorkunit> wu;
  757. if (!isECLAgent)
  758. {
  759. uniqueWUIDs.setValue(wuid, true);
  760. const char *cluster = serverStatusNode.queryProp("Cluster");
  761. StringBuffer queueName;
  762. if (cluster) // backward compat check.
  763. getClusterThorQueueName(queueName, cluster);
  764. else
  765. queueName.append(targetCluster->queueName.get());
  766. createActiveWorkUnit(wu, context, wuid, !strieq(targetClusterName, instance.str()) ? instance.str() : NULL, 0, serverName,
  767. queueName, instance.str(), targetClusterName, false);
  768. if (wu->getStateID() == WUStateRunning) //'aborting' may be another possible status
  769. {
  770. if (subgraph > -1 && sgDuration > -1)
  771. {
  772. wu->setGraphName(graph);
  773. wu->setDuration(durationStr.str());
  774. wu->setGID(subgraphStr.str());
  775. }
  776. if (serverStatusNode.getPropInt("@memoryBlocked ", 0) != 0)
  777. wu->setMemoryBlocked(1);
  778. }
  779. }
  780. else
  781. {
  782. createActiveWorkUnit(wu, context, wuid, instance.str(), 0, serverName, serverName, instance.str(), targetClusterName, false);
  783. if (targetCluster->clusterType == ThorLCRCluster)
  784. wu->setClusterType(CLUSTER_TYPE_THOR);
  785. else if (targetCluster->clusterType == RoxieCluster)
  786. wu->setClusterType(CLUSTER_TYPE_ROXIE);
  787. else
  788. wu->setClusterType(CLUSTER_TYPE_HTHOR);
  789. wu->setClusterQueueName(targetCluster->queueName.get());
  790. if (wu->getStateID() != WUStateRunning)
  791. {
  792. const char *extra = wu->getExtra();
  793. if (wu->getStateID() != WUStateBlocked || !extra || !*extra) // Blocked on persist treated as running here
  794. {
  795. aws.append(*wu.getLink());
  796. jobQueue->countQueuedJobs++;
  797. continue;
  798. }
  799. }
  800. if (serverStatusNode.getPropInt("@memoryBlocked ", 0) != 0)
  801. wu->setMemoryBlocked(1);
  802. }
  803. aws.append(*wu.getLink());
  804. jobQueue->countRunningJobs++;
  805. }
  806. }
  807. }
  808. void CWsSMCEx::readWUsAndStateFromJobQueue(IEspContext& context, CIArrayOf<CWsSMCTargetCluster>& targetClusters, BoolHash& uniqueWUIDs, IArrayOf<IEspActiveWorkunit>& aws)
  809. {
  810. ForEachItemIn(i, targetClusters)
  811. readWUsAndStateFromJobQueue(context, targetClusters.item(i), uniqueWUIDs, aws);
  812. }
  813. void CWsSMCEx::readWUsAndStateFromJobQueue(IEspContext& context, CWsSMCTargetCluster& targetCluster, BoolHash& uniqueWUIDs, IArrayOf<IEspActiveWorkunit>& aws)
  814. {
  815. if (targetCluster.clusterType == ThorLCRCluster)
  816. {
  817. readWUsAndStateFromJobQueue(context, targetCluster, targetCluster.clusterQueue, NULL, uniqueWUIDs, aws);
  818. targetCluster.queueStatus.set(targetCluster.clusterQueue.queueState);
  819. }
  820. if (targetCluster.agentQueue.queueName.length())
  821. {
  822. readWUsAndStateFromJobQueue(context, targetCluster, targetCluster.agentQueue, targetCluster.agentQueue.queueName.str(), uniqueWUIDs, aws);
  823. if (targetCluster.clusterType != ThorLCRCluster)
  824. targetCluster.queueStatus.set(targetCluster.agentQueue.queueState);
  825. }
  826. if (targetCluster.serverQueue.queueName.length())
  827. readWUsAndStateFromJobQueue(context, targetCluster, targetCluster.serverQueue, targetCluster.serverQueue.queueName.str(), uniqueWUIDs, aws);
  828. }
  829. CWsSMCTargetCluster* CWsSMCEx::findTargetCluster(const char* clusterName, CIArrayOf<CWsSMCTargetCluster>& targetClusters)
  830. {
  831. ForEachItemIn(i, targetClusters)
  832. {
  833. CWsSMCTargetCluster& targetCluster = targetClusters.item(i);
  834. if (strieq(targetCluster.clusterName.get(), clusterName))
  835. return &targetCluster;
  836. }
  837. return NULL;
  838. }
  839. CWsSMCTargetCluster* CWsSMCEx::findWUClusterInfo(IEspContext& context, const char* wuid, bool isOnECLAgent, CIArrayOf<CWsSMCTargetCluster>& targetClusters,
  840. CIArrayOf<CWsSMCTargetCluster>& targetClusters1, CIArrayOf<CWsSMCTargetCluster>& targetClusters2)
  841. {
  842. SCMStringBuffer clusterName;
  843. try
  844. {
  845. Owned<IWorkUnitFactory> factory = getWorkUnitFactory();
  846. Owned<IConstWorkUnit> cw = factory->openWorkUnit(wuid, false);
  847. if (!cw)
  848. return NULL;
  849. cw->getClusterName(clusterName);
  850. if (!clusterName.length())
  851. return NULL;
  852. }
  853. catch (IException *e)
  854. {//Exception may be thrown when the openWorkUnit() is called inside the CWUWrapper
  855. StringBuffer msg;
  856. WARNLOG("Failed to open workunit %s: %s", wuid, e->errorMessage(msg).str());
  857. e->Release();
  858. return NULL;
  859. }
  860. const char* cluster = clusterName.str();
  861. CWsSMCTargetCluster* targetCluster = findTargetCluster(cluster, targetClusters);
  862. if (targetCluster || !isOnECLAgent)
  863. return targetCluster;
  864. targetCluster = findTargetCluster(cluster, targetClusters1);
  865. if (targetCluster)
  866. return targetCluster;
  867. return findTargetCluster(cluster, targetClusters2);
  868. }
  869. void CWsSMCEx::addWUsToResponse(IEspContext &context, const IArrayOf<IEspActiveWorkunit>& aws, IEspActivityResponse& resp)
  870. {
  871. const char* user = context.queryUserId();
  872. IArrayOf<IEspActiveWorkunit> awsReturned;
  873. ForEachItemIn(i, aws)
  874. {
  875. IEspActiveWorkunit& wu = aws.item(i);
  876. const char* wuid = wu.getWuid();
  877. if (wuid[0] == 'D')//DFU WU
  878. {
  879. awsReturned.append(*LINK(&wu));
  880. continue;
  881. }
  882. try
  883. {
  884. //if no access, throw an exception and go to the 'catch' section.
  885. const char* owner = wu.getOwner();
  886. context.validateFeatureAccess((!owner || !*owner || (user && streq(user, owner))) ? OWN_WU_ACCESS : OTHERS_WU_ACCESS, SecAccess_Read, true);
  887. awsReturned.append(*LINK(&wu));
  888. continue;
  889. }
  890. catch (IException *e)
  891. { //if the wu cannot be opened for some reason, the openWorkUnit() inside the CActiveWorkunitWrapper() may throw an exception.
  892. //We do not want the exception stops this process of retrieving/showing all active WUs. And that WU should still be displayed
  893. //with the exception.
  894. StringBuffer msg;
  895. Owned<IEspActiveWorkunit> cw = new CActiveWorkunitWrapper(wuid, "", "", e->errorMessage(msg).str(), "normal");
  896. cw->setStateID(WUStateUnknown);
  897. cw->setServer(wu.getServer());
  898. cw->setQueueName(wu.getQueueName());
  899. const char* instanceName = wu.getInstance();
  900. const char* targetClusterName = wu.getTargetClusterName();
  901. if (instanceName && *instanceName)
  902. cw->setInstance(instanceName); // JCSMORE In thor case at least, if queued it is unknown which instance it will run on..
  903. if (targetClusterName && *targetClusterName)
  904. cw->setTargetClusterName(targetClusterName);
  905. awsReturned.append(*cw.getLink());
  906. e->Release();
  907. }
  908. }
  909. resp.setRunning(awsReturned);
  910. return;
  911. }
  912. void CWsSMCEx::setActivityResponse(IEspContext &context, ActivityInfo* activityInfo, IEspActivityRequest &req, IEspActivityResponse& resp)
  913. {
  914. double version = context.getClientVersion();
  915. const char* sortBy = req.getSortBy();
  916. bool descending = req.getDescending();
  917. if (version >= 1.16)
  918. {
  919. IArrayOf<IEspTargetCluster> thorClusters;
  920. IArrayOf<IEspTargetCluster> hthorClusters;
  921. IArrayOf<IEspTargetCluster> roxieClusters;
  922. setESPTargetClusters(context, activityInfo->thorTargetClusters, thorClusters);
  923. setESPTargetClusters(context, activityInfo->roxieTargetClusters, roxieClusters);
  924. setESPTargetClusters(context, activityInfo->hthorTargetClusters, hthorClusters);
  925. sortTargetClusters(thorClusters, sortBy, descending);
  926. sortTargetClusters(roxieClusters, sortBy, descending);
  927. SecAccessFlags access;
  928. if (context.authorizeFeature(THORQUEUE_FEATURE, access) && access>=SecAccess_Full)
  929. resp.setAccessRight("Access_Full");
  930. resp.setSortBy(sortBy);
  931. resp.setDescending(descending);
  932. resp.setThorClusterList(thorClusters);
  933. resp.setRoxieClusterList(roxieClusters);
  934. resp.setHThorClusterList(hthorClusters);
  935. resp.setServerJobQueues(activityInfo->serverJobQueues);
  936. }
  937. else
  938. {//for backward compatible
  939. IArrayOf<IEspThorCluster> thorClusters;
  940. ForEachItemIn(i, activityInfo->thorTargetClusters)
  941. {
  942. CWsSMCTargetCluster& targetCluster = activityInfo->thorTargetClusters.item(i);
  943. Owned<IEspThorCluster> respThorCluster = new CThorCluster("", "");
  944. respThorCluster->setClusterName(targetCluster.clusterName.get());
  945. respThorCluster->setQueueStatus(targetCluster.queueStatus.get());
  946. if (version >= 1.03)
  947. respThorCluster->setQueueName(targetCluster.queueName.get());
  948. if (version >= 1.11)
  949. respThorCluster->setClusterSize(targetCluster.clusterSize);
  950. thorClusters.append(*respThorCluster.getClear());
  951. }
  952. resp.setThorClusters(thorClusters);
  953. if (version > 1.06)
  954. {
  955. IArrayOf<IEspRoxieCluster> roxieClusters;
  956. ForEachItemIn(i, activityInfo->roxieTargetClusters)
  957. {
  958. CWsSMCTargetCluster& targetCluster = activityInfo->roxieTargetClusters.item(i);
  959. Owned<IEspRoxieCluster> respRoxieCluster = new CRoxieCluster("", "");
  960. respRoxieCluster->setClusterName(targetCluster.clusterName.get());
  961. respRoxieCluster->setQueueStatus(targetCluster.queueStatus.get());
  962. respRoxieCluster->setQueueName(targetCluster.queueName.get());
  963. if (version >= 1.11)
  964. respRoxieCluster->setClusterSize(targetCluster.clusterSize);
  965. roxieClusters.append(*respRoxieCluster.getClear());
  966. }
  967. resp.setRoxieClusters(roxieClusters);
  968. }
  969. if (version > 1.10)
  970. {
  971. resp.setSortBy(sortBy);
  972. resp.setDescending(req.getDescending());
  973. }
  974. if (version > 1.11)
  975. {
  976. IArrayOf<IEspHThorCluster> hThorClusters;
  977. ForEachItemIn(i, activityInfo->hthorTargetClusters)
  978. {
  979. CWsSMCTargetCluster& targetCluster = activityInfo->hthorTargetClusters.item(i);
  980. Owned<IEspHThorCluster> respHThorCluster = new CHThorCluster("", "");
  981. respHThorCluster->setClusterName(targetCluster.clusterName.get());
  982. respHThorCluster->setQueueStatus(targetCluster.queueStatus.get());
  983. respHThorCluster->setQueueName(targetCluster.queueName.get());
  984. respHThorCluster->setClusterSize(targetCluster.clusterSize);
  985. hThorClusters.append(*respHThorCluster.getClear());
  986. }
  987. resp.setHThorClusters(hThorClusters);
  988. SecAccessFlags access;
  989. if (context.authorizeFeature(THORQUEUE_FEATURE, access) && access>=SecAccess_Full)
  990. resp.setAccessRight("Access_Full");
  991. }
  992. if (version > 1.03)
  993. resp.setServerJobQueues(activityInfo->serverJobQueues);
  994. }
  995. resp.setDFUJobs(activityInfo->DFURecoveryJobs);
  996. addWUsToResponse(context, activityInfo->aws, resp);
  997. return;
  998. }
  999. void CWsSMCEx::setESPTargetClusters(IEspContext& context, const CIArrayOf<CWsSMCTargetCluster>& targetClusters, IArrayOf<IEspTargetCluster>& respTargetClusters)
  1000. {
  1001. ForEachItemIn(i, targetClusters)
  1002. {
  1003. Owned<IEspTargetCluster> respTargetCluster = new CTargetCluster("", "");
  1004. setESPTargetCluster(context, targetClusters.item(i), respTargetCluster);
  1005. respTargetClusters.append(*respTargetCluster.getClear());
  1006. }
  1007. }
  1008. void CWsSMCEx::addServerJobQueue(IArrayOf<IEspServerJobQueue>& jobQueues, const char* queueName, const char* serverName,
  1009. const char* serverType, const char* networkAddress, unsigned port)
  1010. {
  1011. if (!queueName || !*queueName || !serverName || !*serverName || !serverType || !*serverType)
  1012. return;
  1013. StringBuffer queueState;
  1014. StringBuffer queueStateDetails;
  1015. Owned<IJobQueue> queue = createJobQueue(queueName);
  1016. if (queue->stopped(queueStateDetails))
  1017. queueState.set("stopped");
  1018. else if (queue->paused(queueStateDetails))
  1019. queueState.set("paused");
  1020. else
  1021. queueState.set("running");
  1022. addServerJobQueue(jobQueues, queueName, serverName, serverType, networkAddress, port, queueState.str(), queueStateDetails.str());
  1023. }
  1024. void CWsSMCEx::addServerJobQueue(IArrayOf<IEspServerJobQueue>& jobQueues, const char* queueName, const char* serverName,
  1025. const char* serverType, const char* networkAddress, unsigned port, const char* queueState, const char* queueStateDetails)
  1026. {
  1027. if (!queueName || !*queueName || !serverName || !*serverName || !serverType || !*serverType)
  1028. return;
  1029. if (!queueState || !*queueState)
  1030. queueState = "running";
  1031. Owned<IEspServerJobQueue> jobQueue = createServerJobQueue("", "");
  1032. jobQueue->setQueueName(queueName);
  1033. jobQueue->setServerName(serverName);
  1034. jobQueue->setServerType(serverType);
  1035. if (networkAddress && *networkAddress)
  1036. {
  1037. jobQueue->setNetworkAddress(networkAddress);
  1038. jobQueue->setPort(port);
  1039. }
  1040. setServerJobQueueStatus(jobQueue, queueState, queueStateDetails);
  1041. jobQueues.append(*jobQueue.getClear());
  1042. }
  1043. void CWsSMCEx::setServerJobQueueStatus(double version, IEspServerJobQueue* jobQueue, const char* status, const char* details)
  1044. {
  1045. if (!status || !*status)
  1046. return;
  1047. jobQueue->setQueueStatus(status);
  1048. if (version >= 1.17)
  1049. setServerJobQueueStatusDetails(jobQueue, status, details);
  1050. }
  1051. void CWsSMCEx::setServerJobQueueStatus(IEspServerJobQueue* jobQueue, const char* status, const char* details)
  1052. {
  1053. if (!status || !*status)
  1054. return;
  1055. jobQueue->setQueueStatus(status);
  1056. setServerJobQueueStatusDetails(jobQueue, status, details);
  1057. }
  1058. void CWsSMCEx::setServerJobQueueStatusDetails(IEspServerJobQueue* jobQueue, const char* status, const char* details)
  1059. {
  1060. StringBuffer queueState;
  1061. if (details && *details)
  1062. queueState.appendf("queue %s; %s;", status, details);
  1063. else
  1064. queueState.appendf("queue %s;", status);
  1065. jobQueue->setStatusDetails(queueState.str());
  1066. }
  1067. void CWsSMCEx::addCapabilities(IPropertyTree* pFeatureNode, const char* access,
  1068. IArrayOf<IEspCapability>& capabilities)
  1069. {
  1070. StringBuffer xpath(access);
  1071. xpath.append("/Capability");
  1072. Owned<IPropertyTreeIterator> it = pFeatureNode->getElements(xpath.str());
  1073. ForEach(*it)
  1074. {
  1075. IPropertyTree* pCapabilityNode = &it->query();
  1076. IEspCapability* pCapability = new CCapability("ws_smc");
  1077. pCapability->setName( pCapabilityNode->queryProp("@name") );
  1078. pCapability->setDescription( pCapabilityNode->queryProp("@description") );
  1079. capabilities.append(*pCapability);
  1080. }
  1081. }
  1082. static void checkAccess(IEspContext &context, const char* feature,int level)
  1083. {
  1084. if (!context.validateFeatureAccess(feature, level, false))
  1085. throw MakeStringException(ECLWATCH_THOR_QUEUE_ACCESS_DENIED, "Failed to access the queue functions. Permission denied.");
  1086. }
  1087. bool CWsSMCEx::onMoveJobDown(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
  1088. {
  1089. try
  1090. {
  1091. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1092. {
  1093. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1094. QueueLock lock(queue);
  1095. unsigned index=queue->findRank(req.getWuid());
  1096. if(index<queue->ordinality())
  1097. {
  1098. Owned<IJobQueueItem> item0 = queue->getItem(index);
  1099. Owned<IJobQueueItem> item = queue->getItem(index+1);
  1100. if(item && item0 && (item0->getPriority() == item->getPriority()))
  1101. queue->moveAfter(req.getWuid(),item->queryWUID());
  1102. }
  1103. }
  1104. AccessSuccess(context, "Changed job priority %s",req.getWuid());
  1105. clearActivityInfoCache();
  1106. resp.setRedirectUrl("/WsSMC/");
  1107. }
  1108. catch(IException* e)
  1109. {
  1110. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1111. }
  1112. return true;
  1113. }
  1114. bool CWsSMCEx::onMoveJobUp(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
  1115. {
  1116. try
  1117. {
  1118. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1119. {
  1120. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1121. QueueLock lock(queue);
  1122. unsigned index=queue->findRank(req.getWuid());
  1123. if(index>0 && index<queue->ordinality())
  1124. {
  1125. Owned<IJobQueueItem> item0 = queue->getItem(index);
  1126. Owned<IJobQueueItem> item = queue->getItem(index-1);
  1127. if(item && item0 && (item0->getPriority() == item->getPriority()))
  1128. queue->moveBefore(req.getWuid(),item->queryWUID());
  1129. }
  1130. }
  1131. AccessSuccess(context, "Changed job priority %s",req.getWuid());
  1132. clearActivityInfoCache();
  1133. resp.setRedirectUrl("/WsSMC/");
  1134. }
  1135. catch(IException* e)
  1136. {
  1137. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1138. }
  1139. return true;
  1140. }
  1141. bool CWsSMCEx::onMoveJobBack(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
  1142. {
  1143. try
  1144. {
  1145. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1146. {
  1147. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1148. QueueLock lock(queue);
  1149. unsigned index=queue->findRank(req.getWuid());
  1150. if(index<queue->ordinality())
  1151. {
  1152. Owned<IJobQueueItem> item = queue->getItem(index);
  1153. int priority0 = item->getPriority();
  1154. unsigned biggestIndoxInSamePriority = index;
  1155. unsigned nextIndex = biggestIndoxInSamePriority + 1;
  1156. while (nextIndex<queue->ordinality())
  1157. {
  1158. item.setown(queue->getItem(nextIndex));
  1159. if (priority0 != item->getPriority())
  1160. {
  1161. break;
  1162. }
  1163. biggestIndoxInSamePriority = nextIndex;
  1164. nextIndex++;
  1165. }
  1166. if (biggestIndoxInSamePriority != index)
  1167. {
  1168. item.setown(queue->getItem(biggestIndoxInSamePriority));
  1169. queue->moveAfter(req.getWuid(), item->queryWUID());
  1170. }
  1171. }
  1172. }
  1173. AccessSuccess(context, "Changed job priority %s",req.getWuid());
  1174. clearActivityInfoCache();
  1175. resp.setRedirectUrl("/WsSMC/");
  1176. }
  1177. catch(IException* e)
  1178. {
  1179. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1180. }
  1181. return true;
  1182. }
  1183. bool CWsSMCEx::onMoveJobFront(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
  1184. {
  1185. try
  1186. {
  1187. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1188. {
  1189. Owned<IJobQueue> queue=createJobQueue(req.getQueueName());
  1190. QueueLock lock(queue);
  1191. unsigned index=queue->findRank(req.getWuid());
  1192. if (index>0 && index<queue->ordinality())
  1193. {
  1194. Owned<IJobQueueItem> item = queue->getItem(index);
  1195. int priority0 = item->getPriority();
  1196. unsigned smallestIndoxInSamePriority = index;
  1197. int nextIndex = smallestIndoxInSamePriority - 1;
  1198. while (nextIndex >= 0)
  1199. {
  1200. item.setown(queue->getItem(nextIndex));
  1201. if (priority0 != item->getPriority())
  1202. {
  1203. break;
  1204. }
  1205. smallestIndoxInSamePriority = nextIndex;
  1206. nextIndex--;
  1207. }
  1208. if (smallestIndoxInSamePriority != index)
  1209. {
  1210. item.setown(queue->getItem(smallestIndoxInSamePriority));
  1211. queue->moveBefore(req.getWuid(), item->queryWUID());
  1212. }
  1213. }
  1214. }
  1215. AccessSuccess(context, "Changed job priority %s",req.getWuid());
  1216. clearActivityInfoCache();
  1217. resp.setRedirectUrl("/WsSMC/");
  1218. }
  1219. catch(IException* e)
  1220. {
  1221. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1222. }
  1223. return true;
  1224. }
  1225. bool CWsSMCEx::onRemoveJob(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
  1226. {
  1227. try
  1228. {
  1229. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1230. secAbortWorkUnit(req.getWuid(), *context.querySecManager(), *context.queryUser());
  1231. {
  1232. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1233. QueueLock lock(queue);
  1234. unsigned index=queue->findRank(req.getWuid());
  1235. if(index<queue->ordinality())
  1236. {
  1237. if(!queue->cancelInitiateConversation(req.getWuid()))
  1238. throw MakeStringException(ECLWATCH_CANNOT_DELETE_WORKUNIT,"Failed to remove the workunit %s",req.getWuid());
  1239. }
  1240. }
  1241. AccessSuccess(context, "Removed job %s",req.getWuid());
  1242. clearActivityInfoCache();
  1243. resp.setRedirectUrl("/WsSMC/");
  1244. }
  1245. catch(IException* e)
  1246. {
  1247. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1248. }
  1249. return true;
  1250. }
  1251. bool CWsSMCEx::onStopQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
  1252. {
  1253. try
  1254. {
  1255. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1256. {
  1257. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1258. StringBuffer info;
  1259. queue->stop(createQueueActionInfo(context, "stopped", req, info));
  1260. }
  1261. AccessSuccess(context, "Stopped queue %s", req.getCluster());
  1262. clearActivityInfoCache();
  1263. double version = context.getClientVersion();
  1264. if (version >= 1.19)
  1265. getStatusServerInfo(context, req.getServerType(), req.getCluster(), req.getNetworkAddress(), req.getPort(), resp.updateStatusServerInfo());
  1266. resp.setRedirectUrl("/WsSMC/");
  1267. }
  1268. catch(IException* e)
  1269. {
  1270. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1271. }
  1272. return true;
  1273. }
  1274. bool CWsSMCEx::onResumeQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
  1275. {
  1276. try
  1277. {
  1278. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1279. {
  1280. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1281. StringBuffer info;
  1282. queue->resume(createQueueActionInfo(context, "resumed", req, info));
  1283. }
  1284. AccessSuccess(context, "Resumed queue %s", req.getCluster());
  1285. clearActivityInfoCache();
  1286. double version = context.getClientVersion();
  1287. if (version >= 1.19)
  1288. getStatusServerInfo(context, req.getServerType(), req.getCluster(), req.getNetworkAddress(), req.getPort(), resp.updateStatusServerInfo());
  1289. resp.setRedirectUrl("/WsSMC/");
  1290. }
  1291. catch(IException* e)
  1292. {
  1293. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1294. }
  1295. return true;
  1296. }
  1297. const char* CWsSMCEx::createQueueActionInfo(IEspContext &context, const char* state, IEspSMCQueueRequest &req, StringBuffer& info)
  1298. {
  1299. StringBuffer peer, currentTime;
  1300. context.getPeer(peer);
  1301. const char* userId = context.queryUserId();
  1302. if (!userId || !*userId)
  1303. userId = "Unknown user";
  1304. CDateTime now;
  1305. now.setNow();
  1306. now.getString(currentTime);
  1307. info.appendf("%s by <%s> at <%s> from <%s>", state, userId, currentTime.str(), peer.str());
  1308. const char* comment = req.getComment();
  1309. if (comment && *comment)
  1310. info.append(": ' ").append(comment).append("'");
  1311. return info.str();
  1312. }
  1313. bool CWsSMCEx::onPauseQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
  1314. {
  1315. try
  1316. {
  1317. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1318. {
  1319. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1320. StringBuffer info;
  1321. queue->pause(createQueueActionInfo(context, "paused", req, info));
  1322. }
  1323. AccessSuccess(context, "Paused queue %s", req.getCluster());
  1324. clearActivityInfoCache();
  1325. double version = context.getClientVersion();
  1326. if (version >= 1.19)
  1327. getStatusServerInfo(context, req.getServerType(), req.getCluster(), req.getNetworkAddress(), req.getPort(), resp.updateStatusServerInfo());
  1328. resp.setRedirectUrl("/WsSMC/");
  1329. }
  1330. catch(IException* e)
  1331. {
  1332. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1333. }
  1334. return true;
  1335. }
  1336. bool CWsSMCEx::onClearQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
  1337. {
  1338. try
  1339. {
  1340. checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
  1341. {
  1342. Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
  1343. QueueLock lock(queue);
  1344. for(unsigned i=0;i<queue->ordinality();i++)
  1345. {
  1346. Owned<IJobQueueItem> item = queue->getItem(i);
  1347. secAbortWorkUnit(item->queryWUID(), *context.querySecManager(), *context.queryUser());
  1348. }
  1349. queue->clear();
  1350. }
  1351. AccessSuccess(context, "Cleared queue %s",req.getCluster());
  1352. clearActivityInfoCache();
  1353. double version = context.getClientVersion();
  1354. if (version >= 1.19)
  1355. getStatusServerInfo(context, req.getServerType(), req.getCluster(), req.getNetworkAddress(), req.getPort(), resp.updateStatusServerInfo());
  1356. resp.setRedirectUrl("/WsSMC/");
  1357. }
  1358. catch(IException* e)
  1359. {
  1360. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1361. }
  1362. return true;
  1363. }
  1364. void CWsSMCEx::setJobPriority(IWorkUnitFactory* factory, const char* wuid, const char* queueName, WUPriorityClass& priority)
  1365. {
  1366. if (!wuid || !*wuid)
  1367. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Workunit ID not specified.");
  1368. if (!queueName || !*queueName)
  1369. throw MakeStringException(ECLWATCH_INVALID_INPUT, "queue not specified.");
  1370. Owned<IWorkUnit> lw = factory->updateWorkUnit(wuid);
  1371. if (!lw)
  1372. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Cannot update Workunit %s", wuid);
  1373. lw->setPriority(priority);
  1374. // set job priority to queue
  1375. int priorityValue = lw->getPriorityValue();
  1376. {
  1377. CriticalBlock b(crit);
  1378. Owned<IJobQueue> queue = createJobQueue(queueName);
  1379. QueueLock lock(queue);
  1380. queue->changePriority(wuid,priorityValue);
  1381. }
  1382. return;
  1383. }
  1384. bool CWsSMCEx::onSetJobPriority(IEspContext &context, IEspSMCPriorityRequest &req, IEspSMCPriorityResponse &resp)
  1385. {
  1386. try
  1387. {
  1388. WUPriorityClass priority = PriorityClassNormal;
  1389. if(strieq(req.getPriority(),"high"))
  1390. priority = PriorityClassHigh;
  1391. else if(strieq(req.getPriority(),"low"))
  1392. priority = PriorityClassLow;
  1393. {
  1394. Owned<IWorkUnitFactory> factory = getSecWorkUnitFactory(*context.querySecManager(), *context.queryUser());
  1395. IArrayOf<IConstSMCJob>& jobs = req.getSMCJobs();
  1396. if (!jobs.length())
  1397. setJobPriority(factory, req.getWuid(), req.getQueueName(), priority);
  1398. else
  1399. {
  1400. ForEachItemIn(i, jobs)
  1401. {
  1402. IConstSMCJob &item = jobs.item(i);
  1403. const char *wuid = item.getWuid();
  1404. const char *queueName = item.getQueueName();
  1405. if (wuid && *wuid && queueName && *queueName)
  1406. setJobPriority(factory, wuid, queueName, priority);
  1407. }
  1408. }
  1409. }
  1410. clearActivityInfoCache();
  1411. resp.setRedirectUrl("/WsSMC/");
  1412. }
  1413. catch(IException* e)
  1414. {
  1415. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1416. }
  1417. return true;
  1418. }
  1419. bool CWsSMCEx::onGetThorQueueAvailability(IEspContext &context, IEspGetThorQueueAvailabilityRequest &req, IEspGetThorQueueAvailabilityResponse& resp)
  1420. {
  1421. try
  1422. {
  1423. if (!context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, false))
  1424. throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Failed to get Thor Queue availability. Permission denied.");
  1425. StringArray thorNames, groupNames, targetNames, queueNames;
  1426. getEnvironmentThorClusterNames(thorNames, groupNames, targetNames, queueNames);
  1427. IArrayOf<IEspThorCluster> ThorClusters;
  1428. ForEachItemIn(x, thorNames)
  1429. {
  1430. const char* targetName = targetNames.item(x);
  1431. const char* queueName = queueNames.item(x);
  1432. IEspThorCluster* returnCluster = new CThorCluster("","");
  1433. returnCluster->setClusterName(targetName);
  1434. returnCluster->setQueueName(queueName);
  1435. StringBuffer info;
  1436. Owned<IJobQueue> queue = createJobQueue(queueName);
  1437. if(queue->stopped(info))
  1438. returnCluster->setQueueStatus("stopped");
  1439. else if (queue->paused(info))
  1440. returnCluster->setQueueStatus("paused");
  1441. else
  1442. returnCluster->setQueueStatus("running");
  1443. unsigned enqueued=0;
  1444. unsigned connected=0;
  1445. unsigned waiting=0;
  1446. queue->getStats(connected,waiting,enqueued);
  1447. returnCluster->setQueueAvailable(waiting);
  1448. returnCluster->setJobsRunning(connected - waiting);
  1449. returnCluster->setJobsInQueue(enqueued);
  1450. ThorClusters.append(*returnCluster);
  1451. }
  1452. resp.setThorClusters(ThorClusters);
  1453. }
  1454. catch(IException* e)
  1455. {
  1456. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1457. }
  1458. return true;
  1459. }
  1460. bool CWsSMCEx::onSetBanner(IEspContext &context, IEspSetBannerRequest &req, IEspSetBannerResponse& resp)
  1461. {
  1462. try
  1463. {
  1464. #ifdef _USE_OPENLDAP
  1465. CLdapSecManager* secmgr = dynamic_cast<CLdapSecManager*>(context.querySecManager());
  1466. if(!secmgr || !secmgr->isSuperUser(context.queryUser()))
  1467. throw MakeStringException(ECLWATCH_SUPER_USER_ACCESS_DENIED, "access denied, administrators only.");
  1468. #endif
  1469. StringBuffer chatURLStr, bannerStr;
  1470. const char* chatURL = req.getChatURL();
  1471. const char* banner = req.getBannerContent();
  1472. //Only display valid strings
  1473. if (chatURL)
  1474. {
  1475. const char* pStr = chatURL;
  1476. for (unsigned i = 0; i < strlen(chatURL); i++)
  1477. {
  1478. if ((pStr[0] > 31) && (pStr[0] < 127))
  1479. chatURLStr.append(pStr[0]);
  1480. pStr++;
  1481. }
  1482. }
  1483. if (banner)
  1484. {
  1485. const char* pStr = banner;
  1486. for (unsigned i = 0; i < strlen(banner); i++)
  1487. {
  1488. if ((pStr[0] > 31) && (pStr[0] < 127))
  1489. bannerStr.append(pStr[0]);
  1490. pStr++;
  1491. }
  1492. }
  1493. chatURLStr.trim();
  1494. bannerStr.trim();
  1495. if (!req.getBannerAction_isNull() && req.getBannerAction() && (bannerStr.length() < 1))
  1496. {
  1497. throw MakeStringException(ECLWATCH_MISSING_BANNER_CONTENT, "If a Banner is enabled, the Banner content must be specified.");
  1498. }
  1499. if (!req.getEnableChatURL_isNull() && req.getEnableChatURL() && (!req.getChatURL() || !*req.getChatURL()))
  1500. {
  1501. throw MakeStringException(ECLWATCH_MISSING_CHAT_URL, "If a Chat is enabled, the Chat URL must be specified.");
  1502. }
  1503. m_ChatURL = chatURLStr;
  1504. m_Banner = bannerStr;
  1505. const char* bannerSize = req.getBannerSize();
  1506. if (bannerSize && *bannerSize)
  1507. m_BannerSize.clear().append(bannerSize);
  1508. const char* bannerColor = req.getBannerColor();
  1509. if (bannerColor && *bannerColor)
  1510. m_BannerColor.clear().append(bannerColor);
  1511. const char* bannerScroll = req.getBannerScroll();
  1512. if (bannerScroll && *bannerScroll)
  1513. m_BannerScroll.clear().append(bannerScroll);
  1514. m_BannerAction = 0;
  1515. if(!req.getBannerAction_isNull())
  1516. m_BannerAction = req.getBannerAction();
  1517. m_EnableChatURL = 0;
  1518. if(!req.getEnableChatURL_isNull())
  1519. m_EnableChatURL = req.getEnableChatURL();
  1520. resp.setRedirectUrl("/WsSMC/Activity");
  1521. }
  1522. catch(IException* e)
  1523. {
  1524. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1525. }
  1526. return true;
  1527. }
  1528. bool CWsSMCEx::onNotInCommunityEdition(IEspContext &context, IEspNotInCommunityEditionRequest &req, IEspNotInCommunityEditionResponse &resp)
  1529. {
  1530. return true;
  1531. }
  1532. bool CWsSMCEx::onBrowseResources(IEspContext &context, IEspBrowseResourcesRequest & req, IEspBrowseResourcesResponse & resp)
  1533. {
  1534. try
  1535. {
  1536. if (!context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, false))
  1537. throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Failed to Browse Resources. Permission denied.");
  1538. double version = context.getClientVersion();
  1539. Owned<IEnvironmentFactory> factory = getEnvironmentFactory();
  1540. Owned<IConstEnvironment> constEnv = factory->openEnvironment();
  1541. //The resource files will be downloaded from the same box of ESP (not dali)
  1542. StringBuffer ipStr;
  1543. IpAddress ipaddr = queryHostIP();
  1544. ipaddr.getIpText(ipStr);
  1545. if (ipStr.length() > 0)
  1546. {
  1547. resp.setNetAddress(ipStr.str());
  1548. Owned<IConstMachineInfo> machine = constEnv->getMachineByAddress(ipStr.str());
  1549. if (machine)
  1550. {
  1551. int os = machine->getOS();
  1552. resp.setOS(os);
  1553. }
  1554. }
  1555. if (m_PortalURL.length() > 0)
  1556. resp.setPortalURL(m_PortalURL.str());
  1557. #ifndef USE_RESOURCE
  1558. if (version > 1.12)
  1559. resp.setUseResource(false);
  1560. #else
  1561. if (version > 1.12)
  1562. resp.setUseResource(true);
  1563. //Now, get a list of resources stored inside the ESP box
  1564. IArrayOf<IEspHPCCResourceRepository> resourceRepositories;
  1565. Owned<IPropertyTree> pEnvRoot = &constEnv->getPTree();
  1566. const char* ossInstall = pEnvRoot->queryProp("EnvSettings/path");
  1567. if (!ossInstall || !*ossInstall)
  1568. {
  1569. WARNLOG("Failed to get EnvSettings/Path in environment settings.");
  1570. return true;
  1571. }
  1572. StringBuffer path;
  1573. path.appendf("%s/componentfiles/files/downloads", ossInstall);
  1574. Owned<IFile> f = createIFile(path.str());
  1575. if(!f->exists() || !f->isDirectory())
  1576. {
  1577. WARNLOG("Invalid resource folder");
  1578. return true;
  1579. }
  1580. Owned<IDirectoryIterator> di = f->directoryFiles(NULL, false, true);
  1581. if(di.get() == NULL)
  1582. {
  1583. WARNLOG("Resource folder is empty.");
  1584. return true;
  1585. }
  1586. ForEach(*di)
  1587. {
  1588. if (!di->isDir())
  1589. continue;
  1590. StringBuffer folder, path0, tmpBuf;
  1591. di->getName(folder);
  1592. if (folder.length() == 0)
  1593. continue;
  1594. path0.appendf("%s/%s/description.xml", path.str(), folder.str());
  1595. Owned<IFile> f0 = createIFile(path0.str());
  1596. if(!f0->exists())
  1597. {
  1598. WARNLOG("Description file not found for %s", folder.str());
  1599. continue;
  1600. }
  1601. OwnedIFileIO rIO = f0->openShared(IFOread,IFSHfull);
  1602. if(!rIO)
  1603. {
  1604. WARNLOG("Failed to open the description file for %s", folder.str());
  1605. continue;
  1606. }
  1607. offset_t fileSize = f0->size();
  1608. tmpBuf.ensureCapacity((unsigned)fileSize);
  1609. tmpBuf.setLength((unsigned)fileSize);
  1610. size32_t nRead = rIO->read(0, (size32_t) fileSize, (char*)tmpBuf.str());
  1611. if (nRead != fileSize)
  1612. {
  1613. WARNLOG("Failed to read the description file for %s", folder.str());
  1614. continue;
  1615. }
  1616. Owned<IPropertyTree> desc = createPTreeFromXMLString(tmpBuf.str());
  1617. if (!desc)
  1618. {
  1619. WARNLOG("Invalid description file for %s", folder.str());
  1620. continue;
  1621. }
  1622. Owned<IPropertyTreeIterator> fileIterator = desc->getElements("file");
  1623. if (!fileIterator->first())
  1624. {
  1625. WARNLOG("Invalid description file for %s", folder.str());
  1626. continue;
  1627. }
  1628. IArrayOf<IEspHPCCResource> resourcs;
  1629. do {
  1630. IPropertyTree &fileItem = fileIterator->query();
  1631. const char* filename = fileItem.queryProp("filename");
  1632. if (!filename || !*filename)
  1633. continue;
  1634. const char* name0 = fileItem.queryProp("name");
  1635. const char* description0 = fileItem.queryProp("description");
  1636. const char* version0 = fileItem.queryProp("version");
  1637. Owned<IEspHPCCResource> onefile = createHPCCResource();
  1638. onefile->setFileName(filename);
  1639. if (name0 && *name0)
  1640. onefile->setName(name0);
  1641. if (description0 && *description0)
  1642. onefile->setDescription(description0);
  1643. if (version0 && *version0)
  1644. onefile->setVersion(version0);
  1645. resourcs.append(*onefile.getLink());
  1646. } while (fileIterator->next());
  1647. if (resourcs.ordinality())
  1648. {
  1649. StringBuffer path1;
  1650. path1.appendf("%s/%s", path.str(), folder.str());
  1651. Owned<IEspHPCCResourceRepository> oneRepository = createHPCCResourceRepository();
  1652. oneRepository->setName(folder.str());
  1653. oneRepository->setPath(path1.str());
  1654. oneRepository->setHPCCResources(resourcs);
  1655. resourceRepositories.append(*oneRepository.getLink());
  1656. }
  1657. }
  1658. if (resourceRepositories.ordinality())
  1659. resp.setHPCCResourceRepositories(resourceRepositories);
  1660. #endif
  1661. }
  1662. catch(IException* e)
  1663. {
  1664. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1665. }
  1666. return true;
  1667. }
  1668. int CWsSMCSoapBindingEx::onGetForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, const char *service, const char *method)
  1669. {
  1670. try
  1671. {
  1672. if(stricmp(method,"NotInCommunityEdition")==0)
  1673. {
  1674. StringBuffer page, url, link;
  1675. request->getParameter("EEPortal", url);
  1676. if (url.length() > 0)
  1677. link.appendf("Further information can be found at <a href=\"%s\" target=\"_blank\">%s</a>.", url.str(), url.str());
  1678. page.append(
  1679. "<html>"
  1680. "<head>"
  1681. "<meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\" />"
  1682. "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/default.css\"/>"
  1683. "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/yui/build/fonts/fonts-min.css\" />"
  1684. "<title>Advanced feature in Enterprise Edition</title>"
  1685. "</head>"
  1686. "<body>"
  1687. "<h3 style=\"text-align:centre;\">Advanced feature in the Enterprise Edition</h4>"
  1688. "<p style=\"text-align:centre;\">Support for this feature is coming soon. ");
  1689. if (link.length() > 0)
  1690. page.append(link.str());
  1691. page.append("</p></body>"
  1692. "</html>");
  1693. response->setContent(page.str());
  1694. response->setContentType("text/html");
  1695. response->send();
  1696. return 0;
  1697. }
  1698. else if(stricmp(method,"DisabledInThisVersion")==0)
  1699. {
  1700. StringBuffer page;
  1701. page.append(
  1702. "<html>"
  1703. "<head>"
  1704. "<meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\" />"
  1705. "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/default.css\"/>"
  1706. "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/yui/build/fonts/fonts-min.css\" />"
  1707. "<title>Disabled Feature in This Version</title>"
  1708. "</head>"
  1709. "<body>"
  1710. "<h3 style=\"text-align:centre;\">Disabled Feature in This Version</h4>"
  1711. "<p style=\"text-align:centre;\">This feature is disabled in this version. ");
  1712. page.append("</p></body>"
  1713. "</html>");
  1714. response->setContent(page.str());
  1715. response->setContentType("text/html");
  1716. response->send();
  1717. return 0;
  1718. }
  1719. }
  1720. catch(IException* e)
  1721. {
  1722. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1723. }
  1724. return onGetForm(context, request, response, service, method);
  1725. }
  1726. inline const char *controlCmdMessage(int cmd)
  1727. {
  1728. switch (cmd)
  1729. {
  1730. case CRoxieControlCmd_ATTACH:
  1731. return "<control:unlockDali/>";
  1732. case CRoxieControlCmd_DETACH:
  1733. return "<control:lockDali/>";
  1734. case CRoxieControlCmd_RELOAD:
  1735. return "<control:reload/>";
  1736. case CRoxieControlCmd_STATE:
  1737. return "<control:state/>";
  1738. default:
  1739. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Unknown Roxie Control Command.");
  1740. }
  1741. return NULL;
  1742. }
  1743. bool CWsSMCEx::onRoxieControlCmd(IEspContext &context, IEspRoxieControlCmdRequest &req, IEspRoxieControlCmdResponse &resp)
  1744. {
  1745. if (!context.validateFeatureAccess(ROXIE_CONTROL_URL, SecAccess_Full, false))
  1746. throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Cannot Access Roxie Control. Permission denied.");
  1747. const char *process = req.getProcessCluster();
  1748. if (!process || !*process)
  1749. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Process cluster not specified.");
  1750. const char *controlReq = controlCmdMessage(req.getCommand());
  1751. SocketEndpointArray addrs;
  1752. getRoxieProcessServers(process, addrs);
  1753. if (!addrs.length())
  1754. throw MakeStringException(ECLWATCH_CANNOT_GET_ENV_INFO, "Process cluster not found.");
  1755. Owned<IPropertyTree> controlResp = sendRoxieControlAllNodes(addrs.item(0), controlReq, true, req.getWait());
  1756. if (!controlResp)
  1757. throw MakeStringException(ECLWATCH_INTERNAL_ERROR, "Failed to get control response from roxie.");
  1758. IArrayOf<IEspRoxieControlEndpointInfo> respEndpoints;
  1759. Owned<IPropertyTreeIterator> roxieEndpoints = controlResp->getElements("Endpoint");
  1760. ForEach(*roxieEndpoints)
  1761. {
  1762. IPropertyTree &roxieEndpoint = roxieEndpoints->query();
  1763. Owned<IEspRoxieControlEndpointInfo> respEndpoint = createRoxieControlEndpointInfo();
  1764. respEndpoint->setAddress(roxieEndpoint.queryProp("@ep"));
  1765. respEndpoint->setStatus(roxieEndpoint.queryProp("Status"));
  1766. if (roxieEndpoint.hasProp("Dali/@connected"))
  1767. respEndpoint->setAttached(roxieEndpoint.getPropBool("Dali/@connected"));
  1768. if (roxieEndpoint.hasProp("State/@hash"))
  1769. respEndpoint->setStateHash(roxieEndpoint.queryProp("State/@hash"));
  1770. respEndpoints.append(*respEndpoint.getClear());
  1771. }
  1772. resp.setEndpoints(respEndpoints);
  1773. return true;
  1774. }
  1775. bool CWsSMCEx::onGetStatusServerInfo(IEspContext &context, IEspGetStatusServerInfoRequest &req, IEspGetStatusServerInfoResponse &resp)
  1776. {
  1777. getStatusServerInfo(context, req.getServerType(), req.getServerName(), req.getNetworkAddress(), req.getPort(), resp.updateStatusServerInfo());
  1778. return true;
  1779. }
  1780. void CWsSMCEx::getStatusServerInfo(IEspContext &context, const char *serverType, const char *server, const char *networkAddress, unsigned port,
  1781. IEspStatusServerInfo& statusServerInfo)
  1782. {
  1783. if (!serverType || !*serverType)
  1784. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Server type not specified.");
  1785. Owned<ActivityInfo> activityInfo = getActivityInfo(context);
  1786. if (!activityInfo)
  1787. throw MakeStringException(ECLWATCH_INTERNAL_ERROR, "Failed to get Activity Info cache.");
  1788. if (strieq(serverType,STATUS_SERVER_THOR))
  1789. {
  1790. setTargetClusterInfo(context, serverType, server, activityInfo->thorTargetClusters, activityInfo->aws, statusServerInfo);
  1791. }
  1792. else if (strieq(serverType,STATUS_SERVER_ROXIE))
  1793. {
  1794. setTargetClusterInfo(context, serverType, server, activityInfo->roxieTargetClusters, activityInfo->aws, statusServerInfo);
  1795. }
  1796. else if (strieq(serverType,STATUS_SERVER_HTHOR))
  1797. {
  1798. setTargetClusterInfo(context, serverType, server, activityInfo->hthorTargetClusters, activityInfo->aws, statusServerInfo);
  1799. }
  1800. else if (strieq(serverType,STATUS_SERVER_DFUSERVER))
  1801. {
  1802. setServerQueueInfo(context, serverType, server, activityInfo->serverJobQueues, activityInfo->aws, statusServerInfo);
  1803. }
  1804. else
  1805. {
  1806. setServerQueueInfo(context, serverType, networkAddress, port, activityInfo->serverJobQueues, activityInfo->aws, statusServerInfo);
  1807. }
  1808. }
  1809. void CWsSMCEx::setTargetClusterInfo(IEspContext &context, const char *serverType, const char *clusterName, const CIArrayOf<CWsSMCTargetCluster>& targetClusters,
  1810. const IArrayOf<IEspActiveWorkunit>& aws, IEspStatusServerInfo& statusServerInfo)
  1811. {
  1812. if (!clusterName || !*clusterName)
  1813. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Cluster name not specified.");
  1814. IEspTargetCluster& clusterInfo = statusServerInfo.updateTargetClusterInfo();
  1815. ForEachItemIn(i, targetClusters)
  1816. {
  1817. CWsSMCTargetCluster& targetCluster = targetClusters.item(i);
  1818. const char* name = targetCluster.clusterName.get();
  1819. if (name && strieq(name, clusterName))
  1820. {
  1821. setESPTargetCluster(context, targetCluster, &clusterInfo);
  1822. break;
  1823. }
  1824. }
  1825. setActiveWUs(context, serverType, clusterName, clusterInfo.getQueueName(), aws, statusServerInfo);
  1826. }
  1827. void CWsSMCEx::setServerQueueInfo(IEspContext &context, const char *serverType, const char *serverName, const IArrayOf<IEspServerJobQueue>& serverJobQueues,
  1828. const IArrayOf<IEspActiveWorkunit>& aws, IEspStatusServerInfo& statusServerInfo)
  1829. {
  1830. if (!serverName || !*serverName)
  1831. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Server name not specified.");
  1832. ForEachItemIn(i, serverJobQueues)
  1833. {
  1834. IEspServerJobQueue& serverJobQueue = serverJobQueues.item(i);
  1835. const char* name = serverJobQueue.getServerName();
  1836. if (name && strieq(name, serverName))
  1837. {
  1838. IEspServerJobQueue& serverQueue = statusServerInfo.updateServerInfo();
  1839. serverQueue.copy(serverJobQueue);
  1840. break;
  1841. }
  1842. }
  1843. setActiveWUs(context, serverType, serverName, aws, statusServerInfo);
  1844. }
  1845. void CWsSMCEx::setServerQueueInfo(IEspContext &context, const char *serverType, const char *networkAddress, unsigned port, const IArrayOf<IEspServerJobQueue>& serverJobQueues,
  1846. const IArrayOf<IEspActiveWorkunit>& aws, IEspStatusServerInfo& statusServerInfo)
  1847. {
  1848. if (!networkAddress || !*networkAddress)
  1849. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Network address not specified.");
  1850. ForEachItemIn(i, serverJobQueues)
  1851. {
  1852. IEspServerJobQueue& serverJobQueue = serverJobQueues.item(i);
  1853. const char* ipAddress = serverJobQueue.getNetworkAddress();
  1854. unsigned thePort = serverJobQueue.getPort();
  1855. if (ipAddress && strieq(ipAddress, networkAddress) && (thePort == port))
  1856. {
  1857. IEspServerJobQueue& serverQueue = statusServerInfo.updateServerInfo();
  1858. serverQueue.copy(serverJobQueue);
  1859. break;
  1860. }
  1861. }
  1862. VStringBuffer instance("%s_on_%s:%d", serverType, networkAddress, port);
  1863. setActiveWUs(context, serverType, instance.str(), aws, statusServerInfo);
  1864. }
  1865. void CWsSMCEx::setESPTargetCluster(IEspContext &context, const CWsSMCTargetCluster& targetCluster, IEspTargetCluster* espTargetCluster)
  1866. {
  1867. espTargetCluster->setClusterName(targetCluster.clusterName.get());
  1868. espTargetCluster->setClusterSize(targetCluster.clusterSize);
  1869. espTargetCluster->setClusterType(targetCluster.clusterType);
  1870. espTargetCluster->setQueueName(targetCluster.queueName.get());
  1871. espTargetCluster->setQueueStatus(targetCluster.queueStatus.get());
  1872. setClusterStatus(context, targetCluster, espTargetCluster);
  1873. }
  1874. void CWsSMCEx::setActiveWUs(IEspContext &context, const char *serverType, const char *clusterName, const char *queueName, const IArrayOf<IEspActiveWorkunit>& aws, IEspStatusServerInfo& statusServerInfo)
  1875. {
  1876. const char* clusterType = CLUSTER_TYPE_THOR;
  1877. if (strieq(serverType,STATUS_SERVER_ROXIE))
  1878. clusterType = CLUSTER_TYPE_ROXIE;
  1879. else if (strieq(serverType,STATUS_SERVER_HTHOR))
  1880. clusterType = CLUSTER_TYPE_HTHOR;
  1881. IArrayOf<IEspActiveWorkunit> awsOnThisQueue;
  1882. ForEachItemIn(i, aws)
  1883. {
  1884. IEspActiveWorkunit& wu = aws.item(i);
  1885. const char* wuid = wu.getWuid();
  1886. if (!wuid || !*wuid)
  1887. continue;
  1888. const char* wuServerType = wu.getServer();
  1889. const char* wuClusterName = wu.getTargetClusterName();
  1890. if (!wuServerType || !wuClusterName || !strieq(serverType, wuServerType) || !strieq(clusterName, wuClusterName))
  1891. {
  1892. const char* wuClusterType = wu.getClusterType();
  1893. const char* wuClusterQueueName = wu.getClusterQueueName();
  1894. if (!wuClusterType || !wuClusterQueueName || !strieq(clusterType, wuClusterType) || !strieq(queueName, wuClusterQueueName))
  1895. continue;
  1896. }
  1897. Owned<IEspActiveWorkunit> wuOnThisQueue = new CActiveWorkunitWrapper(wuid, "", "", "", "");
  1898. setActiveWUs(context, wu, wuOnThisQueue);
  1899. awsOnThisQueue.append(*wuOnThisQueue.getLink());
  1900. }
  1901. statusServerInfo.setWorkunits(awsOnThisQueue);
  1902. }
  1903. void CWsSMCEx::setActiveWUs(IEspContext &context, const char *serverType, const char *instance, const IArrayOf<IEspActiveWorkunit>& aws, IEspStatusServerInfo& statusServerInfo)
  1904. {
  1905. IArrayOf<IEspActiveWorkunit> awsOnThisQueue;
  1906. ForEachItemIn(i, aws)
  1907. {
  1908. IEspActiveWorkunit& wu = aws.item(i);
  1909. const char* wuid = wu.getWuid();
  1910. if (!wuid || !*wuid)
  1911. continue;
  1912. const char* wuInstance = wu.getInstance();
  1913. if (!wuInstance || !strieq(wuInstance, instance))
  1914. continue;
  1915. Owned<IEspActiveWorkunit> wuOnThisQueue = new CActiveWorkunitWrapper(wuid, "", "", "", "");
  1916. setActiveWUs(context, wu, wuOnThisQueue);
  1917. awsOnThisQueue.append(*wuOnThisQueue.getLink());
  1918. }
  1919. statusServerInfo.setWorkunits(awsOnThisQueue);
  1920. }
  1921. void CWsSMCEx::setActiveWUs(IEspContext &context, IEspActiveWorkunit& wu, IEspActiveWorkunit* wuToSet)
  1922. {
  1923. try
  1924. {
  1925. const char* user = context.queryUserId();
  1926. const char* owner = wu.getOwner();
  1927. //if no access, throw an exception and go to the 'catch' section.
  1928. context.validateFeatureAccess((!owner || !*owner || (user && streq(user, owner))) ? OWN_WU_ACCESS : OTHERS_WU_ACCESS, SecAccess_Read, true);
  1929. wuToSet->copy(wu);
  1930. }
  1931. catch (IException *e)
  1932. { //if the wu cannot be opened for some reason, the openWorkUnit() inside the CActiveWorkunitWrapper() may throw an exception.
  1933. //We do not want the exception stops this process of retrieving/showing all active WUs. And that WU should still be displayed
  1934. //with the exception.
  1935. wuToSet->setStateID(WUStateUnknown);
  1936. wuToSet->setServer(wu.getServer());
  1937. wuToSet->setQueueName(wu.getQueueName());
  1938. const char* instanceName = wu.getInstance();
  1939. const char* targetClusterName = wu.getTargetClusterName();
  1940. if (instanceName && *instanceName)
  1941. wuToSet->setInstance(instanceName); // JCSMORE In thor case at least, if queued it is unknown which instance it will run on..
  1942. if (targetClusterName && *targetClusterName)
  1943. wuToSet->setTargetClusterName(targetClusterName);
  1944. e->Release();
  1945. }
  1946. }