Quellcode durchsuchen

[oozie] Fix clone workflow action

Add missing setup command
Romain Rigaux vor 13 Jahren
Ursprung
Commit
8f42218

+ 89 - 0
apps/oozie/src/oozie/management/commands/oozie_setup.py

@@ -0,0 +1,89 @@
+#!/usr/bin/env python
+# Licensed to Cloudera, Inc. under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  Cloudera, Inc. licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import logging
+import os
+import posixpath
+
+from django.core import management
+from django.core.management.base import NoArgsCommand
+
+from hadoop import cluster
+
+from oozie.conf import LOCAL_SAMPLE_DATA_DIR, LOCAL_SAMPLE_DIR
+from oozie.models import Workflow
+
+
+LOG = logging.getLogger(__name__)
+
+
+class Command(NoArgsCommand):
+  def handle_noargs(self, **options):
+    remote_fs = cluster.get_hdfs()
+    remote_dir = Workflow.objects.create_data_dir(remote_fs)
+
+    # Copy sample binaries
+    for demo in ('lib', 'pig'):
+      local_dir = posixpath.join(LOCAL_SAMPLE_DIR.get(), demo)
+      remote_data_dir = posixpath.join(remote_dir, demo)
+      LOG.info('Copying workflows %s to %s\n' % (local_dir, remote_data_dir))
+      copy_dir(local_dir, remote_fs, remote_data_dir)
+
+    # Copy sample data
+    local_dir = LOCAL_SAMPLE_DATA_DIR.get()
+    remote_data_dir = posixpath.join(remote_dir, 'data')
+    LOG.info('Copying data %s to %s\n' % (local_dir, remote_data_dir))
+    copy_dir(local_dir, remote_fs, remote_data_dir)
+
+    # Load jobs
+    management.call_command('loaddata', 'apps/oozie/src/oozie/fixtures/initial_data.json', verbosity=2)
+
+  def has_been_setup(self):
+    return False
+
+
+def copy_dir(local_dir, remote_fs, remote_dir, mode=755):
+  remote_fs.mkdir(remote_dir, mode=mode)
+
+  for f in os.listdir(local_dir):
+    local_src = os.path.join(local_dir, f)
+    remote_dst = posixpath.join(remote_dir, f)
+    copy_file(local_src, remote_fs, remote_dst)
+
+
+CHUNK_SIZE = 1024 * 1024
+
+def copy_file(local_src, remote_fs, remote_dst):
+  if remote_fs.exists(remote_dst):
+    LOG.info('%s already exists.  Skipping.' % remote_dst)
+    return
+  else:
+    LOG.info('%s does not exist. trying to copy' % remote_dst)
+
+  if os.path.isfile(local_src):
+    src = file(local_src)
+    try:
+      remote_fs.create(remote_dst, permission=01755)
+      chunk = src.read(CHUNK_SIZE)
+      while chunk:
+        remote_fs.append(remote_dst, chunk)
+        chunk = src.read(CHUNK_SIZE)
+      LOG.info('Copied %s -> %s' % (local_src, remote_dst))
+    finally:
+      src.close()
+  else:
+    LOG.info('Skipping %s (not a file)' % local_src)

+ 12 - 11
apps/oozie/src/oozie/models.py

@@ -36,7 +36,7 @@ from desktop.lib import django_mako
 from hadoop.fs.hadoopfs import Hdfs
 from liboozie.submittion import Submission
 
-from oozie.conf import REMOTE_SAMPLE_DIR
+from oozie.conf import REMOTE_SAMPLE_DIR, REMOTE_DEPLOYMENT_DIR
 from timezones import TIMEZONES
 
 
@@ -143,23 +143,24 @@ class WorkflowManager(models.Manager):
 
   @classmethod
   def create_data_dir(cls, fs):
-    # If needed, create the remote home and data directories
-    remote_data_dir = REMOTE_SAMPLE_DIR.get()
+    # If needed, create the remote home, deployment and data directories
+    directories = (REMOTE_DEPLOYMENT_DIR.get(), REMOTE_SAMPLE_DIR.get())
     user = fs.user
 
     try:
       fs.setuser(fs.DEFAULT_USER)
-      if not fs.exists(remote_data_dir):
-        remote_home_dir = Hdfs.join('/user', fs.user)
-        if remote_data_dir.startswith(remote_home_dir):
-          # Home is 755
-          fs.create_home_dir(remote_home_dir)
-        # Shared by all the users
-        fs.mkdir(remote_data_dir, 01777)
+      for directory in directories:
+        if not fs.exists(directory):
+          remote_home_dir = Hdfs.join('/user', fs.user)
+          if directory.startswith(remote_home_dir):
+            # Home is 755
+            fs.create_home_dir(remote_home_dir)
+          # Shared by all the users
+          fs.mkdir(directory, 01777)
     finally:
       fs.setuser(user)
 
-    return remote_data_dir
+    return REMOTE_SAMPLE_DIR.get()
 
 
 class Workflow(Job):

