123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689 |
- /*##############################################################################
- HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems.
- Licensed under the Apache License, Version 2.0 (the "License");
- you may not use this file except in compliance with the License.
- You may obtain a copy of the License at
- http://www.apache.org/licenses/LICENSE-2.0
- Unless required by applicable law or agreed to in writing, software
- distributed under the License is distributed on an "AS IS" BASIS,
- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- See the License for the specific language governing permissions and
- limitations under the License.
- ############################################################################## */
- #pragma warning (disable : 4786)
- #include "build-config.h"
- #ifdef _USE_OPENLDAP
- #include "ldapsecurity.ipp"
- #endif
- #include "ws_smcService.hpp"
- #include "wshelpers.hpp"
- #include "dalienv.hpp"
- #include "WUWrapper.hpp"
- #include "wujobq.hpp"
- #include "dfuwu.hpp"
- #include "exception_util.hpp"
- #include "roxiecontrol.hpp"
- static const char* FEATURE_URL = "SmcAccess";
- const char* THORQUEUE_FEATURE = "ThorQueueAccess";
- static const char* ROXIE_CONTROL_URL = "RoxieControlAccess";
- const char* PERMISSIONS_FILENAME = "espsmc_permissions.xml";
- void AccessSuccess(IEspContext& context, char const * msg,...) __attribute__((format(printf, 2, 3)));
- void AccessSuccess(IEspContext& context, char const * msg,...)
- {
- StringBuffer buf;
- buf.appendf("User %s: ",context.queryUserId());
- va_list args;
- va_start(args, msg);
- buf.valist_appendf(msg, args);
- va_end(args);
- AUDIT(AUDIT_TYPE_ACCESS_SUCCESS,buf.str());
- }
- void AccessFailure(IEspContext& context, char const * msg,...) __attribute__((format(printf, 2, 3)));
- void AccessFailure(IEspContext& context, char const * msg,...)
- {
- StringBuffer buf;
- buf.appendf("User %s: ",context.queryUserId());
- va_list args;
- va_start(args, msg);
- buf.valist_appendf(msg, args);
- va_end(args);
- AUDIT(AUDIT_TYPE_ACCESS_FAILURE,buf.str());
- }
- struct QueueWrapper
- {
- QueueWrapper(const char* targetName, const char* queueExt)
- {
- StringBuffer name;
- name.append(targetName).append('.').append(queueExt);
- queue.setown(createJobQueue(name.str()));
- }
- QueueWrapper(const char* queueName)
- {
- queue.setown(createJobQueue(queueName));
- }
- operator IJobQueue*() { return queue.get(); }
- IJobQueue* operator->() { return queue.get(); }
- Owned<IJobQueue> queue;
- };
- struct QueueLock
- {
- QueueLock(IJobQueue* q): queue(q) { queue->lock(); }
- ~QueueLock()
- {
- queue->unlock();
- }
- Linked<IJobQueue> queue;
- };
- void CWsSMCEx::init(IPropertyTree *cfg, const char *process, const char *service)
- {
- if (!daliClientActive())
- {
- ERRLOG("No Dali Connection Active.");
- throw MakeStringException(-1, "No Dali Connection Active. Please Specify a Dali to connect to in you configuration file");
- }
- m_BannerAction = 0;
- m_EnableChatURL = false;
- m_BannerSize = "4";
- m_BannerColor = "red";
- m_BannerScroll = "2";
- StringBuffer xpath;
- xpath.appendf("Software/EspProcess[@name='%s']/@portalurl", process);
- const char* portalURL = cfg->queryProp(xpath.str());
- if (portalURL && *portalURL)
- m_PortalURL.append(portalURL);
- }
- static void countProgress(IPropertyTree *t,unsigned &done,unsigned &total)
- {
- total = 0;
- done = 0;
- Owned<IPropertyTreeIterator> it = t->getElements("DFT/progress");
- ForEach(*it) {
- IPropertyTree &e=it->query();
- if (e.getPropInt("@done",0))
- done++;
- total++;
- }
- }
- struct CActiveWorkunitWrapper: public CActiveWorkunit
- {
- CActiveWorkunitWrapper(IEspContext &context, const char* wuid,unsigned index=0): CActiveWorkunit("","")
- {
- double version = context.getClientVersion();
- CWUWrapper wu(wuid, context);
- StringBuffer stateStr;
- SCMStringBuffer state,owner,jobname;
- setWuid(wuid);
- wu->getStateDesc(state);
- if(index)
- stateStr.appendf("queued(%d) [%s]", index, state.str());
- else
- stateStr.set(state.str());
- setState(stateStr.str());
- setStateID(wu->getState());
- if ((version > 1.09) && (wu->getState() == WUStateFailed))
- setWarning("The job will ultimately not complete. Please check ECLAgent.");
- setOwner(wu->getUser(owner).str());
- setJobname(wu->getJobName(jobname).str());
- switch(wu->getPriority())
- {
- case PriorityClassHigh: setPriority("high"); break;
- default:
- case PriorityClassNormal: setPriority("normal"); break;
- case PriorityClassLow: setPriority("low"); break;
- }
- if (version > 1.08 && wu->isPausing())
- {
- setIsPausing(true);
- }
- if (version > 1.14)
- {
- SCMStringBuffer clusterName;
- setClusterName(wu->getClusterName(clusterName).str());
- }
- }
- CActiveWorkunitWrapper(const char* wuid,const char* owner, const char* jobname, const char* state, const char* priority): CActiveWorkunit("","")
- {
- setWuid(wuid);
- setState(state);
- setOwner(owner);
- setJobname(jobname);
- setPriority(priority);
- }
- };
- bool CWsSMCEx::onIndex(IEspContext &context, IEspSMCIndexRequest &req, IEspSMCIndexResponse &resp)
- {
- resp.setRedirectUrl("/");
- return true;
- }
- static int stringcmp(const char **a, const char **b)
- {
- return strcmp(*a, *b);
- }
- bool isInWuList(IArrayOf<IEspActiveWorkunit>& aws, const char* wuid)
- {
- bool bFound = false;
- if (wuid && *wuid && (aws.length() > 0))
- {
- ForEachItemIn(k, aws)
- {
- IEspActiveWorkunit& wu = aws.item(k);
- const char* wuid0 = wu.getWuid();
- const char* server0 = wu.getServer();
- if (wuid0 && !strcmp(wuid0, wuid) && (!server0 || strcmp(server0, "ECLagent")))
- {
- bFound = true;
- break;
- }
- }
- }
- return bFound;
- }
- void addQueuedWorkUnits(const char *queueName, CJobQueueContents &contents, IArrayOf<IEspActiveWorkunit> &aws, IEspContext &context, const char *serverName, const char *instanceName)
- {
- Owned<IJobQueueIterator> iter = contents.getIterator();
- unsigned count=0;
- ForEach(*iter)
- {
- if (!isInWuList(aws, iter->query().queryWUID()))
- {
- try
- {
- Owned<IEspActiveWorkunit> wu(new CActiveWorkunitWrapper(context, iter->query().queryWUID(),++count));
- wu->setServer(serverName);
- wu->setInstance(instanceName); // JCSMORE In thor case at least, if queued it is unknown which instance it will run on..
- wu->setQueueName(queueName);
- aws.append(*wu.getLink());
- }
- catch (IException *e)
- {
- // JCSMORE->KWang what is this handling? Why would this succeeed and above fail?
- StringBuffer msg;
- Owned<IEspActiveWorkunit> wu(new CActiveWorkunitWrapper(iter->query().queryWUID(), "", "", e->errorMessage(msg).str(), "normal"));
- wu->setServer(serverName);
- wu->setInstance(instanceName);
- wu->setQueueName(queueName);
- aws.append(*wu.getLink());
- }
- }
- }
- }
- void CWsSMCEx::getQueueState(int runningJobsInQueue, StringBuffer& queueState, BulletType& bulletType)
- {
- bool queuePausedOrStopped = false;
- if ((queueState.length() > 0) && (strieq(queueState.str(),"stopped") || strieq(queueState.str(),"paused")))
- queuePausedOrStopped = true;
- else
- queueState.set("running");
- bulletType = bulletGreen;
- if (NotFound == runningJobsInQueue)
- {
- if (queuePausedOrStopped)
- bulletType = bulletWhite;
- else
- bulletType = bulletError;
- }
- else if (runningJobsInQueue > 0)
- {
- if (queuePausedOrStopped)
- bulletType = bulletOrange;
- else
- bulletType = bulletGreen;
- }
- else if (queuePausedOrStopped)
- bulletType = bulletYellow;
- return;
- }
- void CWsSMCEx::readClusterTypeAndQueueName(CConstWUClusterInfoArray& clusters, const char* clusterName, StringBuffer& clusterType, SCMStringBuffer& clusterQueue)
- {
- if (!clusterName || !*clusterName)
- return;
- ForEachItemIn(cl, clusters)
- {
- IConstWUClusterInfo &cluster = clusters.item(cl);
- SCMStringBuffer str;
- cluster.getName(str);
- if (!streq(str.str(), clusterName))
- continue;
- if (cluster.getPlatform() == HThorCluster)
- {
- cluster.getAgentQueue(clusterQueue);
- clusterType.set("HThor");
- }
- else if (cluster.getPlatform() == RoxieCluster)
- {
- cluster.getAgentQueue(clusterQueue);
- clusterType.set("Roxie");
- }
- else
- {
- cluster.getThorQueue(clusterQueue);
- clusterType.set("Thor");
- }
- break;
- }
- return;
- }
- void CWsSMCEx::addRunningWUs(IEspContext &context, IPropertyTree& node, CConstWUClusterInfoArray& clusters,
- IArrayOf<IEspActiveWorkunit>& aws, BoolHash& uniqueWUIDs,
- StringArray& runningQueueNames, int* runningJobsInQueue)
- {
- StringBuffer instance;
- StringBuffer qname;
- int serverID = -1;
- const char* name = node.queryProp("@name");
- if (name && *name)
- {
- node.getProp("@queue", qname);
- if (0 == stricmp("ThorMaster", name))
- {
- node.getProp("@thorname",instance);
- }
- else if (0 == stricmp(name, "ECLAgent"))
- {
- qname.append(name);
- }
- if ((instance.length()==0))
- {
- instance.append( !strcmp(name, "ECLagent") ? "ECL agent" : name);
- instance.append(" on ").append(node.queryProp("@node"));
- }
- }
- if (qname.length() > 0)
- {
- StringArray qlist;
- qlist.appendListUniq(qname.str(), ",");
- ForEachItemIn(q, qlist)
- {
- const char *_qname = qlist.item(q);
- serverID = runningQueueNames.find(_qname);
- if (NotFound == serverID)
- {
- serverID = runningQueueNames.ordinality(); // i.e. last
- runningQueueNames.append(_qname);
- }
- }
- }
- Owned<IPropertyTreeIterator> wuids(node.getElements("WorkUnit"));
- ForEach(*wuids)
- {
- const char* wuid=wuids->query().queryProp(NULL);
- if (!wuid)
- continue;
- if (streq(qname.str(), "ECLagent") && uniqueWUIDs.getValue(wuid))
- continue;
- try
- {
- IEspActiveWorkunit* wu=new CActiveWorkunitWrapper(context,wuid);
- const char* servername = node.queryProp("@name");
- const char *cluster = node.queryProp("Cluster");
- wu->setServer(servername);
- wu->setInstance(instance.str());
- StringBuffer queueName;
- if (cluster) // backward compat check.
- getClusterThorQueueName(queueName, cluster);
- else
- queueName.append(qname);
- serverID = runningQueueNames.find(queueName.str());
- wu->setQueueName(queueName);
- double version = context.getClientVersion();
- if (version > 1.01)
- {
- if (wu->getStateID() == WUStateRunning)
- {
- int sg_duration = node.getPropInt("@sg_duration", -1);
- const char* graph = node.queryProp("@graph");
- int subgraph = node.getPropInt("@subgraph", -1);
- if (subgraph > -1 && sg_duration > -1)
- {
- StringBuffer durationStr;
- StringBuffer subgraphStr;
- durationStr.appendf("%d min", sg_duration);
- subgraphStr.appendf("%d", subgraph);
- wu->setGraphName(graph);
- wu->setDuration(durationStr.str());
- wu->setGID(subgraphStr.str());
- }
- int memoryBlocked = node.getPropInt("@memoryBlocked ", 0);
- if (memoryBlocked != 0)
- {
- wu->setMemoryBlocked(1);
- }
- if (serverID > -1)
- {
- runningJobsInQueue[serverID]++;
- }
- if ((version > 1.14) && streq(queueName.str(), "ECLagent"))
- {
- const char* clusterName = wu->getClusterName();
- if (clusterName && *clusterName)
- {
- StringBuffer clusterType;
- SCMStringBuffer clusterQueue;
- readClusterTypeAndQueueName(clusters, clusterName, clusterType, clusterQueue);
- wu->setClusterType(clusterType.str());
- wu->setClusterQueueName(clusterQueue.str());
- }
- }
- }
- }
- uniqueWUIDs.setValue(wuid, true);
- aws.append(*wu);
- }
- catch (IException *e)
- {
- StringBuffer msg;
- Owned<IEspActiveWorkunit> wu(new CActiveWorkunitWrapper(wuid, "", "", e->errorMessage(msg).str(), "normal"));
- wu->setServer(node.queryProp("@name"));
- wu->setInstance(instance.str());
- wu->setQueueName(qname.str());
- aws.append(*wu.getLink());
- }
- }
- return;
- }
- // This method reads job information from both /Status/Servers and IJobQueue.
- //
- // Each server component (a thor cluster, a dfuserver, or an eclagent) is one 'Server' branch under
- // /Status/Servers. A 'Server' branch has a @queue which indicates the queue name of the server.
- // A 'Server' branch also contains the information about running WUs on that 'Server'. This
- // method reads the information. Those WUs are displays under that server (identified by its queue name)
- // on Activity page.
- //
- // For the WUs list inside /Status/Servers/Server[@name=ECLagent] but not list under other 'Server', the
- // existing code has to find out WUID and @clusterName of the WU. Then, uses @clusterName to find out the
- // queue name in IConstWUClusterInfo. Those WUs list under that server (identified by its queue name) with
- // a note 'on ECLagent'. TBD: the logic here will be simpler if the /Status/Servers/Server is named the
- // instance and/or cluster.
- //
- // In order to get information about queued WUs, this method gets queue names from both IConstWUClusterInfo
- // and other environment functions. Each of those queue names is linked to one IJobQueues. From the
- // IJobQueues, this method reads queued jobs for each server component and list them under the server
- // component (identified by its queue name).
- bool CWsSMCEx::onActivity(IEspContext &context, IEspActivityRequest &req, IEspActivityResponse& resp)
- {
- context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, true);
- try
- {
- const char* build_ver = getBuildVersion();
- resp.setBuild(build_ver);
- double version = context.getClientVersion();
- bool isSuperUser = true;
- #ifdef _USE_OPENLDAP
- CLdapSecManager* secmgr = dynamic_cast<CLdapSecManager*>(context.querySecManager());
- if(secmgr && !secmgr->isSuperUser(context.queryUser()))
- isSuperUser = false;
- #endif
- if(isSuperUser && req.getFromSubmitBtn())
- {
- StringBuffer chatURLStr, bannerStr;
- const char* chatURL = req.getChatURL();
- const char* banner = req.getBannerContent();
- //Only display valid strings
- if (chatURL)
- {
- const char* pStr = chatURL;
- for (unsigned i = 0; i < strlen(chatURL); i++)
- {
- if ((pStr[0] > 31) && (pStr[0] < 127))
- chatURLStr.append(pStr[0]);
- pStr++;
- }
- }
- if (banner)
- {
- const char* pStr = banner;
- for (unsigned i = 0; i < strlen(banner); i++)
- {
- if ((pStr[0] > 31) && (pStr[0] < 127))
- bannerStr.append(pStr[0]);
- pStr++;
- }
- }
- chatURLStr.trim();
- bannerStr.trim();
- if (!req.getBannerAction_isNull() && req.getBannerAction() && (bannerStr.length() < 1))
- {
- throw MakeStringException(ECLWATCH_MISSING_BANNER_CONTENT, "If a Banner is enabled, the Banner content must be specified.");
- }
- if (!req.getEnableChatURL_isNull() && req.getEnableChatURL() && (chatURLStr.length() < 1))
- {
- throw MakeStringException(ECLWATCH_MISSING_CHAT_URL, "If a Chat is enabled, the Chat URL must be specified.");
- }
- m_ChatURL = chatURLStr;
- m_Banner = bannerStr;
- const char* bannerSize = req.getBannerSize();
- if (bannerSize && *bannerSize)
- m_BannerSize.clear().append(bannerSize);
- const char* bannerColor = req.getBannerColor();
- if (bannerColor && *bannerColor)
- m_BannerColor.clear().append(bannerColor);
- const char* bannerScroll = req.getBannerScroll();
- if (bannerScroll && *bannerScroll)
- m_BannerScroll.clear().append(bannerScroll);
- m_BannerAction = req.getBannerAction();
- if(!req.getEnableChatURL_isNull())
- m_EnableChatURL = req.getEnableChatURL();
- }
- if (version > 1.05)
- {
- if (version > 1.11)
- resp.setSuperUser(isSuperUser);
- resp.setShowBanner(m_BannerAction);
- resp.setShowChatURL(m_EnableChatURL);
- resp.setBannerContent(m_Banner.str());
- resp.setBannerSize(m_BannerSize.str());
- resp.setBannerColor(m_BannerColor.str());
- resp.setChatURL(m_ChatURL.str());
- if (version > 1.07)
- {
- resp.setBannerScroll(m_BannerScroll.str());
- }
- }
- Owned<IRemoteConnection> connEnv = querySDS().connect("Environment", myProcessSession(), RTM_LOCK_READ, SDS_LOCK_TIMEOUT);
- IPropertyTree* envRoot = connEnv->queryRoot();
- if (!envRoot)
- throw MakeStringException(ECLWATCH_CANNOT_GET_ENV_INFO,"Failed to get environment information.");
- CConstWUClusterInfoArray clusters;
- getEnvironmentClusterInfo(envRoot, clusters);
- BoolHash uniqueWUIDs;
- Owned<IRemoteConnection> conn = querySDS().connect("/Status/Servers",myProcessSession(),RTM_LOCK_READ,30000);
- StringArray runningQueueNames;
- int runningJobsInQueue[256];
- for (int i = 0; i < 256; i++)
- runningJobsInQueue[i] = 0;
- IArrayOf<IEspActiveWorkunit> aws;
- if (conn.get())
- {
- Owned<IPropertyTreeIterator> it(conn->queryRoot()->getElements("Server[@name!=\"ECLagent\"]"));
- ForEach(*it)
- addRunningWUs(context, it->query(), clusters, aws, uniqueWUIDs, runningQueueNames, runningJobsInQueue);
- Owned<IPropertyTreeIterator> it1(conn->queryRoot()->getElements("Server[@name=\"ECLagent\"]"));
- ForEach(*it1)
- addRunningWUs(context, it1->query(), clusters, aws, uniqueWUIDs, runningQueueNames, runningJobsInQueue);
- }
- SecAccessFlags access;
- bool fullAccess=(context.authorizeFeature(THORQUEUE_FEATURE, access) && access>=SecAccess_Full);
- IArrayOf<IEspThorCluster> ThorClusters;
- IArrayOf<IEspHThorCluster> HThorClusters;
- IArrayOf<IEspRoxieCluster> RoxieClusters;
- ForEachItemIn(c, clusters)
- {
- IConstWUClusterInfo &cluster = clusters.item(c);
- SCMStringBuffer str;
- if (cluster.getThorProcesses().ordinality())
- {
- IEspThorCluster* returnCluster = new CThorCluster("","");
- returnCluster->setThorLCR(ThorLCRCluster == cluster.getPlatform() ? "withLCR" : "noLCR");
- str.clear();
- returnCluster->setClusterName(cluster.getName(str).str());
- str.clear();
- const char *queueName = cluster.getThorQueue(str).str();
- returnCluster->setQueueName(queueName);
- StringBuffer queueState;
- CJobQueueContents contents;
- Owned<IJobQueue> queue = createJobQueue(queueName);
- queue->copyItemsAndState(contents, queueState);
- addQueuedWorkUnits(queueName, contents, aws, context, "ThorMaster", NULL);
- BulletType bulletType = bulletGreen;
- int serverID = runningQueueNames.find(queueName);
- int numRunningJobsInQueue = (NotFound != serverID) ? runningJobsInQueue[serverID] : -1;
- getQueueState(numRunningJobsInQueue, queueState, bulletType);
- StringBuffer agentQueueState;
- CJobQueueContents agentContents;
- SCMStringBuffer str1;
- const char *agentQueueName = cluster.getAgentQueue(str1).str();
- Owned<IJobQueue> agentQueue = createJobQueue(agentQueueName);
- agentQueue->copyItemsAndState(agentContents, agentQueueState);
- //Use the same 'queueName' because the job belongs to the same cluster
- addQueuedWorkUnits(queueName, agentContents, aws, context, "ThorMaster", NULL);
- if (bulletType == bulletGreen)
- {//If ThorQueue is normal, check the AgentQueue
- serverID = runningQueueNames.find(agentQueueName);
- numRunningJobsInQueue = (NotFound != serverID) ? runningJobsInQueue[serverID] : -1;
- getQueueState(numRunningJobsInQueue, queueState, bulletType);
- }
- returnCluster->setQueueStatus(queueState.str());
- if (version > 1.06)
- returnCluster->setQueueStatus2(bulletType);
- if (version > 1.10)
- returnCluster->setClusterSize(cluster.getSize());
- addToThorClusterList(ThorClusters, returnCluster, req.getSortBy(), req.getDescending());
- }
- if (version > 1.06)
- {
- str.clear();
- if (cluster.getRoxieProcess(str).length())
- {
- IEspRoxieCluster* returnCluster = new CRoxieCluster("","");
- str.clear();
- returnCluster->setClusterName(cluster.getName(str).str());
- str.clear();
- returnCluster->setQueueName(cluster.getAgentQueue(str).str());
- str.clear();
- const char *queueName = cluster.getAgentQueue(str).str();
- StringBuffer queueState;
- CJobQueueContents contents;
- Owned<IJobQueue> queue = createJobQueue(queueName);
- queue->copyItemsAndState(contents, queueState);
- addQueuedWorkUnits(queueName, contents, aws, context, "RoxieServer", NULL);
- BulletType bulletType = bulletGreen;
- int serverID = runningQueueNames.find(queueName);
- int numRunningJobsInQueue = (NotFound != serverID) ? runningJobsInQueue[serverID] : -1;
- getQueueState(numRunningJobsInQueue, queueState, bulletType);
- returnCluster->setQueueStatus(queueState.str());
- returnCluster->setQueueStatus2(bulletType);
- if (version > 1.10)
- returnCluster->setClusterSize(cluster.getSize());
- addToRoxieClusterList(RoxieClusters, returnCluster, req.getSortBy(), req.getDescending());
- }
- }
- if (version > 1.11 && (cluster.getPlatform() == HThorCluster))
- {
- IEspHThorCluster* returnCluster = new CHThorCluster("","");
- str.clear();
- returnCluster->setClusterName(cluster.getName(str).str());
- str.clear();
- returnCluster->setQueueName(cluster.getAgentQueue(str).str());
- str.clear();
- const char *queueName = cluster.getAgentQueue(str).str();
- StringBuffer queueState;
- CJobQueueContents contents;
- Owned<IJobQueue> queue = createJobQueue(queueName);
- queue->copyItemsAndState(contents, queueState);
- addQueuedWorkUnits(queueName, contents, aws, context, "HThorServer", NULL);
- BulletType bulletType = bulletGreen;
- int serverID = runningQueueNames.find(queueName);
- int numRunningJobsInQueue = (NotFound != serverID) ? runningJobsInQueue[serverID] : -1;
- getQueueState(numRunningJobsInQueue, queueState, bulletType);
- returnCluster->setQueueStatus(queueState.str());
- returnCluster->setQueueStatus2(bulletType);
- HThorClusters.append(*returnCluster);
- }
- }
- resp.setThorClusters(ThorClusters);
- if (version > 1.06)
- resp.setRoxieClusters(RoxieClusters);
- if (version > 1.10)
- {
- resp.setSortBy(req.getSortBy());
- resp.setDescending(req.getDescending());
- }
- if (version > 1.11)
- {
- resp.setHThorClusters(HThorClusters);
- if (fullAccess)
- resp.setAccessRight("Access_Full");
- }
- IArrayOf<IEspServerJobQueue> serverJobQueues;
- IArrayOf<IConstTpEclServer> eclccservers;
- CTpWrapper dummy;
- dummy.getTpEclCCServers(envRoot->queryBranch("Software"), eclccservers);
- ForEachItemIn(x1, eclccservers)
- {
- IConstTpEclServer& eclccserver = eclccservers.item(x1);
- const char* serverName = eclccserver.getName();
- if (!serverName || !*serverName)
- continue;
- Owned <IStringIterator> targetClusters = getTargetClusters(eqEclCCServer, serverName);
- if (!targetClusters->first())
- continue;
- ForEach (*targetClusters)
- {
- SCMStringBuffer targetCluster;
- targetClusters->str(targetCluster);
- StringBuffer queueName;
- StringBuffer queueState;
- CJobQueueContents contents;
- getClusterEclCCServerQueueName(queueName, targetCluster.str());
- Owned<IJobQueue> queue = createJobQueue(queueName);
- queue->copyItemsAndState(contents, queueState);
- unsigned count=0;
- Owned<IJobQueueIterator> iter = contents.getIterator();
- ForEach(*iter)
- {
- if (isInWuList(aws, iter->query().queryWUID()))
- continue;
- Owned<IEspActiveWorkunit> wu(new CActiveWorkunitWrapper(context, iter->query().queryWUID(),++count));
- wu->setServer("ECLCCserver");
- wu->setInstance(serverName);
- wu->setQueueName(queueName);
- aws.append(*wu.getLink());
- }
- addServerJobQueue(serverJobQueues, queueName, queueState.str(), serverName, "ECLCCserver");
- }
- }
- StringBuffer dirxpath;
- dirxpath.appendf("Software/%s", eqDfu);
- Owned<IPropertyTreeIterator> services = envRoot->getElements(dirxpath);
- if (services->first())
- {
- do
- {
- IPropertyTree &serviceTree = services->query();
- const char *queuename = serviceTree.queryProp("@queue");
- const char *serverName = serviceTree.queryProp("@name");
- if (queuename && *queuename)
- {
- StringArray queues;
- loop
- {
- StringAttr subq;
- const char *comma = strchr(queuename,',');
- if (comma)
- subq.set(queuename,comma-queuename);
- else
- subq.set(queuename);
- bool added;
- const char *s = strdup(subq.get());
- queues.bAdd(s, stringcmp, added);
- if (!added)
- free((void *)s);
- if (!comma)
- break;
- queuename = comma+1;
- if (!*queuename)
- break;
- }
- ForEachItemIn(q, queues)
- {
- const char *queueName = queues.item(q);
- StringAttrArray wulist;
- unsigned running = queuedJobs(queueName, wulist);
- ForEachItemIn(i, wulist)
- {
- const char *wuid = wulist.item(i).text.get();
- try
- {
- StringBuffer jname, uname, state;
- Owned<IConstDFUWorkUnit> wu = getDFUWorkUnitFactory()->openWorkUnit(wuid, false);
- if (wu)
- {
- wu->getUser(uname);
- wu->getJobName(jname);
- if (i<running)
- state.append("running");
- else
- state.append("queued");
- Owned<IEspActiveWorkunit> wu1(new CActiveWorkunitWrapper(wuid, uname.str(), jname.str(), state.str(), "normal"));
- wu1->setServer("DFUserver");
- wu1->setInstance(serverName);
- wu1->setQueueName(queueName);
- aws.append(*wu1.getLink());
- }
- }
- catch (IException *e)
- {
- StringBuffer msg;
- Owned<IEspActiveWorkunit> wu1(new CActiveWorkunitWrapper(wuid, "", "", e->errorMessage(msg).str(), "normal"));
- wu1->setServer("DFUserver");
- wu1->setInstance(serverName);
- wu1->setQueueName(queueName);
- aws.append(*wu1.getLink());
- }
- }
- addServerJobQueue(serverJobQueues, queueName, serverName, "DFUserver");
- }
- }
- } while (services->next());
- }
- resp.setRunning(aws);
- if (version > 1.03)
- resp.setServerJobQueues(serverJobQueues);
- IArrayOf<IEspDFUJob> jobs;
- conn.setown(querySDS().connect("DFU/RECOVERY",myProcessSession(),0, INFINITE));
- if (conn)
- {
- Owned<IPropertyTreeIterator> it(conn->queryRoot()->getElements("job"));
- ForEach(*it)
- {
- IPropertyTree &e=it->query();
- if (e.getPropBool("Running",false))
- {
- unsigned done;
- unsigned total;
- countProgress(&e,done,total);
- Owned<IEspDFUJob> job = new CDFUJob("","");
- job->setTimeStarted(e.queryProp("@time_started"));
- job->setDone(done);
- job->setTotal(total);
- StringBuffer cmd;
- cmd.append(e.queryProp("@command")).append(" ").append(e.queryProp("@command_parameters"));
- job->setCommand(cmd.str());
- jobs.append(*job.getLink());
- }
- }
- }
- resp.setDFUJobs(jobs);
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- void CWsSMCEx::addServerJobQueue(IArrayOf<IEspServerJobQueue>& jobQueues, const char* queueName, const char* serverName, const char* serverType)
- {
- if (!queueName || !*queueName || !serverName || !*serverName || !serverType || !*serverType)
- return;
- Owned<IEspServerJobQueue> jobQueue = createServerJobQueue("", "");
- jobQueue->setQueueName(queueName);
- jobQueue->setServerName(serverName);
- jobQueue->setServerType(serverType);
- Owned<IJobQueue> queue = createJobQueue(queueName);
- if (queue->stopped())
- jobQueue->setQueueStatus("stopped");
- else if (queue->paused())
- jobQueue->setQueueStatus("paused");
- else
- jobQueue->setQueueStatus("running");
- jobQueues.append(*jobQueue.getClear());
- }
- void CWsSMCEx::addServerJobQueue(IArrayOf<IEspServerJobQueue>& jobQueues, const char* queueName, const char* queueState, const char* serverName, const char* serverType)
- {
- if (!queueName || !*queueName || !serverName || !*serverName || !serverType || !*serverType)
- return;
- Owned<IEspServerJobQueue> jobQueue = createServerJobQueue("", "");
- jobQueue->setQueueName(queueName);
- jobQueue->setServerName(serverName);
- jobQueue->setServerType(serverType);
- if (queueState && (strieq(queueState,"stopped") || strieq(queueState,"paused")))
- jobQueue->setQueueStatus(queueState);
- else
- jobQueue->setQueueStatus("running");
- jobQueues.append(*jobQueue.getClear());
- }
- void CWsSMCEx::addToThorClusterList(IArrayOf<IEspThorCluster>& clusters, IEspThorCluster* cluster, const char* sortBy, bool descending)
- {
- if (clusters.length() < 1)
- {
- clusters.append(*cluster);
- return;
- }
- const char* clusterName = cluster->getClusterName();
- unsigned clusterSize = cluster->getClusterSize();
- bool clusterAdded = false;
- ForEachItemIn(i, clusters)
- {
- int strCmp = 0;
- IEspThorCluster& cluster1 = clusters.item(i);
- if (!sortBy || !*sortBy || strieq(sortBy, "name"))
- {
- strCmp = strcmp(cluster1.getClusterName(), clusterName);
- }
- else
- {//size
- //strCmp = cluster1.getClusterSize() - clusterSize;
- int si = cluster1.getClusterSize();
- strCmp = si - clusterSize;
- }
- if ((descending && (strCmp < 0)) || (!descending && (strCmp > 0)))
- {
- clusters.add(*cluster, i);
- clusterAdded = true;
- break;
- }
- }
- if (!clusterAdded)
- clusters.append(*cluster);
- return;
- }
- void CWsSMCEx::addToRoxieClusterList(IArrayOf<IEspRoxieCluster>& clusters, IEspRoxieCluster* cluster, const char* sortBy, bool descending)
- {
- if (clusters.length() < 1)
- {
- clusters.append(*cluster);
- return;
- }
- const char* clusterName = cluster->getClusterName();
- unsigned clusterSize = cluster->getClusterSize();
- bool clusterAdded = false;
- ForEachItemIn(i, clusters)
- {
- int strCmp = 0;
- IEspRoxieCluster& cluster1 = clusters.item(i);
- if (!sortBy || !*sortBy || strieq(sortBy, "name"))
- {
- strCmp = strcmp(cluster1.getClusterName(), clusterName);
- }
- else
- {//size
- strCmp = cluster1.getClusterSize() - clusterSize;
- }
- if ((descending && (strCmp < 0)) || (!descending && (strCmp > 0)))
- {
- clusters.add(*cluster, i);
- clusterAdded = true;
- break;
- }
- }
- if (!clusterAdded)
- clusters.append(*cluster);
- return;
- }
- void CWsSMCEx::addCapabilities(IPropertyTree* pFeatureNode, const char* access,
- IArrayOf<IEspCapability>& capabilities)
- {
- StringBuffer xpath(access);
- xpath.append("/Capability");
- Owned<IPropertyTreeIterator> it = pFeatureNode->getElements(xpath.str());
- ForEach(*it)
- {
- IPropertyTree* pCapabilityNode = &it->query();
- IEspCapability* pCapability = new CCapability("ws_smc");
- pCapability->setName( pCapabilityNode->queryProp("@name") );
- pCapability->setDescription( pCapabilityNode->queryProp("@description") );
- capabilities.append(*pCapability);
- }
- }
- static void checkAccess(IEspContext &context, const char* feature,int level)
- {
- if (!context.validateFeatureAccess(feature, level, false))
- throw MakeStringException(ECLWATCH_THOR_QUEUE_ACCESS_DENIED, "Failed to access the queue functions. Permission denied.");
- }
- bool CWsSMCEx::onMoveJobDown(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
- unsigned index=queue->findRank(req.getWuid());
- if(index<queue->ordinality())
- {
- IJobQueueItem * item0 = queue->getItem(index);
- IJobQueueItem * item = queue->getItem(index+1);
- if(item && item0 && (item0->getPriority() == item->getPriority()))
- queue->moveAfter(req.getWuid(),item->queryWUID());
- }
- AccessSuccess(context, "Changed job priority %s",req.getWuid());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onMoveJobUp(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
- unsigned index=queue->findRank(req.getWuid());
- if(index>0 && index<queue->ordinality())
- {
- IJobQueueItem * item0 = queue->getItem(index);
- IJobQueueItem * item = queue->getItem(index-1);
- if(item && item0 && (item0->getPriority() == item->getPriority()))
- queue->moveBefore(req.getWuid(),item->queryWUID());
- }
- AccessSuccess(context, "Changed job priority %s",req.getWuid());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onMoveJobBack(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
-
- unsigned index=queue->findRank(req.getWuid());
- if(index<queue->ordinality())
- {
- int priority0 = queue->getItem(index)->getPriority();
- unsigned biggestIndoxInSamePriority = index;
- unsigned nextIndex = biggestIndoxInSamePriority + 1;
- while (nextIndex<queue->ordinality())
- {
- IJobQueueItem * item = queue->getItem(nextIndex);
- if (priority0 != item->getPriority())
- {
- break;
- }
- biggestIndoxInSamePriority = nextIndex;
- nextIndex++;
- }
- if (biggestIndoxInSamePriority != index)
- {
- IJobQueueItem * item = queue->getItem(biggestIndoxInSamePriority);
- queue->moveAfter(req.getWuid(),item->queryWUID());
- }
- }
- AccessSuccess(context, "Changed job priority %s",req.getWuid());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onMoveJobFront(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
-
- unsigned index=queue->findRank(req.getWuid());
- if(index>0 && index<queue->ordinality())
- {
- int priority0 = queue->getItem(index)->getPriority();
- unsigned smallestIndoxInSamePriority = index;
- int nextIndex = smallestIndoxInSamePriority - 1;
- while (nextIndex >= 0)
- {
- IJobQueueItem * item = queue->getItem(nextIndex);
- if (priority0 != item->getPriority())
- {
- break;
- }
- smallestIndoxInSamePriority = nextIndex;
- nextIndex--;
- }
- if (smallestIndoxInSamePriority != index)
- {
- IJobQueueItem * item = queue->getItem(smallestIndoxInSamePriority);
- queue->moveBefore(req.getWuid(),item->queryWUID());
- }
- }
- AccessSuccess(context, "Changed job priority %s",req.getWuid());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onRemoveJob(IEspContext &context, IEspSMCJobRequest &req, IEspSMCJobResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- secAbortWorkUnit(req.getWuid(), *context.querySecManager(), *context.queryUser());
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
-
- unsigned index=queue->findRank(req.getWuid());
- if(index<queue->ordinality())
- {
- if(!queue->cancelInitiateConversation(req.getWuid()))
- throw MakeStringException(ECLWATCH_CANNOT_DELETE_WORKUNIT,"Failed to remove the workunit %s",req.getWuid());
- }
- AccessSuccess(context, "Removed job %s",req.getWuid());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onStopQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- queue->stop();
- AccessSuccess(context, "Stopped queue %s",req.getCluster());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onResumeQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- queue->resume();
- AccessSuccess(context, "Resumed queue %s",req.getCluster());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onPauseQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- queue->pause();
- AccessSuccess(context, "Paused queue %s",req.getCluster());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onClearQueue(IEspContext &context, IEspSMCQueueRequest &req, IEspSMCQueueResponse &resp)
- {
- try
- {
- checkAccess(context,THORQUEUE_FEATURE,SecAccess_Full);
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- {
- QueueLock lock(queue);
- for(unsigned i=0;i<queue->ordinality();i++)
- secAbortWorkUnit(queue->getItem(i)->queryWUID(), *context.querySecManager(), *context.queryUser());
- queue->clear();
- }
- AccessSuccess(context, "Cleared queue %s",req.getCluster());
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onSetJobPriority(IEspContext &context, IEspSMCPriorityRequest &req, IEspSMCPriorityResponse &resp)
- {
- try
- {
- Owned<IWorkUnitFactory> factory = getSecWorkUnitFactory(*context.querySecManager(), *context.queryUser());
- Owned<IWorkUnit> lw = factory->updateWorkUnit(req.getWuid());
- if(stricmp(req.getPriority(),"high")==0)
- lw->setPriority(PriorityClassHigh);
- else if(stricmp(req.getPriority(),"normal")==0)
- lw->setPriority(PriorityClassNormal);
- else if(stricmp(req.getPriority(),"low")==0)
- lw->setPriority(PriorityClassLow);
- // set job priority
- int priority = lw->getPriorityValue();
- Owned<IJobQueue> queue = createJobQueue(req.getQueueName());
- QueueLock lock(queue);
- queue->changePriority(req.getWuid(),priority);
- resp.setRedirectUrl("/WsSMC/");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onGetThorQueueAvailability(IEspContext &context, IEspGetThorQueueAvailabilityRequest &req, IEspGetThorQueueAvailabilityResponse& resp)
- {
- try
- {
- if (!context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, false))
- throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Failed to get Thor Queue availability. Permission denied.");
- StringArray thorNames, groupNames, targetNames, queueNames;
- getEnvironmentThorClusterNames(thorNames, groupNames, targetNames, queueNames);
- IArrayOf<IEspThorCluster> ThorClusters;
- ForEachItemIn(x, thorNames)
- {
- const char* targetName = targetNames.item(x);
- const char* queueName = queueNames.item(x);
- IEspThorCluster* returnCluster = new CThorCluster("","");
-
- returnCluster->setClusterName(targetName);
- returnCluster->setQueueName(queueName);
- Owned<IJobQueue> queue = createJobQueue(queueName);
- if(queue->stopped())
- returnCluster->setQueueStatus("stopped");
- else if (queue->paused())
- returnCluster->setQueueStatus("paused");
- else
- returnCluster->setQueueStatus("running");
- unsigned enqueued=0;
- unsigned connected=0;
- unsigned waiting=0;
- queue->getStats(connected,waiting,enqueued);
- returnCluster->setQueueAvailable(waiting);
- returnCluster->setJobsRunning(connected - waiting);
- returnCluster->setJobsInQueue(enqueued);
- ThorClusters.append(*returnCluster);
- }
- resp.setThorClusters(ThorClusters);
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onSetBanner(IEspContext &context, IEspSetBannerRequest &req, IEspSetBannerResponse& resp)
- {
- try
- {
- #ifdef _USE_OPENLDAP
- CLdapSecManager* secmgr = dynamic_cast<CLdapSecManager*>(context.querySecManager());
- if(!secmgr || !secmgr->isSuperUser(context.queryUser()))
- throw MakeStringException(ECLWATCH_SUPER_USER_ACCESS_DENIED, "access denied, administrators only.");
- #endif
- StringBuffer chatURLStr, bannerStr;
- const char* chatURL = req.getChatURL();
- const char* banner = req.getBannerContent();
- //Only display valid strings
- if (chatURL)
- {
- const char* pStr = chatURL;
- for (unsigned i = 0; i < strlen(chatURL); i++)
- {
- if ((pStr[0] > 31) && (pStr[0] < 127))
- chatURLStr.append(pStr[0]);
- pStr++;
- }
- }
- if (banner)
- {
- const char* pStr = banner;
- for (unsigned i = 0; i < strlen(banner); i++)
- {
- if ((pStr[0] > 31) && (pStr[0] < 127))
- bannerStr.append(pStr[0]);
- pStr++;
- }
- }
- chatURLStr.trim();
- bannerStr.trim();
- if (!req.getBannerAction_isNull() && req.getBannerAction() && (bannerStr.length() < 1))
- {
- throw MakeStringException(ECLWATCH_MISSING_BANNER_CONTENT, "If a Banner is enabled, the Banner content must be specified.");
- }
- if (!req.getEnableChatURL_isNull() && req.getEnableChatURL() && (!req.getChatURL() || !*req.getChatURL()))
- {
- throw MakeStringException(ECLWATCH_MISSING_CHAT_URL, "If a Chat is enabled, the Chat URL must be specified.");
- }
- m_ChatURL = chatURLStr;
- m_Banner = bannerStr;
- const char* bannerSize = req.getBannerSize();
- if (bannerSize && *bannerSize)
- m_BannerSize.clear().append(bannerSize);
- const char* bannerColor = req.getBannerColor();
- if (bannerColor && *bannerColor)
- m_BannerColor.clear().append(bannerColor);
- const char* bannerScroll = req.getBannerScroll();
- if (bannerScroll && *bannerScroll)
- m_BannerScroll.clear().append(bannerScroll);
- m_BannerAction = 0;
- if(!req.getBannerAction_isNull())
- m_BannerAction = req.getBannerAction();
- m_EnableChatURL = 0;
- if(!req.getEnableChatURL_isNull())
- m_EnableChatURL = req.getEnableChatURL();
- resp.setRedirectUrl("/WsSMC/Activity");
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- bool CWsSMCEx::onNotInCommunityEdition(IEspContext &context, IEspNotInCommunityEditionRequest &req, IEspNotInCommunityEditionResponse &resp)
- {
- return true;
- }
- bool CWsSMCEx::onBrowseResources(IEspContext &context, IEspBrowseResourcesRequest & req, IEspBrowseResourcesResponse & resp)
- {
- try
- {
- if (!context.validateFeatureAccess(FEATURE_URL, SecAccess_Read, false))
- throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Failed to Browse Resources. Permission denied.");
- double version = context.getClientVersion();
- Owned<IEnvironmentFactory> factory = getEnvironmentFactory();
- Owned<IConstEnvironment> constEnv = factory->openEnvironmentByFile();
- //The resource files will be downloaded from the same box of ESP (not dali)
- StringBuffer ipStr;
- IpAddress ipaddr = queryHostIP();
- ipaddr.getIpText(ipStr);
- if (ipStr.length() > 0)
- {
- resp.setNetAddress(ipStr.str());
- Owned<IConstMachineInfo> machine = constEnv->getMachineByAddress(ipStr.str());
- if (machine)
- {
- int os = machine->getOS();
- resp.setOS(os);
- }
- }
- if (m_PortalURL.length() > 0)
- resp.setPortalURL(m_PortalURL.str());
- #ifndef USE_RESOURCE
- if (version > 1.12)
- resp.setUseResource(false);
- #else
- if (version > 1.12)
- resp.setUseResource(true);
- //Now, get a list of resources stored inside the ESP box
- IArrayOf<IEspHPCCResourceRepository> resourceRepositories;
- Owned<IPropertyTree> pEnvRoot = &constEnv->getPTree();
- const char* ossInstall = pEnvRoot->queryProp("EnvSettings/path");
- if (!ossInstall || !*ossInstall)
- {
- WARNLOG("Failed to get EnvSettings/Path in environment settings.");
- return true;
- }
- StringBuffer path;
- path.appendf("%s/componentfiles/files/downloads", ossInstall);
- Owned<IFile> f = createIFile(path.str());
- if(!f->exists() || !f->isDirectory())
- {
- WARNLOG("Invalid resource folder");
- return true;
- }
- Owned<IDirectoryIterator> di = f->directoryFiles(NULL, false, true);
- if(di.get() == NULL)
- {
- WARNLOG("Resource folder is empty.");
- return true;
- }
- ForEach(*di)
- {
- if (!di->isDir())
- continue;
- StringBuffer folder, path0, tmpBuf;
- di->getName(folder);
- if (folder.length() == 0)
- continue;
- path0.appendf("%s/%s/description.xml", path.str(), folder.str());
- Owned<IFile> f0 = createIFile(path0.str());
- if(!f0->exists())
- {
- WARNLOG("Description file not found for %s", folder.str());
- continue;
- }
- OwnedIFileIO rIO = f0->openShared(IFOread,IFSHfull);
- if(!rIO)
- {
- WARNLOG("Failed to open the description file for %s", folder.str());
- continue;
- }
- offset_t fileSize = f0->size();
- tmpBuf.ensureCapacity((unsigned)fileSize);
- tmpBuf.setLength((unsigned)fileSize);
- size32_t nRead = rIO->read(0, (size32_t) fileSize, (char*)tmpBuf.str());
- if (nRead != fileSize)
- {
- WARNLOG("Failed to read the description file for %s", folder.str());
- continue;
- }
- Owned<IPropertyTree> desc = createPTreeFromXMLString(tmpBuf.str());
- if (!desc)
- {
- WARNLOG("Invalid description file for %s", folder.str());
- continue;
- }
- Owned<IPropertyTreeIterator> fileIterator = desc->getElements("file");
- if (!fileIterator->first())
- {
- WARNLOG("Invalid description file for %s", folder.str());
- continue;
- }
- IArrayOf<IEspHPCCResource> resourcs;
- do {
- IPropertyTree &fileItem = fileIterator->query();
- const char* filename = fileItem.queryProp("filename");
- if (!filename || !*filename)
- continue;
- const char* name0 = fileItem.queryProp("name");
- const char* description0 = fileItem.queryProp("description");
- const char* version0 = fileItem.queryProp("version");
- Owned<IEspHPCCResource> onefile = createHPCCResource();
- onefile->setFileName(filename);
- if (name0 && *name0)
- onefile->setName(name0);
- if (description0 && *description0)
- onefile->setDescription(description0);
- if (version0 && *version0)
- onefile->setVersion(version0);
- resourcs.append(*onefile.getLink());
- } while (fileIterator->next());
- if (resourcs.ordinality())
- {
- StringBuffer path1;
- path1.appendf("%s/%s", path.str(), folder.str());
- Owned<IEspHPCCResourceRepository> oneRepository = createHPCCResourceRepository();
- oneRepository->setName(folder.str());
- oneRepository->setPath(path1.str());
- oneRepository->setHPCCResources(resourcs);
- resourceRepositories.append(*oneRepository.getLink());
- }
- }
- if (resourceRepositories.ordinality())
- resp.setHPCCResourceRepositories(resourceRepositories);
- #endif
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return true;
- }
- int CWsSMCSoapBindingEx::onGetForm(IEspContext &context, CHttpRequest* request, CHttpResponse* response, const char *service, const char *method)
- {
- try
- {
- if(stricmp(method,"NotInCommunityEdition")==0)
- {
- StringBuffer page, url, link;
- request->getParameter("EEPortal", url);
- if (url.length() > 0)
- link.appendf("Further information can be found at <a href=\"%s\" target=\"_blank\">%s</a>.", url.str(), url.str());
- page.append(
- "<html>"
- "<head>"
- "<meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\" />"
- "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/default.css\"/>"
- "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/yui/build/fonts/fonts-min.css\" />"
- "<title>Advanced feature in Enterprise Edition</title>"
- "</head>"
- "<body>"
- "<h3 style=\"text-align:centre;\">Advanced feature in the Enterprise Edition</h4>"
- "<p style=\"text-align:centre;\">Support for this feature is coming soon. ");
- if (link.length() > 0)
- page.append(link.str());
- page.append("</p></body>"
- "</html>");
- response->setContent(page.str());
- response->setContentType("text/html");
- response->send();
- return 0;
- }
- else if(stricmp(method,"DisabledInThisVersion")==0)
- {
- StringBuffer page;
- page.append(
- "<html>"
- "<head>"
- "<meta http-equiv=\"Content-Type\" content=\"text/html; charset=utf-8\" />"
- "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/default.css\"/>"
- "<link rel=\"stylesheet\" type=\"text/css\" href=\"/esp/files/yui/build/fonts/fonts-min.css\" />"
- "<title>Disabled Feature in This Version</title>"
- "</head>"
- "<body>"
- "<h3 style=\"text-align:centre;\">Disabled Feature in This Version</h4>"
- "<p style=\"text-align:centre;\">This feature is disabled in this version. ");
- page.append("</p></body>"
- "</html>");
- response->setContent(page.str());
- response->setContentType("text/html");
- response->send();
- return 0;
- }
- }
- catch(IException* e)
- {
- FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
- }
- return onGetForm(context, request, response, service, method);
- }
- inline const char *controlCmdMessage(int cmd)
- {
- switch (cmd)
- {
- case CRoxieControlCmd_ATTACH:
- return "<control:unlockDali/>";
- case CRoxieControlCmd_DETACH:
- return "<control:lockDali/>";
- case CRoxieControlCmd_RELOAD:
- return "<control:reload/>";
- case CRoxieControlCmd_STATE:
- return "<control:state/>";
- default:
- throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Unknown Roxie Control Command.");
- }
- return NULL;
- }
- bool CWsSMCEx::onRoxieControlCmd(IEspContext &context, IEspRoxieControlCmdRequest &req, IEspRoxieControlCmdResponse &resp)
- {
- if (!context.validateFeatureAccess(ROXIE_CONTROL_URL, SecAccess_Full, false))
- throw MakeStringException(ECLWATCH_SMC_ACCESS_DENIED, "Cannot Access Roxie Control. Permission denied.");
- const char *process = req.getProcessCluster();
- if (!process || !*process)
- throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Process cluster not specified.");
- const char *controlReq = controlCmdMessage(req.getCommand());
- SocketEndpointArray addrs;
- getRoxieProcessServers(process, addrs);
- if (!addrs.length())
- throw MakeStringException(ECLWATCH_CANNOT_GET_ENV_INFO, "Process cluster not found.");
- Owned<IPropertyTree> controlResp = sendRoxieControlAllNodes(addrs.item(0), controlReq, true, req.getWait());
- if (!controlResp)
- throw MakeStringException(ECLWATCH_INTERNAL_ERROR, "Failed to get control response from roxie.");
- IArrayOf<IEspRoxieControlEndpointInfo> respEndpoints;
- Owned<IPropertyTreeIterator> roxieEndpoints = controlResp->getElements("Endpoint");
- ForEach(*roxieEndpoints)
- {
- IPropertyTree &roxieEndpoint = roxieEndpoints->query();
- Owned<IEspRoxieControlEndpointInfo> respEndpoint = createRoxieControlEndpointInfo();
- respEndpoint->setAddress(roxieEndpoint.queryProp("@ep"));
- respEndpoint->setStatus(roxieEndpoint.queryProp("Status"));
- if (roxieEndpoint.hasProp("Dali/@connected"))
- respEndpoint->setAttached(roxieEndpoint.getPropBool("Dali/@connected"));
- if (roxieEndpoint.hasProp("State/@hash"))
- respEndpoint->setStateHash(roxieEndpoint.queryProp("State/@hash"));
- respEndpoints.append(*respEndpoint.getClear());
- }
- resp.setEndpoints(respEndpoints);
- return true;
- }
|