utils.py 6.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209
  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. from future import standard_library
  18. standard_library.install_aliases()
  19. from builtins import str
  20. from past.builtins import basestring
  21. import json
  22. import logging
  23. import re
  24. import urllib.parse
  25. from datetime import datetime
  26. from dateutil import tz
  27. from dateutil import parser
  28. from django.utils.formats import localize_input
  29. from django.utils.translation import ugettext as _
  30. from desktop.lib.parameterization import find_variables
  31. from liboozie.oozie_api import get_oozie, DEFAULT_USER
  32. LOG = logging.getLogger(__name__)
  33. JSON_FIELDS = ('parameters', 'job_properties', 'files', 'archives', 'prepares', 'params',
  34. 'deletes', 'mkdirs', 'moves', 'chmods', 'touchzs')
  35. BOOLEAN_FIELDS = ('propagate_configuration','capture_output')
  36. NUMBER_FIELDS_OR_NULL = ('sub_workflow',)
  37. GMT_TIME_FORMAT = "%Y-%m-%dT%H:%MGMT%z"
  38. UTC_TIME_FORMAT = "%Y-%m-%dT%H:%MZ"
  39. FREQUENCY_REGEX = r'^\$\{coord:(?P<frequency_unit>\w+)\((?P<frequency_number>\d+)\)\}$'
  40. def format_field_value(field, value):
  41. if field in JSON_FIELDS:
  42. if isinstance(value, basestring):
  43. value = json.loads(value)
  44. value = [item for item in value if isinstance(item, dict) and item.get('name')]
  45. return json.dumps(value)
  46. if field in NUMBER_FIELDS_OR_NULL:
  47. if not isinstance(value, int) and value is not None:
  48. return int(value)
  49. if field in BOOLEAN_FIELDS:
  50. return str(value).lower() == 'true'
  51. return value
  52. def format_dict_field_values(dictionary):
  53. for key in dictionary:
  54. dictionary[key] = format_field_value(key, dictionary[key])
  55. return dictionary
  56. def model_to_dict(model):
  57. from django.db import models
  58. dictionary = {}
  59. for field in model._meta.fields:
  60. try:
  61. attr = getattr(model, field.name, None)
  62. if isinstance(attr, models.Model):
  63. dictionary[field.name] = attr.id
  64. elif isinstance(attr, datetime):
  65. dictionary[field.name] = str(attr)
  66. else:
  67. dictionary[field.name] = attr
  68. except Exception as e:
  69. LOG.debug(_("Could not set field %(field)s: %(exception)s") % {'field': field.name, 'exception': str(e)})
  70. return dictionary
  71. def sanitize_node_dict(node_dict):
  72. for field in ['node_ptr', 'workflow']:
  73. if field in node_dict:
  74. del node_dict[field]
  75. return node_dict
  76. def workflow_to_dict(workflow):
  77. workflow_dict = model_to_dict(workflow)
  78. node_list = [node.get_full_node() for node in workflow.node_list]
  79. nodes = [model_to_dict(node) for node in node_list]
  80. for index, node in enumerate(node_list):
  81. nodes[index]['child_links'] = [model_to_dict(link) for link in node.get_all_children_links()]
  82. workflow_dict['nodes'] = nodes
  83. return workflow_dict
  84. def smart_path(path, mapping=None, is_coordinator=False):
  85. # Try to prepend home_dir and FS scheme if needed.
  86. # If path starts by a parameter try to get its value from the list of parameters submitted by the user or the coordinator.
  87. # This dynamic checking enable the use of <prepares> statements in a workflow scheduled manually of by a coordinator.
  88. # The logic is a bit complicated but Oozie is not consistent with data paths, prepare, coordinator paths and Fs action.
  89. if mapping is None:
  90. mapping = {}
  91. path = path.strip()
  92. if not path.startswith('$') and not path.startswith('/') and not urllib.parse.urlsplit(path).scheme:
  93. path = '/user/%(username)s/%(path)s' % {
  94. 'username': '${coord:user()}' if is_coordinator else '${wf:user()}',
  95. 'path': path
  96. }
  97. if path.startswith('$'):
  98. variables = find_variables(path)
  99. for var in variables:
  100. prefix = '${%s}' % var
  101. if path.startswith(prefix):
  102. if var in mapping:
  103. if not urllib.parse.urlsplit(mapping[var]).scheme and not mapping[var].startswith('$'):
  104. path = '%(nameNode)s%(path)s' % {'nameNode': '${nameNode}', 'path': path}
  105. else:
  106. if not urllib.parse.urlsplit(path).scheme:
  107. path = '%(nameNode)s%(path)s' % {'nameNode': '${nameNode}', 'path': path}
  108. return path
  109. def contains_symlink(path, mapping):
  110. vars = find_variables(path)
  111. return any([var in mapping and '#' in mapping[var] for var in vars]) or '#' in path
  112. def utc_datetime_format(utc_time):
  113. if utc_time and type(utc_time) is datetime:
  114. return utc_time.strftime(UTC_TIME_FORMAT)
  115. return utc_time
  116. def oozie_to_django_datetime(dt_string):
  117. try:
  118. return localize_input(datetime.strptime(dt_string, UTC_TIME_FORMAT))
  119. except ValueError:
  120. pass
  121. try:
  122. return localize_input(datetime.strptime(dt_string, GMT_TIME_FORMAT))
  123. except ValueError:
  124. pass
  125. return None
  126. class InvalidFrequency(Exception):
  127. pass
  128. def oozie_to_hue_frequency(frequency_string):
  129. """
  130. Get frequency number and units from frequency, which must be of the format
  131. "${coord:$unit($number)}".
  132. frequency units and number are just different parts of the EL function.
  133. Returns:
  134. A tuple of the frequency unit and number
  135. Raises:
  136. InvalidFrequency: If the `frequency_string` does not match the frequency pattern.
  137. """
  138. matches = re.match(FREQUENCY_REGEX, frequency_string)
  139. if matches:
  140. return matches.group('frequency_unit'), matches.group('frequency_number')
  141. else:
  142. raise InvalidFrequency(_('invalid frequency: %s') % frequency_string)
  143. def convert_to_server_timezone(date, local_tz='UTC', server_tz=None, user=DEFAULT_USER):
  144. api = get_oozie(user)
  145. if server_tz is None:
  146. oozie_conf = api.get_configuration()
  147. server_tz = oozie_conf.get('oozie.processing.timezone') or 'UTC'
  148. if date and date.startswith('$'):
  149. return date
  150. # To support previously created jobs
  151. if date.endswith('Z'):
  152. date = date[:-1]
  153. local_tz = 'UTC'
  154. try:
  155. date_local_tz = parser.parse(date)
  156. date_local_tz = date_local_tz.replace(tzinfo=tz.gettz(local_tz))
  157. date_server_tz = date_local_tz.astimezone(tz.gettz(server_tz))
  158. # Oozie timezone is either UTC or GMT(+/-)####
  159. if 'UTC' == server_tz:
  160. return date_server_tz.strftime('%Y-%m-%dT%H:%M') + u'Z'
  161. else:
  162. return date_server_tz.strftime('%Y-%m-%dT%H:%M') + date_server_tz.strftime('%z')
  163. except TypeError as ValueError:
  164. LOG.error("Failed to convert Oozie timestamp: %s" % date)
  165. return None