+ 10 - 5
apps/oozie/src/oozie/tests.py

@@ -203,15 +203,15 @@ class TestEditor:
     assert_not_equal(self.wf.deployment_dir, wf2.deployment_dir)
 
 
-  def test_clone_node(self):
+  def test_clone_action(self):
+    # Need to be tested in edit:workflow too
     action1 = Node.objects.get(name='action-name-1')
 
     node_count = self.wf.actions.count()
     assert_true(1, len(action1.get_children()))
 
-    response = self.c.get(reverse('oozie:clone_action', args=[action1.id]), {}, follow=True)
+    response = self.c.post(reverse('oozie:clone_action', args=[action1.id]), {}, follow=True)
 
-    assert_equal(200, response.status_code)
     assert_not_equal(action1.id, action1.get_children()[1].id)
     assert_true(2, len(action1.get_children()))
     assert_equal(node_count + 1, self.wf.actions.count())
@@ -294,19 +294,23 @@ class TestEditor:
   def test_edit_workflow(self):
     response = self.c.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
     assert_true('Editor' in response.content, response.content)
+    assert_true('Workflow wf-name-1' in response.content, response.content)
 
     # Edit
     finish = SHARE_JOBS.set_for_testing(True)
     try:
       response = self.c.post(reverse('oozie:edit_workflow', args=[self.wf.id]), {})
-      assert_true('wf-name-1' in response.content, response.content)
+      assert_true('jHueNotify.error' in response.content, response.content)
     finally:
       finish()
 
+    # Build POST dict from the forms and test this
+    raise SkipTest
+
     finish = SHARE_JOBS.set_for_testing(True)
     try:
       response = self.c.post(reverse('oozie:edit_workflow', args=[self.wf.id]), WORKFLOW_DICT)
-      assert_true('wf-name-1' in response.content, response.content)
+      assert_false('jHueNotify.error' in response.content, response.content)
     finally:
       finish()
 
@@ -314,6 +318,7 @@ class TestEditor:
   def test_workflow_permissions(self):
     response = self.c.get(reverse('oozie:edit_workflow', args=[self.wf.id]))
     assert_true('Editor' in response.content, response.content)
+    assert_true('Workflow wf-name-1' in response.content, response.content)
     assert_false(self.wf.is_shared)
 
     # Login as someone else

+ 19 - 13
apps/oozie/src/oozie/views/editor.py

@@ -110,7 +110,9 @@ def can_edit_job(user, job):
 
 
 def can_edit_job_or_exception(request, job):
-  if not can_edit_job(request.user, job):
+  if can_edit_job(request.user, job):
+    return True
+  else:
     raise PopupException('Not allowed to modified this job')
 
 
@@ -218,10 +220,10 @@ def edit_workflow(request, workflow):
   history = History.objects.filter(submitter=request.user, job=workflow)
 
   if request.method == 'POST' and can_edit_job_or_exception(request, workflow):
-    workflow_form = WorkflowForm(request.POST, instance=workflow)
-    actions_formset = WorkflowFormSet(request.POST, request.FILES, instance=workflow)
-
     try:
+      workflow_form = WorkflowForm(request.POST, instance=workflow)
+      actions_formset = WorkflowFormSet(request.POST, request.FILES, instance=workflow)
+
       if 'clone_action' in request.POST: return clone_action(request, action=request.POST['clone_action'])
       if 'delete_action' in request.POST: return delete_action(request, action=request.POST['delete_action'])
       if 'move_up_action' in request.POST: return move_up_action(request, action=request.POST['move_up_action'])
@@ -233,9 +235,9 @@ def edit_workflow(request, workflow):
         return redirect(reverse('oozie:list_workflows'))
     except Exception, e:
       request.error(_('Sorry, this operation is not supported: %(error)s') % {'error': e})
-  else:
-    workflow_form = WorkflowForm(instance=workflow)
-    actions_formset = WorkflowFormSet(instance=workflow)
+
+  workflow_form = WorkflowForm(instance=workflow)
+  actions_formset = WorkflowFormSet(instance=workflow)
 
   return render('editor/edit_workflow.mako', request, {
     'workflow_form': workflow_form,
@@ -411,13 +413,17 @@ def delete_action(request, action):
 
 
 @check_action_access_permission
+@check_action_edition_permission
 def clone_action(request, action):
-  # Really weird: action is like a clone object with the old id here
-  action_id = action.id
-  workflow = action.workflow
-  clone = action.clone()
-  workflow.add_action(clone, action_id)
-  return redirect(reverse('oozie:edit_workflow', kwargs={'workflow': workflow.id}))
+  if request.method == 'POST':
+    # Really weird: action is like a clone object with the old id here
+    action_id = action.id
+    workflow = action.workflow
+    clone = action.clone()
+    workflow.add_action(clone, action_id)
+    return redirect(reverse('oozie:edit_workflow', kwargs={'workflow': workflow.id}))
+  else:
+    raise PopupException(_('A POST request is required.'))
 
 
 @check_action_access_permission