models.py 73 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845
  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 StringIO
  21. import time
  22. import zipfile
  23. from datetime import datetime, timedelta
  24. from string import Template
  25. from itertools import chain
  26. from django.db import models
  27. from django.db.models import Q
  28. from django.core.urlresolvers import reverse
  29. from django.core.validators import RegexValidator
  30. from django.contrib.auth.models import User
  31. from django.contrib.contenttypes import generic
  32. from django.contrib.contenttypes.models import ContentType
  33. from django.forms.models import inlineformset_factory
  34. from django.utils.encoding import force_unicode, smart_str
  35. from django.utils.translation import ugettext as _, ugettext_lazy as _t
  36. from desktop.log.access import access_warn
  37. from desktop.lib import django_mako
  38. from desktop.lib.exceptions_renderable import PopupException
  39. from desktop.lib.json_utils import JSONEncoderForHTML
  40. from desktop.models import Document
  41. from hadoop.fs.exceptions import WebHdfsException
  42. from hadoop.fs.hadoopfs import Hdfs
  43. from liboozie.conf import SECURITY_ENABLED
  44. from liboozie.submittion import Submission
  45. from liboozie.submittion import create_directories
  46. from oozie.conf import REMOTE_SAMPLE_DIR
  47. from oozie.utils import utc_datetime_format
  48. from oozie.timezones import TIMEZONES
  49. LOG = logging.getLogger(__name__)
  50. PATH_MAX = 512
  51. name_validator = RegexValidator(regex='^[a-zA-Z_][\-_a-zA-Z0-9]{1,39}$',
  52. message=_('Enter a valid value: combination of 2 - 40 letters and digits starting by a letter'))
  53. class JobManager(models.Manager):
  54. def can_read(self, user, job_id):
  55. job = Job.objects.select_related().get(pk=job_id).get_full_node()
  56. return job.can_read(user)
  57. def can_read_or_exception(self, request, job_id, exception_class=PopupException):
  58. if job_id is None:
  59. return
  60. try:
  61. job = Job.objects.select_related().get(pk=job_id).get_full_node()
  62. if job.can_read(request.user):
  63. return job
  64. else:
  65. message = _("Permission denied. %(username)s does not have the permissions required to access job %(id)s") % \
  66. {'username': request.user.username, 'id': job.id}
  67. access_warn(request, message)
  68. request.error(message)
  69. raise exception_class(message)
  70. except Job.DoesNotExist:
  71. raise exception_class(_('job %(id)s does not exist') % {'id': job_id})
  72. def can_edit_or_exception(self, request, job, exception_class=PopupException):
  73. if job.is_editable(request.user):
  74. return True
  75. else:
  76. raise exception_class(_('Not allowed to modified this job'))
  77. class Job(models.Model):
  78. """
  79. Base class for Oozie Workflows, Coordinators and Bundles.
  80. """
  81. owner = models.ForeignKey(User, db_index=True, verbose_name=_t('Owner'), help_text=_t('Person who can modify the job.')) # Deprecated
  82. name = models.CharField(max_length=40, blank=False, validators=[name_validator], # Deprecated
  83. help_text=_t('Name of the job, which must be unique per user.'), verbose_name=_t('Name'))
  84. description = models.CharField(max_length=1024, blank=True, verbose_name=_t('Description'), # Deprecated
  85. help_text=_t('The purpose of the job.'))
  86. last_modified = models.DateTimeField(auto_now=True, db_index=True, verbose_name=_t('Last modified'))
  87. schema_version = models.CharField(max_length=128, verbose_name=_t('Schema version'),
  88. help_text=_t('The version of the XML schema used to talk to Oozie.'))
  89. deployment_dir = models.CharField(max_length=1024, blank=True, verbose_name=_t('HDFS deployment directory'),
  90. help_text=_t('The path on the HDFS where all the workflows and '
  91. 'dependencies must be uploaded.'))
  92. is_shared = models.BooleanField(default=False, db_index=True, verbose_name=_t('Is shared'), # Deprecated
  93. help_text=_t('Enable other users to have access to this job.'))
  94. parameters = models.TextField(default='[{"name":"oozie.use.system.libpath","value":"true"}]', verbose_name=_t('Oozie parameters'),
  95. help_text=_t('Parameters used at the submission time (e.g. market=US, oozie.use.system.libpath=true).'))
  96. is_trashed = models.BooleanField(default=False, db_index=True, verbose_name=_t('Is trashed'), blank=True,# Deprecated
  97. help_text=_t('If this job is trashed.'))
  98. doc = generic.GenericRelation(Document, related_name='oozie_doc')
  99. objects = JobManager()
  100. unique_together = ('owner', 'name')
  101. def delete(self, skip_trash=False, *args, **kwargs):
  102. if skip_trash:
  103. self.doc.all().delete()
  104. return super(Job, self).delete(*args, **kwargs)
  105. else:
  106. for job in self.doc.all():
  107. job.send_to_trash()
  108. return self
  109. def restore(self):
  110. self.doc.get().restore_from_trash()
  111. return self
  112. def save(self):
  113. super(Job, self).save()
  114. if not self.deployment_dir:
  115. default_dir = Hdfs.join(REMOTE_SAMPLE_DIR.get(), '_%s_-oozie-%s-%s' % (self.owner.username, self.id, time.time()))
  116. self.deployment_dir = default_dir
  117. super(Job, self).save()
  118. def is_deployed(self, fs):
  119. return self.deployment_dir != '' and fs.exists(self.deployment_dir)
  120. def __str__(self):
  121. res = '%s - %s' % (force_unicode(self.name), self.owner)
  122. return force_unicode(res)
  123. def get_full_node(self):
  124. try:
  125. return self.workflow
  126. except Workflow.DoesNotExist:
  127. pass
  128. try:
  129. return self.coordinator
  130. except Coordinator.DoesNotExist:
  131. pass
  132. try:
  133. return self.bundle
  134. except Bundle.DoesNotExist:
  135. pass
  136. def get_type(self):
  137. return self.get_full_node().get_type()
  138. def get_absolute_url(self):
  139. return self.get_full_node().get_absolute_url()
  140. def get_parameters(self):
  141. return json.loads(self.parameters)
  142. @property
  143. def parameters_escapejs(self):
  144. return self._escapejs_parameters_list(self.parameters)
  145. def _escapejs_parameters_list(self, parameters):
  146. return json.dumps(json.loads(parameters), cls=JSONEncoderForHTML)
  147. @property
  148. def status(self):
  149. # TODO
  150. if self.is_shared:
  151. return _('shared')
  152. else:
  153. return _('personal')
  154. def find_all_parameters(self):
  155. params = self.find_parameters()
  156. for param in self.get_parameters():
  157. params[param['name'].strip()] = param['value']
  158. return [{'name': name, 'value': value} for name, value in params.iteritems()]
  159. def can_read(self, user):
  160. try:
  161. return self.doc.get().can_read(user)
  162. except Exception, e:
  163. LOG.error('can_read failed because the object has more than one document: %s' % self.doc.all())
  164. raise e
  165. def is_editable(self, user):
  166. """Only owners or admins can modify a job."""
  167. return user.is_superuser or self.owner == user
  168. class WorkflowManager(models.Manager):
  169. def new_workflow(self, owner):
  170. workflow = Workflow(owner=owner, schema_version='uri:oozie:workflow:0.4')
  171. kill = Kill(name='kill', workflow=workflow, node_type=Kill.node_type)
  172. end = End(name='end', workflow=workflow, node_type=End.node_type)
  173. start = Start(name='start', workflow=workflow, node_type=Start.node_type)
  174. to = Link(parent=start, child=end, name='to')
  175. related = Link(parent=start, child=end, name='related')
  176. workflow.start = start
  177. workflow.end = end
  178. return workflow
  179. def initialize(self, workflow, fs=None):
  180. Kill.objects.create(name='kill', workflow=workflow, node_type=Kill.node_type)
  181. end = End.objects.create(name='end', workflow=workflow, node_type=End.node_type)
  182. start = Start.objects.create(name='start', workflow=workflow, node_type=Start.node_type)
  183. link = Link(parent=start, child=end, name='to')
  184. link.save()
  185. Link.objects.create(parent=start, child=end, name='related')
  186. workflow.start = start
  187. workflow.end = end
  188. workflow.save()
  189. Document.objects.link(workflow, owner=workflow.owner, name=workflow.name, description=workflow.description)
  190. if fs:
  191. self.check_workspace(workflow, fs)
  192. def check_workspace(self, workflow, fs):
  193. create_directories(fs, [REMOTE_SAMPLE_DIR.get()])
  194. create_directories(fs)
  195. if workflow.is_shared:
  196. perms = 0755
  197. else:
  198. perms = 0711
  199. Submission(workflow.owner, workflow, fs, None, {})._create_dir(workflow.deployment_dir, perms=perms)
  200. def destroy(self, workflow, fs):
  201. Submission(workflow.owner, workflow, fs, None, {}).remove_deployment_dir()
  202. try:
  203. workflow.coordinator_set.update(workflow=None) # In Django 1.3 could do ON DELETE set NULL
  204. except:
  205. pass
  206. workflow.save()
  207. workflow.delete(skip_trash=True)
  208. def managed(self):
  209. return self.filter(managed=True)
  210. def unmanaged(self):
  211. return self.filter(managed=False)
  212. class Workflow(Job):
  213. is_single = models.BooleanField(default=False)
  214. start = models.ForeignKey('Start', related_name='start_workflow', blank=True, null=True)
  215. end = models.ForeignKey('End', related_name='end_workflow', blank=True, null=True)
  216. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  217. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  218. 'Properties specified in the Job Properties element override properties specified in the '
  219. 'files specified in the Job XML element.'))
  220. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  221. help_text=_t('Job configuration properties used by all the actions of the workflow '
  222. '(e.g. mapred.job.queue.name=production)'))
  223. managed = models.BooleanField(default=True)
  224. objects = WorkflowManager()
  225. HUE_ID = 'hue-id-w'
  226. ICON = '/oozie/static/art/icon_oozie_workflow_24.png'
  227. METADATA_FORMAT_VERSION = "0.0.1"
  228. def get_type(self):
  229. return 'workflow'
  230. def get_properties(self):
  231. return json.loads(self.job_properties)
  232. def clone(self, fs, new_owner=None):
  233. source_deployment_dir = self.deployment_dir # Needed
  234. nodes = self.node_set.all()
  235. copy_doc = self.doc.get().copy()
  236. links = Link.objects.filter(parent__workflow=self)
  237. copy = self
  238. copy.pk = None
  239. copy.id = None
  240. copy.name += '-copy'
  241. copy.deployment_dir = ''
  242. if new_owner is not None:
  243. copy.owner = new_owner
  244. copy.save()
  245. old_nodes_mapping = {}
  246. for node in nodes:
  247. prev_id = node.id
  248. node = node.get_full_node()
  249. node.pk = None
  250. node.id = None
  251. node.workflow = copy
  252. node.save()
  253. old_nodes_mapping[prev_id] = node
  254. for link in links:
  255. link.pk = None
  256. link.id = None
  257. link.parent = old_nodes_mapping[link.parent.id]
  258. link.child = old_nodes_mapping[link.child.id]
  259. link.save()
  260. copy.start = old_nodes_mapping[self.start.id]
  261. copy.end = old_nodes_mapping[self.end.id]
  262. copy.save()
  263. copy_doc.name = copy.name
  264. copy_doc.owner = copy.owner
  265. copy_doc.save()
  266. copy.doc.add(copy_doc)
  267. try:
  268. if copy.is_shared:
  269. perms = 0755
  270. else:
  271. perms = 0711
  272. fs.copy_remote_dir(source_deployment_dir, copy.deployment_dir, owner=copy.owner, dir_mode=perms)
  273. except WebHdfsException, e:
  274. msg = _('The copy of the deployment directory failed: %s.') % e
  275. LOG.error(msg)
  276. raise PopupException(msg)
  277. # Reload workflow from DB... clears relationship cache
  278. copy = Workflow.objects.get(id=copy.id)
  279. return copy
  280. @property
  281. def job_properties_escapejs(self):
  282. return self._escapejs_parameters_list(self.job_properties)
  283. def has_cycle(self):
  284. """
  285. Topological sort for detecting cycles in the directed graph.
  286. """
  287. queue = set([self.start])
  288. removed_edges = set()
  289. while queue:
  290. node = queue.pop()
  291. edges = set(node.get_children_links())
  292. for edge in edges:
  293. removed_edges.add(edge)
  294. # Edge has no other incoming edges
  295. if not set(edge.child.get_parent_links()) - removed_edges:
  296. queue.add(edge.child)
  297. graph_edges = set([edge for node in self.node_set.all() for edge in node.get_children_links()])
  298. return len(graph_edges - removed_edges) > 0 # Graph does not have unseen edges
  299. def find_parameters(self):
  300. params = set()
  301. for node in self.node_list:
  302. if hasattr(node, 'find_parameters'):
  303. params.update(node.find_parameters())
  304. return dict([(param, '') for param in list(params)])
  305. @property
  306. def actions(self):
  307. return Action.objects.filter(workflow=self, node_type__in=Action.types)
  308. @property
  309. def node_list(self):
  310. """Return a flatten node list ordered by the hierarchy of the nodes in the workflow"""
  311. def flatten(nodes):
  312. flat = []
  313. if type(nodes) == list:
  314. for node in nodes:
  315. flat.extend(flatten(node))
  316. else:
  317. flat.append(nodes)
  318. return flat
  319. def from_iterable(iterables):
  320. # Python 2.6 chain.from_iterable(['ABC', 'DEF']) --> A B C D E F
  321. for it in iterables:
  322. for element in it:
  323. yield element
  324. return list(chain(from_iterable([flatten(row) for row in self.get_hierarchy()])))
  325. @classmethod
  326. def get_application_path_key(cls):
  327. return 'oozie.wf.application.path'
  328. @classmethod
  329. def get_application_filename(cls):
  330. return 'workflow.xml'
  331. def get_absolute_url(self):
  332. if self.doc.get().extra == 'jobsub':
  333. return '/jobsub/#edit-design/%s' % self.id
  334. else:
  335. return reverse('oozie:edit_workflow', kwargs={'workflow': self.id}) + '#editWorkflow'
  336. def get_hierarchy(self):
  337. node = Start.objects.get(workflow=self) # Uncached version of start.
  338. kill = Kill.objects.get(workflow=node.workflow)
  339. # Special case: manage error email actions separately
  340. try:
  341. kill_nodes = [Link.objects.filter(child=kill).get(name='ok').parent, kill]
  342. except Link.DoesNotExist:
  343. kill_nodes = [kill]
  344. return self.get_hierarchy_rec(node=node) + [kill_nodes, [End.objects.get(workflow=node.workflow)]]
  345. def get_hierarchy_rec(self, node=None):
  346. if node is None:
  347. node = self.start
  348. if node.id is None:
  349. return []
  350. node = node.get_full_node()
  351. parents = node.get_parents()
  352. if isinstance(node, End):
  353. return [] # Not returning the end node
  354. elif isinstance(node, Decision):
  355. children = node.get_children('start')
  356. return [[node] + [[self.get_hierarchy_rec(node=child) for child in children],
  357. node.get_child_end()]] + self.get_hierarchy_rec(node.get_child_end().get_child('to'))
  358. elif isinstance(node, DecisionEnd):
  359. return []
  360. elif isinstance(node, Fork):
  361. children = node.get_children('start')
  362. return [[node] + [[self.get_hierarchy_rec(node=child) for child in children],
  363. node.get_child_join()]] + self.get_hierarchy_rec(node.get_child_join().get_child('to'))
  364. elif isinstance(node, Join):
  365. return []
  366. else:
  367. child = Link.objects.filter(parent=node).exclude(name__in=['related', 'kill', 'error'])[0].child
  368. return [node] + self.get_hierarchy_rec(child)
  369. def gen_status_graph(self, oozie_workflow):
  370. from oozie.forms import NodeMetaForm # Circular dependency
  371. actions = oozie_workflow.get_working_actions()
  372. controls = oozie_workflow.get_control_flow_actions()
  373. WorkflowFormSet = inlineformset_factory(Workflow, Node, form=NodeMetaForm, max_num=0, can_order=False, can_delete=False)
  374. forms = WorkflowFormSet(instance=self).forms
  375. template = 'editor/gen/workflow-graph-status.xml.mako'
  376. index = dict([(form.instance.id, form) for form in forms])
  377. actions_index = dict([(action.name, action) for action in actions])
  378. controls_index = dict([(control.name.strip(':'), control) for control in controls])
  379. return django_mako.render_to_string(template, {'nodes': self.get_hierarchy(), 'index': index, 'actions': actions_index, 'controls': controls_index})
  380. @classmethod
  381. def gen_status_graph_from_xml(cls, user, oozie_workflow):
  382. from oozie.importlib.workflows import import_workflow # Circular dependency
  383. try:
  384. workflow = Workflow.objects.new_workflow(user)
  385. workflow.save()
  386. try:
  387. import_workflow(workflow, oozie_workflow.definition)
  388. graph = workflow.gen_status_graph(oozie_workflow)
  389. return graph, workflow.node_list
  390. except Exception, e:
  391. LOG.warn('Workflow %s could not be converted to a graph: %s' % (oozie_workflow.id, e))
  392. finally:
  393. workflow.delete()
  394. return None, []
  395. def to_xml(self, mapping=None):
  396. if mapping is None:
  397. mapping = {}
  398. tmpl = 'editor/gen/workflow.xml.mako'
  399. xml = re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'workflow': self, 'mapping': mapping}))
  400. return force_unicode(xml)
  401. def compress(self, mapping=None, fp=StringIO.StringIO()):
  402. metadata = {
  403. 'version': Workflow.METADATA_FORMAT_VERSION,
  404. 'nodes': {},
  405. 'attributes': {
  406. 'description': self.description,
  407. 'deployment_dir': self.deployment_dir
  408. }
  409. }
  410. for node in self.node_list:
  411. if hasattr(node, 'jar_path'):
  412. metadata['nodes'][node.name] = {
  413. 'attributes': {
  414. 'jar_path': node.jar_path
  415. }
  416. }
  417. xml = self.to_xml(mapping=mapping)
  418. zfile = zipfile.ZipFile(fp, 'w')
  419. zfile.writestr("workflow.xml", smart_str(xml))
  420. zfile.writestr("workflow-metadata.json", smart_str(json.dumps(metadata)))
  421. zfile.close()
  422. return fp
  423. @classmethod
  424. def decompress(cls, fp):
  425. zfile = zipfile.ZipFile(fp, 'r')
  426. metadata_json = zfile.read('workflow-metadata.json')
  427. metadata = json.loads(metadata_json)
  428. workflow_xml = zfile.read('workflow.xml')
  429. return workflow_xml, metadata
  430. class Link(models.Model):
  431. # Links to exclude when using get_children_link(), get_parent_links() in the API
  432. META_LINKS = ('related',)
  433. parent = models.ForeignKey('Node', related_name='child_node')
  434. child = models.ForeignKey('Node', related_name='parent_node', verbose_name='')
  435. name = models.CharField(max_length=40)
  436. comment = models.CharField(max_length=1024, default='', blank=True)
  437. def __unicode__(self):
  438. return '%s %s %s' % (self.parent, self.child, self.name)
  439. class Node(models.Model):
  440. """
  441. Base class for the Oozie WorkflowAction or ControlFlow Nodes.
  442. http://nightly.cloudera.com/cdh4/cdh/4/oozie-3.1.3-cdh4.0.0-SNAPSHOT/WorkflowFunctionalSpec.html#a3_Workflow_Nodes
  443. The Node model is an abstract base class. All concrete actions derive from it.
  444. And it provides something for the Action or ControlFlow to reference.
  445. See https://docs.djangoproject.com/en/dev/topics/db/models/#multi-table-inheritance
  446. """
  447. PARAM_FIELDS = ()
  448. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name'),
  449. help_text=_t('Name of the action, which must be unique by workflow.'))
  450. description = models.CharField(max_length=1024, blank=True, default='', verbose_name=_t('Description'),
  451. help_text=_t('The purpose of the action.'))
  452. node_type = models.CharField(max_length=64, blank=False, verbose_name=_t('Type'),
  453. help_text=_t('The type of action (e.g. MapReduce, Pig...)'))
  454. workflow = models.ForeignKey(Workflow)
  455. children = models.ManyToManyField('self', related_name='parents', symmetrical=False, through=Link)
  456. unique_together = ('workflow', 'name')
  457. def get_full_node(self):
  458. if self.node_type == Mapreduce.node_type:
  459. node = self.mapreduce
  460. elif self.node_type == Pig.node_type:
  461. node = self.pig
  462. elif self.node_type == Hive.node_type:
  463. node = self.hive
  464. elif self.node_type == Sqoop.node_type:
  465. node = self.sqoop
  466. elif self.node_type == Ssh.node_type:
  467. node = self.ssh
  468. elif self.node_type == Shell.node_type:
  469. node = self.shell
  470. elif self.node_type == DistCp.node_type:
  471. node = self.distcp
  472. elif self.node_type == Fs.node_type:
  473. node = self.fs
  474. elif self.node_type == Email.node_type:
  475. node = self.email
  476. elif self.node_type == SubWorkflow.node_type:
  477. node = self.subworkflow
  478. elif self.node_type == Streaming.node_type:
  479. node = self.streaming
  480. elif self.node_type == Java.node_type:
  481. node = self.java
  482. elif self.node_type == Generic.node_type:
  483. node = self.generic
  484. elif self.node_type == Start.node_type:
  485. node = self.start
  486. elif self.node_type == End.node_type:
  487. node = self.end
  488. elif self.node_type == Kill.node_type:
  489. node = self.kill
  490. elif self.node_type == Fork.node_type:
  491. node = self.fork
  492. elif self.node_type == Join.node_type:
  493. node = self.join
  494. elif self.node_type == Decision.node_type:
  495. node = self.decision
  496. elif self.node_type == DecisionEnd.node_type:
  497. node = self.decisionend
  498. else:
  499. raise Exception(_('Unknown Node type: %s. Was it set at its creation?'), (self.node_type,))
  500. return node
  501. def find_parameters(self):
  502. return find_parameters(self, self.PARAM_FIELDS)
  503. def __unicode__(self):
  504. if self.name != '':
  505. return '%s' % self.name
  506. else:
  507. return '%s-%s' % (self.node_type, self.id)
  508. def to_xml(self, mapping=None):
  509. if mapping is None:
  510. mapping = {}
  511. node = self.get_full_node()
  512. data = {
  513. 'node': node,
  514. 'mapping': mapping
  515. }
  516. return django_mako.render_to_string(node.get_template_name(), data)
  517. # Can't use through relation directly with this Django version?
  518. # https://docs.djangoproject.com/en/1.2/topics/db/models/#intermediary-manytomany
  519. def get_link(self, name=None):
  520. if name is None:
  521. return Link.objects.exclude(name__in=Link.META_LINKS).get(parent=self)
  522. else:
  523. return Link.objects.exclude(name__in=Link.META_LINKS).get(parent=self, name=name)
  524. def get_child_link(self, name=None):
  525. return self.get_link(name)
  526. def get_child(self, name=None):
  527. """Includes DecisionEnd nodes"""
  528. return self.get_link(name).child.get_full_node()
  529. def get_oozie_child(self, name=None):
  530. """Resolves DecisionEnd nodes"""
  531. child = self.get_link(name).child.get_full_node()
  532. if child and child.node_type == DecisionEnd.node_type:
  533. child = child.get_oozie_child('to')
  534. return child
  535. def get_children(self, name=None):
  536. if name is not None:
  537. return [link.child for link in Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self, name=name)]
  538. else:
  539. return [link.child for link in Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self)]
  540. def get_parent(self, name=None):
  541. if name is not None:
  542. return self.get_parent_link(name).parent.get_full_node()
  543. else:
  544. return self.get_parent_link().parent.get_full_node()
  545. def get_parents(self):
  546. return [link.parent for link in self.get_parent_links()]
  547. def get_parent_link(self, name=None):
  548. if name is not None:
  549. return Link.objects.get(child=self, name=name)
  550. else:
  551. return Link.objects.get(child=self)
  552. def get_parent_links(self):
  553. return Link.objects.filter(child=self).exclude(name__in=Link.META_LINKS)
  554. def get_children_links(self, name=None):
  555. if name is None:
  556. return Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self)
  557. else:
  558. return Link.objects.exclude(name__in=Link.META_LINKS).filter(parent=self, name=name)
  559. def get_all_children_links(self):
  560. return Link.objects.filter(parent=self)
  561. def get_template_name(self):
  562. return 'editor/gen/workflow-%s.xml.mako' % self.node_type
  563. def is_visible(self):
  564. return True
  565. def add_node(self, child):
  566. raise NotImplementedError(_("%(node_type)s has not implemented the 'add_node' method.") % {
  567. 'node_type': self.node_type
  568. })
  569. class Action(Node):
  570. types = ()
  571. class Meta:
  572. # Cloning does not work anymore if not abstract
  573. abstract = True
  574. def add_node(self, child):
  575. Link.objects.filter(parent=self, name='ok').delete()
  576. Link.objects.create(parent=self, child=child, name='ok')
  577. if not Link.objects.filter(parent=self, name='error').exists():
  578. Link.objects.create(parent=self, child=Kill.objects.get(name='kill', workflow=self.workflow), name='error')
  579. # The fields with '[]' as default value are JSON dictionaries
  580. # When adding a new action, also update
  581. # - Action.types below
  582. # - Node.get_full_node()
  583. # - forms.py _node_type_TO_FORM_CLS
  584. # - workflow.js
  585. # - maybe actions_utils.mako
  586. class Mapreduce(Action):
  587. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'jar_path', 'prepares')
  588. node_type = 'mapreduce'
  589. files = models.TextField(default="[]", verbose_name=_t('Files'),
  590. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  591. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  592. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  593. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  594. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  595. jar_path = models.CharField(max_length=PATH_MAX, verbose_name=_t('Jar name'),
  596. help_text=_t('Name or path to the %(program)s jar file on HDFS. E.g. examples.jar.') % {'program': 'MapReduce'})
  597. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  598. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  599. 'This should be used exclusively for directory cleanup.'))
  600. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  601. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  602. 'Properties specified in the Job Properties element override properties specified in the '
  603. 'files specified in the Job XML element.'))
  604. def get_properties(self):
  605. return json.loads(self.job_properties)
  606. def get_files(self):
  607. return json.loads(self.files)
  608. def get_archives(self):
  609. return json.loads(self.archives)
  610. def get_prepares(self):
  611. return json.loads(self.prepares)
  612. class Streaming(Action):
  613. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'mapper', 'reducer')
  614. node_type = "streaming"
  615. files = models.TextField(default="[]", verbose_name=_t('Files'),
  616. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  617. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  618. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  619. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  620. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  621. mapper = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Mapper'),
  622. help_text=_t('The executable/script to be used as mapper.'))
  623. reducer = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Reducer'),
  624. help_text=_t('The executable/script to be used as reducer.'))
  625. def get_properties(self):
  626. return json.loads(self.job_properties)
  627. def get_files(self):
  628. return json.loads(self.files)
  629. def get_archives(self):
  630. return json.loads(self.archives)
  631. class Java(Action):
  632. PARAM_FIELDS = ('files', 'archives', 'jar_path', 'main_class', 'args',
  633. 'java_opts', 'job_properties', 'prepares')
  634. node_type = "java"
  635. files = models.TextField(default="[]", verbose_name=_t('Files'),
  636. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  637. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  638. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  639. jar_path = models.CharField(max_length=PATH_MAX, blank=False, verbose_name=_t('Jar name'),
  640. help_text=_t('Name or path to the %(program)s jar file on HDFS. E.g. examples.jar.') % {'program': 'Java'})
  641. main_class = models.CharField(max_length=256, blank=False, verbose_name=_t('Main class'),
  642. help_text=_t('Full name of the Java class. E.g. org.apache.hadoop.examples.Grep'))
  643. args = models.TextField(blank=True, verbose_name=_t('Arguments'),
  644. help_text=_t('Arguments of the main method. The value of each arg element is considered a single argument '
  645. 'and they are passed to the main method in the same order.'))
  646. java_opts = models.CharField(max_length=256, blank=True, verbose_name=_t('Java options'),
  647. help_text=_t('Command-line parameters used to start the JVM that will execute '
  648. 'the Java application. Using this element is equivalent to using the mapred.child.java.opts '
  649. 'configuration property. E.g. -Dexample-property=hue'))
  650. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  651. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  652. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  653. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  654. 'This should be used exclusively for directory cleanup.'))
  655. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  656. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  657. 'Properties specified in the Job Properties element override properties specified in the '
  658. 'files specified in the Job XML element.'))
  659. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  660. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  661. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  662. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  663. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  664. def get_properties(self):
  665. return json.loads(self.job_properties)
  666. def get_files(self):
  667. return json.loads(self.files)
  668. def get_archives(self):
  669. return json.loads(self.archives)
  670. def get_prepares(self):
  671. return json.loads(self.prepares)
  672. class Pig(Action):
  673. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  674. node_type = 'pig'
  675. script_path = models.CharField(max_length=256, blank=False, verbose_name=_t('Script name'),
  676. help_text=_t('Script name or path to the Pig script. E.g. my_script.pig.'))
  677. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  678. help_text=_t('The Pig parameters of the script. e.g. "-param", "INPUT=${inputDir}"'))
  679. files = models.TextField(default="[]", verbose_name=_t('Files'),
  680. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  681. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  682. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  683. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  684. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  685. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  686. help_text=_t('List of absolute paths to delete and then to create before starting the application. '
  687. 'This should be used exclusively for directory cleanup.'))
  688. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  689. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  690. 'Properties specified in the Job Properties element override properties specified in the '
  691. 'files specified in the Job XML element.'))
  692. def get_properties(self):
  693. return json.loads(self.job_properties)
  694. def get_files(self):
  695. return json.loads(self.files)
  696. def get_archives(self):
  697. return json.loads(self.archives)
  698. def get_params(self):
  699. return json.loads(self.params)
  700. def get_prepares(self):
  701. return json.loads(self.prepares)
  702. class Hive(Action):
  703. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  704. node_type = 'hive'
  705. script_path = models.CharField(max_length=256, blank=False, verbose_name=_t('Script name'),
  706. help_text=_t('Script name or path to the %(type)s script. E.g. my_script.sql.') % {'type': node_type.title()})
  707. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  708. help_text=_t('The %(type)s parameters of the script. E.g. N=5, INPUT=${inputDir}') % {'type': node_type.title()})
  709. files = models.TextField(default="[]", verbose_name=_t('Files'),
  710. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  711. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  712. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  713. job_properties = models.TextField(default='[{"name":"oozie.hive.defaults","value":"hive-site.xml"}]',
  714. verbose_name=_t('Hadoop job properties'),
  715. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  716. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  717. help_text=_t('List of absolute paths to delete, then create, before starting the application. '
  718. 'This should be used exclusively for directory cleanup.'))
  719. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  720. help_text=_t('Refer to a Hive hive-site.xml file bundled in the workflow deployment directory. '))
  721. def get_properties(self):
  722. return json.loads(self.job_properties)
  723. def get_files(self):
  724. return json.loads(self.files)
  725. def get_archives(self):
  726. return json.loads(self.archives)
  727. def get_params(self):
  728. return json.loads(self.params)
  729. def get_prepares(self):
  730. return json.loads(self.prepares)
  731. class Sqoop(Action):
  732. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  733. node_type = 'sqoop'
  734. script_path = models.TextField(blank=True, verbose_name=_t('Command'), default='',
  735. help_text=_t('The full %(type)s command. Either put it here or split it by spaces and insert the parts as multiple parameters below.')
  736. % {'type': node_type.title()})
  737. params = models.TextField(default="[]", verbose_name=_t('Parameters'),
  738. help_text=_t('If no command is specified, split the command by spaces and insert the %(type)s parameters '
  739. 'here e.g. import, --connect, jdbc:hsqldb:file:db.hsqldb, ...') % {'type': node_type.title()})
  740. files = models.TextField(default="[]", verbose_name=_t('Files'),
  741. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  742. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  743. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  744. job_properties = models.TextField(default='[]',
  745. verbose_name=_t('Hadoop job properties'),
  746. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  747. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  748. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  749. 'This should be used exclusively for directory cleanup'))
  750. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  751. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  752. 'Properties specified in the Job Properties element override properties specified in the '
  753. 'files specified in the Job XML element.'))
  754. def get_properties(self):
  755. return json.loads(self.job_properties)
  756. def get_files(self):
  757. return json.loads(self.files)
  758. def get_archives(self):
  759. return json.loads(self.archives)
  760. def get_params(self):
  761. return json.loads(self.params)
  762. def get_prepares(self):
  763. return json.loads(self.prepares)
  764. class Ssh(Action):
  765. PARAM_FIELDS = ('user', 'host', 'command', 'params')
  766. node_type = 'ssh'
  767. user = models.CharField(max_length=64, verbose_name=_t('User'),
  768. help_text=_t('User executing the shell command.'))
  769. host = models.CharField(max_length=256, verbose_name=_t('Host'),
  770. help_text=_t('Where the shell will be executed.'))
  771. command = models.CharField(max_length=256, verbose_name=_t('%(type)s command') % {'type': node_type.title()},
  772. help_text=_t('The command that will be executed.'))
  773. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  774. help_text=_t('The arguments of the %(type)s command.') % {'type': node_type.title()})
  775. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  776. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  777. 'command output must be in Java properties file format and it must not exceed 2KB. '
  778. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  779. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  780. def get_params(self):
  781. return json.loads(self.params)
  782. class Shell(Action):
  783. PARAM_FIELDS = ('files', 'archives', 'job_properties', 'params', 'prepares')
  784. node_type = 'shell'
  785. command = models.CharField(max_length=256, blank=False, verbose_name=_t('%(type)s command') % {'type': node_type.title()},
  786. help_text=_t('The path of the Shell command to execute.'))
  787. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  788. help_text=_t('The arguments of Shell command can then be specified using one or more argument element.'))
  789. files = models.TextField(default="[]", verbose_name=_t('Files'),
  790. help_text=_t('List of names or paths of files to be added to the distributed cache and the task running directory.'))
  791. archives = models.TextField(default="[]", verbose_name=_t('Archives'),
  792. help_text=_t('List of names or paths of the archives to be added to the distributed cache.'))
  793. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  794. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production)'))
  795. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  796. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  797. 'This should be used exclusively for directory cleanup'))
  798. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  799. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  800. 'Properties specified in the Job Properties element override properties specified in the '
  801. 'files specified in the Job XML element.'))
  802. capture_output = models.BooleanField(default=False, verbose_name=_t('Capture output'),
  803. help_text=_t('Capture output of the stdout of the %(program)s command execution. The %(program)s '
  804. 'command output must be in Java Properties file format and it must not exceed 2KB. '
  805. 'From within the workflow definition, the output of an %(program)s action node is accessible '
  806. 'via the String action:output(String node, String key) function') % {'program': node_type.title()})
  807. def get_properties(self):
  808. return json.loads(self.job_properties)
  809. def get_files(self):
  810. return json.loads(self.files)
  811. def get_archives(self):
  812. return json.loads(self.archives)
  813. def get_params(self):
  814. return json.loads(self.params)
  815. def get_prepares(self):
  816. return json.loads(self.prepares)
  817. class DistCp(Action):
  818. PARAM_FIELDS = ('job_properties', 'params', 'prepares')
  819. node_type = 'distcp'
  820. params = models.TextField(default="[]", verbose_name=_t('Arguments'),
  821. help_text=_t('The arguments of the %(type)s command. Put options first, then source paths, then destination path.')
  822. % {'type': node_type.title()})
  823. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  824. help_text=_t('For the job configuration (e.g. mapred.job.queue.name=production'))
  825. prepares = models.TextField(default="[]", verbose_name=_t('Prepares'),
  826. help_text=_t('List of absolute paths to delete then to create before starting the application. '
  827. 'This should be used exclusively for directory cleanup'))
  828. job_xml = models.CharField(max_length=PATH_MAX, default='', blank=True, verbose_name=_t('Job XML'),
  829. help_text=_t('Refer to a Hadoop JobConf job.xml file bundled in the workflow deployment directory. '
  830. 'Properties specified in the Job Properties element override properties specified in the '
  831. 'files specified in the Job XML element.'))
  832. def get_properties(self):
  833. return json.loads(self.job_properties)
  834. def get_params(self):
  835. return json.loads(self.params)
  836. def get_prepares(self):
  837. return json.loads(self.prepares)
  838. class Fs(Action):
  839. PARAM_FIELDS = ('deletes', 'mkdirs', 'moves', 'chmods', 'touchzs')
  840. node_type = 'fs'
  841. deletes = models.TextField(default="[]", verbose_name=_t('Delete path'), blank=True,
  842. help_text=_t('Delete the specified path, if it is a directory it deletes recursively all its content and '
  843. 'then deletes the directory.'))
  844. mkdirs = models.TextField(default="[]", verbose_name=_t('Create directory'), blank=True,
  845. help_text=_t('Create the specified directory, it creates all missing directories in the path. '
  846. 'If the directory already exist it does a no-op.'))
  847. moves = models.TextField(default="[]", verbose_name=_t('Move file'), blank=True,
  848. help_text=_t('Move a file or directory to another path.'))
  849. chmods = models.TextField(default="[]", verbose_name=_t('Change permissions'), blank=True,
  850. help_text=_t('Change the permissions for the specified path. Permissions can be specified using the Unix Symbolic '
  851. 'representation (e.g. -rwxrw-rw-) or an octal representation (755).'))
  852. touchzs = models.TextField(default="[]", verbose_name=_t('Create or touch a file'), blank=True,
  853. help_text=_t('Creates a zero length file in the specified path if none exists or touch it.'))
  854. def get_deletes(self):
  855. return json.loads(self.deletes)
  856. def get_mkdirs(self):
  857. return json.loads(self.mkdirs)
  858. def get_moves(self):
  859. return json.loads(self.moves)
  860. def get_chmods(self):
  861. return json.loads(self.chmods)
  862. def get_touchzs(self):
  863. return json.loads(self.touchzs)
  864. class Email(Action):
  865. PARAM_FIELDS = ('to', 'cc', 'subject', 'body')
  866. node_type = 'email'
  867. to = models.TextField(default='', verbose_name=_t('TO addresses'), help_text=_t('Comma-separated values.'))
  868. cc = models.TextField(default='', verbose_name=_t('CC addresses (optional)'), blank=True, help_text=_t('Comma-separated values.'))
  869. subject = models.TextField(default='', verbose_name=_t('Subject'), help_text=_t('Plain-text.'))
  870. body = models.TextField(default='', verbose_name=_t('Body'), help_text=_t('Plain-text.'))
  871. class SubWorkflow(Action):
  872. PARAM_FIELDS = ('subworkflow', 'propagate_configuration', 'job_properties')
  873. node_type = 'subworkflow'
  874. sub_workflow = models.ForeignKey(Workflow, db_index=True, verbose_name=_t('Sub-workflow'),
  875. help_text=_t('The sub-workflow application to include. You must own all the sub-workflows.'))
  876. propagate_configuration = models.BooleanField(default=True, verbose_name=_t('Propagate configuration'), blank=True,
  877. help_text=_t('If the workflow job configuration should be propagated to the child workflow.'))
  878. job_properties = models.TextField(default='[]', verbose_name=_t('Hadoop job properties'),
  879. help_text=_t('Can be used to specify the job properties that are required to run the child workflow job.'))
  880. def get_properties(self):
  881. return json.loads(self.job_properties)
  882. class Generic(Action):
  883. PARAM_FIELDS = ('xml',)
  884. node_type = 'generic'
  885. xml = models.TextField(default='', verbose_name=_t('XML of the custom action'),
  886. help_text=_t('This will be inserted verbatim in the action %(action)s. '
  887. 'E.g. all the XML content like %(xml_action)s '
  888. 'will be inserted into the action and produce %(full_action)s') % {
  889. 'action': '<action name="email">...</action>',
  890. 'xml_action': '<email><cc>hue@hue.org</cc></email>',
  891. 'full_action': '<action name="email"><email><cc>hue@hue.org</cc></email><ok/><error/></action>'})
  892. Action.types = (Mapreduce.node_type, Streaming.node_type, Java.node_type, Pig.node_type, Hive.node_type, Sqoop.node_type, Ssh.node_type, Shell.node_type,
  893. DistCp.node_type, Fs.node_type, Email.node_type, SubWorkflow.node_type, Generic.node_type)
  894. class ControlFlow(Node):
  895. """
  896. http://incubator.apache.org/oozie/docs/3.2.0-incubating/docs/WorkflowFunctionalSpec.html#a3.1_Control_Flow_Nodes
  897. """
  898. class Meta:
  899. abstract = True
  900. def get_xml(self):
  901. return django_mako.render_to_string(self.get_template_name(), {})
  902. def is_visible(self):
  903. return True
  904. # Could not make this abstract
  905. class Start(ControlFlow):
  906. node_type = 'start'
  907. def add_node(self, child):
  908. Link.objects.filter(parent=self).delete()
  909. link = Link.objects.create(parent=self, child=child, name='to')
  910. class End(ControlFlow):
  911. node_type = 'end'
  912. def add_node(self, child):
  913. raise RuntimeError(_("End should not have any children."))
  914. class Kill(ControlFlow):
  915. node_type = 'kill'
  916. message = models.CharField(max_length=256, blank=False, default='Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]')
  917. def add_node(self, child):
  918. raise RuntimeError(_("Kill should not have any children."))
  919. def is_visible(self):
  920. return False
  921. class Fork(ControlFlow):
  922. """
  923. A Fork can be converted into a Decision node.
  924. """
  925. node_type = 'fork'
  926. def is_visible(self):
  927. return True
  928. def get_child_join(self):
  929. return Link.objects.get(parent=self, name='related').child.get_full_node()
  930. def convert_to_decision(self):
  931. self.remove_join()
  932. decision = Decision.objects.create(workflow=self.workflow, node_type=Decision.node_type)
  933. decision.save()
  934. links = self.get_all_children_links()
  935. has_default = False
  936. for link in links:
  937. if link.name == 'default':
  938. has_default = True
  939. link.parent = decision
  940. # Defaults to end
  941. if not has_default:
  942. link = Link.objects.create(name="default", parent=decision, child=self.workflow.end)
  943. link.save()
  944. self.delete()
  945. return decision
  946. def remove_join(self):
  947. join = self.get_child_join()
  948. after_join = join.get_child('to')
  949. for parent in join.get_parent_actions():
  950. link = parent.get_link('ok')
  951. link.child = after_join
  952. link.save()
  953. # Automatically delete links thought foreign keys
  954. join.delete()
  955. class Join(ControlFlow):
  956. node_type = 'join'
  957. def is_visible(self):
  958. return True
  959. def get_parent_fork(self):
  960. return self.get_parent_link('related').parent.get_full_node()
  961. def get_parent_actions(self):
  962. return [link.parent for link in self.get_parent_links()]
  963. class Decision(ControlFlow):
  964. """
  965. Essentially a fork where only one of the paths of execution are chosen.
  966. Graphically, this is represented the same way as a fork.
  967. The DecisionEnd node is not represented in Oozie, only in Hue.
  968. """
  969. node_type = 'decision'
  970. def get_child_end(self):
  971. return Link.objects.get(parent=self, name='related').child.get_full_node()
  972. def is_visible(self):
  973. return True
  974. def update_description(self):
  975. self.description = ', '.join(self.get_children_links().values_list('comment', flat=True))
  976. self.save()
  977. class DecisionEnd(ControlFlow):
  978. """
  979. Defines the end of a join.
  980. This node exists purely in the Hue application to provide a smooth transition
  981. from Decision to Endself.
  982. NOTE: NOT AN OOZIE NODE
  983. """
  984. node_type = 'decisionend'
  985. def is_visible(self):
  986. return False
  987. def get_parent_decision(self):
  988. return self.get_parent_link('related').parent.get_full_node()
  989. def get_parent_actions(self):
  990. return [link.parent for link in self.get_parent_links()]
  991. def to_xml(self, mapping):
  992. return ''
  993. FREQUENCY_UNITS = (('minutes', _('Minutes')),
  994. ('hours', _('Hours')),
  995. ('days', _('Days')),
  996. ('months', _('Months')))
  997. FREQUENCY_NUMBERS = [(i, i) for i in xrange(1, 61)]
  998. DATASET_FREQUENCY = ['MINUTE', 'HOUR', 'DAY', 'MONTH', 'YEAR']
  999. class Coordinator(Job):
  1000. frequency_number = models.SmallIntegerField(default=1, choices=FREQUENCY_NUMBERS, verbose_name=_t('Frequency number'),
  1001. help_text=_t('The number of units of the rate at which '
  1002. 'data is periodically created.'))
  1003. frequency_unit = models.CharField(max_length=20, choices=FREQUENCY_UNITS, default='days', verbose_name=_t('Frequency unit'),
  1004. help_text=_t('The unit of the rate at which data is periodically created.'))
  1005. timezone = models.CharField(max_length=24, choices=TIMEZONES, default='America/Los_Angeles', verbose_name=_t('Timezone'),
  1006. help_text=_t('The timezone of the coordinator. Only used for managing the daylight saving time changes when combining several coordinators.'))
  1007. start = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  1008. help_text=_t('When to start the first workflow.'))
  1009. end = models.DateTimeField(default=datetime.today() + timedelta(days=3), verbose_name=_t('End'),
  1010. help_text=_t('When to start the last workflow.'))
  1011. workflow = models.ForeignKey(Workflow, null=True, verbose_name=_t('Workflow'),
  1012. help_text=_t('The workflow to schedule repeatedly.'))
  1013. timeout = models.SmallIntegerField(null=True, blank=True, verbose_name=_t('Timeout'),
  1014. help_text=_t('Number of minutes the coordinator action will be in '
  1015. 'WAITING or READY status before giving up on its execution.'))
  1016. concurrency = models.PositiveSmallIntegerField(null=True, blank=True, choices=FREQUENCY_NUMBERS, verbose_name=_t('Concurrency'),
  1017. help_text=_t('The number of coordinator actions that are allowed to run concurrently (RUNNING status) '
  1018. 'before the coordinator engine starts throttling them.'))
  1019. execution = models.CharField(max_length=10, null=True, blank=True, verbose_name=_t('Execution'),
  1020. choices=(('FIFO', _t('FIFO (oldest first) default')),
  1021. ('LIFO', _t('LIFO (newest first)')),
  1022. ('LAST_ONLY', _t('LAST_ONLY (discards all older materializations)'))),
  1023. help_text=_t('Execution strategy of its coordinator actions when there is backlog of coordinator '
  1024. 'actions in the coordinator engine. The different execution strategies are \'oldest first\', '
  1025. '\'newest first\' and \'last one only\'. A backlog normally happens because of delayed '
  1026. 'input data, concurrency control or because manual re-runs of coordinator jobs.'))
  1027. throttle = models.PositiveSmallIntegerField(null=True, blank=True, choices=FREQUENCY_NUMBERS, verbose_name=_t('Throttle'),
  1028. help_text=_t('The materialization or creation throttle value for its coordinator actions. '
  1029. 'Number of maximum coordinator actions that are allowed to be in WAITING state concurrently.'))
  1030. job_properties = models.TextField(default='[]', verbose_name=_t('Workflow properties'),
  1031. help_text=_t('Additional properties to transmit to the workflow, e.g. limit=100, and EL functions, e.g. username=${coord:user()}'))
  1032. HUE_ID = 'hue-id-c'
  1033. ICON = '/oozie/static/art/icon_oozie_coordinator_24.png'
  1034. METADATA_FORMAT_VERSION = "0.0.1"
  1035. def get_type(self):
  1036. return 'coordinator'
  1037. def to_xml(self, mapping=None):
  1038. if mapping is None:
  1039. mapping = {}
  1040. tmpl = "editor/gen/coordinator.xml.mako"
  1041. return re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {'coord': self, 'mapping': mapping})).encode('utf-8', 'xmlcharrefreplace')
  1042. def clone(self, new_owner=None):
  1043. copy_doc = self.doc.get()
  1044. datasets = Dataset.objects.filter(coordinator=self)
  1045. data_inputs = DataInput.objects.filter(coordinator=self)
  1046. data_outputs = DataOutput.objects.filter(coordinator=self)
  1047. copy = self
  1048. copy.pk = None
  1049. copy.id = None
  1050. copy.name += '-copy'
  1051. copy.deployment_dir = ''
  1052. if new_owner is not None:
  1053. copy.owner = new_owner
  1054. copy.save()
  1055. old_dataset_mapping = {}
  1056. for dataset in datasets:
  1057. prev_id = dataset.id
  1058. dataset.pk = None
  1059. dataset.id = None
  1060. dataset.coordinator = copy
  1061. dataset.save()
  1062. old_dataset_mapping[prev_id] = dataset
  1063. for data_input in data_inputs:
  1064. data_input.pk = None
  1065. data_input.id = None
  1066. data_input.coordinator = copy
  1067. data_input.dataset = old_dataset_mapping[data_input.dataset.id]
  1068. data_input.save()
  1069. for data_output in data_outputs:
  1070. data_output.pk = None
  1071. data_output.id = None
  1072. data_output.coordinator = copy
  1073. data_output.dataset = old_dataset_mapping[data_output.dataset.id]
  1074. data_output.save()
  1075. copy_doc.pk = None
  1076. copy_doc.id = None
  1077. copy_doc.owner = copy.owner
  1078. copy_doc.save()
  1079. copy.doc.add(copy_doc)
  1080. return copy
  1081. @classmethod
  1082. def get_application_path_key(cls):
  1083. return 'oozie.coord.application.path'
  1084. @classmethod
  1085. def get_application_filename(cls):
  1086. return 'coordinator.xml'
  1087. def get_properties(self):
  1088. props = json.loads(self.job_properties)
  1089. index = [prop['name'] for prop in props]
  1090. for prop in self.workflow.get_parameters():
  1091. if not prop['name'] in index:
  1092. props.append(prop)
  1093. index.append(prop['name'])
  1094. # Remove DataInputs and DataOutputs
  1095. datainput_names = [_input.name for _input in self.datainput_set.all()]
  1096. dataoutput_names = [_output.name for _output in self.dataoutput_set.all()]
  1097. removable_names = datainput_names + dataoutput_names
  1098. props = filter(lambda prop: prop['name'] not in removable_names, props)
  1099. return props
  1100. @property
  1101. def job_properties_escapejs(self):
  1102. return self._escapejs_parameters_list(self.job_properties)
  1103. @property
  1104. def start_utc(self):
  1105. return utc_datetime_format(self.start)
  1106. @property
  1107. def end_utc(self):
  1108. return utc_datetime_format(self.end)
  1109. def get_absolute_url(self):
  1110. return reverse('oozie:edit_coordinator', kwargs={'coordinator': self.id})
  1111. @property
  1112. def frequency(self):
  1113. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1114. @property
  1115. def text_frequency(self):
  1116. return '%(number)d %(unit)s' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1117. def find_parameters(self):
  1118. params = self.workflow.find_parameters()
  1119. for param in find_parameters(self, ['job_properties']):
  1120. params[param] = ''
  1121. for dataset in self.dataset_set.all():
  1122. for param in find_parameters(dataset, ['uri']):
  1123. if param not in set(DATASET_FREQUENCY):
  1124. params[param] = ''
  1125. for ds in self.datainput_set.all():
  1126. params.pop(ds.name, None)
  1127. for ds in self.dataoutput_set.all():
  1128. params.pop(ds.name, None)
  1129. for wf_param in json.loads(self.job_properties):
  1130. params.pop(wf_param['name'], None)
  1131. return params
  1132. def compress(self, mapping=None, fp=StringIO.StringIO()):
  1133. metadata = {
  1134. 'version': Coordinator.METADATA_FORMAT_VERSION,
  1135. 'workflow': self.workflow.name,
  1136. 'attributes': {
  1137. 'description': self.description,
  1138. 'deployment_dir': self.deployment_dir
  1139. }
  1140. }
  1141. xml = self.to_xml(mapping=mapping)
  1142. zfile = zipfile.ZipFile(fp, 'w')
  1143. zfile.writestr("coordinator.xml", smart_str(xml))
  1144. zfile.writestr("coordinator-metadata.json", smart_str(json.dumps(metadata)))
  1145. zfile.close()
  1146. return fp
  1147. @classmethod
  1148. def decompress(cls, fp):
  1149. zfile = zipfile.ZipFile(fp, 'r')
  1150. metadata_json = zfile.read('coordinator-metadata.json')
  1151. metadata = json.loads(metadata_json)
  1152. xml = zfile.read('coordinator.xml')
  1153. return xml, metadata
  1154. class DatasetManager(models.Manager):
  1155. def can_read_or_exception(self, request, dataset_id):
  1156. if dataset_id is None:
  1157. return
  1158. try:
  1159. dataset = Dataset.objects.get(pk=dataset_id)
  1160. if dataset.coordinator.can_read(request.user):
  1161. return dataset
  1162. else:
  1163. message = _("Permission denied. %(username)s does not have the permissions to access dataset %(id)s.") % \
  1164. {'username': request.user.username, 'id': dataset.id}
  1165. access_warn(request, message)
  1166. request.error(message)
  1167. raise PopupException(message)
  1168. except Dataset.DoesNotExist:
  1169. raise PopupException(_('dataset %(id)s not exist') % {'id': dataset_id})
  1170. class Dataset(models.Model):
  1171. """
  1172. http://oozie.apache.org/docs/3.3.0/CoordinatorFunctionalSpec.html#a6.3._Synchronous_Coordinator_Application_Definition
  1173. """
  1174. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name'),
  1175. help_text=_t('The name of the dataset.'))
  1176. description = models.CharField(max_length=1024, blank=True, default='', verbose_name=_t('Description'),
  1177. help_text=_t('A description of the dataset.'))
  1178. start = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  1179. help_text=_t(' The UTC datetime of the initial instance of the dataset. The initial instance also provides '
  1180. 'the baseline datetime to compute instances of the dataset using multiples of the frequency.'))
  1181. frequency_number = models.SmallIntegerField(default=1, choices=FREQUENCY_NUMBERS, verbose_name=_t('Frequency number'),
  1182. help_text=_t('The number of units of the rate at which '
  1183. 'data is periodically created.'))
  1184. frequency_unit = models.CharField(max_length=20, choices=FREQUENCY_UNITS, default='days', verbose_name=_t('Frequency unit'),
  1185. help_text=_t('The unit of the rate at which data is periodically created.'))
  1186. uri = models.CharField(max_length=1024, default='/data/${YEAR}${MONTH}${DAY}', verbose_name=_t('URI'),
  1187. help_text=_t('The URI template that identifies the dataset and can be resolved into concrete URIs to identify a particular '
  1188. 'dataset instance. The URI consist of constants (e.g. ${YEAR}/${MONTH}) and '
  1189. 'configuration properties (e.g. /home/${USER}/projects/${PROJECT})'))
  1190. timezone = models.CharField(max_length=24, choices=TIMEZONES, default='America/Los_Angeles', verbose_name=_t('Timezone'),
  1191. help_text=_t('The timezone of the dataset. Only used for managing the daylight saving time changes when combining several datasets.'))
  1192. done_flag = models.CharField(max_length=64, blank=True, default='', verbose_name=_t('Done flag'),
  1193. help_text=_t('The done file for the data set. If the Done flag is not specified, then Oozie '
  1194. 'configures Hadoop to create a _SUCCESS file in the output directory. If Done '
  1195. 'flag is set to empty, then Coordinator looks for the existence of the directory itself.'))
  1196. coordinator = models.ForeignKey(Coordinator, verbose_name=_t('Coordinator'),
  1197. help_text=_t('The coordinator associated with this data.'))
  1198. instance_choice = models.CharField(max_length=10, default='default', verbose_name=_t('Instance type'),
  1199. help_text=_t('Customize the date instance(s), e.g. define a range of dates, use EL functions...'))
  1200. advanced_start_instance = models.CharField(max_length=128, default='0', verbose_name=_t('Start instance'),
  1201. help_text=_t('Shift the frequency for gettting past/future start date or enter verbatim the Oozie start instance, e.g. ${coord:current(0)}'))
  1202. advanced_end_instance = models.CharField(max_length=128, blank=True, default='0', verbose_name=_t('End instance'),
  1203. help_text=_t('Optional: Shift the frequency for gettting past/future end dates or enter verbatim the Oozie end instance.'))
  1204. objects = DatasetManager()
  1205. unique_together = ('coordinator', 'name')
  1206. def __unicode__(self):
  1207. return '%s' % (self.name,)
  1208. @property
  1209. def start_utc(self):
  1210. return utc_datetime_format(self.start)
  1211. @property
  1212. def frequency(self):
  1213. return '${coord:%(unit)s(%(number)d)}' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1214. @property
  1215. def text_frequency(self):
  1216. return '%(number)d %(unit)s' % {'unit': self.frequency_unit, 'number': self.frequency_number}
  1217. @property
  1218. def start_instance(self):
  1219. if not self.is_advanced_start_instance:
  1220. return int(self.advanced_start_instance)
  1221. else:
  1222. return 0
  1223. @property
  1224. def is_advanced_start_instance(self):
  1225. return not self.is_int(self.advanced_start_instance)
  1226. def is_int(self, text):
  1227. try:
  1228. int(text)
  1229. return True
  1230. except ValueError:
  1231. return False
  1232. @property
  1233. def end_instance(self):
  1234. if not self.is_advanced_end_instance:
  1235. return int(self.advanced_end_instance)
  1236. else:
  1237. return 0
  1238. @property
  1239. def is_advanced_end_instance(self):
  1240. return not self.is_int(self.advanced_end_instance)
  1241. class DataInput(models.Model):
  1242. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name of an input variable in the workflow.'),
  1243. help_text=_t('The name of the variable of the workflow to automatically fill up.'))
  1244. dataset = models.OneToOneField(Dataset, verbose_name=_t('The dataset representing format of the data input.'),
  1245. help_text=_t('The pattern of the input data we want to process.'))
  1246. coordinator = models.ForeignKey(Coordinator)
  1247. unique_together = ('coordinator', 'name')
  1248. class DataOutput(models.Model):
  1249. name = models.CharField(max_length=40, validators=[name_validator], verbose_name=_t('Name of an output variable in the workflow'),
  1250. help_text=_t('The name of the variable of the workflow to automatically filled up.'))
  1251. dataset = models.OneToOneField(Dataset, verbose_name=_t('The dataset representing the format of the data output.'),
  1252. help_text=_t('The pattern of the output data we want to generate.'))
  1253. coordinator = models.ForeignKey(Coordinator)
  1254. unique_together = ('coordinator', 'name')
  1255. class BundledCoordinator(models.Model):
  1256. bundle = models.ForeignKey('Bundle', verbose_name=_t('Bundle'),
  1257. help_text=_t('The bundle regrouping all the coordinators.'))
  1258. coordinator = models.ForeignKey(Coordinator, verbose_name=_t('Coordinator'),
  1259. help_text=_t('The coordinator to batch with other coordinators.'))
  1260. parameters = models.TextField(default='[{"name":"oozie.use.system.libpath","value":"true"}]', verbose_name=_t('Parameters'),
  1261. help_text=_t('Constants used at the submission time (e.g. market=US, oozie.use.system.libpath=true).'))
  1262. def get_parameters(self):
  1263. return json.loads(self.parameters)
  1264. class Bundle(Job):
  1265. kick_off_time = models.DateTimeField(default=datetime.today(), verbose_name=_t('Start'),
  1266. help_text=_t('When to start the first coordinators.'))
  1267. coordinators = models.ManyToManyField(Coordinator, through='BundledCoordinator')
  1268. HUE_ID = 'hue-id-b'
  1269. ICON = '/oozie/static/art/icon_oozie_bundle_24.png'
  1270. METADATA_FORMAT_VERSION = '0.0.1'
  1271. def get_type(self):
  1272. return 'bundle'
  1273. def to_xml(self, mapping=None):
  1274. if mapping is None:
  1275. mapping = {}
  1276. tmpl = "editor/gen/bundle.xml.mako"
  1277. return force_unicode(
  1278. re.sub(re.compile('\s*\n+', re.MULTILINE), '\n', django_mako.render_to_string(tmpl, {
  1279. 'bundle': self,
  1280. 'mapping': mapping
  1281. })))
  1282. def clone(self, new_owner=None):
  1283. bundleds = BundledCoordinator.objects.filter(bundle=self)
  1284. copy_doc = self.doc.get()
  1285. copy = self
  1286. copy.pk = None
  1287. copy.id = None
  1288. copy.name += '-copy'
  1289. copy.deployment_dir = ''
  1290. if new_owner is not None:
  1291. copy.owner = new_owner
  1292. copy.save()
  1293. for bundled in bundleds:
  1294. bundled.pk = None
  1295. bundled.id = None
  1296. bundled.bundle = copy
  1297. bundled.save()
  1298. copy_doc.pk = None
  1299. copy_doc.id = None
  1300. copy_doc.owner = copy.owner
  1301. copy_doc.save()
  1302. copy.doc.add(copy_doc)
  1303. return copy
  1304. @classmethod
  1305. def get_application_path_key(cls):
  1306. return 'oozie.bundle.application.path'
  1307. @classmethod
  1308. def get_application_filename(cls):
  1309. return 'bundle.xml'
  1310. def get_absolute_url(self):
  1311. return reverse('oozie:edit_bundle', kwargs={'bundle': self.id})
  1312. def find_parameters(self):
  1313. params = {}
  1314. for bundled in BundledCoordinator.objects.filter(bundle=self):
  1315. for param in bundled.coordinator.find_parameters():
  1316. params[param] = ''
  1317. for param in bundled.get_parameters():
  1318. params.pop(param['name'], None)
  1319. return params
  1320. @property
  1321. def kick_off_time_utc(self):
  1322. return utc_datetime_format(self.kick_off_time)
  1323. def compress(self, mapping=None, fp=StringIO.StringIO()):
  1324. metadata = {
  1325. 'version': Bundle.METADATA_FORMAT_VERSION,
  1326. 'attributes': {
  1327. 'description': self.description,
  1328. 'deployment_dir': self.deployment_dir
  1329. }
  1330. }
  1331. xml = self.to_xml(mapping=mapping)
  1332. zfile = zipfile.ZipFile(fp, 'w')
  1333. zfile.writestr("bundle.xml", smart_str(xml))
  1334. zfile.writestr("bundle-metadata.json", smart_str(json.dumps(metadata)))
  1335. zfile.close()
  1336. return fp
  1337. @classmethod
  1338. def decompress(cls, fp):
  1339. zfile = zipfile.ZipFile(fp, 'r')
  1340. metadata_json = zfile.read('bundle-metadata.json')
  1341. metadata = json.loads(metadata_json)
  1342. xml = zfile.read('bundle.xml')
  1343. return xml, metadata
  1344. class HistoryManager(models.Manager):
  1345. def create_from_submission(self, submission):
  1346. History.objects.create(submitter=submission.user,
  1347. oozie_job_id=submission.oozie_id,
  1348. job=submission.job,
  1349. properties=json.dumps(submission.properties))
  1350. class History(models.Model):
  1351. """
  1352. Contains information on submitted workflows/coordinators.
  1353. """
  1354. submitter = models.ForeignKey(User, db_index=True)
  1355. submission_date = models.DateTimeField(auto_now=True, db_index=True)
  1356. oozie_job_id = models.CharField(max_length=128)
  1357. job = models.ForeignKey(Job, db_index=True)
  1358. properties = models.TextField()
  1359. objects = HistoryManager()
  1360. @property
  1361. def properties_dict(self):
  1362. return json.loads(self.properties)
  1363. def get_absolute_oozie_url(self):
  1364. view = 'oozie:list_oozie_workflow'
  1365. if self.oozie_job_id.endswith('C'):
  1366. view = 'oozie:list_oozie_coordinator'
  1367. elif self.oozie_job_id.endswith('B'):
  1368. view = 'oozie:list_oozie_bundle'
  1369. return reverse(view, kwargs={'job_id': self.oozie_job_id})
  1370. def get_workflow(self):
  1371. if self.oozie_job_id.endswith('W'):
  1372. return self.job.get_full_node()
  1373. def get_coordinator(self):
  1374. if self.oozie_job_id.endswith('C'):
  1375. return self.job.get_full_node()
  1376. @classmethod
  1377. def get_workflow_from_config(self, conf_dict):
  1378. try:
  1379. return Workflow.objects.get(id=conf_dict.get(Workflow.HUE_ID))
  1380. except Workflow.DoesNotExist:
  1381. pass
  1382. @classmethod
  1383. def get_coordinator_from_config(self, conf_dict):
  1384. try:
  1385. return Coordinator.objects.get(id=conf_dict.get(Coordinator.HUE_ID))
  1386. except Coordinator.DoesNotExist:
  1387. pass
  1388. @classmethod
  1389. def cross_reference_submission_history(cls, user, oozie_id, coordinator_job_id):
  1390. # Try do get the history
  1391. history = None
  1392. try:
  1393. history = History.objects.get(oozie_job_id=oozie_id)
  1394. if history.job.owner != user:
  1395. history = None
  1396. except History.DoesNotExist:
  1397. pass
  1398. return history
  1399. def find_parameters(instance, fields=None):
  1400. """Find parameters in the given fields"""
  1401. if fields is None:
  1402. fields = [field.name for field in instance._meta.fields]
  1403. params = []
  1404. for field in fields:
  1405. data = getattr(instance, field)
  1406. if isinstance(data, basestring):
  1407. for match in Template.pattern.finditer(data):
  1408. name = match.group('braced')
  1409. if name is not None:
  1410. params.append(name)
  1411. return params
  1412. # See http://wiki.apache.org/hadoop/JobConfFile
  1413. _STD_PROPERTIES = [
  1414. 'mapred.input.dir',
  1415. 'mapred.output.dir',
  1416. 'mapred.job.name',
  1417. 'mapred.job.queue.name',
  1418. 'mapred.mapper.class',
  1419. 'mapred.reducer.class',
  1420. 'mapred.combiner.class',
  1421. 'mapred.partitioner.class',
  1422. 'mapred.map.tasks',
  1423. 'mapred.reduce.tasks',
  1424. 'mapred.input.format.class',
  1425. 'mapred.output.format.class',
  1426. 'mapred.input.key.class',
  1427. 'mapred.input.value.class',
  1428. 'mapred.output.key.class',
  1429. 'mapred.output.value.class',
  1430. 'mapred.mapoutput.key.class',
  1431. 'mapred.mapoutput.value.class',
  1432. 'mapred.combine.buffer.size',
  1433. 'mapred.min.split.size',
  1434. 'mapred.speculative.execution',
  1435. 'mapred.map.tasks.speculative.execution',
  1436. 'mapred.reduce.tasks.speculative.execution',
  1437. 'mapred.queue.default.acl-administer-jobs',
  1438. ]
  1439. _STD_PROPERTIES_JSON = json.dumps(_STD_PROPERTIES)
  1440. ACTION_TYPES = {
  1441. Mapreduce.node_type: Mapreduce,
  1442. Streaming.node_type: Streaming,
  1443. Java.node_type: Java,
  1444. Pig.node_type: Pig,
  1445. Hive.node_type: Hive,
  1446. Sqoop.node_type: Sqoop,
  1447. Ssh.node_type: Ssh,
  1448. Shell.node_type: Shell,
  1449. DistCp.node_type: DistCp,
  1450. Fs.node_type: Fs,
  1451. Email.node_type: Email,
  1452. SubWorkflow.node_type: SubWorkflow,
  1453. Generic.node_type: Generic,
  1454. }
  1455. CONTROL_TYPES = {
  1456. Fork.node_type: Fork,
  1457. Join.node_type: Join,
  1458. Decision.node_type: Decision,
  1459. DecisionEnd.node_type: DecisionEnd,
  1460. Start.node_type: Start,
  1461. End.node_type: End,
  1462. }
  1463. NODE_TYPES = ACTION_TYPES.copy()
  1464. NODE_TYPES.update(CONTROL_TYPES)