Merge pull request pulp#2920 from dkliban/2496-worker-queryset

Problem: Stale worker documents present in the db

server/pulp/server/async/tasks.py

@@ -213,7 +213,7 @@ def _get_unreserved_worker():
     """
 
     # Build a mapping of queue names to Worker objects
-    workers_dict = dict((worker['name'], worker) for worker in Worker.objects())
+    workers_dict = dict((worker['name'], worker) for worker in Worker.objects.get_online())
     worker_names = workers_dict.keys()
     reserved_names = [r['worker_name'] for r in ReservedResource.objects.all()]
 

server/pulp/server/db/model/__init__.py

@@ -26,7 +26,8 @@
 from pulp.server.db.fields import ISO8601StringField, UTCDateTimeField
 from pulp.server.db.model.reaper_base import ReaperMixin
 from pulp.server.db.model import base
-from pulp.server.db.querysets import CriteriaQuerySet, RepoQuerySet, RepositoryContentUnitQuerySet
+from pulp.server.db.querysets import (CriteriaQuerySet, RepoQuerySet, RepositoryContentUnitQuerySet,
+                                      WorkerQuerySet)
 from pulp.server.managers import factory
 from pulp.server.util import Singleton
 from pulp.server.webservices.views import serializers
@@ -411,7 +412,7 @@ class Worker(AutoRetryDocument):
     meta = {'collection': 'workers',
             'indexes': [],  # this is a small collection that does not need an index
             'allow_inheritance': False,
-            'queryset_class': CriteriaQuerySet}
+            'queryset_class': WorkerQuerySet}
 
     @property
     def queue_name(self):

server/pulp/server/db/querysets.py

@@ -1,10 +1,13 @@
+from datetime import datetime, timedelta
 from gettext import gettext as _
 import operator
 
 from mongoengine import Q
 from mongoengine.queryset import DoesNotExist, QuerySetNoCache
 from pymongo import ASCENDING
 
+from pulp.common import constants
+from pulp.common.dateutils import ensure_tz
 from pulp.server import exceptions as pulp_exceptions
 
 
@@ -116,6 +119,27 @@ def get_or_404(self, **kwargs):
             raise pulp_exceptions.MissingResource(**kwargs)
 
 
+class WorkerQuerySet(CriteriaQuerySet):
+    """
+    Custom queryset for workers
+    """
+
+    def get_online(self):
+        """
+        Returns a queryset with a subset of Worker documents.
+
+        The queryset is filtered to remove any Worker document that has not been updated in the
+        last 25 seconds.
+
+        :return: mongoengine queryset object
+        :rtype:  mongoengine.queryset.QuerySet
+        """
+        query_set = self
+        now = ensure_tz(datetime.utcnow())
+        oldest_heartbeat_time = now - timedelta(seconds=constants.PULP_PROCESS_TIMEOUT_INTERVAL)
+        return query_set.filter(last_heartbeat__gte=oldest_heartbeat_time)
+
+
 class RepoQuerySet(CriteriaQuerySet):
     """
     Custom queryset for repositories.

server/pulp/server/managers/status.py

@@ -25,7 +25,7 @@ def get_workers():
     :returns:          list of workers with their heartbeats
     :rtype:            list
     """
-    return Worker.objects()
+    return Worker.objects.get_online()
 
 
 def get_mongo_conn_status():

server/test/unit/server/async/test_tasks.py

@@ -1000,7 +1000,8 @@ def test_reserved_resources_queried_correctly(self, mock_reserved_resource):
     @mock.patch('pulp.server.async.tasks.ReservedResource')
     def test_worker_returned_when_one_worker_is_not_reserved(self, mock_reserved_resource,
                                                              mock_worker_objects):
-        mock_worker_objects.return_value = [{'name': 'a'}, {'name': 'b'}]
+        get_online = mock_worker_objects.get_online
+        get_online.return_value = [{'name': 'a'}, {'name': 'b'}]
         mock_reserved_resource.objects.all.return_value = [{'worker_name': 'a'}]
         result = tasks._get_unreserved_worker()
         self.assertEqual(result, {'name': 'b'})

server/test/unit/server/db/test_model.py

@@ -22,7 +22,7 @@
 from pulp.server.exceptions import PulpCodedException
 from pulp.server.db import model
 from pulp.server.db.fields import ISO8601StringField
-from pulp.server.db.querysets import CriteriaQuerySet
+from pulp.server.db.querysets import CriteriaQuerySet, WorkerQuerySet
 from pulp.server.webservices.views import serializers
 
 
@@ -707,7 +707,7 @@ def test_meta_inheritance(self):
         self.assertEqual(model.Worker._meta['allow_inheritance'], False)
 
     def test_meta_queryset(self):
-        self.assertEqual(model.Worker._meta['queryset_class'], CriteriaQuerySet)
+        self.assertEqual(model.Worker._meta['queryset_class'], WorkerQuerySet)
         self.assertTrue(issubclass(model.Worker.objects.__class__, QuerySetNoCache))
 
 

server/test/unit/server/managers/test_status.py

@@ -16,8 +16,9 @@ def test_get_version(self, mock_get_distribution):
 
     @patch('pulp.server.db.model.Worker.objects')
     def test_get_workers(self, mock_worker_objects):
-        mock_worker_objects.return_value = [{"last_heartbeat": "123456", "name": "some_worker_1"},
-                                            {"last_heartbeat": "123456", "name": "some_worker_2"}]
+        get_online = mock_worker_objects.get_online
+        get_online.return_value = [{"last_heartbeat": "123456", "name": "some_worker_1"},
+                                   {"last_heartbeat": "123456", "name": "some_worker_2"}]
 
         self.assertEquals(status_manager.get_workers(), [{"last_heartbeat": "123456",
                                                           "name": "some_worker_1"},