| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270327132723273327432753276327732783279328032813282328332843285328632873288328932903291329232933294329532963297329832993300330133023303330433053306330733083309331033113312331333143315331633173318331933203321332233233324332533263327332833293330333133323333333433353336333733383339334033413342334333443345334633473348334933503351335233533354335533563357335833593360336133623363336433653366336733683369337033713372337333743375337633773378337933803381338233833384338533863387338833893390339133923393339433953396339733983399340034013402340334043405340634073408340934103411341234133414341534163417341834193420342134223423342434253426342734283429343034313432343334343435343634373438343934403441344234433444344534463447344834493450345134523453345434553456345734583459346034613462346334643465346634673468346934703471347234733474347534763477347834793480348134823483348434853486348734883489349034913492349334943495349634973498349935003501350235033504350535063507350835093510351135123513351435153516351735183519352035213522352335243525352635273528352935303531353235333534353535363537353835393540354135423543354435453546354735483549355035513552355335543555355635573558355935603561356235633564356535663567356835693570357135723573357435753576357735783579358035813582358335843585358635873588358935903591359235933594359535963597359835993600360136023603360436053606360736083609361036113612361336143615361636173618361936203621362236233624362536263627362836293630363136323633363436353636363736383639364036413642364336443645364636473648364936503651365236533654365536563657365836593660366136623663366436653666366736683669367036713672367336743675367636773678367936803681368236833684368536863687368836893690369136923693369436953696369736983699370037013702370337043705370637073708370937103711371237133714371537163717371837193720372137223723372437253726372737283729373037313732373337343735373637373738373937403741374237433744374537463747374837493750375137523753375437553756375737583759376037613762376337643765376637673768376937703771377237733774377537763777377837793780378137823783378437853786378737883789379037913792379337943795379637973798379938003801380238033804380538063807380838093810381138123813381438153816381738183819382038213822382338243825382638273828382938303831383238333834383538363837383838393840384138423843384438453846384738483849385038513852385338543855385638573858385938603861386238633864386538663867386838693870387138723873387438753876387738783879388038813882388338843885388638873888388938903891389238933894389538963897389838993900390139023903390439053906390739083909391039113912391339143915391639173918391939203921392239233924392539263927392839293930393139323933393439353936393739383939394039413942394339443945394639473948394939503951395239533954395539563957395839593960396139623963396439653966396739683969397039713972397339743975397639773978397939803981398239833984398539863987398839893990399139923993399439953996399739983999400040014002400340044005400640074008400940104011401240134014401540164017401840194020402140224023402440254026402740284029403040314032403340344035403640374038403940404041404240434044404540464047404840494050405140524053405440554056405740584059406040614062406340644065406640674068406940704071407240734074407540764077407840794080408140824083408440854086408740884089409040914092409340944095409640974098409941004101410241034104410541064107410841094110411141124113411441154116411741184119412041214122412341244125412641274128412941304131413241334134413541364137413841394140414141424143414441454146414741484149415041514152415341544155415641574158415941604161416241634164416541664167416841694170417141724173417441754176417741784179418041814182418341844185418641874188418941904191419241934194419541964197419841994200420142024203420442054206420742084209421042114212421342144215421642174218421942204221422242234224422542264227 |
- #!/usr/bin/env python
- # Licensed to Cloudera, Inc. under one
- # or more contributor license agreements. See the NOTICE file
- # distributed with this work for additional information
- # regarding copyright ownership. Cloudera, Inc. licenses this file
- # to you under the Apache License, Version 2.0 (the
- # "License"); you may not use this file except in compliance
- # with the License. You may obtain a copy of the License at
- #
- # http://www.apache.org/licenses/LICENSE-2.0
- #
- # Unless required by applicable law or agreed to in writing, software
- # distributed under the License is distributed on an "AS IS" BASIS,
- # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- # See the License for the specific language governing permissions and
- # limitations under the License.
- import json
- import logging
- import os
- import re
- import time
- import uuid
- from datetime import datetime, timedelta
- from dateutil.parser import parse
- from string import Template
- from xml.sax.saxutils import escape
- from django.core.urlresolvers import reverse
- from django.db.models import Q
- from django.utils.encoding import force_unicode
- from django.utils.translation import ugettext as _
- from desktop.conf import USE_DEFAULT_CONFIGURATION
- from desktop.lib import django_mako
- from desktop.lib.exceptions_renderable import PopupException
- from desktop.lib.i18n import smart_str
- from desktop.lib.json_utils import JSONEncoderForHTML
- from desktop.models import DefaultConfiguration, Document2, Document
- from hadoop.fs.hadoopfs import Hdfs
- from hadoop.fs.exceptions import WebHdfsException
- from liboozie.conf import SECURITY_ENABLED
- from liboozie.oozie_api import get_oozie
- from liboozie.submission2 import Submission
- from liboozie.submission2 import create_directories
- from notebook.models import Notebook
- from oozie.conf import REMOTE_SAMPLE_DIR
- from oozie.utils import utc_datetime_format, UTC_TIME_FORMAT, convert_to_server_timezone
- from oozie.importlib.workflows import generate_v2_graph_nodes, MalformedWfDefException, InvalidTagWithNamespaceException
- LOG = logging.getLogger(__name__)
- class Job(object):
- def find_all_parameters(self, with_lib_path=True):
- params = self.find_parameters()
- for param in self.parameters:
- params[param['name'].strip()] = param['value']
- if params.get('nominal_time') == '':
- params['nominal_time'] = datetime.today().strftime(UTC_TIME_FORMAT)
- return [{'name': name, 'value': value} for name, value in params.iteritems() if with_lib_path or name != 'oozie.use.system.libpath']
- @classmethod
- def get_workspace(cls, user):
- if not isinstance(user, basestring):
- user = user.username
- return (REMOTE_SAMPLE_DIR.get() + '/hue-oozie-$TIME').replace('$USER', user).replace('$TIME', str(time.time()))
- @property
- def validated_name(self):
- return ''.join(escape(self.name[:40], entities={
- "'": "'",
- "\"": """
- }))
- def __str__(self):
- return '%s' % force_unicode(self.name)
- def deployment_dir(self):
- return None
- def check_workspace(self, fs, user):
- # Create optional default root workspace for the first submission
- if REMOTE_SAMPLE_DIR.get() == REMOTE_SAMPLE_DIR.config.default_value:
- create_directories(fs, [REMOTE_SAMPLE_DIR.get()])
- Submission(user, self, fs, None, {})._create_dir(self.deployment_dir)
- Submission(user, self, fs, None, {})._create_dir(Hdfs.join(self.deployment_dir, 'lib'))
- def import_workspace(self, fs, source_deployment_dir, owner):
- try:
- fs.copy_remote_dir(source_deployment_dir, self.deployment_dir, owner=owner)
- except WebHdfsException, e:
- msg = _('The copy of the deployment directory failed: %s.') % e
- LOG.error(msg)
- raise PopupException(msg)
- class WorkflowConfiguration(object):
- APP_NAME = 'oozie-workflow'
- SLA_DEFAULT = [
- {'key': 'enabled', 'value': False}, # Always first element
- {'key': 'nominal-time', 'value': '${nominal_time}'},
- {'key': 'should-start', 'value': ''},
- {'key': 'should-end', 'value': '${30 * MINUTES}'},
- {'key': 'max-duration', 'value': ''},
- {'key': 'alert-events', 'value': ''},
- {'key': 'alert-contact', 'value': ''},
- {'key': 'notification-msg', 'value': ''},
- {'key': 'upstream-apps', 'value': ''},
- ]
- PROPERTIES = [
- {
- "multiple": True,
- "defaultValue": [
- {
- 'name': 'oozie.use.system.libpath',
- 'value': True
- }
- ],
- "value": [
- {
- 'name': 'oozie.use.system.libpath',
- 'value': True
- }
- ],
- "nice_name": _("Variables"),
- "key": "parameters",
- "help_text": _("Add one or more Oozie workflow job parameters."),
- "type": "parameters"
- }, {
- "multiple": False,
- "defaultValue": '',
- "value": '',
- "nice_name": _("Workspace"),
- "key": "deployment_dir",
- "help_text": _("Specify the deployment directory."),
- "type": "hdfs-file"
- }, {
- "multiple": True,
- "defaultValue": [],
- "value": [],
- "nice_name": _("Hadoop Properties"),
- "key": "properties",
- "help_text": _("Hadoop configuration properties."),
- "type": "settings"
- }, {
- "multiple": False,
- "defaultValue": True,
- "value": True,
- "nice_name": _("Show graph arrows"),
- "key": "show_arrows",
- "help_text": _("Toggles display of graph arrows."),
- "type": "boolean"
- }, {
- "multiple": False,
- "defaultValue": "uri:oozie:workflow:0.5",
- "value": "uri:oozie:workflow:0.5",
- "nice_name": _("Version"),
- "key": "schema_version",
- "help_text": _("Oozie XML Schema Version"),
- "type": "string",
- "options": [
- "uri:oozie:workflow:0.5",
- "uri:oozie:workflow:0.4.5",
- "uri:oozie:workflow:0.4",
- ]
- }, {
- "multiple": False,
- "defaultValue": '',
- "value": '',
- "nice_name": _("Job XML"),
- "key": "job_xml",
- "help_text": _("Oozie Job XML file"),
- "type": "hdfs-file"
- }, {
- "multiple": False,
- "defaultValue": False,
- "value": False,
- "nice_name": _("SLA Enabled"),
- "key": "sla_enabled",
- "help_text": _("SLA Enabled"),
- "type": "boolean"
- }, {
- "multiple": False,
- "defaultValue": SLA_DEFAULT,
- "value": SLA_DEFAULT,
- "nice_name": _("SLA Configuration"),
- "key": "sla",
- "help_text": _("Oozie SLA properties"),
- "type": "settings",
- "options": [prop['key'] for prop in SLA_DEFAULT]
- }
- ]
- class Workflow(Job):
- XML_FILE_NAME = 'workflow.xml'
- PROPERTY_APP_PATH = 'oozie.wf.application.path'
- HUE_ID = 'hue-id-w'
- def __init__(self, data=None, document=None, workflow=None, user=None):
- self.document = document
- self.user = user
- if document is not None:
- self.data = document.data
- elif data is not None:
- self.data = data
- else:
- if not workflow:
- workflow = self.get_default_workflow()
- workflow['properties'] = self.get_workflow_properties_for_user(user, workflow)
- self.data = json.dumps({
- 'layout': [{
- "size":12, "rows":[
- {"widgets":[{"size":12, "name":"Start", "id":"3f107997-04cc-8733-60a9-a4bb62cebffc", "widgetType":"start-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
- {"widgets":[{"size":12, "name":"End", "id":"33430f0f-ebfa-c3ec-f237-3e77efa03d0a", "widgetType":"end-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
- {"widgets":[{"size":12, "name":"Kill", "id":"17c9c895-5a16-7443-bb81-f34b30b21548", "widgetType":"kill-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]}
- ],
- "drops":[ "temp"],
- "klass":"card card-home card-column span12"
- }],
- 'workflow': workflow
- })
- @classmethod
- def get_application_path_key(cls):
- return 'oozie.wf.application.path'
- @classmethod
- def gen_workflow_data_from_xml(cls, user, oozie_workflow):
- node_list = []
- try:
- node_list = generate_v2_graph_nodes(oozie_workflow.definition)
- except MalformedWfDefException, e:
- LOG.exception("Could not find any nodes in Workflow definition. Maybe it's malformed?")
- except InvalidTagWithNamespaceException, e:
- LOG.exception(
- "Tag with namespace %(namespace)s is not valid. Please use one of the following namespaces: %(namespaces)s" % {
- 'namespace': e.namespace,
- 'namespaces': e.namespaces
- })
- _to_lowercase(node_list)
- adj_list = _create_graph_adjaceny_list(node_list)
- node_hierarchy = ['start']
- _get_hierarchy_from_adj_list(adj_list, adj_list['start']['ok_to'], node_hierarchy)
- _update_adj_list(adj_list)
- nodes_uuid_set = set()
- wf_rows = _create_workflow_layout(node_hierarchy, adj_list, nodes_uuid_set)
- data = {'layout': [{}], 'workflow': {}}
- if wf_rows:
- data['layout'][0]['rows'] = wf_rows
- wf_nodes = []
- nodes_uuid_set = set()
- _dig_nodes(node_hierarchy, adj_list, user, wf_nodes, nodes_uuid_set)
- data['workflow']['nodes'] = wf_nodes
- data['workflow']['id'] = '123'
- data['workflow']['properties'] = cls.get_workflow_properties_for_user(user, workflow=None)
- data['workflow']['properties'].update({
- 'deployment_dir': '/user/hue/oozie/workspaces/hue-oozie-1452553957.19'
- })
- return data
- @classmethod
- def get_default_workflow(cls):
- return {
- "id": None,
- "uuid": None,
- "name": "My Workflow",
- "nodes": [
- {"id": "3f107997-04cc-8733-60a9-a4bb62cebffc", "name": "Start", "type": "start-widget", "properties": {},
- "children": [{'to': '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'}]},
- {"id": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a", "name": "End", "type": "end-widget", "properties": {},
- "children": []},
- {"id": "17c9c895-5a16-7443-bb81-f34b30b21548", "name": "Kill", "type": "kill-widget",
- "properties": {'message': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]')},
- "children": []}
- ]
- }
- @classmethod
- def get_workflow_properties_for_user(cls, user, workflow=None):
- workflow = workflow if workflow is not None else {}
- properties = workflow.get('properties', None)
- if not properties:
- config = None
- properties = cls.get_properties()
- if user is not None:
- if USE_DEFAULT_CONFIGURATION.get():
- config = DefaultConfiguration.objects.get_configuration_for_user(app=WorkflowConfiguration.APP_NAME, user=user)
- if config is not None:
- properties.update(config.properties_dict)
- properties.update({
- 'wf1_id': None,
- 'description': ''
- })
- return properties
- @staticmethod
- def get_properties():
- return dict((prop['key'], prop['value']) for prop in WorkflowConfiguration.PROPERTIES)
- @property
- def id(self):
- return self.document.id if self.document else None
- @property
- def uuid(self):
- return self.document.uuid
- @property
- def name(self):
- _data = self.get_data()
- return _data['workflow']['name']
- @property
- def deployment_dir(self):
- _data = self.get_data()
- return _data['workflow']['properties']['deployment_dir']
- @property
- def parameters(self):
- _data = self.get_data()
- return _data['workflow']['properties']['parameters']
- @property
- def sla_enabled(self):
- _data = self.get_data()
- return _data['workflow']['properties']['sla'][0].get('value')
- @property
- def has_some_slas(self):
- return self.sla_enabled or any([node.sla_enabled for node in self.nodes])
- @property
- def credentials(self):
- return list(set([cred for node in self.nodes for cred in node.data['properties']['credentials']]))
- @property
- def sla(self):
- _data = self.get_data()
- return _data['workflow']['properties']['sla']
- @property
- def nodes(self):
- _data = self.get_data()
- return [Node(node, self.user) for node in _data['workflow']['nodes']]
- def find_parameters(self):
- params = set()
- for param in find_dollar_braced_variables(self.name):
- params.add(param)
- if self.sla_enabled:
- for param in find_json_parameters(self.sla):
- params.add(param)
- parameters = dict([(param, '') for param in list(params)])
- for node in self.nodes:
- if 'document' in node.data['type']:
- for param in node.data['properties']['parameters']:
- if param['value'] and '=' in param['value']:
- name, val = param['value'].split('=', 1)
- parameters[name] = val
- extra_fields = []
- if node.data['properties'].get('key_tab_path'):
- extra_fields.append('key_tab_path')
- if node.data['properties'].get('user_principal'):
- extra_fields.append('user_principal')
- extra = find_parameters(node, fields=extra_fields)
- else:
- extra = node.find_parameters()
- if extra:
- parameters.update(dict([(param, '') for param in list(extra)]))
- return parameters
- def get_json(self):
- _data = self.get_data()
- return json.dumps(_data)
- def get_data(self):
- _data = json.loads(self.data)
- if self.document is not None:
- _data['workflow']['id'] = self.document.id
- _data['workflow']['dependencies'] = list(self.document.dependencies.values('uuid', ))
- else:
- _data['workflow']['dependencies'] = []
- if 'parameters' not in _data['workflow']['properties']:
- _data['workflow']['properties']['parameters'] = [
- {'name': 'oozie.use.system.libpath', 'value': True},
- ]
- if 'show_arrows' not in _data['workflow']['properties']:
- _data['workflow']['properties']['show_arrows'] = True
- for node in _data['workflow']['nodes']:
- if 'credentials' in node['properties']: # If node is an Action
- if 'retry_max' not in node['properties']: # When displaying a workflow
- node['properties']['retry_max'] = []
- if 'retry_interval' not in node['properties']:
- node['properties']['retry_interval'] = []
- # Backward compatibility
- _upgrade_older_node(node)
- return _data
- def to_xml(self, mapping=None):
- if mapping is None:
- mapping = {}
- tmpl = 'editor2/gen/workflow.xml.mako'
- data = self.get_data()
- 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
- node_mapping = dict([(node.id, node) for node in nodes])
- sub_wfs_ids = [node.data['properties']['workflow'] for node in nodes if node.data['type'] == 'subworkflow']
- workflow_mapping = dict(
- [(workflow.uuid, Workflow(document=workflow, user=self.user)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)])
- xml = re.sub(re.compile('>\s*\n+', re.MULTILINE), '>\n', django_mako.render_to_string(tmpl, {
- 'wf': self,
- 'workflow': data['workflow'],
- 'nodes': nodes,
- 'mapping': mapping,
- 'node_mapping': node_mapping,
- 'workflow_mapping': workflow_mapping
- }))
- return force_unicode(xml.strip())
- def get_absolute_url(self):
- return reverse('oozie:edit_workflow') + '?workflow=%s' % self.id if self.document else ''
- def override_subworkflow_id(self, sub_wf_action, workflow_id):
- _data = self.get_data()
- action = [_action for _action in _data['workflow']['nodes'] if _action['id'] == sub_wf_action.id]
- if action:
- action[0]['properties']['job_properties'].append({'name': Workflow.HUE_ID, 'value': workflow_id})
- self.data = json.dumps(_data)
- def update_name(self, name):
- _data = self.get_data()
- _data['workflow']['name'] = name
- self.data = json.dumps(_data)
- def set_workspace(self, user):
- _data = json.loads(self.data)
- _data['workflow']['properties']['deployment_dir'] = Job.get_workspace(user)
- self.data = json.dumps(_data)
- def create_single_action_workflow_data(self, node_id):
- _data = json.loads(self.data)
- start_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'Start'][0]
- submit_node = [node for node in _data['workflow']['nodes'] if node['id'] == node_id][0]
- end_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'End'][0]
- kill_node = [node for node in _data['workflow']['nodes'] if node['name'] == 'Kill'][0]
- # Modify children to point Start -> Submit_node -> End/Kill
- start_node['children'] = [{'to': submit_node['id']}]
- submit_node['children'] = [{'to': end_node['id']}, {'error': kill_node['id']}]
- _data['workflow']['properties']['deployment_dir'] = None
- # Recursively find the widget node
- def _get_node(rows, node_id):
- for row in rows:
- if not row['widgets']:
- for col in row['columns']:
- node = _get_node(col['rows'], node_id)
- if node:
- return node
- elif row['widgets'][0]['id'] == node_id:
- return row
- # Create wf data with above nodes
- return json.dumps({
- 'layout': [{
- "size": 12,
- "rows": [
- [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'Start'][0],
- _get_node(_data['layout'][0]['rows'], node_id),
- [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'End'][0],
- [row for row in _data['layout'][0]['rows'] if row['widgets'] and row['widgets'][0]['name'] == 'Kill'][0]
- ],
- "drops": ["temp"],
- "klass": "card card-home card-column span12"
- }],
- 'workflow': {
- "id": None,
- "uuid": None,
- "name": _data['workflow']['name'],
- "properties": _data['workflow']['properties'],
- "nodes": [start_node, submit_node, end_node, kill_node]
- }
- })
- # Updates node_list to lowercase names
- # To avoid case-sensitive failures
- def _to_lowercase(node_list):
- for node in node_list:
- for key in node.keys():
- if hasattr(node[key], 'lower'):
- node[key] = node[key].lower()
- def _update_adj_list(adj_list):
- uuids = {}
- id = 1
- first_kill_node_seen = False
- for node in adj_list.keys():
- adj_list[node]['id'] = id
- # Oozie uses same action for streaming and mapreduce but Hue manages them differently
- if adj_list[node]['node_type'] == 'map-reduce':
- if 'streaming' in adj_list[node]['name']:
- adj_list[node]['node_type'] = 'streaming'
- else:
- adj_list[node]['node_type'] = 'mapreduce'
- elif adj_list[node]['node_type'] == 'sub-workflow':
- adj_list[node]['node_type'] = 'subworkflow'
- if adj_list[node]['node_type'] == 'kill':
- # JS requires at least one of the kill nodes to have this Id
- if not first_kill_node_seen:
- adj_list[node]['uuid'] = '17c9c895-5a16-7443-bb81-f34b30b21548'
- first_kill_node_seen = True
- else:
- adj_list[node]['uuid'] = str(uuid.uuid4())
- elif adj_list[node]['node_type'] == 'start':
- adj_list[node]['uuid'] = '3f107997-04cc-8733-60a9-a4bb62cebffc'
- elif adj_list[node]['node_type'] == 'end':
- adj_list[node]['uuid'] = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
- else:
- adj_list[node]['uuid'] = node[-4:] + str(uuid.uuid4())[4:]
- uuids[id] = adj_list[node]['uuid']
- id += 1
- return adj_list
- def _dig_nodes(nodes, adj_list, user, wf_nodes, nodes_uuid_set):
- for node in nodes:
- if type(node) != list:
- node = adj_list[node]
- if node['uuid'] not in nodes_uuid_set:
- properties = {}
- if '%s-widget' % node['node_type'] in NODES:
- properties = dict(NODES['%s-widget' % node['node_type']].get_fields())
- if node['node_type'] == 'pig':
- properties['script_path'] = node.get('pig').get('script_path')
- elif node['node_type'] == 'spark':
- properties['class'] = node.get('spark').get('class')
- properties['jars'] = node.get('spark').get('jar')
- elif node['node_type'] == 'hive' or node['node_type'] == 'hive2':
- properties['script_path'] = node.get('hive').get('script')
- elif node['node_type'] == 'java':
- properties['main_class'] = node.get('java').get('main-class')
- elif node['node_type'] == 'sqoop':
- properties['command'] = node.get('sqoop').get('command')
- elif node['node_type'] == 'mapreduce':
- properties['job_properties'] = node.get('job_properties')
- elif node['node_type'] == 'shell':
- properties['shell_command'] = node.get('shell').get('command')
- elif node['node_type'] == 'ssh':
- properties['user'] = '%s@%s' % (node.get('ssh').get('user'), node.get('ssh').get('host'))
- properties['ssh_command'] = node.get('ssh').get('command')
- elif node['node_type'] == 'fs':
- properties['touchzs'] = node.get('fs').get('touchzs')
- properties['mkdirs'] = node.get('fs').get('mkdirs')
- properties['moves'] = node.get('fs').get('moves')
- properties['deletes'] = node.get('fs').get('deletes')
- elif node['node_type'] == 'email':
- properties['to'] = node.get('email').get('to')
- properties['subject'] = node.get('email').get('subject')
- elif node['node_type'] == 'streaming':
- properties['mapper'] = node.get('streaming').get('mapper')
- properties['reducer'] = node.get('streaming').get('reducer')
- elif node['node_type'] == 'distcp':
- properties['distcp_parameters'] = node.get('params')
- elif node['node_type'] == 'subworkflow':
- properties['app-path'] = node.get('subworkflow').get('app-path')
- properties['workflow'] = node.get('uuid')
- properties['job_properties'] = []
- properties['sla'] = ''
- children = []
- if node['node_type'] in ('fork', 'decision'):
- for key in node.keys():
- if key.startswith('path'):
- children.append({'to': adj_list[node[key]]['uuid'], 'condition': '${ 1 gt 0 }'})
- if node['node_type'] == 'decision':
- children.append({'to': adj_list[node['default']]['uuid'], 'condition': '${ 1 gt 0 }'})
- else:
- if node.get('ok_to'):
- children.append({'to': adj_list[node['ok_to']]['uuid']})
- if node.get('error_to'):
- children.append({'error': adj_list[node['error_to']]['uuid']})
- nodes_uuid_set.add(node['uuid'])
- wf_nodes.append({
- "id": node['uuid'],
- "name": '%s-%s' % (node['node_type'].split('-')[0], node['uuid'][:4]),
- "type": "%s-widget" % node['node_type'],
- "properties": properties,
- "children": children
- })
- else:
- _dig_nodes(node, adj_list, user, wf_nodes, nodes_uuid_set)
- def _create_workflow_layout(nodes, adj_list, nodes_uuid_set, size=12):
- wf_rows = []
- for node in nodes:
- if type(node) == list and len(node) == 1:
- node = node[0]
- if type(node) != list:
- _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node]['uuid'],
- 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":[]}]})
- else:
- if adj_list[node[0]]['node_type'] in ('fork', 'decision'):
- _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id=adj_list[node[0]]['uuid'],
- 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":[]}]})
- wf_rows.append({
- "id": str(uuid.uuid4()),
- "widgets":[
- ],
- "columns":[
- {
- "id": str(uuid.uuid4()),
- "size": (size / len(node[1])),
- "rows":
- [{
- "id": str(uuid.uuid4()),
- "widgets": c['widgets'],
- "columns":c.get('columns') or []
- } for c in col],
- "klass":"card card-home card-column span%s" % (size / len(node[1]))
- }
- for col in [_create_workflow_layout(item, adj_list, nodes_uuid_set, size) for item in node[1]]
- ]
- })
- if adj_list[node[0]]['node_type'] == 'fork':
- 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":[]}]})
- else:
- wf_rows.append(_create_workflow_layout(node, adj_list, nodes_uuid_set, size))
- return wf_rows
- # Prevent duplicate nodes in graph layout
- def _append_to_wf_rows(wf_rows, nodes_uuid_set, row_id, row):
- if row['widgets'][0]['id'] not in nodes_uuid_set:
- nodes_uuid_set.add(row['widgets'][0]['id'])
- wf_rows.append(row)
- def _get_hierarchy_from_adj_list(adj_list, curr_node, node_hierarchy):
- _get_hierarchy_from_adj_list_helper(adj_list, curr_node, node_hierarchy)
- # Add End and Kill nodes to node_hierarchy
- for key in adj_list.keys():
- if adj_list[key]['node_type'] == 'kill':
- node_hierarchy.append([adj_list[key]['name']])
- node_hierarchy.append([adj_list[key]['name'] for key in adj_list.keys() if adj_list[key]['node_type'] == 'end'])
- def _get_hierarchy_from_adj_list_helper(adj_list, curr_node, node_hierarchy):
- if not curr_node or adj_list[curr_node]['node_type'] in ('join', 'end', 'kill'):
- return curr_node
- elif adj_list[curr_node]['node_type'] in ('fork', 'decision'):
- branch_nodes = []
- branch_nodes.append(curr_node)
- join_node = None
- children = []
- for key in adj_list[curr_node].keys():
- if key.startswith('path') or key == 'default':
- child = []
- return_node = _get_hierarchy_from_adj_list_helper(adj_list, adj_list[curr_node][key], child)
- join_node = return_node if not join_node else join_node
- if child:
- children.append(child)
- branch_nodes.append(children)
- if adj_list[curr_node]['node_type'] == 'fork':
- branch_nodes.append(join_node)
- node_hierarchy.append(branch_nodes)
- return _get_hierarchy_from_adj_list_helper(adj_list, adj_list[join_node]['ok_to'], node_hierarchy)
- node_hierarchy.append(branch_nodes)
- return join_node
- else:
- node_hierarchy.append(curr_node)
- return _get_hierarchy_from_adj_list_helper(adj_list, adj_list[curr_node]['ok_to'], node_hierarchy)
- def _create_graph_adjaceny_list(nodes):
- start_node = [node for node in nodes if node.get('node_type') == 'start'][0]
- adj_list = {'start': start_node}
- for node in nodes:
- if node and node.get('node_type') != 'start':
- adj_list[node['name']] = node
- return adj_list
- class Node():
- def __init__(self, data, user=None):
- self.data = data
- self.user = user
- self._augment_data()
- def to_xml(self, mapping=None, node_mapping=None, workflow_mapping=None):
- if mapping is None:
- mapping = {}
- if node_mapping is None:
- node_mapping = {}
- if workflow_mapping is None:
- workflow_mapping = {}
- if self.data['type'] in ('hive2', 'hive-document') and not self.data['properties']['jdbc_url']:
- self.data['properties']['jdbc_url'] = _get_hiveserver2_url()
- if self.data['type'] == 'fork':
- links = [link for link in self.data['children'] if link['to'] in node_mapping]
- if len(links) != len(self.data['children']):
- LOG.warn('Fork has some children links that do not exist, ignoring them: links %s, existing links %s, links %s, existing links %s' \
- % (len(links), len(self.data['children']), links, self.data['children']))
- self.data['children'] = links
- if self.data['type'] == JavaDocumentAction.TYPE:
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- properties = notebook.get_data()['snippets'][0]['properties']
- self.data['properties']['main_class'] = properties['class']
- self.data['properties']['app_jar'] = properties['app_jar'] # Not used here
- self.data['properties']['files'] = [{'value': f['path']} for f in properties['files']]
- self.data['properties']['arguments'] = [{'value': prop} for prop in properties['arguments']]
- elif self.data['type'] == SparkDocumentAction.TYPE or self.data['type'] == 'spark-document':
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- properties = notebook.get_data()['snippets'][0]['properties']
- if self.data['type'] == 'spark-document': # Oozie Document Action
- self.data['properties']['app_name'] = properties['app_name']
- self.data['properties']['class'] = properties['class']
- self.data['properties']['jars'] = os.path.basename(properties['jars'][0])
- self.data['properties']['files'] = [{'value': f} for f in properties['jars']] + [{'value': f['path']} for f in properties['files']]
- self.data['properties']['spark_arguments'] = [{'value': prop} for prop in properties['spark_arguments']]
- self.data['properties']['spark_opts'] = ' '.join(properties['spark_opts'])
- if len(properties['jars']) > 1:
- self.data['properties']['spark_opts'] += ' --py-files ' + ','.join([os.path.basename(f) for f in properties['jars'][1:]])
- elif self.data['type'] == PigDocumentAction.TYPE:
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- action = notebook.get_data()['snippets'][0]
- name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
- self.data['properties']['script_path'] = "${wf:appPath()}/" + name + ".pig"
- self.data['properties']['parameters'] = [{'value': prop} for prop in action['properties']['parameters']]
- self.data['properties']['arguments'] = []
- job_properties = []
- for prop in action['properties']['hadoopProperties']:
- name, value = prop.split('=', 1)
- job_properties.append({'name': name, 'value': value})
- self.data['properties']['job_properties'] = job_properties
- self.data['properties']['files'] = [{'value': prop} for prop in action['properties']['parameters']]
- elif self.data['type'] == SqoopDocumentAction.TYPE:
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- action = notebook.get_data()['snippets'][0]
- name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
- command = action['statement']
- if command.startswith('sqoop '):
- _ignore, command = command.split('sqoop ', 1)
- self.data['properties']['command'] = command
- self.data['properties']['files'] = [{'value': f['path']} for f in action['properties']['files']]
- self.data['properties']['arguments'] = []
- elif self.data['type'] == DistCpDocumentAction.TYPE:
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- action = notebook.get_data()['snippets'][0]
- name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
- self.data['properties']['source_path'] = action['properties']['source_path']
- self.data['properties']['destination_path'] = action['properties']['destination_path']
- elif self.data['type'] == ShellDocumentAction.TYPE:
- if self.data['properties'].get('uuid'):
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- action = notebook.get_data()['snippets'][0]
- name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
- self.data['properties']['shell_command'] = action['properties']['command_path']
- self.data['properties']['env_var'] = [{'value': prop} for prop in action['properties']['env_var']]
- self.data['properties']['capture_output'] = action['properties']['capture_output']
- self.data['properties']['arguments'] = [{'value': prop} for prop in action['properties']['arguments']]
- 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']]
- self.data['properties']['archives'] = [{'value': prop} for prop in action['properties']['archives']]
- elif self.data['type'] == MapReduceDocumentAction.TYPE:
- notebook = Notebook(document=Document2.objects.get_by_uuid(user=self.user, uuid=self.data['properties']['uuid']))
- action = notebook.get_data()['snippets'][0]
- name = '%s-%s' % (self.data['type'].split('-')[0], self.data['id'][:4])
- self.data['properties']['app_jar'] = action['properties']['app_jar']
- self.data['properties']['arguments'] = []
- self.data['properties']['parameters'] = []
- job_properties = []
- for prop in action['properties']['hadoopProperties']:
- name, value = prop.split('=', 1)
- job_properties.append({'name': name, 'value': value})
- self.data['properties']['job_properties'] = job_properties
- self.data['properties']['files'] = [{'value': prop} for prop in action['properties']['files']]
- self.data['properties']['archives'] = [{'value': prop} for prop in action['properties']['archives']]
- elif self.data['type'] == ImpalaAction.TYPE or self.data['type'] == ImpalaDocumentAction.TYPE:
- shell_command_name = self.data['name'] + '.sh'
- self.data['properties']['shell_command'] = shell_command_name
- self.data['properties']['env_var'] = []
- self.data['properties']['capture_output'] = False
- self.data['properties']['arguments'] = []
- if self.data['type'] == ImpalaAction.TYPE:
- script_path = self.data['properties'].get('script_path')
- else:
- script_path = self.data['name'] + '.sql'
- files = [{'value': shell_command_name}, {'value': script_path}]
- if self.data['properties']['key_tab_path']:
- files.append({'value': self.data['properties']['key_tab_path']})
- self.data['properties']['files'] = files
- self.data['properties']['archives'] = []
- data = {
- 'node': self.data,
- 'mapping': mapping,
- 'node_mapping': node_mapping,
- 'workflow_mapping': workflow_mapping
- }
- if mapping.get('send_email'):
- if self.data['type'] == KillAction.TYPE and not self.data['properties'].get('enableMail'):
- self.data['properties']['enableMail'] = True
- self.data['properties']['to'] = self.user.email
- self.data['properties']['subject'] = _("${wf:name()} execution failure")
- self.data['properties']['body'] = _("Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]")
- if self.data['type'] == EndNode.TYPE:
- self.data['properties']['enableMail'] = True
- self.data['properties']['to'] = self.user.email
- self.data['properties']['subject'] = _("${wf:name()} execution successful")
- if mapping.get('send_result_path'):
- if self.data['type'] == EndNode.TYPE:
- self.data['properties']['body'] = 'View result file at %(send_result_browse_url)s' % mapping
- return django_mako.render_to_string(self.get_template_name(), data)
- @property
- def id(self):
- return self.data['id']
- @property
- def name(self):
- return self.data['name']
- @property
- def sla_enabled(self):
- return 'sla' in self.data['properties'] and self.data['properties']['sla'] and self.data['properties']['sla'][0].get('value')
- def _augment_data(self):
- self.data['type'] = self.data['type'].replace('-widget', '')
- self.data['uuid'] = self.data['id']
- # Action Node
- if 'credentials' not in self.data['properties']:
- self.data['properties']['credentials'] = []
- if 'prepares' not in self.data['properties']:
- self.data['properties']['prepares'] = []
- if 'job_xml' not in self.data['properties']:
- self.data['properties']['job_xml'] = []
- if 'properties' not in self.data['properties']:
- self.data['properties']['properties'] = []
- if 'params' not in self.data['properties']:
- self.data['properties']['params'] = []
- if 'files' not in self.data['properties']:
- self.data['properties']['files'] = []
- if 'archives' not in self.data['properties']:
- self.data['properties']['archives'] = []
- if 'sla' not in self.data['properties']:
- self.data['properties']['sla'] = WorkflowConfiguration.SLA_DEFAULT
- if 'retry_max' not in self.data['properties']:
- self.data['properties']['retry_max'] = []
- if 'retry_interval' not in self.data['properties']:
- self.data['properties']['retry_interval'] = []
- # Backward compatibility
- _upgrade_older_node(self.data)
- def get_template_name(self):
- node_type = self.data['type']
- if self.data['type'] == JavaDocumentAction.TYPE:
- node_type = JavaAction.TYPE
- elif self.data['type'] == ImpalaAction.TYPE or self.data['type'] == ImpalaDocumentAction.TYPE:
- node_type = ShellAction.TYPE
- return 'editor2/gen/workflow-%s.xml.mako' % node_type
- def find_parameters(self):
- return find_parameters(self) + (find_parameters(self, ['sla']) if self.sla_enabled else [])
- def _upgrade_older_node(node):
- if node['type'] in ('sqoop', 'sqoop-widget') and 'arguments' not in node['properties']:
- node['properties']['arguments'] = node['properties']['parameters']
- if node['type'] in ('kill', 'kill-widget') and 'to' not in node['properties']:
- node['properties']['enableMail'] = False
- node['properties']['to'] = ''
- node['properties']['cc'] = ''
- node['properties']['subject'] = ''
- node['properties']['body'] = ''
- if node['type'] in ('end', 'end-widget') and 'to' not in node['properties']:
- node['properties']['enableMail'] = False
- node['properties']['to'] = ''
- node['properties']['cc'] = ''
- node['properties']['subject'] = ''
- node['properties']['body'] = ''
- node['properties']['content_type'] = 'text/plain'
- node['properties']['attachment'] = ''
- if node['type'] == 'email-widget' and 'bcc' not in node['properties']:
- node['properties']['bcc'] = ''
- node['properties']['content_type'] = 'text/plain'
- node['properties']['attachment'] = ''
- if node['type'] == 'spark-widget' and 'files' not in node['properties']:
- node['properties']['files'] = []
- if (node['type'] == 'hive2-widget' or node['type'] == 'hive-document-widget') and 'arguments' not in node['properties']:
- node['properties']['arguments'] = []
- class Action(object):
- @classmethod
- def get_fields(cls):
- credentials = [cls.DEFAULT_CREDENTIALS] if hasattr(cls, 'DEFAULT_CREDENTIALS') and cls.DEFAULT_CREDENTIALS else []
- return [(f['name'], f['value']) for f in cls.FIELDS.itervalues()] + [('sla', WorkflowConfiguration.SLA_DEFAULT), ('credentials', credentials)]
- class StartNode(Action):
- TYPE = 'start'
- FIELDS = {}
- class EndNode(Action):
- TYPE = 'end'
- FIELDS = {}
- class PigAction(Action):
- TYPE = 'pig'
- FIELDS = {
- 'script_path': {
- 'name': 'script_path',
- 'label': _('Script'),
- 'value': '',
- 'help_text': _('Path to the script on HDFS.'),
- 'type': ''
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The Pig parameters of the script without -param. e.g. INPUT=${inputDir}'),
- 'type': ''
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('The Pig parameters of the script as is. e.g. -param, INPUT=${inputDir}'),
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': [],
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['script_path']]
- class JavaAction(Action):
- TYPE = 'java'
- FIELDS = {
- 'jar_path': {
- 'name': 'jar_path',
- 'label': _('Jar name'),
- 'value': '',
- 'help_text': _('Path to the jar on HDFS.'),
- 'type': ''
- },
- 'main_class': {
- 'name': 'main_class',
- 'label': _('Main class'),
- 'value': '',
- 'help_text': _('Java class. e.g. org.apache.hadoop.examples.Grep'),
- 'type': 'text'
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('Arguments of the main method. The value of each arg element is considered a single argument '
- 'and they are passed to the main method in the same order.'),
- 'type': ''
- },
- 'java_opts': {
- 'name': 'java_opts',
- 'label': _('Java options'),
- 'value': [],
- 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b'),
- 'type': ''
- },
- 'capture_output': {
- 'name': 'capture_output',
- 'label': _('Capture output'),
- 'value': False,
- 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
- 'command output must be in Java Properties file format and it must not exceed 2KB. '
- 'From within the workflow definition, the output of an %(program)s action node is accessible '
- 'via the String action:output(String node, String key) function') % {'program': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': [],
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['jar_path'], cls.FIELDS['main_class']]
- class HiveAction(Action):
- TYPE = 'hive'
- DEFAULT_CREDENTIALS = 'hcat'
- FIELDS = {
- 'script_path': {
- 'name': 'script_path',
- 'label': _('Script'),
- 'value': '',
- 'help_text': _('Path to the script on HDFS.'),
- 'type': ''
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'hive_xml': {
- 'name': 'hive_xml',
- 'label': _('Hive XML'),
- 'value': '',
- 'help_text': _('Refer to a hive-site.xml for connecting to Hive'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['script_path'], cls.FIELDS['hive_xml']]
- def _get_hiveserver2_url():
- try:
- from beeswax.hive_site import hiveserver2_jdbc_url
- return hiveserver2_jdbc_url()
- except Exception, e:
- # Might fail is Hive is disabled
- LOG.exception('Could not guess HiveServer2 URL: %s' % smart_str(e))
- return 'jdbc:hive2://localhost:10000/default'
- class HiveServer2Action(Action):
- TYPE = 'hive2'
- DEFAULT_CREDENTIALS = 'hive2'
- FIELDS = {
- 'script_path': {
- 'name': 'script_path',
- 'label': _('Script'),
- 'value': '',
- 'help_text': _('Path to the script on HDFS.'),
- 'type': ''
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('Arguments for beeline. E.g. --showHeader=true, -Djavax.net.ssl.trustStore=/etc/cdep-ssl-conf/CA_STANDARD/truststore.jks'),
- 'type': []
- },
- 'jdbc_url': {
- 'name': 'jdbc_url',
- 'label': _('HiveServer2 URL'),
- 'value': "",
- 'help_text': _('e.g. jdbc:hive2://localhost:10000/default. JDBC URL for the Hive Server 2.'),
- 'type': ''
- },
- 'password': {
- 'name': 'password',
- 'label': _('Password'),
- 'value': '',
- '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 '
- 'something requiring a password (e.g. LDAP); non-secured Hive Server 2 or Kerberized Hive Server 2 don\'t require a password.'),
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['script_path']]
- def _get_impala_url():
- try:
- from impala.dbms import get_query_server_config
- return get_query_server_config()['server_host']
- except Exception, e:
- # Might fail is Impala is disabled
- LOG.exception('Could not get Impalad URL: %s' % smart_str(e))
- return 'localhost'
- class ImpalaAction(HiveServer2Action):
- # Executed as shell action until Oozie supports an Impala Action
- TYPE = 'impala'
- DEFAULT_CREDENTIALS = '' # None at this time, need to upload user keytab
- FIELDS = HiveServer2Action.FIELDS.copy()
- del FIELDS['jdbc_url']
- del FIELDS['password']
- FIELDS['impalad_host'] = {
- 'name': 'impalad_host',
- 'label': _('Impalad hostname'),
- 'value': "",
- 'help_text': _('e.g. impalad-001.cluster.com. The hostname of the Impalad to send the query to.'),
- 'type': ''
- }
- FIELDS['key_tab_path'] = {
- 'name': 'key_tab_path',
- 'label': _('Keytab path'),
- 'value': '${key_tab_path}' if SECURITY_ENABLED.get() else '',
- 'help_text': _('Path to the keytab to use when on a secure cluster, e.g. /user/joe/joe.keytab.'),
- 'type': ''
- }
- FIELDS['user_principal'] = {
- 'name': 'user_principal',
- 'label': _('User principal'),
- 'value': '${user_principal}' if SECURITY_ENABLED.get() else '',
- 'help_text': _('Name of the principal to use in the kinit, e.g.: kinit -k -t /home/joe/joe.keytab joe@PROD.EDH.'),
- 'type': ''
- }
- class SubWorkflowAction(Action):
- TYPE = 'subworkflow'
- FIELDS = {
- 'workflow': {
- 'name': 'workflow',
- 'label': _('Sub-workflow'),
- 'value': None,
- 'help_text': _('The sub-workflow application to include. You must own all the sub-workflows'),
- 'type': 'workflow'
- },
- 'propagate_configuration': {
- 'name': 'propagate_configuration',
- 'label': _('Propagate configuration'),
- 'value': True,
- 'help_text': _('If the workflow job configuration should be propagated to the child workflow.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('Can be used to specify the job properties that are required to run the child workflow job.'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['workflow']]
- class SqoopAction(Action):
- TYPE = 'sqoop'
- FIELDS = {
- 'command': {
- 'name': 'command',
- 'label': _('Sqoop command'),
- 'value': 'import --connect jdbc:hsqldb:file:db.hsqldb --table TT --target-dir hdfs://localhost:8020/user/foo -m 1',
- '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},
- 'type': 'textarea'
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('If no command is specified, split the command by spaces and insert the %(type)s parameters '
- 'here e.g. import, --connect, jdbc:hsqldb:file:db.hsqldb, ...') % {'type': TYPE},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['command']]
- class MapReduceAction(Action):
- TYPE = 'mapreduce'
- FIELDS = {
- 'jar_path': {
- 'name': 'jar_path',
- 'label': _('Jar name'),
- 'value': '',
- 'help_text': _('Path to the jar on HDFS.'),
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['jar_path']]
- class ShellAction(Action):
- TYPE = 'shell'
- FIELDS = {
- 'shell_command': {
- 'name': 'shell_command',
- 'label': _('Shell command'),
- 'value': '',
- 'help_text': _('Shell command to execute, e.g script.sh'),
- 'type': ''
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('One arg, e.g. -l, --help'),
- 'type': ''
- },
- 'env_var': {
- 'name': 'env_var',
- 'label': _('Environment variables'),
- 'value': [],
- 'help_text': _('e.g. MAX=10 or PATH=$PATH:mypath'),
- 'type': ''
- },
- 'capture_output': {
- 'name': 'capture_output',
- 'label': _('Capture output'),
- 'value': True,
- 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
- 'command output must be in Java Properties file format and it must not exceed 2KB. '
- 'From within the workflow definition, the output of an %(program)s action node is accessible '
- 'via the String action:output(String node, String key) function') % {'program': TYPE},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['shell_command']]
- class SshAction(Action):
- TYPE = 'ssh'
- FIELDS = {
- 'host': {
- 'name': 'host',
- 'label': _('User and Host'),
- 'value': 'user@host.com',
- 'help_text': _('Where the shell will be executed.'),
- 'type': 'text'
- },
- 'ssh_command': {
- 'name': 'ssh_command',
- 'label': _('Ssh command'),
- 'value': 'ls',
- 'help_text': _('The path of the Shell command to execute.'),
- 'type': 'textarea'
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('One arg, e.g. -l, --help'),
- 'type': ''
- },
- 'capture_output': {
- 'name': 'capture_output',
- 'label': _('Capture output'),
- 'value': True,
- 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
- 'command output must be in Java Properties file format and it must not exceed 2KB. '
- 'From within the workflow definition, the output of an %(program)s action node is accessible '
- 'via the String action:output(String node, String key) function') % {'program': TYPE},
- 'type': ''
- },
- # Common
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['host'], cls.FIELDS['ssh_command']]
- class FsAction(Action):
- TYPE = 'fs'
- FIELDS = {
- 'deletes': {
- 'name': 'deletes',
- 'label': _('Delete path'),
- 'value': [],
- 'help_text': _('Deletes recursively all content.'),
- 'type': ''
- },
- 'mkdirs': {
- 'name': 'mkdirs',
- 'label': _('Create directory'),
- 'value': [],
- 'help_text': _('Sub directories are created if needed.'),
- 'type': ''
- },
- 'moves': {
- 'name': 'moves',
- 'label': _('Move file or directory'),
- 'value': [],
- 'help_text': _('Destination.'),
- 'type': ''
- },
- 'chmods': {
- 'name': 'chmods',
- 'label': _('Change permissions'),
- 'value': [],
- 'help_text': _('File or directory.'),
- 'type': ''
- },
- 'touchzs': {
- 'name': 'touchzs',
- 'label': _('Create or touch a file'),
- 'value': [],
- 'help_text': _('Or update its modification date.'),
- 'type': ''
- },
- 'chgrps': {
- 'name': 'chgrps',
- 'label': _('Change the group'),
- 'value': [],
- 'help_text': _('File or directory.'),
- 'type': ''
- },
- # Common
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return []
- class EmailAction(Action):
- TYPE = 'email'
- FIELDS = {
- 'to': {
- 'name': 'to',
- 'label': _('To addresses'),
- 'value': '',
- 'help_text': _('Comma-separated values'),
- 'type': 'text'
- },
- 'cc': {
- 'name': 'cc',
- 'label': _('cc'),
- 'value': '',
- 'help_text': _('Comma-separated values'),
- 'type': 'text'
- },
- 'bcc': {
- 'name': 'bcc',
- 'label': _('bcc'),
- 'value': '',
- 'help_text': _('Comma-separated values'),
- 'type': 'text'
- },
- 'subject': {
- 'name': 'subject',
- 'label': _('Subject'),
- 'value': '',
- 'help_text': _('Plain-text'),
- 'type': 'text'
- },
- 'body': {
- 'name': 'body',
- 'label': _('Body'),
- 'value': '',
- 'help_text': _('Plain-text'),
- 'type': 'textarea'
- },
- 'attachment': {
- 'name': 'attachment',
- 'label': _('Attachment'),
- 'value': '',
- 'help_text': _('Comma separated list of HDFS files.'),
- 'type': ''
- },
- 'content_type': {
- 'name': 'content_type',
- 'label': _('Content-type'),
- 'value': 'text/plain',
- 'help_text': _('Default is text/plain'),
- 'type': 'text'
- },
- # Common
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['to'], cls.FIELDS['subject'], cls.FIELDS['body']]
- class StreamingAction(Action):
- TYPE = 'streaming'
- FIELDS = {
- 'mapper': {
- 'name': 'mapper',
- 'label': _('Mapper'),
- 'value': '',
- 'help_text': _('The executable/script to be used as mapper.'),
- 'type': ''
- },
- 'reducer': {
- 'name': 'reducer',
- 'label': _('Reducer'),
- 'value': '',
- 'help_text': _('The executable/script to be used as reducer.'),
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.')
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.')
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production')
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.')
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml')
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['mapper'], cls.FIELDS['reducer']]
- class DistCpAction(Action):
- TYPE = 'distcp'
- FIELDS = {
- 'distcp_parameters': {
- 'name': 'distcp_parameters',
- 'label': _('Arguments'),
- 'value': [{'value': ''}, {'value': ''}],
- 'help_text': _('Options first, then source / destination paths'),
- 'type': 'distcp'
- },
- # Common
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.')
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production')
- },
- 'java_opts': {
- 'name': 'java_opts',
- 'label': _('Java options'),
- 'value': '',
- 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b')
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['distcp_parameters']]
- class SparkAction(Action):
- TYPE = 'spark'
- FIELDS = {
- 'spark_master': {
- 'name': 'spark_master',
- 'label': _('Spark Master'),
- 'value': 'yarn',
- 'help_text': _('Ex: spark://host:port, mesos://host:port, yarn, or local.'),
- 'type': ''
- },
- 'mode': {
- 'name': 'mode',
- 'label': _('Mode'),
- 'value': 'client',
- 'help_text': _('e.g. Client cluster'),
- 'type': ''
- },
- 'app_name': {
- 'name': 'app_name',
- 'label': _('App name'),
- 'value': 'MySpark',
- 'help_text': _('The name of the spark application'),
- 'type': ''
- },
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Path to file to put in the running directory.'),
- 'type': ''
- },
- 'class': {
- 'name': 'class',
- 'label': _('Main class'),
- 'value': '',
- 'help_text': _("Only if using jars, e.g. org.apache.spark.examples.mllib.JavaALS"),
- 'type': 'text'
- },
- 'jars': {
- 'name': 'jars',
- 'label': _('Jar/py name'),
- 'value': '',
- 'help_text': _('Name of main file added in Files.'),
- 'type': 'text'
- },
- 'spark_opts': {
- 'name': 'spark_opts',
- 'label': _('Options list'),
- 'value': '',
- 'help_text': _('Ex: --executor-memory 20G --num-executors 50'),
- 'type': ''
- },
- 'spark_arguments': {
- 'name': 'spark_arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('Arguments, one by one, e.g. 1000, /path/a.')
- },
- # Common
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production')
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.')
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['files'], cls.FIELDS['jars']]
- class KillAction(Action):
- TYPE = 'kill'
- FIELDS = {
- 'message': {
- 'name': 'message',
- 'label': _('Message'),
- 'value': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]'),
- 'help_text': _('Message to display when the workflow fails. Can contain some EL functions.'),
- 'type': 'textarea'
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['message']]
- class JoinAction(Action):
- TYPE = 'join'
- FIELDS = {}
- @classmethod
- def get_mandatory_fields(cls):
- return []
- class GenericAction(Action):
- TYPE = 'generic'
- FIELDS = {
- 'xml': {
- 'name': 'xml',
- 'label': _('XML of the action'),
- 'value': '<my_action>\n</my_action>',
- 'help_text': _('Insert verbatim the XML of the action to insert into the workflow.'),
- 'type': 'textarea'
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['xml']]
- class ForkNode(Action):
- TYPE = 'fork'
- FIELDS = {}
- @classmethod
- def get_mandatory_fields(cls):
- return []
- class HiveDocumentAction(Action):
- TYPE = 'hive-document'
- DEFAULT_CREDENTIALS = 'hive2'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Hive query'),
- 'value': '',
- 'help_text': _('Select a saved Hive query you want to schedule.'),
- 'type': 'hive'
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The parameters of the script. E.g. N=5, INPUT=${inputDir}'),
- 'type': ''
- },
- # Common
- 'jdbc_url': {
- 'name': 'jdbc_url',
- 'label': _('HiveServer2 URL'),
- 'value': "",
- 'help_text': _('e.g. jdbc:hive2://localhost:10000/default. JDBC URL for the Hive Server 2.'),
- 'type': ''
- },
- 'password': {
- 'name': 'password',
- 'label': _('Password'),
- 'value': '',
- '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 '
- 'something requiring a password (e.g. LDAP); non-secured Hive Server 2 or Kerberized Hive Server 2 don\'t require a password.'),
- 'type': ''
- },
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class ImpalaDocumentAction(HiveDocumentAction):
- TYPE = 'impala-document'
- DEFAULT_CREDENTIALS = '' # None at this time, need to upload user keytab
- FIELDS = HiveServer2Action.FIELDS.copy()
- del FIELDS['jdbc_url']
- del FIELDS['password']
- FIELDS['impalad_host'] = {
- 'name': 'impalad_host',
- 'label': _('Impalad hostname'),
- 'value': "",
- 'help_text': _('e.g. impalad-001.cluster.com (optional)'),
- 'type': ''
- }
- FIELDS['key_tab_path'] = {
- 'name': 'key_tab_path',
- 'label': _('Keytab path'),
- 'value': '${key_tab_path}' if SECURITY_ENABLED.get() else '',
- 'help_text': _('Path to the keytab to use when on a secure cluster, e.g. /user/joe/joe.keytab.'),
- 'type': ''
- }
- FIELDS['user_principal'] = {
- 'name': 'user_principal',
- 'label': _('User principal'),
- 'value': '${user_principal}' if SECURITY_ENABLED.get() else '',
- 'help_text': _('Name of the principal to use in the kinit, e.g.: kinit -k -t /home/joe/joe.keytab joe@PROD.EDH.'),
- 'type': ''
- }
- FIELDS['uuid'] = {
- 'name': 'uuid',
- 'label': _('Impala query'),
- 'value': '',
- 'help_text': _('Select a saved Impala query you want to schedule.'),
- 'type': 'impala'
- }
- class JavaDocumentAction(Action):
- TYPE = 'java-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Java program'),
- 'value': '',
- 'help_text': _('Select a saved Java program you want to schedule.'),
- 'type': 'java'
- },
- 'arguments': {
- 'name': 'arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('Arguments of the main method. The value of each arg element is considered a single argument '
- 'and they are passed to the main method in the same order.'),
- 'type': ''
- },
- 'java_opts': {
- 'name': 'java_opts',
- 'label': _('Java options'),
- 'value': [],
- 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b'),
- 'type': ''
- },
- 'capture_output': {
- 'name': 'capture_output',
- 'label': _('Capture output'),
- 'value': False,
- 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
- 'command output must be in Java Properties file format and it must not exceed 2KB. '
- 'From within the workflow definition, the output of an %(program)s action node is accessible '
- 'via the String action:output(String node, String key) function') % {'program': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': [],
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class SparkDocumentAction(Action):
- TYPE = 'spark2-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Spark program'),
- 'value': '',
- 'help_text': _('Select a saved Spark program you want to schedule.'),
- 'type': 'spark'
- },
- 'spark_master': {
- 'name': 'spark_master',
- 'label': _('Spark Master'),
- 'value': 'yarn',
- 'help_text': _('Ex: spark://host:port, mesos://host:port, yarn, or local.'),
- 'type': ''
- },
- 'mode': {
- 'name': 'mode',
- 'label': _('Mode'),
- 'value': 'client',
- 'help_text': _('e.g. Client cluster'),
- 'type': ''
- },
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Path to file to put in the running directory.'),
- 'type': ''
- },
- 'spark_arguments': {
- 'name': 'spark_arguments',
- 'label': _('Arguments'),
- 'value': [],
- 'help_text': _('Arguments, one by one, e.g. 1000, /path/a.')
- },
- 'parameters': { # For Oozie Action Document
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production')
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.')
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class PigDocumentAction(Action):
- TYPE = 'pig-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Pig script'),
- 'value': '',
- 'help_text': _('Select a saved Spark program you want to schedule.'),
- 'type': 'pig'
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': [],
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class SqoopDocumentAction(Action):
- TYPE = 'sqoop-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Sqoop command'),
- 'value': '',
- 'help_text': _('Select a saved Sqoop program you want to schedule.'),
- 'type': 'sqoop'
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class DistCpDocumentAction(Action):
- TYPE = 'distcp-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('DistCp program'),
- 'value': '',
- 'help_text': _('Select a saved DistCp program you want to schedule.'),
- 'type': 'distcp-doc'
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.')
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production')
- },
- 'java_opts': {
- 'name': 'java_opts',
- 'label': _('Java options'),
- 'value': '',
- 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b')
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class ShellDocumentAction(Action):
- TYPE = 'shell-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('Shell program'),
- 'value': '',
- 'help_text': _('Select a saved Shell program you want to schedule.'),
- 'type': 'shell-doc'
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class MapReduceDocumentAction(Action):
- TYPE = 'mapreduce-document'
- FIELDS = {
- 'uuid': {
- 'name': 'uuid',
- 'label': _('MapReduce program'),
- 'value': '',
- 'help_text': _('Select a saved MapReduce program you want to schedule.'),
- 'type': 'mapreduce-doc'
- },
- 'parameters': {
- 'name': 'parameters',
- 'label': _('Parameters'),
- 'value': [],
- 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
- 'type': ''
- },
- # Common
- 'files': {
- 'name': 'files',
- 'label': _('Files'),
- 'value': [],
- 'help_text': _('Files put in the running directory.'),
- 'type': ''
- },
- 'archives': {
- 'name': 'archives',
- 'label': _('Archives'),
- 'value': [],
- 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
- 'type': ''
- },
- 'job_properties': {
- 'name': 'job_properties',
- 'label': _('Hadoop job properties'),
- 'value': [],
- 'help_text': _('value, e.g. production'),
- 'type': ''
- },
- 'prepares': {
- 'name': 'prepares',
- 'label': _('Prepares'),
- 'value': [],
- 'help_text': _('Path to manipulate before starting the application.'),
- 'type': ''
- },
- 'job_xml': {
- 'name': 'job_xml',
- 'label': _('Job XML'),
- 'value': '',
- 'help_text': _('Refer to a Hadoop JobConf job.xml'),
- 'type': ''
- },
- 'retry_max': {
- 'name': 'retry_max',
- 'label': _('Max retry'),
- 'value': [],
- 'help_text': _('Number of times, default is 3'),
- 'type': ''
- },
- 'retry_interval': {
- 'name': 'retry_interval',
- 'label': _('Retry interval'),
- 'value': [],
- 'help_text': _('Wait time in minutes, default is 10'),
- 'type': ''
- }
- }
- @classmethod
- def get_mandatory_fields(cls):
- return [cls.FIELDS['uuid']]
- class DecisionNode(Action):
- TYPE = 'decision'
- FIELDS = {}
- @classmethod
- def get_mandatory_fields(cls):
- return []
- NODES = {
- 'start-widget': StartNode,
- 'end-widget': EndNode,
- 'pig-widget': PigAction,
- 'java-widget': JavaAction,
- 'hive-widget': HiveAction,
- 'hive2-widget': HiveServer2Action,
- 'impala-widget': ImpalaAction,
- 'sqoop-widget': SqoopAction,
- 'mapreduce-widget': MapReduceAction,
- 'subworkflow-widget': SubWorkflowAction,
- 'shell-widget': ShellAction,
- 'ssh-widget': SshAction,
- 'fs-widget': FsAction,
- 'email-widget': EmailAction,
- 'streaming-widget': StreamingAction,
- 'distcp-widget': DistCpAction,
- 'kill-widget': KillAction,
- 'join-widget': JoinAction,
- 'fork-widget': ForkNode,
- 'decision-widget': DecisionNode,
- 'spark-widget': SparkAction,
- 'generic-widget': GenericAction,
- 'hive-document-widget': HiveDocumentAction,
- 'impala-document-widget': ImpalaDocumentAction,
- 'java-document-widget': JavaDocumentAction,
- 'spark-document-widget': SparkDocumentAction,
- 'pig-document-widget': PigDocumentAction,
- 'sqoop-document-widget': SqoopDocumentAction,
- 'distcp-document-widget': DistCpDocumentAction,
- 'shell-document-widget': ShellDocumentAction,
- 'mapreduce-document-widget': MapReduceDocumentAction
- }
- WORKFLOW_NODE_PROPERTIES = {}
- for node in NODES.itervalues():
- WORKFLOW_NODE_PROPERTIES.update(node.FIELDS)
- def find_parameters(instance, fields=None):
- """Find parameters in the given fields"""
- if fields is None:
- fields = NODES['%s-widget' % instance.data['type']].FIELDS.keys()
- params = []
- for field in fields:
- data = instance.data['properties'][field]
- if field == 'sla' and not instance.sla_enabled:
- continue
- if isinstance(data, list):
- params.extend(find_json_parameters(data))
- elif isinstance(data, basestring):
- for match in Template.pattern.finditer(data):
- name = match.group('braced')
- if name is not None:
- params.append(name)
- return params
- def find_json_parameters(fields):
- # Input is list of json dict
- params = []
- for field in fields:
- for data in field.values():
- if isinstance(data, basestring):
- for match in Template.pattern.finditer(data):
- name = match.group('braced')
- if name is not None:
- params.append(name)
- return params
- def find_dollar_variables(text):
- return re.findall('[^\n\\\\]\$([^\{ \'\"\-;\(\)]+)', text, re.MULTILINE)
- def find_dollar_braced_variables(text):
- vars = set()
- for var in re.findall('\$\{([A-Za-z0-9:_-]+)\}', text, re.MULTILINE):
- if ':' in var:
- var = var.split(':', 1)[1]
- vars.add(var)
- return list(vars)
- def import_workflow_from_hue_3_7(old_wf):
- """
- Example of data to transform
- [<Start: start>, <Pig: Pig>, [<Kill: kill>], [<End: end>]]
- [<Start: start>, <Java: TeraGenWorkflow>, <Java: TeraSort>, [<Kill: kill>], [<End: end>]]
- [<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>]]
- """
- uuids = {}
- old_nodes = old_wf.get_hierarchy()
- wf = Workflow()
- wf_rows = []
- wf_nodes = []
- data = wf.get_data()
- # UUIDs node mapping
- for node in old_wf.node_list:
- if node.name == 'kill':
- node_uuid = '17c9c895-5a16-7443-bb81-f34b30b21548'
- elif node.name == 'start':
- node_uuid = '3f107997-04cc-8733-60a9-a4bb62cebffc'
- elif node.name == 'end':
- node_uuid = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
- else:
- node_uuid = str(uuid.uuid4())
- uuids[node.id] = node_uuid
- # Workflow
- data['workflow']['uuid'] = str(uuid.uuid4())
- data['workflow']['name'] = old_wf.name
- data['workflow']['properties']['properties'] = json.loads(old_wf.job_properties)
- data['workflow']['properties']['job_xml'] = old_wf.job_xml
- data['workflow']['properties']['description'] = old_wf.description
- data['workflow']['properties']['schema_version'] = old_wf.schema_version
- data['workflow']['properties']['deployment_dir'] = old_wf.deployment_dir
- data['workflow']['properties']['parameters'] = json.loads(old_wf.parameters)
- data['workflow']['properties']['description'] = old_wf.description
- data['workflow']['properties']['sla'] = old_wf.sla
- data['workflow']['properties']['sla_enabled'] = old_wf.sla_enabled
- data['workflow']['properties']['imported'] = True
- data['workflow']['properties']['wf1_id'] = old_wf.id
- # Layout
- rows = data['layout'][0]['rows']
- def _create_layout(nodes, size=12):
- wf_rows = []
- for node in nodes:
- if type(node) == list and len(node) == 1:
- node = node[0]
- if type(node) != list:
- 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":[]}]})
- else:
- if node[0].node_type == 'fork':
- 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":[]}]})
- wf_rows.append({
- "id": str(uuid.uuid4()),
- "widgets":[
- ],
- "columns":[
- {
- "id": str(uuid.uuid4()),
- "size": (size / len(node[1])),
- "rows":
- [{
- "id": str(uuid.uuid4()),
- "widgets": c['widgets'],
- "columns":[]
- }
- for c in col] if type(col) == list else [{
- "id": str(uuid.uuid4()),
- "widgets": col['widgets'],
- "columns":[]
- }
- ]
- ,
- "klass":"card card-home card-column span%s" % (size / len(node[1]))
- }
- for col in _create_layout(node[1], size)
- ]
- })
- 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":[]}]})
- else:
- wf_rows.append(_create_layout(node, size))
- return wf_rows
- wf_rows = _create_layout(old_nodes)
- if wf_rows:
- data['layout'][0]['rows'] = [data['layout'][0]['rows'][0]] + wf_rows + [data['layout'][0]['rows'][-1]]
- # Content
- def _dig_nodes(nodes):
- for node in nodes:
- if type(node) != list:
- properties = {}
- if '%s-widget' % node.node_type in NODES:
- properties = dict(NODES['%s-widget' % node.node_type].get_fields())
- if node.node_type == 'pig':
- properties['script_path'] = node.script_path
- properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'hive':
- properties['script_path'] = node.script_path
- properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['hive_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'java':
- properties['jar_path'] = node.jar_path
- properties['main_class'] = node.main_class
- properties['arguments'] = [{'value': arg} for arg in node.args.split(' ')]
- properties['java_opts'] = node.java_opts
- properties['capture_output'] = node.capture_output
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'sqoop':
- properties['command'] = node.script_path
- properties['parameters'] = json.loads(node.params)
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'mapreduce':
- properties['jar_path'] = node.jar_path
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'shell':
- properties['shell_command'] = node.command
- properties['arguments'] = json.loads(node.params)
- properties['capture_output'] = node.capture_output
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'ssh':
- properties['user'] = '%s@%s' % (node.user, node.host)
- properties['ssh_command'] = node.command
- properties['params'] = json.loads(node.params)
- properties['capture_output'] = node.capture_output
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'fs':
- properties['deletes'] = [{'value': f['name']} for f in json.loads(node.deletes)]
- properties['mkdirs'] = [{'value': f['name']} for f in json.loads(node.mkdirs)]
- properties['moves'] = json.loads(node.moves)
- chmods = json.loads(node.chmods)
- for c in chmods:
- c['value'] = c['path']
- c['dir_files'] = False
- properties['chmods'] = chmods
- properties['touchzs'] = [{'value': f['name']} for f in json.loads(node.touchzs)]
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'email':
- properties['to'] = node.to
- properties['cc'] = node.cc
- properties['subject'] = node.subject
- properties['body'] = node.body
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'streaming':
- properties['mapper'] = node.mapper
- properties['reducer'] = node.reducer
- properties['files'] = [{'value': f} for f in json.loads(node.files)]
- properties['archives'] = json.loads(node.archives)
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['job_xml'] = node.job_xml
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- elif node.node_type == 'distcp':
- properties['distcp_parameters'] = json.loads(node.params)
- properties['java_opts'] = node.job_xml
- properties['job_properties'] = json.loads(node.job_properties)
- properties['prepares'] = json.loads(node.prepares)
- properties['description'] = node.description
- properties['sla'] = node.sla
- properties['sla_enabled'] = node.sla_enabled
- wf_nodes.append({
- "id": uuids[node.id],
- "name": '%s-%s' % (node.node_type.split('-')[0], uuids[node.id][:4]),
- "type": "%s-widget" % node.node_type,
- "properties": properties,
- "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()]
- })
- else:
- _dig_nodes(node)
- _dig_nodes(old_nodes)
- data['workflow']['nodes'] = wf_nodes
- return Workflow(data=json.dumps(data))
- class Coordinator(Job):
- XML_FILE_NAME = 'coordinator.xml'
- PROPERTY_APP_PATH = 'oozie.coord.application.path'
- HUE_ID = 'hue-id-c'
- def __init__(self, data=None, json_data=None, document=None):
- self.document = document
- if document is not None:
- self._data = json.loads(document.data)
- elif json_data is not None:
- self._data = json.loads(json_data)
- elif data is not None:
- self._data = data
- else:
- self._data = {
- 'id': None,
- 'uuid': None,
- 'name': 'My Coordinator',
- 'variables': [], # Aka workflow parameters
- 'properties': {
- 'description': '',
- 'deployment_dir': '',
- 'schema_version': 'uri:oozie:coordinator:0.2',
- 'frequency_number': 1,
- 'frequency_unit': 'days',
- 'cron_frequency': '0 0 * * *',
- 'cron_advanced': False,
- 'timezone': '',
- 'start': '${start_date}',
- 'end': '${end_date}',
- 'workflow': None,
- 'document': None,
- 'timeout': None,
- 'concurrency': None,
- 'execution': None,
- 'throttle': None,
- 'job_xml': '',
- 'credentials': [],
- 'parameters': [
- {'name': 'oozie.use.system.libpath', 'value': True},
- {'name': 'start_date', 'value': datetime.today().strftime('%Y-%m-%dT%H:%M')},
- {'name': 'end_date', 'value': (datetime.today() + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M')}
- ],
- 'sla': WorkflowConfiguration.SLA_DEFAULT
- }
- }
- @property
- def id(self):
- return self.document.id
- @property
- def uuid(self):
- return self.document.uuid
- def get_data_for_json(self):
- _data = self.data.copy()
- start_date = filter(lambda a: a['name'] == 'start_date', self._data['properties']['parameters'])
- if start_date and type(start_date[0]['value']) == datetime:
- start_date[0]['value'] = start_date[0]['value'].strftime('%Y-%m-%dT%H:%M:%S')
- end_date = filter(lambda a: a['name'] == 'end_date', self._data['properties']['parameters'])
- if end_date and type(end_date[0]['value']) == datetime:
- end_date[0]['value'] = end_date[0]['value'].strftime('%Y-%m-%dT%H:%M:%S')
- return _data
- def to_json(self):
- return json.dumps(self.get_data_for_json())
- def to_json_for_html(self):
- return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
- @property
- def data(self):
- if type(self._data['properties']['start']) != datetime and not '$' in self._data['properties']['start']:
- self._data['properties']['start'] = parse(self._data['properties']['start'])
- if type(self._data['properties']['end']) != datetime and not '$' in self._data['properties']['end']:
- self._data['properties']['end'] = parse(self._data['properties']['end'])
- if self.document is not None:
- self._data['id'] = self.document.id
- if 'document' not in self._data['properties']:
- self._data['properties']['document'] = None
- return self._data
- @property
- def name(self):
- from notebook.connectors.oozie_batch import OozieApi # Import dependency
- if self.data['properties']['document']:
- return _("%s for %s") % (OozieApi.SCHEDULE_JOB_PREFIX, self.data['name'] or self.data['type'])
- else:
- return self.data['name']
- def set_workspace(self, user):
- self.data['properties']['deployment_dir'] = Job.get_workspace(user)
- @property
- def deployment_dir(self):
- return self.data['properties']['deployment_dir']
- def find_parameters(self):
- params = set()
- for param in find_dollar_braced_variables(self.name):
- params.add(param)
- for param in find_json_parameters([self.data['properties']]):
- params.add(param)
- for param in find_json_parameters(self.data['variables']):
- if param not in ('MINUTE', 'HOUR', 'DAY', 'MONTH', 'YEAR') and not param.startswith('coord:'):
- params.add(param)
- if self.sla_enabled:
- for param in find_json_parameters(self.sla):
- params.add(param)
- # Get missed params from workflow
- for prop in self.workflow.find_parameters():
- if not prop in params:
- params.add(prop)
- # Remove the ones filled up by coordinator
- removable_names = [ds['workflow_variable'] for ds in self.data['variables']]
- return dict([(param, '') for param in list(params) if param not in removable_names])
- @property
- def sla_enabled(self):
- return self.data['properties']['sla'][0].get('value')
- @property
- def sla(self):
- return self.data['properties']['sla']
- @property
- def parameters(self):
- return self.data['properties']['parameters']
- @property
- def datasets(self):
- return self.inputDatasets + self.outputDatasets
- @property
- def inputDatasets(self):
- return [Dataset(dataset, self) for dataset in self.data['variables'] if dataset['dataset_type'] == 'input_path']
- @property
- def outputDatasets(self):
- return [Dataset(dataset, self) for dataset in self.data['variables'] if dataset['dataset_type'] == 'output_path']
- @property
- def start_server_tz(self):
- return self.data['properties']['start']
- @property
- def end_server_tz(self):
- return self.data['properties']['end']
- @property
- def frequency(self):
- return '${coord:%(unit)s(%(number)d)}' % {'unit': self.data['properties']['frequency_unit'], 'number': self.data['properties']['frequency_number']}
- @property
- def cron_frequency(self):
- data_dict = self.data['properties']
- if 'cron_frequency' in data_dict:
- return data_dict['cron_frequency']
- else:
- # Backward compatibility
- freq = '0 0 * * *'
- if data_dict['frequency_number'] == 1:
- if data_dict['frequency_unit'] == 'minutes':
- freq = '* * * * *'
- elif data_dict['frequency_unit'] == 'hours':
- freq = '0 * * * *'
- elif data_dict['frequency_unit'] == 'days':
- freq = '0 0 * * *'
- elif data_dict['frequency_unit'] == 'months':
- freq = '0 0 0 * *'
- return {'frequency': freq, 'isAdvancedCron': False}
- def to_xml(self, mapping=None):
- if mapping is None:
- mapping = {}
- tmpl = "editor2/gen/coordinator.xml.mako"
- return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
- def clear_workflow_params(self):
- # Repopulated in the config properties
- self.data['variables'] = [dataset for dataset in self.data['variables'] if dataset['dataset_type'] != 'parameter']
- @property
- def properties(self):
- props = [{'name': dataset['workflow_variable'], 'value': dataset['dataset_variable']} for dataset in self.data['variables'] if dataset['dataset_type'] == 'parameter']
- props += self.data['properties']['parameters']
- return props
- @property
- def workflow(self):
- if self.document is None:
- raise PopupException(_('Cannot return workflow since document attribute is None.'))
- # Integrated scheduler
- if self.data['properties']['document']:
- document = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['document'])
- wf_doc = WorkflowBuilder().create_workflow(document=document, user=self.document.owner, managed=True)
- wf = Workflow(data=wf_doc.data)
- wf_doc.delete()
- return wf
- else:
- wf_doc = Document2.objects.get_by_uuid(user=self.document.owner, uuid=self.data['properties']['workflow'])
- return Workflow(document=wf_doc)
- def get_absolute_url(self):
- return reverse('oozie:edit_coordinator') + '?coordinator=%s' % self.id
- @classmethod
- def get_application_path_key(cls):
- return 'oozie.coord.application.path'
- class Dataset():
- def __init__(self, data, coordinator):
- self._data = data
- self.coordinator = coordinator
- @property
- def data(self):
- self._data['name'] = self._data['workflow_variable']
- return self._data
- @property
- def frequency(self):
- if self.data['same_frequency']:
- if self.coordinator.cron_frequency == '* * * * *':
- frequency_unit = 'minutes'
- elif self.coordinator.cron_frequency == '0 * * * *':
- frequency_unit = 'hours'
- elif self.coordinator.cron_frequency == '0 0 * * *':
- frequency_unit = 'days'
- elif self.coordinator.cron_frequency == '0 0 0 * *':
- frequency_unit = 'months'
- else:
- raise PopupException(_('The frequency of the workflow parameter "%s" cannot be guessed from the frequency of the coordinator.'
- ' It so needs to be specified manually.') % self.data['name'])
- frequency_number = 1
- else:
- frequency_unit = self.data['frequency_unit']
- frequency_number = self.data['frequency_number']
- return '${coord:%(unit)s(%(number)s)}' % {'unit': frequency_unit, 'number': frequency_number}
- @property
- def start_server_tz(self):
- if self.data['same_start']:
- return self.coordinator.start_server_tz
- else:
- return convert_to_server_timezone(self.data['start'], self.data['timezone'])
- @property
- def timezone(self):
- if self.data['same_timezone']:
- return self.coordinator.data['properties']['timezone']
- else:
- return self.data['timezone']
- @property
- def start_instance(self):
- if not self.is_advanced_start_instance:
- return int(self.data['advanced_start_instance'])
- else:
- return 0
- @property
- def is_advanced_start_instance(self):
- return not self.is_int(self.data['advanced_start_instance'])
- def is_int(self, text):
- try:
- int(text)
- return True
- except ValueError:
- return False
- @property
- def end_instance(self):
- if not self.is_advanced_end_instance:
- return int(self.data['advanced_end_instance'])
- else:
- return 0
- @property
- def is_advanced_end_instance(self):
- return not self.is_int(self.data['advanced_end_instance'])
- class Bundle(Job):
- XML_FILE_NAME = 'bundle.xml'
- PROPERTY_APP_PATH = 'oozie.bundle.application.path'
- HUE_ID = 'hue-id-b'
- def __init__(self, data=None, json_data=None, document=None):
- self.document = document
- if document is not None:
- self._data = json.loads(document.data)
- elif json_data is not None:
- self._data = json.loads(json_data)
- elif data is not None:
- self._data = data
- else:
- self._data = {
- 'id': None,
- 'uuid': None,
- 'name': 'My Bundle',
- 'coordinators': [],
- 'properties': {
- 'description': '',
- 'deployment_dir': '',
- 'schema_version': 'uri:oozie:bundle:0.2',
- 'kickoff': datetime.today(),
- 'parameters': [{'name': 'oozie.use.system.libpath', 'value': 'true'}]
- }
- }
- @property
- def id(self):
- return self.document.id
- @property
- def uuid(self):
- return self.document.uuid
- def get_data_for_json(self):
- _data = self.data.copy()
- _data['properties']['kickoff'] = _data['properties']['kickoff'].strftime('%Y-%m-%dT%H:%M:%S')
- return _data
- def to_json(self):
- return json.dumps(self.get_data_for_json())
- def to_json_for_html(self):
- return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
- @property
- def data(self):
- if type(self._data['properties']['kickoff']) == unicode:
- self._data['properties']['kickoff'] = parse(self._data['properties']['kickoff'])
- if self.document is not None:
- self._data['id'] = self.document.id
- return self._data
- def to_xml(self, mapping=None):
- if mapping is None:
- mapping = {}
- mapping.update(dict(list(self.get_coordinator_docs().values('uuid', 'name'))))
- tmpl = "editor2/gen/bundle.xml.mako"
- return force_unicode(
- re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
- 'bundle': self,
- 'mapping': mapping
- })))
- def get_coordinator_docs(self):
- coordinator_ids = [coordinator['coordinator'] for coordinator in self.data['coordinators']]
- return Document2.objects.filter(type='oozie-coordinator2', uuid__in=coordinator_ids)
- def get_coordinator_objects(self):
- return [Coordinator(document=doc) for doc in self.get_coordinator_docs()]
- @property
- def name(self):
- return self.data['name']
- @property
- def parameters(self):
- return self.data['properties']['parameters']
- @property
- def kick_off_time_utc(self):
- return utc_datetime_format(self.data['properties']['kickoff'])
- def set_workspace(self, user):
- self.data['properties']['deployment_dir'] = Job.get_workspace(user)
- @property
- def deployment_dir(self):
- return self.data['properties']['deployment_dir']
- def find_parameters(self):
- params = set()
- for param in find_dollar_braced_variables(self.name):
- params.add(param)
- for coord in self.get_coordinator_objects():
- params.update(coord.find_parameters())
- for param in find_json_parameters([self.data['properties']]):
- params.add(param)
- # Remove the ones filled up by bundle
- removable_names = [p['name'] for coord in self.data['coordinators'] for p in coord['properties']]
- return dict([(param, '') for param in list(params) if param not in removable_names])
- def get_absolute_url(self):
- return reverse('oozie:edit_bundle') + '?bundle=%s' % self.id
- @classmethod
- def get_application_path_key(cls):
- return 'oozie.bundle.application.path'
- class History(object):
- @classmethod
- def get_workflow_from_config(self, conf_dict):
- try:
- doc = Document2.objects.get(type='oozie-workflow2', id=conf_dict.get(Workflow.HUE_ID))
- return Workflow(document=doc)
- except Document2.DoesNotExist:
- pass
- @classmethod
- def get_coordinator_from_config(self, conf_dict):
- try:
- doc = Document2.objects.get(type='oozie-coordinator2', id=conf_dict.get(Coordinator.HUE_ID))
- return Coordinator(document=doc)
- except Document2.DoesNotExist:
- pass
- @classmethod
- def get_bundle_from_config(self, conf_dict):
- try:
- doc = Document2.objects.get(type='oozie-bundle2', id=conf_dict.get(Bundle.HUE_ID))
- return Bundle(document=doc)
- except Document2.DoesNotExist:
- pass
- def _import_workspace(fs, user, job):
- source_workspace_dir = job.deployment_dir
- job.set_workspace(user)
- job.check_workspace(fs, user)
- job.import_workspace(fs, source_workspace_dir, user)
- def _save_workflow(workflow, layout, user, fs=None):
- if workflow.get('id'):
- workflow_doc = Document2.objects.get(id=workflow['id'])
- else:
- workflow_doc = Document2.objects.create(name=workflow['name'], uuid=workflow['uuid'], type='oozie-workflow2', owner=user, description=workflow['properties']['description'])
- Document.objects.link(workflow_doc, owner=workflow_doc.owner, name=workflow_doc.name, description=workflow_doc.description, extra='workflow2')
- # Excludes all the sub-workflow and Hive dependencies. Contains list of history and coordinator dependencies.
- workflow_doc.dependencies = workflow_doc.dependencies.exclude(Q(is_history=False) & Q(type__in=['oozie-workflow2', 'query-hive', 'query-java']))
- dependencies = \
- [node['properties']['workflow'] for node in workflow['nodes'] if node['type'] == 'subworkflow-widget'] + \
- [node['properties']['uuid'] for node in workflow['nodes'] if 'document-widget' in node['type'] and node['properties'].get('uuid')]
- if dependencies:
- dependency_docs = Document2.objects.filter(uuid__in=dependencies)
- workflow_doc.dependencies.add(*dependency_docs)
- if workflow['properties'].get('imported'): # We convert from and old workflow format (3.8 <) to the latest
- workflow['properties']['imported'] = False
- workflow_instance = Workflow(workflow=workflow, user=user)
- _import_workspace(fs, user, workflow_instance)
- workflow['properties']['deployment_dir'] = workflow_instance.deployment_dir
- workflow_doc.update_data({'workflow': workflow})
- workflow_doc.update_data({'layout': layout})
- workflow_doc1 = workflow_doc.doc.get()
- workflow_doc.name = workflow_doc1.name = workflow['name']
- workflow_doc.description = workflow_doc1.description = workflow['properties']['description']
- workflow_doc.save()
- workflow_doc1.save()
- return workflow_doc
- class WorkflowBuilder():
- """
- Building a workflow that has saved Documents for nodes (e.g Saved Hive query, saved Pig script...).
- """
- def create_workflow(self, user, document=None, name=None, managed=False):
- nodes = []
- documents = [document]
- if name is None:
- name = _('Schedule of ') + ','.join([document.name or document.type for document in documents])
- for document in documents:
- if document.type == 'query-java':
- node = self.get_java_document_node(document)
- elif document.type == 'query-hive':
- node = self.get_hive_document_node(document, user)
- elif document.type == 'query-impala':
- node = self.get_impala_document_node(document, user)
- elif document.type == 'query-spark2':
- node = self.get_spark_document_node(document, user)
- elif document.type == 'query-pig':
- node = self.get_pig_document_node(document, user)
- elif document.type == 'query-sqoop1':
- node = self.get_sqoop_document_node(document, user)
- elif document.type == 'query-distcp':
- node = self.get_distcp_document_node(document, user)
- elif document.type == 'query-shell':
- node = self.get_shell_document_node(document, user)
- elif document.type == 'query-mapreduce':
- node = self.get_mapreduce_document_node(document, user)
- else:
- raise PopupException(_('Snippet type %s is not supported in batch execution.') % document.type)
- nodes.append(node)
- workflow_doc = self.get_workflow(nodes, name, document.uuid, user, managed=managed)
- workflow_doc.dependencies.add(*documents)
- return workflow_doc
- def create_notebook_workflow(self, user, notebook=None, name=None, managed=False):
- nodes = []
- if name is None:
- name = _('Schedule of ') + ','.join([snippet['name'] or snippet['type'] for snippet in notebook['snippets']])
- for snippet in notebook['snippets']:
- if snippet['type'] == 'java':
- node = self.get_java_snippet_node(snippet)
- elif snippet['type'] == 'hive':
- node = self.get_hive_snippet_node(snippet, user)
- elif snippet['type'] == 'impala':
- node = self.get_impala_snippet_node(snippet, user)
- elif snippet['type'] == 'shell':
- node = self.get_shell_snippet_node(snippet)
- else:
- raise PopupException(_('Snippet type %s is not supported in batch execution.') % snippet)
- nodes.append(node)
- workflow_doc = self.get_workflow(nodes, name, notebook['uuid'], user, managed=managed) # TODO optionally save
- return workflow_doc
- def get_document_parameters(self, document):
- notebook = Notebook(document=document)
- parameters = find_dollar_braced_variables(notebook.get_str())
- return [{u'value': u'%s=${%s}' % (p, p)} for p in parameters]
- def _get_hive_node(self, node_id, user, is_document_node=False):
- api = get_oozie(user)
- credentials = [HiveDocumentAction.DEFAULT_CREDENTIALS] if api.security_enabled else []
- return {
- u'id': node_id,
- u'name': u'hive-%s' % node_id[:4],
- u"type": u"hive-document-widget", # if is_document_node else u"hive2-widget",
- u'properties': {
- u'files': [],
- u'job_xml': u'',
- u'retry_interval': [],
- u'retry_max': [],
- u'job_properties': [],
- u'arguments': [],
- u'parameters': [],
- u'sla': [
- {u'key': u'enabled', u'value': False},
- {u'key': u'nominal-time', u'value': u'${nominal_time}'},
- {u'key': u'should-start', u'value': u''},
- {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
- {u'key': u'max-duration', u'value': u''},
- {u'key': u'alert-events', u'value': u''},
- {u'key': u'alert-contact', u'value': u''},
- {u'key': u'notification-msg', u'value': u''},
- {u'key': u'upstream-apps', u'value': u''},
- ],
- u'archives': [],
- u'prepares': [],
- u'credentials': credentials,
- u'password': u'',
- u'jdbc_url': u'',
- },
- u'children': [
- {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
- {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
- ],
- u'actionParameters': [],
- }
- def get_hive_snippet_node(self, snippet, user):
- node = self._get_hive_node(snippet['id'], user)
- node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in snippet['variables']]
- node['properties']['statements'] = 'USE %s;\n\n%s' % (snippet['database'], snippet['statement_raw'])
- return node
- def get_hive_document_node(self, document, user):
- node = self._get_hive_node(document.uuid, user, is_document_node=True)
- notebook = Notebook(document=document)
- node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in notebook.get_data()['snippets'][0]['variables']]
- node['properties']['uuid'] = document.uuid
- return node
- def _get_impala_node(self, node_id, user, is_document_node=False):
- credentials = []
- return {
- u'id': node_id,
- u'name': u'impala-%s' % node_id[:4],
- u"type": u"impala-document-widget",
- u'properties': {
- u'files': [],
- u'job_xml': u'',
- u'retry_interval': [],
- u'retry_max': [],
- u'job_properties': [],
- u'arguments': [],
- u'parameters': [],
- u'sla': [
- {u'key': u'enabled', u'value': False},
- {u'key': u'nominal-time', u'value': u'${nominal_time}'},
- {u'key': u'should-start', u'value': u''},
- {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
- {u'key': u'max-duration', u'value': u''},
- {u'key': u'alert-events', u'value': u''},
- {u'key': u'alert-contact', u'value': u''},
- {u'key': u'notification-msg', u'value': u''},
- {u'key': u'upstream-apps', u'value': u''},
- ],
- u'archives': [],
- u'prepares': [],
- u'credentials': credentials,
- u'impalad_host': u'',
- u'key_tab_path': u'',
- u'user_principal': u''
- },
- u'children': [
- {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
- {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
- ],
- u'actionParameters': [],
- }
- def get_impala_snippet_node(self, snippet, user):
- node = self._get_impala_node(snippet['id'], user)
- node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in snippet['variables']]
- node['properties']['statements'] = 'USE %s;\n\n%s' % (snippet['database'], snippet['statement_raw'])
- return node
- def get_impala_document_node(self, document, user):
- node = self._get_impala_node(document.uuid, user, is_document_node=True)
- notebook = Notebook(document=document)
- node['properties']['parameters'] = [{'value': '%(name)s=%(value)s' % v} for v in notebook.get_data()['snippets'][0]['variables']]
- node['properties']['uuid'] = document.uuid
- return node
- def _get_spark_node(self, node_id, user, is_document_node=False):
- credentials = []
- return {
- u'id': node_id,
- u'name': u'spark2-%s' % node_id[:4],
- u"type": u"spark2-document-widget", # if is_document_node else u"hive2-widget",
- u'properties': {
- u'files': [],
- u'job_xml': u'',
- u'retry_interval': [],
- u'retry_max': [],
- u'job_properties': [],
- u'sla': [
- {u'key': u'enabled', u'value': False},
- {u'key': u'nominal-time', u'value': u'${nominal_time}'},
- {u'key': u'should-start', u'value': u''},
- {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
- {u'key': u'max-duration', u'value': u''},
- {u'key': u'alert-events', u'value': u''},
- {u'key': u'alert-contact', u'value': u''},
- {u'key': u'notification-msg', u'value': u''},
- {u'key': u'upstream-apps', u'value': u''},
- ],
- u'archives': [],
- u'prepares': [],
- u'credentials': credentials,
- u'spark_master': u'yarn',
- u'mode': u'client',
- u'app_name': u'BatchSpark2'
- },
- u'children': [
- {u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'},
- {u'error': u'17c9c895-5a16-7443-bb81-f34b30b21548'}
- ],
- u'actionParameters': [],
- }
- def get_spark_snippet_node(self, snippet):
- credentials = []
- node_id = snippet.get('id', str(uuid.uuid4()))
- node = self._get_java_node(node_id, credentials)
- node['properties']['class'] = snippet['properties']['class']
- node['properties']['jars'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
- node['properties']['spark_opts'] = [{'value': f['path']} for f in snippet['properties']['files']]
- node['properties']['spark_arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
- return node
- def get_spark_document_node(self, document, user):
- node = self._get_spark_node(document.uuid, user, is_document_node=True)
- node['properties']['uuid'] = document.uuid
- return node
- def get_sqoop_document_node(self, document, user):
- node = self._get_sqoop_node(document.uuid, is_document_node=True)
- node['properties']['uuid'] = document.uuid
- return node
- def _get_sqoop_node(self, node_id, credentials=None, is_document_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'sqoop-%s' % node_id[:4],
- "type": "sqoop-document-widget",
- "properties":{
- "command": "",
- "arguments": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": [],
- "files": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def get_distcp_document_node(self, document, user):
- node = self._get_distcp_node(document.uuid, is_document_node=True)
- node['properties']['uuid'] = document.uuid
- return node
- def _get_distcp_node(self, node_id, credentials=None, is_document_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'distcp-%s' % node_id[:4],
- "type": "distcp-document-widget",
- "properties":{
- "source_path": "",
- "destination_path": "",
- "arguments": [],
- "java_opts": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def get_shell_document_node(self, document, user):
- node = self._get_shell_node(document.uuid)
- node['properties']['uuid'] = document.uuid
- return node
- def get_shell_snippet_node(self, snippet):
- node = self._get_shell_node(snippet['id'])
- node['properties']['shell_command'] = snippet['properties'].get('shell_command')
- node['properties']['arguments'] = snippet['properties'].get('arguments')
- node['properties']['archives'] = snippet['properties'].get('archives')
- node['properties']['files'] = snippet['properties'].get('files')
- node['properties']['env_var'] = snippet['properties'].get('env_var')
- return node
- def _get_shell_node(self, node_id, credentials=None, is_snippet_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'shell-%s' % node_id[:4],
- "type": "shell-document-widget",
- "properties":{
- "command_path": "",
- "env_var": [],
- "arguments": [],
- "java_opts": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "capture_output": True,
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def get_mapreduce_document_node(self, document, user):
- node = self._get_mapreduce_node(document.uuid, is_document_node=True)
- node['properties']['uuid'] = document.uuid
- return node
- def _get_mapreduce_node(self, node_id, credentials=None, is_document_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'mapreduce-%s' % node_id[:4],
- "type": "mapreduce-document-widget",
- "properties":{
- "jar_path": "",
- "arguments": [],
- "java_opts": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def get_pig_document_node(self, document, user):
- node = self._get_pig_node(document.uuid, is_document_node=True)
- node['properties']['uuid'] = document.uuid
- return node
- def _get_pig_node(self, node_id, credentials=None, is_document_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'pig-%s' % node_id[:4],
- "type": "pig-document-widget",
- "properties":{
- "job_xml": [],
- "jar_path": "",
- "java_opts": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def _get_java_node(self, node_id, credentials=None, is_document_node=False):
- if credentials is None:
- credentials = []
- return {
- "id": node_id,
- 'name': 'java-%s' % node_id[:4],
- "type": "java-document-widget" if is_document_node else "java-widget",
- "properties":{
- "job_xml": [],
- "jar_path": "",
- "java_opts": [],
- "retry_max": [],
- "retry_interval": [],
- "job_properties": [],
- "capture_output": False,
- "prepares": [],
- "credentials": credentials,
- "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"}],
- "archives": []
- },
- "children": [
- {"to": "33430f0f-ebfa-c3ec-f237-3e77efa03d0a"},
- {"error": "17c9c895-5a16-7443-bb81-f34b30b21548"}
- ],
- "actionParameters": [],
- "actionParametersFetched": False
- }
- def get_java_snippet_node(self, snippet):
- credentials = []
- node_id = snippet.get('id', str(uuid.uuid4()))
- node = self._get_java_node(node_id, credentials)
- node['properties']['main_class'] = snippet['properties']['class']
- node['properties']['app_jar'] = snippet['properties']['app_jar'] # Not used, submission add it to oozie.libpath instead
- node['properties']['files'] = [{'value': f['path']} for f in snippet['properties']['files']]
- node['properties']['arguments'] = [{'value': f} for f in snippet['properties']['arguments']]
- return node
- def get_java_document_node(self, document):
- credentials = []
- node = self._get_java_node(document.uuid, credentials, is_document_node=True)
- node['uuid'] = document.uuid
- return node
- def get_workflow(self, nodes, name, doc_uuid, user, managed=False):
- parameters = []
- data = {
- u'workflow': {
- u'name': name,
- u'nodes': [{
- u'name': u'Start',
- u'properties': {},
- u'actionParametersFetched': False,
- u'id': u'3f107997-04cc-8733-60a9-a4bb62cebffc',
- u'type': u'start-widget',
- u'children': [{u'to': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a'}],
- u'actionParameters': [],
- }, {
- u'name': u'End',
- u'properties': {
- u'body': u'',
- u'cc': u'',
- u'to': u'',
- u'enableMail': False,
- u'message': u'Workflow ${wf:id()} finished',
- u'subject': u'',
- u'attachment': u''
- },
- u'actionParametersFetched': False,
- u'id': u'33430f0f-ebfa-c3ec-f237-3e77efa03d0a',
- u'type': u'end-widget',
- u'children': [],
- u'actionParameters': [],
- }, {
- u'name': u'Kill',
- u'properties': {
- u'body': u'',
- u'cc': u'',
- u'to': u'',
- u'enableMail': False,
- u'message': u'Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]',
- u'subject': u'',
- u'attachment': u''
- },
- u'actionParametersFetched': False,
- u'id': u'17c9c895-5a16-7443-bb81-f34b30b21548',
- u'type': u'kill-widget',
- u'children': [],
- u'actionParameters': [],
- }
- ],
- u'properties': {
- u'job_xml': u'',
- u'description': u'',
- u'wf1_id': None,
- u'sla_enabled': False,
- u'deployment_dir': Job.get_workspace(user),
- u'schema_version': u'uri:oozie:workflow:0.5',
- u'sla': [
- {u'key': u'enabled', u'value': False},
- {u'key': u'nominal-time', u'value': u'${nominal_time}'},
- {u'key': u'should-start', u'value': u''},
- {u'key': u'should-end', u'value': u'${30 * MINUTES}'},
- {u'key': u'max-duration', u'value': u''},
- {u'key': u'alert-events', u'value': u''},
- {u'key': u'alert-contact', u'value': u''},
- {u'key': u'notification-msg', u'value': u''},
- {u'key': u'upstream-apps', u'value': u''},
- ],
- u'show_arrows': True,
- u'parameters': parameters,
- u'properties': [],
- },
- u'uuid': str(uuid.uuid4()),
- }
- }
- _prev_node = data['workflow']['nodes'][0]
- for node in nodes:
- data['workflow']['nodes'].append(node)
- _prev_node['children'][0]['to'] = node['id'] # We link nodes
- _prev_node = node
- workflow_doc = _save_workflow(data['workflow'], {}, user)
- workflow_doc.is_managed = managed
- workflow_doc.save()
- return workflow_doc
|