Converts tasking system to use Django models

.travis.yml

@@ -8,6 +8,10 @@ addons:
 install:
     - "pip install -r test_requirements.txt"
     - "pushd app/ && python setup.py develop && popd"
+    - "pushd common/ && python setup.py develop && popd"
+    - "pushd exceptions/ && python setup.py develop && popd"
+    - "pushd plugin/ && python setup.py develop && popd"
+    - "pushd tasking/ && python setup.py develop && popd"
 before_script:
     - psql -U postgres -c "CREATE USER pulp WITH SUPERUSER LOGIN;"
     - psql -U postgres -c "CREATE DATABASE pulp OWNER pulp;"

app/etc/pulp/server.yaml

@@ -42,3 +42,16 @@
 #     pulp.app:
 #       level: INFO
 
+# Broker configuration
+#
+# `broker`: Broker configuration for Pulp. By default, Pulp uses the Qpid
+# broker without SSL. It is possible to change this by providing a different
+# broker string. The remaining options are used to configure SSL settings.
+#
+# broker:
+#   url: qpid://localhost/
+#   celery_require_ssl: false
+#   cacert: /etc/pki/pulp/qpid/ca.crt
+#   keyfile: /etc/pki/pulp/qpid/client.crt
+#   certfile: /etc/pki/pulp/qpid/client.crt
+#   login_method:

app/pulp/app/models/catalog.py

@@ -22,11 +22,11 @@ class DownloadCatalog(Model):
     Relations:
 
     :cvar artifact: The artifact that is expected to be present at ``url``.
-    :type artifact: pulp.platform.models.Artifact
+    :type artifact: pulp.app.models.Artifact
 
     :cvar importer: The importer that contains the configuration necessary
                     to access ``url``.
-    :type importer: pulp.platform.models.Importer
+    :type importer: pulp.app.models.Importer
     """
     # Although there is a Django field for URLs based on CharField, there is
     # not technically any limit on URL lengths so it's simplier to allow any

app/pulp/app/models/progress.py

@@ -7,7 +7,7 @@
 from django.db import models
 
 from pulp.app.models import Model, Task
-from pulp.app.tasks.task_system import get_current_task_id
+from pulp.tasking import get_current_task_id
 
 
 _logger = logging.getLogger(__name__)
@@ -44,14 +44,14 @@ class ProgressReport(Model):
     SKIPPED = 'skipped'
     RUNNING = 'running'
     COMPLETED = 'completed'
-    ERRORED = 'errored'
+    FAILED = 'failed'
     CANCELED = 'canceled'
     STATES = (
         (WAITING, 'Waiting'),
         (SKIPPED, 'Skipped'),
         (RUNNING, 'Running'),
         (COMPLETED, 'Completed'),
-        (ERRORED, 'Errored'),
+        (FAILED, 'Failed'),
         (CANCELED, 'Canceled')
     )
     message = models.TextField()
@@ -88,9 +88,9 @@ def __enter__(self):
 
     def __exit__(self, type, value, traceback):
         """
-        Update the progress report state to COMPLETED or ERRORED.
+        Update the progress report state to COMPLETED or FAILED.
 
-        If an exception occurs the progress report state is saved as ERRORED and the exception is
+        If an exception occurs the progress report state is saved as FAILED and the exception is
         not suppressed. If the context manager exited without exception the progress report state
         is saved as COMPLETED.
 
@@ -100,7 +100,7 @@ def __exit__(self, type, value, traceback):
             self.state = self.COMPLETED
             self.save()
         else:
-            self.state = self.ERRORED
+            self.state = self.FAILED
             self.save()
 
 
@@ -122,13 +122,13 @@ class ProgressSpinner(ProgressReport):
         >>> metadata_progress.save()
 
     The ProgressSpinner() is a context manager that provides automatic state transitions for the
-    RUNNING COMPLETED and ERRORED states. Use it as follows:
+    RUNNING COMPLETED and FAILED states. Use it as follows:
 
         >>> spinner = ProgressSpinner('Publishing Metadata')
         >>> with spinner:
         >>>     # spinner is at 'running'
         >>>     publish_metadata()
-        >>>     # spinner is at 'completed' if no exception or 'errored' if an exception was raised
+        >>>     # spinner is at 'completed' if no exception or 'failed' if an exception was raised
 
     You can also use this short form:
 
@@ -162,7 +162,7 @@ class ProgressBar(ProgressReport):
         >>> progress_bar.save()
 
     The ProgressBar() is a context manager that provides automatic state transitions for the RUNNING
-    COMPLETED and ERRORED states. The increment() method can be called in the loop as work is
+    COMPLETED and FAILED states. The increment() method can be called in the loop as work is
     completed. Use it as follows:
 
         >>> progress_bar = ProgressBar(message='Publishing files', total=len(files_iterator))
