models2.py 56 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774
  1. #!/usr/bin/env python
  2. # Licensed to Cloudera, Inc. under one
  3. # or more contributor license agreements. See the NOTICE file
  4. # distributed with this work for additional information
  5. # regarding copyright ownership. Cloudera, Inc. licenses this file
  6. # to you under the Apache License, Version 2.0 (the
  7. # "License"); you may not use this file except in compliance
  8. # with the License. You may obtain a copy of the License at
  9. #
  10. # http://www.apache.org/licenses/LICENSE-2.0
  11. #
  12. # Unless required by applicable law or agreed to in writing, software
  13. # distributed under the License is distributed on an "AS IS" BASIS,
  14. # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15. # See the License for the specific language governing permissions and
  16. # limitations under the License.
  17. import json
  18. import logging
  19. import re
  20. import time
  21. import uuid
  22. from datetime import datetime, timedelta
  23. from dateutil.parser import parse
  24. from string import Template
  25. from django.utils.encoding import force_unicode
  26. from desktop.lib.json_utils import JSONEncoderForHTML
  27. from django.utils.translation import ugettext as _
  28. from desktop.lib import django_mako
  29. from desktop.models import Document2
  30. from hadoop.fs.hadoopfs import Hdfs
  31. from liboozie.submission2 import Submission
  32. from liboozie.submission2 import create_directories
  33. from oozie.conf import REMOTE_SAMPLE_DIR
  34. from oozie.models import Workflow as OldWorflows
  35. from oozie.utils import utc_datetime_format
  36. LOG = logging.getLogger(__name__)
  37. class Job(object):
  38. def find_all_parameters(self, with_lib_path=True):
  39. params = self.find_parameters()
  40. for param in self.parameters:
  41. params[param['name'].strip()] = param['value']
  42. return [{'name': name, 'value': value} for name, value in params.iteritems() if with_lib_path or name != 'oozie.use.system.libpath']
  43. @classmethod
  44. def get_workspace(cls, user):
  45. return (REMOTE_SAMPLE_DIR.get() + '/hue-oozie-$TIME').replace('$USER', user.username).replace('$TIME', str(time.time()))
  46. @property
  47. def validated_name(self):
  48. good_name = []
  49. for c in self.name[:40]:
  50. if not good_name:
  51. if not re.match('[a-zA-Z_]', c):
  52. c = '_'
  53. else:
  54. if not re.match('[\-_a-zA-Z0-9]', c):
  55. c = '_'
  56. good_name.append(c)
  57. return ''.join(good_name)
  58. class Workflow(Job):
  59. XML_FILE_NAME = 'workflow.xml'
  60. PROPERTY_APP_PATH = 'oozie.wf.application.path'
  61. SLA_DEFAULT = [
  62. {'key': 'enabled', 'value': False},
  63. {'key': 'nominal-time', 'value': '${nominal_time}'},
  64. {'key': 'should-start', 'value': ''},
  65. {'key': 'should-end', 'value': '${30 * MINUTES}'},
  66. {'key': 'max-duration', 'value': ''},
  67. {'key': 'alert-events', 'value': ''},
  68. {'key': 'alert-contact', 'value': ''},
  69. {'key': 'notification-msg', 'value': ''},
  70. {'key': 'upstream-apps', 'value': ''},
  71. ]
  72. HUE_ID = 'hue-id-w'
  73. def __init__(self, data=None, document=None, workflow=None):
  74. self.document = document
  75. if document is not None:
  76. self.data = document.data
  77. elif data is not None:
  78. self.data = data
  79. else:
  80. self.data = json.dumps({
  81. 'layout': [{
  82. "size":12, "rows":[
  83. {"widgets":[{"size":12, "name":"Start", "id":"3f107997-04cc-8733-60a9-a4bb62cebffc", "widgetType":"start-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
  84. {"widgets":[{"size":12, "name":"End", "id":"33430f0f-ebfa-c3ec-f237-3e77efa03d0a", "widgetType":"end-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]},
  85. {"widgets":[{"size":12, "name":"Kill", "id":"17c9c895-5a16-7443-bb81-f34b30b21548", "widgetType":"kill-widget", "properties":{}, "offset":0, "isLoading":False, "klass":"card card-widget span12"}]}
  86. ],
  87. "drops":[ "temp"],
  88. "klass":"card card-home card-column span12"
  89. }],
  90. 'workflow': workflow if workflow is not None else {
  91. "id": None,
  92. "uuid": None,
  93. "name": "My Workflow",
  94. "properties": {
  95. "description": "",
  96. "job_xml": "",
  97. "sla_enabled": False,
  98. "schema_version": "uri:oozie:workflow:0.5",
  99. "sla_workflow_enabled": False,
  100. "credentials": [],
  101. "properties": [],
  102. "sla": Workflow.SLA_DEFAULT,
  103. "show_arrows": True,
  104. "wf1_id": None
  105. },
  106. "nodes":[
  107. {"id":"3f107997-04cc-8733-60a9-a4bb62cebffc","name":"Start","type":"start-widget","properties":{},"children":[{'to': '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'}]},
  108. {"id":"33430f0f-ebfa-c3ec-f237-3e77efa03d0a","name":"End","type":"end-widget","properties":{},"children":[]},
  109. {"id":"17c9c895-5a16-7443-bb81-f34b30b21548","name":"Kill","type":"kill-widget","properties":{'message': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]')},"children":[]}
  110. ]
  111. }
  112. })
  113. @property
  114. def id(self):
  115. return self.document.id
  116. @property
  117. def uuid(self):
  118. return self.document.uuid
  119. def get_json(self):
  120. _data = self.get_data()
  121. return json.dumps(_data)
  122. def get_data(self):
  123. _data = json.loads(self.data)
  124. if self.document is not None:
  125. _data['workflow']['id'] = self.document.id
  126. _data['workflow']['dependencies'] = list(self.document.dependencies.values('uuid',))
  127. else:
  128. _data['workflow']['dependencies'] = []
  129. if 'parameters' not in _data['workflow']['properties']:
  130. _data['workflow']['properties']['parameters'] = [
  131. {'name': 'oozie.use.system.libpath', 'value': True},
  132. ]
  133. if 'show_arrows' not in _data['workflow']['properties']:
  134. _data['workflow']['properties']['show_arrows'] = True
  135. return _data
  136. def to_xml(self, mapping=None):
  137. if mapping is None:
  138. mapping = {}
  139. tmpl = 'editor/gen2/workflow.xml.mako'
  140. data = self.get_data()
  141. 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
  142. node_mapping = dict([(node.id, node) for node in nodes])
  143. sub_wfs_ids = [node.data['properties']['workflow'] for node in nodes if node.data['type'] == 'subworkflow']
  144. workflow_mapping = dict([(workflow.uuid, Workflow(document=workflow)) for workflow in Document2.objects.filter(uuid__in=sub_wfs_ids)])
  145. xml = re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
  146. 'wf': self,
  147. 'workflow': data['workflow'],
  148. 'nodes': nodes,
  149. 'mapping': mapping,
  150. 'node_mapping': node_mapping,
  151. 'workflow_mapping': workflow_mapping
  152. }))
  153. return force_unicode(xml)
  154. @property
  155. def name(self):
  156. _data = self.get_data()
  157. return _data['workflow']['name']
  158. def update_name(self, name):
  159. _data = self.get_data()
  160. _data['workflow']['name'] = name
  161. self.data = json.dumps(_data)
  162. @property
  163. def deployment_dir(self):
  164. _data = self.get_data()
  165. return _data['workflow']['properties']['deployment_dir']
  166. @property
  167. def parameters(self):
  168. _data = self.get_data()
  169. return _data['workflow']['properties']['parameters']
  170. @property
  171. def sla_enabled(self):
  172. _data = self.get_data()
  173. return _data['workflow']['properties']['sla_enabled']
  174. @property
  175. def sla(self):
  176. _data = self.get_data()
  177. return _data['workflow']['properties']['sla']
  178. @property
  179. def nodes(self):
  180. _data = self.get_data()
  181. return [Node(node) for node in _data['workflow']['nodes']]
  182. def find_parameters(self):
  183. params = set()
  184. if self.sla_enabled:
  185. for param in find_json_parameters(self.sla):
  186. params.add(param)
  187. for node in self.nodes:
  188. params.update(node.find_parameters())
  189. return dict([(param, '') for param in list(params)])
  190. def set_workspace(self, user):
  191. _data = json.loads(self.data)
  192. _data['workflow']['properties']['deployment_dir'] = Job.get_workspace(user)
  193. self.data = json.dumps(_data)
  194. def check_workspace(self, fs, user):
  195. # Create optional root workspace for the first submission
  196. root = REMOTE_SAMPLE_DIR.get().rsplit('/', 1)
  197. if len(root) > 1 and '$' not in root[0]:
  198. create_directories(fs, [root[0]])
  199. Submission(user, self, fs, None, {})._create_dir(self.deployment_dir)
  200. Submission(user, self, fs, None, {})._create_dir(Hdfs.join(self.deployment_dir, 'lib'))
  201. class Node():
  202. def __init__(self, data):
  203. self.data = data
  204. self._augment_data()
  205. def to_xml(self, mapping=None, node_mapping=None, workflow_mapping=None):
  206. if mapping is None:
  207. mapping = {}
  208. if node_mapping is None:
  209. node_mapping = {}
  210. if workflow_mapping is None:
  211. workflow_mapping = {}
  212. data = {
  213. 'node': self.data,
  214. 'mapping': mapping,
  215. 'node_mapping': node_mapping,
  216. 'workflow_mapping': workflow_mapping
  217. }
  218. return django_mako.render_to_string(self.get_template_name(), data)
  219. @property
  220. def id(self):
  221. return self.data['id']
  222. @property
  223. def name(self):
  224. return self.data['name']
  225. @property
  226. def sla_enabled(self):
  227. _data = self.get_data()
  228. return _data['workflow']['properties']['sla_enabled']
  229. def _augment_data(self):
  230. self.data['type'] = self.data['type'].replace('-widget', '')
  231. self.data['uuid'] = self.data['id']
  232. # Action Node
  233. if 'credentials' not in self.data['properties']:
  234. self.data['properties']['credentials'] = []
  235. if 'prepares' not in self.data['properties']:
  236. self.data['properties']['prepares'] = []
  237. if 'job_xml' not in self.data['properties']:
  238. self.data['properties']['job_xml'] = []
  239. if 'properties' not in self.data['properties']:
  240. self.data['properties']['properties'] = []
  241. if 'params' not in self.data['properties']:
  242. self.data['properties']['params'] = []
  243. if 'files' not in self.data['properties']:
  244. self.data['properties']['files'] = []
  245. if 'archives' not in self.data['properties']:
  246. self.data['properties']['archives'] = []
  247. if 'sla_enabled' not in self.data['properties']:
  248. self.data['properties']['sla_enabled'] = False
  249. if 'sla' not in self.data['properties']:
  250. self.data['properties']['sla'] = []
  251. def get_template_name(self):
  252. return 'editor/gen2/workflow-%s.xml.mako' % self.data['type']
  253. def find_parameters(self):
  254. return find_parameters(self)
  255. class Action(object):
  256. @classmethod
  257. def get_fields(cls):
  258. return [(f['name'], f['value']) for f in cls.FIELDS.itervalues()] + [('sla', Workflow.SLA_DEFAULT), ('credentials', [])]
  259. class StartNode(Action):
  260. TYPE = 'start'
  261. FIELDS = {}
  262. class EndNode(Action):
  263. TYPE = 'end'
  264. FIELDS = {}
  265. class PigAction(Action):
  266. TYPE = 'pig'
  267. FIELDS = {
  268. 'script_path': {
  269. 'name': 'script_path',
  270. 'label': _('Script'),
  271. 'value': '',
  272. 'help_text': _('Path to the script on HDFS.'),
  273. 'type': ''
  274. },
  275. 'parameters': {
  276. 'name': 'parameters',
  277. 'label': _('Parameters'),
  278. 'value': [],
  279. 'help_text': _('The Pig parameters of the script without -param. e.g. INPUT=${inputDir}'),
  280. 'type': ''
  281. },
  282. 'arguments': {
  283. 'name': 'arguments',
  284. 'label': _('Arguments'),
  285. 'value': [],
  286. 'help_text': _('The Pig parameters of the script as is. e.g. -param, INPUT=${inputDir}'),
  287. 'type': ''
  288. },
  289. # Common
  290. 'files': {
  291. 'name': 'files',
  292. 'label': _('Files'),
  293. 'value': [],
  294. 'help_text': _('Files put in the running directory.'),
  295. 'type': ''
  296. },
  297. 'archives': {
  298. 'name': 'archives',
  299. 'label': _('Archives'),
  300. 'value': [],
  301. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  302. 'type': ''
  303. },
  304. 'job_properties': {
  305. 'name': 'job_properties',
  306. 'label': _('Hadoop job properties'),
  307. 'value': [],
  308. 'help_text': _('value, e.g. production'),
  309. 'type': ''
  310. },
  311. 'prepares': {
  312. 'name': 'prepares',
  313. 'label': _('Prepares'),
  314. 'value': [],
  315. 'help_text': _('Path to manipulate before starting the application.'),
  316. 'type': ''
  317. },
  318. 'job_xml': {
  319. 'name': 'job_xml',
  320. 'label': _('Job XML'),
  321. 'value': [],
  322. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  323. 'type': ''
  324. }
  325. }
  326. @classmethod
  327. def get_mandatory_fields(cls):
  328. return [cls.FIELDS['script_path']]
  329. class JavaAction(Action):
  330. TYPE = 'java'
  331. FIELDS = {
  332. 'jar_path': {
  333. 'name': 'jar_path',
  334. 'label': _('Jar name'),
  335. 'value': '',
  336. 'help_text': _('Path to the jar on HDFS.'),
  337. 'type': ''
  338. },
  339. 'main_class': {
  340. 'name': 'main_class',
  341. 'label': _('Main class'),
  342. 'value': '',
  343. 'help_text': _('Java class. e.g. org.apache.hadoop.examples.Grep'),
  344. 'type': 'text'
  345. },
  346. 'arguments': {
  347. 'name': 'arguments',
  348. 'label': _('Arguments'),
  349. 'value': [],
  350. 'help_text': _('Arguments of the main method. The value of each arg element is considered a single argument '
  351. 'and they are passed to the main method in the same order.'),
  352. 'type': ''
  353. },
  354. 'java_opts': {
  355. 'name': 'java_opts',
  356. 'label': _('Java options'),
  357. 'value': [],
  358. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b'),
  359. 'type': ''
  360. },
  361. 'capture_output': {
  362. 'name': 'capture_output',
  363. 'label': _('Capture output'),
  364. 'value': False,
  365. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  366. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  367. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  368. 'via the String action:output(String node, String key) function') % {'program': TYPE.title()},
  369. 'type': ''
  370. },
  371. # Common
  372. 'files': {
  373. 'name': 'files',
  374. 'label': _('Files'),
  375. 'value': [],
  376. 'help_text': _('Files put in the running directory.'),
  377. 'type': ''
  378. },
  379. 'archives': {
  380. 'name': 'archives',
  381. 'label': _('Archives'),
  382. 'value': [],
  383. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  384. 'type': ''
  385. },
  386. 'job_properties': {
  387. 'name': 'job_properties',
  388. 'label': _('Hadoop job properties'),
  389. 'value': [],
  390. 'help_text': _('value, e.g. production'),
  391. 'type': ''
  392. },
  393. 'prepares': {
  394. 'name': 'prepares',
  395. 'label': _('Prepares'),
  396. 'value': [],
  397. 'help_text': _('Path to manipulate before starting the application.'),
  398. 'type': ''
  399. },
  400. 'job_xml': {
  401. 'name': 'job_xml',
  402. 'label': _('Job XML'),
  403. 'value': [],
  404. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  405. 'type': ''
  406. }
  407. }
  408. @classmethod
  409. def get_mandatory_fields(cls):
  410. return [cls.FIELDS['jar_path'], cls.FIELDS['main_class']]
  411. class HiveAction(Action):
  412. TYPE = 'hive'
  413. FIELDS = {
  414. 'script_path': {
  415. 'name': 'script_path',
  416. 'label': _('Script'),
  417. 'value': '',
  418. 'help_text': _('Path to the script on HDFS.'),
  419. 'type': ''
  420. },
  421. 'parameters': {
  422. 'name': 'parameters',
  423. 'label': _('Parameters'),
  424. 'value': [],
  425. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  426. 'type': ''
  427. },
  428. # Common
  429. 'files': {
  430. 'name': 'files',
  431. 'label': _('Files'),
  432. 'value': [],
  433. 'help_text': _('Files put in the running directory.'),
  434. 'type': ''
  435. },
  436. 'archives': {
  437. 'name': 'archives',
  438. 'label': _('Archives'),
  439. 'value': [],
  440. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  441. 'type': ''
  442. },
  443. 'job_properties': {
  444. 'name': 'job_properties',
  445. 'label': _('Hadoop job properties'),
  446. 'value': [],
  447. 'help_text': _('value, e.g. production'),
  448. 'type': ''
  449. },
  450. 'prepares': {
  451. 'name': 'prepares',
  452. 'label': _('Prepares'),
  453. 'value': [],
  454. 'help_text': _('Path to manipulate before starting the application.'),
  455. 'type': ''
  456. },
  457. 'hive_xml': {
  458. 'name': 'hive_xml',
  459. 'label': _('Hive XML'),
  460. 'value': '',
  461. 'help_text': _('Refer to a hive-site.xml for connecting to Hive'),
  462. 'type': ''
  463. }
  464. }
  465. @classmethod
  466. def get_mandatory_fields(cls):
  467. return [cls.FIELDS['script_path'], cls.FIELDS['hive_xml']]
  468. class HiveServer2Action(Action):
  469. TYPE = 'hive2'
  470. FIELDS = {
  471. 'script_path': {
  472. 'name': 'script_path',
  473. 'label': _('Script'),
  474. 'value': '',
  475. 'help_text': _('Path to the script on HDFS.'),
  476. 'type': ''
  477. },
  478. 'parameters': {
  479. 'name': 'parameters',
  480. 'label': _('Parameters'),
  481. 'value': [],
  482. 'help_text': _('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': TYPE.title()},
  483. 'type': ''
  484. },
  485. # Common
  486. 'jdbc_url': {
  487. 'name': 'jdbc_url',
  488. 'label': _('JDBC URL'),
  489. 'value': 'jdbc:hive2://localhost:10000/default',
  490. 'help_text': _('JDBC URL for the Hive Server 2. Beeline will use this to know where to connect to.'),
  491. 'type': ''
  492. },
  493. 'password': {
  494. 'name': 'password',
  495. 'label': _('Password'),
  496. 'value': '',
  497. '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 '
  498. 'something requiring a password (e.g. LDAP); non-secured Hive Server 2 or Kerberized Hive Server 2 don\'t require a password.'),
  499. 'type': ''
  500. },
  501. 'files': {
  502. 'name': 'files',
  503. 'label': _('Files'),
  504. 'value': [],
  505. 'help_text': _('Files put in the running directory.'),
  506. 'type': ''
  507. },
  508. 'archives': {
  509. 'name': 'archives',
  510. 'label': _('Archives'),
  511. 'value': [],
  512. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  513. 'type': ''
  514. },
  515. 'job_properties': {
  516. 'name': 'job_properties',
  517. 'label': _('Hadoop job properties'),
  518. 'value': [],
  519. 'help_text': _('value, e.g. production'),
  520. 'type': ''
  521. },
  522. 'prepares': {
  523. 'name': 'prepares',
  524. 'label': _('Prepares'),
  525. 'value': [],
  526. 'help_text': _('Path to manipulate before starting the application.'),
  527. 'type': ''
  528. },
  529. 'job_xml': {
  530. 'name': 'job_xml',
  531. 'label': _('Job XML'),
  532. 'value': '',
  533. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  534. 'type': ''
  535. }
  536. }
  537. @classmethod
  538. def get_mandatory_fields(cls):
  539. return [cls.FIELDS['script_path']]
  540. class SubWorkflowAction(Action):
  541. TYPE = 'subworkflow'
  542. FIELDS = {
  543. 'workflow': {
  544. 'name': 'workflow',
  545. 'label': _('Sub-workflow'),
  546. 'value': None,
  547. 'help_text': _('The sub-workflow application to include. You must own all the sub-workflows'),
  548. 'type': 'workflow'
  549. },
  550. 'propagate_configuration': {
  551. 'name': 'propagate_configuration',
  552. 'label': _('Propagate configuration'),
  553. 'value': True,
  554. 'help_text': _('If the workflow job configuration should be propagated to the child workflow.'),
  555. 'type': ''
  556. },
  557. 'job_properties': {
  558. 'name': 'job_properties',
  559. 'label': _('Hadoop job properties'),
  560. 'value': [],
  561. 'help_text': _('Can be used to specify the job properties that are required to run the child workflow job.'),
  562. 'type': ''
  563. }
  564. }
  565. @classmethod
  566. def get_mandatory_fields(cls):
  567. return [cls.FIELDS['workflow']]
  568. class SqoopAction(Action):
  569. TYPE = 'sqoop'
  570. FIELDS = {
  571. 'command': {
  572. 'name': 'command',
  573. 'label': _('Sqoop command'),
  574. 'value': 'import --connect jdbc:hsqldb:file:db.hsqldb --table TT --target-dir hdfs://localhost:8020/user/foo -m 1',
  575. '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},
  576. 'type': 'textarea'
  577. },
  578. 'parameters': {
  579. 'name': 'parameters',
  580. 'label': _('Arguments'),
  581. 'value': [],
  582. 'help_text': _('If no command is specified, split the command by spaces and insert the %(type)s parameters '
  583. 'here e.g. import, --connect, jdbc:hsqldb:file:db.hsqldb, ...') % {'type': TYPE},
  584. 'type': ''
  585. },
  586. # Common
  587. 'files': {
  588. 'name': 'files',
  589. 'label': _('Files'),
  590. 'value': [],
  591. 'help_text': _('Files put in the running directory.'),
  592. 'type': ''
  593. },
  594. 'archives': {
  595. 'name': 'archives',
  596. 'label': _('Archives'),
  597. 'value': [],
  598. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  599. 'type': ''
  600. },
  601. 'job_properties': {
  602. 'name': 'job_properties',
  603. 'label': _('Hadoop job properties'),
  604. 'value': [],
  605. 'help_text': _('value, e.g. production'),
  606. 'type': ''
  607. },
  608. 'prepares': {
  609. 'name': 'prepares',
  610. 'label': _('Prepares'),
  611. 'value': [],
  612. 'help_text': _('Path to manipulate before starting the application.'),
  613. 'type': ''
  614. },
  615. 'job_xml': {
  616. 'name': 'job_xml',
  617. 'label': _('Job XML'),
  618. 'value': '',
  619. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  620. 'type': ''
  621. }
  622. }
  623. @classmethod
  624. def get_mandatory_fields(cls):
  625. return [cls.FIELDS['command']]
  626. class MapReduceAction(Action):
  627. TYPE = 'mapreduce'
  628. FIELDS = {
  629. 'jar_path': {
  630. 'name': 'jar_path',
  631. 'label': _('Jar name'),
  632. 'value': '',
  633. 'help_text': _('Path to the jar on HDFS.'),
  634. 'type': ''
  635. },
  636. # Common
  637. 'files': {
  638. 'name': 'files',
  639. 'label': _('Files'),
  640. 'value': [],
  641. 'help_text': _('Files put in the running directory.'),
  642. 'type': ''
  643. },
  644. 'archives': {
  645. 'name': 'archives',
  646. 'label': _('Archives'),
  647. 'value': [],
  648. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  649. 'type': ''
  650. },
  651. 'job_properties': {
  652. 'name': 'job_properties',
  653. 'label': _('Hadoop job properties'),
  654. 'value': [],
  655. 'help_text': _('value, e.g. production'),
  656. 'type': ''
  657. },
  658. 'prepares': {
  659. 'name': 'prepares',
  660. 'label': _('Prepares'),
  661. 'value': [],
  662. 'help_text': _('Path to manipulate before starting the application.'),
  663. 'type': ''
  664. },
  665. 'job_xml': {
  666. 'name': 'job_xml',
  667. 'label': _('Job XML'),
  668. 'value': '',
  669. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  670. 'type': ''
  671. }
  672. }
  673. @classmethod
  674. def get_mandatory_fields(cls):
  675. return [cls.FIELDS['jar_path']]
  676. class ShellAction(Action):
  677. TYPE = 'shell'
  678. FIELDS = {
  679. 'shell_command': {
  680. 'name': 'shell_command',
  681. 'label': _('Shell command'),
  682. 'value': '',
  683. 'help_text': _('Shell command to execute, e.g script.sh'),
  684. 'type': ''
  685. },
  686. 'arguments': {
  687. 'name': 'arguments',
  688. 'label': _('Arguments'),
  689. 'value': [],
  690. 'help_text': _('One arg, e.g. -l, --help'),
  691. 'type': ''
  692. },
  693. 'env_var': {
  694. 'name': 'env_var',
  695. 'label': _('Environment variables'),
  696. 'value': [],
  697. 'help_text': _('e.g. MAX=10 or PATH=$PATH:mypath'),
  698. 'type': ''
  699. },
  700. 'capture_output': {
  701. 'name': 'capture_output',
  702. 'label': _('Capture output'),
  703. 'value': True,
  704. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  705. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  706. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  707. 'via the String action:output(String node, String key) function') % {'program': TYPE},
  708. 'type': ''
  709. },
  710. # Common
  711. 'files': {
  712. 'name': 'files',
  713. 'label': _('Files'),
  714. 'value': [],
  715. 'help_text': _('Files put in the running directory.'),
  716. 'type': ''
  717. },
  718. 'archives': {
  719. 'name': 'archives',
  720. 'label': _('Archives'),
  721. 'value': [],
  722. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.'),
  723. 'type': ''
  724. },
  725. 'job_properties': {
  726. 'name': 'job_properties',
  727. 'label': _('Hadoop job properties'),
  728. 'value': [],
  729. 'help_text': _('value, e.g. production'),
  730. 'type': ''
  731. },
  732. 'prepares': {
  733. 'name': 'prepares',
  734. 'label': _('Prepares'),
  735. 'value': [],
  736. 'help_text': _('Path to manipulate before starting the application.'),
  737. 'type': ''
  738. },
  739. 'job_xml': {
  740. 'name': 'job_xml',
  741. 'label': _('Job XML'),
  742. 'value': '',
  743. 'help_text': _('Refer to a Hadoop JobConf job.xml'),
  744. 'type': ''
  745. }
  746. }
  747. @classmethod
  748. def get_mandatory_fields(cls):
  749. return [cls.FIELDS['shell_command']]
  750. class SshAction(Action):
  751. TYPE = 'ssh'
  752. FIELDS = {
  753. 'host': {
  754. 'name': 'host',
  755. 'label': _('User and Host'),
  756. 'value': 'user@host.com',
  757. 'help_text': _('Where the shell will be executed.'),
  758. 'type': 'text'
  759. },
  760. 'ssh_command': {
  761. 'name': 'ssh_command',
  762. 'label': _('Ssh command'),
  763. 'value': 'ls',
  764. 'help_text': _('The path of the Shell command to execute.'),
  765. 'type': 'textarea'
  766. },
  767. 'arguments': {
  768. 'name': 'arguments',
  769. 'label': _('Arguments'),
  770. 'value': [],
  771. 'help_text': _('One arg, e.g. -l, --help'),
  772. 'type': ''
  773. },
  774. 'capture_output': {
  775. 'name': 'capture_output',
  776. 'label': _('Capture output'),
  777. 'value': True,
  778. 'help_text': _('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  779. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  780. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  781. 'via the String action:output(String node, String key) function') % {'program': TYPE},
  782. 'type': ''
  783. },
  784. }
  785. @classmethod
  786. def get_mandatory_fields(cls):
  787. return [cls.FIELDS['host'], cls.FIELDS['ssh_command']]
  788. class FsAction(Action):
  789. TYPE = 'fs'
  790. FIELDS = {
  791. 'deletes': {
  792. 'name': 'deletes',
  793. 'label': _('Delete path'),
  794. 'value': [],
  795. 'help_text': _('Deletes recursively all content.'),
  796. 'type': ''
  797. },
  798. 'mkdirs': {
  799. 'name': 'mkdirs',
  800. 'label': _('Create directory'),
  801. 'value': [],
  802. 'help_text': _('Sub directories are created if needed.'),
  803. 'type': ''
  804. },
  805. 'moves': {
  806. 'name': 'moves',
  807. 'label': _('Move file or directory'),
  808. 'value': [],
  809. 'help_text': _('Destination.'),
  810. 'type': ''
  811. },
  812. 'chmods': {
  813. 'name': 'chmods',
  814. 'label': _('Change permissions'),
  815. 'value': [],
  816. 'help_text': _('File or directory.'),
  817. 'type': ''
  818. },
  819. 'touchzs': {
  820. 'name': 'touchzs',
  821. 'label': _('Create or touch a file'),
  822. 'value': [],
  823. 'help_text': _('Or update its modification date.'),
  824. 'type': ''
  825. },
  826. 'chgrps': {
  827. 'name': 'chgrps',
  828. 'label': _('Change the group'),
  829. 'value': [],
  830. 'help_text': _('File or directory.'),
  831. 'type': ''
  832. }
  833. }
  834. @classmethod
  835. def get_mandatory_fields(cls):
  836. return [cls.FIELDS['deletes'], cls.FIELDS['mkdirs'], cls.FIELDS['moves'], cls.FIELDS['chmods']]
  837. class EmailAction(Action):
  838. TYPE = 'email'
  839. FIELDS = {
  840. 'to': {
  841. 'name': 'to',
  842. 'label': _('To addresses'),
  843. 'value': '',
  844. 'help_text': _('Comma-separated values'),
  845. 'type': 'text'
  846. },
  847. 'cc': {
  848. 'name': 'cc',
  849. 'label': _('Cc addresses (optional)'),
  850. 'value': '',
  851. 'help_text': _('Comma-separated values'),
  852. 'type': 'text'
  853. },
  854. 'subject': {
  855. 'name': 'subject',
  856. 'label': _('Subject'),
  857. 'value': '',
  858. 'help_text': _('Plain-text'),
  859. 'type': 'text'
  860. },
  861. 'body': {
  862. 'name': 'body',
  863. 'label': _('Body'),
  864. 'value': '',
  865. 'help_text': _('Plain-text'),
  866. 'type': 'textarea'
  867. },
  868. }
  869. @classmethod
  870. def get_mandatory_fields(cls):
  871. return [cls.FIELDS['to'], cls.FIELDS['subject'], cls.FIELDS['body']]
  872. class StreamingAction(Action):
  873. TYPE = 'streaming'
  874. FIELDS = {
  875. 'mapper': {
  876. 'name': 'mapper',
  877. 'label': _('Mapper'),
  878. 'value': '',
  879. 'help_text': _('The executable/script to be used as mapper.'),
  880. 'type': ''
  881. },
  882. 'reducer': {
  883. 'name': 'reducer',
  884. 'label': _('Reducer'),
  885. 'value': '',
  886. 'help_text': _('The executable/script to be used as reducer.'),
  887. 'type': ''
  888. },
  889. # Common
  890. 'files': {
  891. 'name': 'files',
  892. 'label': _('Files'),
  893. 'value': [],
  894. 'help_text': _('Files put in the running directory.')
  895. },
  896. 'archives': {
  897. 'name': 'archives',
  898. 'label': _('Archives'),
  899. 'value': [],
  900. 'help_text': _('zip, tar and tgz/tar.gz uncompressed into the running directory.')
  901. },
  902. 'job_properties': {
  903. 'name': 'job_properties',
  904. 'label': _('Hadoop job properties'),
  905. 'value': [],
  906. 'help_text': _('value, e.g. production')
  907. },
  908. 'prepares': {
  909. 'name': 'prepares',
  910. 'label': _('Prepares'),
  911. 'value': [],
  912. 'help_text': _('Path to manipulate before starting the application.')
  913. },
  914. 'job_xml': {
  915. 'name': 'job_xml',
  916. 'label': _('Job XML'),
  917. 'value': '',
  918. 'help_text': _('Refer to a Hadoop JobConf job.xml')
  919. }
  920. }
  921. @classmethod
  922. def get_mandatory_fields(cls):
  923. return [cls.FIELDS['mapper'], cls.FIELDS['reducer']]
  924. class DistCpAction(Action):
  925. TYPE = 'distcp'
  926. FIELDS = {
  927. 'distcp_parameters': {
  928. 'name': 'distcp_parameters',
  929. 'label': _('Arguments'),
  930. 'value': [{'value': ''}, {'value': ''}],
  931. 'help_text': _('Options first, then source / destination paths'),
  932. 'type': 'distcp'
  933. },
  934. # Common
  935. 'prepares': {
  936. 'name': 'prepares',
  937. 'label': _('Prepares'),
  938. 'value': [],
  939. 'help_text': _('Path to manipulate before starting the application.')
  940. },
  941. 'job_properties': {
  942. 'name': 'job_properties',
  943. 'label': _('Hadoop job properties'),
  944. 'value': [],
  945. 'help_text': _('value, e.g. production')
  946. },
  947. 'java_opts': {
  948. 'name': 'java_opts',
  949. 'label': _('Java options'),
  950. 'value': '',
  951. 'help_text': _('Parameters for the JVM, e.g. -Dprop1=a -Dprop2=b')
  952. }
  953. }
  954. @classmethod
  955. def get_mandatory_fields(cls):
  956. return [cls.FIELDS['distcp_parameters']]
  957. class KillAction(Action):
  958. TYPE = 'kill'
  959. FIELDS = {
  960. 'message': {
  961. 'name': 'message',
  962. 'label': _('Message'),
  963. 'value': _('Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]'),
  964. 'help_text': _('Message to display when the workflow fails. Can contain some EL functions.'),
  965. 'type': 'textarea'
  966. }
  967. }
  968. @classmethod
  969. def get_mandatory_fields(cls):
  970. return [cls.FIELDS['message']]
  971. class JoinAction(Action):
  972. TYPE = 'join'
  973. FIELDS = {}
  974. @classmethod
  975. def get_mandatory_fields(cls):
  976. return []
  977. class ForkNode(Action):
  978. TYPE = 'fork'
  979. FIELDS = {}
  980. @classmethod
  981. def get_mandatory_fields(cls):
  982. return []
  983. class DecisionNode(Action):
  984. TYPE = 'decision'
  985. FIELDS = {}
  986. @classmethod
  987. def get_mandatory_fields(cls):
  988. return []
  989. NODES = {
  990. 'start-widget': StartNode,
  991. 'end-widget': EndNode,
  992. 'pig-widget': PigAction,
  993. 'java-widget': JavaAction,
  994. 'hive-widget': HiveAction,
  995. 'hive2-widget': HiveServer2Action,
  996. 'sqoop-widget': SqoopAction,
  997. 'mapreduce-widget': MapReduceAction,
  998. 'subworkflow-widget': SubWorkflowAction,
  999. 'shell-widget': ShellAction,
  1000. 'ssh-widget': SshAction,
  1001. 'fs-widget': FsAction,
  1002. 'email-widget': EmailAction,
  1003. 'streaming-widget': StreamingAction,
  1004. 'distcp-widget': DistCpAction,
  1005. 'kill-widget': KillAction,
  1006. 'join-widget': JoinAction,
  1007. 'fork-widget': ForkNode,
  1008. 'decision-widget': DecisionNode,
  1009. }
  1010. WORKFLOW_NODE_PROPERTIES = {}
  1011. for node in NODES.itervalues():
  1012. WORKFLOW_NODE_PROPERTIES.update(node.FIELDS)
  1013. def find_parameters(instance, fields=None):
  1014. """Find parameters in the given fields"""
  1015. if fields is None:
  1016. fields = NODES['%s-widget' % instance.data['type']].FIELDS.keys()
  1017. params = []
  1018. for field in fields:
  1019. data = instance.data['properties'][field]
  1020. if field == 'sla' and not instance.sla_enabled:
  1021. continue
  1022. if isinstance(data, list):
  1023. params.extend(find_json_parameters(data))
  1024. elif isinstance(data, basestring):
  1025. for match in Template.pattern.finditer(data):
  1026. name = match.group('braced')
  1027. if name is not None:
  1028. params.append(name)
  1029. return params
  1030. def find_json_parameters(fields):
  1031. # Input is list of json dict
  1032. params = []
  1033. for field in fields:
  1034. for data in field.values():
  1035. if isinstance(data, basestring):
  1036. for match in Template.pattern.finditer(data):
  1037. name = match.group('braced')
  1038. if name is not None:
  1039. params.append(name)
  1040. return params
  1041. def find_dollar_variables(text):
  1042. return re.findall('[^\n\\\\]\$([^\{ \'\"\-;\(\)]+)', text, re.MULTILINE)
  1043. def find_dollar_braced_variables(text):
  1044. vars = set()
  1045. for var in re.findall('\$\{(.+)\}', text, re.MULTILINE):
  1046. if ':' in var:
  1047. var = var.split(':', 1)[1]
  1048. vars.add(var)
  1049. return list(vars)
  1050. def import_workflow_from_hue_3_7(old_wf):
  1051. """
  1052. Example of data to transform
  1053. [<Start: start>, <Pig: Pig>, [<Kill: kill>], [<End: end>]]
  1054. [<Start: start>, <Java: TeraGenWorkflow>, <Java: TeraSort>, [<Kill: kill>], [<End: end>]]
  1055. [<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>]]
  1056. """
  1057. uuids = {}
  1058. old_nodes = old_wf.get_hierarchy()
  1059. wf = Workflow()
  1060. wf_rows = []
  1061. wf_nodes = []
  1062. data = wf.get_data()
  1063. # UUIDs node mapping
  1064. for node in old_wf.node_list:
  1065. if node.name == 'kill':
  1066. node_uuid = '17c9c895-5a16-7443-bb81-f34b30b21548'
  1067. elif node.name == 'start':
  1068. node_uuid = '3f107997-04cc-8733-60a9-a4bb62cebffc'
  1069. elif node.name == 'end':
  1070. node_uuid = '33430f0f-ebfa-c3ec-f237-3e77efa03d0a'
  1071. else:
  1072. node_uuid = str(uuid.uuid4())
  1073. uuids[node.id] = node_uuid
  1074. # Workflow
  1075. data['workflow']['uuid'] = str(uuid.uuid4())
  1076. data['workflow']['name'] = old_wf.name
  1077. data['workflow']['properties']['properties'] = json.loads(old_wf.job_properties)
  1078. data['workflow']['properties']['job_xml'] = old_wf.job_xml
  1079. data['workflow']['properties']['description'] = old_wf.description
  1080. data['workflow']['properties']['schema_version'] = old_wf.schema_version
  1081. data['workflow']['properties']['deployment_dir'] = old_wf.deployment_dir
  1082. data['workflow']['properties']['parameters'] = json.loads(old_wf.parameters)
  1083. data['workflow']['properties']['description'] = old_wf.description
  1084. data['workflow']['properties']['sla'] = old_wf.sla
  1085. data['workflow']['properties']['sla_enabled'] = old_wf.sla_enabled
  1086. data['workflow']['properties']['imported'] = True
  1087. data['workflow']['properties']['wf1_id'] = old_wf.id
  1088. # Layout
  1089. rows = data['layout'][0]['rows']
  1090. def _create_layout(nodes, size=12):
  1091. wf_rows = []
  1092. for node in nodes:
  1093. if type(node) == list and len(node) == 1:
  1094. node = node[0]
  1095. if type(node) != list:
  1096. 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":[]}]})
  1097. else:
  1098. if node[0].node_type == 'fork':
  1099. 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":[]}]})
  1100. wf_rows.append({
  1101. "id": str(uuid.uuid4()),
  1102. "widgets":[
  1103. ],
  1104. "columns":[
  1105. {
  1106. "id": str(uuid.uuid4()),
  1107. "size": (size / len(node[1])),
  1108. "rows":
  1109. [{
  1110. "id": str(uuid.uuid4()),
  1111. "widgets": c['widgets'],
  1112. "columns":[]
  1113. }
  1114. for c in col] if type(col) == list else [{
  1115. "id": str(uuid.uuid4()),
  1116. "widgets": col['widgets'],
  1117. "columns":[]
  1118. }
  1119. ]
  1120. ,
  1121. "klass":"card card-home card-column span%s" % (size / len(node[1]))
  1122. }
  1123. for col in _create_layout(node[1], size)
  1124. ]
  1125. })
  1126. 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":[]}]})
  1127. else:
  1128. wf_rows.append(_create_layout(node, size))
  1129. return wf_rows
  1130. wf_rows = _create_layout(old_nodes)
  1131. if wf_rows:
  1132. data['layout'][0]['rows'] = [data['layout'][0]['rows'][0]] + wf_rows + [data['layout'][0]['rows'][-1]]
  1133. # Content
  1134. def _dig_nodes(nodes):
  1135. for node in nodes:
  1136. if type(node) != list:
  1137. properties = {}
  1138. if '%s-widget' % node.node_type in NODES:
  1139. properties = dict(NODES['%s-widget' % node.node_type].get_fields())
  1140. if node.node_type == 'pig':
  1141. properties['script_path'] = node.script_path
  1142. properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
  1143. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1144. properties['archives'] = json.loads(node.archives)
  1145. properties['job_properties'] = json.loads(node.job_properties)
  1146. properties['prepares'] = json.loads(node.prepares)
  1147. properties['job_xml'] = node.job_xml
  1148. properties['description'] = node.description
  1149. properties['sla'] = node.sla
  1150. properties['sla_enabled'] = node.sla_enabled
  1151. elif node.node_type == 'hive':
  1152. properties['script_path'] = node.script_path
  1153. properties['parameters'] = [param for param in json.loads(node.params) if param['value'] != '-param']
  1154. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1155. properties['archives'] = json.loads(node.archives)
  1156. properties['job_properties'] = json.loads(node.job_properties)
  1157. properties['prepares'] = json.loads(node.prepares)
  1158. properties['hive_xml'] = node.job_xml
  1159. properties['description'] = node.description
  1160. properties['sla'] = node.sla
  1161. properties['sla_enabled'] = node.sla_enabled
  1162. elif node.node_type == 'java':
  1163. properties['jar_path'] = node.jar_path
  1164. properties['main_class'] = node.main_class
  1165. properties['arguments'] = [{'value': arg} for arg in node.args.split(' ')]
  1166. properties['java_opts'] = node.java_opts
  1167. properties['capture_output'] = node.capture_output
  1168. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1169. properties['archives'] = json.loads(node.archives)
  1170. properties['job_properties'] = json.loads(node.job_properties)
  1171. properties['prepares'] = json.loads(node.prepares)
  1172. properties['job_xml'] = node.job_xml
  1173. properties['description'] = node.description
  1174. properties['sla'] = node.sla
  1175. properties['sla_enabled'] = node.sla_enabled
  1176. elif node.node_type == 'sqoop':
  1177. properties['command'] = node.script_path
  1178. properties['parameters'] = json.loads(node.params)
  1179. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1180. properties['archives'] = json.loads(node.archives)
  1181. properties['job_properties'] = json.loads(node.job_properties)
  1182. properties['prepares'] = json.loads(node.prepares)
  1183. properties['job_xml'] = node.job_xml
  1184. properties['description'] = node.description
  1185. properties['sla'] = node.sla
  1186. properties['sla_enabled'] = node.sla_enabled
  1187. elif node.node_type == 'mapreduce':
  1188. properties['jar_path'] = node.jar_path
  1189. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1190. properties['archives'] = json.loads(node.archives)
  1191. properties['job_properties'] = json.loads(node.job_properties)
  1192. properties['prepares'] = json.loads(node.prepares)
  1193. properties['job_xml'] = node.job_xml
  1194. properties['description'] = node.description
  1195. properties['sla'] = node.sla
  1196. properties['sla_enabled'] = node.sla_enabled
  1197. elif node.node_type == 'shell':
  1198. properties['shell_command'] = node.command
  1199. properties['arguments'] = json.loads(node.params)
  1200. properties['capture_output'] = node.capture_output
  1201. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1202. properties['archives'] = json.loads(node.archives)
  1203. properties['job_properties'] = json.loads(node.job_properties)
  1204. properties['prepares'] = json.loads(node.prepares)
  1205. properties['job_xml'] = node.job_xml
  1206. properties['description'] = node.description
  1207. properties['sla'] = node.sla
  1208. properties['sla_enabled'] = node.sla_enabled
  1209. elif node.node_type == 'ssh':
  1210. properties['user'] = '%s@%s' % (node.user, node.host)
  1211. properties['ssh_command'] = node.command
  1212. properties['params'] = json.loads(node.params)
  1213. properties['capture_output'] = node.capture_output
  1214. properties['description'] = node.description
  1215. properties['sla'] = node.sla
  1216. properties['sla_enabled'] = node.sla_enabled
  1217. elif node.node_type == 'fs':
  1218. properties['deletes'] = [{'value': f['name']} for f in json.loads(node.deletes)]
  1219. properties['mkdirs'] = [{'value': f['name']} for f in json.loads(node.mkdirs)]
  1220. properties['moves'] = json.loads(node.moves)
  1221. chmods = json.loads(node.chmods)
  1222. for c in chmods:
  1223. c['value'] = c['path']
  1224. c['dir_files'] = False
  1225. properties['chmods'] = chmods
  1226. properties['touchzs'] = [{'value': f['name']} for f in json.loads(node.touchzs)]
  1227. properties['description'] = node.description
  1228. properties['sla'] = node.sla
  1229. properties['sla_enabled'] = node.sla_enabled
  1230. elif node.node_type == 'email':
  1231. properties['to'] = node.to
  1232. properties['cc'] = node.cc
  1233. properties['subject'] = node.subject
  1234. properties['body'] = node.body
  1235. properties['description'] = node.description
  1236. properties['sla'] = node.sla
  1237. properties['sla_enabled'] = node.sla_enabled
  1238. elif node.node_type == 'streaming':
  1239. properties['mapper'] = node.mapper
  1240. properties['reducer'] = node.reducer
  1241. properties['files'] = [{'value': f} for f in json.loads(node.files)]
  1242. properties['archives'] = json.loads(node.archives)
  1243. properties['job_properties'] = json.loads(node.job_properties)
  1244. properties['prepares'] = json.loads(node.prepares)
  1245. properties['job_xml'] = node.job_xml
  1246. properties['description'] = node.description
  1247. properties['sla'] = node.sla
  1248. properties['sla_enabled'] = node.sla_enabled
  1249. elif node.node_type == 'distcp':
  1250. properties['distcp_parameters'] = json.loads(node.params)
  1251. properties['java_opts'] = node.job_xml
  1252. properties['job_properties'] = json.loads(node.job_properties)
  1253. properties['prepares'] = json.loads(node.prepares)
  1254. properties['description'] = node.description
  1255. properties['sla'] = node.sla
  1256. properties['sla_enabled'] = node.sla_enabled
  1257. wf_nodes.append({
  1258. "id": uuids[node.id],
  1259. "name": '%s-%s' % (node.node_type.split('-')[0], uuids[node.id][:4]),
  1260. "type": "%s-widget" % node.node_type,
  1261. "properties": properties,
  1262. "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()]
  1263. })
  1264. else:
  1265. _dig_nodes(node)
  1266. _dig_nodes(old_nodes)
  1267. data['workflow']['nodes'] = wf_nodes
  1268. return Workflow(data=json.dumps(data))
  1269. class Coordinator(Job):
  1270. XML_FILE_NAME = 'coordinator.xml'
  1271. PROPERTY_APP_PATH = 'oozie.coord.application.path'
  1272. HUE_ID = 'hue-id-c'
  1273. def __init__(self, data=None, json_data=None, document=None):
  1274. self.document = document
  1275. if document is not None:
  1276. self._data = json.loads(document.data)
  1277. elif json_data is not None:
  1278. self._data = json.loads(json_data)
  1279. elif data is not None:
  1280. self._data = data
  1281. else:
  1282. self._data = {
  1283. 'id': None,
  1284. 'uuid': None,
  1285. 'name': 'My Coordinator',
  1286. 'variables': [], # Aka workflow parameters
  1287. 'properties': {
  1288. 'deployment_dir': '',
  1289. 'schema_version': 'uri:oozie:coordinator:0.2',
  1290. 'frequency_number': 1,
  1291. 'frequency_unit': 'days',
  1292. 'cron_frequency': '0 0 * * *',
  1293. 'cron_advanced': False,
  1294. 'timezone': 'America/Los_Angeles',
  1295. 'start': '${start_date}',
  1296. 'end': '${end_date}',
  1297. 'workflow': None,
  1298. 'timeout': None,
  1299. 'concurrency': None,
  1300. 'execution': None,
  1301. 'throttle': None,
  1302. 'job_xml': '',
  1303. 'sla_enabled': False,
  1304. 'sla_workflow_enabled': False,
  1305. 'credentials': [],
  1306. 'parameters': [
  1307. {'name': 'oozie.use.system.libpath', 'value': True},
  1308. {'name': 'start_date', 'value': datetime.today().strftime('%Y-%m-%dT%H:%M:%S')},
  1309. {'name': 'end_date', 'value': (datetime.today() + timedelta(days=7)).strftime('%Y-%m-%dT%H:%M:%S')}
  1310. ],
  1311. 'sla': Workflow.SLA_DEFAULT
  1312. }
  1313. }
  1314. @property
  1315. def id(self):
  1316. return self.document.id
  1317. @property
  1318. def uuid(self):
  1319. return self.document.uuid
  1320. def get_data_for_json(self):
  1321. _data = self.data.copy()
  1322. if type(self._data['properties']['start']) == datetime:
  1323. _data['properties']['start'] = _data['properties']['start'].strftime('%Y-%m-%dT%H:%M:%S')
  1324. if type(self._data['properties']['end']) == datetime:
  1325. _data['properties']['end'] = _data['properties']['end'].strftime('%Y-%m-%dT%H:%M:%S')
  1326. return _data
  1327. def to_json(self):
  1328. return json.dumps(self.get_data_for_json())
  1329. def to_json_for_html(self):
  1330. return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
  1331. @property
  1332. def data(self):
  1333. if type(self._data['properties']['start']) != datetime and not '$' in self._data['properties']['start']:
  1334. self._data['properties']['start'] = parse(self._data['properties']['start'])
  1335. if type(self._data['properties']['end']) != datetime and not '$' in self._data['properties']['end']:
  1336. self._data['properties']['end'] = parse(self._data['properties']['end'])
  1337. if self.document is not None:
  1338. self._data['id'] = self.document.id
  1339. return self._data
  1340. @property
  1341. def name(self):
  1342. return self.data['name']
  1343. def set_workspace(self, user):
  1344. self.data['properties']['deployment_dir'] = Job.get_workspace(user)
  1345. @property
  1346. def deployment_dir(self):
  1347. return self.data['properties']['deployment_dir']
  1348. def find_parameters(self):
  1349. params = set()
  1350. for param in find_json_parameters([self.data['properties']]):
  1351. params.add(param)
  1352. if self.sla_enabled:
  1353. for param in find_json_parameters(self.sla):
  1354. params.add(param)
  1355. # get missed params from wf
  1356. # for prop in self.workflow.get_parameters():
  1357. # if not prop['name'] in index:
  1358. # props.append(prop)
  1359. # index.append(prop['name'])
  1360. #
  1361. # # Remove DataInputs and DataOutputs
  1362. # datainput_names = [_input.name for _input in self.datainput_set.all()]
  1363. # dataoutput_names = [_output.name for _output in self.dataoutput_set.all()]
  1364. # removable_names = datainput_names + dataoutput_names
  1365. # props = filter(lambda prop: prop['name'] not in removable_names, props)
  1366. # get $params in wf properties
  1367. # [{'name': parameter['workflow_variable'], 'value': parameter['dataset_variable']} for parameter in self.data['variables'] if parameter['dataset_type'] == 'parameter']
  1368. return dict([(param, '') for param in list(params)])
  1369. @property
  1370. def sla_enabled(self):
  1371. return self.data['properties']['sla_enabled']
  1372. @property
  1373. def sla(self):
  1374. return self.data['properties']['sla']
  1375. @property
  1376. def parameters(self):
  1377. return self.data['properties']['parameters']
  1378. @property
  1379. def datasets(self):
  1380. return self.inputDatasets + self.outputDatasets
  1381. @property
  1382. def inputDatasets(self):
  1383. return [Dataset(dataset) for dataset in self.data['variables'] if dataset['dataset_type'] == 'input_path']
  1384. @property
  1385. def outputDatasets(self):
  1386. return [Dataset(dataset) for dataset in self.data['variables'] if dataset['dataset_type'] == 'output_path']
  1387. @property
  1388. def start_utc(self):
  1389. return utc_datetime_format(self.data['properties']['start'])
  1390. @property
  1391. def end_utc(self):
  1392. return utc_datetime_format(self.data['properties']['end'])
  1393. @property
  1394. def frequency(self):
  1395. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.data['properties']['frequency_unit'], 'number': self.data['properties']['frequency_number']}
  1396. @property
  1397. def cron_frequency(self):
  1398. data_dict = self.data['properties']
  1399. if 'cron_frequency' in data_dict:
  1400. return data_dict['cron_frequency']
  1401. else:
  1402. # Backward compatibility
  1403. freq = '0 0 * * *'
  1404. if data_dict['frequency_number'] == 1:
  1405. if data_dict['frequency_number'] == 'MINUTES':
  1406. freq = '* * * * *'
  1407. elif data_dict['frequency_number'] == 'HOURS':
  1408. freq = '0 * * * *'
  1409. elif data_dict['frequency_number'] == 'DAYS':
  1410. freq = '0 0 * * *'
  1411. elif data_dict['frequency_number'] == 'MONTH':
  1412. freq = '0 0 * * *'
  1413. return {'frequency': freq, 'isAdvancedCron': False}
  1414. def to_xml(self, mapping=None):
  1415. if mapping is None:
  1416. mapping = {}
  1417. tmpl = "editor/gen2/coordinator.xml.mako"
  1418. return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
  1419. @property
  1420. def properties(self):
  1421. props = [{'name': dataset['workflow_variable'], 'value': dataset['dataset_variable']} for dataset in self.data['variables'] if dataset['dataset_type'] == 'parameter']
  1422. props += self.data['properties']['properties']
  1423. return props
  1424. class Dataset():
  1425. def __init__(self, data):
  1426. self._data = data
  1427. @property
  1428. def data(self):
  1429. if type(self._data['start']) == unicode:
  1430. self._data['start'] = parse(self._data['start'])
  1431. self._data['name'] = self._data['workflow_variable']
  1432. return self._data
  1433. @property
  1434. def frequency(self):
  1435. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.data['frequency_unit'], 'number': self.data['frequency_number']}
  1436. @property
  1437. def start_utc(self):
  1438. return utc_datetime_format(self.data['start'])
  1439. @property
  1440. def start_instance(self):
  1441. if not self.is_advanced_start_instance:
  1442. return int(self.data['advanced_start_instance'])
  1443. else:
  1444. return 0
  1445. @property
  1446. def is_advanced_start_instance(self):
  1447. return not self.is_int(self.data['advanced_start_instance'])
  1448. def is_int(self, text):
  1449. try:
  1450. int(text)
  1451. return True
  1452. except ValueError:
  1453. return False
  1454. @property
  1455. def end_instance(self):
  1456. if not self.is_advanced_end_instance:
  1457. return int(self.data['advanced_end_instance'])
  1458. else:
  1459. return 0
  1460. @property
  1461. def is_advanced_end_instance(self):
  1462. return not self.is_int(self.data['advanced_end_instance'])
  1463. class Bundle(Job):
  1464. XML_FILE_NAME = 'bundle.xml'
  1465. PROPERTY_APP_PATH = 'oozie.bundle.application.path'
  1466. HUE_ID = 'hue-id-b'
  1467. def __init__(self, data=None, json_data=None, document=None):
  1468. self.document = document
  1469. if document is not None:
  1470. self._data = json.loads(document.data)
  1471. elif json_data is not None:
  1472. self._data = json.loads(json_data)
  1473. elif data is not None:
  1474. self._data = data
  1475. else:
  1476. self._data = {
  1477. 'id': None,
  1478. 'uuid': None,
  1479. 'name': 'My Bundle',
  1480. 'coordinators': [],
  1481. 'properties': {
  1482. 'deployment_dir': '',
  1483. 'schema_version': 'uri:oozie:bundle:0.2',
  1484. 'kickoff': datetime.today(),
  1485. 'parameters': [{'name': 'oozie.use.system.libpath', 'value': True}]
  1486. }
  1487. }
  1488. @property
  1489. def id(self):
  1490. return self.document.id
  1491. @property
  1492. def uuid(self):
  1493. return self.document.uuid
  1494. def get_data_for_json(self):
  1495. _data = self.data.copy()
  1496. _data['properties']['kickoff'] = _data['properties']['kickoff'].strftime('%Y-%m-%dT%H:%M:%S')
  1497. return _data
  1498. def to_json(self):
  1499. return json.dumps(self.get_data_for_json())
  1500. def to_json_for_html(self):
  1501. return json.dumps(self.get_data_for_json(), cls=JSONEncoderForHTML)
  1502. @property
  1503. def data(self):
  1504. if type(self._data['properties']['kickoff']) == unicode:
  1505. self._data['properties']['kickoff'] = parse(self._data['properties']['kickoff'])
  1506. if self.document is not None:
  1507. self._data['id'] = self.document.id
  1508. return self._data
  1509. def to_xml(self, mapping=None):
  1510. if mapping is None:
  1511. mapping = {}
  1512. mapping.update(dict(list(Document2.objects.filter(type='oozie-coordinator2', uuid__in=self.data['coordinators']).values('uuid', 'name'))))
  1513. tmpl = "editor/gen2/bundle.xml.mako"
  1514. return force_unicode(
  1515. re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
  1516. 'bundle': self,
  1517. 'mapping': mapping
  1518. })))
  1519. @property
  1520. def name(self):
  1521. return self.data['name']
  1522. @property
  1523. def parameters(self):
  1524. return self.data['properties']['parameters']
  1525. @property
  1526. def kick_off_time_utc(self):
  1527. return utc_datetime_format(self.data['properties']['kickoff'])
  1528. def set_workspace(self, user):
  1529. self.data['properties']['deployment_dir'] = Job.get_workspace(user)
  1530. @property
  1531. def deployment_dir(self):
  1532. return self.data['properties']['deployment_dir']
  1533. def find_parameters(self):
  1534. return {}