models.py 79 KB

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