@@ -172,7 +172,7 @@ class ProgressBar(ProgressReport):
         >>>     for file in files_iterator:
         >>>         handle(file)
         >>>         progress_bar.increment()  # increments and saves
-        >>> # progress_bar is at 'completed' if no exception or 'errored' if an exception was raised
+        >>> # progress_bar is at 'completed' if no exception or 'failed' if an exception was raised
 
     A convenience method called iter() allows you to avoid calling increment() directly:
 

app/pulp/app/models/task.py

@@ -27,7 +27,7 @@ class ReservedResource(Model):
     resource = models.TextField()
 
     task = models.OneToOneField("Task")
-    worker = models.ForeignKey("Worker")
+    worker = models.ForeignKey("Worker", on_delete=models.CASCADE, related_name="reservations")
 
 
 class Worker(Model):
@@ -43,7 +43,7 @@ class Worker(Model):
     :type last_heartbeat: models.DateTimeField
     """
     name = models.TextField(db_index=True, unique=True)
-    last_heartbeat = models.DateTimeField()
+    last_heartbeat = models.DateTimeField(auto_now=True)
 
 
 class TaskLock(Model):
@@ -92,8 +92,8 @@ class Task(Model):
     :cvar finished_at: The time the task finished executing
     :type finished_at: models.DateTimeField
 
-    :cvar error: Collection of errors that might have occurred while task was running
-    :type error: models.JSONField
+    :cvar non_fatal_errors: Dictionary of non-fatal errors that occurred while task was running.
+    :type non_fatal_errors: models.JSONField
 
     :cvar result: Return value of the task
     :type result: models.JSONField
@@ -113,7 +113,7 @@ class Task(Model):
     RUNNING = 'running'
     SUSPENDED = 'suspended'
     COMPLETED = 'completed'
-    ERRORED = 'errored'
+    FAILED = 'failed'
     CANCELED = 'canceled'
     STATES = (
         (WAITING, 'Waiting'),
@@ -122,7 +122,7 @@ class Task(Model):
         (RUNNING, 'Running'),
         (SUSPENDED, 'Suspended'),
         (COMPLETED, 'Completed'),
-        (ERRORED, 'Errored'),
+        (FAILED, 'Failed'),
         (CANCELED, 'Canceled')
     )
     group = models.UUIDField(null=True)
@@ -131,11 +131,11 @@ class Task(Model):
     started_at = models.DateTimeField(null=True)
     finished_at = models.DateTimeField(null=True)
 
-    error = JSONField()
+    non_fatal_errors = JSONField()
     result = JSONField()
 
     parent = models.ForeignKey("Task", null=True, related_name="spawned_tasks")
-    worker = models.ForeignKey("Worker", null=True)
+    worker = models.ForeignKey("Worker", null=True, related_name="tasks")
 
 
 class TaskTag(Model):

app/pulp/app/settings.py

@@ -177,6 +177,14 @@
             },
         }
     },
+    'broker': {
+        'url': 'qpid://localhost/',
+        'celery_require_ssl': False,
+        'cacert': '/etc/pki/pulp/qpid/ca.crt',
+        'keyfile': '/etc/pki/pulp/qpid/client.crt',
+        'certfile': '/etc/pki/pulp/qpid/client.crt',
+        'login_method': None
+    },
 }
 
 

common/pulp/common/constants.py

@@ -0,0 +1,17 @@
+# Call States
+
+CALL_WAITING_STATE = 'waiting'
+CALL_SKIPPED_STATE = 'skipped'
+CALL_ACCEPTED_STATE = 'accepted'
+CALL_RUNNING_STATE = 'running'
+CALL_SUSPENDED_STATE = 'suspended'
+CALL_FINISHED_STATE = 'finished'
+CALL_ERROR_STATE = 'error'
+CALL_CANCELED_STATE = 'canceled'
+
+CALL_INCOMPLETE_STATES = (CALL_WAITING_STATE, CALL_ACCEPTED_STATE, CALL_RUNNING_STATE,
+                          CALL_SUSPENDED_STATE)
+CALL_COMPLETE_STATES = (CALL_SKIPPED_STATE, CALL_FINISHED_STATE, CALL_ERROR_STATE,
+                        CALL_CANCELED_STATE)
+CALL_STATES = (CALL_WAITING_STATE, CALL_SKIPPED_STATE, CALL_ACCEPTED_STATE, CALL_RUNNING_STATE,
+               CALL_SUSPENDED_STATE, CALL_FINISHED_STATE, CALL_ERROR_STATE, CALL_CANCELED_STATE)

common/pulp/common/error_codes.py

@@ -0,0 +1,160 @@
+from collections import namedtuple
+from gettext import gettext as _
+
+
+Error = namedtuple('Error', ['code', 'message', 'required_fields'])
+"""
+The error named tuple has 4 components:
+code: The 7 character uniquely identifying code for this error, 3 A-Z identifying the module
+      followed by 4 numeric characters for the msg id.  All general pulp server errors start
+      with PLP
+message: The message that will be printed for this error
+required_files: A list of required fields for printing the message
+"""
+
+# The PLP0000 error is to wrap non-pulp exceptions
+PLP0000 = Error("PLP0000",
+                "%(message)s", ['message'])
+PLP0001 = Error("PLP0001",
+                _("A general pulp exception occurred"), [])
+PLP0002 = Error(
+    "PLP0002",
+    _("Errors occurred updating bindings on consumers for repo %(repo_id)s and distributor "
+      "%(distributor_id)s"),
+    ['repo_id', 'distributor_id'])
+PLP0003 = Error("PLP0003",
+                _("Errors occurred removing bindings on consumers while deleting a distributor for "
+                  "repo %(repo_id)s and distributor %(distributor_id)s"),
+                ['repo_id', 'distributor_id'])
+PLP0004 = Error("PLP0004",
+                _("Errors occurred creating bindings for the repository group %(group_id)s.  "
+                  "Binding creation was attempted for the repository %(repo_id)s and "
+                  "distributor %(distributor_id)s"),
+                ['repo_id', 'distributor_id', 'group_id'])
+
+PLP0005 = Error("PLP0005",
+                _("Errors occurred deleting bindings for the repository group %(group_id)s.  "
+                  "Binding deletion was attempted for the repository %(repo_id)s and "
+                  "distributor %(distributor_id)s"),
+                ['repo_id', 'distributor_id', 'group_id'])
+PLP0006 = Error("PLP0006", _("Errors occurred while updating the distributor configuration for "
+                             "repository %(repo_id)s"),
+                ['repo_id'])
+PLP0007 = Error("PLP0007",
+                _("Error occurred while cascading delete of repository %(repo_id)s to distributor "
+                  "bindings associated with it."),
+                ['repo_id'])
+PLP0008 = Error("PLP0008",
+                _("Error raising error %(code)s.  "
+                  "The field [%(field)s] was not included in the error_data."),
+                ['code', 'field'])
+PLP0009 = Error("PLP0009", _("Missing resource(s): %(resources)s"), ['resources'])
+PLP0010 = Error("PLP0010", _("Conflicting operation reasons: %(reasons)s"), ['reasons'])
+PLP0011 = Error("PLP0011", _("Operation timed out after: %(timeout)s"), ['timeout'])
+PLP0012 = Error("PLP0012", _("Operation postponed"), [])
+PLP0014 = Error("PLP0014", _('Operation not implemented: %(operation_name)s'), ['operation_name'])
+PLP0015 = Error("PLP0015", _('Invalid properties: %(properties)s'), ['properties'])
+PLP0016 = Error("PLP0016", _('Missing values for: %(properties)s'), ['properties'])
+PLP0017 = Error("PLP0017", _('Unsupported properties: %(properties)s'), ['properties'])
+PLP0018 = Error("PLP0018", _('Duplicate resource: %(resource_id)s'), ['resource_id'])
+PLP0019 = Error("PLP0019", _('Pulp only accepts input encoded in UTF-8: %(value)s'), ['value'])
+PLP0020 = Error("PLP0020",
+                _("Errors occurred installing content for the consumer group %(group_id)s."),
+                ['group_id'])
+PLP0021 = Error("PLP0021",
+                _("Errors occurred updating content for the consumer group %(group_id)s."),
+                ['group_id'])
+PLP0022 = Error("PLP0022",
+                _("Errors occurred uninstalling content for the consumer group %(group_id)s."),
+                ['group_id'])
+PLP0023 = Error("PLP0023", _("Task is already in a complete state: %(task_id)s"), ['task_id'])
+PLP0024 = Error("PLP0024",
+                _("There are no Celery workers found in the system for reserved task work. "
+                  "Please ensure that there is at least one Celery worker running, and that the "
+                  "celerybeat service is also running."),
+                [])
+PLP0025 = Error("PLP0025", _("Authentication failed."), [])
+PLP0026 = Error(
+    "PLP0026", _("Permission denied: user %(user)s cannot perform %(operation)s."),
+    ['user', 'operation'])
+PLP0027 = Error(
+    "PLP0027", _("Authentication with username %(user)s failed: invalid SSL certificate."),
+    ['user'])
+PLP0028 = Error(
+    "PLP0028", _("Authentication with username %(user)s failed: invalid oauth credentials."),
+    ['user'])
+PLP0029 = Error(
+    "PLP0029",
+    _("Authentication with username %(user)s failed: preauthenticated remote user is missing."),
+    ['user'])
+PLP0030 = Error(
+    "PLP0030",
+    _("Authentication with username %(user)s failed: invalid username or password"), ['user'])
+PLP0031 = Error("PLP0031", _("Content source %(id)s could not be found at %(url)s"), ['id', 'url'])
+PLP0032 = Error(
+    "PLP0032", _("Task %(task_id)s encountered one or more failures during execution."),
+    ['task_id'])
+PLP0034 = Error("PLP0034", _("The distributor %(distributor_id)s indicated a failed response when "
+                             "publishing repository %(repo_id)s."),
+                ['distributor_id', 'repo_id'])
+PLP0037 = Error(
+    "PLP0037",
+    _("Content import of %(path)s failed - must be an existing file."),
+    ['path'])
+PLP0038 = Error("PLP0038", _("The unit model with id %(model_id)s and class "
+                             "%(model_class)s failed to register. Another model has already "
+                             "been registered with the same id."), ['model_id', 'model_class'])
+PLP0040 = Error("PLP0040", _("Database 'seeds' config must include at least one hostname:port "
+                             "value. Refer to /etc/pulp/server.conf for proper use."), [])
+PLP0041 = Error("PLP0041", _("Database 'replica_set' config must be specified when more than one "
+                             "seed is provided. Refer to /etc/pulp/server.conf for proper use."),
+                [])
+PLP0042 = Error("PLP0042", _("This request is forbidden."), [])
+PLP0043 = Error("PLP0043", _("Database 'write_concern' config can only be 'majority' or 'all'. "
+                             "Refer to /etc/pulp/server.conf for proper use."), [])
+PLP0044 = Error("PLP0044", _("The target importer does not support the types from the source"), [])
+PLP0045 = Error("PLP0045", _("The repository cannot be exported because some units are "
+                             "not downloaded."), [])
+PLP0046 = Error("PLP0046", _("The repository group cannot be exported because these repos have "
+                             "units that are not downloaded: %(repos)s"), ['repos'])
+PLP0047 = Error("PLP0047", _("The importer %(importer_id)s indicated a failed response when "
+                             "uploading %(unit_type)s unit to repository %(repo_id)s."),
+                ['importer_id', 'unit_type', 'repo_id'])
+PLP0048 = Error("PLP0048", _("The file is expected to be present, but is not, for unit %(unit)s"),
+                ['unit'])
+
+# Create a section for general validation errors (PLP1000 - PLP2999)
+# Validation problems should be reported with a general PLP1000 error with a more specific
+# error message nested inside of it.
+PLP1000 = Error("PLP1000", _("A validation error occurred."), [])
+PLP1001 = Error("PLP1001", _("The consumer %(consumer_id)s does not exist."), ['consumer_id'])
+PLP1002 = Error("PLP1002", _("The field %(field)s must have a value specified."), ['field'])
+PLP1003 = Error(
+    "PLP1003",
+    _("The value specified for the field %(field)s must be made up of letters, numbers, "
+      "underscores, or hyphens with no spaces."),
+    ['field'])
+PLP1004 = Error(
+    "PLP1004",
+    _("An object of type %(type)s already exists in the database with an id of %(object_id)s"),
+    ['type', 'object_id'])
+PLP1005 = Error("PLP1005", _("The checksum type '%(checksum_type)s' is unknown."),
+                ['checksum_type'])
+PLP1006 = Error(
+    "PLP1006", _("The value specified for the field %(field)s may not start with %(value)s."),
+    ['field', 'value'])
+PLP1007 = Error("PLP1007", _("The relative path specified must not point outside of the parent"
+                             " directory:  %(path)s"), ['path'])
+PLP1008 = Error("PLP1008", _("The importer type %(importer_type_id)s does not exist"),
+                ['importer_type_id'])
+PLP1009 = Error("PLP1009", _("The request body does not contain valid JSON"), [])
+PLP1010 = Error("PLP1010", _("Provided value %(value)s for field %(field)s must be of type "
+                             "%(field_type)s."), ["value", "field", "field_type"])
+PLP1011 = Error("PLP1011", _("Invalid task state passed to purge: %(state)s."), ["state"])
+PLP1012 = Error("PLP1012", _("No task state given to parameters list for delete."), [])
+PLP1013 = Error("PLP1013", _("Checksum does not match calculated value."), [])
+PLP1014 = Error("PLP1014", _("The configuration value for the key '%(key)s' in "
+                             "section '%(section)s' is not valid for the following "
+                             "reason: %(reason)s"), ["key", "section", "reason"])
+PLP1015 = Error("PLP1015", _("The JSON data must be of type '%(data_type)s'."),
+                ['data_type'])

exceptions/pulp/exceptions/__init__.py

@@ -0,0 +1,2 @@
+from .base import PulpCodedException, PulpException
+from .http import MissingResource