Update the clustering and job processor app to use the ClusterScope setting and not their own settings singleton
This commit is contained in:
Roberto Rosario
@@ -1,88 +1 @@
|
||||
from __future__ import absolute_import
|
||||
|
||||
import atexit
|
||||
import logging
|
||||
import psutil
|
||||
from multiprocessing import active_children
|
||||
|
||||
from django.db import transaction, DatabaseError
|
||||
from django.dispatch import receiver
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
from scheduler.api import LocalScheduler
|
||||
from navigation.api import bind_links, register_model_list_columns
|
||||
from project_setup.api import register_setup
|
||||
from common.utils import encapsulate
|
||||
|
||||
from clustering.models import Node
|
||||
from clustering.signals import node_died, node_heartbeat
|
||||
|
||||
from .models import JobQueue, JobProcessingConfig, JobQueueItem, Worker
|
||||
from .tasks import job_queue_poll, house_keeping
|
||||
from .links import (node_workers, job_queues, tool_link,
|
||||
job_queue_items_pending, job_queue_items_error, job_queue_items_active,
|
||||
job_queue_config_edit, setup_link, job_queue_start, job_queue_stop,
|
||||
job_requeue, job_delete, worker_terminate)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@transaction.commit_on_success
|
||||
def add_job_queue_jobs():
|
||||
job_processor_scheduler = LocalScheduler('job_processor', _(u'Job processor'))
|
||||
try:
|
||||
job_processor_scheduler.add_interval_job('job_queue_poll', _(u'Poll a job queue for pending jobs.'), job_queue_poll, seconds=JobProcessingConfig.get().job_queue_poll_interval)
|
||||
job_processor_scheduler.add_interval_job('house_keeping', _(u'Poll a job queue for pending jobs.'), house_keeping, seconds=JobProcessingConfig.get().dead_job_removal_interval)
|
||||
except DatabaseError:
|
||||
transaction.rollback()
|
||||
|
||||
job_processor_scheduler.start()
|
||||
|
||||
|
||||
add_job_queue_jobs()
|
||||
register_setup(setup_link)
|
||||
bind_links([JobQueue, 'job_queues'], [job_queues], menu_name='secondary_menu')
|
||||
bind_links([JobQueue], [job_queue_start, job_queue_stop, job_queue_items_pending, job_queue_items_active, job_queue_items_error])
|
||||
bind_links([Node], [node_workers])
|
||||
bind_links(['job_queue_config_edit'], [job_queue_config_edit], menu_name='secondary_menu')
|
||||
bind_links([JobQueueItem], [job_requeue, job_delete])
|
||||
bind_links([Worker], [worker_terminate])
|
||||
|
||||
Node.add_to_class('workers', lambda node: node.worker_set)
|
||||
|
||||
register_model_list_columns(Node, [
|
||||
{
|
||||
'name': _(u'active workers'),
|
||||
'attribute': encapsulate(lambda x: x.workers().all().count())
|
||||
},
|
||||
])
|
||||
|
||||
|
||||
@receiver(node_died, dispatch_uid='process_dead_workers')
|
||||
def process_dead_workers(sender, node, **kwargs):
|
||||
logger.debug('received signal')
|
||||
for dead_worker in node.worker_set.all():
|
||||
if dead_worker.job_queue_item:
|
||||
dead_worker.job_queue_item.requeue(force=True, at_top=True)
|
||||
dead_worker.delete()
|
||||
|
||||
|
||||
@receiver(node_heartbeat, dispatch_uid='node_processes')
|
||||
def node_processes(sender, node, **kwargs):
|
||||
logger.debug('update current node\'s processes')
|
||||
all_active_pids = psutil.get_pid_list()
|
||||
# Remove stale workers based on current child pids
|
||||
for dead_worker in node.worker_set.exclude(pid__in=all_active_pids):
|
||||
if dead_worker.job_queue_item:
|
||||
dead_worker.job_queue_item.requeue(force=True, at_top=True)
|
||||
dead_worker.delete()
|
||||
|
||||
|
||||
def kill_all_node_processes():
|
||||
logger.debug('terminating this node\'s all processes')
|
||||
for process in active_children():
|
||||
process.terminate()
|
||||
process.join()
|
||||
|
||||
|
||||
atexit.register(kill_all_node_processes)
|
||||
|
||||
@@ -45,9 +45,6 @@ job_queue_items_active = Link(text=_(u'active jobs'), view='job_queue_items_acti
|
||||
job_queue_start = Link(text=_(u'start'), view='job_queue_start', args='object.pk', icon=icon_job_queue_start, permissions=[PERMISSION_JOB_QUEUE_START_STOP], condition=is_not_running)
|
||||
job_queue_stop = Link(text=_(u'stop'), view='job_queue_stop', args='object.pk', icon=icon_job_queue_stop, permissions=[PERMISSION_JOB_QUEUE_START_STOP], condition=is_running)
|
||||
|
||||
job_queue_config_edit = Link(text=_(u'edit job processing configuration'), view='job_queue_config_edit', permissions=[PERMISSION_JOB_PROCESSING_CONFIGURATION])
|
||||
setup_link = Link(text=_(u'job processing configuration'), view='job_queue_config_edit', permissions=[PERMISSION_JOB_PROCESSING_CONFIGURATION])
|
||||
|
||||
job_requeue = Link(text=_(u'requeue job'), view='job_requeue', args='object.pk', icon=icon_job_requeue, permissions=[PERMISSION_JOB_REQUEUE], condition=is_in_error_state)
|
||||
job_delete = Link(text=_(u'delete job'), view='job_delete', args='object.pk', icon=icon_job_delete, permissions=[PERMISSION_JOB_DELETE], condition=is_in_pending_state)
|
||||
|
||||
|
||||
@@ -0,0 +1,67 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import datetime
|
||||
from south.db import db
|
||||
from south.v2 import SchemaMigration
|
||||
from django.db import models
|
||||
|
||||
|
||||
class Migration(SchemaMigration):
|
||||
|
||||
def forwards(self, orm):
|
||||
# Deleting model 'JobProcessingConfig'
|
||||
db.delete_table('job_processor_jobprocessingconfig')
|
||||
|
||||
|
||||
def backwards(self, orm):
|
||||
# Adding model 'JobProcessingConfig'
|
||||
db.create_table('job_processor_jobprocessingconfig', (
|
||||
('lock_id', self.gf('django.db.models.fields.CharField')(default=1, max_length=1, unique=True)),
|
||||
('job_queue_poll_interval', self.gf('django.db.models.fields.PositiveIntegerField')(default=2)),
|
||||
('dead_job_removal_interval', self.gf('django.db.models.fields.PositiveIntegerField')(default=5)),
|
||||
('id', self.gf('django.db.models.fields.AutoField')(primary_key=True)),
|
||||
))
|
||||
db.send_create_signal('job_processor', ['JobProcessingConfig'])
|
||||
|
||||
|
||||
models = {
|
||||
'clustering.node': {
|
||||
'Meta': {'object_name': 'Node'},
|
||||
'cpuload': ('django.db.models.fields.FloatField', [], {'default': '0', 'blank': 'True'}),
|
||||
'heartbeat': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime(2012, 9, 15, 0, 0)', 'blank': 'True'}),
|
||||
'hostname': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '255'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'memory_usage': ('django.db.models.fields.FloatField', [], {'default': '0', 'blank': 'True'}),
|
||||
'state': ('django.db.models.fields.CharField', [], {'default': "'d'", 'max_length': '4'})
|
||||
},
|
||||
'job_processor.jobqueue': {
|
||||
'Meta': {'ordering': "('priority',)", 'object_name': 'JobQueue'},
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'name': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '32'}),
|
||||
'priority': ('django.db.models.fields.IntegerField', [], {'default': '0'}),
|
||||
'state': ('django.db.models.fields.CharField', [], {'default': "'r'", 'max_length': '4'}),
|
||||
'unique_jobs': ('django.db.models.fields.BooleanField', [], {'default': 'True'})
|
||||
},
|
||||
'job_processor.jobqueueitem': {
|
||||
'Meta': {'ordering': "('creation_datetime',)", 'object_name': 'JobQueueItem'},
|
||||
'creation_datetime': ('django.db.models.fields.DateTimeField', [], {}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'job_queue': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['job_processor.JobQueue']"}),
|
||||
'job_type': ('django.db.models.fields.CharField', [], {'max_length': '32'}),
|
||||
'kwargs': ('django.db.models.fields.TextField', [], {}),
|
||||
'result': ('django.db.models.fields.TextField', [], {'blank': 'True'}),
|
||||
'state': ('django.db.models.fields.CharField', [], {'default': "'p'", 'max_length': '4'}),
|
||||
'unique_id': ('django.db.models.fields.CharField', [], {'unique': 'True', 'max_length': '64', 'blank': 'True'})
|
||||
},
|
||||
'job_processor.worker': {
|
||||
'Meta': {'ordering': "('creation_datetime',)", 'unique_together': "(('node', 'pid'),)", 'object_name': 'Worker'},
|
||||
'creation_datetime': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime(2012, 9, 15, 0, 0)'}),
|
||||
'heartbeat': ('django.db.models.fields.DateTimeField', [], {'default': 'datetime.datetime(2012, 9, 15, 0, 0)', 'blank': 'True'}),
|
||||
'id': ('django.db.models.fields.AutoField', [], {'primary_key': 'True'}),
|
||||
'job_queue_item': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['job_processor.JobQueueItem']", 'null': 'True', 'blank': 'True'}),
|
||||
'node': ('django.db.models.fields.related.ForeignKey', [], {'to': "orm['clustering.Node']"}),
|
||||
'pid': ('django.db.models.fields.PositiveIntegerField', [], {'max_length': '255'}),
|
||||
'state': ('django.db.models.fields.CharField', [], {'default': "'r'", 'max_length': '4'})
|
||||
}
|
||||
}
|
||||
|
||||
complete_apps = ['job_processor']
|
||||
@@ -16,7 +16,6 @@ from django.utils.translation import ugettext_lazy as _
|
||||
from django.utils.translation import ugettext
|
||||
from django.utils.simplejson import loads, dumps
|
||||
|
||||
from common.models import Singleton
|
||||
from clustering.models import Node
|
||||
|
||||
from .literals import (JOB_STATE_CHOICES, JOB_STATE_PENDING,
|
||||
@@ -293,14 +292,3 @@ class Worker(models.Model):
|
||||
verbose_name = _(u'worker')
|
||||
verbose_name_plural = _(u'workers')
|
||||
unique_together = ('node', 'pid')
|
||||
|
||||
|
||||
class JobProcessingConfig(Singleton):
|
||||
job_queue_poll_interval = models.PositiveIntegerField(verbose_name=(u'job queue poll interval (in seconds)'), default=DEFAULT_JOB_QUEUE_POLL_INTERVAL)
|
||||
dead_job_removal_interval = models.PositiveIntegerField(verbose_name=(u'dead job check and removal interval (in seconds)'), help_text=_(u'Interval of time to check the cluster for and remove unresponsive jobs.'), default=DEFAULT_DEAD_JOB_REMOVAL_INTERVAL)
|
||||
|
||||
def __unicode__(self):
|
||||
return ugettext(u'Job queues configuration')
|
||||
|
||||
class Meta:
|
||||
verbose_name = verbose_name_plural = _(u'job queues configuration')
|
||||
|
||||
86
apps/job_processor/post_init.py
Normal file
86
apps/job_processor/post_init.py
Normal file
@@ -0,0 +1,86 @@
|
||||
from __future__ import absolute_import
|
||||
|
||||
import atexit
|
||||
import logging
|
||||
import psutil
|
||||
from multiprocessing import active_children
|
||||
|
||||
from django.db import transaction, DatabaseError
|
||||
from django.dispatch import receiver
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
from scheduler.api import LocalScheduler
|
||||
from navigation.api import bind_links, register_model_list_columns
|
||||
from common.utils import encapsulate
|
||||
from clustering.models import Node
|
||||
from clustering.signals import node_died, node_heartbeat
|
||||
|
||||
from .models import JobQueue, JobQueueItem, Worker
|
||||
from .tasks import job_queue_poll, house_keeping
|
||||
from .links import (node_workers, job_queues, tool_link,
|
||||
job_queue_items_pending, job_queue_items_error, job_queue_items_active,
|
||||
job_queue_start, job_queue_stop, job_requeue, job_delete, worker_terminate)
|
||||
from .settings import QUEUE_POLL_INTERVAL, DEAD_JOB_REMOVAL_INTERVAL
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@transaction.commit_on_success
|
||||
def add_job_queue_jobs():
|
||||
job_processor_scheduler = LocalScheduler('job_processor', _(u'Job processor'))
|
||||
try:
|
||||
# TODO: auto convert settings to int
|
||||
job_processor_scheduler.add_interval_job('job_queue_poll', _(u'Poll a job queue for pending jobs.'), job_queue_poll, seconds=int(QUEUE_POLL_INTERVAL))
|
||||
job_processor_scheduler.add_interval_job('house_keeping', _(u'Poll a job queue for pending jobs.'), house_keeping, seconds=int(DEAD_JOB_REMOVAL_INTERVAL))
|
||||
except DatabaseError:
|
||||
transaction.rollback()
|
||||
|
||||
job_processor_scheduler.start()
|
||||
|
||||
|
||||
@receiver(node_died, dispatch_uid='process_dead_workers')
|
||||
def process_dead_workers(sender, node, **kwargs):
|
||||
logger.debug('received signal')
|
||||
for dead_worker in node.worker_set.all():
|
||||
if dead_worker.job_queue_item:
|
||||
dead_worker.job_queue_item.requeue(force=True, at_top=True)
|
||||
dead_worker.delete()
|
||||
|
||||
|
||||
@receiver(node_heartbeat, dispatch_uid='node_processes')
|
||||
def node_processes(sender, node, **kwargs):
|
||||
logger.debug('update current node\'s processes')
|
||||
all_active_pids = psutil.get_pid_list()
|
||||
# Remove stale workers based on current child pids
|
||||
for dead_worker in node.worker_set.exclude(pid__in=all_active_pids):
|
||||
if dead_worker.job_queue_item:
|
||||
dead_worker.job_queue_item.requeue(force=True, at_top=True)
|
||||
dead_worker.delete()
|
||||
|
||||
|
||||
def kill_all_node_processes():
|
||||
logger.debug('terminating this node\'s all processes')
|
||||
for process in active_children():
|
||||
process.terminate()
|
||||
process.join()
|
||||
|
||||
|
||||
def init_job_processor():
|
||||
atexit.register(kill_all_node_processes)
|
||||
|
||||
add_job_queue_jobs()
|
||||
bind_links([JobQueue, 'job_queues'], [job_queues], menu_name='secondary_menu')
|
||||
bind_links([JobQueue], [job_queue_start, job_queue_stop, job_queue_items_pending, job_queue_items_active, job_queue_items_error])
|
||||
bind_links([Node], [node_workers])
|
||||
bind_links([JobQueueItem], [job_requeue, job_delete])
|
||||
bind_links([Worker], [worker_terminate])
|
||||
|
||||
Node.add_to_class('workers', lambda node: node.worker_set)
|
||||
|
||||
register_model_list_columns(Node, [
|
||||
{
|
||||
'name': _(u'active workers'),
|
||||
'attribute': encapsulate(lambda x: x.workers().all().count())
|
||||
},
|
||||
])
|
||||
|
||||
@@ -9,8 +9,6 @@ urlpatterns = patterns('job_processor.views',
|
||||
url(r'^queue/(?P<job_queue_pk>\d+)/items/active/$', 'job_queue_items', {'active_jobs' :True}, 'job_queue_items_active'),
|
||||
url(r'^queue/(?P<job_queue_pk>\d+)/start/$', 'job_queue_start', (), 'job_queue_start'),
|
||||
url(r'^queue/(?P<job_queue_pk>\d+)/stop/$', 'job_queue_stop', (), 'job_queue_stop'),
|
||||
|
||||
url(r'^config/edit/$', 'job_queue_config_edit', (), 'job_queue_config_edit'),
|
||||
|
||||
url(r'^job/(?P<job_item_pk>\d+)/requeue/$', 'job_requeue', (), 'job_requeue'),
|
||||
url(r'^job/(?P<job_item_pk>\d+)/delete/$', 'job_delete', (), 'job_delete'),
|
||||
|
||||
@@ -14,8 +14,7 @@ from clustering.models import Node
|
||||
from permissions.models import Permission
|
||||
|
||||
from .exceptions import JobQueueAlreadyStopped, JobQueueAlreadyStarted
|
||||
from .forms import JobProcessingConfigForm
|
||||
from .models import JobQueue, JobProcessingConfig, JobQueueItem, Worker
|
||||
from .models import JobQueue, JobQueueItem, Worker
|
||||
from .permissions import (PERMISSION_JOB_QUEUE_VIEW,
|
||||
PERMISSION_JOB_PROCESSING_CONFIGURATION, PERMISSION_JOB_QUEUE_START_STOP,
|
||||
PERMISSION_JOB_REQUEUE)
|
||||
@@ -163,38 +162,7 @@ def job_queue_items(request, job_queue_pk, pending_jobs=False, error_jobs=False,
|
||||
return render_to_response('generic_list.html', context,
|
||||
context_instance=RequestContext(request))
|
||||
|
||||
|
||||
def job_queue_config_edit(request):
|
||||
Permission.objects.check_permissions(request.user, [PERMISSION_JOB_PROCESSING_CONFIGURATION])
|
||||
|
||||
job_processing_config = JobProcessingConfig.get()
|
||||
|
||||
post_action_redirect = None
|
||||
|
||||
previous = request.POST.get('previous', request.GET.get('previous', request.META.get('HTTP_REFERER', '/')))
|
||||
next = request.POST.get('next', request.GET.get('next', post_action_redirect if post_action_redirect else request.META.get('HTTP_REFERER', '/')))
|
||||
|
||||
|
||||
if request.method == 'POST':
|
||||
form = JobProcessingConfigForm(data=request.POST)
|
||||
if form.is_valid():
|
||||
try:
|
||||
form.save()
|
||||
except Exception, exc:
|
||||
messages.error(request, _(u'Error trying to edit job processing configuration; %s') % exc)
|
||||
else:
|
||||
messages.success(request, _(u'Job processing configuration edited successfully.'))
|
||||
return HttpResponseRedirect(next)
|
||||
else:
|
||||
form = JobProcessingConfigForm(instance=job_processing_config)
|
||||
|
||||
return render_to_response('generic_form.html', {
|
||||
'form': form,
|
||||
'object': job_processing_config,
|
||||
'title': _(u'Edit job processing configuration')
|
||||
}, context_instance=RequestContext(request))
|
||||
|
||||
|
||||
|
||||
def job_queue_stop(request, job_queue_pk):
|
||||
job_queue = get_object_or_404(JobQueue, pk=job_queue_pk)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user