Convert job ID to uuid

In some CI runs it was observed that unexpected results were being
returned for middleware jobs. This commit converts our job ids
from being monotonically incrementing integer to proper uuid so
that the job id that client is trying to track is guaranteed to
uniquely identify it regardless of which HA node is being
connected to.

Author: anodos325

src/middlewared/middlewared/job.py

@@ -1,6 +1,5 @@
 import asyncio
 import contextlib
-from collections import OrderedDict
 import copy
 import enum
 import errno
@@ -17,6 +16,7 @@
 from middlewared.pipe import Pipes
 from middlewared.utils.privilege import credential_is_limited_to_own_jobs, credential_has_full_admin
 from middlewared.utils.time_utils import utc_now
+from uuid import uuid4
 
 
 logger = logging.getLogger(__name__)
@@ -225,8 +225,7 @@ class JobsDeque:
 
     def __init__(self, maxlen=1000):
         self.maxlen = maxlen
-        self.count = 0
-        self.__dict = OrderedDict()
+        self.__dict = {}
         with contextlib.suppress(FileNotFoundError):
             shutil.rmtree(LOGS_DIR)
 
@@ -244,7 +243,6 @@ def _get_next_id(self):
         return self.count
 
     def add(self, job):
-        job.set_id(self._get_next_id())
         if len(self.__dict) > self.maxlen:
             for old_job_id, old_job in self.__dict.items():
                 if old_job.state in (State.SUCCESS, State.FAILED, State.ABORTED):
@@ -291,7 +289,7 @@ def __init__(self, middleware, method_name, serviceobj, method, args, options, p
         self.app = app
         self.audit_callback = audit_callback
 
-        self.id = None
+        self.id = str(uuid4())
         self.lock = None
         self.result = None
         self.error = None
@@ -377,9 +375,6 @@ def get_lock_name(self):
                                 errno.EINVAL)
         return lock_name
 
-    def set_id(self, id_):
-        self.id = id_
-
     def set_result(self, result):
         self.result = result
 

src/middlewared/middlewared/service/core_service.py

@@ -109,7 +109,7 @@ def __job_by_credential_and_id(self, credential, job_id, access):
     @filterable
     @filterable_returns(Dict(
         'job',
-        Int('id'),
+        Str('id'),
         Str('method'),
         List('arguments'),
         Bool('transient'),
@@ -178,7 +178,7 @@ def get_jobs(self, app, filters, options):
         return jobs
 
     @no_authz_required
-    @accepts(Int('id'), Str('filename'), Bool('buffered', default=False))
+    @accepts(Str('id'), Str('filename'), Bool('buffered', default=False))
     @pass_app(rest=True)
     async def job_download_logs(self, app, id_, filename, buffered):
         """
@@ -195,15 +195,15 @@ async def job_download_logs(self, app, id_, filename, buffered):
         return (await self._download(app, 'filesystem.get', [job.logs_path], filename, buffered))[1]
 
     @no_authz_required
-    @accepts(Int('id'))
+    @accepts(Str('id'))
     @job()
     async def job_wait(self, job, id_):
         target_job = self.__job_by_credential_and_id(job.credentials, id_, JobAccess.READ)
 
         return await job.wrap(target_job)
 
     @private
-    @accepts(Int('id'), Dict(
+    @accepts(Str('id'), Dict(
         'job-update',
         Dict('progress', additional_attrs=True),
     ))