Convert async workethreads to use classes

This refactors the worker thread class and image request class.
Now there is one general submit function in the worker thread
class.  The ImageRequest class has been replaced with a set
of AsyncRequests classes for each operation.  This makes
the code a bit cleaner and the inputs for each async tasks
are more clearly defined.

Author: scanon

imagegw/shifter_imagegw/config.py

@@ -71,9 +71,6 @@ def __init__(self, data=None):
             data = config['Platforms'][platform]
             self.Platforms[platform] = Platform(data)
 
-        self.Locations = config['Locations']
-        self.Platofrms = config['Platforms']
-
         loglevel = config.get('LogLevel', 'info').lower()
         loglevel_map = {
             'debug': logging.DEBUG,

imagegw/shifter_imagegw/imagemngr.py

@@ -34,6 +34,9 @@
 from pymongo import MongoClient
 import pymongo.errors
 from shifter_imagegw.imageworker import WorkerThreads
+from shifter_imagegw.imageworker import PullRequest
+from shifter_imagegw.imageworker import ImportRequest
+from shifter_imagegw.imageworker import ExpireRequest
 from shifter_imagegw.config import Config
 import grp
 
@@ -170,7 +173,7 @@ def _isadmin(self, session, system=None):
         admins = self.platforms[system].admins
         user = session.user
         if user in admins:
-            self.logger.debug('user {user} is an admin')
+            self.logger.debug(f'user {user} is an admin')
             return True
         return False
 
@@ -509,7 +512,14 @@ def pull(self, session, image):
             request['session'] = session
             self.logger.debug("Calling do pull with queue="
                               f"{request['system']}")
-            self.workers.dopull(ident, request)
+            pr = PullRequest(self.config,
+                             session.system,
+                             request['tag'],
+                             ident,
+                             session,
+                             useracl=request['userACL'],
+                             groupacl=request['groupACL'])
+            self.workers.submit(pr)
 
             memo = "pull request queued " \
                    f"s={request['system']} tag={request['tag']}"
@@ -561,7 +571,13 @@ def mngrimport(self, session, image):
         request['session'] = session
         self.logger.debug("Calling wrkimport with queue="
                           f"{request['system']}")
-        self.workers.dowrkimport(ident, request)
+        ir = ImportRequest(self.config,
+                           session.system,
+                           image['tag'],
+                           ident,
+                           session,
+                           image['filepath'])
+        self.workers.submit(ir)
 
         memo = "import request queued " \
                f"s={request['system']} tag={request['tag']}"
@@ -774,8 +790,13 @@ def expire_id(self, rec, ident):
         """ Helper function to expire by id """
         memo = f"Calling do expire id={ident}"
         self.logger.debug(memo)
+        er = ExpireRequest(self.config,
+                           rec['system'],
+                           rec['tag'],
+                           rec['id'],
+                           ident)
 
-        self.workers.doexpire(ident, rec)
+        self.workers.submit(er)
         self.logger.info("expire request queued "
                          f"s={rec['system']} tag={ident}")
 
@@ -795,7 +816,13 @@ def expire(self, session, image):
         memo = "Calling do expire with " \
                f"queue={image['system']} id={ident}"
         self.logger.debug(memo)
-        self.workers.doexpire(ident, rec)
+        er = ExpireRequest(self.config,
+                           rec['system'],
+                           rec['tag'],
+                           rec['id'],
+                           ident)
+
+        self.workers.submit(er)
 
         memo = "expire request queued " \
                f"s={image['system']} tag={image['tag']}"

imagegw/shifter_imagegw/transfer.py

@@ -60,9 +60,9 @@ def _exec_and_log(cmd, logger):
     stderr = bstderr.decode("utf-8")
     if logger is not None:
         if stdout is not None and len(stdout) > 0:
-            logger.debug("{cmd[0]} stdout: {stdout.strip()}")
+            logger.debug(f"{cmd[0]} stdout: {stdout.strip()}")
         if stderr is not None and len(stderr) > 0:
-            logger.error("{cmd[0]} stdout: {stdoerr.strip()}")
+            logger.error(f"{cmd[0]} stderr: {stderr.strip()}")
     return proc.returncode
 
 
@@ -84,9 +84,9 @@ def _get_stdout_and_log(cmd, logger=None):
     stderr = bstderr.decode("utf-8")
     if logger is not None:
         if stdout is not None and len(stdout) > 0:
-            logger.debug("{cmd[0]} stdout: {stdout.strip()}")
+            logger.debug(f"{cmd[0]} stdout: {stdout.strip()}")
         if stderr is not None and len(stderr) > 0:
-            logger.error("{cmd[0]} stdout: {stdoerr.strip()}")
+            logger.error(f"{cmd[0]} stdout: {stderr.strip()}")
             # push this error back to calling function so
             # it can be reported sensibly
             rerror = "%s %s" % (cmd[0], stderr.strip())

imagegw/test/auth_test.py

@@ -19,7 +19,7 @@
 from shifter_imagegw.config import Config
 import warnings
 warnings.filterwarnings('ignore', category=UserWarning, module='pymunge.raw')
-from shifter_imagegw.auth import authenticate
+from shifter_imagegw.auth import authenticate  # noqa
 
 
 def test_authenticate(mocker):

imagegw/test/imagemngr_test.py

@@ -1,4 +1,3 @@
-from shifter_imagegw.imageworker import WorkerThreads
 import os
 import pytest
 import time
@@ -8,13 +7,12 @@
 from copy import deepcopy
 from time import sleep
 from pymongo import MongoClient
-from shifter_imagegw.imagemngr import ImageMngr
 from multiprocessing.pool import ThreadPool
 from random import randint
 from shifter_imagegw.config import Config
-import warnings
-warnings.filterwarnings('ignore', category=UserWarning, module='pymunge.raw')
-from shifter_imagegw.auth import Session  # noqa
+from shifter_imagegw.models import Session
+from shifter_imagegw.imageworker import WorkerThreads
+from shifter_imagegw.imagemngr import ImageMngr
 
 """
 Shifter, Copyright (c) 2015, The Regents of the University of California,
@@ -42,6 +40,7 @@ def __init__(self, q):
         self.mode = 1
         self.q = q
         self.pools = ThreadPool(processes=2)
+        self.op = "pull"
 
     def set_mode(self, mode):
         self.mode = mode
@@ -58,7 +57,7 @@ def pull(self, request, updater):
         states = ('PULLING', 'EXAMINATION', 'CONVERSION', 'TRANSFER')
         for state in states:
             updater.update_status(state, state)
-            sleep(1)
+            sleep(0.1)
         ret = {
             'id': '%x' % randint(0, 100000),
             'entrypoint': ['./blah'],
@@ -73,7 +72,7 @@ def wrkimport(self, request, updater):
         states = ('HASHING', 'TRANSFER', 'READY')
         for state in states:
             updater.update_status(state, state)
-            sleep(1)
+            sleep(0.1)
         ret = {
             'id': '%x' % randint(0, 100000),
             'entrypoint': ['./blah'],
@@ -84,6 +83,15 @@ def wrkimport(self, request, updater):
         updater.update_status(state, state, ret)
         return ret
 
+    def submit(self, req):
+        req.updater.update_method = self.updater
+        if self.op == "pull":
+            self.pools.apply_async(self.pull, [req, req.updater],
+                                   {}, None, req.updater.failed)
+        elif self.op == "import":
+            self.pools.apply_async(self.wrkimport, [req, req.updater],
+                                   {}, None, req.updater.failed)
+
 
 @pytest.fixture(autouse=True)
 def set_path(monkeypatch):
@@ -642,9 +650,9 @@ def test_pull2(ctx):
     pr = ctx.pull
     # Do the pull
     session = ctx.session
-    rec1 = ctx.mtm.pull(session, pr)  # ,delay=False)
+    rec1 = ctx.mtm.pull(session, pr)
     pr['tag'] = ctx.tag2
-    rec2 = ctx.mtm.pull(session, pr)  # ,delay=False)
+    rec2 = ctx.mtm.pull(session, pr)
     assert rec1
     id1 = rec1['_id']
     assert rec2
@@ -947,6 +955,7 @@ def test_import(ctx):
     }
     # Do the pull
     session = ctx.session
+    ctx.mtm.workers.op = "import"
     rec = ctx.mtm.mngrimport(session, pr)  # ,delay=False)
     assert rec
     id = rec['_id']
@@ -968,26 +977,6 @@ def test_acl_update_denied(ctx):
     pass
 
 
-def test_expire_remote(ctx):
-    system = ctx.system
-    record = good_record(ctx)
-    # Create a fake record in mongo
-    id = ctx.images.insert_one(record).inserted_id
-    assert id
-    # Create a bogus image file
-    file, metafile = create_fakeimage(ctx, system, record['id'],
-                                      ctx.format)
-    session = ctx.admin_session
-    er = {'system': system, 'tag': ctx.tag, 'itype': ctx.itype}
-    rec = ctx.m.expire(session, er)  # ,delay=False)
-    assert rec
-    time.sleep(2)
-    state = ctx.m.get_state(id)
-    assert state == 'EXPIRED'
-    assert os.path.exists(file) is False
-    assert os.path.exists(metafile) is False
-
-
 def test_expire_local(ctx):
     record = good_record(ctx)
     system = 'systemb'
@@ -1000,7 +989,7 @@ def test_expire_local(ctx):
                                       ctx.format)
     session = ctx.admin_session
     er = {'system': system, 'tag': ctx.tag, 'itype': ctx.itype}
-    rec = ctx.m.expire(session, er)  # ,delay=False)
+    rec = ctx.m.expire(session, er)
     assert rec
     time.sleep(2)
     state = ctx.m.get_state(id)