|
|
@@ -21,15 +21,14 @@ import time
|
|
|
import subprocess
|
|
|
import threading
|
|
|
|
|
|
-from nose.plugins.skip import SkipTest
|
|
|
-from nose.tools import assert_equal, assert_true
|
|
|
+from nose.tools import assert_equal
|
|
|
|
|
|
-from desktop.lib.django_test_util import make_logged_in_client
|
|
|
from desktop.lib.paths import get_run_root
|
|
|
|
|
|
from liboozie.oozie_api import get_oozie
|
|
|
from liboozie.conf import OOZIE_URL
|
|
|
from hadoop import pseudo_hdfs4
|
|
|
+import atexit
|
|
|
|
|
|
|
|
|
_oozie_running = False
|
|
|
@@ -37,12 +36,11 @@ _oozie_lock = threading.Lock()
|
|
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
-# TODO HUE-752
|
|
|
+
|
|
|
class OozieServerProvider(object):
|
|
|
"""
|
|
|
Setup a Oozie server.
|
|
|
"""
|
|
|
- STATES_WF_COMPLETION = ('SUCCEEDED' , 'KILLED', 'FAILED')
|
|
|
OOZIE_TEST_PORT = '18080'
|
|
|
OOZIE_HOME = get_run_root('ext/oozie/oozie')
|
|
|
|
|
|
@@ -50,37 +48,32 @@ class OozieServerProvider(object):
|
|
|
|
|
|
@classmethod
|
|
|
def setup_class(cls):
|
|
|
- raise SkipTest
|
|
|
-
|
|
|
cls.cluster = pseudo_hdfs4.shared_cluster()
|
|
|
- cls.client = make_logged_in_client()
|
|
|
cls.oozie, callback = cls._get_shared_oozie_server()
|
|
|
- cls.shutdown = [ callback ]
|
|
|
+ cls.shutdown = [callback]
|
|
|
|
|
|
- def wait_until_completion(self, jobid, timeout=60.0, step=1):
|
|
|
+ @classmethod
|
|
|
+ def wait_until_completion(cls, oozie_jobid, timeout=70.0, step=1):
|
|
|
sleep = 0
|
|
|
- workflow = None
|
|
|
+ job = cls.oozie.get_job(oozie_jobid)
|
|
|
if step < 0:
|
|
|
step = 1
|
|
|
start = time.time()
|
|
|
|
|
|
- while not self.is_job_completed(workflow) and time.time() - start < timeout:
|
|
|
+ while job.is_running() and time.time() - start < timeout:
|
|
|
time.sleep(sleep)
|
|
|
sleep = sleep + step
|
|
|
- LOG.info('Checking status of %s...' % jobid)
|
|
|
- workflow = self.oozie.get_job(jobid)
|
|
|
- if not self.is_job_completed(workflow):
|
|
|
- logs = self.oozie.get_job_log(jobid)
|
|
|
- raise Exception("%s took too long to complete: %s" % (jobid, logs))
|
|
|
+ LOG.info('Checking status of %s...' % oozie_jobid)
|
|
|
+ job = cls.oozie.get_job(oozie_jobid)
|
|
|
+ LOG.info('Status: %s' % job)
|
|
|
|
|
|
- return workflow
|
|
|
+ if job.is_running():
|
|
|
+ logs = cls.oozie.get_job_log(oozie_jobid)
|
|
|
+ raise Exception("%s took too long to complete: %s" % (oozie_jobid, logs))
|
|
|
+ else:
|
|
|
+ LOG.info('Job duration %s: %d' % (job.id, time.time() - start))
|
|
|
|
|
|
- def is_job_completed(self, workflow):
|
|
|
- # Only for Workflows so far
|
|
|
- if workflow is None:
|
|
|
- return False
|
|
|
- LOG.info(workflow.status)
|
|
|
- return workflow.status in OozieServerProvider.STATES_WF_COMPLETION
|
|
|
+ return job
|
|
|
|
|
|
@classmethod
|
|
|
def _start_oozie(cls, cluster):
|
|
|
@@ -98,6 +91,19 @@ class OozieServerProvider(object):
|
|
|
process = subprocess.Popen(args=args, env=env, cwd=cluster._tmpdir, stdin=subprocess.PIPE)
|
|
|
return process
|
|
|
|
|
|
+ @classmethod
|
|
|
+ def _reset_oozie(cls):
|
|
|
+ env = os.environ
|
|
|
+
|
|
|
+ args = ['rm', '-r', OozieServerProvider.OOZIE_HOME + '/data/oozie-db']
|
|
|
+ LOG.info("Executing %s, env %s" % (args, env))
|
|
|
+ subprocess.call(args, env=env)
|
|
|
+
|
|
|
+ args = [OozieServerProvider.OOZIE_HOME + '/bin/ooziedb.sh', 'create', '-sqlfile', 'oozie.sql', '-run']
|
|
|
+ LOG.info("Executing %s, env %s" % (args, env))
|
|
|
+ subprocess.call(args, env=env)
|
|
|
+
|
|
|
+
|
|
|
@classmethod
|
|
|
def _get_shared_oozie_server(cls):
|
|
|
global _oozie_running
|
|
|
@@ -105,16 +111,29 @@ class OozieServerProvider(object):
|
|
|
|
|
|
_oozie_lock.acquire()
|
|
|
if not _oozie_running:
|
|
|
+ LOG.info('\nStarting a Mini Oozie. Requires "tools/jenkins/jenkins.sh" to be previously ran.\n')
|
|
|
+ LOG.info('See https://issues.cloudera.org/browse/HUE-861\n')
|
|
|
+
|
|
|
finish = (
|
|
|
OOZIE_URL.set_for_testing("http://localhost:%s/oozie" % OozieServerProvider.OOZIE_TEST_PORT),
|
|
|
)
|
|
|
|
|
|
cluster = pseudo_hdfs4.shared_cluster()
|
|
|
+ cls._reset_oozie()
|
|
|
+ p = cls._start_oozie(cluster)
|
|
|
+
|
|
|
+ def kill():
|
|
|
+ LOG.info("Killing Oozie server (pid %d)." % p.pid)
|
|
|
+ os.kill(p.pid, 9)
|
|
|
+ p.wait()
|
|
|
+ atexit.register(kill)
|
|
|
|
|
|
start = time.time()
|
|
|
started = False
|
|
|
sleep = 0.01
|
|
|
- while not started and time.time() - start < 20.0:
|
|
|
+
|
|
|
+ while not started and time.time() - start < 30.0:
|
|
|
+ status = None
|
|
|
try:
|
|
|
LOG.info('Check Oozie status...')
|
|
|
status = get_oozie().get_oozie_status()
|
|
|
@@ -123,8 +142,8 @@ class OozieServerProvider(object):
|
|
|
break
|
|
|
time.sleep(sleep)
|
|
|
sleep *= 2
|
|
|
- except:
|
|
|
- LOG.info('Oozie server status not NORMAL yet.')
|
|
|
+ except Exception, e:
|
|
|
+ LOG.info('Oozie server status not NORMAL yet: %s - %s' % (status, e))
|
|
|
time.sleep(sleep)
|
|
|
sleep *= 2
|
|
|
pass
|
|
|
@@ -143,39 +162,7 @@ class OozieServerProvider(object):
|
|
|
return get_oozie(), callback
|
|
|
|
|
|
|
|
|
-class TestoozieWithHadoop(OozieServerProvider):
|
|
|
+class TestMiniOozie(OozieServerProvider):
|
|
|
+
|
|
|
def test_oozie_status(self):
|
|
|
- assert_equal(self.oozie.get_oozie_status()['systemMode'], 'NORMAL')
|
|
|
-
|
|
|
- def test_oozie_example(self):
|
|
|
- jobid = None
|
|
|
-
|
|
|
- try:
|
|
|
- self.cluster.fs.setuser('hue')
|
|
|
- self.cluster.fs.create_home_dir()
|
|
|
- home = self.cluster.fs.get_home_dir()
|
|
|
-
|
|
|
- self.cluster.put(OozieServerProvider + '/oozie/examples', home)
|
|
|
- self.cluster.put(OozieServerProvider + '/oozie/examples/input-data/text/data.txt', home)
|
|
|
- self.cluster.chmod(home, '0777')
|
|
|
-
|
|
|
- application_path = self.cluster._fs_default_name + home + '/examples/apps/map-reduce'
|
|
|
- assert_true(self.cluster.fs.exists(home + '/examples/apps/map-reduce'))
|
|
|
-
|
|
|
- jobid = self.oozie.submit_workflow(application_path, {
|
|
|
- 'nameNode': self.cluster._fs_default_name,
|
|
|
- 'jobTracker': self.cluster.mapred_job_tracker,
|
|
|
- 'queueName': 'default',
|
|
|
- 'examplesRoot': 'examples',
|
|
|
- 'outputDir': 'test-out'})
|
|
|
- assert_true(jobid)
|
|
|
-
|
|
|
- self.oozie.job_control(jobid, 'start')
|
|
|
- workflow = self.wait_until_completion(jobid)
|
|
|
-
|
|
|
- assert_equal('SUCCEEDED', workflow.status)
|
|
|
- assert_true(self.cluster.fs.exists('output-data'))
|
|
|
- except:
|
|
|
- if jobid is not None:
|
|
|
- print self.oozie.get_job_log(jobid)
|
|
|
- raise
|
|
|
+ assert_equal(get_oozie().get_oozie_status()['systemMode'], 'NORMAL')
|