Type annotations

env/pyvenv.cfg

@@ -0,0 +1,3 @@
+home = /usr/bin
+include-system-site-packages = false
+version = 3.5.2

jobManager.py

@@ -1,4 +1,19 @@
 from __future__ import print_function
+import copy
+import time
+import logging
+import threading
+from typing import Optional
+from config import Config
+from tangoObjects import TangoQueue
+from worker import Worker
+from preallocator import Preallocator
+from jobQueue import JobQueue
+# from tango import *
+import tango
+from datetime import datetime
+from builtins import str
+
 #
 # JobManager - Thread that assigns jobs to worker threads
 #
@@ -13,21 +28,11 @@
 from builtins import object
 from future import standard_library
 standard_library.install_aliases()
-from builtins import str
-import threading, logging, time, copy
 
-from datetime import datetime
-from tango import *
-from jobQueue import JobQueue
-from preallocator import Preallocator
-from worker import Worker
-
-from tangoObjects import TangoQueue
-from config import Config
 
 class JobManager(object):
 
-    def __init__(self, queue):
+    def __init__(self, queue: JobQueue) -> None:
         self.daemon = True
         self.jobQueue = queue
         self.preallocator = self.jobQueue.preallocator
@@ -37,19 +42,19 @@ def __init__(self, queue):
         self.nextId = 10000
         self.running = False
 
-    def start(self):
+    def start(self) -> None:
         if self.running:
             return
         thread = threading.Thread(target=self.__manage)
         thread.daemon = True
         thread.start()
 
-    def run(self):
+    def run(self) -> None:
         if self.running:
             return
         self.__manage()
 
-    def _getNextID(self):
+    def _getNextID(self) -> int:
         """ _getNextID - returns next ID to be used for a job-associated
         VM.  Job-associated VM's have 5-digit ID numbers between 10000
         and 99999.
@@ -60,78 +65,83 @@ def _getNextID(self):
             self.nextId = 10000
         return id
 
-    def __manage(self):
+    def __manage(self) -> None:
         self.running = True
         while True:
-            id = self.jobQueue.getNextPendingJob()
-
-            if id:
-                job = self.jobQueue.get(id)
-
-                # job could no longer exist if it was completed by someone else
-                if job == None:
-                    continue
-
-                if not job.accessKey and Config.REUSE_VMS:
-                    id, vm = self.jobQueue.getNextPendingJobReuse(id)
-                    job = self.jobQueue.get(id)
-                    if job == None:
-                        continue
-
-                try:
-                    # Mark the job assigned
-                    self.jobQueue.assignJob(job.id)
-                    # if the job has specified an account
-                    # create an VM on the account and run on that instance
-                    if job.accessKeyId:
-                        from vmms.ec2SSH import Ec2SSH
-                        vmms = Ec2SSH(job.accessKeyId, job.accessKey)
-                        newVM = copy.deepcopy(job.vm)
-                        newVM.id = self._getNextID()
-                        preVM = vmms.initializeVM(newVM)
+            # Gets the next pending job/ id
+            # Blocks until we get a next job
+            job = self.jobQueue.getNextPendingJob()
+
+            if not job.accessKey and Config.REUSE_VMS:
+                vm = self.jobQueue.reuseVM(job)
+
+            try:
+                # Mark the job assigned
+                job.makeAssigned()
+                # if the job has specified an account
+                # create an VM on the account and run on that instance
+                if job.accessKeyId:
+                    from vmms.ec2SSH import Ec2SSH
+                    vmms = Ec2SSH(job.accessKeyId, job.accessKey)
+                    newVM = copy.deepcopy(job.vm)
+                    newVM.id = self._getNextID()
+                    preVM = vmms.initializeVM(newVM)
+
+                else:
+                    # Try to find a vm on the free list and allocate it to
+                    # the worker if successful.
+                    if Config.REUSE_VMS:
+                        preVM = vm
                     else:
-                        # Try to find a vm on the free list and allocate it to
-                        # the worker if successful.
-                        if Config.REUSE_VMS:
-                            preVM = vm
-                        else:
-                            preVM = self.preallocator.allocVM(job.vm.name)
-                        vmms = self.vmms[job.vm.vmms]  # Create new vmms object
-
-                    # Now dispatch the job to a worker
-                    self.log.info("Dispatched job %s:%d to %s [try %d]" %
-                                  (job.name, job.id, preVM.name, job.retries))
-                    job.appendTrace(
-                        "%s|Dispatched job %s:%d [try %d]" %
-                        (datetime.utcnow().ctime(), job.name, job.id, job.retries))
-
-                    Worker(
-                        job,
-                        vmms,
-                        self.jobQueue,
-                        self.preallocator,
-                        preVM
-                    ).start()
-
-                except Exception as err:
-                    self.jobQueue.makeDead(job.id, str(err))
+                        preVM = self.preallocator.allocVM(job.vm.name)
+                    vmms = self.vmms[job.vm.vmms]  # Create new vmms object
 
+                if (preVM is not None):
+                    # Log the preallocated vm only in the case where a vm 
+                    # was allocated
+                    self.log.info("Dispatched job %s:%d to %s [try %d]" %
+                              (job.name, job.id, preVM.name, job.retries))
+                else:
+                    self.log.info("No VM preallocated for job %s:%d [try %d]" %
+                              (job.name, job.id, job.retries))
+
+
+
+                job.appendTrace(
+                   "%s|Dispatched job %s:%d [try %d]" %
+                   (datetime.utcnow().ctime(), job.name, job.id, job.retries))
+
+                # Now dispatch the job to a worker 
+                Worker(
+                    job,
+                    vmms,
+                    self.jobQueue,
+                    self.preallocator,
+                    preVM
+                 ).start()
+            except Exception as err:
+                print("EXCEPTION")
+                print(str(err))
+
             # Sleep for a bit and then check again
             time.sleep(Config.DISPATCH_PERIOD)
 
+    @staticmethod
+    def runJobManager(mock_vmms: Optional[str]=None) -> None:
+        t = tango.TangoServer(mock_vmms)
+        t.log.debug("Resetting Tango VMs")
+        t.resetTango(t.preallocator.vmms)
+        for key in t.preallocator.machines.keys():
+            t.preallocator.machines.set(key, [[], TangoQueue(key)])
+        jobs = JobManager(t.jobQueue)
+        print("Starting the stand-alone Tango JobManager", mock_vmms)
+        jobs.run()
+
 
 if __name__ == "__main__":
 
     if not Config.USE_REDIS:
         print("You need to have Redis running to be able to initiate stand-alone\
          JobManager")
     else:
-        tango = TangoServer()
-        tango.log.debug("Resetting Tango VMs")
-        tango.resetTango(tango.preallocator.vmms)
-        for key in tango.preallocator.machines.keys():
-            tango.preallocator.machines.set(key, [[], TangoQueue(key)])
-        jobs = JobManager(tango.jobQueue)
-
-        print("Starting the stand-alone Tango JobManager")
-        jobs.run()
+        JobManager.runJobManager()