ws_fsService.cpp 132 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558255925602561256225632564256525662567256825692570257125722573257425752576257725782579258025812582258325842585258625872588258925902591259225932594259525962597259825992600260126022603260426052606260726082609261026112612261326142615261626172618261926202621262226232624262526262627262826292630263126322633263426352636263726382639264026412642264326442645264626472648264926502651265226532654265526562657265826592660266126622663266426652666266726682669267026712672267326742675267626772678267926802681268226832684268526862687268826892690269126922693269426952696269726982699270027012702270327042705270627072708270927102711271227132714271527162717271827192720272127222723272427252726272727282729273027312732273327342735273627372738273927402741274227432744274527462747274827492750275127522753275427552756275727582759276027612762276327642765276627672768276927702771277227732774277527762777277827792780278127822783278427852786278727882789279027912792279327942795279627972798279928002801280228032804280528062807280828092810281128122813281428152816281728182819282028212822282328242825282628272828282928302831283228332834283528362837283828392840284128422843284428452846284728482849285028512852285328542855285628572858285928602861286228632864286528662867286828692870287128722873287428752876287728782879288028812882288328842885288628872888288928902891289228932894289528962897289828992900290129022903290429052906290729082909291029112912291329142915291629172918291929202921292229232924292529262927292829292930293129322933293429352936293729382939294029412942294329442945294629472948294929502951295229532954295529562957295829592960296129622963296429652966296729682969297029712972297329742975297629772978297929802981298229832984298529862987298829892990299129922993299429952996299729982999300030013002300330043005300630073008300930103011301230133014301530163017301830193020302130223023302430253026302730283029303030313032303330343035303630373038303930403041304230433044304530463047304830493050305130523053305430553056305730583059306030613062306330643065306630673068306930703071307230733074307530763077307830793080308130823083308430853086308730883089309030913092309330943095309630973098309931003101310231033104310531063107310831093110311131123113311431153116311731183119312031213122312331243125312631273128312931303131313231333134313531363137313831393140314131423143314431453146314731483149315031513152315331543155315631573158315931603161316231633164316531663167316831693170317131723173317431753176317731783179318031813182318331843185318631873188318931903191319231933194319531963197319831993200320132023203320432053206320732083209321032113212321332143215321632173218321932203221322232233224322532263227322832293230323132323233323432353236323732383239324032413242324332443245324632473248324932503251325232533254325532563257325832593260326132623263326432653266326732683269327032713272327332743275327632773278327932803281328232833284328532863287328832893290329132923293329432953296329732983299330033013302330333043305330633073308330933103311331233133314331533163317331833193320332133223323332433253326332733283329333033313332333333343335333633373338333933403341334233433344334533463347334833493350335133523353335433553356335733583359336033613362336333643365336633673368336933703371337233733374337533763377337833793380338133823383338433853386338733883389339033913392339333943395339633973398339934003401340234033404340534063407340834093410341134123413341434153416341734183419342034213422342334243425342634273428342934303431343234333434343534363437343834393440344134423443344434453446344734483449345034513452345334543455345634573458345934603461346234633464346534663467346834693470347134723473347434753476347734783479348034813482348334843485348634873488348934903491349234933494349534963497349834993500350135023503350435053506350735083509351035113512351335143515351635173518351935203521352235233524352535263527352835293530353135323533353435353536353735383539354035413542354335443545354635473548354935503551355235533554355535563557355835593560356135623563356435653566356735683569357035713572357335743575357635773578357935803581358235833584358535863587358835893590359135923593359435953596359735983599360036013602360336043605360636073608360936103611361236133614361536163617361836193620362136223623362436253626362736283629363036313632363336343635363636373638363936403641364236433644364536463647364836493650365136523653365436553656365736583659366036613662366336643665366636673668366936703671
  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. #pragma warning (disable : 4129)
  15. #include <math.h>
  16. #include "jsocket.hpp"
  17. #include "dasds.hpp"
  18. #include "dadfs.hpp"
  19. #include "dautils.hpp"
  20. #include "daclient.hpp"
  21. #include "wshelpers.hpp"
  22. #include "dfuwu.hpp"
  23. #include "workunit.hpp"
  24. #include "ws_fsService.hpp"
  25. #ifdef _WIN32
  26. #include "windows.h"
  27. #endif
  28. #include "jlog.hpp"
  29. #include "dalienv.hpp"
  30. #include "dfuutil.hpp"
  31. #include "portlist.h"
  32. #include "sacmd.hpp"
  33. #include "exception_util.hpp"
  34. #include "LogicFileWrapper.hpp"
  35. #include "dameta.hpp"
  36. #include "daqueue.hpp"
  37. #define DFU_WU_URL "DfuWorkunitsAccess"
  38. #define DFU_EX_URL "DfuExceptionsAccess"
  39. #define FILE_SPRAY_URL "FileSprayAccess"
  40. #define FILE_DESPRAY_URL "FileDesprayAccess"
  41. #define WUDETAILS_REFRESH_MINS 1
  42. const unsigned dropZoneFileSearchMaxFiles = 1000;
  43. void SetResp(StringBuffer &resp, IConstDFUWorkUnit * wu, bool array);
  44. int Schedule::run()
  45. {
  46. PROGLOG("DfuWorkunit WUSchedule Thread started.");
  47. while(!stopping)
  48. {
  49. unsigned int waitTimeMillies = 1000*60;
  50. if (!detached)
  51. {
  52. try
  53. {
  54. if (waitTimeMillies == (unsigned)-1)
  55. {
  56. PROGLOG("WS_FS WUSchedule Thread Re-started.");
  57. waitTimeMillies = 1000*60;
  58. }
  59. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  60. Owned<IConstDFUWorkUnitIterator> itr = factory->getWorkUnitsByState(DFUstate_scheduled);
  61. itr->first();
  62. while(itr->isValid())
  63. {
  64. Owned<IConstDFUWorkUnit> wu = itr->get();
  65. CDateTime dt, now;
  66. now.setNow();
  67. try
  68. {
  69. wu->getTimeScheduled(dt);
  70. if (now.compare(dt) > 0)
  71. {
  72. StringAttr wuid(wu->queryId());
  73. wu.clear();
  74. submitDFUWorkUnit(wuid.get());
  75. }
  76. }
  77. catch(IException *e)
  78. {
  79. StringBuffer msg;
  80. IERRLOG("Exception %d:%s in WsWorkunits Schedule::run", e->errorCode(), e->errorMessage(msg).str());
  81. e->Release();
  82. }
  83. itr->next();
  84. }
  85. }
  86. catch(IException *e)
  87. {
  88. StringBuffer msg;
  89. IERRLOG("Exception %d:%s in WS_FS Schedule::run", e->errorCode(), e->errorMessage(msg).str());
  90. e->Release();
  91. }
  92. catch(...)
  93. {
  94. IERRLOG("Unknown exception in WS_FS Schedule::run");
  95. }
  96. }
  97. else
  98. {
  99. OWARNLOG("Detached from DALI, WS_FS schedule interrupted");
  100. waitTimeMillies = (unsigned)-1;
  101. }
  102. semSchedule.wait(waitTimeMillies);
  103. }
  104. return 0;
  105. }
  106. void CFileSprayEx::init(IPropertyTree *cfg, const char *process, const char *service)
  107. {
  108. #ifndef _CONTAINERIZED
  109. StringBuffer xpath;
  110. xpath.clear().appendf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/QueueLabel", process, service);
  111. cfg->getProp(xpath.str(), m_QueueLabel);
  112. StringArray qlist;
  113. getDFUServerQueueNames(qlist, nullptr);
  114. if (qlist.ordinality())
  115. {
  116. if (!m_QueueLabel.length())
  117. m_QueueLabel.append(qlist.item(0));
  118. else
  119. {
  120. bool found = false;
  121. ForEachItemIn(i, qlist)
  122. {
  123. const char* qname = qlist.item(i);
  124. if (qname && strieq(qname, m_QueueLabel.str()))
  125. {
  126. found = true;
  127. break;
  128. }
  129. }
  130. if (!found)
  131. throw MakeStringException(-1, "Invalid DFU Queue Label %s in configuration file", m_QueueLabel.str());
  132. }
  133. }
  134. xpath.setf("Software/EspProcess[@name=\"%s\"]/@PageCacheTimeoutSeconds", process);
  135. if (cfg->hasProp(xpath.str()))
  136. setPageCacheTimeoutMilliSeconds(cfg->getPropInt(xpath.str()));
  137. xpath.setf("Software/EspProcess[@name=\"%s\"]/@MaxPageCacheItems", process);
  138. if (cfg->hasProp(xpath.str()))
  139. setMaxPageCacheItems(cfg->getPropInt(xpath.str()));
  140. xpath.setf("Software/EspProcess[@name=\"%s\"]/EspService[@name=\"%s\"]/MonitorQueueLabel", process, service);
  141. cfg->getProp(xpath.str(), m_MonitorQueueLabel);
  142. directories.set(cfg->queryPropTree("Software/Directories"));
  143. #else
  144. // Using the first queue for now.
  145. // TODO: Re-design support for multiple queues
  146. Owned<IPropertyTreeIterator> dfuQueues = getComponentConfigSP()->getElements("dfuQueues");
  147. ForEach(*dfuQueues)
  148. {
  149. IPropertyTree & dfuQueue = dfuQueues->query();
  150. const char * dfuName = dfuQueue.queryProp("@name");
  151. if (!isEmptyString(dfuName))
  152. {
  153. getDfuQueueName(m_QueueLabel, dfuName);
  154. getDfuMonitorQueueName(m_MonitorQueueLabel, dfuName);
  155. break;
  156. }
  157. }
  158. #endif
  159. DBGLOG("queueLabel=%s", m_QueueLabel.str());
  160. DBGLOG("monitorQueueLabel=%s", m_MonitorQueueLabel.str());
  161. if (!daliClientActive())
  162. {
  163. OERRLOG("No Dali Connection Active.");
  164. throw MakeStringException(-1, "No Dali Connection Active. Please Specify a Dali to connect to in your configuration file");
  165. }
  166. m_sched.start();
  167. }
  168. StringBuffer& CFileSprayEx::getAcceptLanguage(IEspContext& context, StringBuffer& acceptLanguage)
  169. {
  170. context.getAcceptLanguage(acceptLanguage);
  171. if (!acceptLanguage.length())
  172. {
  173. acceptLanguage.set("en");
  174. return acceptLanguage;
  175. }
  176. acceptLanguage.setLength(2);
  177. VStringBuffer languageFile("%ssmc_xslt/nls/%s/hpcc.xml", getCFD(), acceptLanguage.str());
  178. if (!checkFileExists(languageFile.str()))
  179. acceptLanguage.set("en");
  180. return acceptLanguage;
  181. }
  182. void ParsePath(const char * fullPath, StringBuffer &ip, StringBuffer &filePath, StringBuffer &title)
  183. {
  184. ip.clear();
  185. filePath.clear();
  186. title.clear();
  187. if(fullPath == NULL || *fullPath == '\0')
  188. return;
  189. const char* ptr = fullPath;
  190. if(*ptr == '\\' && *(ptr+1) == '\\')
  191. {
  192. ptr += 2;
  193. while(*ptr != '\0' && *ptr != '\\')
  194. ptr++;
  195. ip.append(ptr - fullPath - 2, fullPath + 2);
  196. }
  197. filePath.append(ptr);
  198. ptr = fullPath + strlen(fullPath) - 1;
  199. while(ptr > fullPath && *ptr != '\\')
  200. ptr--;
  201. title.append(ptr + 1);
  202. }
  203. const char * const NODATETIME="1970-01-01T00:00:00Z";
  204. // Assign from a dfuwu workunit structure to an esp request workunit structure.
  205. static void DeepAssign(IEspContext &context, IConstDFUWorkUnit *src, IEspDFUWorkunit &dest)
  206. {
  207. if(src == NULL)
  208. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "'Source DFU workunit' doesn't exist.");
  209. StringBuffer tmp;
  210. double version = context.getClientVersion();
  211. dest.setID(src->queryId());
  212. #ifndef _CONTAINERIZED
  213. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory(true);
  214. Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
  215. Owned<IPropertyTree> root = &constEnv->getPTree();
  216. if (src->getClusterName(tmp.clear()).length()!=0)
  217. {
  218. char *clusterName = (char *)tmp.str();
  219. if (clusterName && *clusterName)
  220. {
  221. StringBuffer clusterNameForDisplay(clusterName);
  222. Owned<IPropertyTreeIterator> clusters= root->getElements("Software/Topology/Cluster");
  223. if (clusters->first())
  224. {
  225. do {
  226. IPropertyTree &cluster = clusters->query();
  227. const char* name = cluster.queryProp("@name");
  228. if (!name || !*name)
  229. continue;
  230. Owned<IPropertyTreeIterator> thorClusters= cluster.getElements(eqThorCluster);
  231. Owned<IPropertyTreeIterator> roxieClusters= cluster.getElements(eqRoxieCluster);
  232. if (thorClusters->first() || roxieClusters->first())
  233. {
  234. if (thorClusters->first())
  235. {
  236. IPropertyTree &thorCluster = thorClusters->query();
  237. const char* process = thorCluster.queryProp("@process");
  238. if (process && *process)
  239. {
  240. if (clusterName && !stricmp(clusterName, process))
  241. {
  242. clusterNameForDisplay.clear().append(name);
  243. break;
  244. }
  245. }
  246. }
  247. if (roxieClusters->first())
  248. {
  249. IPropertyTree &roxieCluster = roxieClusters->query();
  250. const char* process = roxieCluster.queryProp("@process");
  251. if (process && *process)
  252. {
  253. if (clusterName && !stricmp(clusterName, name))
  254. {
  255. clusterNameForDisplay.clear().append(name);
  256. break;
  257. }
  258. }
  259. }
  260. }
  261. } while (clusters->next());
  262. }
  263. dest.setClusterName(clusterNameForDisplay.str());
  264. }
  265. }
  266. #else
  267. src->getClusterName(tmp.clear());
  268. dest.setClusterName(tmp.str());
  269. #endif
  270. if ((version > 1.05) && src->getDFUServerName(tmp.clear()).length())
  271. dest.setDFUServerName(tmp.str());
  272. if (src->getJobName(tmp.clear()).length()!=0)
  273. dest.setJobName(tmp.str());
  274. else
  275. dest.setJobName("");
  276. if (src->getQueue(tmp.clear()).length()!=0)
  277. dest.setQueue(tmp.str());
  278. if (src->getUser(tmp.clear()).length()!=0)
  279. dest.setUser(tmp.str());
  280. dest.setIsProtected(src->isProtected());
  281. dest.setCommand(src->getCommand());
  282. IConstDFUprogress *prog = src->queryProgress();
  283. if (prog != NULL)
  284. {
  285. DFUstate state = prog->getState();
  286. dest.setState(state);
  287. StringBuffer statemsg;
  288. encodeDFUstate(state,statemsg);
  289. dest.setStateMessage(statemsg.str());
  290. CDateTime startAt;
  291. CDateTime stoppAt;
  292. prog->getTimeStarted(startAt);
  293. prog->getTimeStopped(stoppAt);
  294. StringBuffer tmpstr;
  295. startAt.getDateString(tmpstr);
  296. tmpstr.append(" ");
  297. startAt.getTimeString(tmpstr);
  298. dest.setTimeStarted(tmpstr.str());
  299. tmpstr.clear();
  300. stoppAt.getDateString(tmpstr);
  301. tmpstr.append(" ");
  302. stoppAt.getTimeString(tmpstr);
  303. dest.setTimeStopped(tmpstr.str());
  304. StringBuffer prgmsg;
  305. prog->formatProgressMessage(prgmsg);
  306. dest.setProgressMessage(prgmsg.str());
  307. prog->formatSummaryMessage(prgmsg.clear());
  308. dest.setSummaryMessage(prgmsg.str());
  309. unsigned secs = prog->getSecsLeft();
  310. if(secs > 0)
  311. dest.setSecsLeft(secs);
  312. dest.setPercentDone(prog->getPercentDone());
  313. }
  314. IConstDFUoptions *options = src->queryOptions();
  315. if(options)
  316. {
  317. dest.setReplicate(options->getReplicate());
  318. dest.setOverwrite(options->getOverwrite());
  319. }
  320. IConstDFUfileSpec * file = src->querySource();
  321. if (file != NULL)
  322. {
  323. //if (file->getTitle(tmp.clear()).length()!=0)
  324. // dest.setSourceTitle(tmp.str());
  325. StringBuffer lfn;
  326. file->getLogicalName(lfn);
  327. if (lfn.length() != 0)
  328. {
  329. dest.setSourceLogicalName(lfn.str());
  330. }
  331. else
  332. dest.setSourceFormat(file->getFormat());
  333. if (file->getRawDirectory(tmp.clear()).length()!=0)
  334. dest.setSourceDirectory(tmp.str());
  335. SocketEndpoint srcdali;
  336. StringBuffer srcdaliip;
  337. file->getForeignDali(srcdali);
  338. srcdali.getIpText(srcdaliip);
  339. if(srcdaliip.length() > 0 && strcmp(srcdaliip.str(), "0.0.0.0") != 0)
  340. dest.setSourceDali(srcdaliip.str());
  341. StringBuffer diffkeyname;
  342. file->getDiffKey(diffkeyname);
  343. if(diffkeyname.length() > 0)
  344. dest.setSourceDiffKeyName(diffkeyname.str());
  345. StringBuffer socket, dir, title;
  346. unsigned np = file->getNumParts(0); // should handle multiple clusters?
  347. if (lfn.length() == 0) { // no logical name
  348. if (np == 1)
  349. {
  350. Owned<IFileDescriptor> info;
  351. try
  352. {
  353. info.setown(file->getFileDescriptor());
  354. if(info)
  355. {
  356. Owned<INode> node = info->getNode(0);
  357. if (node)
  358. {
  359. node->endpoint().getIpText(socket);
  360. dest.setSourceIP(socket.str());
  361. }
  362. const char *defaultdir = info->queryDefaultDir();
  363. if (defaultdir&&*defaultdir)
  364. addPathSepChar(dir.append(defaultdir));
  365. file->getRawFileMask(dir);
  366. dest.setSourceFilePath(dir.str());
  367. }
  368. }
  369. catch(IException *e)
  370. {
  371. EXCLOG(e,"DeepAssign getFileDescriptor");
  372. e->Release();
  373. }
  374. }
  375. }
  376. if (np)
  377. dest.setSourceNumParts(np);
  378. unsigned rs = file->getRecordSize();
  379. if (rs)
  380. dest.setSourceRecordSize(rs);
  381. StringBuffer rowtag;
  382. file->getRowTag(rowtag);
  383. if(rowtag.length() > 0)
  384. dest.setRowTag(rowtag.str());
  385. if (version >= 1.04 && (file->getFormat() == DFUff_csv))
  386. {
  387. StringBuffer separate, terminate, quote, escape;
  388. bool quotedTerminator;
  389. file->getCsvOptions(separate,terminate,quote, escape, quotedTerminator);
  390. if(separate.length() > 0)
  391. dest.setSourceCsvSeparate(separate.str());
  392. if(terminate.length() > 0)
  393. dest.setSourceCsvTerminate(terminate.str());
  394. if(quote.length() > 0)
  395. dest.setSourceCsvQuote(quote.str());
  396. if((version >= 1.05) && (escape.length() > 0))
  397. dest.setSourceCsvEscape(escape.str());
  398. if(version >=1.10)
  399. dest.setQuotedTerminator(quotedTerminator);
  400. }
  401. }
  402. file = src->queryDestination();
  403. if (file != NULL)
  404. {
  405. StringBuffer lfn;
  406. file->getLogicalName(lfn);
  407. if (lfn.length() != 0)
  408. dest.setDestLogicalName(lfn.str());
  409. else
  410. dest.setDestFormat(file->getFormat());
  411. if (file->getRawDirectory(tmp.clear()).length()!=0)
  412. dest.setDestDirectory(tmp.str());
  413. if (file->getGroupName(0,tmp.clear()).length()!=0) // should handle multiple clusters?
  414. {
  415. char *clusterName = (char *)tmp.str();
  416. if (clusterName)
  417. dest.setDestGroupName(clusterName);
  418. }
  419. if (version >= 1.21)
  420. dest.setPreserveFileParts(file->getWrap());
  421. StringBuffer socket, dir, title;
  422. unsigned np = file->getNumParts(0); // should handle multiple clusters?
  423. if (lfn.length() == 0) { // no logical name
  424. if (np == 1)
  425. {
  426. Owned<IFileDescriptor> info;
  427. try
  428. {
  429. info.setown(file->getFileDescriptor());
  430. if(info)
  431. {
  432. Owned<INode> node = info->getNode(0);
  433. if (node)
  434. {
  435. node->endpoint().getIpText(socket);
  436. dest.setDestIP(socket.str());
  437. }
  438. const char *defaultdir = info->queryDefaultDir();
  439. if (defaultdir&&*defaultdir)
  440. addPathSepChar(dir.append(defaultdir));
  441. file->getRawFileMask(dir);
  442. dest.setDestFilePath(dir.str());
  443. }
  444. }
  445. catch(IException *e)
  446. {
  447. EXCLOG(e,"DeepAssign getFileDescriptor dest");
  448. e->Release();
  449. }
  450. }
  451. }
  452. if (np)
  453. dest.setDestNumParts(np);
  454. unsigned rs = file->getRecordSize();
  455. if (rs)
  456. dest.setDestRecordSize(rs);
  457. dest.setCompress(file->isCompressed());
  458. }
  459. // monitor stuff
  460. IConstDFUmonitor *monitor = src->queryMonitor();
  461. if (monitor) {
  462. monitor->getEventName(tmp.clear());
  463. if (tmp.length())
  464. dest.setMonitorEventName(tmp.str());
  465. bool sub = monitor->getSub();
  466. dest.setMonitorSub(sub);
  467. unsigned sl = monitor->getShotLimit();
  468. if (sl)
  469. dest.setMonitorShotLimit(sl);
  470. }
  471. }
  472. bool CFileSprayEx::ParseLogicalPath(const char * pLogicalPath, const char* groupName, const char* cluster,
  473. StringBuffer &folder, StringBuffer &title, StringBuffer &defaultFolder, StringBuffer &defaultReplicateFolder)
  474. {
  475. if(!pLogicalPath || !*pLogicalPath)
  476. return false;
  477. folder.clear();
  478. title.clear();
  479. defaultFolder.clear();
  480. defaultReplicateFolder.clear();
  481. DFD_OS os = DFD_OSdefault;
  482. if(groupName != NULL && *groupName != '\0')
  483. {
  484. #ifdef _CONTAINERIZED
  485. Owned<IPropertyTree> plane = getStoragePlane(groupName);
  486. if (plane)
  487. defaultFolder.append(plane->queryProp("@prefix"));
  488. #else
  489. StringBuffer basedir;
  490. GroupType groupType;
  491. Owned<IGroup> group = queryNamedGroupStore().lookup(groupName, basedir, groupType);
  492. if (group) {
  493. switch (queryOS(group->queryNode(0).endpoint())) {
  494. case MachineOsW2K:
  495. os = DFD_OSwindows; break;
  496. case MachineOsSolaris:
  497. case MachineOsLinux:
  498. os = DFD_OSunix; break;
  499. }
  500. if (directories.get())
  501. {
  502. switch (groupType)
  503. {
  504. case grp_roxie:
  505. getConfigurationDirectory(directories, "data", "roxie", cluster, defaultFolder);
  506. getConfigurationDirectory(directories, "data2", "roxie", cluster, defaultReplicateFolder);
  507. // MORE - should extend to systems with higher redundancy
  508. break;
  509. case grp_hthor:
  510. getConfigurationDirectory(directories, "data", "eclagent", cluster, defaultFolder);
  511. break;
  512. case grp_thor:
  513. default:
  514. getConfigurationDirectory(directories, "data", "thor", cluster, defaultFolder);
  515. getConfigurationDirectory(directories, "mirror", "thor", cluster, defaultReplicateFolder);
  516. }
  517. }
  518. }
  519. else
  520. {
  521. // Error here?
  522. }
  523. #endif
  524. }
  525. getLFNDirectoryUsingBaseDir(folder, pLogicalPath, defaultFolder.str());
  526. const char *n = pLogicalPath;
  527. const char* p;
  528. do {
  529. p = strstr(n,"::");
  530. if(p)
  531. n = p+2;
  532. } while(p);
  533. title.append(n);
  534. return true;
  535. }
  536. bool CFileSprayEx::ParseLogicalPath(const char * pLogicalPath, StringBuffer &title)
  537. {
  538. if(!pLogicalPath || !*pLogicalPath)
  539. return false;
  540. title.clear();
  541. const char *n = pLogicalPath;
  542. const char* p;
  543. do {
  544. p = strstr(n,"::");
  545. if(p)
  546. n = p+2;
  547. } while(p);
  548. title.append(n);
  549. return true;
  550. }
  551. void setRoxieClusterPartDiskMapping(const char *clusterName, const char *defaultFolder, const char *defaultReplicateFolder,
  552. bool supercopy, IDFUfileSpec *wuFSpecDest, IDFUoptions *wuOptions)
  553. {
  554. ClusterPartDiskMapSpec spec;
  555. spec.setDefaultBaseDir(defaultFolder);
  556. #ifndef _CONTAINERIZED
  557. //In containerized mode, there is no need to replicate files to the local disks of the roxie cluster.
  558. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory(true);
  559. Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
  560. VStringBuffer dirxpath("Software/RoxieCluster[@name=\"%s\"]",clusterName);
  561. Owned<IPropertyTree> pEnvRoot = &constEnv->getPTree();
  562. Owned<IPropertyTreeIterator> processes = pEnvRoot->getElements(dirxpath);
  563. if (!processes->first())
  564. {
  565. DBGLOG("Failed to get RoxieCluster settings");
  566. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_INFO, "Failed to get RoxieCluster settings. The workunit will not be created.");
  567. }
  568. IPropertyTree &process = processes->query();
  569. const char *slaveConfig = process.queryProp("@slaveConfig");
  570. if (!slaveConfig || !*slaveConfig)
  571. {
  572. DBGLOG("Failed to get RoxieCluster settings");
  573. throw MakeStringException(ECLWATCH_INVALID_CLUSTER_INFO, "Failed to get RoxieCluster settings. The workunit will not be created.");
  574. }
  575. bool replicate = false;
  576. unsigned redundancy = 0; // Number of "spare" copies of the data
  577. unsigned channelsPerNode = 1; // Overloaded and cyclic modes
  578. int replicateOffset = 1; // Used In cyclic mode only
  579. unsigned numDataCopies = process.getPropInt("@numDataCopies", 1);
  580. if (strieq(slaveConfig, "overloaded"))
  581. {
  582. channelsPerNode = process.getPropInt("@channelsPernode", 1);
  583. spec.setDefaultReplicateDir(defaultReplicateFolder);
  584. }
  585. else if (strieq(slaveConfig, "full redundancy"))
  586. {
  587. redundancy = numDataCopies-1;
  588. replicateOffset = 0;
  589. replicate = true;
  590. }
  591. else if (strieq(slaveConfig, "cyclic redundancy"))
  592. {
  593. redundancy = numDataCopies-1;
  594. channelsPerNode = numDataCopies;
  595. replicateOffset = process.getPropInt("@cyclicOffset", 1);
  596. spec.setDefaultReplicateDir(defaultReplicateFolder);
  597. replicate = true;
  598. }
  599. spec.setRoxie (redundancy, channelsPerNode, replicateOffset);
  600. wuOptions->setReplicate(replicate);
  601. #endif
  602. if (!supercopy)
  603. spec.setRepeatedCopies(CPDMSRP_lastRepeated,false);
  604. wuFSpecDest->setClusterPartDiskMapSpec(clusterName,spec);
  605. }
  606. StringBuffer& constructFileMask(const char* filename, StringBuffer& filemask)
  607. {
  608. filemask.clear().append(filename).toLowerCase().append("._$P$_of_$N$");
  609. return filemask;
  610. }
  611. bool CFileSprayEx::onDFUWUSearch(IEspContext &context, IEspDFUWUSearchRequest & req, IEspDFUWUSearchResponse & resp)
  612. {
  613. try
  614. {
  615. #ifndef _CONTAINERIZED
  616. //The code below should be only for legacy ECLWatch.
  617. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Read, ECLWATCH_DFU_WU_ACCESS_DENIED, "Access to DFU workunit is denied.");
  618. Owned<IEnvironmentFactory> factory = getEnvironmentFactory(true);
  619. Owned<IConstEnvironment> environment = factory->openEnvironment();
  620. Owned<IPropertyTree> root = &environment->getPTree();
  621. StringArray dfuclusters;
  622. Owned<IPropertyTreeIterator> clusterIterator = root->getElements("Software/Topology/Cluster");
  623. if (clusterIterator->first())
  624. {
  625. do {
  626. IPropertyTree &cluster = clusterIterator->query();
  627. const char *clusterName = cluster.queryProp("@name");
  628. if (!clusterName || !*clusterName)
  629. continue;
  630. dfuclusters.append(clusterName);
  631. } while (clusterIterator->next());
  632. }
  633. resp.setClusterNames(dfuclusters);
  634. #endif
  635. }
  636. catch(IException* e)
  637. {
  638. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  639. }
  640. return true;
  641. }
  642. int readFromCommaSeparatedString(const char *commaSeparatedString, StringBuffer* output)
  643. {
  644. int numOfItems = 0;
  645. if (commaSeparatedString && *commaSeparatedString)
  646. {
  647. char *pStr = (char *) commaSeparatedString;
  648. while (pStr)
  649. {
  650. char item[1024];
  651. bool bFoundComma = false;
  652. int len = strlen(pStr);
  653. for (int i = 0; i < len; i++)
  654. {
  655. char *pStr1 = pStr + i;
  656. if (pStr1[0] != ',')
  657. continue;
  658. strncpy(item, pStr, pStr1 - pStr);
  659. item[pStr1 - pStr] = 0;
  660. bFoundComma = true;
  661. if (i < len - 1)
  662. pStr = pStr1 + 1;
  663. else
  664. pStr = NULL;
  665. break;
  666. }
  667. if (!bFoundComma && len > 0)
  668. {
  669. strcpy(item, pStr);
  670. pStr = NULL;
  671. }
  672. output[numOfItems] = item;
  673. numOfItems++;
  674. }
  675. }
  676. return numOfItems;
  677. }
  678. bool CFileSprayEx::GetArchivedDFUWorkunits(IEspContext &context, IEspGetDFUWorkunits &req, IEspGetDFUWorkunitsResponse &resp)
  679. {
  680. StringBuffer user;
  681. context.getUserID(user);
  682. SocketEndpoint ep;
  683. getSashaServiceEP(ep, "dfuwu-archiver", true);
  684. Owned<INode> sashaserver = createINode(ep);
  685. __int64 count=req.getPageSize();
  686. if(count < 1)
  687. count=100;
  688. __int64 begin=req.getPageStartFrom();
  689. if (begin < 0)
  690. begin = 0;
  691. Owned<ISashaCommand> cmd = createSashaCommand();
  692. cmd->setAction(SCA_LIST);
  693. cmd->setOnline(false);
  694. cmd->setArchived(true);
  695. cmd->setDFU(true);
  696. cmd->setLimit((int) count+1);
  697. cmd->setStart((int)begin);
  698. if(req.getCluster() && *req.getCluster())
  699. cmd->setCluster(req.getCluster());
  700. if(req.getOwner() && *req.getOwner())
  701. cmd->setOwner(req.getOwner());
  702. if(req.getJobname() && *req.getJobname())
  703. cmd->setJobName(req.getJobname());
  704. if(req.getStateReq() && *req.getStateReq())
  705. cmd->setState(req.getStateReq());
  706. cmd->setOutputFormat("owner,jobname,cluster,state,command");//date range/owner/jobname/state*/
  707. if (!cmd->send(sashaserver))
  708. {
  709. StringBuffer url;
  710. throw MakeStringException(ECLWATCH_CANNOT_CONNECT_ARCHIVE_SERVER,
  711. "Sasha (%s) took too long to respond from: Get archived workUnits.",
  712. ep.getUrlStr(url).str());
  713. }
  714. IArrayOf<IEspDFUWorkunit> results;
  715. __int64 actualCount = cmd->numIds();
  716. StringBuffer s;
  717. for (unsigned j=0;j<actualCount;j++)
  718. {
  719. const char *wuidStr = cmd->queryId(j);
  720. if (!wuidStr)
  721. continue;
  722. StringBuffer strArray[6];
  723. readFromCommaSeparatedString(wuidStr, strArray);
  724. //skip any workunits without access
  725. Owned<IEspDFUWorkunit> resultWU = createDFUWorkunit("", "");
  726. resultWU->setArchived(true);
  727. if (strArray[0].length() > 0)
  728. resultWU->setID(strArray[0].str());
  729. if (strArray[1].length() > 0)
  730. resultWU->setUser(strArray[1].str());
  731. if (strArray[2].length() > 0)
  732. resultWU->setJobName(strArray[2].str());
  733. if (strArray[3].length() > 0)
  734. resultWU->setClusterName(strArray[3].str());
  735. if (strArray[4].length() > 0)
  736. resultWU->setStateMessage(strArray[4].str());
  737. if (strArray[5].length() > 0)
  738. resultWU->setCommand(atoi(strArray[5].str()));
  739. results.append(*resultWU.getLink());
  740. }
  741. resp.setPageStartFrom(begin+1);
  742. resp.setNextPage(-1);
  743. if(count < actualCount)
  744. {
  745. if (results.length() > count)
  746. {
  747. results.pop();
  748. }
  749. resp.setNextPage(begin + count);
  750. resp.setPageEndAt(begin + count);
  751. }
  752. else
  753. {
  754. resp.setPageEndAt(begin + actualCount);
  755. }
  756. if(begin > 0)
  757. {
  758. resp.setFirst(false);
  759. if (begin - count > 0)
  760. resp.setPrevPage(begin - count);
  761. else
  762. resp.setPrevPage(0);
  763. }
  764. resp.setPageSize(count);
  765. resp.setResults(results);
  766. StringBuffer basicQuery;
  767. if (req.getStateReq() && *req.getStateReq())
  768. {
  769. resp.setStateReq(req.getStateReq());
  770. addToQueryString(basicQuery, "StateReq", req.getStateReq());
  771. }
  772. if (req.getCluster() && *req.getCluster())
  773. {
  774. resp.setCluster(req.getCluster());
  775. addToQueryString(basicQuery, "Cluster", req.getCluster());
  776. }
  777. if (req.getOwner() && *req.getOwner())
  778. {
  779. resp.setOwner(req.getOwner());
  780. addToQueryString(basicQuery, "Owner", req.getOwner());
  781. }
  782. if (req.getType() && *req.getType())
  783. {
  784. resp.setType(req.getType());
  785. addToQueryString(basicQuery, "Type", req.getType());
  786. }
  787. resp.setFilters(basicQuery.str());
  788. resp.setBasicQuery(basicQuery.str());
  789. return true;
  790. }
  791. bool CFileSprayEx::getOneDFUWorkunit(IEspContext& context, const char* wuid, IEspGetDFUWorkunitsResponse& resp)
  792. {
  793. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  794. Owned<IConstDFUWorkUnit> wu = factory->openWorkUnit(wuid, false);
  795. if (!wu)
  796. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Dfu workunit %s not found.", wuid);
  797. Owned<IEspDFUWorkunit> resultWU = createDFUWorkunit();
  798. resultWU->setID(wuid);
  799. resultWU->setCommand(wu->getCommand());
  800. resultWU->setIsProtected(wu->isProtected());
  801. StringBuffer jobname, user, cluster;
  802. resultWU->setJobName(wu->getJobName(jobname).str());
  803. resultWU->setUser(wu->getUser(user).str());
  804. const char* clusterName = wu->getClusterName(cluster).str();
  805. if (clusterName && *clusterName)
  806. {
  807. #ifdef _CONTAINERIZED
  808. Owned<IStringIterator> targets = getContainerTargetClusters(nullptr, clusterName);
  809. #else
  810. Owned<IStringIterator> targets = getTargetClusters(nullptr, clusterName);
  811. #endif
  812. if (!targets->first())
  813. resultWU->setClusterName(clusterName);
  814. else
  815. {
  816. SCMStringBuffer targetCluster;
  817. targets->str(targetCluster);
  818. resultWU->setClusterName(targetCluster.str());
  819. }
  820. }
  821. IConstDFUprogress* prog = wu->queryProgress();
  822. if (prog)
  823. {
  824. StringBuffer statemsg;
  825. DFUstate state = prog->getState();
  826. encodeDFUstate(state, statemsg);
  827. resultWU->setState(state);
  828. resultWU->setStateMessage(statemsg.str());
  829. resultWU->setPercentDone(prog->getPercentDone());
  830. }
  831. IArrayOf<IEspDFUWorkunit> result;
  832. result.append(*resultWU.getClear());
  833. resp.setResults(result);
  834. return true;
  835. }
  836. bool CFileSprayEx::onGetDFUWorkunits(IEspContext &context, IEspGetDFUWorkunits &req, IEspGetDFUWorkunitsResponse &resp)
  837. {
  838. try
  839. {
  840. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Read, ECLWATCH_DFU_WU_ACCESS_DENIED, "Access to DFU workunit is denied.");
  841. StringBuffer wuidStr(req.getWuid());
  842. const char* wuid = wuidStr.trim().str();
  843. if (wuid && *wuid && looksLikeAWuid(wuid, 'D'))
  844. return getOneDFUWorkunit(context, wuid, resp);
  845. double version = context.getClientVersion();
  846. if (version > 1.02)
  847. {
  848. const char *type = req.getType();
  849. if (type && *type && !stricmp(type, "archived workunits"))
  850. {
  851. return GetArchivedDFUWorkunits(context, req, resp);
  852. }
  853. }
  854. StringBuffer clusterReq;
  855. const char *clusterName = req.getCluster();
  856. if(clusterName && *clusterName)
  857. {
  858. clusterReq.append(clusterName);
  859. }
  860. #ifndef _CONTAINERIZED
  861. //The code is used to find out a target cluster from a process cluster.
  862. //It is not needed for containerized environment.
  863. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory(true);
  864. Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
  865. Owned<IPropertyTree> root = &constEnv->getPTree();
  866. StringArray targetClusters, clusterProcesses;
  867. Owned<IPropertyTreeIterator> clusters= root->getElements("Software/Topology/Cluster");
  868. if (clusters->first())
  869. {
  870. do {
  871. IPropertyTree &cluster = clusters->query();
  872. const char* name = cluster.queryProp("@name");
  873. if (!name || !*name)
  874. continue;
  875. Owned<IPropertyTreeIterator> thorClusters= cluster.getElements(eqThorCluster);
  876. Owned<IPropertyTreeIterator> roxieClusters= cluster.getElements(eqRoxieCluster);
  877. if (thorClusters->first() || roxieClusters->first())
  878. {
  879. bool bFound = false;
  880. if (thorClusters->first())
  881. {
  882. IPropertyTree &thorCluster = thorClusters->query();
  883. const char* process = thorCluster.queryProp("@process");
  884. if (process && *process)
  885. {
  886. targetClusters.append(name);
  887. clusterProcesses.append(process);
  888. if (clusterName && !stricmp(clusterName, name))
  889. {
  890. clusterReq.clear().append(process);
  891. }
  892. }
  893. }
  894. if (!bFound && roxieClusters->first())
  895. {
  896. IPropertyTree &roxieCluster = roxieClusters->query();
  897. const char* process = roxieCluster.queryProp("@process");
  898. if (process && *process)
  899. {
  900. targetClusters.append(name);
  901. clusterProcesses.append(process);
  902. if (clusterName && !stricmp(clusterName, name))
  903. {
  904. clusterReq.clear().append(process);
  905. }
  906. }
  907. }
  908. }
  909. } while (clusters->next());
  910. }
  911. #endif
  912. __int64 pagesize = req.getPageSize();
  913. __int64 pagefrom = req.getPageStartFrom();
  914. __int64 displayFrom = 0;
  915. if (pagesize < 1)
  916. {
  917. pagesize = 100;
  918. }
  919. if (pagefrom > 0)
  920. {
  921. displayFrom = pagefrom;
  922. }
  923. DFUsortfield sortorder[2] = {DFUsf_wuid, DFUsf_term};
  924. sortorder[0] = (DFUsortfield) (DFUsf_wuid + DFUsf_reverse);
  925. if(req.getSortby() && *req.getSortby())
  926. {
  927. const char *sortby = req.getSortby();
  928. if (!stricmp(sortby, "Owner"))
  929. sortorder[0] = DFUsf_user;
  930. else if (!stricmp(sortby, "JobName"))
  931. sortorder[0] = DFUsf_job;
  932. else if (!stricmp(sortby, "Cluster"))
  933. sortorder[0] = DFUsf_cluster;
  934. else if (!stricmp(sortby, "State"))
  935. sortorder[0] = DFUsf_state;
  936. else if (!stricmp(sortby, "Type"))
  937. sortorder[0] = DFUsf_command;
  938. else if (!stricmp(sortby, "Protected"))
  939. sortorder[0] = DFUsf_protected;
  940. else if (!stricmp(sortby, "PCTDone"))
  941. sortorder[0] = (DFUsortfield) (DFUsf_pcdone | DFUsf_numeric);
  942. else
  943. sortorder[0] = DFUsf_wuid;
  944. bool descending = req.getDescending();
  945. if (descending)
  946. sortorder[0] = (DFUsortfield) (sortorder[0] | DFUsf_reverse);
  947. }
  948. DFUsortfield filters[10];
  949. unsigned short filterCount = 0;
  950. MemoryBuffer filterbuf;
  951. if(req.getStateReq() && *req.getStateReq())
  952. {
  953. filters[filterCount] = DFUsf_state;
  954. filterCount++;
  955. if (stricmp(req.getStateReq(), "unknown") != 0)
  956. filterbuf.append(req.getStateReq());
  957. else
  958. filterbuf.append("");
  959. }
  960. if(wuid && *wuid)
  961. {
  962. filters[filterCount] = DFUsf_wildwuid;
  963. filterCount++;
  964. filterbuf.append(wuid);
  965. }
  966. if(clusterName && *clusterName)
  967. {
  968. filters[filterCount] = DFUsf_cluster;
  969. filterCount++;
  970. filterbuf.append(clusterReq.str());
  971. }
  972. if(req.getOwner() && *req.getOwner())
  973. {
  974. filters[filterCount] = DFUsortfield (DFUsf_user | DFUsf_nocase);
  975. filterCount++;
  976. filterbuf.append(req.getOwner());
  977. }
  978. if(req.getJobname() && *req.getJobname())
  979. {
  980. filters[filterCount] = DFUsortfield (DFUsf_job | DFUsf_nocase);
  981. filterCount++;
  982. filterbuf.append(req.getJobname());
  983. }
  984. filters[filterCount] = DFUsf_term;
  985. __int64 cacheHint = req.getCacheHint();
  986. if (cacheHint < 0) //Not set yet
  987. cacheHint = 0;
  988. IArrayOf<IEspDFUWorkunit> result;
  989. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  990. unsigned numWUs;
  991. PROGLOG("GetDFUWorkunits: getWorkUnitsSorted");
  992. Owned<IConstDFUWorkUnitIterator> itr = factory->getWorkUnitsSorted(sortorder, filters, filterbuf.bufferBase(), (int) displayFrom, (int) pagesize+1, req.getOwner(), &cacheHint, &numWUs);
  993. if (version >= 1.07)
  994. resp.setCacheHint(cacheHint);
  995. PROGLOG("GetDFUWorkunits: getWorkUnitsSorted done");
  996. //unsigned actualCount = 0;
  997. itr->first();
  998. while(itr->isValid())
  999. {
  1000. Owned<IConstDFUWorkUnit> wu = itr->get();
  1001. //actualCount++;
  1002. Owned<IEspDFUWorkunit> resultWU = createDFUWorkunit("", "");
  1003. resultWU->setID(wu->queryId());
  1004. StringBuffer jobname, user, cluster;
  1005. resultWU->setJobName(wu->getJobName(jobname).str());
  1006. resultWU->setCommand(wu->getCommand());
  1007. resultWU->setUser(wu->getUser(user).str());
  1008. const char* clusterName = wu->getClusterName(cluster).str();
  1009. if (!isEmptyString(clusterName))
  1010. {
  1011. #ifdef _CONTAINERIZED
  1012. resultWU->setClusterName(clusterName);
  1013. #else
  1014. //For bare metal environment, the wu->getClusterName() may return a process
  1015. //cluster. It has to be converted to a target cluster.
  1016. StringBuffer clusterForDisplay(clusterName);
  1017. if (clusterProcesses.ordinality())
  1018. {
  1019. for (unsigned i = 0; i < clusterProcesses.length(); i++)
  1020. {
  1021. const char* clusterProcessName = clusterProcesses.item(i);
  1022. if (!stricmp(clusterProcessName, clusterName))
  1023. {
  1024. clusterForDisplay.clear().append(targetClusters.item(i));
  1025. break;
  1026. }
  1027. }
  1028. }
  1029. resultWU->setClusterName(clusterForDisplay.str());
  1030. #endif
  1031. }
  1032. resultWU->setIsProtected(wu->isProtected());
  1033. IConstDFUprogress *prog = wu->queryProgress();
  1034. if (prog != NULL)
  1035. {
  1036. DFUstate state = prog->getState();
  1037. resultWU->setState(state);
  1038. StringBuffer statemsg;
  1039. encodeDFUstate(state,statemsg);
  1040. resultWU->setStateMessage(statemsg.str());
  1041. resultWU->setPercentDone(prog->getPercentDone());
  1042. }
  1043. result.append(*resultWU.getLink());
  1044. itr->next();
  1045. }
  1046. if (result.length() > pagesize)
  1047. result.pop();
  1048. resp.setPageSize(pagesize);
  1049. resp.setNumWUs(numWUs);
  1050. resp.setPageStartFrom(displayFrom + 1);
  1051. if(displayFrom + pagesize < numWUs)
  1052. {
  1053. resp.setNextPage(displayFrom + pagesize);
  1054. resp.setPageEndAt(pagefrom + pagesize);
  1055. __int64 last = displayFrom + pagesize;
  1056. while (last + pagesize < numWUs)
  1057. {
  1058. last += pagesize;
  1059. }
  1060. resp.setLastPage(last);
  1061. }
  1062. else
  1063. {
  1064. resp.setNextPage(-1);
  1065. resp.setPageEndAt(numWUs);
  1066. }
  1067. if(displayFrom > 0)
  1068. {
  1069. resp.setFirst(false);
  1070. if (displayFrom - pagesize > 0)
  1071. resp.setPrevPage(displayFrom - pagesize);
  1072. else
  1073. resp.setPrevPage(0);
  1074. }
  1075. StringBuffer basicQuery;
  1076. if (req.getStateReq() && *req.getStateReq())
  1077. {
  1078. resp.setStateReq(req.getStateReq());
  1079. addToQueryString(basicQuery, "StateReq", req.getStateReq());
  1080. }
  1081. if (req.getCluster() && *req.getCluster())
  1082. {
  1083. resp.setCluster(req.getCluster());
  1084. addToQueryString(basicQuery, "Cluster", req.getCluster());
  1085. }
  1086. if (req.getOwner() && *req.getOwner())
  1087. {
  1088. resp.setOwner(req.getOwner());
  1089. addToQueryString(basicQuery, "Owner", req.getOwner());
  1090. }
  1091. resp.setFilters(basicQuery.str());
  1092. if (req.getSortby() && *req.getSortby())
  1093. {
  1094. resp.setSortby(req.getSortby());
  1095. if (req.getDescending())
  1096. resp.setDescending(req.getDescending());
  1097. StringBuffer strbuf(req.getSortby());
  1098. strbuf.append("=");
  1099. String str1(strbuf.str());
  1100. String str(basicQuery.str());
  1101. if (str.indexOf(str1) < 0)
  1102. {
  1103. addToQueryString(basicQuery, "Sortby", req.getSortby());
  1104. if (req.getDescending())
  1105. addToQueryString(basicQuery, "Descending", "1");
  1106. }
  1107. }
  1108. resp.setBasicQuery(basicQuery.str());
  1109. resp.setResults(result);
  1110. }
  1111. catch(IException* e)
  1112. {
  1113. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1114. }
  1115. return true;
  1116. }
  1117. void CFileSprayEx::addToQueryString(StringBuffer &queryString, const char *name, const char *value)
  1118. {
  1119. if (queryString.length() > 0)
  1120. {
  1121. queryString.append("&amp;");
  1122. }
  1123. queryString.append(name);
  1124. queryString.append("=");
  1125. queryString.append(value);
  1126. }
  1127. void CFileSprayEx::getInfoFromSasha(IEspContext &context, const char *sashaServer, const char* wuid, IEspDFUWorkunit *info)
  1128. {
  1129. Owned<ISashaCommand> cmd = createSashaCommand();
  1130. cmd->addId(wuid);
  1131. cmd->setAction(SCA_GET);
  1132. cmd->setArchived(true);
  1133. cmd->setDFU(true);
  1134. SocketEndpoint ep(sashaServer);
  1135. Owned<INode> node = createINode(ep);
  1136. if (!cmd->send(node,1*60*1000))
  1137. {
  1138. StringBuffer url;
  1139. throw MakeStringException(ECLWATCH_CANNOT_CONNECT_ARCHIVE_SERVER,
  1140. "Sasha (%s) took too long to respond from: Get information for %s.",
  1141. ep.getUrlStr(url).str(), wuid);
  1142. }
  1143. if (cmd->numIds()==0)
  1144. {
  1145. DBGLOG("Could not read archived %s",wuid);
  1146. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT,"Cannot read workunit %s.",wuid);
  1147. }
  1148. unsigned num = cmd->numResults();
  1149. if (num < 1)
  1150. return;
  1151. StringBuffer res;
  1152. cmd->getResult(0,res);
  1153. if(res.length() < 1)
  1154. return;
  1155. Owned<IPropertyTree> wu = createPTreeFromXMLString(res.str());
  1156. if (!wu)
  1157. return;
  1158. const char * command = wu->queryProp("@command");
  1159. const char * submitID = wu->queryProp("@submitID");
  1160. const char * cluster = wu->queryProp("@clusterName");
  1161. const char * queue = wu->queryProp("@queue");
  1162. const char * jobName = wu->queryProp("@jobName");
  1163. const char * protectedWU = wu->queryProp("@protected");
  1164. info->setID(wuid);
  1165. info->setArchived(true);
  1166. if (command && *command)
  1167. info->setCommandMessage(command);
  1168. if (cluster && *cluster)
  1169. info->setClusterName(cluster);
  1170. if (submitID && *submitID)
  1171. info->setUser(submitID);
  1172. if (queue && *queue)
  1173. info->setQueue(queue);
  1174. if (jobName && *jobName)
  1175. info->setJobName(jobName);
  1176. if (protectedWU && stricmp(protectedWU, "0"))
  1177. info->setIsProtected(true);
  1178. else
  1179. info->setIsProtected(false);
  1180. IPropertyTree *source = wu->queryPropTree("Source");
  1181. if(source)
  1182. {
  1183. const char * directory = source->queryProp("@directory");
  1184. const char * name = source->queryProp("@name");
  1185. if (directory && *directory)
  1186. info->setSourceDirectory(directory);
  1187. if (name && *name)
  1188. info->setSourceLogicalName(name);
  1189. }
  1190. IPropertyTree *dest = wu->queryPropTree("Destination");
  1191. if(dest)
  1192. {
  1193. const char * directory = dest->queryProp("@directory");
  1194. int numParts = dest->getPropInt("@numparts", -1);
  1195. if (directory && *directory)
  1196. info->setDestDirectory(directory);
  1197. if (numParts > 0)
  1198. info->setDestNumParts(numParts);
  1199. }
  1200. IPropertyTree *progress = wu->queryPropTree("Progress");
  1201. if(progress)
  1202. {
  1203. const char * state = progress->queryProp("@state");
  1204. const char * timeStarted = progress->queryProp("@timestarted");
  1205. const char * timeStopped = progress->queryProp("@timestopped");
  1206. if (state && *state)
  1207. info->setStateMessage(state);
  1208. if (timeStarted && *timeStarted)
  1209. {
  1210. StringBuffer startStr(timeStarted);
  1211. startStr.replace('T', ' ');
  1212. info->setTimeStarted(startStr.str());
  1213. }
  1214. if (timeStopped && *timeStopped)
  1215. {
  1216. StringBuffer stopStr(timeStopped);
  1217. stopStr.replace('T', ' ');
  1218. info->setTimeStopped(stopStr.str());
  1219. }
  1220. }
  1221. return;
  1222. }
  1223. bool CFileSprayEx::getArchivedWUInfo(IEspContext &context, IEspGetDFUWorkunit &req, IEspGetDFUWorkunitResponse &resp)
  1224. {
  1225. const char *wuid = req.getWuid();
  1226. if (wuid && *wuid)
  1227. {
  1228. StringBuffer serviceEndpoint;
  1229. getSashaService(serviceEndpoint, "sasha-dfuwu-archiver", true);
  1230. getInfoFromSasha(context, serviceEndpoint, wuid, &resp.updateResult());
  1231. return true;
  1232. }
  1233. return false;
  1234. }
  1235. bool CFileSprayEx::onGetDFUWorkunit(IEspContext &context, IEspGetDFUWorkunit &req, IEspGetDFUWorkunitResponse &resp)
  1236. {
  1237. try
  1238. {
  1239. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Read, ECLWATCH_DFU_WU_ACCESS_DENIED, "Access to DFU workunit is denied.");
  1240. const char* wuid = req.getWuid();
  1241. if (!wuid || !*wuid)
  1242. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Dfu workunit ID not specified.");
  1243. bool found = false;
  1244. double version = context.getClientVersion();
  1245. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1246. Owned<IConstDFUWorkUnit> wu = factory->openWorkUnit(wuid, false);
  1247. if(wu)
  1248. {
  1249. IEspDFUWorkunit &result = resp.updateResult();
  1250. PROGLOG("GetDFUWorkunit: %s", wuid);
  1251. DeepAssign(context, wu, result);
  1252. int n = resp.getResult().getState();
  1253. if (n == DFUstate_scheduled || n == DFUstate_queued || n == DFUstate_started)
  1254. {
  1255. resp.setAutoRefresh(WUDETAILS_REFRESH_MINS);
  1256. }
  1257. found = true;
  1258. }
  1259. else if ((version > 1.02) && getArchivedWUInfo(context, req, resp))
  1260. {
  1261. found = true;
  1262. }
  1263. if (!found)
  1264. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Dfu workunit %s not found.", req.getWuid());
  1265. }
  1266. catch(IException* e)
  1267. {
  1268. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1269. }
  1270. return true;
  1271. }
  1272. bool CFileSprayEx::onGetDFUProgress(IEspContext &context, IEspProgressRequest &req, IEspProgressResponse &resp)
  1273. {
  1274. try
  1275. {
  1276. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Read, ECLWATCH_DFU_WU_ACCESS_DENIED, "Access to DFU workunit is denied.");
  1277. const char* wuid = req.getWuid();
  1278. if(!wuid || !*wuid)
  1279. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Workunit ID not specified.");
  1280. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1281. Owned<IConstDFUWorkUnit> wu = factory->openWorkUnit(req.getWuid(), false);
  1282. if(!wu)
  1283. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Dfu workunit %s not found.", req.getWuid());
  1284. resp.setWuid(req.getWuid());
  1285. PROGLOG("GetDFUProgress: %s", wuid);
  1286. IConstDFUprogress *prog = wu->queryProgress();
  1287. if (prog)
  1288. {
  1289. resp.setPercentDone(prog->getPercentDone());
  1290. resp.setKbPerSec(prog->getKbPerSec());
  1291. resp.setKbPerSecAve(prog->getKbPerSecAve());
  1292. resp.setSecsLeft(prog->getSecsLeft());
  1293. StringBuffer statestr;
  1294. encodeDFUstate(prog->getState(), statestr);
  1295. resp.setState(statestr.str());
  1296. resp.setSlavesDone(prog->getSlavesDone());
  1297. StringBuffer msg;
  1298. prog->formatProgressMessage(msg);
  1299. resp.setProgressMessage(msg.str());
  1300. prog->formatSummaryMessage(msg.clear());
  1301. resp.setSummaryMessage(msg.str());
  1302. prog->getTimeTaken(msg.clear());
  1303. resp.setTimeTaken(msg.str());
  1304. }
  1305. }
  1306. catch(IException* e)
  1307. {
  1308. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1309. }
  1310. return true;
  1311. }
  1312. bool CFileSprayEx::onCreateDFUWorkunit(IEspContext &context, IEspCreateDFUWorkunit &req, IEspCreateDFUWorkunitResponse &resp)
  1313. {
  1314. try
  1315. {
  1316. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to create DFU workunit. Permission denied.");
  1317. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1318. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  1319. setDFUServerQueueReq(req.getDFUServerQueue(), wu);
  1320. setUserAuth(context, wu);
  1321. wu->commit();
  1322. const char * d = wu->queryId();
  1323. IEspDFUWorkunit &result = resp.updateResult();
  1324. DeepAssign(context, wu, result);
  1325. result.setOverwrite(false);
  1326. result.setReplicate(true);
  1327. }
  1328. catch(IException* e)
  1329. {
  1330. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1331. }
  1332. return true;
  1333. }
  1334. bool CFileSprayEx::onUpdateDFUWorkunit(IEspContext &context, IEspUpdateDFUWorkunit &req, IEspUpdateDFUWorkunitResponse &resp)
  1335. {
  1336. try
  1337. {
  1338. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to update DFU workunit. Permission denied.");
  1339. IConstDFUWorkunit & reqWU = req.getWu();
  1340. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1341. Owned<IDFUWorkUnit> wu = factory->updateWorkUnit(reqWU.getID());
  1342. if(!wu)
  1343. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Dfu workunit %s not found.", reqWU.getID());
  1344. PROGLOG("UpdateDFUWorkunit: %s", reqWU.getID());
  1345. IDFUprogress *prog = wu->queryUpdateProgress();
  1346. if (prog && req.getStateOrig() != reqWU.getState())
  1347. {
  1348. if (prog->getState() != req.getStateOrig())
  1349. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT,"Cannot update DFU workunit %s because its state has been changed internally. Please refresh the page and try again.",reqWU.getID());
  1350. prog->setState((enum DFUstate)reqWU.getState());
  1351. }
  1352. const char* clusterOrig = req.getClusterOrig();
  1353. const char* cluster = reqWU.getClusterName();
  1354. if(cluster && (!clusterOrig || stricmp(clusterOrig, cluster)))
  1355. {
  1356. wu->setClusterName(reqWU.getClusterName());
  1357. }
  1358. const char* jobNameOrig = req.getJobNameOrig();
  1359. const char* jobName = reqWU.getJobName();
  1360. if(jobName && (!jobNameOrig || stricmp(jobNameOrig, jobName)))
  1361. {
  1362. wu->setJobName(jobName);
  1363. }
  1364. if (reqWU.getIsProtected() != req.getIsProtectedOrig())
  1365. wu->protect(reqWU.getIsProtected());
  1366. wu->commit();
  1367. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(reqWU.getID()).str());
  1368. }
  1369. catch(IException* e)
  1370. {
  1371. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1372. }
  1373. return true;
  1374. }
  1375. bool markWUFailed(IDFUWorkUnitFactory *f, const char *wuid)
  1376. {
  1377. Owned<IDFUWorkUnit> wu = f->updateWorkUnit(wuid);
  1378. if(!wu)
  1379. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Dfu workunit %s not found.", wuid);
  1380. IDFUprogress *prog = wu->queryUpdateProgress();
  1381. if(!prog)
  1382. throw MakeStringException(ECLWATCH_PROGRESS_INFO_NOT_FOUND, "progress information not found for workunit %s.", wuid);
  1383. else if(prog->getState() == DFUstate_started)
  1384. throw MakeStringException(ECLWATCH_CANNOT_DELETE_WORKUNIT, "Cannot delete workunit %s because its state is Started.", wuid);
  1385. else
  1386. {
  1387. prog->setState(DFUstate_failed);
  1388. return true;
  1389. }
  1390. return false;
  1391. }
  1392. static unsigned NumOfDFUWUActionNames = 6;
  1393. static const char *DFUWUActionNames[] = { "Delete", "Protect" , "Unprotect" , "Restore" , "SetToFailed", "Archive" };
  1394. bool CFileSprayEx::onDFUWorkunitsAction(IEspContext &context, IEspDFUWorkunitsActionRequest &req, IEspDFUWorkunitsActionResponse &resp)
  1395. {
  1396. try
  1397. {
  1398. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to update DFU workunit. Permission denied.");
  1399. CDFUWUActions action = req.getType();
  1400. if (action == DFUWUActions_Undefined)
  1401. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Action not defined.");
  1402. const char* actionStr = (action < NumOfDFUWUActionNames) ? DFUWUActionNames[action] : "Unknown";
  1403. StringArray& wuids = req.getWuids();
  1404. if (!wuids.ordinality())
  1405. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Workunit not defined.");
  1406. if ((action == CDFUWUActions_Restore) || (action == CDFUWUActions_Archive))
  1407. {
  1408. StringBuffer msg;
  1409. ForEachItemIn(i, wuids)
  1410. {
  1411. StringBuffer wuidStr(wuids.item(i));
  1412. const char* wuid = wuidStr.trim().str();
  1413. if (isEmptyString(wuid))
  1414. msg.appendf("Empty Workunit ID at %u. ", i);
  1415. }
  1416. if (!msg.isEmpty())
  1417. throw makeStringException(ECLWATCH_INVALID_INPUT, msg);
  1418. Owned<ISashaCommand> cmd = archiveOrRestoreWorkunits(wuids, nullptr, action == CDFUWUActions_Archive, true);
  1419. IArrayOf<IEspDFUActionResult> results;
  1420. ForEachItemIn(x, wuids)
  1421. {
  1422. Owned<IEspDFUActionResult> res = createDFUActionResult();
  1423. res->setID(wuids.item(x));
  1424. res->setAction(actionStr);
  1425. StringBuffer reply;
  1426. if (action == CDFUWUActions_Restore)
  1427. reply.set("Restore ID: ");
  1428. else
  1429. reply.set("Archive ID: ");
  1430. if (cmd->getId(x, reply))
  1431. res->setResult(reply.str());
  1432. else
  1433. res->setResult("Failed to get Archive/restore ID.");
  1434. results.append(*res.getLink());
  1435. }
  1436. resp.setDFUActionResults(results);
  1437. return true;
  1438. }
  1439. IArrayOf<IEspDFUActionResult> results;
  1440. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1441. for(unsigned i = 0; i < wuids.ordinality(); ++i)
  1442. {
  1443. const char* wuid = wuids.item(i);
  1444. Owned<IEspDFUActionResult> res = createDFUActionResult("", "");
  1445. res->setID(wuid);
  1446. res->setAction(actionStr);
  1447. try
  1448. {
  1449. PROGLOG("%s %s", actionStr, wuid);
  1450. switch (action)
  1451. {
  1452. case CDFUWUActions_Delete:
  1453. if (!markWUFailed(factory, wuid))
  1454. throw MakeStringException(ECLWATCH_CANNOT_DELETE_WORKUNIT, "Failed to mark workunit failed.");
  1455. if (!factory->deleteWorkUnit(wuid))
  1456. throw MakeStringException(ECLWATCH_CANNOT_DELETE_WORKUNIT, "Failed in deleting workunit.");
  1457. res->setResult("Success");
  1458. break;
  1459. case CDFUWUActions_Protect:
  1460. case CDFUWUActions_Unprotect:
  1461. case CDFUWUActions_SetToFailed:
  1462. Owned<IDFUWorkUnit> wu = factory->updateWorkUnit(wuid);
  1463. if(!wu.get())
  1464. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Failed in calling updateWorkUnit().");
  1465. switch (action)
  1466. {
  1467. case CDFUWUActions_Protect:
  1468. wu->protect(true);
  1469. break;
  1470. case CDFUWUActions_Unprotect:
  1471. wu->protect(false);
  1472. break;
  1473. case CDFUWUActions_SetToFailed:
  1474. IDFUprogress *prog = wu->queryUpdateProgress();
  1475. if (!prog)
  1476. throw MakeStringException(ECLWATCH_CANNOT_UPDATE_WORKUNIT, "Failed in calling queryUpdateProgress().");
  1477. prog->setState(DFUstate_failed);
  1478. break;
  1479. }
  1480. wu->commit();
  1481. res->setResult("Success");
  1482. break;
  1483. }
  1484. PROGLOG("%s %s done", actionStr, wuid);
  1485. }
  1486. catch (IException *e)
  1487. {
  1488. StringBuffer eMsg, failedMsg("Failed: ");
  1489. res->setResult(failedMsg.append(e->errorMessage(eMsg)).str());
  1490. e->Release();
  1491. }
  1492. results.append(*res.getLink());
  1493. }
  1494. resp.setDFUActionResults(results);
  1495. }
  1496. catch(IException* e)
  1497. {
  1498. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1499. }
  1500. return true;
  1501. }
  1502. bool CFileSprayEx::onDeleteDFUWorkunits(IEspContext &context, IEspDeleteDFUWorkunits &req, IEspDeleteDFUWorkunitsResponse &resp)
  1503. {
  1504. try
  1505. {
  1506. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to delete DFU workunit. Permission denied.");
  1507. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1508. StringArray & wuids = req.getWuids();
  1509. for(unsigned i = 0; i < wuids.ordinality(); ++i)
  1510. {
  1511. const char* wuid = wuids.item(i);
  1512. if (markWUFailed(factory, wuid))
  1513. {
  1514. factory->deleteWorkUnit(wuid);
  1515. PROGLOG("DeleteDFUWorkunits: %s deleted", wuid);
  1516. }
  1517. }
  1518. resp.setRedirectUrl("/FileSpray/GetDFUWorkunits");
  1519. }
  1520. catch(IException* e)
  1521. {
  1522. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1523. }
  1524. return true;
  1525. }
  1526. bool CFileSprayEx::onDeleteDFUWorkunit(IEspContext &context, IEspDeleteDFUWorkunit &req, IEspDeleteDFUWorkunitResponse &resp)
  1527. {
  1528. try
  1529. {
  1530. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to delete DFU workunit. Permission denied.");
  1531. const char* wuid = req.getWuid();
  1532. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1533. if (markWUFailed(factory, wuid))
  1534. {
  1535. resp.setResult(factory->deleteWorkUnit(wuid));
  1536. PROGLOG("DeleteDFUWorkunit: %s deleted", wuid);
  1537. }
  1538. else
  1539. resp.setResult(false);
  1540. resp.setRedirectUrl("/FileSpray/GetDFUWorkunits");
  1541. }
  1542. catch(IException* e)
  1543. {
  1544. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1545. }
  1546. return true;
  1547. }
  1548. bool CFileSprayEx::onSubmitDFUWorkunit(IEspContext &context, IEspSubmitDFUWorkunit &req, IEspSubmitDFUWorkunitResponse &resp)
  1549. {
  1550. try
  1551. {
  1552. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to submit DFU workunit. Permission denied.");
  1553. if (!req.getWuid() || !*req.getWuid())
  1554. throw MakeStringException(ECLWATCH_MISSING_PARAMS, "Workunit ID required");
  1555. PROGLOG("SubmitDFUWorkunit: %s", req.getWuid());
  1556. submitDFUWorkUnit(req.getWuid());
  1557. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(req.getWuid()).str());
  1558. }
  1559. catch(IException* e)
  1560. {
  1561. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1562. }
  1563. return true;
  1564. }
  1565. bool CFileSprayEx::onAbortDFUWorkunit(IEspContext &context, IEspAbortDFUWorkunit &req, IEspAbortDFUWorkunitResponse &resp)
  1566. {
  1567. try
  1568. {
  1569. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Write, ECLWATCH_DFU_WU_ACCESS_DENIED, "Failed to abort DFU workunit. Permission denied.");
  1570. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1571. Owned<IDFUWorkUnit> wu = factory->updateWorkUnit(req.getWuid());
  1572. if(!wu)
  1573. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Dfu workunit %s not found.", req.getWuid());
  1574. PROGLOG("AbortDFUWorkunit: %s", req.getWuid());
  1575. wu->requestAbort();
  1576. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(req.getWuid()).str());
  1577. }
  1578. catch(IException* e)
  1579. {
  1580. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1581. }
  1582. return true;
  1583. }
  1584. bool CFileSprayEx::onGetDFUExceptions(IEspContext &context, IEspGetDFUExceptions &req, IEspGetDFUExceptionsResponse &resp)
  1585. {
  1586. try
  1587. {
  1588. context.ensureFeatureAccess(DFU_EX_URL, SecAccess_Read, ECLWATCH_DFU_EX_ACCESS_DENIED, "Failed to get DFU Exceptions. Permission denied.");
  1589. IArrayOf<IEspDFUException> result;
  1590. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1591. Owned<IDFUWorkUnit> wu = factory->updateWorkUnit(req.getWuid());
  1592. if(!wu)
  1593. throw MakeStringException(ECLWATCH_CANNOT_GET_WORKUNIT, "Dfu workunit %s not found.", req.getWuid());
  1594. PROGLOG("GetDFUExceptions: %s", req.getWuid());
  1595. Owned<IExceptionIterator> itr = wu->getExceptionIterator();
  1596. itr->first();
  1597. while(itr->isValid())
  1598. {
  1599. Owned<IEspDFUException> resultE = createDFUException("", "");
  1600. IException &e = itr->query();
  1601. resultE->setCode(e.errorCode());
  1602. StringBuffer msg;
  1603. resultE->setMessage(e.errorMessage(msg).str());
  1604. result.append(*resultE.getLink());
  1605. itr->next();
  1606. }
  1607. resp.setResult(result);
  1608. }
  1609. catch(IException* e)
  1610. {
  1611. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1612. }
  1613. return true;
  1614. }
  1615. void CFileSprayEx::readAndCheckSpraySourceReq(MemoryBuffer& srcxml, const char* srcIP, const char* srcPath, const char* srcPlane,
  1616. StringBuffer& sourceIPReq, StringBuffer& sourcePathReq)
  1617. {
  1618. StringBuffer sourcePath;
  1619. if(srcxml.length() == 0)
  1620. {
  1621. if (!isEmptyString(srcPlane))
  1622. {
  1623. Owned<IPropertyTree> dropZone = getDropZonePlane(srcPlane);
  1624. if (!dropZone)
  1625. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Unknown landing zone: %s", srcPlane);
  1626. const char * dropZonePlanePath = dropZone->queryProp("@prefix");
  1627. if (isAbsolutePath(srcPath))
  1628. {
  1629. if (!startsWith(srcPath,dropZonePlanePath))
  1630. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid source path");
  1631. sourcePath.append(srcPath).trim();
  1632. }
  1633. else
  1634. {
  1635. sourcePath.append(dropZonePlanePath);
  1636. addNonEmptyPathSepChar(sourcePath);
  1637. sourcePath.append(srcPath).trim();
  1638. }
  1639. const char * hostGroup = dropZone->queryProp("@hostGroup");
  1640. if (hostGroup)
  1641. {
  1642. Owned<IPropertyTree> match = getHostGroup(hostGroup,true);
  1643. if (!isEmptyString(srcIP))
  1644. {
  1645. // Already have srcIP. Just need to check that the ip is valid for storage plane.
  1646. bool ipAddressMatches = false;
  1647. Owned<IPropertyTreeIterator> hostIter = match->getElements("hosts");
  1648. ForEach (*hostIter)
  1649. {
  1650. const char *knownIP = hostIter->query().queryProp(nullptr);
  1651. if (strcmp(knownIP, srcIP)==0)
  1652. {
  1653. ipAddressMatches=true;
  1654. break;
  1655. }
  1656. }
  1657. if (!ipAddressMatches)
  1658. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "srcip %s is not valid storage plane %s", srcIP, srcPlane);
  1659. sourceIPReq.set(srcIP);
  1660. }
  1661. else
  1662. sourceIPReq.set(match->queryProp("hosts[1]"));
  1663. }
  1664. else
  1665. {
  1666. sourceIPReq.set("localhost"); // storage plane will be mounted when not using hostgroup
  1667. }
  1668. }
  1669. else
  1670. {
  1671. sourcePath.append(srcPath).trim();
  1672. sourceIPReq.set(srcIP).trim();
  1673. }
  1674. if (sourceIPReq.isEmpty())
  1675. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Source network IP not specified.");
  1676. if (sourcePath.isEmpty())
  1677. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Source path not specified.");
  1678. }
  1679. getStandardPosixPath(sourcePathReq, sourcePath.str());
  1680. }
  1681. static void checkValidDfuQueue(const char * dfuQueue)
  1682. {
  1683. if (isEmptyString(dfuQueue))
  1684. return;
  1685. #ifndef _CONTAINERIZED
  1686. Owned<IEnvironmentFactory> envFactory = getEnvironmentFactory(true);
  1687. Owned<IConstEnvironment> constEnv = envFactory->openEnvironment();
  1688. if (!constEnv->isValidDfuQueueName(dfuQueue))
  1689. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid DFU server queue name:'%s'", dfuQueue);
  1690. #else
  1691. bool isValidDfuQueueName = false;
  1692. Owned<IPropertyTreeIterator> dfuServers = getComponentConfigSP()->getElements("dfuQueues");
  1693. ForEach(*dfuServers)
  1694. {
  1695. IPropertyTree & dfuServer = dfuServers->query();
  1696. const char * dfuServerName = dfuServer.queryProp("@name");
  1697. StringBuffer knownDfuQueueName;
  1698. getDfuQueueName(knownDfuQueueName, dfuServerName);
  1699. if (streq(dfuQueue, knownDfuQueueName))
  1700. {
  1701. isValidDfuQueueName = true;
  1702. break;
  1703. }
  1704. }
  1705. if (!isValidDfuQueueName)
  1706. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Invalid DFU server queue name: '%s'", dfuQueue);
  1707. #endif
  1708. }
  1709. bool CFileSprayEx::onSprayFixed(IEspContext &context, IEspSprayFixed &req, IEspSprayFixedResponse &resp)
  1710. {
  1711. try
  1712. {
  1713. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Write, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do Spray. Permission denied.");
  1714. StringBuffer destFolder, destTitle, defaultFolder, defaultReplicateFolder;
  1715. const char* destNodeGroup = req.getDestGroup();
  1716. if (isEmptyString(destNodeGroup))
  1717. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Destination node group not specified.");
  1718. MemoryBuffer& srcxml = (MemoryBuffer&)req.getSrcxml();
  1719. StringBuffer sourceIPReq, sourcePathReq;
  1720. readAndCheckSpraySourceReq(srcxml, req.getSourceIP(), req.getSourcePath(), req.getSourcePlane(), sourceIPReq, sourcePathReq);
  1721. const char* srcip = sourceIPReq.str();
  1722. const char* srcfile = sourcePathReq.str();
  1723. const char* destname = req.getDestLogicalName();
  1724. if(isEmptyString(destname))
  1725. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Destination file not specified.");
  1726. CDfsLogicalFileName lfn;
  1727. if (!lfn.setValidate(destname))
  1728. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Invalid destination filename:'%s'", destname);
  1729. destname = lfn.get();
  1730. PROGLOG("SprayFixed: DestLogicalName %s, DestGroup %s", destname, destNodeGroup);
  1731. StringBuffer gName, ipAddr;
  1732. const char *pTr = strchr(destNodeGroup, ' ');
  1733. if (pTr)
  1734. {
  1735. gName.append(pTr - destNodeGroup, destNodeGroup);
  1736. ipAddr.append(pTr+1);
  1737. }
  1738. else
  1739. gName.append(destNodeGroup);
  1740. if (ipAddr.length() > 0)
  1741. ParseLogicalPath(destname, ipAddr.str(), NULL, destFolder, destTitle, defaultFolder, defaultReplicateFolder);
  1742. else
  1743. ParseLogicalPath(destname, destNodeGroup, NULL, destFolder, destTitle, defaultFolder, defaultReplicateFolder);
  1744. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1745. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  1746. wu->setClusterName(gName.str());
  1747. wu->setJobName(destTitle.str());
  1748. const char * dfuQueue = req.getDFUServerQueue();
  1749. checkValidDfuQueue(dfuQueue);
  1750. setDFUServerQueueReq(dfuQueue, wu);
  1751. setUserAuth(context, wu);
  1752. wu->setCommand(DFUcmd_import);
  1753. IDFUfileSpec *source = wu->queryUpdateSource();
  1754. if(srcxml.length() == 0)
  1755. {
  1756. RemoteMultiFilename rmfn;
  1757. SocketEndpoint ep(srcip);
  1758. if (ep.isNull())
  1759. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "SprayFixed to %s: cannot resolve source network IP from %s.", destname, srcip);
  1760. rmfn.setEp(ep);
  1761. StringBuffer fnamebuf(srcfile);
  1762. fnamebuf.trim();
  1763. rmfn.append(fnamebuf.str()); // handles comma separated files
  1764. source->setMultiFilename(rmfn);
  1765. }
  1766. else
  1767. {
  1768. srcxml.append('\0');
  1769. source->setFromXML((const char*)srcxml.toByteArray());
  1770. }
  1771. IDFUfileSpec *destination = wu->queryUpdateDestination();
  1772. bool nosplit = req.getNosplit();
  1773. int recordsize = req.getSourceRecordSize();
  1774. const char* format = req.getSourceFormat();
  1775. if ((recordsize == RECFMVB_RECSIZE_ESCAPE) || (format && strieq(format, "recfmvb")))
  1776. {//recordsize may be set by dfuplus; format may be set by EclWatch
  1777. source->setFormat(DFUff_recfmvb);
  1778. destination->setFormat(DFUff_variable);
  1779. }
  1780. else if ((recordsize == RECFMV_RECSIZE_ESCAPE) || (format && strieq(format, "recfmv")))
  1781. {
  1782. source->setFormat(DFUff_recfmv);
  1783. destination->setFormat(DFUff_variable);
  1784. }
  1785. else if ((recordsize == PREFIX_VARIABLE_RECSIZE_ESCAPE) || (format && strieq(format, "variable")))
  1786. {
  1787. source->setFormat(DFUff_variable);
  1788. destination->setFormat(DFUff_variable);
  1789. }
  1790. else if((recordsize == PREFIX_VARIABLE_BIGENDIAN_RECSIZE_ESCAPE) || (format && strieq(format, "variablebigendian")))
  1791. {
  1792. source->setFormat(DFUff_variablebigendian);
  1793. destination->setFormat(DFUff_variable);
  1794. }
  1795. else if(recordsize == 0 && !nosplit) // -ve record sizes for blocked
  1796. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid record size");
  1797. else
  1798. source->setRecordSize(recordsize);
  1799. destination->setLogicalName(destname);
  1800. destination->setDirectory(destFolder.str());
  1801. StringBuffer fileMask;
  1802. constructFileMask(destTitle.str(), fileMask);
  1803. destination->setFileMask(fileMask.str());
  1804. destination->setGroupName(gName.str());
  1805. const char * encryptkey = req.getEncrypt();
  1806. if(req.getCompress()||(encryptkey&&*encryptkey))
  1807. destination->setCompressed(true);
  1808. ClusterPartDiskMapSpec mspec;
  1809. destination->getClusterPartDiskMapSpec(gName.str(), mspec);
  1810. mspec.setDefaultBaseDir(defaultFolder.str());
  1811. mspec.setDefaultReplicateDir(defaultReplicateFolder.str());
  1812. destination->setClusterPartDiskMapSpec(gName.str(), mspec);
  1813. int repo = req.getReplicateOffset();
  1814. bool isNull = req.getReplicateOffset_isNull();
  1815. if (!isNull && (repo!=1))
  1816. destination->setReplicateOffset(repo);
  1817. if (req.getWrap())
  1818. destination->setWrap(true);
  1819. IDFUoptions *options = wu->queryUpdateOptions();
  1820. const char * decryptkey = req.getDecrypt();
  1821. if ((encryptkey&&*encryptkey)||(decryptkey&&*decryptkey))
  1822. options->setEncDec(encryptkey,decryptkey);
  1823. options->setReplicate(req.getReplicate());
  1824. options->setOverwrite(req.getOverwrite()); // needed if target already exists
  1825. const char* prefix = req.getPrefix();
  1826. if(prefix && *prefix)
  1827. options->setLengthPrefix(prefix);
  1828. if(req.getNosplit())
  1829. options->setNoSplit(true);
  1830. if(req.getNorecover())
  1831. options->setNoRecover(true);
  1832. if(req.getMaxConnections() > 0)
  1833. options->setmaxConnections(req.getMaxConnections());
  1834. if(req.getThrottle() > 0)
  1835. options->setThrottle(req.getThrottle());
  1836. if(req.getTransferBufferSize() > 0)
  1837. options->setTransferBufferSize(req.getTransferBufferSize());
  1838. if (req.getPull())
  1839. options->setPull(true);
  1840. if (req.getPush())
  1841. options->setPush(true);
  1842. if (!req.getNoCommon_isNull())
  1843. options->setNoCommon(req.getNoCommon());
  1844. if (req.getFailIfNoSourceFile())
  1845. options->setFailIfNoSourceFile(true);
  1846. if (req.getRecordStructurePresent())
  1847. options->setRecordStructurePresent(true);
  1848. if (!req.getExpireDays_isNull())
  1849. options->setExpireDays(req.getExpireDays());
  1850. resp.setWuid(wu->queryId());
  1851. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  1852. submitDFUWorkUnit(wu.getClear());
  1853. }
  1854. catch(IException* e)
  1855. {
  1856. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  1857. }
  1858. return true;
  1859. }
  1860. bool CFileSprayEx::onSprayVariable(IEspContext &context, IEspSprayVariable &req, IEspSprayResponse &resp)
  1861. {
  1862. try
  1863. {
  1864. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Write, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do Spray. Permission denied.");
  1865. StringBuffer destFolder, destTitle, defaultFolder, defaultReplicateFolder;
  1866. const char* destNodeGroup = req.getDestGroup();
  1867. if (isEmptyString(destNodeGroup))
  1868. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Destination node group not specified.");
  1869. StringBuffer gName, ipAddr;
  1870. const char *pTr = strchr(destNodeGroup, ' ');
  1871. if (pTr)
  1872. {
  1873. gName.append(pTr - destNodeGroup, destNodeGroup);
  1874. ipAddr.append(pTr+1);
  1875. }
  1876. else
  1877. gName.append(destNodeGroup);
  1878. MemoryBuffer& srcxml = (MemoryBuffer&)req.getSrcxml();
  1879. StringBuffer sourceIPReq, sourcePathReq;
  1880. readAndCheckSpraySourceReq(srcxml, req.getSourceIP(), req.getSourcePath(), req.getSourcePlane(), sourceIPReq, sourcePathReq);
  1881. const char* srcip = sourceIPReq.str();
  1882. const char* srcfile = sourcePathReq.str();
  1883. const char* destname = req.getDestLogicalName();
  1884. if(isEmptyString(destname))
  1885. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Destination file not specified.");
  1886. CDfsLogicalFileName lfn;
  1887. if (!lfn.setValidate(destname))
  1888. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Invalid destination filename:'%s'", destname);
  1889. destname = lfn.get();
  1890. PROGLOG("SprayVariable: DestLogicalName %s, DestGroup %s", destname, destNodeGroup);
  1891. if (ipAddr.length() > 0)
  1892. ParseLogicalPath(destname, ipAddr.str(), NULL, destFolder, destTitle, defaultFolder, defaultReplicateFolder);
  1893. else
  1894. ParseLogicalPath(destname, destNodeGroup, NULL, destFolder, destTitle, defaultFolder, defaultReplicateFolder);
  1895. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  1896. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  1897. wu->setClusterName(gName.str());
  1898. wu->setJobName(destTitle.str());
  1899. const char * dfuQueue = req.getDFUServerQueue();
  1900. checkValidDfuQueue(dfuQueue);
  1901. setDFUServerQueueReq(dfuQueue, wu);
  1902. setUserAuth(context, wu);
  1903. wu->setCommand(DFUcmd_import);
  1904. IDFUfileSpec *source = wu->queryUpdateSource();
  1905. IDFUfileSpec *destination = wu->queryUpdateDestination();
  1906. IDFUoptions *options = wu->queryUpdateOptions();
  1907. if(srcxml.length() == 0)
  1908. {
  1909. RemoteMultiFilename rmfn;
  1910. SocketEndpoint ep(srcip);
  1911. if (ep.isNull())
  1912. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "SprayVariable to %s: cannot resolve source network IP from %s.", destname, srcip);
  1913. rmfn.setEp(ep);
  1914. StringBuffer fnamebuf(srcfile);
  1915. fnamebuf.trim();
  1916. rmfn.append(fnamebuf.str()); // handles comma separated files
  1917. source->setMultiFilename(rmfn);
  1918. }
  1919. else
  1920. {
  1921. srcxml.append('\0');
  1922. source->setFromXML((const char*)srcxml.toByteArray());
  1923. }
  1924. source->setMaxRecordSize(req.getSourceMaxRecordSize());
  1925. source->setFormat((DFUfileformat)req.getSourceFormat());
  1926. StringBuffer rowtag;
  1927. if (req.getIsJSON())
  1928. {
  1929. const char *srcRowPath = req.getSourceRowPath();
  1930. if (!srcRowPath || *srcRowPath != '/')
  1931. rowtag.append("/");
  1932. rowtag.append(srcRowPath);
  1933. }
  1934. else
  1935. rowtag.append(req.getSourceRowTag());
  1936. // if rowTag specified, it means it's xml or json format, otherwise it's csv
  1937. if(rowtag.length())
  1938. {
  1939. source->setRowTag(rowtag);
  1940. options->setKeepHeader(true);
  1941. }
  1942. else
  1943. {
  1944. const char* cs = req.getSourceCsvSeparate();
  1945. if (req.getNoSourceCsvSeparator())
  1946. {
  1947. cs = "";
  1948. }
  1949. else if(cs == NULL || *cs == '\0')
  1950. cs = "\\,";
  1951. const char* ct = req.getSourceCsvTerminate();
  1952. if(ct == NULL || *ct == '\0')
  1953. ct = "\\n,\\r\\n";
  1954. const char* cq = req.getSourceCsvQuote();
  1955. if(cq== NULL)
  1956. cq = "\"";
  1957. source->setCsvOptions(cs, ct, cq, req.getSourceCsvEscape(), req.getQuotedTerminator());
  1958. options->setQuotedTerminator(req.getQuotedTerminator());
  1959. }
  1960. destination->setLogicalName(destname);
  1961. destination->setDirectory(destFolder.str());
  1962. StringBuffer fileMask;
  1963. constructFileMask(destTitle.str(), fileMask);
  1964. destination->setFileMask(fileMask.str());
  1965. destination->setGroupName(gName.str());
  1966. ClusterPartDiskMapSpec mspec;
  1967. destination->getClusterPartDiskMapSpec(gName.str(), mspec);
  1968. mspec.setDefaultBaseDir(defaultFolder.str());
  1969. mspec.setDefaultReplicateDir(defaultReplicateFolder.str());
  1970. destination->setClusterPartDiskMapSpec(gName.str(), mspec);
  1971. const char * encryptkey = req.getEncrypt();
  1972. if(req.getCompress()||(encryptkey&&*encryptkey))
  1973. destination->setCompressed(true);
  1974. const char * decryptkey = req.getDecrypt();
  1975. if ((encryptkey&&*encryptkey)||(decryptkey&&*decryptkey))
  1976. options->setEncDec(encryptkey,decryptkey);
  1977. int repo = req.getReplicateOffset();
  1978. bool isNull = req.getReplicateOffset_isNull();
  1979. if (!isNull && (repo!=1))
  1980. destination->setReplicateOffset(repo);
  1981. options->setReplicate(req.getReplicate());
  1982. options->setOverwrite(req.getOverwrite()); // needed if target already exists
  1983. const char* prefix = req.getPrefix();
  1984. if(prefix && *prefix)
  1985. options->setLengthPrefix(prefix);
  1986. if(req.getNosplit())
  1987. options->setNoSplit(true);
  1988. if(req.getNorecover())
  1989. options->setNoRecover(true);
  1990. if(req.getMaxConnections() > 0)
  1991. options->setmaxConnections(req.getMaxConnections());
  1992. if(req.getThrottle() > 0)
  1993. options->setThrottle(req.getThrottle());
  1994. if(req.getTransferBufferSize() > 0)
  1995. options->setTransferBufferSize(req.getTransferBufferSize());
  1996. if (req.getPull())
  1997. options->setPull(true);
  1998. if (req.getPush())
  1999. options->setPush(true);
  2000. if (req.getFailIfNoSourceFile())
  2001. options->setFailIfNoSourceFile(true);
  2002. if (req.getRecordStructurePresent())
  2003. options->setRecordStructurePresent(true);
  2004. if (!req.getExpireDays_isNull())
  2005. options->setExpireDays(req.getExpireDays());
  2006. resp.setWuid(wu->queryId());
  2007. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2008. submitDFUWorkUnit(wu.getClear());
  2009. }
  2010. catch(IException* e)
  2011. {
  2012. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2013. }
  2014. return true;
  2015. }
  2016. bool CFileSprayEx::onReplicate(IEspContext &context, IEspReplicate &req, IEspReplicateResponse &resp)
  2017. {
  2018. try
  2019. {
  2020. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Write, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do Replicate. Permission denied.");
  2021. const char* srcname = req.getSourceLogicalName();
  2022. if(!srcname || !*srcname)
  2023. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Source logical file not specified.");
  2024. PROGLOG("Replicate %s", srcname);
  2025. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2026. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  2027. StringBuffer jobname("Replicate: ");
  2028. jobname.append(srcname);
  2029. wu->setJobName(jobname.str());
  2030. setDFUServerQueueReq(req.getDFUServerQueue(), wu);
  2031. setUserAuth(context, wu);
  2032. wu->setCommand(DFUcmd_replicate);
  2033. IDFUfileSpec *source = wu->queryUpdateSource();
  2034. if (source)
  2035. {
  2036. source->setLogicalName(srcname);
  2037. int repo = req.getReplicateOffset();
  2038. if (repo!=1)
  2039. source->setReplicateOffset(repo);
  2040. }
  2041. const char* cluster = req.getCluster();
  2042. if(cluster && *cluster)
  2043. {
  2044. IDFUoptions *opt = wu->queryUpdateOptions();
  2045. opt->setReplicateMode(DFURMmissing,cluster,req.getRepeatLast(),req.getOnlyRepeated());
  2046. }
  2047. resp.setWuid(wu->queryId());
  2048. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2049. submitDFUWorkUnit(wu.getClear());
  2050. }
  2051. catch(IException* e)
  2052. {
  2053. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2054. }
  2055. return true;
  2056. }
  2057. void CFileSprayEx::getDropZoneInfoByDestPlane(double clientVersion, const char* destPlane, const char* destFileIn, StringBuffer& destFileOut, StringBuffer& umask, StringBuffer & hostip)
  2058. {
  2059. Owned<IPropertyTree> dropZone = getDropZonePlane(destPlane);
  2060. if (!dropZone)
  2061. throw makeStringExceptionV(ECLWATCH_DROP_ZONE_NOT_FOUND, "Unknown landing zone %s", destPlane);
  2062. StringBuffer fullDropZoneDir(dropZone->queryProp("@prefix"));
  2063. addPathSepChar(fullDropZoneDir);
  2064. if (isAbsolutePath(destFileIn))
  2065. {
  2066. if (!startsWith(destFileIn, fullDropZoneDir))
  2067. throw makeStringExceptionV(ECLWATCH_DROP_ZONE_NOT_FOUND, "No landing zone configured for %s:%s", destPlane, destFileIn);
  2068. }
  2069. else
  2070. {
  2071. destFileOut.append(fullDropZoneDir);
  2072. addNonEmptyPathSepChar(destFileOut);
  2073. }
  2074. destFileOut.append(destFileIn).trim();
  2075. dropZone->getProp("@umask", umask);
  2076. const char * hostGroup = dropZone->queryProp("@hostGroup");
  2077. if (hostGroup)
  2078. {
  2079. Owned<IPropertyTree> match = getHostGroup(hostGroup,true);
  2080. if (!hostip.isEmpty())
  2081. {
  2082. // Already have hostip. Just need to check that the ip is valid for storage plane.
  2083. bool ipAddressMatches = false;
  2084. Owned<IPropertyTreeIterator> hostIter = match->getElements("hosts");
  2085. ForEach (*hostIter)
  2086. {
  2087. const char *knownIP = hostIter->query().queryProp(nullptr);
  2088. if (strcmp(knownIP, hostip)==0)
  2089. {
  2090. ipAddressMatches=true;
  2091. break;
  2092. }
  2093. }
  2094. if (!ipAddressMatches)
  2095. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "destip %s is not valid storage plane %s", hostip.str(), destPlane);
  2096. }
  2097. else
  2098. hostip.set(match->queryProp("hosts[1]"));
  2099. }
  2100. else
  2101. {
  2102. hostip.set("localhost"); // storage plane will be mounted when not using hostgroup
  2103. }
  2104. }
  2105. void CFileSprayEx::getDropZoneInfoByIP(double clientVersion, const char* ip, const char* destFileIn, StringBuffer& destFileOut, StringBuffer& umask)
  2106. {
  2107. #ifndef _CONTAINERIZED
  2108. if (destFileIn && *destFileIn)
  2109. destFileOut.set(destFileIn);
  2110. if (!ip || !*ip)
  2111. throw MakeStringExceptionDirect(ECLWATCH_INVALID_IP, "Network address must be specified for a drop zone!");
  2112. Owned<IEnvironmentFactory> factory = getEnvironmentFactory(true);
  2113. Owned<IConstEnvironment> constEnv = factory->openEnvironment();
  2114. StringBuffer destFile;
  2115. if (isAbsolutePath(destFileIn))
  2116. {
  2117. destFile.set(destFileIn);
  2118. Owned<IConstDropZoneInfo> dropZone = constEnv->getDropZoneByAddressPath(ip, destFile.str());
  2119. if (!dropZone)
  2120. {
  2121. if (constEnv->isDropZoneRestrictionEnabled())
  2122. throw MakeStringException(ECLWATCH_DROP_ZONE_NOT_FOUND, "No drop zone configured for '%s' and '%s'. Check your system drop zone configuration.", ip, destFile.str());
  2123. else
  2124. {
  2125. LOG(MCdebugInfo, unknownJob, "No drop zone configured for '%s' and '%s'. Check your system drop zone configuration.", ip, destFile.str());
  2126. return;
  2127. }
  2128. }
  2129. SCMStringBuffer directory, maskBuf;
  2130. dropZone->getDirectory(directory);
  2131. destFileOut.set(destFile.str());
  2132. dropZone->getUMask(maskBuf);
  2133. if (maskBuf.length())
  2134. umask.set(maskBuf.str());
  2135. return;
  2136. }
  2137. Owned<IConstDropZoneInfoIterator> dropZoneItr = constEnv->getDropZoneIteratorByAddress(ip);
  2138. if (dropZoneItr->count() < 1)
  2139. {
  2140. if (constEnv->isDropZoneRestrictionEnabled())
  2141. throw MakeStringException(ECLWATCH_DROP_ZONE_NOT_FOUND, "Drop zone not found for network address '%s'. Check your system drop zone configuration.", ip);
  2142. else
  2143. {
  2144. LOG(MCdebugInfo, unknownJob, "Drop zone not found for network address '%s'. Check your system drop zone configuration.", ip);
  2145. return;
  2146. }
  2147. }
  2148. bool dzFound = false;
  2149. ForEach(*dropZoneItr)
  2150. {
  2151. IConstDropZoneInfo& dropZoneInfo = dropZoneItr->query();
  2152. SCMStringBuffer dropZoneDirectory, dropZoneUMask;
  2153. dropZoneInfo.getDirectory(dropZoneDirectory);
  2154. dropZoneInfo.getUMask(dropZoneUMask);
  2155. if (!dropZoneDirectory.length())
  2156. continue;
  2157. if (!dzFound)
  2158. {
  2159. dzFound = true;
  2160. destFileOut.set(dropZoneDirectory.str());
  2161. addPathSepChar(destFileOut);
  2162. destFileOut.append(destFileIn);
  2163. if (dropZoneUMask.length())
  2164. umask.set(dropZoneUMask.str());
  2165. }
  2166. else
  2167. {
  2168. if (constEnv->isDropZoneRestrictionEnabled())
  2169. throw MakeStringException(ECLWATCH_INVALID_INPUT, "> 1 drop zones found for network address '%s'.", ip);
  2170. else
  2171. {
  2172. LOG(MCdebugInfo, unknownJob, "> 1 drop zones found for network address '%s'.", ip);
  2173. return;
  2174. }
  2175. }
  2176. }
  2177. if (!dzFound)
  2178. {
  2179. if (constEnv->isDropZoneRestrictionEnabled())
  2180. throw MakeStringException(ECLWATCH_DROP_ZONE_NOT_FOUND, "No valid drop zone found for network address '%s'. Check your system drop zone configuration.", ip);
  2181. else
  2182. LOG(MCdebugInfo, unknownJob, "No valid drop zone found for network address '%s'. Check your system drop zone configuration.", ip);
  2183. }
  2184. #else
  2185. throw makeStringException(-1, "Internal error: CFileSprayEx::getDropZoneInfoByIP should not be called in containerized environment");
  2186. #endif
  2187. }
  2188. static StringBuffer & expandLogicalAsPhysical(StringBuffer & target, const char * name, const char * separator)
  2189. {
  2190. const char * cur = name;
  2191. for (;;)
  2192. {
  2193. const char * colon = strstr(cur, "::");
  2194. if (!colon)
  2195. break;
  2196. //MORE: Process special characters?
  2197. target.append(colon - cur, cur);
  2198. target.append(separator);
  2199. cur = colon + 2;
  2200. }
  2201. return target.append(cur);
  2202. }
  2203. bool CFileSprayEx::onDespray(IEspContext &context, IEspDespray &req, IEspDesprayResponse &resp)
  2204. {
  2205. try
  2206. {
  2207. context.ensureFeatureAccess(FILE_DESPRAY_URL, SecAccess_Write, ECLWATCH_FILE_DESPRAY_ACCESS_DENIED, "Failed to do Despray. Permission denied.");
  2208. const char* srcname = req.getSourceLogicalName();
  2209. if(!srcname || !*srcname)
  2210. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Source logical file not specified.");
  2211. PROGLOG("Despray %s", srcname);
  2212. double version = context.getClientVersion();
  2213. StringBuffer destip(req.getDestIP());
  2214. const char* destPlane = req.getDestPlane();
  2215. #ifdef _CONTAINERIZED
  2216. if (isEmptyString(destPlane))
  2217. destPlane = req.getDestGroup(); // allow eclwatch to continue providing storage plane as 'destgroup' field
  2218. #endif
  2219. StringBuffer destPath;
  2220. StringBuffer implicitDestFile;
  2221. const char* destfile = getStandardPosixPath(destPath, req.getDestPath()).str();
  2222. MemoryBuffer& dstxml = (MemoryBuffer&)req.getDstxml();
  2223. if(dstxml.length() == 0)
  2224. {
  2225. if(isEmptyString(destPlane) && destip.isEmpty())
  2226. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Destination network IP/storage plane not specified.");
  2227. //If the destination filename is not provided, calculate a relative filename from the logical filename
  2228. if(!destfile || !*destfile)
  2229. {
  2230. expandLogicalAsPhysical(implicitDestFile, srcname, "/");
  2231. destfile = implicitDestFile;
  2232. }
  2233. }
  2234. StringBuffer srcTitle;
  2235. ParseLogicalPath(srcname, srcTitle);
  2236. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2237. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  2238. wu->setJobName(srcTitle.str());
  2239. setDFUServerQueueReq(req.getDFUServerQueue(), wu);
  2240. setUserAuth(context, wu);
  2241. wu->setCommand(DFUcmd_export);
  2242. IDFUfileSpec *source = wu->queryUpdateSource();
  2243. IDFUfileSpec *destination = wu->queryUpdateDestination();
  2244. IDFUoptions *options = wu->queryUpdateOptions();
  2245. bool preserveFileParts = req.getWrap();
  2246. source->setLogicalName(srcname);
  2247. if(dstxml.length() == 0)
  2248. {
  2249. StringBuffer destfileWithPath, umask;
  2250. if (!isEmptyString(destPlane))
  2251. getDropZoneInfoByDestPlane(version, destPlane, destfile, destfileWithPath, umask, destip);
  2252. else
  2253. getDropZoneInfoByIP(version, destip, destfile, destfileWithPath, umask);
  2254. RemoteFilename rfn;
  2255. SocketEndpoint ep(destip.str());
  2256. if (ep.isNull())
  2257. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Despray %s: cannot resolve destination network IP from %s.", srcname, destip.str());
  2258. //Ensure the filename is dependent on the file part if parts are being preserved
  2259. if (preserveFileParts && !strstr(destfileWithPath, "$P$"))
  2260. destfileWithPath.append("._$P$_of_$N$");
  2261. rfn.setPath(ep, destfileWithPath.str());
  2262. if (umask.length())
  2263. options->setUMask(umask.str());
  2264. destination->setSingleFilename(rfn);
  2265. }
  2266. else
  2267. {
  2268. dstxml.append('\0');
  2269. destination->setFromXML((const char*)dstxml.toByteArray());
  2270. }
  2271. destination->setTitle(srcTitle.str());
  2272. options->setKeepHeader(true);
  2273. options->setOverwrite(req.getOverwrite()); // needed if target already exists
  2274. const char* splitprefix = req.getSplitprefix();
  2275. if(splitprefix && *splitprefix)
  2276. options->setSplitPrefix(splitprefix);
  2277. if (version > 1.01)
  2278. {
  2279. if(req.getMaxConnections() > 0)
  2280. options->setmaxConnections(req.getMaxConnections());
  2281. else if(req.getSingleConnection())
  2282. options->setmaxConnections(1);
  2283. }
  2284. else
  2285. {
  2286. if(req.getMaxConnections() > 0)
  2287. options->setmaxConnections(req.getMaxConnections());
  2288. }
  2289. if(req.getThrottle() > 0)
  2290. options->setThrottle(req.getThrottle());
  2291. if(req.getTransferBufferSize() > 0)
  2292. options->setTransferBufferSize(req.getTransferBufferSize());
  2293. if(req.getNorecover())
  2294. options->setNoRecover(true);
  2295. if (preserveFileParts) {
  2296. options->setPush(); // I think needed for a despray
  2297. destination->setWrap(true);
  2298. }
  2299. if (req.getMultiCopy())
  2300. destination->setMultiCopy(true);
  2301. const char * encryptkey = req.getEncrypt();
  2302. if(req.getCompress()||(encryptkey&&*encryptkey))
  2303. destination->setCompressed(true);
  2304. const char * decryptkey = req.getDecrypt();
  2305. if ((encryptkey&&*encryptkey)||(decryptkey&&*decryptkey))
  2306. options->setEncDec(encryptkey,decryptkey);
  2307. resp.setWuid(wu->queryId());
  2308. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2309. submitDFUWorkUnit(wu.getClear());
  2310. }
  2311. catch(IException* e)
  2312. {
  2313. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2314. }
  2315. return true;
  2316. }
  2317. bool CFileSprayEx::onCopy(IEspContext &context, IEspCopy &req, IEspCopyResponse &resp)
  2318. {
  2319. try
  2320. {
  2321. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Write, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do Copy. Permission denied.");
  2322. const char* srcname = req.getSourceLogicalName();
  2323. const char* dstname = req.getDestLogicalName();
  2324. if(!srcname || !*srcname)
  2325. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Source logical file not specified.");
  2326. if(!dstname || !*dstname)
  2327. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Destination logical file not specified.");
  2328. PROGLOG("Copy from %s to %s", srcname, dstname);
  2329. StringBuffer destFolder, destTitle, defaultFolder, defaultReplicateFolder;
  2330. StringBuffer srcNodeGroup, destNodeGroup;
  2331. bool bRoxie = false;
  2332. const char* destNodeGroupReq = req.getDestGroup();
  2333. if(!destNodeGroupReq || !*destNodeGroupReq)
  2334. {
  2335. getNodeGroupFromLFN(context, srcname, destNodeGroup);
  2336. DBGLOG("Destination node group not specified, using source node group %s", destNodeGroup.str());
  2337. }
  2338. else
  2339. {
  2340. destNodeGroup = destNodeGroupReq;
  2341. const char* destRoxie = req.getDestGroupRoxie();
  2342. if (destRoxie && !stricmp(destRoxie, "Yes"))
  2343. {
  2344. bRoxie = true;
  2345. }
  2346. }
  2347. CDfsLogicalFileName lfn; // NOTE: must not be moved into block below, or dstname will point to invalid memory
  2348. if (!bRoxie)
  2349. {
  2350. if (!lfn.setValidate(dstname))
  2351. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid destination filename:'%s'", dstname);
  2352. dstname = lfn.get();
  2353. }
  2354. ParseLogicalPath(dstname, destNodeGroup.str(), NULL, destFolder, destTitle, defaultFolder, defaultReplicateFolder);
  2355. StringBuffer fileMask;
  2356. constructFileMask(destTitle.str(), fileMask);
  2357. Owned<IUserDescriptor> udesc=createUserDescriptor();
  2358. const char* srcDali = req.getSourceDali();
  2359. const char* srcu = req.getSrcusername();
  2360. if (!isEmptyString(srcDali) && !isEmptyString(srcu))
  2361. {
  2362. udesc->set(srcu, req.getSrcpassword());
  2363. }
  2364. else
  2365. {
  2366. StringBuffer user, passwd;
  2367. context.getUserID(user);
  2368. context.getPassword(passwd);
  2369. udesc->set(user, passwd);
  2370. }
  2371. CDfsLogicalFileName logicalName;
  2372. logicalName.set(srcname);
  2373. if (!isEmptyString(srcDali))
  2374. {
  2375. SocketEndpoint ep(srcDali);
  2376. if (ep.isNull())
  2377. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Copy %s: cannot resolve SourceDali network IP from %s.", srcname, srcDali);
  2378. logicalName.setForeign(ep,false);
  2379. }
  2380. Owned<IDistributedFile> file = queryDistributedFileDirectory().lookup(logicalName, udesc, false, false, false, nullptr, defaultPrivilegedUser);
  2381. if (!file)
  2382. throw MakeStringException(ECLWATCH_FILE_NOT_EXIST, "Failed to find file: %s", logicalName.get());
  2383. bool supercopy = req.getSuperCopy();
  2384. if (supercopy)
  2385. {
  2386. if (!file->querySuperFile())
  2387. supercopy = false;
  2388. }
  2389. else if (file->querySuperFile() && (file->querySuperFile()->numSubFiles() > 1) && isFileKey(file))
  2390. supercopy = true;
  2391. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2392. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  2393. wu->setJobName(dstname);
  2394. setDFUServerQueueReq(req.getDFUServerQueue(), wu);
  2395. setUserAuth(context, wu);
  2396. if(destNodeGroup.length() > 0)
  2397. wu->setClusterName(destNodeGroup.str());
  2398. if (supercopy)
  2399. wu->setCommand(DFUcmd_supercopy);
  2400. else
  2401. wu->setCommand(DFUcmd_copy);
  2402. IDFUfileSpec *wuFSpecSource = wu->queryUpdateSource();
  2403. IDFUfileSpec *wuFSpecDest = wu->queryUpdateDestination();
  2404. IDFUoptions *wuOptions = wu->queryUpdateOptions();
  2405. wuFSpecSource->setLogicalName(srcname);
  2406. if(srcDali && *srcDali)
  2407. {
  2408. SocketEndpoint ep(srcDali);
  2409. wuFSpecSource->setForeignDali(ep);
  2410. const char* srcusername = req.getSrcusername();
  2411. if(srcusername && *srcusername)
  2412. {
  2413. const char* srcpasswd = req.getSrcpassword();
  2414. wuFSpecSource->setForeignUser(srcusername, srcpasswd);
  2415. }
  2416. }
  2417. wuFSpecDest->setLogicalName(dstname);
  2418. wuFSpecDest->setFileMask(fileMask.str());
  2419. wuOptions->setOverwrite(req.getOverwrite());
  2420. wuOptions->setPreserveCompression(req.getPreserveCompression());
  2421. if (!req.getExpireDays_isNull())
  2422. wuOptions->setExpireDays(req.getExpireDays());
  2423. if(req.getNosplit())
  2424. wuOptions->setNoSplit(true);
  2425. if (!req.getNoCommon_isNull())
  2426. wuOptions->setNoCommon(req.getNoCommon());
  2427. if (bRoxie)
  2428. {
  2429. setRoxieClusterPartDiskMapping(destNodeGroup.str(), defaultFolder.str(), defaultReplicateFolder.str(), supercopy, wuFSpecDest, wuOptions);
  2430. wuFSpecDest->setWrap(true); // roxie always wraps
  2431. if(req.getCompress())
  2432. wuFSpecDest->setCompressed(true);
  2433. if (!supercopy)
  2434. wuOptions->setSuppressNonKeyRepeats(true); // **** only repeat last part when src kind = key
  2435. }
  2436. else
  2437. {
  2438. const char* srcDiffKeyName = req.getSourceDiffKeyName();
  2439. const char* destDiffKeyName = req.getDestDiffKeyName();
  2440. if (srcDiffKeyName&&*srcDiffKeyName)
  2441. wuFSpecSource->setDiffKey(srcDiffKeyName);
  2442. if (destDiffKeyName&&*destDiffKeyName)
  2443. wuFSpecDest->setDiffKey(destDiffKeyName);
  2444. wuFSpecDest->setDirectory(destFolder.str());
  2445. wuFSpecDest->setGroupName(destNodeGroup.str());
  2446. wuFSpecDest->setWrap(req.getWrap());
  2447. const char * encryptkey = req.getEncrypt();
  2448. if(req.getCompress()||(encryptkey&&*encryptkey))
  2449. wuFSpecDest->setCompressed(true);
  2450. wuOptions->setReplicate(req.getReplicate());
  2451. const char * decryptkey = req.getDecrypt();
  2452. if ((encryptkey&&*encryptkey)||(decryptkey&&*decryptkey))
  2453. wuOptions->setEncDec(encryptkey,decryptkey);
  2454. if(req.getNorecover())
  2455. wuOptions->setNoRecover(true);
  2456. if(!req.getNosplit_isNull())
  2457. wuOptions->setNoSplit(req.getNosplit());
  2458. if(req.getMaxConnections() > 0)
  2459. wuOptions->setmaxConnections(req.getMaxConnections());
  2460. if(req.getThrottle() > 0)
  2461. wuOptions->setThrottle(req.getThrottle());
  2462. if(req.getTransferBufferSize() > 0)
  2463. wuOptions->setTransferBufferSize(req.getTransferBufferSize());
  2464. if (req.getPull())
  2465. wuOptions->setPull(true);
  2466. if (req.getPush())
  2467. wuOptions->setPush(true);
  2468. if (req.getIfnewer())
  2469. wuOptions->setIfNewer(true);
  2470. if (req.getNosplit())
  2471. wuOptions->setNoSplit(true);
  2472. ClusterPartDiskMapSpec mspec;
  2473. wuFSpecDest->getClusterPartDiskMapSpec(destNodeGroup.str(), mspec);
  2474. mspec.setDefaultBaseDir(defaultFolder.str());
  2475. mspec.setDefaultReplicateDir(defaultReplicateFolder.str());
  2476. wuFSpecDest->setClusterPartDiskMapSpec(destNodeGroup.str(), mspec);
  2477. }
  2478. resp.setResult(wu->queryId());
  2479. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2480. submitDFUWorkUnit(wu.getClear());
  2481. }
  2482. catch(IException* e)
  2483. {
  2484. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2485. }
  2486. return true;
  2487. }
  2488. bool CFileSprayEx::onRename(IEspContext &context, IEspRename &req, IEspRenameResponse &resp)
  2489. {
  2490. try
  2491. {
  2492. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Write, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do Rename. Permission denied.");
  2493. const char* srcname = req.getSrcname();
  2494. const char* dstname = req.getDstname();
  2495. if(!srcname || !*srcname)
  2496. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Source logical file not specified.");
  2497. if(!dstname || !*dstname)
  2498. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Destination logical file not specified.");
  2499. PROGLOG("Rename from %s to %s", srcname, dstname);
  2500. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2501. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  2502. StringBuffer destTitle;
  2503. ParseLogicalPath(req.getDstname(), destTitle);
  2504. wu->setJobName(destTitle.str());
  2505. setDFUServerQueueReq(req.getDFUServerQueue(), wu);
  2506. setUserAuth(context, wu);
  2507. wu->setCommand(DFUcmd_rename);
  2508. #if 0 // TBD - Handling for multiple clusters? the cluster should be specified by user if needed
  2509. Owned<IUserDescriptor> udesc;
  2510. if(user.length() > 0)
  2511. {
  2512. const char* passwd = context.queryPassword();
  2513. udesc.setown(createUserDescriptor());
  2514. udesc->set(user.str(), passwd);
  2515. Owned<IDistributedFile> df = queryDistributedFileDirectory().lookup(srcname, udesc);
  2516. if(df)
  2517. {
  2518. StringBuffer cluster0;
  2519. df->getClusterName(0,cluster0); // TBD - Handling for multiple clusters?
  2520. if (cluster0.length()!=0)
  2521. {
  2522. wu->setClusterName(cluster0.str());
  2523. }
  2524. else
  2525. {
  2526. const char *cluster = df->queryAttributes().queryProp("@group");
  2527. if (cluster && *cluster)
  2528. {
  2529. wu->setClusterName(cluster);
  2530. }
  2531. }
  2532. }
  2533. }
  2534. #endif
  2535. IDFUfileSpec *source = wu->queryUpdateSource();
  2536. source->setLogicalName(srcname);
  2537. IDFUfileSpec *destination = wu->queryUpdateDestination();
  2538. destination->setLogicalName(dstname);
  2539. IDFUoptions *options = wu->queryUpdateOptions();
  2540. options->setOverwrite(req.getOverwrite());
  2541. resp.setWuid(wu->queryId());
  2542. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2543. submitDFUWorkUnit(wu.getClear());
  2544. }
  2545. catch(IException* e)
  2546. {
  2547. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2548. }
  2549. return true;
  2550. }
  2551. bool CFileSprayEx::onDFUWUFile(IEspContext &context, IEspDFUWUFileRequest &req, IEspDFUWUFileResponse &resp)
  2552. {
  2553. try
  2554. {
  2555. context.ensureFeatureAccess(DFU_WU_URL, SecAccess_Read, ECLWATCH_DFU_WU_ACCESS_DENIED, "Access to DFU workunit is denied.");
  2556. if (*req.getWuid())
  2557. {
  2558. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2559. Owned<IConstDFUWorkUnit> wu = factory->openWorkUnit(req.getWuid(), false);
  2560. if(!wu)
  2561. throw MakeStringException(ECLWATCH_CANNOT_OPEN_WORKUNIT, "Dfu workunit %s not found.", req.getWuid());
  2562. PROGLOG("DFUWUFile: %s", req.getWuid());
  2563. StringBuffer xmlbuf;
  2564. xmlbuf.append("<?xml version=\"1.0\" encoding=\"UTF-8\"?>");
  2565. const char* plainText = req.getPlainText();
  2566. if (plainText && (!stricmp(plainText, "yes")))
  2567. {
  2568. wu->toXML(xmlbuf);
  2569. resp.setFile(xmlbuf.str());
  2570. resp.setFile_mimetype(HTTP_TYPE_TEXT_PLAIN);
  2571. }
  2572. else
  2573. {
  2574. xmlbuf.append("<?xml-stylesheet href=\"../esp/xslt/xmlformatter.xsl\" type=\"text/xsl\"?>");
  2575. wu->toXML(xmlbuf);
  2576. resp.setFile(xmlbuf.str());
  2577. resp.setFile_mimetype(HTTP_TYPE_APPLICATION_XML);
  2578. }
  2579. }
  2580. }
  2581. catch(IException* e)
  2582. {
  2583. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2584. }
  2585. return true;
  2586. }
  2587. bool CFileSprayEx::onFileList(IEspContext &context, IEspFileListRequest &req, IEspFileListResponse &resp)
  2588. {
  2589. try
  2590. {
  2591. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do FileList. Permission denied.");
  2592. const char* path = req.getPath();
  2593. if (!path || !*path)
  2594. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Path not specified.");
  2595. double version = context.getClientVersion();
  2596. const char* netaddr = req.getNetaddr();
  2597. if (!netaddr || !*netaddr)
  2598. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Network address not specified.");
  2599. const char* fileNameMask = req.getMask();
  2600. bool directoryOnly = req.getDirectoryOnly();
  2601. PROGLOG("FileList: Netaddr %s, Path %s", netaddr, path);
  2602. StringBuffer sPath(path);
  2603. const char* osStr = req.getOS();
  2604. if (osStr && *osStr)
  2605. {
  2606. int os = atoi(osStr);
  2607. const char pathSep = (os == OS_WINDOWS) ? '\\' : '/';
  2608. sPath.replace(pathSep=='\\'?'/':'\\', pathSep);
  2609. if (*(sPath.str() + sPath.length() -1) != pathSep)
  2610. sPath.append( pathSep );
  2611. }
  2612. if (!isEmptyString(fileNameMask))
  2613. {
  2614. const char* ext = pathExtension(sPath.str());
  2615. if (ext && !strieq(ext, "cfg") && !strieq(ext, "log"))
  2616. throw MakeStringException(ECLWATCH_ACCESS_TO_FILE_DENIED, "Only cfg or log file allowed.");
  2617. }
  2618. if (!validateDropZonePath(nullptr, netaddr, sPath) && !validateConfigurationDirectory(nullptr, "log", nullptr, nullptr, sPath)) //The path should be the absolute path for the dropzone or log file.
  2619. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Invalid file path %s", sPath.str());
  2620. RemoteFilename rfn;
  2621. SocketEndpoint ep;
  2622. #ifdef MACHINE_IP
  2623. ep.set(MACHINE_IP);
  2624. #else
  2625. ep.set(netaddr);
  2626. if (ep.isNull())
  2627. throw MakeStringException(ECLWATCH_INVALID_INPUT, "FileList: cannot resolve network IP from %s.", netaddr);
  2628. #endif
  2629. rfn.setPath(ep, sPath.str());
  2630. Owned<IFile> f = createIFile(rfn);
  2631. if (f->isDirectory()!=fileBool::foundYes)
  2632. throw MakeStringException(ECLWATCH_INVALID_DIRECTORY, "%s is not a directory.", path);
  2633. IArrayOf<IEspPhysicalFileStruct> files;
  2634. Owned<IDirectoryIterator> di = f->directoryFiles(NULL, false, true);
  2635. if(di.get() != NULL)
  2636. {
  2637. ForEach(*di)
  2638. {
  2639. StringBuffer fname;
  2640. di->getName(fname);
  2641. if (fname.length() == 0 || (directoryOnly && !di->isDir()) || (!di->isDir() && !isEmptyString(fileNameMask) && !WildMatch(fname.str(), fileNameMask, true)))
  2642. continue;
  2643. Owned<IEspPhysicalFileStruct> onefile = createPhysicalFileStruct();
  2644. onefile->setName(fname.str());
  2645. onefile->setIsDir(di->isDir());
  2646. onefile->setFilesize(di->getFileSize());
  2647. CDateTime modtime;
  2648. StringBuffer timestr;
  2649. di->getModifiedTime(modtime);
  2650. unsigned y,m,d,h,min,sec,nsec;
  2651. modtime.getDate(y,m,d,true);
  2652. modtime.getTime(h,min,sec,nsec,true);
  2653. timestr.appendf("%04d-%02d-%02d %02d:%02d:%02d", y,m,d,h,min,sec);
  2654. onefile->setModifiedtime(timestr.str());
  2655. files.append(*onefile.getLink());
  2656. }
  2657. }
  2658. sPath.replace('\\', '/');//XSLT cannot handle backslashes
  2659. resp.setPath(sPath);
  2660. resp.setFiles(files);
  2661. resp.setNetaddr(netaddr);
  2662. if (osStr && *osStr)
  2663. {
  2664. int os = atoi(osStr);
  2665. resp.setOS(os);
  2666. }
  2667. if (!isEmptyString(fileNameMask))
  2668. resp.setMask(fileNameMask);
  2669. if (version >= 1.10)
  2670. {
  2671. StringBuffer acceptLanguage;
  2672. resp.setAcceptLanguage(getAcceptLanguage(context, acceptLanguage).str());
  2673. }
  2674. resp.setDirectoryOnly(directoryOnly);
  2675. }
  2676. catch(IException* e)
  2677. {
  2678. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2679. }
  2680. return true;
  2681. }
  2682. bool CFileSprayEx::checkDropZoneIPAndPath(double clientVersion, const char* dropZoneName, const char* netAddr, const char* path)
  2683. {
  2684. if (isEmptyString(netAddr) || isEmptyString(path))
  2685. throw MakeStringException(ECLWATCH_INVALID_INPUT, "NetworkAddress or Path not defined.");
  2686. bool isIPAddressReq = isIPAddress(netAddr);
  2687. IArrayOf<IConstTpDropZone> allTpDropZones;
  2688. CTpWrapper tpWrapper;
  2689. tpWrapper.getTpDropZones(9999, nullptr, false, allTpDropZones); //version 9999: get the latest information about dropzone
  2690. ForEachItemIn(i, allTpDropZones)
  2691. {
  2692. IConstTpDropZone& dropZone = allTpDropZones.item(i);
  2693. if (!isEmptyString(dropZoneName) && !streq(dropZoneName, dropZone.getName()))
  2694. continue;
  2695. const char* prefix = dropZone.getPath();
  2696. if (strnicmp(path, prefix, strlen(prefix)) != 0)
  2697. continue;
  2698. IArrayOf<IConstTpMachine>& tpMachines = dropZone.getTpMachines();
  2699. ForEachItemIn(ii, tpMachines)
  2700. {
  2701. if (matchNetAddressRequest(netAddr, isIPAddressReq, tpMachines.item(ii)))
  2702. return true;
  2703. }
  2704. }
  2705. return false;
  2706. }
  2707. void CFileSprayEx::addDropZoneFile(IEspContext& context, IDirectoryIterator* di, const char* name, const char pathSep, IArrayOf<IEspPhysicalFileStruct>& files)
  2708. {
  2709. Owned<IEspPhysicalFileStruct> aFile = createPhysicalFileStruct();
  2710. const char* pName = strrchr(name, pathSep);
  2711. if (!pName)
  2712. aFile->setName(name);
  2713. else
  2714. {
  2715. StringBuffer sPath;
  2716. sPath.append(pName - name, name);
  2717. aFile->setPath(sPath.str());
  2718. pName++; //skip the PathSepChar
  2719. aFile->setName(pName);
  2720. }
  2721. aFile->setIsDir(di->isDir());
  2722. CDateTime modtime;
  2723. StringBuffer timestr;
  2724. di->getModifiedTime(modtime);
  2725. unsigned y,m,d,h,min,sec,nsec;
  2726. modtime.getDate(y,m,d,true);
  2727. modtime.getTime(h,min,sec,nsec,true);
  2728. timestr.appendf("%04d-%02d-%02d %02d:%02d:%02d", y,m,d,h,min,sec);
  2729. aFile->setModifiedtime(timestr.str());
  2730. aFile->setFilesize(di->getFileSize());
  2731. files.append(*aFile.getLink());
  2732. }
  2733. void CFileSprayEx::searchDropZoneFiles(IEspContext& context, IpAddress& ip, const char* dir, const char* nameFilter, IArrayOf<IEspPhysicalFileStruct>& files, unsigned& filesFound)
  2734. {
  2735. RemoteFilename rfn;
  2736. SocketEndpoint ep;
  2737. ep.ipset(ip);
  2738. rfn.setPath(ep, dir);
  2739. Owned<IFile> f = createIFile(rfn);
  2740. if(f->isDirectory()!=fileBool::foundYes)
  2741. throw MakeStringException(ECLWATCH_INVALID_DIRECTORY, "%s is not a directory.", dir);
  2742. const char pathSep = getPathSepChar(dir);
  2743. Owned<IDirectoryIterator> di = f->directoryFiles(nameFilter, true, true);
  2744. ForEach(*di)
  2745. {
  2746. StringBuffer fname;
  2747. di->getName(fname);
  2748. if (!fname.length())
  2749. continue;
  2750. filesFound++;
  2751. if (filesFound > dropZoneFileSearchMaxFiles)
  2752. break;
  2753. addDropZoneFile(context, di, fname.str(), pathSep, files);
  2754. }
  2755. }
  2756. bool CFileSprayEx::onDropZoneFileSearch(IEspContext &context, IEspDropZoneFileSearchRequest &req, IEspDropZoneFileSearchResponse &resp)
  2757. {
  2758. try
  2759. {
  2760. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Access, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do FileList. Permission denied.");
  2761. const char* dropZoneName = req.getDropZoneName();
  2762. if (isEmptyString(dropZoneName))
  2763. throw MakeStringException(ECLWATCH_INVALID_INPUT, "DropZone not specified.");
  2764. const char* dropZoneServerReq = req.getServer(); //IP or hostname
  2765. if (isEmptyString(dropZoneServerReq))
  2766. throw MakeStringException(ECLWATCH_INVALID_INPUT, "DropZone server not specified.");
  2767. const char* nameFilter = req.getNameFilter();
  2768. if (isEmptyString(nameFilter))
  2769. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Name Filter not specified.");
  2770. bool validNameFilter = false;
  2771. const char* pNameFilter = nameFilter;
  2772. while (!isEmptyString(pNameFilter))
  2773. {
  2774. if (*pNameFilter != '*')
  2775. {
  2776. validNameFilter = true;
  2777. break;
  2778. }
  2779. pNameFilter++;
  2780. }
  2781. if (!validNameFilter)
  2782. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Invalid Name Filter '*'");
  2783. double version = context.getClientVersion();
  2784. bool serverFound = false;
  2785. unsigned filesFound = 0;
  2786. IArrayOf<IEspPhysicalFileStruct> files;
  2787. bool isIPAddressReq = isIPAddress(dropZoneServerReq);
  2788. IArrayOf<IConstTpDropZone> allTpDropZones;
  2789. CTpWrapper tpWrapper;
  2790. tpWrapper.getTpDropZones(9999, nullptr, false, allTpDropZones); //version 9999: get the latest information about dropzone
  2791. ForEachItemIn(i, allTpDropZones)
  2792. {
  2793. IConstTpDropZone& dropZone = allTpDropZones.item(i);
  2794. const char* name = dropZone.getName();
  2795. if (!streq(dropZoneName, name))
  2796. continue;
  2797. if (req.getECLWatchVisibleOnly() && !dropZone.getECLWatchVisible())
  2798. continue;
  2799. IArrayOf<IConstTpMachine>& tpMachines = dropZone.getTpMachines();
  2800. ForEachItemIn(ii, tpMachines)
  2801. {
  2802. IConstTpMachine& tpMachine = tpMachines.item(ii);
  2803. if (!matchNetAddressRequest(dropZoneServerReq, isIPAddressReq, tpMachine))
  2804. continue;
  2805. IpAddress ipAddress;
  2806. ipAddress.ipset(tpMachine.getNetaddress());
  2807. searchDropZoneFiles(context, ipAddress, dropZone.getPath(), nameFilter, files, filesFound);
  2808. serverFound = true;
  2809. }
  2810. }
  2811. if (!serverFound)
  2812. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Server %s not found in dropzone %s.", dropZoneServerReq, dropZoneName);
  2813. if ((version >= 1.16) && (filesFound > dropZoneFileSearchMaxFiles))
  2814. {
  2815. VStringBuffer msg("More than %u files are found. Only %u files are returned.", dropZoneFileSearchMaxFiles, dropZoneFileSearchMaxFiles);
  2816. resp.setWarning(msg.str());
  2817. }
  2818. resp.setFiles(files);
  2819. }
  2820. catch(IException* e)
  2821. {
  2822. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2823. }
  2824. return true;
  2825. }
  2826. bool CFileSprayEx::onDfuMonitor(IEspContext &context, IEspDfuMonitorRequest &req, IEspDfuMonitorResponse &resp)
  2827. {
  2828. try
  2829. {
  2830. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Failed to do DfuMonitor. Permission denied.");
  2831. Owned<IDFUWorkUnitFactory> factory = getDFUWorkUnitFactory();
  2832. Owned<IDFUWorkUnit> wu = factory->createWorkUnit();
  2833. wu->setQueue(m_MonitorQueueLabel.str());
  2834. StringBuffer user, passwd;
  2835. wu->setUser(context.getUserID(user).str());
  2836. wu->setPassword(context.getPassword(passwd).str());
  2837. wu->setCommand(DFUcmd_monitor);
  2838. IDFUmonitor *monitor = wu->queryUpdateMonitor();
  2839. IDFUfileSpec *source = wu->queryUpdateSource();
  2840. const char *eventname = req.getEventName();
  2841. const char *lname = req.getLogicalName();
  2842. if (lname&&*lname)
  2843. source->setLogicalName(lname);
  2844. else {
  2845. const char *ip = req.getIp();
  2846. const char *filename = req.getFilename();
  2847. if (filename&&*filename) {
  2848. RemoteFilename rfn;
  2849. if (ip&&*ip) {
  2850. SocketEndpoint ep(ip);
  2851. if (ep.isNull())
  2852. throw MakeStringException(ECLWATCH_INVALID_INPUT, "DfuMonitor: cannot resolve network IP from %s.", ip);
  2853. rfn.setPath(ep,filename);
  2854. }
  2855. else
  2856. rfn.setRemotePath(filename);
  2857. source->setSingleFilename(rfn);
  2858. }
  2859. else
  2860. throw MakeStringException(ECLWATCH_INVALID_INPUT, "Neither logical name nor network ip/file specified for monitor.");
  2861. }
  2862. if (eventname)
  2863. monitor->setEventName(eventname);
  2864. monitor->setShotLimit(req.getShotLimit());
  2865. monitor->setSub(req.getSub());
  2866. resp.setWuid(wu->queryId());
  2867. resp.setRedirectUrl(StringBuffer("/FileSpray/GetDFUWorkunit?wuid=").append(wu->queryId()).str());
  2868. submitDFUWorkUnit(wu.getClear());
  2869. }
  2870. catch(IException* e)
  2871. {
  2872. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2873. }
  2874. return true;
  2875. }
  2876. bool CFileSprayEx::onOpenSave(IEspContext &context, IEspOpenSaveRequest &req, IEspOpenSaveResponse &resp)
  2877. {
  2878. try
  2879. {
  2880. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Permission denied.");
  2881. const char* location = req.getLocation();
  2882. const char* path = req.getPath();
  2883. const char* name = req.getName();
  2884. const char* type = req.getType();
  2885. const char* dateTime = req.getDateTime();
  2886. if (location && *location)
  2887. resp.setLocation(location);
  2888. if (path && *path)
  2889. resp.setPath(path);
  2890. if (name && *name)
  2891. resp.setName(name);
  2892. if (type && *type)
  2893. resp.setType(type);
  2894. if (dateTime && *dateTime)
  2895. resp.setDateTime(dateTime);
  2896. if (req.getBinaryFile())
  2897. resp.setViewable(false);
  2898. }
  2899. catch(IException* e)
  2900. {
  2901. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  2902. }
  2903. return true;
  2904. }
  2905. bool CFileSprayEx::getDropZoneFiles(IEspContext &context, const char* dropZone, const char* netaddr, const char* path,
  2906. IEspDropZoneFilesRequest &req, IEspDropZoneFilesResponse &resp)
  2907. {
  2908. if (!checkDropZoneIPAndPath(context.getClientVersion(), dropZone, netaddr, path))
  2909. throw MakeStringException(ECLWATCH_DROP_ZONE_NOT_FOUND, "Dropzone is not found in the environment settings.");
  2910. bool directoryOnly = req.getDirectoryOnly();
  2911. RemoteFilename rfn;
  2912. SocketEndpoint ep;
  2913. #ifdef MACHINE_IP
  2914. ep.set(MACHINE_IP);
  2915. #else
  2916. ep.set(netaddr);
  2917. if (ep.isNull())
  2918. throw MakeStringException(ECLWATCH_INVALID_INPUT, "CFileSprayEx::getDropZoneFiles: cannot resolve network IP from %s.", netaddr);
  2919. #endif
  2920. rfn.setPath(ep, path);
  2921. Owned<IFile> f = createIFile(rfn);
  2922. if(f->isDirectory()!=fileBool::foundYes)
  2923. throw MakeStringException(ECLWATCH_INVALID_DIRECTORY, "%s is not a directory.", path);
  2924. IArrayOf<IEspPhysicalFileStruct> files;
  2925. Owned<IDirectoryIterator> di = f->directoryFiles(NULL, false, true);
  2926. if(di.get() != NULL)
  2927. {
  2928. ForEach(*di)
  2929. {
  2930. StringBuffer fname;
  2931. di->getName(fname);
  2932. if (fname.length() == 0 || (directoryOnly && !di->isDir()))
  2933. continue;
  2934. Owned<IEspPhysicalFileStruct> onefile = createPhysicalFileStruct();
  2935. onefile->setName(fname.str());
  2936. onefile->setIsDir(di->isDir());
  2937. onefile->setFilesize(di->getFileSize());
  2938. CDateTime modtime;
  2939. StringBuffer timestr;
  2940. di->getModifiedTime(modtime);
  2941. unsigned y,m,d,h,min,sec,nsec;
  2942. modtime.getDate(y,m,d,true);
  2943. modtime.getTime(h,min,sec,nsec,true);
  2944. timestr.appendf("%04d-%02d-%02d %02d:%02d:%02d", y,m,d,h,min,sec);
  2945. onefile->setModifiedtime(timestr.str());
  2946. files.append(*onefile.getLink());
  2947. }
  2948. }
  2949. resp.setFiles(files);
  2950. return true;
  2951. }
  2952. //This method returns all dropzones and, if NetAddress and Path specified, returns filtered list of files.
  2953. bool CFileSprayEx::onDropZoneFiles(IEspContext &context, IEspDropZoneFilesRequest &req, IEspDropZoneFilesResponse &resp)
  2954. {
  2955. try
  2956. {
  2957. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Permission denied.");
  2958. const char* netAddress = req.getNetAddress();
  2959. if (!isEmptyString(netAddress))
  2960. {
  2961. IpAddress ipToCheck;
  2962. ipToCheck.ipset(netAddress);
  2963. if (ipToCheck.isNull())
  2964. throw makeStringExceptionV(ECLWATCH_INVALID_INPUT, "Invalid server %s specified.", netAddress);
  2965. }
  2966. bool filesFromALinux = false;
  2967. IArrayOf<IEspDropZone> dropZoneList;
  2968. bool ECLWatchVisibleOnly = req.getECLWatchVisibleOnly();
  2969. bool isIPAddressReq = !isEmptyString(netAddress) && isIPAddress(netAddress);
  2970. IArrayOf<IConstTpDropZone> allTpDropZones;
  2971. CTpWrapper tpWrapper;
  2972. tpWrapper.getTpDropZones(9999, nullptr, false, allTpDropZones); //version 9999: get the latest information about dropzone
  2973. ForEachItemIn(i, allTpDropZones)
  2974. {
  2975. IConstTpDropZone& dropZone = allTpDropZones.item(i);
  2976. if (ECLWatchVisibleOnly && !dropZone.getECLWatchVisible())
  2977. continue;
  2978. const char* dropZoneName = dropZone.getName();
  2979. const char* prefix = dropZone.getPath();
  2980. bool isLinux = getPathSepChar(prefix) == '/' ? true : false;
  2981. IArrayOf<IConstTpMachine>& tpMachines = dropZone.getTpMachines();
  2982. ForEachItemIn(ii, tpMachines)
  2983. {
  2984. const char* machineName = tpMachines.item(ii).getName();
  2985. const char* netAddr = tpMachines.item(ii).getNetaddress();
  2986. if (isEmptyString(machineName) || isEmptyString(netAddr))
  2987. continue;
  2988. Owned<IEspDropZone> dropZone = createDropZone();
  2989. dropZone->setName(dropZoneName);
  2990. dropZone->setPath(prefix);
  2991. dropZone->setComputer(machineName);
  2992. dropZone->setNetAddress(netAddr);
  2993. if (isLinux)
  2994. dropZone->setLinux("true");
  2995. dropZoneList.append(*dropZone.getClear());
  2996. if (!isEmptyString(netAddress) && matchNetAddressRequest(netAddress, isIPAddressReq, tpMachines.item(ii)))
  2997. filesFromALinux = isLinux;
  2998. }
  2999. }
  3000. if (dropZoneList.ordinality())
  3001. resp.setDropZones(dropZoneList);
  3002. const char* dzName = req.getDropZoneName();
  3003. const char* directory = req.getPath();
  3004. const char* subfolder = req.getSubfolder();
  3005. if (isEmptyString(netAddress) || (isEmptyString(directory) && isEmptyString(subfolder)))
  3006. return true;
  3007. StringBuffer netAddressStr, directoryStr, osStr;
  3008. netAddressStr.set(netAddress);
  3009. if (!isEmptyString(directory))
  3010. directoryStr.set(directory);
  3011. if (!isEmptyString(subfolder))
  3012. {
  3013. if (directoryStr.length())
  3014. addPathSepChar(directoryStr);
  3015. directoryStr.append(subfolder);
  3016. }
  3017. addPathSepChar(directoryStr);
  3018. getDropZoneFiles(context, dzName, netAddress, directoryStr.str(), req, resp);
  3019. resp.setDropZoneName(dzName);
  3020. resp.setNetAddress(netAddress);
  3021. resp.setPath(directoryStr.str());
  3022. resp.setOS(filesFromALinux);
  3023. resp.setECLWatchVisibleOnly(ECLWatchVisibleOnly);
  3024. }
  3025. catch(IException* e)
  3026. {
  3027. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  3028. }
  3029. return true;
  3030. }
  3031. bool CFileSprayEx::onDeleteDropZoneFiles(IEspContext &context, IEspDeleteDropZoneFilesRequest &req, IEspDFUWorkunitsActionResponse &resp)
  3032. {
  3033. try
  3034. {
  3035. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Full, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Permission denied.");
  3036. double version = context.getClientVersion();
  3037. const char* dzName = req.getDropZoneName();
  3038. const char* netAddress = req.getNetAddress();
  3039. const char* directory = req.getPath();
  3040. const char* osStr = req.getOS();
  3041. StringArray & files = req.getNames();
  3042. if (!files.ordinality())
  3043. throw MakeStringException(ECLWATCH_INVALID_INPUT, "File not specified.");
  3044. StringBuffer path(directory);
  3045. if (!isEmptyString(osStr))
  3046. {
  3047. char pathSep = (atoi(osStr) == OS_WINDOWS) ? '\\' : '/';
  3048. path.replace(pathSep=='\\' ? '/' : '\\', pathSep);
  3049. }
  3050. addPathSepChar(path, getPathSepChar(path.str()));
  3051. if (!checkDropZoneIPAndPath(version, dzName, netAddress, path.str()))
  3052. throw MakeStringException(ECLWATCH_DROP_ZONE_NOT_FOUND, "Dropzone is not found in the environment settings.");
  3053. RemoteFilename rfn;
  3054. SocketEndpoint ep(netAddress);
  3055. if (ep.isNull())
  3056. throw MakeStringException(ECLWATCH_INVALID_INPUT, "DeleteDropZoneFiles: cannot resolve network IP from %s.", netAddress);
  3057. rfn.setPath(ep, path.str());
  3058. Owned<IFile> f = createIFile(rfn);
  3059. if(f->isDirectory()!=fileBool::foundYes)
  3060. throw MakeStringException(ECLWATCH_INVALID_DIRECTORY, "%s is not a directory.", directory);
  3061. bool bAllSuccess = true;
  3062. IArrayOf<IEspDFUActionResult> results;
  3063. for(unsigned i = 0; i < files.ordinality(); ++i)
  3064. {
  3065. const char* file = files.item(i);
  3066. if (!file || !*file)
  3067. continue;
  3068. PROGLOG("DeleteDropZoneFiles: netAddress %s, path %s, file %s", netAddress, directory, file);
  3069. Owned<IEspDFUActionResult> res = createDFUActionResult("", "");
  3070. res->setID(files.item(i));
  3071. res->setAction("Delete");
  3072. res->setResult("Success");
  3073. try
  3074. {
  3075. StringBuffer fileToDelete(path);
  3076. fileToDelete.append(file);
  3077. rfn.setPath(ep, fileToDelete.str());
  3078. Owned<IFile> rFile = createIFile(rfn);
  3079. if (!rFile->exists())
  3080. res->setResult("Warning: this file does not exist.");
  3081. else
  3082. rFile->remove();
  3083. }
  3084. catch (IException *e)
  3085. {
  3086. bAllSuccess = false;
  3087. StringBuffer eMsg;
  3088. eMsg = e->errorMessage(eMsg);
  3089. e->Release();
  3090. StringBuffer failedMsg("Failed: ");
  3091. failedMsg.append(eMsg);
  3092. res->setResult(failedMsg.str());
  3093. }
  3094. results.append(*res.getLink());
  3095. }
  3096. resp.setFirstColumn("File");
  3097. resp.setDFUActionResults(results);
  3098. }
  3099. catch(IException* e)
  3100. {
  3101. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  3102. }
  3103. return true;
  3104. }
  3105. void CFileSprayEx::appendGroupNode(IArrayOf<IEspGroupNode>& groupNodes, const char* nodeName, const char* clusterType,
  3106. bool replicateOutputs)
  3107. {
  3108. Owned<IEspGroupNode> node = createGroupNode();
  3109. node->setName(nodeName);
  3110. node->setClusterType(clusterType);
  3111. if (replicateOutputs)
  3112. node->setReplicateOutputs(replicateOutputs);
  3113. groupNodes.append(*node.getClear());
  3114. }
  3115. bool CFileSprayEx::onGetSprayTargets(IEspContext &context, IEspGetSprayTargetsRequest &req, IEspGetSprayTargetsResponse &resp)
  3116. {
  3117. try
  3118. {
  3119. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Permission denied.");
  3120. #ifdef _CONTAINERIZED
  3121. IArrayOf<IEspGroupNode> sprayTargets;
  3122. Owned<IPropertyTreeIterator> dataPlanes = getGlobalConfigSP()->getElements("storage/planes[@category='data']");
  3123. ForEach(*dataPlanes)
  3124. {
  3125. IPropertyTree & plane = dataPlanes->query();
  3126. const char * name = plane.queryProp("@name");
  3127. appendGroupNode(sprayTargets, name, "storage plane", false /*replicate outputs*/);
  3128. }
  3129. resp.setGroupNodes(sprayTargets);
  3130. #else
  3131. Owned<IEnvironmentFactory> factory = getEnvironmentFactory(true);
  3132. Owned<IConstEnvironment> environment = factory->openEnvironment();
  3133. Owned<IPropertyTree> root = &environment->getPTree();
  3134. IArrayOf<IEspGroupNode> sprayTargets;
  3135. //Fetch all the group names for all the thor instances (and dedup them)
  3136. BoolHash uniqueThorClusterGroupNames;
  3137. Owned<IPropertyTreeIterator> it = root->getElements("Software/ThorCluster");
  3138. ForEach(*it)
  3139. {
  3140. IPropertyTree& cluster = it->query();
  3141. StringBuffer thorClusterGroupName;
  3142. getClusterGroupName(cluster, thorClusterGroupName);
  3143. if (!thorClusterGroupName.length())
  3144. continue;
  3145. bool* found = uniqueThorClusterGroupNames.getValue(thorClusterGroupName.str());
  3146. if (!found || !*found)
  3147. appendGroupNode(sprayTargets, thorClusterGroupName.str(), "thor", cluster.getPropBool("@replicateOutputs", false));
  3148. }
  3149. //Fetch all the group names for all the hthor instances
  3150. it.setown(root->getElements("Software/EclAgentProcess"));
  3151. ForEach(*it)
  3152. {
  3153. IPropertyTree &cluster = it->query();
  3154. const char* name = cluster.queryProp("@name");
  3155. if (!name || !*name)
  3156. continue;
  3157. unsigned ins = 0;
  3158. Owned<IPropertyTreeIterator> insts = cluster.getElements("Instance");
  3159. ForEach(*insts)
  3160. {
  3161. const char *na = insts->query().queryProp("@netAddress");
  3162. if (!na || !*na)
  3163. continue;
  3164. SocketEndpoint ep(na);
  3165. if (ep.isNull())
  3166. continue;
  3167. ins++;
  3168. VStringBuffer gname("hthor__%s", name);
  3169. if (ins>1)
  3170. gname.append('_').append(ins);
  3171. appendGroupNode(sprayTargets, gname.str(), "hthor", false);
  3172. }
  3173. }
  3174. resp.setGroupNodes(sprayTargets);
  3175. #endif
  3176. }
  3177. catch(IException* e)
  3178. {
  3179. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  3180. }
  3181. return true;
  3182. }
  3183. void CFileSprayEx::setDFUServerQueueReq(const char* dfuServerQueue, IDFUWorkUnit* wu)
  3184. {
  3185. wu->setQueue((dfuServerQueue && *dfuServerQueue) ? dfuServerQueue : m_QueueLabel.str());
  3186. }
  3187. void CFileSprayEx::setUserAuth(IEspContext &context, IDFUWorkUnit* wu)
  3188. {
  3189. StringBuffer user, passwd;
  3190. wu->setUser(context.getUserID(user).str());
  3191. wu->setPassword(context.getPassword(passwd).str());
  3192. }
  3193. bool CFileSprayEx::onGetDFUServerQueues(IEspContext &context, IEspGetDFUServerQueuesRequest &req, IEspGetDFUServerQueuesResponse &resp)
  3194. {
  3195. try
  3196. {
  3197. context.ensureFeatureAccess(FILE_SPRAY_URL, SecAccess_Read, ECLWATCH_FILE_SPRAY_ACCESS_DENIED, "Permission denied.");
  3198. StringArray qlist;
  3199. #ifdef _CONTAINERIZED
  3200. Owned<IPropertyTreeIterator> dfuQueues = getComponentConfigSP()->getElements("dfuQueues");
  3201. ForEach(*dfuQueues)
  3202. {
  3203. IPropertyTree &dfuQueue = dfuQueues->query();
  3204. const char *dfuName = dfuQueue.queryProp("@name");
  3205. assertex(!isEmptyString(dfuName));
  3206. StringBuffer queueLabel;
  3207. getDfuQueueName(queueLabel, dfuName);
  3208. qlist.append(queueLabel);
  3209. }
  3210. #else
  3211. getDFUServerQueueNames(qlist, req.getDFUServerName());
  3212. #endif
  3213. resp.setNames(qlist);
  3214. }
  3215. catch(IException* e)
  3216. {
  3217. FORWARDEXCEPTION(context, e, ECLWATCH_INTERNAL_ERROR);
  3218. }
  3219. return true;
  3220. }