#!/usr/bin/env python
# ResourceAssigner.py: ResourceAssigner listens on the lofar ?? bus and calls onTaskSpecified
#
# Copyright (C) 2015
# ASTRON (Netherlands Institute for Radio Astronomy)
# P.O.Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.
#
# $Id: raservice.py 1580 2015-09-30 14:18:57Z loose $
"""
TaskSpecifiedListener listens to a bus on which specified tasks get published. It will then try
to assign resources to these tasks.
"""
import qpid.messaging
import logging
from datetime import datetime
import time
from lofar.sas.resourceassignment.rataskspecified.RABusListener import RATaskSpecifiedBusListener
from lofar.messaging.RPC import RPC, RPCException
import lofar.sas.resourceassignment.resourceassignmentservice.rpc as rarpc
from lofar.sas.resourceassignment.rataskspecified.config import DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_BUSNAME
from lofar.sas.resourceassignment.rataskspecified.config import DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_SUBJECT
from lofar.sas.resourceassignment.resourceassigner.assignment import ResourceAssigner
logger = logging.getLogger(__name__)
class SpecifiedTaskListener(RATaskSpecifiedBusListener):
def __init__(self,
busname=DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_BUSNAME,
subject=DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_SUBJECT,
broker=None,
assigner=None,
**kwargs):
"""
SpecifiedTaskListener listens on the lofar ?? bus and calls onTaskSpecified
:param busname: valid Qpid address (default: lofar.otdb.status)
:param broker: valid Qpid broker host (default: None, which means localhost)
additional parameters in kwargs:
options= <dict> Dictionary of options passed to QPID
exclusive= <bool> Create an exclusive binding so no other services can consume duplicate messages (default: False)
numthreads= <int> Number of parallel threads processing messages (default: 1)
verbose= <bool> Output extra logging over stdout (default: False)
"""
super(SpecifiedTaskListener, self).__init__(busname=busname, subject=subject, broker=broker, **kwargs)
self.assigner = assigner
if not self.assigner:
self.assigner = ResourceAssigner()
def onTaskSpecified(self, otdb_id, specification_tree):
logger.info('onTaskSpecified: otdb_id=%s' % otdb_id)
self.assigner.doAssignment(specification_tree)
__all__ = ["SpecifiedTaskListener"]
def main():
from optparse import OptionParser
from lofar.messaging import setQpidLogLevel
from lofar.common.util import waitForInterrupt
from lofar.sas.resourceassignment.resourceassignmentservice.config import DEFAULT_BUSNAME as RADB_BUSNAME
from lofar.sas.resourceassignment.resourceassignmentservice.config import DEFAULT_SERVICENAME as RADB_SERVICENAME
from lofar.sas.resourceassignment.resourceassignmentestimator.config import DEFAULT_BUSNAME as RE_BUSNAME
from lofar.sas.resourceassignment.resourceassignmentestimator.config import DEFAULT_SERVICENAME as RE_SERVICENAME
from lofar.sas.otdb.config import DEFAULT_OTDB_SERVICE_BUSNAME, DEFAULT_OTDB_SERVICENAME
from lofar.sas.systemstatus.service.config import DEFAULT_SSDB_BUSNAME
from lofar.sas.systemstatus.service.config import DEFAULT_SSDB_SERVICENAME
from lofar.sas.resourceassignment.resourceassigner.config import DEFAULT_RA_NOTIFICATION_BUSNAME
from lofar.sas.resourceassignment.resourceassigner.config import DEFAULT_RA_NOTIFICATION_PREFIX
# Check the invocation arguments
parser = OptionParser("%prog [options]",
description='runs the resourceassigner service')
parser.add_option('-q', '--broker', dest='broker', type='string',
default=None,
help='Address of the qpid broker, default: localhost')
parser.add_option("--notification_busname", dest="notification_busname", type="string",
default=DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_BUSNAME,
help="Name of the notification bus on which taskspecified messages are published. [default: %default]")
parser.add_option("--notification_subject", dest="notification_subject", type="string",
default=DEFAULT_RA_TASK_SPECIFIED_NOTIFICATION_SUBJECT,
help="Subject of the published taskspecified messages to listen for. [default: %default]")
parser.add_option("--radb_busname", dest="radb_busname", type="string",
default=RADB_BUSNAME,
help="Name of the bus on which the radb service listens. [default: %default]")
parser.add_option("--radb_servicename", dest="radb_servicename", type="string",
default=RADB_SERVICENAME,
help="Name of the radb service. [default: %default]")
parser.add_option("--re_busname", dest="re_busname", type="string",
default=RE_BUSNAME,
help="Name of the bus on which the resource estimator service listens. [default: %default]")
parser.add_option("--re_servicename", dest="re_servicename", type="string",
default=RE_SERVICENAME,
help="Name of the resource estimator service. [default: %default]")
parser.add_option("--otdb_busname", dest="otdb_busname", type="string", default=DEFAULT_OTDB_SERVICE_BUSNAME, help="Name of the bus on which the OTDB service listens, default: %default")
parser.add_option("--otdb_servicename", dest="otdb_servicename", type="string", default=DEFAULT_OTDB_SERVICENAME, help="Name of the OTDB service, default: %default")
parser.add_option("--ssdb_busname", dest="ssdb_busname", type="string",
default=DEFAULT_SSDB_BUSNAME,
help="Name of the bus on which the ssdb service listens. [default: %default]")
parser.add_option("--ssdb_servicename", dest="ssdb_servicename", type="string",
default=DEFAULT_SSDB_SERVICENAME,
help="Name of the ssdb service. [default: %default]")
parser.add_option("--ra_notification_busname", dest="ra_notification_busname", type="string",
default=DEFAULT_RA_NOTIFICATION_BUSNAME,
help="Name of the notification bus on which the resourceassigner publishes its notifications. [default: %default]")
parser.add_option("--ra_notification_prefix", dest="ra_notification_prefix", type="string",
default=DEFAULT_RA_NOTIFICATION_PREFIX,
help="Prefix for the subject of the by the resourceassigner published notification messages. [default: %default]")
parser.add_option('-V', '--verbose', dest='verbose', action='store_true', help='verbose logging')
(options, args) = parser.parse_args()
setQpidLogLevel(logging.INFO)
logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
level=logging.DEBUG if options.verbose else logging.INFO)
with ResourceAssigner(radb_busname=options.radb_busname,
radb_servicename=options.radb_servicename,
re_busname=options.re_busname,
re_servicename=options.re_servicename,
otdb_busname=options.otdb_busname,
otdb_servicename=options.otdb_servicename,
ssdb_busname=options.ssdb_busname,
ssdb_servicename=options.ssdb_servicename,
ra_notification_busname=options.ra_notification_busname,
ra_notification_prefix=options.ra_notification_prefix,
broker=options.broker) as assigner:
with SpecifiedTaskListener(busname=options.notification_busname,
subject=options.notification_subject,
broker=options.broker,
assigner=assigner) as listener:
waitForInterrupt()
if __name__ == '__main__':
main()