models2.py 142 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import json
  18. import logging
  19. import os
  20. import re
  21. import time
  22. import uuid
  23. from datetime import datetime, timedelta
  24. from dateutil.parser import parse
  25. from string import Template
  26. from xml.sax.saxutils import escape
  27. from django.core.urlresolvers import reverse
  28. from django.db.models import Q
  29. from django.utils.encoding import force_unicode
  30. from django.utils.translation import ugettext as _
  31. from desktop.conf import USE_DEFAULT_CONFIGURATION
  32. from desktop.lib import django_mako
  33. from desktop.lib.exceptions_renderable import PopupException
  34. from desktop.lib.i18n import smart_str
  35. from desktop.lib.json_utils import JSONEncoderForHTML
  36. from desktop.models import DefaultConfiguration, Document2, Document
  37. from hadoop.fs.hadoopfs import Hdfs
  38. from hadoop.fs.exceptions import WebHdfsException
  39. from liboozie.conf import SECURITY_ENABLED
  40. from liboozie.oozie_api import get_oozie
  41. from liboozie.submission2 import Submission
  42. from liboozie.submission2 import create_directories
  43. from notebook.models import Notebook
  44. from oozie.conf import REMOTE_SAMPLE_DIR
  45. from oozie.utils import utc_datetime_format, UTC_TIME_FORMAT, convert_to_server_timezone
  46. from oozie.importlib.workflows import generate_v2_graph_nodes, MalformedWfDefException, InvalidTagWithNamespaceException
  47. LOG = logging.getLogger(__name__)
  48. class Job(object):
  49. def find_all_parameters(self, with_lib_path=True):
  50. params = self.find_parameters()
  51. for param in self.parameters:
  52. params[param['name'].strip()] = param['value']
  53. if params.get('nominal_time') == '':
  54. params['nominal_time'] = datetime.today().strftime(UTC_TIME_FORMAT)
  55. return [{'name': name, 'value': value} for name, value in params.iteritems() if with_lib_path or name != 'oozie.use.system.libpath']
  56. @classmethod
  57. def get_workspace(cls, user):
  58. if not isinstance(user, basestring):
  59. user = user.username
  60. return (REMOTE_SAMPLE_DIR.get() + '/hue-oozie-$TIME').replace('$USER', user).replace('$TIME', str(time.time()))
  61. @property
  62. def validated_name(self):
  63. return ''.join(escape(self.name[:40], entities={
  64. "'": "'",
  65. "\"": """
  66. }))
  67. def __str__(self):
  68. return '%s' % force_unicode(self.name)
  69. def deployment_dir(self):
  70. return None
  71. def check_workspace(self, fs, user):
  72. # Create optional default root workspace for the first submission
  73. if REMOTE_SAMPLE_DIR.get() == REMOTE_SAMPLE_DIR.config.default_value:
  74. create_directories(fs, [REMOTE_SAMPLE_DIR.get()])
  75. Submission(user, self, fs, None, {})._create_dir(self.deployment_dir)
  76. Submission(user, self, fs, None, {})._create_dir(Hdfs.join(self.deployment_dir, 'lib'))
  77. def import_workspace(self, fs, source_deployment_dir, owner):
  78. try:
  79. fs.copy_remote_dir(source_deployment_dir, self.deployment_dir, owner=owner)
  80. except WebHdfsException, e:
  81. msg = _('The copy of the deployment directory failed: %s.') % e
  82. LOG.error(msg)
  83. raise PopupException(msg)
  84. class WorkflowConfiguration(object):
  85. APP_NAME = 'oozie-workflow'
  86. SLA_DEFAULT = [
  87. {'key': 'enabled', 'value': False}, # Always first element
  88. {'key': 'nominal-time', 'value': '${nominal_time}'},
  89. {'key': 'should-start', 'value': ''},
  90. {'key': 'should-end', 'value': '${30 * MINUTES}'},
  91. {'key': 'max-duration', 'value': ''},
  92. {'key': 'alert-events', 'value': ''},
  93. {'key': 'alert-contact', 'value': ''},
  94. {'key': 'notification-msg', 'value': ''},
  95. {'key': 'upstream-apps', 'value': ''},
  96. ]
  97. PROPERTIES = [
  98. {
  99. "multiple": True,
  100. "defaultValue": [
  101. {
  102. 'name': 'oozie.use.system.libpath',
  103. 'value': True
  104. }
  105. ],
  106. "value": [
  107. {
  108. 'name': 'oozie.use.system.libpath',
  109. 'value': True
  110. }
  111. ],
  112. "nice_name": _("Variables"),
  113. "key": "parameters",
  114. "help_text": _("Add one or more Oozie workflow job parameters."),
  115. "type": "parameters"
  116. }, {
  117. "multiple": False,
  118. "defaultValue": '',
  119. "value": '',
  120. "nice_name": _("Workspace"),
  121. "key": "deployment_dir",
  122. "help_text": _("Specify the deployment directory."),
  123. "type": "hdfs-file"
  124. }, {
  125. "multiple": True,
  126. "defaultValue": [],
  127. "value": [],
  128. "nice_name": _("Hadoop Properties"),
  129. "key": "properties",
  130. "help_text": _("Hadoop configuration properties."),
  131. "type": "settings"
  132. }, {
  133. "multiple": False,
  134. "defaultValue": True,
  135. "value": True,
  136. "nice_name": _("Show graph arrows"),
  137. "key": "show_arrows",
  138. "help_text": _("Toggles display of graph arrows."),
  139. "type": "boolean"
  140. }, {
  141. "multiple": False,
  142. "defaultValue": "uri:oozie:workflow:0.5",
  143. "value": "uri:oozie:workflow:0.5",
  144. "nice_name": _("Version"),
  145. "key": "schema_version",
  146. "help_text": _("Oozie XML Schema Version"),
  147. "type": "string",
  148. "options": [
  149. "uri:oozie:workflow:0.5",
  150. "uri:oozie:workflow:0.4.5",
  151. "uri:oozie:workflow:0.4",
  152. ]
  153. }, {
  154. "multiple": False,
  155. "defaultValue": '',
  156. "value": '',
  157. "nice_name": _("Job XML"),
  158. "key": "job_xml",
  159. "help_text": _("Oozie Job XML file"),
  160. "type": "hdfs-file"
  161. }, {
  162. "multiple": False,
  163. "defaultValue": False,
  164. "value": False,
  165. "nice_name": _("SLA Enabled"),
  166. "key": "sla_enabled",
  167. "help_text": _("SLA Enabled"),
  168. "type": "boolean"
  169. }, {
  170. "multiple": False,
  171. "defaultValue": SLA_DEFAULT,
  172. "value": SLA_DEFAULT,
  173. "nice_name": _("SLA Configuration"),
  174. "key": "sla",
  175. "help_text": _("Oozie SLA properties"),
  176. "type": "settings",
  177. "options": [prop['key'] for prop in SLA_DEFAULT]
  178. }
  179. ]
  180. class Workflow(Job):
  181. XML_FILE_NAME = 'workflow.xml'
  182. PROPERTY_APP_PATH = 'oozie.wf.application.path'
  183. HUE_ID = 'hue-id-w'
  184. def __init__(self, data=None, document=None, workflow=None, user=None):
  185. self.document = document
  186. self.user = user
  187. if document is not None:
  188. self.data = document.data
  189. elif data is not None:
  190. self.data = data
  191. else:
  192. if not workflow:
  193. workflow = self.get_default_workflow()
  194. workflow['properties'] = self.get_workflow_properties_for_user(user, workflow)
  195. self.data = json.dumps({
  196. 'layout': [{
  197. "size":12, "rows":[
  198. {"widgets":[{"size":12, "name":"Start", "id":"3f107997-04cc-8733-60a9-a4bb62cebffc", "widgetType":"start-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
  199. {"widgets":[{"size":12, "name":"End", "id":"33430f0f-ebfa-c3ec-f237-3e77efa03d0a", "widgetType":"end-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
  200. {"widgets":[{"size":12, "name":"Kill", "id":"17c9c895-5a16-7443-bb81-f34b30b21548", "widgetType":"kill-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]}
  201. ],
  202. "drops":[ "temp"],
  203. "klass":"card card-home card-column span12"
  204. }],
  205. 'workflow': workflow
  206. })
  207. @classmethod
  208. def get_application_path_key(cls):
  209. return 'oozie.wf.application.path'
  210. @classmethod
  211. def gen_workflow_data_from_xml(cls, user, oozie_workflow):
  212. node_list = []
  213. try:
  214. node_list = generate_v2_graph_nodes(oozie_workflow.definition)
  215. except MalformedWfDefException, e:
  216. LOG.exception("Could not find any nodes in Workflow definition. Maybe it's malformed?")
  217. except InvalidTagWithNamespaceException, e:
  218. LOG.exception(
  219. "Tag with namespace %(namespace)s is not valid. Please use one of the following namespaces: %(namespaces)s" % {
  220. 'namespace': e.namespace,
  221. 'namespaces': e.namespaces
  222. })
  223. _to_lowercase(node_list)
  224. adj_list = _create_graph_adjaceny_list(node_list)
  225. node_hierarchy = ['start']
  226. _get_hierarchy_from_adj_list(adj_list, adj_list['start']['ok_to'], node_hierarchy)
  227. _update_adj_list(adj_list)
  228. nodes_uuid_set = set()
  229. wf_rows = _create_workflow_layout(node_hierarchy, adj_list, nodes_uuid_set)
  230. data = {'layout': [{}], 'workflow': {}}
  231. if wf_rows:
  232. data['layout'][0]['rows'] = wf_rows
  233. wf_nodes = []
  234. nodes_uuid_set = set()
  235. _dig_nodes(node_hierarchy, adj_list, user, wf_nodes, nodes_uuid_set)
  236. data['workflow']['nodes'] = wf_nodes
  237. data['workflow']['id'] = '123'
  238. data['workflow']['properties'] = cls.get_workflow_properties_for_user(user, workflow=None)
  239. data['workflow']['properties'].update({
  240. 'deployment_dir': '/user/hue/oozie/workspaces/hue-oozie-1452553957.19'
  241. })
  242. return data
  243. @classmethod
  244. def get_default_workflow(cls):
  245. return {
  246. "id": None,
  247. "uuid": None,
  248. "name": "My Workflow",
  249. "nodes": [
  250. {"id": "3f107997-04cc-8733-60a9-a4bb62cebffc", "name": "Start", "type": "start-widget", "properties": {},
  251. "children": [{'to': '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'}]},
  252. {"id": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a", "name": "End", "type": "end-widget", "properties": {},
  253. "children": []},
  254. {"id": "17c9c895-5a16-7443-bb81-f34b30b21548", "name": "Kill", "type": "kill-widget",
  255. "properties": {'message': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]')},
  256. "children": []}
  257. ]
  258. }
  259. @classmethod
  260. def get_workflow_properties_for_user(cls, user, workflow=None):
  261. workflow = workflow if workflow is not None else {}
  262. properties = workflow.get('properties', None)
  263. if not properties:
  264. config = None
  265. properties = cls.get_properties()
  266. if user is not None:
  267. if USE_DEFAULT_CONFIGURATION.get():
  268. config = DefaultConfiguration.objects.get_configuration_for_user(app=WorkflowConfiguration.APP_NAME, user=user)
  269. if config is not None:
  270. properties.update(config.properties_dict)
  271. properties.update({
  272. 'wf1_id': None,
  273. 'description': ''
  274. })
  275. return properties
  276. @staticmethod
  277. def get_properties():
  278. return dict((prop['key'], prop['value']) for prop in WorkflowConfiguration.PROPERTIES)
  279. @property
  280. def id(self):
  281. return self.document.id if self.document else None
  282. @property
  283. def uuid(self):
  284. return self.document.uuid
  285. @property
  286. def name(self):
  287. _data = self.get_data()
  288. return _data['workflow']['name']
  289. @property
  290. def deployment_dir(self):
  291. _data = self.get_data()
  292. return _data['workflow']['properties']['deployment_dir']
  293. @property
  294. def parameters(self):
  295. _data = self.get_data()
  296. return _data['workflow']['properties']['parameters']
  297. @property
  298. def sla_enabled(self):
  299. _data = self.get_data()
  300. return _data['workflow']['properties']['sla'][0].get('value')
  301. @property
  302. def has_some_slas(self):
  303. return self.sla_enabled or any([node.sla_enabled for node in self.nodes])
  304. @property
  305. def credentials(self):
  306. return list(set([cred for node in self.nodes for cred in node.data['properties']['credentials']]))
  307. @property
  308. def sla(self):
  309. _data = self.get_data()
  310. return _data['workflow']['properties']['sla']
  311. @property
  312. def nodes(self):
  313. _data = self.get_data()
  314. return [Node(node, self.user) for node in _data['workflow']['nodes']]
  315. def find_parameters(self):
  316. params = set()
  317. for param in find_dollar_braced_variables(self.name):
  318. params.add(param)
  319. if self.sla_enabled:
  320. for param in find_json_parameters(self.sla):
  321. params.add(param)
  322. parameters = dict([(param, '') for param in list(params)])
  323. for node in self.nodes:
  324. if 'document' in node.data['type']:
  325. for param in node.data['properties']['parameters']:
  326. if param['value'] and '=' in param['value']:
  327. name, val = param['value'].split('=', 1)
  328. parameters[name] = val
  329. extra_fields = []
  330. if node.data['properties'].get('key_tab_path'):
  331. extra_fields.append('key_tab_path')
  332. if node.data['properties'].get('user_principal'):
  333. extra_fields.append('user_principal')
  334. extra = find_parameters(node, fields=extra_fields)
  335. else:
  336. extra = node.find_parameters()
  337. if extra:
  338. parameters.update(dict([(param, '') for param in list(extra)]))
  339. return parameters
  340. def get_json(self):
  341. _data = self.get_data()
  342. return json.dumps(_data)
  343. def get_data(self):
  344. _data = json.loads(self.data)
  345. if self.document is not None:
  346. _data['workflow']['id'] = self.document.id
  347. _data['workflow']['dependencies'] = list(self.document.dependencies.values('uuid', ))
  348. else:
  349. _data['workflow']['dependencies'] = []
  350. if 'parameters' not in _data['workflow']['properties']:
  351. _data['workflow']['properties']['parameters'] = [
  352. {'name': 'oozie.use.system.libpath', 'value': True},
  353. ]
  354. if 'show_arrows' not in _data['workflow']['properties']:
  355. _data['workflow']['properties']['show_arrows'] = True
  356. for node in _data['workflow']['nodes']:
  357. if 'credentials' in node['properties']: # If node is an Action
  358. if 'retry_max' not in node['properties']: # When displaying a workflow
  359. node['properties']['retry_max'] = []
  360. if 'retry_interval' not in node['properties']:
  361. node['properties']['retry_interval'] = []
  362. # Backward compatibility
  363. _upgrade_older_node(node)
  364. return _data
  365. def to_xml(self, mapping=None):
  366. if mapping is None:
  367. mapping = {}
  368. tmpl = 'editor2/gen/workflow.xml.mako'
  369. data = self.get_data()
  370. nodes = [node for node in self.nodes if node.name != 'End'] + [node for node in self.nodes if node.name == 'End'] # End at the end
  371. node_mapping = dict([(node.id, node) for node in nodes])
  372. sub_wfs_ids = [node.data['properties']['workflow'] for node in nodes if node.data['type'] == 'subworkflow']
  373. workflow_mapping = dict(
  374. [(workflow.uuid, Workflow(document=workflow, user=self.user)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)])
  375. xml = re.sub(re.compile('>\s*\n+', re.MULTILINE), '>\n', django_mako.render_to_string(tmpl, {
  376. 'wf': self,
  377. 'workflow': data['workflow'],
  378. 'nodes': nodes,
  379. 'mapping': mapping,
  380. 'node_mapping': node_mapping,
  381. 'workflow_mapping': workflow_mapping
  382. }))
  383. return force_unicode(xml.strip())
  384. def get_absolute_url(self):
  385. return reverse('oozie:edit_workflow') + '?workflow=%s' % self.id if self.document else ''
  386. def override_subworkflow_id(self, sub_wf_action, workflow_id):
  387. _data = self.get_data()
  388. action = [_action for _action in _data['workflow']['nodes'] if _action['id'] == sub_wf_action.id]
  389. if action:
  390. action[0]['properties']['job_properties'].append({'name': Workflow.HUE_ID, 'value': workflow_id})
  391. self.data = json.dumps(_data)
  392. def update_name(self, name):
  393. _data = self.get_data()
  394. _data['workflow']['name'] = name
  395. self.data = json.dumps(_data)
  396. def set_workspace(self, user):
  397. _data = json.loads(self.data)
  398. _data['workflow']['properties']['deployment_dir'] = Job.get_workspace(user)
  399. self.data = json.dumps(_data)
  400. def create_single_action_workflow_data(self, node_id):
  401. _data = json.loads(self.data)
  402. start_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'Start'][0]
  403. submit_node = [node for node in _data['workflow']['nodes'] if node['id'] == node_id][0]
  404. end_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'End'][0]
  405. kill_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'Kill'][0]
  406. # Modify children to point Start -> Submit_node -> End/Kill
  407. start_node['children'] = [{'to': submit_node['id']}]
  408. submit_node['children'] = [{'to': end_node['id']}, {'error': kill_node['id']}]
  409. _data['workflow']['properties']['deployment_dir'] = None
  410. # Recursively find the widget node
  411. def _get_node(rows, node_id):
  412. for row in rows:
  413. if not row['widgets']:
  414. for col in row['columns']:
  415. node = _get_node(col['rows'], node_id)
  416. if node:
  417. return node
  418. elif row['widgets'][0]['id'] == node_id:
  419. return row
  420. # Create wf data with above nodes
  421. return json.dumps({
  422. 'layout': [{
  423. "size": 12,
  424. "rows": [
  425. [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'Start'][0],
  426. _get_node(_data['layout'][0]['rows'], node_id),
  427. [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'End'][0],
  428. [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'Kill'][0]
  429. ],
  430. "drops": ["temp"],
  431. "klass": "card card-home card-column span12"
  432. }],
  433. 'workflow': {
  434. "id": None,
  435. "uuid": None,
  436. "name": _data['workflow']['name'],
  437. "properties": _data['workflow']['properties'],
  438. "nodes": [start_node, submit_node, end_node, kill_node]
  439. }
  440. })
  441. # Updates node_list to lowercase names
  442. # To avoid case-sensitive failures
  443. def _to_lowercase(node_list):
  444. for node in node_list:
  445. for key in node.keys():
  446. if hasattr(node[key], 'lower'):
  447. node[key] = node[key].lower()
  448. def _update_adj_list(adj_list):
  449. uuids = {}
  450. id = 1
  451. first_kill_node_seen = False
  452. for node in adj_list.keys():
  453. adj_list[node]['id'] = id
  454. # Oozie uses same action for streaming and mapreduce but Hue manages them differently
  455. if adj_list[node]['node_type'] == 'map-reduce':
  456. if 'streaming' in adj_list[node]['name']:
  457. adj_list[node]['node_type'] = 'streaming'
  458. else:
  459. adj_list[node]['node_type'] = 'mapreduce'
  460. elif adj_list[node]['node_type'] == 'sub-workflow':
  461. adj_list[node]['node_type'] = 'subworkflow'
  462. if adj_list[node]['node_type'] == 'kill':
  463. # JS requires at least one of the kill nodes to have this Id
  464. if not first_kill_node_seen:
  465. adj_list[node]['uuid'] = '17c9c895-5a16-7443-bb81-f34b30b21548'
  466. first_kill_node_seen = True
  467. else:
  468. adj_list[node]['uuid'] = str(uuid.uuid4())
  469. elif adj_list[node]['node_type'] == 'start':
  470. adj_list[node]['uuid'] = '3f107997-04cc-8733-60a9-a4bb62cebffc'
  471. elif adj_list[node]['node_type'] == 'end':
  472. adj_list[node]['uuid'] = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
  473. else:
  474. adj_list[node]['uuid'] = node[-4:] + str(uuid.uuid4())[4:]
  475. uuids[id] = adj_list[node]['uuid']
  476. id += 1
  477. return adj_list
  478. def _dig_nodes(nodes, adj_list, user, wf_nodes, nodes_uuid_set):
  479. for node in nodes:
  480. if type(node) != list:
  481. node = adj_list[node]
  482. if node['uuid'] not in nodes_uuid_set:
  483. properties = {}
  484. if '%s-widget' % node['node_type'] in NODES:
  485. properties = dict(NODES['%s-widget' % node['node_type']].get_fields())
  486. if node['node_type'] == 'pig':
  487. properties['script_path'] = node.get('pig').get('script_path')
  488. elif node['node_type'] == 'spark':
  489. properties['class'] = node.get('spark').get('class')
  490. properties['jars'] = node.get('spark').get('jar')
  491. elif node['node_type'] == 'hive' or node['node_type'] == 'hive2':
  492. properties['script_path'] = node.get('hive').get('script')
  493. elif node['node_type'] == 'java':
  494. properties['main_class'] = node.get('java').get('main-class')
  495. elif node['node_type'] == 'sqoop':
  496. properties['command'] = node.get('sqoop').get('command')
  497. elif node['node_type'] == 'mapreduce':
  498. properties['job_properties'] = node.get('job_properties')
  499. elif node['node_type'] == 'shell':
  500. properties['shell_command'] = node.get('shell').get('command')
  501. elif node['node_type'] == 'ssh':
  502. properties['user'] = '%s@%s' % (node.get('ssh').get('user'), node.get('ssh').get('host'))
  503. properties['ssh_command'] = node.get('ssh').get('command')
  504. elif node['node_type'] == 'fs':
  505. properties['touchzs'] = node.get('fs').get('touchzs')
  506. properties['mkdirs'] = node.get('fs').get('mkdirs')
  507. properties['moves'] = node.get('fs').get('moves')
  508. properties['deletes'] = node.get('fs').get('deletes')
  509. elif node['node_type'] == 'email':
  510. properties['to'] = node.get('email').get('to')
  511. properties['subject'] = node.get('email').get('subject')
  512. elif node['node_type'] == 'streaming':
  513. properties['mapper'] = node.get('streaming').get('mapper')
  514. properties['reducer'] = node.get('streaming').get('reducer')
  515. elif node['node_type'] == 'distcp':
  516. properties['distcp_parameters'] = node.get('params')
  517. elif node['node_type'] == 'subworkflow':
  518. properties['app-path'] = node.get('subworkflow').get('app-path')
  519. properties['workflow'] = node.get('uuid')
  520. properties['job_properties'] = []
  521. properties['sla'] = ''
  522. children = []
  523. if node['node_type'] in ('fork', 'decision'):
  524. for key in node.keys():
  525. if key.startswith('path'):
  526. children.append({'to': adj_list[node[key]]['uuid'], 'condition': '${ 1 gt 0 }'})
  527. if node['node_type'] == 'decision':
  528. children.append({'to': adj_list[node['default']]['uuid'], 'condition': '${ 1 gt 0 }'})
  529. else:
  530. if node.get('ok_to'):
  531. children.append({'to': adj_list[node['ok_to']]['uuid']})
  532. if node.get('error_to'):
  533. children.append({'error': adj_list[node['error_to']]['uuid']})
  534. nodes_uuid_set.add(node['uuid'])
  535. wf_nodes.append({
  536. "id": node['uuid'],
  537. "name": '%s-%s' % (node['node_type'].split('-')[0], node['uuid'][:4]),
  538. "type": "%s-widget" % node['node_type'],
  539. "properties": properties,
  540. "children": children
  541. })
  542. else:
  543. _dig_nodes(node, adj_list, user, wf_nodes, nodes_uuid_set)
  544. def _create_workflow_layout(nodes, adj_list, nodes_uuid_set, size=12):
  545. wf_rows = []
  546. for node in nodes:
  547. if type(node) == list and len(node) == 1:
  548. node = node[0]
  549. if type(node) != list:
  550. _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node]['uuid'],
  551. row={"widgets":[{"size":size, "name": adj_list[node]['node_type'], "id": adj_list[node]['uuid'], "widgetType": "%s-widget" % adj_list[node]['node_type'], "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  552. else:
  553. if adj_list[node[0]]['node_type'] in ('fork', 'decision'):
  554. _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node[0]]['uuid'],
  555. row={"widgets":[{"size":size, "name": adj_list[node[0]]['name'], "id": adj_list[node[0]]['uuid'], "widgetType": "%s-widget" % adj_list[node[0]]['node_type'], "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  556. wf_rows.append({
  557. "id": str(uuid.uuid4()),
  558. "widgets":[
  559. ],
  560. "columns":[
  561. {
  562. "id": str(uuid.uuid4()),
  563. "size": (size / len(node[1])),
  564. "rows":
  565. [{
  566. "id": str(uuid.uuid4()),
  567. "widgets": c['widgets'],
  568. "columns":c.get('columns') or []
  569. } for c in col],
  570. "klass":"card card-home card-column span%s" % (size / len(node[1]))
  571. }
  572. for col in [_create_workflow_layout(item, adj_list, nodes_uuid_set, size) for item in node[1]]
  573. ]
  574. })
  575. if adj_list[node[0]]['node_type'] == 'fork':
  576. wf_rows.append({"widgets":[{"size":size, "name": adj_list[node[2]]['name'], "id": adj_list[node[2]]['uuid'], "widgetType": "%s-widget" % adj_list[node[2]]['node_type'], "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  577. else:
  578. wf_rows.append(_create_workflow_layout(node, adj_list, nodes_uuid_set, size))
  579. return wf_rows
  580. # Prevent duplicate nodes in graph layout
  581. def _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id, row):
  582. if row['widgets'][0]['id'] not in nodes_uuid_set:
  583. nodes_uuid_set.add(row['widgets'][0]['id'])
  584. wf_rows.append(row)
  585. def _get_hierarchy_from_adj_list(adj_list, curr_node, node_hierarchy):
  586. _get_hierarchy_from_adj_list_helper(adj_list, curr_node, node_hierarchy)
  587. # Add End and Kill nodes to node_hierarchy
  588. for key in adj_list.keys():
  589. if adj_list[key]['node_type'] == 'kill':
  590. node_hierarchy.append([adj_list[key]['name']])
  591. node_hierarchy.append([adj_list[key]['name'] for key in adj_list.keys() if adj_list[key]['node_type'] == 'end'])
  592. def _get_hierarchy_from_adj_list_helper(adj_list, curr_node, node_hierarchy):
  593. if not curr_node or adj_list[curr_node]['node_type'] in ('join', 'end', 'kill'):
  594. return curr_node
  595. elif adj_list[curr_node]['node_type'] in ('fork', 'decision'):
  596. branch_nodes = []
  597. branch_nodes.append(curr_node)
  598. join_node = None
  599. children = []
  600. for key in adj_list[curr_node].keys():
  601. if key.startswith('path') or key == 'default':
  602. child = []
  603. return_node = _get_hierarchy_from_adj_list_helper(adj_list, adj_list[curr_node][key], child)
  604. join_node = return_node if not join_node else join_node
  605. if child:
  606. children.append(child)
  607. branch_nodes.append(children)
  608. if adj_list[curr_node]['node_type'] == 'fork':
  609. branch_nodes.append(join_node)
  610. node_hierarchy.append(branch_nodes)
  611. return _get_hierarchy_from_adj_list_helper(adj_list, adj_list[join_node]['ok_to'], node_hierarchy)
  612. node_hierarchy.append(branch_nodes)
  613. return join_node
  614. else:
  615. node_hierarchy.append(curr_node)
  616. return _get_hierarchy_from_adj_list_helper(adj_list, adj_list[curr_node]['ok_to'], node_hierarchy)
  617. def _create_graph_adjaceny_list(nodes):
  618. start_node = [node for node in nodes if node.get('node_type') == 'start'][0]
  619. adj_list = {'start': start_node}
  620. for node in nodes:
  621. if node and node.get('node_type') != 'start':
  622. adj_list[node['name']] = node
  623. return adj_list
  624. class Node():
  625. def __init__(self, data, user=None):
  626. self.data = data
  627. self.user = user
  628. self._augment_data()
  629. def to_xml(self, mapping=None, node_mapping=None, workflow_mapping=None):
  630. if mapping is None:
  631. mapping = {}
  632. if node_mapping is None:
  633. node_mapping = {}
  634. if workflow_mapping is None:
  635. workflow_mapping = {}
  636. if self.data['type'] in ('hive2', 'hive-document') and not self.data['properties']['jdbc_url']:
  637. self.data['properties']['jdbc_url'] = _get_hiveserver2_url()
  638. if self.data['type'] == 'fork':
  639. links = [link for link in self.data['children'] if link['to'] in node_mapping]
  640. if len(links) != len(self.data['children']):
  641. LOG.warn('Fork has some children links that do not exist, ignoring them: links %s, existing links %s, links %s, existing links %s' \
  642. % (len(links), len(self.data['children']), links, self.data['children']))
  643. self.data['children'] = links
  644. if self.data['type'] == JavaDocumentAction.TYPE:
  645. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  646. properties = notebook.get_data()['snippets'][0]['properties']
  647. self.data['properties']['main_class'] = properties['class']
  648. self.data['properties']['app_jar'] = properties['app_jar'] # Not used here
  649. self.data['properties']['files'] = [{'value': f['path']} for f in properties['files']]
  650. self.data['properties']['arguments'] = [{'value': prop} for prop in properties['arguments']]
  651. elif self.data['type'] == SparkDocumentAction.TYPE or self.data['type'] == 'spark-document':
  652. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  653. properties = notebook.get_data()['snippets'][0]['properties']
  654. if self.data['type'] == 'spark-document': # Oozie Document Action
  655. self.data['properties']['app_name'] = properties['app_name']
  656. self.data['properties']['class'] = properties['class']
  657. self.data['properties']['jars'] = os.path.basename(properties['jars'][0])
  658. self.data['properties']['files'] = [{'value': f} for f in properties['jars']] + [{'value': f['path']} for f in properties['files']]
  659. self.data['properties']['spark_arguments'] = [{'value': prop} for prop in properties['spark_arguments']]
  660. self.data['properties']['spark_opts'] = ' '.join(properties['spark_opts'])
  661. if len(properties['jars']) > 1:
  662. self.data['properties']['spark_opts'] += ' --py-files ' + ','.join([os.path.basename(f) for f in properties['jars'][1:]])
  663. elif self.data['type'] == PigDocumentAction.TYPE:
  664. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  665. action = notebook.get_data()['snippets'][0]
  666. name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
  667. self.data['properties']['script_path'] = "${wf:appPath()}/" + name + ".pig"
  668. self.data['properties']['parameters'] = [{'value': prop} for prop in action['properties']['parameters']]
  669. self.data['properties']['arguments'] = []
  670. job_properties = []
  671. for prop in action['properties']['hadoopProperties']:
  672. name, value = prop.split('=', 1)
  673. job_properties.append({'name': name, 'value': value})
  674. self.data['properties']['job_properties'] = job_properties
  675. self.data['properties']['files'] = [{'value': prop} for prop in action['properties']['parameters']]
  676. elif self.data['type'] == SqoopDocumentAction.TYPE:
  677. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  678. action = notebook.get_data()['snippets'][0]
  679. name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
  680. command = action['statement']
  681. if command.startswith('sqoop '):
  682. _ignore, command = command.split('sqoop ', 1)
  683. self.data['properties']['command'] = command
  684. self.data['properties']['files'] = [{'value': f['path']} for f in action['properties']['files']]
  685. self.data['properties']['arguments'] = []
  686. elif self.data['type'] == DistCpDocumentAction.TYPE:
  687. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  688. action = notebook.get_data()['snippets'][0]
  689. name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
  690. self.data['properties']['source_path'] = action['properties']['source_path']
  691. self.data['properties']['destination_path'] = action['properties']['destination_path']
  692. elif self.data['type'] == ShellDocumentAction.TYPE:
  693. if self.data['properties'].get('uuid'):
  694. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  695. action = notebook.get_data()['snippets'][0]
  696. name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
  697. self.data['properties']['shell_command'] = action['properties']['command_path']
  698. self.data['properties']['env_var'] = [{'value': prop} for prop in action['properties']['env_var']]
  699. self.data['properties']['capture_output'] = action['properties']['capture_output']
  700. self.data['properties']['arguments'] = [{'value': prop} for prop in action['properties']['arguments']]
  701. self.data['properties']['files'] = ([{'value': action['properties']['command_path']}] if not action['properties'].get('command_path', '').startswith('/') else []) + [{'value': prop} for prop in action['properties']['files']]
  702. self.data['properties']['archives'] = [{'value': prop} for prop in action['properties']['archives']]
  703. elif self.data['type'] == MapReduceDocumentAction.TYPE:
  704. notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
  705. action = notebook.get_data()['snippets'][0]
  706. name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
  707. self.data['properties']['app_jar'] = action['properties']['app_jar']
  708. self.data['properties']['arguments'] = []
  709. self.data['properties']['parameters'] = []
  710. job_properties = []
  711. for prop in action['properties']['hadoopProperties']:
  712. name, value = prop.split('=', 1)
  713. job_properties.append({'name': name, 'value': value})
  714. self.data['properties']['job_properties'] = job_properties
  715. self.data['properties']['files'] = [{'value': prop} for prop in action['properties']['files']]
  716. self.data['properties']['archives'] = [{'value': prop} for prop in action['properties']['archives']]
  717. elif self.data['type'] == ImpalaAction.TYPE or self.data['type'] == ImpalaDocumentAction.TYPE:
  718. shell_command_name = self.data['name'] + '.sh'
  719. self.data['properties']['shell_command'] = shell_command_name
  720. self.data['properties']['env_var'] = []
  721. self.data['properties']['capture_output'] = False
  722. self.data['properties']['arguments'] = []
  723. if self.data['type'] == ImpalaAction.TYPE:
  724. script_path = self.data['properties'].get('script_path')
  725. else:
  726. script_path = self.data['name'] + '.sql'
  727. files = [{'value': shell_command_name}, {'value': script_path}]
  728. if self.data['properties']['key_tab_path']:
  729. files.append({'value': self.data['properties']['key_tab_path']})
  730. self.data['properties']['files'] = files
  731. self.data['properties']['archives'] = []
  732. data = {
  733. 'node': self.data,
  734. 'mapping': mapping,
  735. 'node_mapping': node_mapping,
  736. 'workflow_mapping': workflow_mapping
  737. }
  738. if mapping.get('send_email'):
  739. if self.data['type'] == KillAction.TYPE and not self.data['properties'].get('enableMail'):
  740. self.data['properties']['enableMail'] = True
  741. self.data['properties']['to'] = self.user.email
  742. self.data['properties']['subject'] = _("${wf:name()} execution failure")
  743. self.data['properties']['body'] = _("Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]")
  744. if self.data['type'] == EndNode.TYPE:
  745. self.data['properties']['enableMail'] = True
  746. self.data['properties']['to'] = self.user.email
  747. self.data['properties']['subject'] = _("${wf:name()} execution successful")
  748. if mapping.get('send_result_path'):
  749. if self.data['type'] == EndNode.TYPE:
  750. self.data['properties']['body'] = 'View result file at %(send_result_browse_url)s' % mapping
  751. return django_mako.render_to_string(self.get_template_name(), data)
  752. @property
  753. def id(self):
  754. return self.data['id']
  755. @property
  756. def name(self):
  757. return self.data['name']
  758. @property
  759. def sla_enabled(self):
  760. return 'sla' in self.data['properties'] and self.data['properties']['sla'] and self.data['properties']['sla'][0].get('value')
  761. def _augment_data(self):
  762. self.data['type'] = self.data['type'].replace('-widget', '')
  763. self.data['uuid'] = self.data['id']
  764. # Action Node
  765. if 'credentials' not in self.data['properties']:
  766. self.data['properties']['credentials'] = []
  767. if 'prepares' not in self.data['properties']:
  768. self.data['properties']['prepares'] = []
  769. if 'job_xml' not in self.data['properties']:
  770. self.data['properties']['job_xml'] = []
  771. if 'properties' not in self.data['properties']:
  772. self.data['properties']['properties'] = []
  773. if 'params' not in self.data['properties']:
  774. self.data['properties']['params'] = []
  775. if 'files' not in self.data['properties']:
  776. self.data['properties']['files'] = []
  777. if 'archives' not in self.data['properties']:
  778. self.data['properties']['archives'] = []
  779. if 'sla' not in self.data['properties']:
  780. self.data['properties']['sla'] = WorkflowConfiguration.SLA_DEFAULT
  781. if 'retry_max' not in self.data['properties']:
  782. self.data['properties']['retry_max'] = []
  783. if 'retry_interval' not in self.data['properties']:
  784. self.data['properties']['retry_interval'] = []
  785. # Backward compatibility
  786. _upgrade_older_node(self.data)
  787. def get_template_name(self):
  788. node_type = self.data['type']
  789. if self.data['type'] == JavaDocumentAction.TYPE:
  790. node_type = JavaAction.TYPE
  791. elif self.data['type'] == ImpalaAction.TYPE or self.data['type'] == ImpalaDocumentAction.TYPE:
  792. node_type = ShellAction.TYPE
  793. return 'editor2/gen/workflow-%s.xml.mako' % node_type
  794. def find_parameters(self):
  795. return find_parameters(self) + (find_parameters(self, ['sla']) if self.sla_enabled else [])
  796. def _upgrade_older_node(node):
  797. if node['type'] in ('sqoop', 'sqoop-widget') and 'arguments' not in node['properties']:
  798. node['properties']['arguments'] = node['properties']['parameters']
  799. if node['type'] in ('kill', 'kill-widget') and 'to' not in node['properties']:
  800. node['properties']['enableMail'] = False
  801. node['properties']['to'] = ''
  802. node['properties']['cc'] = ''
  803. node['properties']['subject'] = ''
  804. node['properties']['body'] = ''
  805. if node['type'] in ('end', 'end-widget') and 'to' not in node['properties']:
  806. node['properties']['enableMail'] = False
  807. node['properties']['to'] = ''
  808. node['properties']['cc'] = ''
  809. node['properties']['subject'] = ''
  810. node['properties']['body'] = ''
  811. node['properties']['content_type'] = 'text/plain'
  812. node['properties']['attachment'] = ''
  813. if node['type'] == 'email-widget' and 'bcc' not in node['properties']:
  814. node['properties']['bcc'] = ''
  815. node['properties']['content_type'] = 'text/plain'
  816. node['properties']['attachment'] = ''
  817. if node['type'] == 'spark-widget' and 'files' not in node['properties']:
  818. node['properties']['files'] = []
  819. if (node['type'] == 'hive2-widget' or node['type'] == 'hive-document-widget') and 'arguments' not in node['properties']:
  820. node['properties']['arguments'] = []
  821. class Action(object):
  822. @classmethod
  823. def get_fields(cls):
  824. credentials = [cls.DEFAULT_CREDENTIALS] if hasattr(cls, 'DEFAULT_CREDENTIALS') and cls.DEFAULT_CREDENTIALS else []
  825. return [(f['name'], f['value']) for f in cls.FIELDS.itervalues()] + [('sla', WorkflowConfiguration.SLA_DEFAULT), ('credentials', credentials)]
  826. class StartNode(Action):
  827. TYPE = 'start'
  828. FIELDS = {}
  829. class EndNode(Action):
  830. TYPE = 'end'
  831. FIELDS = {}
  832. class PigAction(Action):
  833. TYPE = 'pig'
  834. FIELDS = {
  835. 'script_path': {
  836. 'name': 'script_path',
  837. 'label': _('Script'),
  838. 'value': '',
  839. 'help_text': _('Path to the script on HDFS.'),
  840. 'type': ''
  841. },
  842. 'parameters': {
  843. 'name': 'parameters',
  844. 'label': _('Parameters'),
  845. 'value': [],
  846. 'help_text': _('The Pig parameters of the script without -param. e.g. INPUT=${inputDir}'),
  847. 'type': ''
  848. },
  849. 'arguments': {
  850. 'name': 'arguments',
  851. 'label': _('Arguments'),
  852. 'value': [],
  853. 'help_text': _('The Pig parameters of the script as is. e.g. -param, INPUT=${inputDir}'),
  854. 'type': ''
  855. },
  856. # Common
  857. 'files': {
  858. 'name': 'files',
  859. 'label': _('Files'),
  860. 'value': [],
  861. 'help_text': _('Files put in the running directory.'),
  862. 'type': ''
  863. },
  864. 'archives': {
  865. 'name': 'archives',
  866. 'label': _('Archives'),
  867. 'value': [],
  868. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  869. 'type': ''
  870. },
  871. 'job_properties': {
  872. 'name': 'job_properties',
  873. 'label': _('Hadoop job properties'),
  874. 'value': [],
  875. 'help_text': _('value, e.g. production'),
  876. 'type': ''
  877. },
  878. 'prepares': {
  879. 'name': 'prepares',
  880. 'label': _('Prepares'),
  881. 'value': [],
  882. 'help_text': _('Path to manipulate before starting the application.'),
  883. 'type': ''
  884. },
  885. 'job_xml': {
  886. 'name': 'job_xml',
  887. 'label': _('Job XML'),
  888. 'value': [],
  889. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  890. 'type': ''
  891. },
  892. 'retry_max': {
  893. 'name': 'retry_max',
  894. 'label': _('Max retry'),
  895. 'value': [],
  896. 'help_text': _('Number of times, default is 3'),
  897. 'type': ''
  898. },
  899. 'retry_interval': {
  900. 'name': 'retry_interval',
  901. 'label': _('Retry interval'),
  902. 'value': [],
  903. 'help_text': _('Wait time in minutes, default is 10'),
  904. 'type': ''
  905. }
  906. }
  907. @classmethod
  908. def get_mandatory_fields(cls):
  909. return [cls.FIELDS['script_path']]
  910. class JavaAction(Action):
  911. TYPE = 'java'
  912. FIELDS = {
  913. 'jar_path': {
  914. 'name': 'jar_path',
  915. 'label': _('Jar name'),
  916. 'value': '',
  917. 'help_text': _('Path to the jar on HDFS.'),
  918. 'type': ''
  919. },
  920. 'main_class': {
  921. 'name': 'main_class',
  922. 'label': _('Main class'),
  923. 'value': '',
  924. 'help_text': _('Java class. e.g. org.apache.hadoop.examples.Grep'),
  925. 'type': 'text'
  926. },
  927. 'arguments': {
  928. 'name': 'arguments',
  929. 'label': _('Arguments'),
  930. 'value': [],
  931. 'help_text': _('Arguments of the main method. The value of each arg element is considered a single argument '
  932. 'and they are passed to the main method in the same order.'),
  933. 'type': ''
  934. },
  935. 'java_opts': {
  936. 'name': 'java_opts',
  937. 'label': _('Java options'),
  938. 'value': [],
  939. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b'),
  940. 'type': ''
  941. },
  942. 'capture_output': {
  943. 'name': 'capture_output',
  944. 'label': _('Capture output'),
  945. 'value': False,
  946. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  947. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  948. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  949. 'via the String action:output(String node, String key) function') % {'program': TYPE.title()},
  950. 'type': ''
  951. },
  952. # Common
  953. 'files': {
  954. 'name': 'files',
  955. 'label': _('Files'),
  956. 'value': [],
  957. 'help_text': _('Files put in the running directory.'),
  958. 'type': ''
  959. },
  960. 'archives': {
  961. 'name': 'archives',
  962. 'label': _('Archives'),
  963. 'value': [],
  964. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  965. 'type': ''
  966. },
  967. 'job_properties': {
  968. 'name': 'job_properties',
  969. 'label': _('Hadoop job properties'),
  970. 'value': [],
  971. 'help_text': _('value, e.g. production'),
  972. 'type': ''
  973. },
  974. 'prepares': {
  975. 'name': 'prepares',
  976. 'label': _('Prepares'),
  977. 'value': [],
  978. 'help_text': _('Path to manipulate before starting the application.'),
  979. 'type': ''
  980. },
  981. 'job_xml': {
  982. 'name': 'job_xml',
  983. 'label': _('Job XML'),
  984. 'value': [],
  985. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  986. 'type': ''
  987. },
  988. 'retry_max': {
  989. 'name': 'retry_max',
  990. 'label': _('Max retry'),
  991. 'value': [],
  992. 'help_text': _('Number of times, default is 3'),
  993. 'type': ''
  994. },
  995. 'retry_interval': {
  996. 'name': 'retry_interval',
  997. 'label': _('Retry interval'),
  998. 'value': [],
  999. 'help_text': _('Wait time in minutes, default is 10'),
  1000. 'type': ''
  1001. }
  1002. }
  1003. @classmethod
  1004. def get_mandatory_fields(cls):
  1005. return [cls.FIELDS['jar_path'], cls.FIELDS['main_class']]
  1006. class HiveAction(Action):
  1007. TYPE = 'hive'
  1008. DEFAULT_CREDENTIALS = 'hcat'
  1009. FIELDS = {
  1010. 'script_path': {
  1011. 'name': 'script_path',
  1012. 'label': _('Script'),
  1013. 'value': '',
  1014. 'help_text': _('Path to the script on HDFS.'),
  1015. 'type': ''
  1016. },
  1017. 'parameters': {
  1018. 'name': 'parameters',
  1019. 'label': _('Parameters'),
  1020. 'value': [],
  1021. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  1022. 'type': ''
  1023. },
  1024. # Common
  1025. 'files': {
  1026. 'name': 'files',
  1027. 'label': _('Files'),
  1028. 'value': [],
  1029. 'help_text': _('Files put in the running directory.'),
  1030. 'type': ''
  1031. },
  1032. 'archives': {
  1033. 'name': 'archives',
  1034. 'label': _('Archives'),
  1035. 'value': [],
  1036. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1037. 'type': ''
  1038. },
  1039. 'job_properties': {
  1040. 'name': 'job_properties',
  1041. 'label': _('Hadoop job properties'),
  1042. 'value': [],
  1043. 'help_text': _('value, e.g. production'),
  1044. 'type': ''
  1045. },
  1046. 'prepares': {
  1047. 'name': 'prepares',
  1048. 'label': _('Prepares'),
  1049. 'value': [],
  1050. 'help_text': _('Path to manipulate before starting the application.'),
  1051. 'type': ''
  1052. },
  1053. 'hive_xml': {
  1054. 'name': 'hive_xml',
  1055. 'label': _('Hive XML'),
  1056. 'value': '',
  1057. 'help_text': _('Refer to a hive-site.xml for connecting to Hive'),
  1058. 'type': ''
  1059. },
  1060. 'retry_max': {
  1061. 'name': 'retry_max',
  1062. 'label': _('Max retry'),
  1063. 'value': [],
  1064. 'help_text': _('Number of times, default is 3'),
  1065. 'type': ''
  1066. },
  1067. 'retry_interval': {
  1068. 'name': 'retry_interval',
  1069. 'label': _('Retry interval'),
  1070. 'value': [],
  1071. 'help_text': _('Wait time in minutes, default is 10'),
  1072. 'type': ''
  1073. }
  1074. }
  1075. @classmethod
  1076. def get_mandatory_fields(cls):
  1077. return [cls.FIELDS['script_path'], cls.FIELDS['hive_xml']]
  1078. def _get_hiveserver2_url():
  1079. try:
  1080. from beeswax.hive_site import hiveserver2_jdbc_url
  1081. return hiveserver2_jdbc_url()
  1082. except Exception, e:
  1083. # Might fail is Hive is disabled
  1084. LOG.exception('Could not guess HiveServer2 URL: %s' % smart_str(e))
  1085. return 'jdbc:hive2://localhost:10000/default'
  1086. class HiveServer2Action(Action):
  1087. TYPE = 'hive2'
  1088. DEFAULT_CREDENTIALS = 'hive2'
  1089. FIELDS = {
  1090. 'script_path': {
  1091. 'name': 'script_path',
  1092. 'label': _('Script'),
  1093. 'value': '',
  1094. 'help_text': _('Path to the script on HDFS.'),
  1095. 'type': ''
  1096. },
  1097. 'parameters': {
  1098. 'name': 'parameters',
  1099. 'label': _('Parameters'),
  1100. 'value': [],
  1101. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  1102. 'type': ''
  1103. },
  1104. 'arguments': {
  1105. 'name': 'arguments',
  1106. 'label': _('Arguments'),
  1107. 'value': [],
  1108. 'help_text': _('Arguments for beeline. E.g. --showHeader=true, -Djavax.net.ssl.trustStore=/etc/cdep-ssl-conf/CA_STANDARD/truststore.jks'),
  1109. 'type': []
  1110. },
  1111. 'jdbc_url': {
  1112. 'name': 'jdbc_url',
  1113. 'label': _('HiveServer2 URL'),
  1114. 'value': "",
  1115. 'help_text': _('e.g. jdbc:hive2://localhost:10000/default. JDBC URL for the Hive Server 2.'),
  1116. 'type': ''
  1117. },
  1118. 'password': {
  1119. 'name': 'password',
  1120. 'label': _('Password'),
  1121. 'value': '',
  1122. 'help_text': _('The password element must contain the password of the current user. However, the password is only used if Hive Server 2 is backed by '
  1123. 'something requiring a password (e.g. LDAP); non-secured Hive Server 2 or Kerberized Hive Server 2 don\'t require a password.'),
  1124. 'type': ''
  1125. },
  1126. # Common
  1127. 'files': {
  1128. 'name': 'files',
  1129. 'label': _('Files'),
  1130. 'value': [],
  1131. 'help_text': _('Files put in the running directory.'),
  1132. 'type': ''
  1133. },
  1134. 'archives': {
  1135. 'name': 'archives',
  1136. 'label': _('Archives'),
  1137. 'value': [],
  1138. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1139. 'type': ''
  1140. },
  1141. 'job_properties': {
  1142. 'name': 'job_properties',
  1143. 'label': _('Hadoop job properties'),
  1144. 'value': [],
  1145. 'help_text': _('value, e.g. production'),
  1146. 'type': ''
  1147. },
  1148. 'prepares': {
  1149. 'name': 'prepares',
  1150. 'label': _('Prepares'),
  1151. 'value': [],
  1152. 'help_text': _('Path to manipulate before starting the application.'),
  1153. 'type': ''
  1154. },
  1155. 'job_xml': {
  1156. 'name': 'job_xml',
  1157. 'label': _('Job XML'),
  1158. 'value': '',
  1159. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1160. 'type': ''
  1161. },
  1162. 'retry_max': {
  1163. 'name': 'retry_max',
  1164. 'label': _('Max retry'),
  1165. 'value': [],
  1166. 'help_text': _('Number of times, default is 3'),
  1167. 'type': ''
  1168. },
  1169. 'retry_interval': {
  1170. 'name': 'retry_interval',
  1171. 'label': _('Retry interval'),
  1172. 'value': [],
  1173. 'help_text': _('Wait time in minutes, default is 10'),
  1174. 'type': ''
  1175. }
  1176. }
  1177. @classmethod
  1178. def get_mandatory_fields(cls):
  1179. return [cls.FIELDS['script_path']]
  1180. def _get_impala_url():
  1181. try:
  1182. from impala.dbms import get_query_server_config
  1183. return get_query_server_config()['server_host']
  1184. except Exception, e:
  1185. # Might fail is Impala is disabled
  1186. LOG.exception('Could not get Impalad URL: %s' % smart_str(e))
  1187. return 'localhost'
  1188. class ImpalaAction(HiveServer2Action):
  1189. # Executed as shell action until Oozie supports an Impala Action
  1190. TYPE = 'impala'
  1191. DEFAULT_CREDENTIALS = '' # None at this time, need to upload user keytab
  1192. FIELDS = HiveServer2Action.FIELDS.copy()
  1193. del FIELDS['jdbc_url']
  1194. del FIELDS['password']
  1195. FIELDS['impalad_host'] = {
  1196. 'name': 'impalad_host',
  1197. 'label': _('Impalad hostname'),
  1198. 'value': "",
  1199. 'help_text': _('e.g. impalad-001.cluster.com. The hostname of the Impalad to send the query to.'),
  1200. 'type': ''
  1201. }
  1202. FIELDS['key_tab_path'] = {
  1203. 'name': 'key_tab_path',
  1204. 'label': _('Keytab path'),
  1205. 'value': '${key_tab_path}' if SECURITY_ENABLED.get() else '',
  1206. 'help_text': _('Path to the keytab to use when on a secure cluster, e.g. /user/joe/joe.keytab.'),
  1207. 'type': ''
  1208. }
  1209. FIELDS['user_principal'] = {
  1210. 'name': 'user_principal',
  1211. 'label': _('User principal'),
  1212. 'value': '${user_principal}' if SECURITY_ENABLED.get() else '',
  1213. 'help_text': _('Name of the principal to use in the kinit, e.g.: kinit -k -t /home/joe/joe.keytab joe@PROD.EDH.'),
  1214. 'type': ''
  1215. }
  1216. class SubWorkflowAction(Action):
  1217. TYPE = 'subworkflow'
  1218. FIELDS = {
  1219. 'workflow': {
  1220. 'name': 'workflow',
  1221. 'label': _('Sub-workflow'),
  1222. 'value': None,
  1223. 'help_text': _('The sub-workflow application to include. You must own all the sub-workflows'),
  1224. 'type': 'workflow'
  1225. },
  1226. 'propagate_configuration': {
  1227. 'name': 'propagate_configuration',
  1228. 'label': _('Propagate configuration'),
  1229. 'value': True,
  1230. 'help_text': _('If the workflow job configuration should be propagated to the child workflow.'),
  1231. 'type': ''
  1232. },
  1233. 'job_properties': {
  1234. 'name': 'job_properties',
  1235. 'label': _('Hadoop job properties'),
  1236. 'value': [],
  1237. 'help_text': _('Can be used to specify the job properties that are required to run the child workflow job.'),
  1238. 'type': ''
  1239. },
  1240. 'retry_max': {
  1241. 'name': 'retry_max',
  1242. 'label': _('Max retry'),
  1243. 'value': [],
  1244. 'help_text': _('Number of times, default is 3'),
  1245. 'type': ''
  1246. },
  1247. 'retry_interval': {
  1248. 'name': 'retry_interval',
  1249. 'label': _('Retry interval'),
  1250. 'value': [],
  1251. 'help_text': _('Wait time in minutes, default is 10'),
  1252. 'type': ''
  1253. }
  1254. }
  1255. @classmethod
  1256. def get_mandatory_fields(cls):
  1257. return [cls.FIELDS['workflow']]
  1258. class SqoopAction(Action):
  1259. TYPE = 'sqoop'
  1260. FIELDS = {
  1261. 'command': {
  1262. 'name': 'command',
  1263. 'label': _('Sqoop command'),
  1264. 'value': 'import --connect jdbc:hsqldb:file:db.hsqldb --table TT --target-dir hdfs://localhost:8020/user/foo -m 1',
  1265. 'help_text': _('The full %(type)s command. Either put it here or split it by spaces and insert the parts as multiple parameters below.') % {'type': TYPE},
  1266. 'type': 'textarea'
  1267. },
  1268. 'arguments': {
  1269. 'name': 'arguments',
  1270. 'label': _('Arguments'),
  1271. 'value': [],
  1272. 'help_text': _('If no command is specified, split the command by spaces and insert the %(type)s parameters '
  1273. 'here e.g. import, --connect, jdbc:hsqldb:file:db.hsqldb, ...') % {'type': TYPE},
  1274. 'type': ''
  1275. },
  1276. # Common
  1277. 'files': {
  1278. 'name': 'files',
  1279. 'label': _('Files'),
  1280. 'value': [],
  1281. 'help_text': _('Files put in the running directory.'),
  1282. 'type': ''
  1283. },
  1284. 'archives': {
  1285. 'name': 'archives',
  1286. 'label': _('Archives'),
  1287. 'value': [],
  1288. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1289. 'type': ''
  1290. },
  1291. 'job_properties': {
  1292. 'name': 'job_properties',
  1293. 'label': _('Hadoop job properties'),
  1294. 'value': [],
  1295. 'help_text': _('value, e.g. production'),
  1296. 'type': ''
  1297. },
  1298. 'prepares': {
  1299. 'name': 'prepares',
  1300. 'label': _('Prepares'),
  1301. 'value': [],
  1302. 'help_text': _('Path to manipulate before starting the application.'),
  1303. 'type': ''
  1304. },
  1305. 'job_xml': {
  1306. 'name': 'job_xml',
  1307. 'label': _('Job XML'),
  1308. 'value': '',
  1309. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1310. 'type': ''
  1311. },
  1312. 'retry_max': {
  1313. 'name': 'retry_max',
  1314. 'label': _('Max retry'),
  1315. 'value': [],
  1316. 'help_text': _('Number of times, default is 3'),
  1317. 'type': ''
  1318. },
  1319. 'retry_interval': {
  1320. 'name': 'retry_interval',
  1321. 'label': _('Retry interval'),
  1322. 'value': [],
  1323. 'help_text': _('Wait time in minutes, default is 10'),
  1324. 'type': ''
  1325. }
  1326. }
  1327. @classmethod
  1328. def get_mandatory_fields(cls):
  1329. return [cls.FIELDS['command']]
  1330. class MapReduceAction(Action):
  1331. TYPE = 'mapreduce'
  1332. FIELDS = {
  1333. 'jar_path': {
  1334. 'name': 'jar_path',
  1335. 'label': _('Jar name'),
  1336. 'value': '',
  1337. 'help_text': _('Path to the jar on HDFS.'),
  1338. 'type': ''
  1339. },
  1340. # Common
  1341. 'files': {
  1342. 'name': 'files',
  1343. 'label': _('Files'),
  1344. 'value': [],
  1345. 'help_text': _('Files put in the running directory.'),
  1346. 'type': ''
  1347. },
  1348. 'archives': {
  1349. 'name': 'archives',
  1350. 'label': _('Archives'),
  1351. 'value': [],
  1352. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1353. 'type': ''
  1354. },
  1355. 'job_properties': {
  1356. 'name': 'job_properties',
  1357. 'label': _('Hadoop job properties'),
  1358. 'value': [],
  1359. 'help_text': _('value, e.g. production'),
  1360. 'type': ''
  1361. },
  1362. 'prepares': {
  1363. 'name': 'prepares',
  1364. 'label': _('Prepares'),
  1365. 'value': [],
  1366. 'help_text': _('Path to manipulate before starting the application.'),
  1367. 'type': ''
  1368. },
  1369. 'job_xml': {
  1370. 'name': 'job_xml',
  1371. 'label': _('Job XML'),
  1372. 'value': '',
  1373. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1374. 'type': ''
  1375. },
  1376. 'retry_max': {
  1377. 'name': 'retry_max',
  1378. 'label': _('Max retry'),
  1379. 'value': [],
  1380. 'help_text': _('Number of times, default is 3'),
  1381. 'type': ''
  1382. },
  1383. 'retry_interval': {
  1384. 'name': 'retry_interval',
  1385. 'label': _('Retry interval'),
  1386. 'value': [],
  1387. 'help_text': _('Wait time in minutes, default is 10'),
  1388. 'type': ''
  1389. }
  1390. }
  1391. @classmethod
  1392. def get_mandatory_fields(cls):
  1393. return [cls.FIELDS['jar_path']]
  1394. class ShellAction(Action):
  1395. TYPE = 'shell'
  1396. FIELDS = {
  1397. 'shell_command': {
  1398. 'name': 'shell_command',
  1399. 'label': _('Shell command'),
  1400. 'value': '',
  1401. 'help_text': _('Shell command to execute, e.g script.sh'),
  1402. 'type': ''
  1403. },
  1404. 'arguments': {
  1405. 'name': 'arguments',
  1406. 'label': _('Arguments'),
  1407. 'value': [],
  1408. 'help_text': _('One arg, e.g. -l, --help'),
  1409. 'type': ''
  1410. },
  1411. 'env_var': {
  1412. 'name': 'env_var',
  1413. 'label': _('Environment variables'),
  1414. 'value': [],
  1415. 'help_text': _('e.g. MAX=10 or PATH=$PATH:mypath'),
  1416. 'type': ''
  1417. },
  1418. 'capture_output': {
  1419. 'name': 'capture_output',
  1420. 'label': _('Capture output'),
  1421. 'value': True,
  1422. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  1423. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  1424. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  1425. 'via the String action:output(String node, String key) function') % {'program': TYPE},
  1426. 'type': ''
  1427. },
  1428. # Common
  1429. 'files': {
  1430. 'name': 'files',
  1431. 'label': _('Files'),
  1432. 'value': [],
  1433. 'help_text': _('Files put in the running directory.'),
  1434. 'type': ''
  1435. },
  1436. 'archives': {
  1437. 'name': 'archives',
  1438. 'label': _('Archives'),
  1439. 'value': [],
  1440. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1441. 'type': ''
  1442. },
  1443. 'job_properties': {
  1444. 'name': 'job_properties',
  1445. 'label': _('Hadoop job properties'),
  1446. 'value': [],
  1447. 'help_text': _('value, e.g. production'),
  1448. 'type': ''
  1449. },
  1450. 'prepares': {
  1451. 'name': 'prepares',
  1452. 'label': _('Prepares'),
  1453. 'value': [],
  1454. 'help_text': _('Path to manipulate before starting the application.'),
  1455. 'type': ''
  1456. },
  1457. 'job_xml': {
  1458. 'name': 'job_xml',
  1459. 'label': _('Job XML'),
  1460. 'value': '',
  1461. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1462. 'type': ''
  1463. },
  1464. 'retry_max': {
  1465. 'name': 'retry_max',
  1466. 'label': _('Max retry'),
  1467. 'value': [],
  1468. 'help_text': _('Number of times, default is 3'),
  1469. 'type': ''
  1470. },
  1471. 'retry_interval': {
  1472. 'name': 'retry_interval',
  1473. 'label': _('Retry interval'),
  1474. 'value': [],
  1475. 'help_text': _('Wait time in minutes, default is 10'),
  1476. 'type': ''
  1477. }
  1478. }
  1479. @classmethod
  1480. def get_mandatory_fields(cls):
  1481. return [cls.FIELDS['shell_command']]
  1482. class SshAction(Action):
  1483. TYPE = 'ssh'
  1484. FIELDS = {
  1485. 'host': {
  1486. 'name': 'host',
  1487. 'label': _('User and Host'),
  1488. 'value': 'user@host.com',
  1489. 'help_text': _('Where the shell will be executed.'),
  1490. 'type': 'text'
  1491. },
  1492. 'ssh_command': {
  1493. 'name': 'ssh_command',
  1494. 'label': _('Ssh command'),
  1495. 'value': 'ls',
  1496. 'help_text': _('The path of the Shell command to execute.'),
  1497. 'type': 'textarea'
  1498. },
  1499. 'arguments': {
  1500. 'name': 'arguments',
  1501. 'label': _('Arguments'),
  1502. 'value': [],
  1503. 'help_text': _('One arg, e.g. -l, --help'),
  1504. 'type': ''
  1505. },
  1506. 'capture_output': {
  1507. 'name': 'capture_output',
  1508. 'label': _('Capture output'),
  1509. 'value': True,
  1510. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  1511. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  1512. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  1513. 'via the String action:output(String node, String key) function') % {'program': TYPE},
  1514. 'type': ''
  1515. },
  1516. # Common
  1517. 'retry_max': {
  1518. 'name': 'retry_max',
  1519. 'label': _('Max retry'),
  1520. 'value': [],
  1521. 'help_text': _('Number of times, default is 3'),
  1522. 'type': ''
  1523. },
  1524. 'retry_interval': {
  1525. 'name': 'retry_interval',
  1526. 'label': _('Retry interval'),
  1527. 'value': [],
  1528. 'help_text': _('Wait time in minutes, default is 10'),
  1529. 'type': ''
  1530. }
  1531. }
  1532. @classmethod
  1533. def get_mandatory_fields(cls):
  1534. return [cls.FIELDS['host'], cls.FIELDS['ssh_command']]
  1535. class FsAction(Action):
  1536. TYPE = 'fs'
  1537. FIELDS = {
  1538. 'deletes': {
  1539. 'name': 'deletes',
  1540. 'label': _('Delete path'),
  1541. 'value': [],
  1542. 'help_text': _('Deletes recursively all content.'),
  1543. 'type': ''
  1544. },
  1545. 'mkdirs': {
  1546. 'name': 'mkdirs',
  1547. 'label': _('Create directory'),
  1548. 'value': [],
  1549. 'help_text': _('Sub directories are created if needed.'),
  1550. 'type': ''
  1551. },
  1552. 'moves': {
  1553. 'name': 'moves',
  1554. 'label': _('Move file or directory'),
  1555. 'value': [],
  1556. 'help_text': _('Destination.'),
  1557. 'type': ''
  1558. },
  1559. 'chmods': {
  1560. 'name': 'chmods',
  1561. 'label': _('Change permissions'),
  1562. 'value': [],
  1563. 'help_text': _('File or directory.'),
  1564. 'type': ''
  1565. },
  1566. 'touchzs': {
  1567. 'name': 'touchzs',
  1568. 'label': _('Create or touch a file'),
  1569. 'value': [],
  1570. 'help_text': _('Or update its modification date.'),
  1571. 'type': ''
  1572. },
  1573. 'chgrps': {
  1574. 'name': 'chgrps',
  1575. 'label': _('Change the group'),
  1576. 'value': [],
  1577. 'help_text': _('File or directory.'),
  1578. 'type': ''
  1579. },
  1580. # Common
  1581. 'retry_max': {
  1582. 'name': 'retry_max',
  1583. 'label': _('Max retry'),
  1584. 'value': [],
  1585. 'help_text': _('Number of times, default is 3'),
  1586. 'type': ''
  1587. },
  1588. 'retry_interval': {
  1589. 'name': 'retry_interval',
  1590. 'label': _('Retry interval'),
  1591. 'value': [],
  1592. 'help_text': _('Wait time in minutes, default is 10'),
  1593. 'type': ''
  1594. }
  1595. }
  1596. @classmethod
  1597. def get_mandatory_fields(cls):
  1598. return []
  1599. class EmailAction(Action):
  1600. TYPE = 'email'
  1601. FIELDS = {
  1602. 'to': {
  1603. 'name': 'to',
  1604. 'label': _('To addresses'),
  1605. 'value': '',
  1606. 'help_text': _('Comma-separated values'),
  1607. 'type': 'text'
  1608. },
  1609. 'cc': {
  1610. 'name': 'cc',
  1611. 'label': _('cc'),
  1612. 'value': '',
  1613. 'help_text': _('Comma-separated values'),
  1614. 'type': 'text'
  1615. },
  1616. 'bcc': {
  1617. 'name': 'bcc',
  1618. 'label': _('bcc'),
  1619. 'value': '',
  1620. 'help_text': _('Comma-separated values'),
  1621. 'type': 'text'
  1622. },
  1623. 'subject': {
  1624. 'name': 'subject',
  1625. 'label': _('Subject'),
  1626. 'value': '',
  1627. 'help_text': _('Plain-text'),
  1628. 'type': 'text'
  1629. },
  1630. 'body': {
  1631. 'name': 'body',
  1632. 'label': _('Body'),
  1633. 'value': '',
  1634. 'help_text': _('Plain-text'),
  1635. 'type': 'textarea'
  1636. },
  1637. 'attachment': {
  1638. 'name': 'attachment',
  1639. 'label': _('Attachment'),
  1640. 'value': '',
  1641. 'help_text': _('Comma separated list of HDFS files.'),
  1642. 'type': ''
  1643. },
  1644. 'content_type': {
  1645. 'name': 'content_type',
  1646. 'label': _('Content-type'),
  1647. 'value': 'text/plain',
  1648. 'help_text': _('Default is text/plain'),
  1649. 'type': 'text'
  1650. },
  1651. # Common
  1652. 'retry_max': {
  1653. 'name': 'retry_max',
  1654. 'label': _('Max retry'),
  1655. 'value': [],
  1656. 'help_text': _('Number of times, default is 3'),
  1657. 'type': ''
  1658. },
  1659. 'retry_interval': {
  1660. 'name': 'retry_interval',
  1661. 'label': _('Retry interval'),
  1662. 'value': [],
  1663. 'help_text': _('Wait time in minutes, default is 10'),
  1664. 'type': ''
  1665. }
  1666. }
  1667. @classmethod
  1668. def get_mandatory_fields(cls):
  1669. return [cls.FIELDS['to'], cls.FIELDS['subject'], cls.FIELDS['body']]
  1670. class StreamingAction(Action):
  1671. TYPE = 'streaming'
  1672. FIELDS = {
  1673. 'mapper': {
  1674. 'name': 'mapper',
  1675. 'label': _('Mapper'),
  1676. 'value': '',
  1677. 'help_text': _('The executable/script to be used as mapper.'),
  1678. 'type': ''
  1679. },
  1680. 'reducer': {
  1681. 'name': 'reducer',
  1682. 'label': _('Reducer'),
  1683. 'value': '',
  1684. 'help_text': _('The executable/script to be used as reducer.'),
  1685. 'type': ''
  1686. },
  1687. # Common
  1688. 'files': {
  1689. 'name': 'files',
  1690. 'label': _('Files'),
  1691. 'value': [],
  1692. 'help_text': _('Files put in the running directory.')
  1693. },
  1694. 'archives': {
  1695. 'name': 'archives',
  1696. 'label': _('Archives'),
  1697. 'value': [],
  1698. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.')
  1699. },
  1700. 'job_properties': {
  1701. 'name': 'job_properties',
  1702. 'label': _('Hadoop job properties'),
  1703. 'value': [],
  1704. 'help_text': _('value, e.g. production')
  1705. },
  1706. 'prepares': {
  1707. 'name': 'prepares',
  1708. 'label': _('Prepares'),
  1709. 'value': [],
  1710. 'help_text': _('Path to manipulate before starting the application.')
  1711. },
  1712. 'job_xml': {
  1713. 'name': 'job_xml',
  1714. 'label': _('Job XML'),
  1715. 'value': '',
  1716. 'help_text': _('Refer to a Hadoop JobConf job.xml')
  1717. },
  1718. 'retry_max': {
  1719. 'name': 'retry_max',
  1720. 'label': _('Max retry'),
  1721. 'value': [],
  1722. 'help_text': _('Number of times, default is 3'),
  1723. 'type': ''
  1724. },
  1725. 'retry_interval': {
  1726. 'name': 'retry_interval',
  1727. 'label': _('Retry interval'),
  1728. 'value': [],
  1729. 'help_text': _('Wait time in minutes, default is 10'),
  1730. 'type': ''
  1731. }
  1732. }
  1733. @classmethod
  1734. def get_mandatory_fields(cls):
  1735. return [cls.FIELDS['mapper'], cls.FIELDS['reducer']]
  1736. class DistCpAction(Action):
  1737. TYPE = 'distcp'
  1738. FIELDS = {
  1739. 'distcp_parameters': {
  1740. 'name': 'distcp_parameters',
  1741. 'label': _('Arguments'),
  1742. 'value': [{'value': ''}, {'value': ''}],
  1743. 'help_text': _('Options first, then source / destination paths'),
  1744. 'type': 'distcp'
  1745. },
  1746. # Common
  1747. 'prepares': {
  1748. 'name': 'prepares',
  1749. 'label': _('Prepares'),
  1750. 'value': [],
  1751. 'help_text': _('Path to manipulate before starting the application.')
  1752. },
  1753. 'job_properties': {
  1754. 'name': 'job_properties',
  1755. 'label': _('Hadoop job properties'),
  1756. 'value': [],
  1757. 'help_text': _('value, e.g. production')
  1758. },
  1759. 'java_opts': {
  1760. 'name': 'java_opts',
  1761. 'label': _('Java options'),
  1762. 'value': '',
  1763. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b')
  1764. },
  1765. 'retry_max': {
  1766. 'name': 'retry_max',
  1767. 'label': _('Max retry'),
  1768. 'value': [],
  1769. 'help_text': _('Number of times, default is 3'),
  1770. 'type': ''
  1771. },
  1772. 'retry_interval': {
  1773. 'name': 'retry_interval',
  1774. 'label': _('Retry interval'),
  1775. 'value': [],
  1776. 'help_text': _('Wait time in minutes, default is 10'),
  1777. 'type': ''
  1778. }
  1779. }
  1780. @classmethod
  1781. def get_mandatory_fields(cls):
  1782. return [cls.FIELDS['distcp_parameters']]
  1783. class SparkAction(Action):
  1784. TYPE = 'spark'
  1785. FIELDS = {
  1786. 'spark_master': {
  1787. 'name': 'spark_master',
  1788. 'label': _('Spark Master'),
  1789. 'value': 'yarn',
  1790. 'help_text': _('Ex: spark://host:port, mesos://host:port, yarn, or local.'),
  1791. 'type': ''
  1792. },
  1793. 'mode': {
  1794. 'name': 'mode',
  1795. 'label': _('Mode'),
  1796. 'value': 'client',
  1797. 'help_text': _('e.g. Client cluster'),
  1798. 'type': ''
  1799. },
  1800. 'app_name': {
  1801. 'name': 'app_name',
  1802. 'label': _('App name'),
  1803. 'value': 'MySpark',
  1804. 'help_text': _('The name of the spark application'),
  1805. 'type': ''
  1806. },
  1807. 'files': {
  1808. 'name': 'files',
  1809. 'label': _('Files'),
  1810. 'value': [],
  1811. 'help_text': _('Path to file to put in the running directory.'),
  1812. 'type': ''
  1813. },
  1814. 'class': {
  1815. 'name': 'class',
  1816. 'label': _('Main class'),
  1817. 'value': '',
  1818. 'help_text': _("Only if using jars, e.g. org.apache.spark.examples.mllib.JavaALS"),
  1819. 'type': 'text'
  1820. },
  1821. 'jars': {
  1822. 'name': 'jars',
  1823. 'label': _('Jar/py name'),
  1824. 'value': '',
  1825. 'help_text': _('Name of main file added in Files.'),
  1826. 'type': 'text'
  1827. },
  1828. 'spark_opts': {
  1829. 'name': 'spark_opts',
  1830. 'label': _('Options list'),
  1831. 'value': '',
  1832. 'help_text': _('Ex: --executor-memory 20G --num-executors 50'),
  1833. 'type': ''
  1834. },
  1835. 'spark_arguments': {
  1836. 'name': 'spark_arguments',
  1837. 'label': _('Arguments'),
  1838. 'value': [],
  1839. 'help_text': _('Arguments, one by one, e.g. 1000, /path/a.')
  1840. },
  1841. # Common
  1842. 'job_properties': {
  1843. 'name': 'job_properties',
  1844. 'label': _('Hadoop job properties'),
  1845. 'value': [],
  1846. 'help_text': _('value, e.g. production')
  1847. },
  1848. 'prepares': {
  1849. 'name': 'prepares',
  1850. 'label': _('Prepares'),
  1851. 'value': [],
  1852. 'help_text': _('Path to manipulate before starting the application.')
  1853. },
  1854. 'job_xml': {
  1855. 'name': 'job_xml',
  1856. 'label': _('Job XML'),
  1857. 'value': '',
  1858. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1859. 'type': ''
  1860. },
  1861. 'retry_max': {
  1862. 'name': 'retry_max',
  1863. 'label': _('Max retry'),
  1864. 'value': [],
  1865. 'help_text': _('Number of times, default is 3'),
  1866. 'type': ''
  1867. },
  1868. 'retry_interval': {
  1869. 'name': 'retry_interval',
  1870. 'label': _('Retry interval'),
  1871. 'value': [],
  1872. 'help_text': _('Wait time in minutes, default is 10'),
  1873. 'type': ''
  1874. }
  1875. }
  1876. @classmethod
  1877. def get_mandatory_fields(cls):
  1878. return [cls.FIELDS['files'], cls.FIELDS['jars']]
  1879. class KillAction(Action):
  1880. TYPE = 'kill'
  1881. FIELDS = {
  1882. 'message': {
  1883. 'name': 'message',
  1884. 'label': _('Message'),
  1885. 'value': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]'),
  1886. 'help_text': _('Message to display when the workflow fails. Can contain some EL functions.'),
  1887. 'type': 'textarea'
  1888. }
  1889. }
  1890. @classmethod
  1891. def get_mandatory_fields(cls):
  1892. return [cls.FIELDS['message']]
  1893. class JoinAction(Action):
  1894. TYPE = 'join'
  1895. FIELDS = {}
  1896. @classmethod
  1897. def get_mandatory_fields(cls):
  1898. return []
  1899. class GenericAction(Action):
  1900. TYPE = 'generic'
  1901. FIELDS = {
  1902. 'xml': {
  1903. 'name': 'xml',
  1904. 'label': _('XML of the action'),
  1905. 'value': '<my_action>\n</my_action>',
  1906. 'help_text': _('Insert verbatim the XML of the action to insert into the workflow.'),
  1907. 'type': 'textarea'
  1908. }
  1909. }
  1910. @classmethod
  1911. def get_mandatory_fields(cls):
  1912. return [cls.FIELDS['xml']]
  1913. class ForkNode(Action):
  1914. TYPE = 'fork'
  1915. FIELDS = {}
  1916. @classmethod
  1917. def get_mandatory_fields(cls):
  1918. return []
  1919. class HiveDocumentAction(Action):
  1920. TYPE = 'hive-document'
  1921. DEFAULT_CREDENTIALS = 'hive2'
  1922. FIELDS = {
  1923. 'uuid': {
  1924. 'name': 'uuid',
  1925. 'label': _('Hive query'),
  1926. 'value': '',
  1927. 'help_text': _('Select a saved Hive query you want to schedule.'),
  1928. 'type': 'hive'
  1929. },
  1930. 'parameters': {
  1931. 'name': 'parameters',
  1932. 'label': _('Parameters'),
  1933. 'value': [],
  1934. 'help_text': _('The parameters of the script. E.g. N=5, INPUT=${inputDir}'),
  1935. 'type': ''
  1936. },
  1937. # Common
  1938. 'jdbc_url': {
  1939. 'name': 'jdbc_url',
  1940. 'label': _('HiveServer2 URL'),
  1941. 'value': "",
  1942. 'help_text': _('e.g. jdbc:hive2://localhost:10000/default. JDBC URL for the Hive Server 2.'),
  1943. 'type': ''
  1944. },
  1945. 'password': {
  1946. 'name': 'password',
  1947. 'label': _('Password'),
  1948. 'value': '',
  1949. 'help_text': _('The password element must contain the password of the current user. However, the password is only used if Hive Server 2 is backed by '
  1950. 'something requiring a password (e.g. LDAP); non-secured Hive Server 2 or Kerberized Hive Server 2 don\'t require a password.'),
  1951. 'type': ''
  1952. },
  1953. 'files': {
  1954. 'name': 'files',
  1955. 'label': _('Files'),
  1956. 'value': [],
  1957. 'help_text': _('Files put in the running directory.'),
  1958. 'type': ''
  1959. },
  1960. 'archives': {
  1961. 'name': 'archives',
  1962. 'label': _('Archives'),
  1963. 'value': [],
  1964. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  1965. 'type': ''
  1966. },
  1967. 'job_properties': {
  1968. 'name': 'job_properties',
  1969. 'label': _('Hadoop job properties'),
  1970. 'value': [],
  1971. 'help_text': _('value, e.g. production'),
  1972. 'type': ''
  1973. },
  1974. 'prepares': {
  1975. 'name': 'prepares',
  1976. 'label': _('Prepares'),
  1977. 'value': [],
  1978. 'help_text': _('Path to manipulate before starting the application.'),
  1979. 'type': ''
  1980. },
  1981. 'job_xml': {
  1982. 'name': 'job_xml',
  1983. 'label': _('Job XML'),
  1984. 'value': '',
  1985. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  1986. 'type': ''
  1987. },
  1988. 'retry_max': {
  1989. 'name': 'retry_max',
  1990. 'label': _('Max retry'),
  1991. 'value': [],
  1992. 'help_text': _('Number of times, default is 3'),
  1993. 'type': ''
  1994. },
  1995. 'retry_interval': {
  1996. 'name': 'retry_interval',
  1997. 'label': _('Retry interval'),
  1998. 'value': [],
  1999. 'help_text': _('Wait time in minutes, default is 10'),
  2000. 'type': ''
  2001. }
  2002. }
  2003. @classmethod
  2004. def get_mandatory_fields(cls):
  2005. return [cls.FIELDS['uuid']]
  2006. class ImpalaDocumentAction(HiveDocumentAction):
  2007. TYPE = 'impala-document'
  2008. DEFAULT_CREDENTIALS = '' # None at this time, need to upload user keytab
  2009. FIELDS = HiveServer2Action.FIELDS.copy()
  2010. del FIELDS['jdbc_url']
  2011. del FIELDS['password']
  2012. FIELDS['impalad_host'] = {
  2013. 'name': 'impalad_host',
  2014. 'label': _('Impalad hostname'),
  2015. 'value': "",
  2016. 'help_text': _('e.g. impalad-001.cluster.com (optional)'),
  2017. 'type': ''
  2018. }
  2019. FIELDS['key_tab_path'] = {
  2020. 'name': 'key_tab_path',
  2021. 'label': _('Keytab path'),
  2022. 'value': '${key_tab_path}' if SECURITY_ENABLED.get() else '',
  2023. 'help_text': _('Path to the keytab to use when on a secure cluster, e.g. /user/joe/joe.keytab.'),
  2024. 'type': ''
  2025. }
  2026. FIELDS['user_principal'] = {
  2027. 'name': 'user_principal',
  2028. 'label': _('User principal'),
  2029. 'value': '${user_principal}' if SECURITY_ENABLED.get() else '',
  2030. 'help_text': _('Name of the principal to use in the kinit, e.g.: kinit -k -t /home/joe/joe.keytab joe@PROD.EDH.'),
  2031. 'type': ''
  2032. }
  2033. FIELDS['uuid'] = {
  2034. 'name': 'uuid',
  2035. 'label': _('Impala query'),
  2036. 'value': '',
  2037. 'help_text': _('Select a saved Impala query you want to schedule.'),
  2038. 'type': 'impala'
  2039. }
  2040. class JavaDocumentAction(Action):
  2041. TYPE = 'java-document'
  2042. FIELDS = {
  2043. 'uuid': {
  2044. 'name': 'uuid',
  2045. 'label': _('Java program'),
  2046. 'value': '',
  2047. 'help_text': _('Select a saved Java program you want to schedule.'),
  2048. 'type': 'java'
  2049. },
  2050. 'arguments': {
  2051. 'name': 'arguments',
  2052. 'label': _('Arguments'),
  2053. 'value': [],
  2054. 'help_text': _('Arguments of the main method. The value of each arg element is considered a single argument '
  2055. 'and they are passed to the main method in the same order.'),
  2056. 'type': ''
  2057. },
  2058. 'java_opts': {
  2059. 'name': 'java_opts',
  2060. 'label': _('Java options'),
  2061. 'value': [],
  2062. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b'),
  2063. 'type': ''
  2064. },
  2065. 'capture_output': {
  2066. 'name': 'capture_output',
  2067. 'label': _('Capture output'),
  2068. 'value': False,
  2069. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  2070. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  2071. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  2072. 'via the String action:output(String node, String key) function') % {'program': TYPE.title()},
  2073. 'type': ''
  2074. },
  2075. # Common
  2076. 'files': {
  2077. 'name': 'files',
  2078. 'label': _('Files'),
  2079. 'value': [],
  2080. 'help_text': _('Files put in the running directory.'),
  2081. 'type': ''
  2082. },
  2083. 'archives': {
  2084. 'name': 'archives',
  2085. 'label': _('Archives'),
  2086. 'value': [],
  2087. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  2088. 'type': ''
  2089. },
  2090. 'job_properties': {
  2091. 'name': 'job_properties',
  2092. 'label': _('Hadoop job properties'),
  2093. 'value': [],
  2094. 'help_text': _('value, e.g. production'),
  2095. 'type': ''
  2096. },
  2097. 'prepares': {
  2098. 'name': 'prepares',
  2099. 'label': _('Prepares'),
  2100. 'value': [],
  2101. 'help_text': _('Path to manipulate before starting the application.'),
  2102. 'type': ''
  2103. },
  2104. 'job_xml': {
  2105. 'name': 'job_xml',
  2106. 'label': _('Job XML'),
  2107. 'value': [],
  2108. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2109. 'type': ''
  2110. },
  2111. 'retry_max': {
  2112. 'name': 'retry_max',
  2113. 'label': _('Max retry'),
  2114. 'value': [],
  2115. 'help_text': _('Number of times, default is 3'),
  2116. 'type': ''
  2117. },
  2118. 'retry_interval': {
  2119. 'name': 'retry_interval',
  2120. 'label': _('Retry interval'),
  2121. 'value': [],
  2122. 'help_text': _('Wait time in minutes, default is 10'),
  2123. 'type': ''
  2124. }
  2125. }
  2126. @classmethod
  2127. def get_mandatory_fields(cls):
  2128. return [cls.FIELDS['uuid']]
  2129. class SparkDocumentAction(Action):
  2130. TYPE = 'spark2-document'
  2131. FIELDS = {
  2132. 'uuid': {
  2133. 'name': 'uuid',
  2134. 'label': _('Spark program'),
  2135. 'value': '',
  2136. 'help_text': _('Select a saved Spark program you want to schedule.'),
  2137. 'type': 'spark'
  2138. },
  2139. 'spark_master': {
  2140. 'name': 'spark_master',
  2141. 'label': _('Spark Master'),
  2142. 'value': 'yarn',
  2143. 'help_text': _('Ex: spark://host:port, mesos://host:port, yarn, or local.'),
  2144. 'type': ''
  2145. },
  2146. 'mode': {
  2147. 'name': 'mode',
  2148. 'label': _('Mode'),
  2149. 'value': 'client',
  2150. 'help_text': _('e.g. Client cluster'),
  2151. 'type': ''
  2152. },
  2153. 'files': {
  2154. 'name': 'files',
  2155. 'label': _('Files'),
  2156. 'value': [],
  2157. 'help_text': _('Path to file to put in the running directory.'),
  2158. 'type': ''
  2159. },
  2160. 'spark_arguments': {
  2161. 'name': 'spark_arguments',
  2162. 'label': _('Arguments'),
  2163. 'value': [],
  2164. 'help_text': _('Arguments, one by one, e.g. 1000, /path/a.')
  2165. },
  2166. 'parameters': { # For Oozie Action Document
  2167. 'name': 'parameters',
  2168. 'label': _('Parameters'),
  2169. 'value': [],
  2170. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  2171. 'type': ''
  2172. },
  2173. # Common
  2174. 'job_properties': {
  2175. 'name': 'job_properties',
  2176. 'label': _('Hadoop job properties'),
  2177. 'value': [],
  2178. 'help_text': _('value, e.g. production')
  2179. },
  2180. 'prepares': {
  2181. 'name': 'prepares',
  2182. 'label': _('Prepares'),
  2183. 'value': [],
  2184. 'help_text': _('Path to manipulate before starting the application.')
  2185. },
  2186. 'job_xml': {
  2187. 'name': 'job_xml',
  2188. 'label': _('Job XML'),
  2189. 'value': '',
  2190. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2191. 'type': ''
  2192. },
  2193. 'retry_max': {
  2194. 'name': 'retry_max',
  2195. 'label': _('Max retry'),
  2196. 'value': [],
  2197. 'help_text': _('Number of times, default is 3'),
  2198. 'type': ''
  2199. },
  2200. 'retry_interval': {
  2201. 'name': 'retry_interval',
  2202. 'label': _('Retry interval'),
  2203. 'value': [],
  2204. 'help_text': _('Wait time in minutes, default is 10'),
  2205. 'type': ''
  2206. }
  2207. }
  2208. @classmethod
  2209. def get_mandatory_fields(cls):
  2210. return [cls.FIELDS['uuid']]
  2211. class PigDocumentAction(Action):
  2212. TYPE = 'pig-document'
  2213. FIELDS = {
  2214. 'uuid': {
  2215. 'name': 'uuid',
  2216. 'label': _('Pig script'),
  2217. 'value': '',
  2218. 'help_text': _('Select a saved Spark program you want to schedule.'),
  2219. 'type': 'pig'
  2220. },
  2221. # Common
  2222. 'files': {
  2223. 'name': 'files',
  2224. 'label': _('Files'),
  2225. 'value': [],
  2226. 'help_text': _('Files put in the running directory.'),
  2227. 'type': ''
  2228. },
  2229. 'archives': {
  2230. 'name': 'archives',
  2231. 'label': _('Archives'),
  2232. 'value': [],
  2233. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  2234. 'type': ''
  2235. },
  2236. 'job_properties': {
  2237. 'name': 'job_properties',
  2238. 'label': _('Hadoop job properties'),
  2239. 'value': [],
  2240. 'help_text': _('value, e.g. production'),
  2241. 'type': ''
  2242. },
  2243. 'prepares': {
  2244. 'name': 'prepares',
  2245. 'label': _('Prepares'),
  2246. 'value': [],
  2247. 'help_text': _('Path to manipulate before starting the application.'),
  2248. 'type': ''
  2249. },
  2250. 'job_xml': {
  2251. 'name': 'job_xml',
  2252. 'label': _('Job XML'),
  2253. 'value': [],
  2254. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2255. 'type': ''
  2256. },
  2257. 'retry_max': {
  2258. 'name': 'retry_max',
  2259. 'label': _('Max retry'),
  2260. 'value': [],
  2261. 'help_text': _('Number of times, default is 3'),
  2262. 'type': ''
  2263. },
  2264. 'retry_interval': {
  2265. 'name': 'retry_interval',
  2266. 'label': _('Retry interval'),
  2267. 'value': [],
  2268. 'help_text': _('Wait time in minutes, default is 10'),
  2269. 'type': ''
  2270. }
  2271. }
  2272. @classmethod
  2273. def get_mandatory_fields(cls):
  2274. return [cls.FIELDS['uuid']]
  2275. class SqoopDocumentAction(Action):
  2276. TYPE = 'sqoop-document'
  2277. FIELDS = {
  2278. 'uuid': {
  2279. 'name': 'uuid',
  2280. 'label': _('Sqoop command'),
  2281. 'value': '',
  2282. 'help_text': _('Select a saved Sqoop program you want to schedule.'),
  2283. 'type': 'sqoop'
  2284. },
  2285. 'parameters': {
  2286. 'name': 'parameters',
  2287. 'label': _('Parameters'),
  2288. 'value': [],
  2289. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  2290. 'type': ''
  2291. },
  2292. # Common
  2293. 'files': {
  2294. 'name': 'files',
  2295. 'label': _('Files'),
  2296. 'value': [],
  2297. 'help_text': _('Files put in the running directory.'),
  2298. 'type': ''
  2299. },
  2300. 'archives': {
  2301. 'name': 'archives',
  2302. 'label': _('Archives'),
  2303. 'value': [],
  2304. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  2305. 'type': ''
  2306. },
  2307. 'job_properties': {
  2308. 'name': 'job_properties',
  2309. 'label': _('Hadoop job properties'),
  2310. 'value': [],
  2311. 'help_text': _('value, e.g. production'),
  2312. 'type': ''
  2313. },
  2314. 'prepares': {
  2315. 'name': 'prepares',
  2316. 'label': _('Prepares'),
  2317. 'value': [],
  2318. 'help_text': _('Path to manipulate before starting the application.'),
  2319. 'type': ''
  2320. },
  2321. 'job_xml': {
  2322. 'name': 'job_xml',
  2323. 'label': _('Job XML'),
  2324. 'value': '',
  2325. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2326. 'type': ''
  2327. },
  2328. 'retry_max': {
  2329. 'name': 'retry_max',
  2330. 'label': _('Max retry'),
  2331. 'value': [],
  2332. 'help_text': _('Number of times, default is 3'),
  2333. 'type': ''
  2334. },
  2335. 'retry_interval': {
  2336. 'name': 'retry_interval',
  2337. 'label': _('Retry interval'),
  2338. 'value': [],
  2339. 'help_text': _('Wait time in minutes, default is 10'),
  2340. 'type': ''
  2341. }
  2342. }
  2343. @classmethod
  2344. def get_mandatory_fields(cls):
  2345. return [cls.FIELDS['uuid']]
  2346. class DistCpDocumentAction(Action):
  2347. TYPE = 'distcp-document'
  2348. FIELDS = {
  2349. 'uuid': {
  2350. 'name': 'uuid',
  2351. 'label': _('DistCp program'),
  2352. 'value': '',
  2353. 'help_text': _('Select a saved DistCp program you want to schedule.'),
  2354. 'type': 'distcp-doc'
  2355. },
  2356. 'parameters': {
  2357. 'name': 'parameters',
  2358. 'label': _('Parameters'),
  2359. 'value': [],
  2360. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  2361. 'type': ''
  2362. },
  2363. # Common
  2364. 'prepares': {
  2365. 'name': 'prepares',
  2366. 'label': _('Prepares'),
  2367. 'value': [],
  2368. 'help_text': _('Path to manipulate before starting the application.')
  2369. },
  2370. 'job_properties': {
  2371. 'name': 'job_properties',
  2372. 'label': _('Hadoop job properties'),
  2373. 'value': [],
  2374. 'help_text': _('value, e.g. production')
  2375. },
  2376. 'java_opts': {
  2377. 'name': 'java_opts',
  2378. 'label': _('Java options'),
  2379. 'value': '',
  2380. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b')
  2381. },
  2382. 'retry_max': {
  2383. 'name': 'retry_max',
  2384. 'label': _('Max retry'),
  2385. 'value': [],
  2386. 'help_text': _('Number of times, default is 3'),
  2387. 'type': ''
  2388. },
  2389. 'retry_interval': {
  2390. 'name': 'retry_interval',
  2391. 'label': _('Retry interval'),
  2392. 'value': [],
  2393. 'help_text': _('Wait time in minutes, default is 10'),
  2394. 'type': ''
  2395. }
  2396. }
  2397. @classmethod
  2398. def get_mandatory_fields(cls):
  2399. return [cls.FIELDS['uuid']]
  2400. class ShellDocumentAction(Action):
  2401. TYPE = 'shell-document'
  2402. FIELDS = {
  2403. 'uuid': {
  2404. 'name': 'uuid',
  2405. 'label': _('Shell program'),
  2406. 'value': '',
  2407. 'help_text': _('Select a saved Shell program you want to schedule.'),
  2408. 'type': 'shell-doc'
  2409. },
  2410. 'parameters': {
  2411. 'name': 'parameters',
  2412. 'label': _('Parameters'),
  2413. 'value': [],
  2414. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  2415. 'type': ''
  2416. },
  2417. # Common
  2418. 'files': {
  2419. 'name': 'files',
  2420. 'label': _('Files'),
  2421. 'value': [],
  2422. 'help_text': _('Files put in the running directory.'),
  2423. 'type': ''
  2424. },
  2425. 'archives': {
  2426. 'name': 'archives',
  2427. 'label': _('Archives'),
  2428. 'value': [],
  2429. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  2430. 'type': ''
  2431. },
  2432. 'job_properties': {
  2433. 'name': 'job_properties',
  2434. 'label': _('Hadoop job properties'),
  2435. 'value': [],
  2436. 'help_text': _('value, e.g. production'),
  2437. 'type': ''
  2438. },
  2439. 'prepares': {
  2440. 'name': 'prepares',
  2441. 'label': _('Prepares'),
  2442. 'value': [],
  2443. 'help_text': _('Path to manipulate before starting the application.'),
  2444. 'type': ''
  2445. },
  2446. 'job_xml': {
  2447. 'name': 'job_xml',
  2448. 'label': _('Job XML'),
  2449. 'value': '',
  2450. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2451. 'type': ''
  2452. },
  2453. 'retry_max': {
  2454. 'name': 'retry_max',
  2455. 'label': _('Max retry'),
  2456. 'value': [],
  2457. 'help_text': _('Number of times, default is 3'),
  2458. 'type': ''
  2459. },
  2460. 'retry_interval': {
  2461. 'name': 'retry_interval',
  2462. 'label': _('Retry interval'),
  2463. 'value': [],
  2464. 'help_text': _('Wait time in minutes, default is 10'),
  2465. 'type': ''
  2466. }
  2467. }
  2468. @classmethod
  2469. def get_mandatory_fields(cls):
  2470. return [cls.FIELDS['uuid']]
  2471. class MapReduceDocumentAction(Action):
  2472. TYPE = 'mapreduce-document'
  2473. FIELDS = {
  2474. 'uuid': {
  2475. 'name': 'uuid',
  2476. 'label': _('MapReduce program'),
  2477. 'value': '',
  2478. 'help_text': _('Select a saved MapReduce program you want to schedule.'),
  2479. 'type': 'mapreduce-doc'
  2480. },
  2481. 'parameters': {
  2482. 'name': 'parameters',
  2483. 'label': _('Parameters'),
  2484. 'value': [],
  2485. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  2486. 'type': ''
  2487. },
  2488. # Common
  2489. 'files': {
  2490. 'name': 'files',
  2491. 'label': _('Files'),
  2492. 'value': [],
  2493. 'help_text': _('Files put in the running directory.'),
  2494. 'type': ''
  2495. },
  2496. 'archives': {
  2497. 'name': 'archives',
  2498. 'label': _('Archives'),
  2499. 'value': [],
  2500. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  2501. 'type': ''
  2502. },
  2503. 'job_properties': {
  2504. 'name': 'job_properties',
  2505. 'label': _('Hadoop job properties'),
  2506. 'value': [],
  2507. 'help_text': _('value, e.g. production'),
  2508. 'type': ''
  2509. },
  2510. 'prepares': {
  2511. 'name': 'prepares',
  2512. 'label': _('Prepares'),
  2513. 'value': [],
  2514. 'help_text': _('Path to manipulate before starting the application.'),
  2515. 'type': ''
  2516. },
  2517. 'job_xml': {
  2518. 'name': 'job_xml',
  2519. 'label': _('Job XML'),
  2520. 'value': '',
  2521. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  2522. 'type': ''
  2523. },
  2524. 'retry_max': {
  2525. 'name': 'retry_max',
  2526. 'label': _('Max retry'),
  2527. 'value': [],
  2528. 'help_text': _('Number of times, default is 3'),
  2529. 'type': ''
  2530. },
  2531. 'retry_interval': {
  2532. 'name': 'retry_interval',
  2533. 'label': _('Retry interval'),
  2534. 'value': [],
  2535. 'help_text': _('Wait time in minutes, default is 10'),
  2536. 'type': ''
  2537. }
  2538. }
  2539. @classmethod
  2540. def get_mandatory_fields(cls):
  2541. return [cls.FIELDS['uuid']]
  2542. class DecisionNode(Action):
  2543. TYPE = 'decision'
  2544. FIELDS = {}
  2545. @classmethod
  2546. def get_mandatory_fields(cls):
  2547. return []
  2548. NODES = {
  2549. 'start-widget': StartNode,
  2550. 'end-widget': EndNode,
  2551. 'pig-widget': PigAction,
  2552. 'java-widget': JavaAction,
  2553. 'hive-widget': HiveAction,
  2554. 'hive2-widget': HiveServer2Action,
  2555. 'impala-widget': ImpalaAction,
  2556. 'sqoop-widget': SqoopAction,
  2557. 'mapreduce-widget': MapReduceAction,
  2558. 'subworkflow-widget': SubWorkflowAction,
  2559. 'shell-widget': ShellAction,
  2560. 'ssh-widget': SshAction,
  2561. 'fs-widget': FsAction,
  2562. 'email-widget': EmailAction,
  2563. 'streaming-widget': StreamingAction,
  2564. 'distcp-widget': DistCpAction,
  2565. 'kill-widget': KillAction,
  2566. 'join-widget': JoinAction,
  2567. 'fork-widget': ForkNode,
  2568. 'decision-widget': DecisionNode,
  2569. 'spark-widget': SparkAction,
  2570. 'generic-widget': GenericAction,
  2571. 'hive-document-widget': HiveDocumentAction,
  2572. 'impala-document-widget': ImpalaDocumentAction,
  2573. 'java-document-widget': JavaDocumentAction,
  2574. 'spark-document-widget': SparkDocumentAction,
  2575. 'pig-document-widget': PigDocumentAction,
  2576. 'sqoop-document-widget': SqoopDocumentAction,
  2577. 'distcp-document-widget': DistCpDocumentAction,
  2578. 'shell-document-widget': ShellDocumentAction,
  2579. 'mapreduce-document-widget': MapReduceDocumentAction
  2580. }
  2581. WORKFLOW_NODE_PROPERTIES = {}
  2582. for node in NODES.itervalues():
  2583. WORKFLOW_NODE_PROPERTIES.update(node.FIELDS)
  2584. def find_parameters(instance, fields=None):
  2585. """Find parameters in the given fields"""
  2586. if fields is None:
  2587. fields = NODES['%s-widget' % instance.data['type']].FIELDS.keys()
  2588. params = []
  2589. for field in fields:
  2590. data = instance.data['properties'][field]
  2591. if field == 'sla' and not instance.sla_enabled:
  2592. continue
  2593. if isinstance(data, list):
  2594. params.extend(find_json_parameters(data))
  2595. elif isinstance(data, basestring):
  2596. for match in Template.pattern.finditer(data):
  2597. name = match.group('braced')
  2598. if name is not None:
  2599. params.append(name)
  2600. return params
  2601. def find_json_parameters(fields):
  2602. # Input is list of json dict
  2603. params = []
  2604. for field in fields:
  2605. for data in field.values():
  2606. if isinstance(data, basestring):
  2607. for match in Template.pattern.finditer(data):
  2608. name = match.group('braced')
  2609. if name is not None:
  2610. params.append(name)
  2611. return params
  2612. def find_dollar_variables(text):
  2613. return re.findall('[^\n\\\\]\$([^\{ \'\"\-;\(\)]+)', text, re.MULTILINE)
  2614. def find_dollar_braced_variables(text):
  2615. vars = set()
  2616. for var in re.findall('\$\{([A-Za-z0-9:_-]+)\}', text, re.MULTILINE):
  2617. if ':' in var:
  2618. var = var.split(':', 1)[1]
  2619. vars.add(var)
  2620. return list(vars)
  2621. def import_workflow_from_hue_3_7(old_wf):
  2622. """
  2623. Example of data to transform
  2624. [<Start: start>, <Pig: Pig>, [<Kill: kill>], [<End: end>]]
  2625. [<Start: start>, <Java: TeraGenWorkflow>, <Java: TeraSort>, [<Kill: kill>], [<End: end>]]
  2626. [<Start: start>, [<Fork: fork-34>, [[<Mapreduce: Sleep-1>, <Mapreduce: Sleep-10>], [<Mapreduce: Sleep-5>, [<Fork: fork-38>, [[<Mapreduce: Sleep-3>], [<Mapreduce: Sleep-4>]], <Join: join-39>]]], <Join: join-35>], [<Kill: kill>], [<End: end>]]
  2627. """
  2628. uuids = {}
  2629. old_nodes = old_wf.get_hierarchy()
  2630. wf = Workflow()
  2631. wf_rows = []
  2632. wf_nodes = []
  2633. data = wf.get_data()
  2634. # UUIDs node mapping
  2635. for node in old_wf.node_list:
  2636. if node.name == 'kill':
  2637. node_uuid = '17c9c895-5a16-7443-bb81-f34b30b21548'
  2638. elif node.name == 'start':
  2639. node_uuid = '3f107997-04cc-8733-60a9-a4bb62cebffc'
  2640. elif node.name == 'end':
  2641. node_uuid = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
  2642. else:
  2643. node_uuid = str(uuid.uuid4())
  2644. uuids[node.id] = node_uuid
  2645. # Workflow
  2646. data['workflow']['uuid'] = str(uuid.uuid4())
  2647. data['workflow']['name'] = old_wf.name
  2648. data['workflow']['properties']['properties'] = json.loads(old_wf.job_properties)
  2649. data['workflow']['properties']['job_xml'] = old_wf.job_xml
  2650. data['workflow']['properties']['description'] = old_wf.description
  2651. data['workflow']['properties']['schema_version'] = old_wf.schema_version
  2652. data['workflow']['properties']['deployment_dir'] = old_wf.deployment_dir
  2653. data['workflow']['properties']['parameters'] = json.loads(old_wf.parameters)
  2654. data['workflow']['properties']['description'] = old_wf.description
  2655. data['workflow']['properties']['sla'] = old_wf.sla
  2656. data['workflow']['properties']['sla_enabled'] = old_wf.sla_enabled
  2657. data['workflow']['properties']['imported'] = True
  2658. data['workflow']['properties']['wf1_id'] = old_wf.id
  2659. # Layout
  2660. rows = data['layout'][0]['rows']
  2661. def _create_layout(nodes, size=12):
  2662. wf_rows = []
  2663. for node in nodes:
  2664. if type(node) == list and len(node) == 1:
  2665. node = node[0]
  2666. if type(node) != list:
  2667. wf_rows.append({"widgets":[{"size":size, "name": node.name.title(), "id": uuids[node.id], "widgetType": "%s-widget" % node.node_type, "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  2668. else:
  2669. if node[0].node_type == 'fork':
  2670. wf_rows.append({"widgets":[{"size":size, "name": 'Fork', "id": uuids[node[0].id], "widgetType": "%s-widget" % node[0].node_type, "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  2671. wf_rows.append({
  2672. "id": str(uuid.uuid4()),
  2673. "widgets":[
  2674. ],
  2675. "columns":[
  2676. {
  2677. "id": str(uuid.uuid4()),
  2678. "size": (size / len(node[1])),
  2679. "rows":
  2680. [{
  2681. "id": str(uuid.uuid4()),
  2682. "widgets": c['widgets'],
  2683. "columns":[]
  2684. }
  2685. for c in col] if type(col) == list else [{
  2686. "id": str(uuid.uuid4()),
  2687. "widgets": col['widgets'],
  2688. "columns":[]
  2689. }
  2690. ]
  2691. ,
  2692. "klass":"card card-home card-column span%s" % (size / len(node[1]))
  2693. }
  2694. for col in _create_layout(node[1], size)
  2695. ]
  2696. })
  2697. wf_rows.append({"widgets":[{"size":size, "name": 'Join', "id": uuids[node[2].id], "widgetType": "%s-widget" % node[2].node_type, "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span%s" % size, "columns":[]}]})
  2698. else:
  2699. wf_rows.append(_create_layout(node, size))
  2700. return wf_rows
  2701. wf_rows = _create_layout(old_nodes)
  2702. if wf_rows:
  2703. data['layout'][0]['rows'] = [data['layout'][0]['rows'][0]] + wf_rows + [data['layout'][0]['rows'][-1]]
  2704. # Content
  2705. def _dig_nodes(nodes):
  2706. for node in nodes:
  2707. if type(node) != list:
  2708. properties = {}
  2709. if '%s-widget' % node.node_type in NODES:
  2710. properties = dict(NODES['%s-widget' % node.node_type].get_fields())
  2711. if node.node_type == 'pig':
  2712. properties['script_path'] = node.script_path
  2713. properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
  2714. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2715. properties['archives'] = json.loads(node.archives)
  2716. properties['job_properties'] = json.loads(node.job_properties)
  2717. properties['prepares'] = json.loads(node.prepares)
  2718. properties['job_xml'] = node.job_xml
  2719. properties['description'] = node.description
  2720. properties['sla'] = node.sla
  2721. properties['sla_enabled'] = node.sla_enabled
  2722. elif node.node_type == 'hive':
  2723. properties['script_path'] = node.script_path
  2724. properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
  2725. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2726. properties['archives'] = json.loads(node.archives)
  2727. properties['job_properties'] = json.loads(node.job_properties)
  2728. properties['prepares'] = json.loads(node.prepares)
  2729. properties['hive_xml'] = node.job_xml
  2730. properties['description'] = node.description
  2731. properties['sla'] = node.sla
  2732. properties['sla_enabled'] = node.sla_enabled
  2733. elif node.node_type == 'java':
  2734. properties['jar_path'] = node.jar_path
  2735. properties['main_class'] = node.main_class
  2736. properties['arguments'] = [{'value': arg} for arg in node.args.split(' ')]
  2737. properties['java_opts'] = node.java_opts
  2738. properties['capture_output'] = node.capture_output
  2739. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2740. properties['archives'] = json.loads(node.archives)
  2741. properties['job_properties'] = json.loads(node.job_properties)
  2742. properties['prepares'] = json.loads(node.prepares)
  2743. properties['job_xml'] = node.job_xml
  2744. properties['description'] = node.description
  2745. properties['sla'] = node.sla
  2746. properties['sla_enabled'] = node.sla_enabled
  2747. elif node.node_type == 'sqoop':
  2748. properties['command'] = node.script_path
  2749. properties['parameters'] = json.loads(node.params)
  2750. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2751. properties['archives'] = json.loads(node.archives)
  2752. properties['job_properties'] = json.loads(node.job_properties)
  2753. properties['prepares'] = json.loads(node.prepares)
  2754. properties['job_xml'] = node.job_xml
  2755. properties['description'] = node.description
  2756. properties['sla'] = node.sla
  2757. properties['sla_enabled'] = node.sla_enabled
  2758. elif node.node_type == 'mapreduce':
  2759. properties['jar_path'] = node.jar_path
  2760. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2761. properties['archives'] = json.loads(node.archives)
  2762. properties['job_properties'] = json.loads(node.job_properties)
  2763. properties['prepares'] = json.loads(node.prepares)
  2764. properties['job_xml'] = node.job_xml
  2765. properties['description'] = node.description
  2766. properties['sla'] = node.sla
  2767. properties['sla_enabled'] = node.sla_enabled
  2768. elif node.node_type == 'shell':
  2769. properties['shell_command'] = node.command
  2770. properties['arguments'] = json.loads(node.params)
  2771. properties['capture_output'] = node.capture_output
  2772. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2773. properties['archives'] = json.loads(node.archives)
  2774. properties['job_properties'] = json.loads(node.job_properties)
  2775. properties['prepares'] = json.loads(node.prepares)
  2776. properties['job_xml'] = node.job_xml
  2777. properties['description'] = node.description
  2778. properties['sla'] = node.sla
  2779. properties['sla_enabled'] = node.sla_enabled
  2780. elif node.node_type == 'ssh':
  2781. properties['user'] = '%s@%s' % (node.user, node.host)
  2782. properties['ssh_command'] = node.command
  2783. properties['params'] = json.loads(node.params)
  2784. properties['capture_output'] = node.capture_output
  2785. properties['description'] = node.description
  2786. properties['sla'] = node.sla
  2787. properties['sla_enabled'] = node.sla_enabled
  2788. elif node.node_type == 'fs':
  2789. properties['deletes'] = [{'value': f['name']} for f in json.loads(node.deletes)]
  2790. properties['mkdirs'] = [{'value': f['name']} for f in json.loads(node.mkdirs)]
  2791. properties['moves'] = json.loads(node.moves)
  2792. chmods = json.loads(node.chmods)
  2793. for c in chmods:
  2794. c['value'] = c['path']
  2795. c['dir_files'] = False
  2796. properties['chmods'] = chmods
  2797. properties['touchzs'] = [{'value': f['name']} for f in json.loads(node.touchzs)]
  2798. properties['description'] = node.description
  2799. properties['sla'] = node.sla
  2800. properties['sla_enabled'] = node.sla_enabled
  2801. elif node.node_type == 'email':
  2802. properties['to'] = node.to
  2803. properties['cc'] = node.cc
  2804. properties['subject'] = node.subject
  2805. properties['body'] = node.body
  2806. properties['description'] = node.description
  2807. properties['sla'] = node.sla
  2808. properties['sla_enabled'] = node.sla_enabled
  2809. elif node.node_type == 'streaming':
  2810. properties['mapper'] = node.mapper
  2811. properties['reducer'] = node.reducer
  2812. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  2813. properties['archives'] = json.loads(node.archives)
  2814. properties['job_properties'] = json.loads(node.job_properties)
  2815. properties['prepares'] = json.loads(node.prepares)
  2816. properties['job_xml'] = node.job_xml
  2817. properties['description'] = node.description
  2818. properties['sla'] = node.sla
  2819. properties['sla_enabled'] = node.sla_enabled
  2820. elif node.node_type == 'distcp':
  2821. properties['distcp_parameters'] = json.loads(node.params)
  2822. properties['java_opts'] = node.job_xml
  2823. properties['job_properties'] = json.loads(node.job_properties)
  2824. properties['prepares'] = json.loads(node.prepares)
  2825. properties['description'] = node.description
  2826. properties['sla'] = node.sla
  2827. properties['sla_enabled'] = node.sla_enabled
  2828. wf_nodes.append({
  2829. "id": uuids[node.id],
  2830. "name": '%s-%s' % (node.node_type.split('-')[0], uuids[node.id][:4]),
  2831. "type": "%s-widget" % node.node_type,
  2832. "properties": properties,
  2833. "children":[{('to' if link.name in ('ok', 'start') else link.name): uuids[link.child.get_full_node().id]} for link in node.get_children_links()]
  2834. })
  2835. else:
  2836. _dig_nodes(node)
  2837. _dig_nodes(old_nodes)
  2838. data['workflow']['nodes'] = wf_nodes
  2839. return Workflow(data=json.dumps(data))
  2840. class Coordinator(Job):
  2841. XML_FILE_NAME = 'coordinator.xml'
  2842. PROPERTY_APP_PATH = 'oozie.coord.application.path'
  2843. HUE_ID = 'hue-id-c'
  2844. def __init__(self, data=None, json_data=None, document=None):
  2845. self.document = document
  2846. if document is not None:
  2847. self._data = json.loads(document.data)
  2848. elif json_data is not None:
  2849. self._data = json.loads(json_data)
  2850. elif data is not None:
  2851. self._data = data
  2852. else:
  2853. self._data = {
  2854. 'id': None,
  2855. 'uuid': None,
  2856. 'name': 'My Coordinator',
  2857. 'variables': [], # Aka workflow parameters
  2858. 'properties': {
  2859. 'description': '',
  2860. 'deployment_dir': '',
  2861. 'schema_version': 'uri:oozie:coordinator:0.2',
  2862. 'frequency_number': 1,
  2863. 'frequency_unit': 'days',
  2864. 'cron_frequency': '0 0 * * *',
  2865. 'cron_advanced': False,
  2866. 'timezone': '',
  2867. 'start': '${start_date}',
  2868. 'end': '${end_date}',
  2869. 'workflow': None,
  2870. 'document': None,
  2871. 'timeout': None,
  2872. 'concurrency': None,
  2873. 'execution': None,
  2874. 'throttle': None,
  2875. 'job_xml': '',
  2876. 'credentials': [],
  2877. 'parameters': [
  2878. {'name': 'oozie.use.system.libpath', 'value': True},
  2879. {'name': 'start_date', 'value': datetime.today().strftime('%Y-%m-%dT%H:%M')},
  2880. {'name': 'end_date', 'value': (datetime.today() + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M')}
  2881. ],
  2882. 'sla': WorkflowConfiguration.SLA_DEFAULT
  2883. }
  2884. }
  2885. @property
  2886. def id(self):
  2887. return self.document.id
  2888. @property
  2889. def uuid(self):
  2890. return self.document.uuid
  2891. def get_data_for_json(self):
  2892. _data = self.data.copy()
  2893. start_date = filter(lambda a: a['name'] == 'start_date', self._data['properties']['parameters'])
  2894. if start_date and type(start_date[0]['value']) == datetime:
  2895. start_date[0]['value'] = start_date[0]['value'].strftime('%Y-%m-%dT%H:%M:%S')
  2896. end_date = filter(lambda a: a['name'] == 'end_date', self._data['properties']['parameters'])
  2897. if end_date and type(end_date[0]['value']) == datetime:
  2898. end_date[0]['value'] = end_date[0]['value'].strftime('%Y-%m-%dT%H:%M:%S')
  2899. return _data
  2900. def to_json(self):
  2901. return json.dumps(self.get_data_for_json())
  2902. def to_json_for_html(self):
  2903. return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
  2904. @property
  2905. def data(self):
  2906. if type(self._data['properties']['start']) != datetime and not '$' in self._data['properties']['start']:
  2907. self._data['properties']['start'] = parse(self._data['properties']['start'])
  2908. if type(self._data['properties']['end']) != datetime and not '$' in self._data['properties']['end']:
  2909. self._data['properties']['end'] = parse(self._data['properties']['end'])
  2910. if self.document is not None:
  2911. self._data['id'] = self.document.id
  2912. if 'document' not in self._data['properties']:
  2913. self._data['properties']['document'] = None
  2914. return self._data
  2915. @property
  2916. def name(self):
  2917. from notebook.connectors.oozie_batch import OozieApi # Import dependency
  2918. if self.data['properties']['document']:
  2919. return _("%s for %s") % (OozieApi.SCHEDULE_JOB_PREFIX, self.data['name'] or self.data['type'])
  2920. else:
  2921. return self.data['name']
  2922. def set_workspace(self, user):
  2923. self.data['properties']['deployment_dir'] = Job.get_workspace(user)
  2924. @property
  2925. def deployment_dir(self):
  2926. return self.data['properties']['deployment_dir']
  2927. def find_parameters(self):
  2928. params = set()
  2929. for param in find_dollar_braced_variables(self.name):
  2930. params.add(param)
  2931. for param in find_json_parameters([self.data['properties']]):
  2932. params.add(param)
  2933. for param in find_json_parameters(self.data['variables']):
  2934. if param not in ('MINUTE', 'HOUR', 'DAY', 'MONTH', 'YEAR') and not param.startswith('coord:'):
  2935. params.add(param)
  2936. if self.sla_enabled:
  2937. for param in find_json_parameters(self.sla):
  2938. params.add(param)
  2939. # Get missed params from workflow
  2940. for prop in self.workflow.find_parameters():
  2941. if not prop in params:
  2942. params.add(prop)
  2943. # Remove the ones filled up by coordinator
  2944. removable_names = [ds['workflow_variable'] for ds in self.data['variables']]
  2945. return dict([(param, '') for param in list(params) if param not in removable_names])
  2946. @property
  2947. def sla_enabled(self):
  2948. return self.data['properties']['sla'][0].get('value')
  2949. @property
  2950. def sla(self):
  2951. return self.data['properties']['sla']
  2952. @property
  2953. def parameters(self):
  2954. return self.data['properties']['parameters']
  2955. @property
  2956. def datasets(self):
  2957. return self.inputDatasets + self.outputDatasets
  2958. @property
  2959. def inputDatasets(self):
  2960. return [Dataset(dataset, self) for dataset in self.data['variables'] if dataset['dataset_type'] == 'input_path']
  2961. @property
  2962. def outputDatasets(self):
  2963. return [Dataset(dataset, self) for dataset in self.data['variables'] if dataset['dataset_type'] == 'output_path']
  2964. @property
  2965. def start_server_tz(self):
  2966. return self.data['properties']['start']
  2967. @property
  2968. def end_server_tz(self):
  2969. return self.data['properties']['end']
  2970. @property
  2971. def frequency(self):
  2972. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.data['properties']['frequency_unit'], 'number': self.data['properties']['frequency_number']}
  2973. @property
  2974. def cron_frequency(self):
  2975. data_dict = self.data['properties']
  2976. if 'cron_frequency' in data_dict:
  2977. return data_dict['cron_frequency']
  2978. else:
  2979. # Backward compatibility
  2980. freq = '0 0 * * *'
  2981. if data_dict['frequency_number'] == 1:
  2982. if data_dict['frequency_unit'] == 'minutes':
  2983. freq = '* * * * *'
  2984. elif data_dict['frequency_unit'] == 'hours':
  2985. freq = '0 * * * *'
  2986. elif data_dict['frequency_unit'] == 'days':
  2987. freq = '0 0 * * *'
  2988. elif data_dict['frequency_unit'] == 'months':
  2989. freq = '0 0 0 * *'
  2990. return {'frequency': freq, 'isAdvancedCron': False}
  2991. def to_xml(self, mapping=None):
  2992. if mapping is None:
  2993. mapping = {}
  2994. tmpl = "editor2/gen/coordinator.xml.mako"
  2995. return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
  2996. def clear_workflow_params(self):
  2997. # Repopulated in the config properties
  2998. self.data['variables'] = [dataset for dataset in self.data['variables'] if dataset['dataset_type'] != 'parameter']
  2999. @property
  3000. def properties(self):
  3001. props = [{'name': dataset['workflow_variable'], 'value': dataset['dataset_variable']} for dataset in self.data['variables'] if dataset['dataset_type'] == 'parameter']
  3002. props += self.data['properties']['parameters']
  3003. return props
  3004. @property
  3005. def workflow(self):
  3006. if self.document is None:
  3007. raise PopupException(_('Cannot return workflow since document attribute is None.'))
  3008. # Integrated scheduler
  3009. if self.data['properties']['document']:
  3010. document = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['document'])
  3011. wf_doc = WorkflowBuilder().create_workflow(document=document, user=self.document.owner, managed=True)
  3012. wf = Workflow(data=wf_doc.data)
  3013. wf_doc.delete()
  3014. return wf
  3015. else:
  3016. wf_doc = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['workflow'])
  3017. return Workflow(document=wf_doc)
  3018. def get_absolute_url(self):
  3019. return reverse('oozie:edit_coordinator') + '?coordinator=%s' % self.id
  3020. @classmethod
  3021. def get_application_path_key(cls):
  3022. return 'oozie.coord.application.path'
  3023. class Dataset():
  3024. def __init__(self, data, coordinator):
  3025. self._data = data
  3026. self.coordinator = coordinator
  3027. @property
  3028. def data(self):
  3029. self._data['name'] = self._data['workflow_variable']
  3030. return self._data
  3031. @property
  3032. def frequency(self):
  3033. if self.data['same_frequency']:
  3034. if self.coordinator.cron_frequency == '* * * * *':
  3035. frequency_unit = 'minutes'
  3036. elif self.coordinator.cron_frequency == '0 * * * *':
  3037. frequency_unit = 'hours'
  3038. elif self.coordinator.cron_frequency == '0 0 * * *':
  3039. frequency_unit = 'days'
  3040. elif self.coordinator.cron_frequency == '0 0 0 * *':
  3041. frequency_unit = 'months'
  3042. else:
  3043. raise PopupException(_('The frequency of the workflow parameter "%s" cannot be guessed from the frequency of the coordinator.'
  3044. ' It so needs to be specified manually.') % self.data['name'])
  3045. frequency_number = 1
  3046. else:
  3047. frequency_unit = self.data['frequency_unit']
  3048. frequency_number = self.data['frequency_number']
  3049. return '${coord:%(unit)s(%(number)s)}' % {'unit': frequency_unit, 'number': frequency_number}
  3050. @property
  3051. def start_server_tz(self):
  3052. if self.data['same_start']:
  3053. return self.coordinator.start_server_tz
  3054. else:
  3055. return convert_to_server_timezone(self.data['start'], self.data['timezone'])
  3056. @property
  3057. def timezone(self):
  3058. if self.data['same_timezone']:
  3059. return self.coordinator.data['properties']['timezone']
  3060. else:
  3061. return self.data['timezone']
  3062. @property
  3063. def start_instance(self):
  3064. if not self.is_advanced_start_instance:
  3065. return int(self.data['advanced_start_instance'])
  3066. else:
  3067. return 0
  3068. @property
  3069. def is_advanced_start_instance(self):
  3070. return not self.is_int(self.data['advanced_start_instance'])
  3071. def is_int(self, text):
  3072. try:
  3073. int(text)
  3074. return True
  3075. except ValueError:
  3076. return False
  3077. @property
  3078. def end_instance(self):
  3079. if not self.is_advanced_end_instance:
  3080. return int(self.data['advanced_end_instance'])
  3081. else:
  3082. return 0
  3083. @property
  3084. def is_advanced_end_instance(self):
  3085. return not self.is_int(self.data['advanced_end_instance'])
  3086. class Bundle(Job):
  3087. XML_FILE_NAME = 'bundle.xml'
  3088. PROPERTY_APP_PATH = 'oozie.bundle.application.path'
  3089. HUE_ID = 'hue-id-b'
  3090. def __init__(self, data=None, json_data=None, document=None):
  3091. self.document = document
  3092. if document is not None:
  3093. self._data = json.loads(document.data)
  3094. elif json_data is not None:
  3095. self._data = json.loads(json_data)
  3096. elif data is not None:
  3097. self._data = data
  3098. else:
  3099. self._data = {
  3100. 'id': None,
  3101. 'uuid': None,
  3102. 'name': 'My Bundle',
  3103. 'coordinators': [],
  3104. 'properties': {
  3105. 'description': '',
  3106. 'deployment_dir': '',
  3107. 'schema_version': 'uri:oozie:bundle:0.2',
  3108. 'kickoff': datetime.today(),
  3109. 'parameters': [{'name': 'oozie.use.system.libpath', 'value': 'true'}]
  3110. }
  3111. }
  3112. @property
  3113. def id(self):
  3114. return self.document.id
  3115. @property
  3116. def uuid(self):
  3117. return self.document.uuid
  3118. def get_data_for_json(self):
  3119. _data = self.data.copy()
  3120. _data['properties']['kickoff'] = _data['properties']['kickoff'].strftime('%Y-%m-%dT%H:%M:%S')
  3121. return _data
  3122. def to_json(self):
  3123. return json.dumps(self.get_data_for_json())
  3124. def to_json_for_html(self):
  3125. return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
  3126. @property
  3127. def data(self):
  3128. if type(self._data['properties']['kickoff']) == unicode:
  3129. self._data['properties']['kickoff'] = parse(self._data['properties']['kickoff'])
  3130. if self.document is not None:
  3131. self._data['id'] = self.document.id
  3132. return self._data
  3133. def to_xml(self, mapping=None):
  3134. if mapping is None:
  3135. mapping = {}
  3136. mapping.update(dict(list(self.get_coordinator_docs().values('uuid', 'name'))))
  3137. tmpl = "editor2/gen/bundle.xml.mako"
  3138. return force_unicode(
  3139. re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
  3140. 'bundle': self,
  3141. 'mapping': mapping
  3142. })))
  3143. def get_coordinator_docs(self):
  3144. coordinator_ids = [coordinator['coordinator'] for coordinator in self.data['coordinators']]
  3145. return Document2.objects.filter(type='oozie-coordinator2', uuid__in=coordinator_ids)
  3146. def get_coordinator_objects(self):
  3147. return [Coordinator(document=doc) for doc in self.get_coordinator_docs()]
  3148. @property
  3149. def name(self):
  3150. return self.data['name']
  3151. @property
  3152. def parameters(self):
  3153. return self.data['properties']['parameters']
  3154. @property
  3155. def kick_off_time_utc(self):
  3156. return utc_datetime_format(self.data['properties']['kickoff'])
  3157. def set_workspace(self, user):
  3158. self.data['properties']['deployment_dir'] = Job.get_workspace(user)
  3159. @property
  3160. def deployment_dir(self):
  3161. return self.data['properties']['deployment_dir']
  3162. def find_parameters(self):
  3163. params = set()
  3164. for param in find_dollar_braced_variables(self.name):
  3165. params.add(param)
  3166. for coord in self.get_coordinator_objects():
  3167. params.update(coord.find_parameters())
  3168. for param in find_json_parameters([self.data['properties']]):
  3169. params.add(param)
  3170. # Remove the ones filled up by bundle
  3171. removable_names = [p['name'] for coord in self.data['coordinators'] for p in coord['properties']]
  3172. return dict([(param, '') for param in list(params) if param not in removable_names])
  3173. def get_absolute_url(self):
  3174. return reverse('oozie:edit_bundle') + '?bundle=%s' % self.id
  3175. @classmethod
  3176. def get_application_path_key(cls):
  3177. return 'oozie.bundle.application.path'
  3178. class History(object):
  3179. @classmethod
  3180. def get_workflow_from_config(self, conf_dict):
  3181. try:
  3182. doc = Document2.objects.get(type='oozie-workflow2', id=conf_dict.get(Workflow.HUE_ID))
  3183. return Workflow(document=doc)
  3184. except Document2.DoesNotExist:
  3185. pass
  3186. @classmethod
  3187. def get_coordinator_from_config(self, conf_dict):
  3188. try:
  3189. doc = Document2.objects.get(type='oozie-coordinator2', id=conf_dict.get(Coordinator.HUE_ID))
  3190. return Coordinator(document=doc)
  3191. except Document2.DoesNotExist:
  3192. pass
  3193. @classmethod
  3194. def get_bundle_from_config(self, conf_dict):
  3195. try:
  3196. doc = Document2.objects.get(type='oozie-bundle2', id=conf_dict.get(Bundle.HUE_ID))
  3197. return Bundle(document=doc)
  3198. except Document2.DoesNotExist:
  3199. pass
  3200. def _import_workspace(fs, user, job):
  3201. source_workspace_dir = job.deployment_dir
  3202. job.set_workspace(user)
  3203. job.check_workspace(fs, user)
  3204. job.import_workspace(fs, source_workspace_dir, user)
  3205. def _save_workflow(workflow, layout, user, fs=None):
  3206. if workflow.get('id'):
  3207. workflow_doc = Document2.objects.get(id=workflow['id'])
  3208. else:
  3209. workflow_doc = Document2.objects.create(name=workflow['name'], uuid=workflow['uuid'], type='oozie-workflow2', owner=user, description=workflow['properties']['description'])
  3210. Document.objects.link(workflow_doc, owner=workflow_doc.owner, name=workflow_doc.name, description=workflow_doc.description, extra='workflow2')
  3211. # Excludes all the sub-workflow and Hive dependencies. Contains list of history and coordinator dependencies.
  3212. workflow_doc.dependencies = workflow_doc.dependencies.exclude(Q(is_history=False) & Q(type__in=['oozie-workflow2', 'query-hive', 'query-java']))
  3213. dependencies = \
  3214. [node['properties']['workflow'] for node in workflow['nodes'] if node['type'] == 'subworkflow-widget'] + \
  3215. [node['properties']['uuid'] for node in workflow['nodes'] if 'document-widget' in node['type'] and node['properties'].get('uuid')]
  3216. if dependencies:
  3217. dependency_docs = Document2.objects.filter(uuid__in=dependencies)
  3218. workflow_doc.dependencies.add(*dependency_docs)
  3219. if workflow['properties'].get('imported'): # We convert from and old workflow format (3.8 <) to the latest
  3220. workflow['properties']['imported'] = False
  3221. workflow_instance = Workflow(workflow=workflow, user=user)
  3222. _import_workspace(fs, user, workflow_instance)
  3223. workflow['properties']['deployment_dir'] = workflow_instance.deployment_dir
  3224. workflow_doc.update_data({'workflow': workflow})
  3225. workflow_doc.update_data({'layout': layout})
  3226. workflow_doc1 = workflow_doc.doc.get()
  3227. workflow_doc.name = workflow_doc1.name = workflow['name']
  3228. workflow_doc.description = workflow_doc1.description = workflow['properties']['description']
  3229. workflow_doc.save()
  3230. workflow_doc1.save()
  3231. return workflow_doc
  3232. class WorkflowBuilder():
  3233. """
  3234. Building a workflow that has saved Documents for nodes (e.g Saved Hive query, saved Pig script...).
  3235. """
  3236. def create_workflow(self, user, document=None, name=None, managed=False):
  3237. nodes = []
  3238. documents = [document]
  3239. if name is None:
  3240. name = _('Schedule of ') + ','.join([document.name or document.type for document in documents])
  3241. for document in documents:
  3242. if document.type == 'query-java':
  3243. node = self.get_java_document_node(document)
  3244. elif document.type == 'query-hive':
  3245. node = self.get_hive_document_node(document, user)
  3246. elif document.type == 'query-impala':
  3247. node = self.get_impala_document_node(document, user)
  3248. elif document.type == 'query-spark2':
  3249. node = self.get_spark_document_node(document, user)
  3250. elif document.type == 'query-pig':
  3251. node = self.get_pig_document_node(document, user)
  3252. elif document.type == 'query-sqoop1':
  3253. node = self.get_sqoop_document_node(document, user)
  3254. elif document.type == 'query-distcp':
  3255. node = self.get_distcp_document_node(document, user)
  3256. elif document.type == 'query-shell':
  3257. node = self.get_shell_document_node(document, user)
  3258. elif document.type == 'query-mapreduce':
  3259. node = self.get_mapreduce_document_node(document, user)
  3260. else:
  3261. raise PopupException(_('Snippet type %s is not supported in batch execution.') % document.type)
  3262. nodes.append(node)
  3263. workflow_doc = self.get_workflow(nodes, name, document.uuid, user, managed=managed)
  3264. workflow_doc.dependencies.add(*documents)
  3265. return workflow_doc
  3266. def create_notebook_workflow(self, user, notebook=None, name=None, managed=False):
  3267. nodes = []
  3268. if name is None:
  3269. name = _('Schedule of ') + ','.join([snippet['name'] or snippet['type'] for snippet in notebook['snippets']])
  3270. for snippet in notebook['snippets']:
  3271. if snippet['type'] == 'java':
  3272. node = self.get_java_snippet_node(snippet)
  3273. elif snippet['type'] == 'hive':
  3274. node = self.get_hive_snippet_node(snippet, user)
  3275. elif snippet['type'] == 'impala':
  3276. node = self.get_impala_snippet_node(snippet, user)
  3277. elif snippet['type'] == 'shell':
  3278. node = self.get_shell_snippet_node(snippet)
  3279. else:
  3280. raise PopupException(_('Snippet type %s is not supported in batch execution.') % snippet)
  3281. nodes.append(node)
  3282. workflow_doc = self.get_workflow(nodes, name, notebook['uuid'], user, managed=managed) # TODO optionally save
  3283. return workflow_doc
  3284. def get_document_parameters(self, document):
  3285. notebook = Notebook(document=document)
  3286. parameters = find_dollar_braced_variables(notebook.get_str())
  3287. return [{u'value': u'%s=${%s}' % (p, p)} for p in parameters]
  3288. def _get_hive_node(self, node_id, user, is_document_node=False):
  3289. api = get_oozie(user)
  3290. credentials = [HiveDocumentAction.DEFAULT_CREDENTIALS] if api.security_enabled else []
  3291. return {
  3292. u'id': node_id,
  3293. u'name': u'hive-%s' % node_id[:4],
  3294. u"type": u"hive-document-widget", # if is_document_node else u"hive2-widget",
  3295. u'properties': {
  3296. u'files': [],
  3297. u'job_xml': u'',
  3298. u'retry_interval': [],
  3299. u'retry_max': [],
  3300. u'job_properties': [],
  3301. u'arguments': [],
  3302. u'parameters': [],
  3303. u'sla': [
  3304. {u'key': u'enabled', u'value': False},
  3305. {u'key': u'nominal-time', u'value': u'${nominal_time}'},
  3306. {u'key': u'should-start', u'value': u''},
  3307. {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
  3308. {u'key': u'max-duration', u'value': u''},
  3309. {u'key': u'alert-events', u'value': u''},
  3310. {u'key': u'alert-contact', u'value': u''},
  3311. {u'key': u'notification-msg', u'value': u''},
  3312. {u'key': u'upstream-apps', u'value': u''},
  3313. ],
  3314. u'archives': [],
  3315. u'prepares': [],
  3316. u'credentials': credentials,
  3317. u'password': u'',
  3318. u'jdbc_url': u'',
  3319. },
  3320. u'children': [
  3321. {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
  3322. {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
  3323. ],
  3324. u'actionParameters': [],
  3325. }
  3326. def get_hive_snippet_node(self, snippet, user):
  3327. node = self._get_hive_node(snippet['id'], user)
  3328. node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in snippet['variables']]
  3329. node['properties']['statements'] = 'USE %s;\n\n%s' % (snippet['database'], snippet['statement_raw'])
  3330. return node
  3331. def get_hive_document_node(self, document, user):
  3332. node = self._get_hive_node(document.uuid, user, is_document_node=True)
  3333. notebook = Notebook(document=document)
  3334. node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in notebook.get_data()['snippets'][0]['variables']]
  3335. node['properties']['uuid'] = document.uuid
  3336. return node
  3337. def _get_impala_node(self, node_id, user, is_document_node=False):
  3338. credentials = []
  3339. return {
  3340. u'id': node_id,
  3341. u'name': u'impala-%s' % node_id[:4],
  3342. u"type": u"impala-document-widget",
  3343. u'properties': {
  3344. u'files': [],
  3345. u'job_xml': u'',
  3346. u'retry_interval': [],
  3347. u'retry_max': [],
  3348. u'job_properties': [],
  3349. u'arguments': [],
  3350. u'parameters': [],
  3351. u'sla': [
  3352. {u'key': u'enabled', u'value': False},
  3353. {u'key': u'nominal-time', u'value': u'${nominal_time}'},
  3354. {u'key': u'should-start', u'value': u''},
  3355. {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
  3356. {u'key': u'max-duration', u'value': u''},
  3357. {u'key': u'alert-events', u'value': u''},
  3358. {u'key': u'alert-contact', u'value': u''},
  3359. {u'key': u'notification-msg', u'value': u''},
  3360. {u'key': u'upstream-apps', u'value': u''},
  3361. ],
  3362. u'archives': [],
  3363. u'prepares': [],
  3364. u'credentials': credentials,
  3365. u'impalad_host': u'',
  3366. u'key_tab_path': u'',
  3367. u'user_principal': u''
  3368. },
  3369. u'children': [
  3370. {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
  3371. {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
  3372. ],
  3373. u'actionParameters': [],
  3374. }
  3375. def get_impala_snippet_node(self, snippet, user):
  3376. node = self._get_impala_node(snippet['id'], user)
  3377. node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in snippet['variables']]
  3378. node['properties']['statements'] = 'USE %s;\n\n%s' % (snippet['database'], snippet['statement_raw'])
  3379. return node
  3380. def get_impala_document_node(self, document, user):
  3381. node = self._get_impala_node(document.uuid, user, is_document_node=True)
  3382. notebook = Notebook(document=document)
  3383. node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in notebook.get_data()['snippets'][0]['variables']]
  3384. node['properties']['uuid'] = document.uuid
  3385. return node
  3386. def _get_spark_node(self, node_id, user, is_document_node=False):
  3387. credentials = []
  3388. return {
  3389. u'id': node_id,
  3390. u'name': u'spark2-%s' % node_id[:4],
  3391. u"type": u"spark2-document-widget", # if is_document_node else u"hive2-widget",
  3392. u'properties': {
  3393. u'files': [],
  3394. u'job_xml': u'',
  3395. u'retry_interval': [],
  3396. u'retry_max': [],
  3397. u'job_properties': [],
  3398. u'sla': [
  3399. {u'key': u'enabled', u'value': False},
  3400. {u'key': u'nominal-time', u'value': u'${nominal_time}'},
  3401. {u'key': u'should-start', u'value': u''},
  3402. {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
  3403. {u'key': u'max-duration', u'value': u''},
  3404. {u'key': u'alert-events', u'value': u''},
  3405. {u'key': u'alert-contact', u'value': u''},
  3406. {u'key': u'notification-msg', u'value': u''},
  3407. {u'key': u'upstream-apps', u'value': u''},
  3408. ],
  3409. u'archives': [],
  3410. u'prepares': [],
  3411. u'credentials': credentials,
  3412. u'spark_master': u'yarn',
  3413. u'mode': u'client',
  3414. u'app_name': u'BatchSpark2'
  3415. },
  3416. u'children': [
  3417. {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
  3418. {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
  3419. ],
  3420. u'actionParameters': [],
  3421. }
  3422. def get_spark_snippet_node(self, snippet):
  3423. credentials = []
  3424. node_id = snippet.get('id', str(uuid.uuid4()))
  3425. node = self._get_java_node(node_id, credentials)
  3426. node['properties']['class'] = snippet['properties']['class']
  3427. node['properties']['jars'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
  3428. node['properties']['spark_opts'] = [{'value': f['path']} for f in snippet['properties']['files']]
  3429. node['properties']['spark_arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
  3430. return node
  3431. def get_spark_document_node(self, document, user):
  3432. node = self._get_spark_node(document.uuid, user, is_document_node=True)
  3433. node['properties']['uuid'] = document.uuid
  3434. return node
  3435. def get_sqoop_document_node(self, document, user):
  3436. node = self._get_sqoop_node(document.uuid, is_document_node=True)
  3437. node['properties']['uuid'] = document.uuid
  3438. return node
  3439. def _get_sqoop_node(self, node_id, credentials=None, is_document_node=False):
  3440. if credentials is None:
  3441. credentials = []
  3442. return {
  3443. "id": node_id,
  3444. 'name': 'sqoop-%s' % node_id[:4],
  3445. "type": "sqoop-document-widget",
  3446. "properties":{
  3447. "command": "",
  3448. "arguments": [],
  3449. "retry_max": [],
  3450. "retry_interval": [],
  3451. "job_properties": [],
  3452. "prepares": [],
  3453. "credentials": credentials,
  3454. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3455. "archives": [],
  3456. "files": []
  3457. },
  3458. "children": [
  3459. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3460. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3461. ],
  3462. "actionParameters": [],
  3463. "actionParametersFetched": False
  3464. }
  3465. def get_distcp_document_node(self, document, user):
  3466. node = self._get_distcp_node(document.uuid, is_document_node=True)
  3467. node['properties']['uuid'] = document.uuid
  3468. return node
  3469. def _get_distcp_node(self, node_id, credentials=None, is_document_node=False):
  3470. if credentials is None:
  3471. credentials = []
  3472. return {
  3473. "id": node_id,
  3474. 'name': 'distcp-%s' % node_id[:4],
  3475. "type": "distcp-document-widget",
  3476. "properties":{
  3477. "source_path": "",
  3478. "destination_path": "",
  3479. "arguments": [],
  3480. "java_opts": [],
  3481. "retry_max": [],
  3482. "retry_interval": [],
  3483. "job_properties": [],
  3484. "prepares": [],
  3485. "credentials": credentials,
  3486. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3487. "archives": []
  3488. },
  3489. "children": [
  3490. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3491. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3492. ],
  3493. "actionParameters": [],
  3494. "actionParametersFetched": False
  3495. }
  3496. def get_shell_document_node(self, document, user):
  3497. node = self._get_shell_node(document.uuid)
  3498. node['properties']['uuid'] = document.uuid
  3499. return node
  3500. def get_shell_snippet_node(self, snippet):
  3501. node = self._get_shell_node(snippet['id'])
  3502. node['properties']['shell_command'] = snippet['properties'].get('shell_command')
  3503. node['properties']['arguments'] = snippet['properties'].get('arguments')
  3504. node['properties']['archives'] = snippet['properties'].get('archives')
  3505. node['properties']['files'] = snippet['properties'].get('files')
  3506. node['properties']['env_var'] = snippet['properties'].get('env_var')
  3507. return node
  3508. def _get_shell_node(self, node_id, credentials=None, is_snippet_node=False):
  3509. if credentials is None:
  3510. credentials = []
  3511. return {
  3512. "id": node_id,
  3513. 'name': 'shell-%s' % node_id[:4],
  3514. "type": "shell-document-widget",
  3515. "properties":{
  3516. "command_path": "",
  3517. "env_var": [],
  3518. "arguments": [],
  3519. "java_opts": [],
  3520. "retry_max": [],
  3521. "retry_interval": [],
  3522. "job_properties": [],
  3523. "capture_output": True,
  3524. "prepares": [],
  3525. "credentials": credentials,
  3526. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3527. "archives": []
  3528. },
  3529. "children": [
  3530. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3531. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3532. ],
  3533. "actionParameters": [],
  3534. "actionParametersFetched": False
  3535. }
  3536. def get_mapreduce_document_node(self, document, user):
  3537. node = self._get_mapreduce_node(document.uuid, is_document_node=True)
  3538. node['properties']['uuid'] = document.uuid
  3539. return node
  3540. def _get_mapreduce_node(self, node_id, credentials=None, is_document_node=False):
  3541. if credentials is None:
  3542. credentials = []
  3543. return {
  3544. "id": node_id,
  3545. 'name': 'mapreduce-%s' % node_id[:4],
  3546. "type": "mapreduce-document-widget",
  3547. "properties":{
  3548. "jar_path": "",
  3549. "arguments": [],
  3550. "java_opts": [],
  3551. "retry_max": [],
  3552. "retry_interval": [],
  3553. "job_properties": [],
  3554. "prepares": [],
  3555. "credentials": credentials,
  3556. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3557. "archives": []
  3558. },
  3559. "children": [
  3560. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3561. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3562. ],
  3563. "actionParameters": [],
  3564. "actionParametersFetched": False
  3565. }
  3566. def get_pig_document_node(self, document, user):
  3567. node = self._get_pig_node(document.uuid, is_document_node=True)
  3568. node['properties']['uuid'] = document.uuid
  3569. return node
  3570. def _get_pig_node(self, node_id, credentials=None, is_document_node=False):
  3571. if credentials is None:
  3572. credentials = []
  3573. return {
  3574. "id": node_id,
  3575. 'name': 'pig-%s' % node_id[:4],
  3576. "type": "pig-document-widget",
  3577. "properties":{
  3578. "job_xml": [],
  3579. "jar_path": "",
  3580. "java_opts": [],
  3581. "retry_max": [],
  3582. "retry_interval": [],
  3583. "job_properties": [],
  3584. "prepares": [],
  3585. "credentials": credentials,
  3586. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3587. "archives": []
  3588. },
  3589. "children": [
  3590. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3591. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3592. ],
  3593. "actionParameters": [],
  3594. "actionParametersFetched": False
  3595. }
  3596. def _get_java_node(self, node_id, credentials=None, is_document_node=False):
  3597. if credentials is None:
  3598. credentials = []
  3599. return {
  3600. "id": node_id,
  3601. 'name': 'java-%s' % node_id[:4],
  3602. "type": "java-document-widget" if is_document_node else "java-widget",
  3603. "properties":{
  3604. "job_xml": [],
  3605. "jar_path": "",
  3606. "java_opts": [],
  3607. "retry_max": [],
  3608. "retry_interval": [],
  3609. "job_properties": [],
  3610. "capture_output": False,
  3611. "prepares": [],
  3612. "credentials": credentials,
  3613. "sla": [{"value":False, "key":"enabled"}, {"value":"${nominal_time}", "key":"nominal-time"}, {"value":"", "key":"should-start"}, {"value":"${30 * MINUTES}", "key":"should-end"}, {"value":"", "key":"max-duration"}, {"value":"", "key":"alert-events"}, {"value":"", "key":"alert-contact"}, {"value":"", "key":"notification-msg"}, {"value":"", "key":"upstream-apps"}],
  3614. "archives": []
  3615. },
  3616. "children": [
  3617. {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
  3618. {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
  3619. ],
  3620. "actionParameters": [],
  3621. "actionParametersFetched": False
  3622. }
  3623. def get_java_snippet_node(self, snippet):
  3624. credentials = []
  3625. node_id = snippet.get('id', str(uuid.uuid4()))
  3626. node = self._get_java_node(node_id, credentials)
  3627. node['properties']['main_class'] = snippet['properties']['class']
  3628. node['properties']['app_jar'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
  3629. node['properties']['files'] = [{'value': f['path']} for f in snippet['properties']['files']]
  3630. node['properties']['arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
  3631. return node
  3632. def get_java_document_node(self, document):
  3633. credentials = []
  3634. node = self._get_java_node(document.uuid, credentials, is_document_node=True)
  3635. node['uuid'] = document.uuid
  3636. return node
  3637. def get_workflow(self, nodes, name, doc_uuid, user, managed=False):
  3638. parameters = []
  3639. data = {
  3640. u'workflow': {
  3641. u'name': name,
  3642. u'nodes': [{
  3643. u'name': u'Start',
  3644. u'properties': {},
  3645. u'actionParametersFetched': False,
  3646. u'id': u'3f107997-04cc-8733-60a9-a4bb62cebffc',
  3647. u'type': u'start-widget',
  3648. u'children': [{u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'}],
  3649. u'actionParameters': [],
  3650. }, {
  3651. u'name': u'End',
  3652. u'properties': {
  3653. u'body': u'',
  3654. u'cc': u'',
  3655. u'to': u'',
  3656. u'enableMail': False,
  3657. u'message': u'Workflow ${wf:id()} finished',
  3658. u'subject': u'',
  3659. u'attachment': u''
  3660. },
  3661. u'actionParametersFetched': False,
  3662. u'id': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a',
  3663. u'type': u'end-widget',
  3664. u'children': [],
  3665. u'actionParameters': [],
  3666. }, {
  3667. u'name': u'Kill',
  3668. u'properties': {
  3669. u'body': u'',
  3670. u'cc': u'',
  3671. u'to': u'',
  3672. u'enableMail': False,
  3673. u'message': u'Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]',
  3674. u'subject': u'',
  3675. u'attachment': u''
  3676. },
  3677. u'actionParametersFetched': False,
  3678. u'id': u'17c9c895-5a16-7443-bb81-f34b30b21548',
  3679. u'type': u'kill-widget',
  3680. u'children': [],
  3681. u'actionParameters': [],
  3682. }
  3683. ],
  3684. u'properties': {
  3685. u'job_xml': u'',
  3686. u'description': u'',
  3687. u'wf1_id': None,
  3688. u'sla_enabled': False,
  3689. u'deployment_dir': Job.get_workspace(user),
  3690. u'schema_version': u'uri:oozie:workflow:0.5',
  3691. u'sla': [
  3692. {u'key': u'enabled', u'value': False},
  3693. {u'key': u'nominal-time', u'value': u'${nominal_time}'},
  3694. {u'key': u'should-start', u'value': u''},
  3695. {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
  3696. {u'key': u'max-duration', u'value': u''},
  3697. {u'key': u'alert-events', u'value': u''},
  3698. {u'key': u'alert-contact', u'value': u''},
  3699. {u'key': u'notification-msg', u'value': u''},
  3700. {u'key': u'upstream-apps', u'value': u''},
  3701. ],
  3702. u'show_arrows': True,
  3703. u'parameters': parameters,
  3704. u'properties': [],
  3705. },
  3706. u'uuid': str(uuid.uuid4()),
  3707. }
  3708. }
  3709. _prev_node = data['workflow']['nodes'][0]
  3710. for node in nodes:
  3711. data['workflow']['nodes'].append(node)
  3712. _prev_node['children'][0]['to'] = node['id'] # We link nodes
  3713. _prev_node = node
  3714. workflow_doc = _save_workflow(data['workflow'], {}, user)
  3715. workflow_doc.is_managed = managed
  3716. workflow_doc.save()
  3717. return workflow_doc