diff --git a/.gitattributes b/.gitattributes
index 99beefd7e47c8e8b93d16f10977724b561e75a84..30a9f9311c220aa948db39056a14cfa283f41a71 100644
--- a/.gitattributes
+++ b/.gitattributes
@@ -2623,6 +2623,9 @@ LCS/Messaging/python/messaging/test/t_messagebus.sh -text
LCS/Messaging/python/messaging/test/t_messages.py -text
LCS/Messaging/python/messaging/test/t_messages.run -text
LCS/Messaging/python/messaging/test/t_messages.sh -text
+LCS/Messaging/python/messaging/test/t_service_message_handler.py -text
+LCS/Messaging/python/messaging/test/t_service_message_handler.run -text svneol=unset#application/x-shellscript
+LCS/Messaging/python/messaging/test/t_service_message_handler.sh -text svneol=unset#application/x-shellscript
LCS/Messaging/src/CMakeLists.txt -text
LCS/Messaging/src/DefaultSettings.cc -text
LCS/Messaging/src/EventMessage.cc -text
@@ -4822,6 +4825,7 @@ SAS/OTDB/bin/momIDs -text
SAS/OTDB/bin/repairTree.py -text
SAS/OTDB/bin/revertDefaultTemplates.py -text
SAS/OTDB/include/OTDB/DefaultTemplate.h -text
+SAS/OTDB/sql/README! -text
SAS/OTDB/sql/assignProcessType_func.sql -text
SAS/OTDB/sql/campaignAPI.sql -text
SAS/OTDB/sql/create_rules.sql -text
@@ -4835,6 +4839,7 @@ SAS/OTDB/sql/getDefaultTemplates_func.sql -text
SAS/OTDB/sql/getModifiedTrees_func.sql -text
SAS/OTDB/sql/getMomID2treeID_func.sql -text
SAS/OTDB/sql/getSchedulerInfo_func.sql -text
+SAS/OTDB/sql/getStateChanges_func.sql -text
SAS/OTDB/sql/getTreeGroup_func.sql -text
SAS/OTDB/sql/getTreesInPeriod_func.sql -text
SAS/OTDB/sql/getVTitem_func.sql -text
@@ -4847,6 +4852,21 @@ SAS/OTDB/src/setStatus.conf -text
SAS/OTDB/test/tBrokenHardware.cc -text
SAS/OTDB/test/tMetadata.cc -text
SAS/OTDB/test/tQueryPIC.cc -text
+SAS/OTDB/test/t_getTreeGroup.py -text
+SAS/OTDB/test/t_getTreeGroup.run -text svneol=unset#application/x-shellscript
+SAS/OTDB/test/t_getTreeGroup.sh -text svneol=unset#application/x-shellscript
+SAS/OTDB/test/unittest_db.dump.gz -text svneol=unset#application/x-gzip
+SAS/OTDB_Services/CMakeLists.txt -text
+SAS/OTDB_Services/TreeService.py -text
+SAS/OTDB_Services/TreeStatusEvents.py -text
+SAS/OTDB_Services/test/CMakeLists.txt -text
+SAS/OTDB_Services/test/t_TreeService.py -text
+SAS/OTDB_Services/test/t_TreeService.run -text svneol=unset#application/x-shellscript
+SAS/OTDB_Services/test/t_TreeService.sh -text svneol=unset#application/x-shellscript
+SAS/OTDB_Services/test/t_TreeStatusEvents.py -text
+SAS/OTDB_Services/test/t_TreeStatusEvents.run -text svneol=unset#application/x-shellscript
+SAS/OTDB_Services/test/t_TreeStatusEvents.sh -text svneol=unset#application/x-shellscript
+SAS/OTDB_Services/test/unittest_db.dump.gz -text svneol=unset#application/x-gzip
SAS/ResourceAssignment/CMakeLists.txt -text
SAS/ResourceAssignment/ResourceAssignmentEditor/CMakeLists.txt -text
SAS/ResourceAssignment/ResourceAssignmentEditor/bin/CMakeLists.txt -text
diff --git a/CMake/LofarPackageList.cmake b/CMake/LofarPackageList.cmake
index 48daf66e7307fec9bfd284dedba80bc651d0c5fd..9379cd95d424db26e7d30b675184e38f37ecddfe 100644
--- a/CMake/LofarPackageList.cmake
+++ b/CMake/LofarPackageList.cmake
@@ -135,6 +135,7 @@ if(NOT DEFINED LOFAR_PACKAGE_LIST_INCLUDED)
set(CobaltTest_SOURCE_DIR ${CMAKE_SOURCE_DIR}/RTCP/Cobalt/CobaltTest)
set(BrokenAntennaInfo_SOURCE_DIR ${CMAKE_SOURCE_DIR}/RTCP/Cobalt/BrokenAntennaInfo)
set(OTDB_SOURCE_DIR ${CMAKE_SOURCE_DIR}/SAS/OTDB)
+ set(OTDB_Services_SOURCE_DIR ${CMAKE_SOURCE_DIR}/SAS/OTDB_Services)
set(OTB_SOURCE_DIR ${CMAKE_SOURCE_DIR}/SAS/OTB)
set(OTDB_SQL_SOURCE_DIR ${CMAKE_SOURCE_DIR}/SAS/OTDB/sql)
set(Scheduler_SOURCE_DIR ${CMAKE_SOURCE_DIR}/SAS/Scheduler)
diff --git a/LCS/Messaging/python/messaging/RPC.py b/LCS/Messaging/python/messaging/RPC.py
index fcc7868aac43437b5a871b3ec0b4f8c6b664e660..2e0f199e7eabe16003dd49a29ee552afcf8f3115 100644
--- a/LCS/Messaging/python/messaging/RPC.py
+++ b/LCS/Messaging/python/messaging/RPC.py
@@ -21,9 +21,13 @@
# RPC invocation with possible timeout
from lofar.messaging.messagebus import ToBus, FromBus
-from lofar.messaging.messages import ServiceMessage, ReplyMessage
+from lofar.messaging.messages import RequestMessage, ReplyMessage, analyze_args, args_as_content
import uuid
+class RPCException(Exception):
+ "Exception occured in the RPC code itself, like time-out, invalid message received, etc."
+ pass
+
class RPC():
"""
This class provides an easy way to invoke a Remote Rrocedure Call to a
@@ -37,65 +41,64 @@ class RPC():
As a side-effect the sender and session are destroyed.
"""
- def __init__(self, bus, service, timeout=None, ForwardExceptions=None, Verbose=None):
- """
- Initialize an Remote procedure call using:
- bus= <str> Bus Name
- service= <str> Service Name
+ def __init__(self, service, **kwargs ):
+ """
+ Initialize an Remote procedure call using:
+ service= <str> Service Name
+ busname= <str> Bus Name
timeout= <float> Time to wait in seconds before the call is considered a failure.
Verbose= <bool> If True output extra logging to stdout.
Use with extra care: ForwardExceptions= <bool>
- This enables forwarding exceptions from the server side tobe raised at the client side durting RPC invocation.
- """
- self.timeout = timeout
- self.ForwardExceptions = False
- self.Verbose = False
- if ForwardExceptions is True:
- self.ForwardExceptions = True
- if Verbose is True:
- self.Verbose = True
- self.BusName = bus
+ This enables forwarding exceptions from the server side tobe raised at the client side durting RPC invocation.
+ """
+ self.timeout = kwargs.pop("timeout", None)
+ self.ForwardExceptions = kwargs.pop("ForwardExceptions", False)
+ self.Verbose = kwargs.pop("Verbose", False)
+ self.BusName = kwargs.pop("busname", None)
self.ServiceName = service
if self.BusName is None:
self.Request = ToBus(self.ServiceName)
else:
- self.Request = ToBus(self.BusName + "/" + self.ServiceName)
+ self.Request = ToBus("%s/%s" % (self.BusName, self.ServiceName))
+ if len(kwargs):
+ raise AttributeError("Unexpected argument passed to RPC class: %s" %( kwargs ))
def __enter__(self):
- """
- Internal use only. (handles scope 'with')
- """
+ """
+ Internal use only. (handles scope 'with')
+ """
self.Request.open()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
- """
- Internal use only. (handles scope 'with')
- """
+ """
+ Internal use only. (handles scope 'with')
+ """
self.Request.close()
- def __call__(self, msg, timeout=None):
- """
- Enable the use of the object to directly invoke the RPC.
+ def __call__(self, *args, **kwargs):
+ """
+ Enable the use of the object to directly invoke the RPC.
example:
- with RPC(bus,service) as myrpc:
+ with RPC(bus,service) as myrpc:
result=myrpc(request)
-
- """
- if timeout is None:
- timeout = self.timeout
+
+ """
+ timeout = kwargs.pop("timeout", self.timeout)
+ Content = args_as_content(*args, **kwargs)
+ HasArgs, HasKwArgs = analyze_args(args, kwargs)
# create unique reply address for this rpc call
- options={'create':'always','delete':'receiver'}
- ReplyAddress= "reply." + str(uuid.uuid4())
+ options = {'create':'always','delete':'receiver'}
+ ReplyAddress = "reply.%s" % (str(uuid.uuid4()))
if self.BusName is None:
- Reply = FromBus(ReplyAddress+" ; "+str(options))
+ Reply = FromBus("%s ; %s" %(ReplyAddress,str(options)))
else:
- Reply = FromBus(self.BusName + "/" + ReplyAddress)
+ Reply = FromBus("%s/%s" % (self.BusName, ReplyAddress))
with Reply:
- MyMsg = ServiceMessage(msg, ReplyAddress)
+ MyMsg = RequestMessage(Content, ReplyAddress , has_args=HasArgs, has_kwargs=HasKwArgs)
MyMsg.ttl = timeout
self.Request.send(MyMsg)
answer = Reply.receive(timeout)
@@ -106,15 +109,15 @@ class RPC():
status["state"] = "TIMEOUT"
status["errmsg"] = "RPC Timed out"
status["backtrace"] = ""
- return (None, status)
+ raise RPCException(status)
# Check for illegal message type
if isinstance(answer, ReplyMessage) is False:
# if we come here we had a Time-Out
status["state"] = "ERROR"
- status["errmsg"] = "Incorrect messagetype (" + str(type(answer)) + ") received."
+ status["errmsg"] = "Incorrect messagetype (%s) received." % (str(type(answer)))
status["backtrace"] = ""
- return (None, status)
+ raise RPCException(status)
# return content and status if status is 'OK'
if (answer.status == "OK"):
@@ -129,7 +132,7 @@ class RPC():
status["state"] = "ERROR"
status["errmsg"] = "Return state in message not found"
status["backtrace"] = ""
- return (Null, status)
+ raise RPCException(status)
# Does the client expect us to throw the exception?
if self.ForwardExceptions is True:
@@ -139,5 +142,7 @@ class RPC():
instance = excep_class_(answer.backtrace)
raise (instance)
else:
- raise (Exception(answer.errmsg))
- return (None,status)
+ raise RPCException(answer.errmsg)
+ return (None, status)
+
+__all__ = ["RPC", "RPCException"]
diff --git a/LCS/Messaging/python/messaging/Service.py b/LCS/Messaging/python/messaging/Service.py
index aacc512325f3b3b0c7dcd1d935e5b115cc53dcfe..1c759817b8d78fd4eb6923dec9e1d58a7833b97b 100644
--- a/LCS/Messaging/python/messaging/Service.py
+++ b/LCS/Messaging/python/messaging/Service.py
@@ -21,7 +21,7 @@
#
from lofar.messaging.messagebus import ToBus,FromBus
-from lofar.messaging.messages import ReplyMessage,ServiceMessage
+from lofar.messaging.messages import ReplyMessage,RequestMessage
import threading
import time
import uuid
@@ -31,44 +31,103 @@ import logging
logger = logging.getLogger(__name__)
+class MessageHandlerInterface(object):
+ """
+ Interface class for tuning the handling of a message by the Service class.
+ The class defines some (placeholders for) functions that the Service class calls
+ during the handling of messages. It can for instance be used to maintain a database connection.
+
+ The pseudocode of the Service class is:
+ Service(busname, function or from_MessageHandlerInterface_derived_class, ..., HandlerArguments={})
+
+ handler = <from_MessageHandlerInterface_derived_class>(HandlerArguments)
+ handler.prepare_loop()
+ while alive:
+ handler.prepare_receive()
+ msg = wait for messages()
+ handler.handle_message(msg)
+ handler.finalize_handling(handling_result)
+ handler.finalize_loop()
+ """
+ def __init__(self, **kwargs):
+ pass
+
+ def prepare_loop(self):
+ "Called before main processing loop is entered."
+ pass
+
+ def prepare_receive(self):
+ "Called in main processing loop just before a blocking wait for messages is done."
+ pass
+
+ def handle_message(self, msg):
+ "Function the should handle the received message and return a result."
+ raise NotImplementedError("OOPS! YOU ENDED UP IN THE MESSAGE HANDLER OF THE ABSTRACT BASE CLASS!")
+
+ def finalize_handling(self, successful):
+ "Called in the main loop after the result was send back to the requester."
+ "@successful@ reflects the state of the handling: true/false"
+ pass
+
+ def finalize_loop(self):
+ "Called after main processing loop is finished."
+ pass
+
+
# create service:
-class Service:
+class Service(object):
"""
Service class for registering python functions with a Service name on a message bus.
- create new service with Service( BusName, ServiceName, ServiceHandler )
- Additional options:
- options=<dict> for the QPID connection
- numthreads=<int> amount of threads processing messages
- startonwith=<bool> automatically start listening when in scope using 'with'
- verbose=<bool> show debug text
+ create new service with Service(busname, servicename, servicehandler)
+ busname <string> The name of the messagebus (queue or exchange) the service whould listen on.
+ servicename <string> The name that the user should use the invocate the servicehandler.
+ servicehandler <...> May be a function of an class that is derived from the MessageHandlerInterface.
+ The service uses this function or class for the handling of the messages.
+ Optional arguments:
+ options <dict> For the QPID connection
+ exclusive <bool> To create eclusive access to this messagebus. Default:True
+ numthreads <int> Amount of threads processing messages. Default:1
+ parsefullmessage <bool> Pass full message of only message content to the service handler. Default:False.
+ startonwith <bool> Automatically start listening when in scope using 'with'
+ verbose <bool> Show debug text. Default:False
+ handler_args <dict> Arguments that are passed to the constructor of the servicehandler is case the servicehandler
+ is a class in stead of a function.
"""
- def __init__(self, busname, servicename, servicehandler, options=None, exclusive=True, numthreads=1, parsefullmessage=False, startonwith=False, verbose=False):
+ def __init__(self, servicename, servicehandler, **kwargs):
"""
- Initialize Service object with busname (str) ,servicename (str) and servicehandler function.
+ Initialize Service object with servicename (str) and servicehandler function.
additional parameters:
+ busname= <string> Name of the bus in case exchanges are used in stead of queues
options= <dict> Dictionary of options passed to QPID
exclusive= <bool> Create an exclusive binding so no other services can consume duplicate messages (default: True)
- numthreads= <int> Number of parallel threads processing messages (default: 1)
+ numthreads= <int> Number of parallel threads processing messages (default: 1)
verbose= <bool> Output extra logging over stdout (default: False)
"""
- self.BusName = busname
- self.ServiceName = servicename
- self.ServiceHandler = servicehandler
- self.connected = False
- self.running = False
- self.exclusive = exclusive
- self.link_uuid = str(uuid.uuid4())
- self._numthreads = numthreads
- self.Verbose = verbose
- self.options = {"capacity": numthreads*20}
- self.parsefullmessage=parsefullmessage
- self.startonwith = startonwith
+ self.service_name = servicename
+ self.service_handler = servicehandler
+ self.connected = False
+ self.running = [False]
+ self.link_uuid = str(uuid.uuid4())
+ self.busname = kwargs.pop("busname", None)
+ self.exclusive = kwargs.pop("exclusive", True)
+ self._numthreads = kwargs.pop("numthreads", 1)
+ self.verbose = kwargs.pop("verbose", False)
+ self.options = {"capacity": self._numthreads*20}
+ options = kwargs.pop("options", None)
+ self.parsefullmessage = kwargs.pop("parsefullmessage", False)
+ self.startonwith = kwargs.pop("startonwith", None)
+ self.handler_args = kwargs.pop("handler_args", None)
+ self.listening = False
+ if len(kwargs):
+ raise AttributeError("Unexpected argument passed to Service class: %s", kwargs)
# Set appropriate flags for exclusive binding
- if self.exclusive is True:
- self.options["link"] = '{name:"' + self.link_uuid + '", x-bindings:[{key:' + self.ServiceName + ', arguments: {"qpid.exclusive-binding":True}}]}'
+ if self.exclusive == True:
+ self.options["link"] = '{name:"' + self.link_uuid + \
+ '", x-bindings:[{key:' + self.service_name + \
+ ', arguments: {"qpid.exclusive-binding":True}}]}'
# only add options if it is given as a dictionary
if isinstance(options,dict):
@@ -77,46 +136,84 @@ class Service:
def _debug(self, txt):
"""
- Internal use only.
- """
- if self.Verbose is True:
+ Internal use only.
+ """
+ if self.verbose == True:
logger.debug("[Service: %s]", txt)
- def StartListening(self, numthreads=None):
+ def start_listening(self, numthreads=None):
"""
- Start the background threads and process incoming messages.
- """
- if numthreads is not None:
+ Start the background threads and process incoming messages.
+ """
+ if self.listening == True:
+ return
+
+ # Usually a service will be listening on a 'bus' implemented by a topic exchange
+ if self.busname != None:
+ self.listener = FromBus(self.busname+"/"+self.service_name, options=self.options)
+ self.reply_bus = ToBus(self.busname)
+ self.listener.open()
+ self.reply_bus.open()
+ # Handle case when queues are used
+ else:
+ # assume that we are listening on a queue and therefore we cannot use a generic ToBus() for replies.
+ self.listener = FromBus(self.service_name, options=self.options)
+ self.listener.open()
+ self.reply_bus=None
+
+ self.connected = True
+
+ if numthreads != None:
self._numthreads = numthreads
- if self.connected is False:
- raise Exception("StartListening Called on closed connections")
- self.running = True
+ # use a list to ensure that threads always 'see' changes in the running state.
+ self.running = [ True ]
self._tr = []
self.reccounter = []
self.okcounter =[]
for i in range(self._numthreads):
- self._tr.append(threading.Thread(target=self._loop, args=[i]))
+ # set up service_handler
+ if str(type(self.service_handler)) == "<type 'instancemethod'>" or \
+ str(type(self.service_handler)) == "<type 'function'>":
+ thread_service_handler = MessageHandlerInterface()
+ thread_service_handler.handle_message = self.service_handler
+ else:
+ thread_service_handler = self.service_handler(**self.handler_args)
+ if not isinstance(thread_service_handler, MessageHandlerInterface):
+ raise TypeError("Servicehandler argument must by a function or a derived class from MessageHandlerInterface.")
+
+ self._tr.append(threading.Thread(target=self._loop,
+ kwargs={"index":i, "service_handler":thread_service_handler}))
self.reccounter.append(0)
self.okcounter.append(0)
self._tr[i].start()
+ self.listening = True
- def StopListening(self):
- """
- Stop the background threads that listen to incoming messages.
- """
+ def stop_listening(self):
+ """
+ Stop the background threads that listen to incoming messages.
+ """
# stop all running threads
- if self.running is True:
- self.running = False
+ if self.running[0] == True:
+ self.running[0] = False
for i in range(self._numthreads):
self._tr[i].join()
- logger.info("Thread %2d: STOPPED Listening for messages on Bus %s and service name %s." % (i, self.BusName, self.ServiceName))
+ logger.info("Thread %2d: STOPPED Listening for messages on Bus %s and service name %s." % (i, self.busname, self.service_name))
logger.info(" %d messages received and %d processed OK." % (self.reccounter[i], self.okcounter[i]))
+ self.listening = False
+ # close the listeners
+ if self.connected == True:
+ if isinstance(self.listener, FromBus):
+ self.listener.close()
+ if isinstance(self.reply_bus, ToBus):
+ self.reply_bus.close()
+ self.connected = False
- def WaitForInterrupt(self):
- """
- Useful (low cpu load) loop that waits for keyboard interrupt.
- """
+
+ def wait_for_interrupt(self):
+ """
+ Useful (low cpu load) loop that waits for keyboard interrupt.
+ """
looping = True
while looping:
try:
@@ -128,40 +225,16 @@ class Service:
def __enter__(self):
"""
- Internal use only. Handles scope with keyword 'with'
+ Internal use only. Handles scope with keyword 'with'
"""
- # Usually a service will be listening on a 'bus' implemented by a topic exchange
- if self.BusName is not None:
- self.Listen = FromBus(self.BusName+"/"+self.ServiceName, options=self.options)
- self.Reply = ToBus(self.BusName)
- self.Listen.open()
- self.Reply.open()
- # Handle case when queues are used
- else:
- # assume that we are listening on a queue and therefore we cannot use a generic ToBus() for replies.
- self.Listen = FromBus(self.ServiceName, options=self.options)
- self.Listen.open()
- self.Reply=None
-
- self.connected = True
-
- # If required start listening on 'with'
- if self.startonwith is True:
- self.StartListening()
+ self.start_listening()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""
Internal use only. Handles scope with keyword 'with'
"""
- self.StopListening()
- # close the listeners
- if self.connected is True:
- self.connected = False
- if isinstance(self.Listen, FromBus):
- self.Listen.close()
- if isinstance(self.Reply, ToBus):
- self.Reply.close()
+ self.stop_listening()
def _send_reply(self, replymessage, status, reply_to, errtxt="",backtrace=""):
"""
@@ -169,62 +242,123 @@ class Service:
"""
# Compose Reply message from reply and status.
if isinstance(replymessage,ReplyMessage):
- ToSend = replymessage
+ reply_msg = replymessage
else:
- ToSend = ReplyMessage(replymessage, reply_to)
- ToSend.status = status
- ToSend.errmsg = errtxt
- ToSend.backtrace = backtrace
+ reply_msg = ReplyMessage(replymessage, reply_to)
+ reply_msg.status = status
+ reply_msg.errmsg = errtxt
+ reply_msg.backtrace = backtrace
- # show the message content if required by the Verbose flag.
- if self.Verbose is True:
- ToSend.show()
+ # show the message content if required by the verbose flag.
+ if self.verbose == True:
+ reply_msg.show()
# send the result to the RPC client
- if isinstance(self.Reply,ToBus):
- ToSend.subject = reply_to
- self.Reply.send(ToSend)
+ if '/' in reply_to:
+ # sometimes clients (JAVA) setup the reply_to field as "exchange/key; {options}"
+ # make sure we can deal with that.
+ reply_address=reply_to.split('/')
+ num_parts=len(reply_address)
+ reply_busname=reply_address[num_parts-2]
+ subject=reply_address[num_parts-1]
+ try:
+ with ToBus(reply_busname) as dest:
+ # remove any extra field if present
+ if ';' in subject:
+ subject = subject.split(';')[0]
+ reply_msg.subject=subject
+ dest.send(reply_msg)
+ except MessageBusError as e:
+ logger.error("Failed to send reply message to reply address %s on messagebus %s." %(subject,reply_busname))
+ return
+
+ if isinstance(self.reply_bus,ToBus):
+ reply_msg.subject = reply_to
+ try:
+ self.reply_bus.send(reply_msg)
+ except MessageBusError as e:
+ logger.error("Failed to send reply message to reply address %s on messagebus %s." %(reply_to,self.busname))
+ return
else:
+ # the reply address is not in a default known format
+ # and we do not have a default bus destination
+ # we will try to deliver the message anyway.
try:
with ToBus(reply_to) as dest:
- dest.send(ToSend)
+ dest.send(reply_msg)
except MessageBusError as e:
- logger.error("Failed to send reply to reply address %s" %(reply_to))
+ logger.error("Failed to send reply messgage to reply address %s" %(reply_to))
- def _loop(self, index):
+ def _loop(self, **kwargs):
"""
Internal use only. Message listener loop that receives messages and starts the attached function with the message content as argument.
"""
- logger.info( "Thread %d START Listening for messages on Bus %s and service name %s." %(index, self.BusName, self.ServiceName))
- while self.running:
+ thread_idx = kwargs.pop("index")
+ service_handler = kwargs.pop("service_handler")
+ logger.info( "Thread %d START Listening for messages on Bus %s and service name %s." %(thread_idx, self.busname, self.service_name))
+ try:
+ service_handler.prepare_loop()
+ except Exception as e:
+ logger.error("prepare_loop() failed with %s", e)
+
+ while self.running[0]:
+ try:
+ service_handler.prepare_receive()
+ except Exception as e:
+ logger.error("prepare_receive() failed with %s", e)
+ continue
+
try:
# get the next message
- msg = self.Listen.receive(1)
+ msg = self.listener.receive(1)
# retry if timed-out
if msg is None:
continue
# report if messages are not Service Messages
- if isinstance(msg, ServiceMessage) is not True:
- logger.error( "Received wrong messagetype %s, ServiceMessage expected." %(str(type(msg))))
- self.Listen.ack(msg)
+ if isinstance(msg, RequestMessage) is not True:
+ logger.error( "Received wrong messagetype %s, RequestMessage expected." %(str(type(msg))))
+ self.listener.ack(msg)
continue
# Keep track of number of received messages
- self.reccounter[index] += 1
+ self.reccounter[thread_idx] += 1
# Execute the service handler function and send reply back to client
try:
self._debug("Running handler")
if self.parsefullmessage is True:
- replymessage = self.ServiceHandler(msg)
+ replymessage = service_handler.handle_message(msg)
else:
- replymessage = self.ServiceHandler(msg.content)
+ # check for positional arguments and named arguments
+ if msg.has_args=="True":
+ rpcargs=msg.content
+ if msg.has_kwargs=="True":
+ # both positional and named arguments
+ rpckwargs=rpcargs[-1]
+ del rpcargs[-1]
+ rpcargs=tuple(rpcargs)
+ replymessage = service_handler.handle_message(*rpcargs,**rpckwargs)
+ else:
+ # only positional arguments
+ rpcargs=tuple(rpcargs)
+ replymessage = service_handler.handle_message(*rpcargs)
+ else:
+ if msg.has_kwargs=="True":
+ # only named arguments
+ replymessage = service_handler.handle_message(**(msg.content))
+ else:
+ replymessage = service_handler.handle_message(msg.content)
+
self._debug("finished handler")
self._send_reply(replymessage,"OK",msg.reply_to)
- self.okcounter[index] += 1
- self.Listen.ack(msg)
+ self.okcounter[thread_idx] += 1
+ self.listener.ack(msg)
+ try:
+ service_handler.finalize_handling(True)
+ except Exception as e:
+ logger.error("finalize_handling() failed with %s", e)
continue
except Exception as e:
@@ -242,16 +376,28 @@ class Service:
del rawbacktrace[-1]
backtrace = ''.join(rawbacktrace).encode('latin-1').decode('unicode_escape')
self._debug(backtrace)
- if self.Verbose is True:
+ if self.verbose is True:
logger.info("[Service:] Status: %s", str(status))
logger.info("[Service:] ERRTXT: %s", str(errtxt))
logger.info("[Service:] BackTrace: %s", str( backtrace ))
self._send_reply(None, status, msg.reply_to, errtxt=errtxt, backtrace=backtrace)
+ try:
+ service_handler.finalize_handling(False)
+ except Exception as e:
+ logger.error("finalize_handling() failed with %s", e)
+ continue
except Exception as e:
# Unknown problem in the library. Report this and continue.
excinfo = sys.exc_info()
logger.error("[Service:] ERROR during processing of incoming message.")
traceback.print_exception(*excinfo)
- logger.info( "Thread %d: Resuming listening on bus %s for service %s" % (index, self.BusName, self.ServiceName))
+ logger.info("Thread %d: Resuming listening on bus %s for service %s" %
+ (thread_idx, self.busname, self.service_name))
+
+ try:
+ service_handler.finalize_loop()
+ except Exception as e:
+ logger.error("finalize_loop() failed with %s", e)
+__all__ = ["Service", "MessageHandlerInterface"]
diff --git a/LCS/Messaging/python/messaging/__init__.py b/LCS/Messaging/python/messaging/__init__.py
index ab3e262510f54c3fa75b93fc31e859908c22085f..995e83d2a4e8209510346fd76b098ade3174d8b0 100644
--- a/LCS/Messaging/python/messaging/__init__.py
+++ b/LCS/Messaging/python/messaging/__init__.py
@@ -27,5 +27,5 @@ Module initialization file.
from exceptions import *
from messages import *
from messagebus import *
-from RPC import RPC
-from Service import Service
+from RPC import *
+from Service import *
diff --git a/LCS/Messaging/python/messaging/messages.py b/LCS/Messaging/python/messaging/messages.py
index e934e0d2d50fafa8f8dccd8842d1016312e7f8c7..c79e5c0461611b14959c20b6d108ef818ed1b7f0 100644
--- a/LCS/Messaging/python/messaging/messages.py
+++ b/LCS/Messaging/python/messaging/messages.py
@@ -96,6 +96,33 @@ def to_qpid_message(msg):
return msg.qpid_msg
raise InvalidMessage("Invalid message type: %r" % type(msg))
+def analyze_args(args,kwargs):
+ HasKwArgs=(len(kwargs)>0)
+ # more than one argument given?
+ HasMultipleArgs=(len(args)> 1 ) or (( len(kwargs)>0 ) and (len(args)>0))
+ return (HasMultipleArgs,HasKwArgs)
+
+def args_as_content(*args,**kwargs):
+ """
+ Convert positional args and named args into a message body.
+ :param msg: Message to be converted into a Qpid message.
+ :return: Qpid message
+ :raise InvalidMessage if `msg` cannot be converted into a Qpid message.
+ """
+ HasMultipleArgs,HasKwArgs = analyze_args(args, kwargs)
+ if HasMultipleArgs:
+ # convert arguments to list
+ Content = list(args)
+ if HasKwArgs:
+ # if both positional and named arguments then
+ # we add the kwargs dictionary as the last item in the list
+ Content.append(kwargs)
+ return Content
+ if HasKwArgs:
+ # we have only one named argument
+ return kwargs
+ # we have only one positional argument
+ return list(args)[0]
class MessageFactory(Factory):
"""
@@ -158,7 +185,11 @@ class LofarMessage(object):
self.__dict__['_qpid_msg'] = content
else:
try:
- self.__dict__['_qpid_msg'] = qpid.messaging.Message(content)
+ if isinstance(content,basestring):
+ self.__dict__['_qpid_msg'] = qpid.messaging.Message(unicode(content))
+ else:
+ self.__dict__['_qpid_msg'] = qpid.messaging.Message(content)
+
except KeyError:
raise InvalidMessage(
"Unsupported content type: %r" % type(content))
@@ -241,9 +272,11 @@ class EventMessage(LofarMessage):
will be stored in a persistent queue for later delivery.
"""
- def __init__(self, content=None):
+ def __init__(self, content=None, context=None):
super(EventMessage, self).__init__(content)
- self.durable = True
+ if (context!=None):
+ self.durable = True
+ self.subject = context
class MonitoringMessage(LofarMessage):
@@ -268,18 +301,22 @@ class ProgressMessage(LofarMessage):
super(ProgressMessage, self).__init__(content)
-class ServiceMessage(LofarMessage):
+class RequestMessage(LofarMessage):
"""
Message class used for service messages. Service messages are
request-reply type of messages. They are typically used to query a
subsystem. A service message must contain a valid ``ReplyTo`` property.
"""
- def __init__(self, content=None, reply_to=None):
- super(ServiceMessage, self).__init__(content)
+ def __init__(self, content=None, reply_to=None,**kwargs): #reply_to=None, has_args=None, has_kwargs=None):
+ super(RequestMessage, self).__init__(content)
if (reply_to!=None):
- self.reply_to = reply_to
-
+ #if (len(kwargs)>0):
+ #reply_to = kwargs.pop("reply_to",None)
+ #if (reply_to!=None):
+ self.reply_to = reply_to
+ self.has_args = str(kwargs.pop("has_args",False))
+ self.has_kwargs = str(kwargs.pop("has_kwargs",False))
class ReplyMessage(LofarMessage):
"""
@@ -298,8 +335,8 @@ class ReplyMessage(LofarMessage):
MESSAGE_FACTORY.register("EventMessage", EventMessage)
MESSAGE_FACTORY.register("MonitoringMessage", MonitoringMessage)
MESSAGE_FACTORY.register("ProgressMessage", ProgressMessage)
-MESSAGE_FACTORY.register("ServiceMessage", ServiceMessage)
+MESSAGE_FACTORY.register("RequestMessage", RequestMessage)
MESSAGE_FACTORY.register("ReplyMessage", ReplyMessage)
__all__ = ["EventMessage", "MonitoringMessage", "ProgressMessage",
- "ServiceMessage", "ReplyMessage"]
+ "RequestMessage", "ReplyMessage"]
diff --git a/LCS/Messaging/python/messaging/test/CMakeLists.txt b/LCS/Messaging/python/messaging/test/CMakeLists.txt
index 12bdeb7d46f7690b42347c09a2cbcb0d24f5e889..b4816dd96b7b390e47416c128a367e64345c46e6 100644
--- a/LCS/Messaging/python/messaging/test/CMakeLists.txt
+++ b/LCS/Messaging/python/messaging/test/CMakeLists.txt
@@ -5,3 +5,4 @@ include(LofarCTest)
lofar_add_test(t_messages)
lofar_add_test(t_messagebus)
lofar_add_test(t_RPC)
+lofar_add_test(t_service_message_handler)
diff --git a/LCS/Messaging/python/messaging/test/t_RPC.py b/LCS/Messaging/python/messaging/test/t_RPC.py
index 35aec35ad63b3e02224aa205b79b0535cfc22d48..f1968d7870a88896b22445655b9b74da96203006 100644
--- a/LCS/Messaging/python/messaging/test/t_RPC.py
+++ b/LCS/Messaging/python/messaging/test/t_RPC.py
@@ -112,38 +112,38 @@ if __name__ == '__main__':
busname = sys.argv[1] if len(sys.argv) > 1 else "simpletest"
# Register functs as a service handler listening at busname and ServiceName
- serv1 = Service(busname, "ErrorService", ErrorFunc, numthreads=1)
- serv2 = Service(busname, "ExceptionService", ExceptionFunc, numthreads=1)
- serv3 = Service(busname, "StringService", StringFunc, numthreads=1)
- serv4 = Service(busname, "ListService", ListFunc, numthreads=1)
- serv5 = Service(busname, "DictService", DictFunc, numthreads=1)
+ serv1 = Service("ErrorService", ErrorFunc, busname=busname, numthreads=1)
+ serv2 = Service("ExceptionService", ExceptionFunc, busname=busname, numthreads=1)
+ serv3 = Service("StringService", StringFunc, busname=busname, numthreads=1)
+ serv4 = Service("ListService", ListFunc, busname=busname, numthreads=1)
+ serv5 = Service("DictService", DictFunc, busname=busname, numthreads=1)
# 'with' sets up the connection context and defines the scope of the service.
with nested(serv1, serv2, serv3, serv4, serv5):
# Start listening in the background. This will start as many threads as defined by the instance
- serv1.StartListening()
- serv2.StartListening()
- serv3.StartListening()
- serv4.StartListening()
- serv5.StartListening()
+ serv1.start_listening()
+ serv2.start_listening()
+ serv3.start_listening()
+ serv4.start_listening()
+ serv5.start_listening()
# Redo all tests but via through RPC
# ErrorFunc
- with RPC(busname, "ErrorService") as rpc:
+ with RPC("ErrorService", busname=busname) as rpc:
try:
result = rpc("aap noot mies")
except UserException as e:
pass
# ExceptionFunc
- with RPC(busname, "ExceptionService") as rpc:
+ with RPC("ExceptionService", busname=busname) as rpc:
try:
result = rpc("aap noot mies")
except IndexError as e:
pass
# StringFunc
- with RPC(busname, "StringService") as rpc:
+ with RPC("StringService", busname=busname) as rpc:
try:
result = rpc([25])
except InvalidArgType as e:
@@ -153,7 +153,7 @@ if __name__ == '__main__':
raise Exception("String function failed:{}".format(result))
# ListFunc
- with RPC(busname, "ListService") as rpc:
+ with RPC("ListService", busname=busname) as rpc:
try:
result = rpc("25")
except InvalidArgType as e:
@@ -163,7 +163,7 @@ if __name__ == '__main__':
raise Exception("List function failed:{}".format(result))
# DictFunc
- with RPC(busname, "DictService") as rpc:
+ with RPC("DictService", busname=busname) as rpc:
try:
result = rpc([25])
except InvalidArgType as e:
@@ -175,8 +175,8 @@ if __name__ == '__main__':
print "Functions tested with RPC: All OK"
# Tell all background listener threads to stop and wait for them to finish.
- serv1.StopListening()
- serv2.StopListening()
- serv3.StopListening()
- serv4.StopListening()
- serv5.StopListening()
+ serv1.stop_listening()
+ serv2.stop_listening()
+ serv3.stop_listening()
+ serv4.stop_listening()
+ serv5.stop_listening()
diff --git a/LCS/Messaging/python/messaging/test/t_messagebus.py b/LCS/Messaging/python/messaging/test/t_messagebus.py
index 1f511b18cb9ea4cf97b33fcb7a8d2905716ff713..68a454f9cfc404ecbb92ed2390a358c10678155a 100644
--- a/LCS/Messaging/python/messaging/test/t_messagebus.py
+++ b/LCS/Messaging/python/messaging/test/t_messagebus.py
@@ -33,7 +33,7 @@ from lofar.messaging.messages import *
from lofar.messaging.messagebus import *
from lofar.messaging.exceptions import MessageBusError, InvalidMessage
-TIMEOUT = 0.1
+TIMEOUT = 1.0
# ======== FromBus unit tests ======== #
@@ -245,9 +245,8 @@ class SendReceiveMessage(unittest.TestCase):
Helper class that implements the send/receive logic and message checks.
:param send_msg: Message to send
"""
- with self.tobus:
+ with self.tobus, self.frombus:
self.tobus.send(send_msg)
- with self.frombus:
recv_msg = self.frombus.receive(timeout=TIMEOUT)
self.frombus.ack(recv_msg)
self.assertEqual(
@@ -278,12 +277,12 @@ class SendReceiveMessage(unittest.TestCase):
content = {"Progress": "Message"}
self._test_sendrecv(ProgressMessage(content))
- def test_sendrecv_service_message(self):
+ def test_sendrecv_request_message(self):
"""
- Test send/receive of an ServiceMessage, containing a byte array.
+ Test send/receive of an RequestMessage, containing a byte array.
"""
- content = struct.pack("17B", *(ord(c)+32 for c in "A service message"))
- self._test_sendrecv(ServiceMessage(content, reply_to=QUEUE))
+ content = {"request": "Do Something", "argument": "Very Often"}
+ self._test_sendrecv(RequestMessage(content, reply_to=QUEUE))
if __name__ == '__main__':
diff --git a/LCS/Messaging/python/messaging/test/t_messages.py b/LCS/Messaging/python/messaging/test/t_messages.py
index a6cf4f2b1b91964a07870c820e43a220cb5700b7..ed360c35aef784c2e1bea4f4312ca37e7163b909 100644
--- a/LCS/Messaging/python/messaging/test/t_messages.py
+++ b/LCS/Messaging/python/messaging/test/t_messages.py
@@ -235,7 +235,7 @@ class ContentLofarMessage(unittest.TestCase):
content = "ASCII string"
msg = LofarMessage(content)
self.assertEqual((msg.content, msg.content_type),
- (content, None))
+ (unicode(content), 'text/plain'))
def test_construct_from_unicode(self):
"""
@@ -265,15 +265,15 @@ class ContentLofarMessage(unittest.TestCase):
self.assertEqual((msg.content, msg.content_type),
(content, "amqp/map"))
- def test_construct_from_binary(self):
- """
- Test that an LofarMessage can be constructed from binary data.
- Use struct.pack() to create a byte array
- """
- content = struct.pack("<256B", *range(256))
- msg = LofarMessage(content)
- self.assertEqual((msg.content, msg.content_type),
- (content, None))
+ # def test_construct_from_binary(self):
+ # """
+ # Test that an LofarMessage can be constructed from binary data.
+ # Use struct.pack() to create a byte array
+ # """
+ # content = struct.pack("<256B", *range(256))
+ # msg = LofarMessage(content)
+ # self.assertEqual((msg.content, msg.content_type),
+ # (content, None))
def test_construct_from_unsupported(self):
"""
diff --git a/LCS/Messaging/python/messaging/test/t_service_message_handler.py b/LCS/Messaging/python/messaging/test/t_service_message_handler.py
new file mode 100644
index 0000000000000000000000000000000000000000..653a4854ff9428971e8593d2e6ffdbb289923e02
--- /dev/null
+++ b/LCS/Messaging/python/messaging/test/t_service_message_handler.py
@@ -0,0 +1,196 @@
+#!/usr/bin/env python
+"""
+Program to test the RPC and Service class of the Messaging package.
+It defines 5 functions and first calls those functions directly to check
+that the functions are OK. Next the same tests are done with the RPC and
+Service classes in between. This should give the same results.
+"""
+import logging
+import sys
+import time
+from lofar.messaging import *
+
+logging.basicConfig(stream=sys.stdout, level=logging.WARNING)
+
+class UserException(Exception):
+ "Always thrown in one of the functions"
+ pass
+class InvalidArgType(Exception):
+ "Thrown when the input is wrong for one of the functions"
+ pass
+
+# create several function:
+def ErrorFunc(input_value):
+ " Always thrown a predefined exception"
+ raise UserException("Exception thrown by the user")
+
+def ExceptionFunc(input_value):
+ "Generate a exception not caught by the function"
+ a = "aap"
+ b = a[23]
+
+def StringFunc(input_value):
+ "Convert the string to uppercase."
+ if not isinstance(input_value, str) and not isinstance(input_value, unicode):
+ raise InvalidArgType("Input value must be of the type 'string'")
+ return input_value.upper()
+
+class OnlyMessageHandling(MessageHandlerInterface):
+ def __init__(self, **kwargs):
+ MessageHandlerInterface.__init__(self)
+ print "Creation of OnlyMessageHandling class: %s" % kwargs
+ self.handle_message = kwargs.pop("function")
+ self.args = kwargs
+
+class FullMessageHandling(MessageHandlerInterface):
+ def __init__(self, **kwargs):
+ MessageHandlerInterface.__init__(self)
+ print "Creation of FullMessageHandling class: %s" % kwargs
+ self.handle_message = kwargs.pop("function")
+ self.args = kwargs
+ def prepare_loop(self):
+ print "FullMessageHandling prepare_loop: %s" % self.args
+ def prepare_receive(self):
+ print "FullMessageHandling prepare_receive: %s" % self.args
+ def finalize_handling(self, successful):
+ print "FullMessageHandling finalize_handling: %s" % self.args
+ def finalize_loop(self):
+ print "FullMessageHandling finalize_loop: %s" % self.args
+
+class FailingMessageHandling(MessageHandlerInterface):
+ def __init__(self, **kwargs):
+ MessageHandlerInterface.__init__(self)
+ print "Creation of FailingMessageHandling class: %s" % kwargs
+ self.handle_message = kwargs.pop("function")
+ self.args = kwargs
+ self.counter = 0
+ def prepare_loop(self):
+ print "FailingMessageHandling prepare_loop: %s" % self.args
+ raise UserException("oops in prepare_loop()")
+ def prepare_receive(self):
+ # allow one succesfull call otherwise the main loop never accepts the message :-)
+ print "FailingMessageHandling prepare_receive: %s" % self.args
+ if self.counter:
+ time.sleep(1) # Prevent running around too fast
+ raise UserException("oops in prepare_receive(%d)" % self.counter)
+ else:
+ self.counter = self.counter + 1
+ def finalize_handling(self, successful):
+ print "FailingMessageHandling finalize_handling: %s, %s" % (self.args, successful)
+ raise UserException("oops in finalize_handling()")
+ def finalize_loop(self):
+ print "FailingMessageHandling finalize_loop: %s" % self.args
+ raise UserException("oops in finalize_loop()")
+
+if __name__ == '__main__':
+ busname = sys.argv[1] if len(sys.argv) > 1 else "simpletest"
+
+ # Register functs as a service handler listening at busname and ServiceName
+ serv1_plain = Service("String1Service", StringFunc, busname=busname, numthreads=1, startonwith=True)
+ serv1_minimal_class = Service("String2Service", OnlyMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : StringFunc})
+ serv1_full_class = Service("String3Service", FullMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : StringFunc})
+ serv1_failing_class = Service("String4Service", FailingMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : StringFunc})
+
+ # 'with' sets up the connection context and defines the scope of the service.
+ with serv1_plain, serv1_minimal_class, serv1_full_class, serv1_failing_class:
+ # Redo string tests via RPC
+ with RPC("String1Service", ForwardExceptions=True, busname=busname) as rpc:
+ result = rpc("aap noot mies")
+ if result[0] != "AAP NOOT MIES":
+ raise Exception("String function failed of String1Service:{}".format(result))
+ print "string1Service is OK"
+
+ with RPC("String2Service", ForwardExceptions=True, busname=busname) as rpc:
+ result = rpc("aap noot mies")
+ if result[0] != "AAP NOOT MIES":
+ raise Exception("String function failed of String2Service:{}".format(result))
+ print "string2Service is OK"
+
+ with RPC("String3Service", ForwardExceptions=True, busname=busname) as rpc:
+ result = rpc("aap noot mies")
+ if result[0] != "AAP NOOT MIES":
+ raise Exception("String function failed of String3Service:{}".format(result))
+ print "string3Service is OK"
+
+ with RPC("String4Service", ForwardExceptions=True, busname=busname) as rpc:
+ result = rpc("aap noot mies")
+ if result[0] != "AAP NOOT MIES":
+ raise Exception("String function failed of String4Service:{}".format(result))
+ print "string4Service is OK"
+
+ # Register functs as a service handler listening at busname and ServiceName
+ serv2_plain = Service("Error1Service", ErrorFunc, busname=busname, numthreads=1, startonwith=True)
+ serv2_minimal_class = Service("Error2Service", OnlyMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ErrorFunc})
+ serv2_full_class = Service("Error3Service", FullMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ErrorFunc})
+ serv2_failing_class = Service("Error4Service", FailingMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ErrorFunc})
+
+ # 'with' sets up the connection context and defines the scope of the service.
+ with serv2_plain, serv2_minimal_class, serv2_full_class, serv2_failing_class:
+ # Redo Error tests via RPC
+ with RPC("Error1Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except RPCException as e:
+ print "Error1Service is OK"
+
+ with RPC("Error2Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except RPCException as e:
+ print "Error2Service is OK"
+
+ with RPC("Error3Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except RPCException as e:
+ print "Error3Service is OK"
+
+ with RPC("Error4Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except Exception as e:
+ print "Error4Service is OK"
+
+ # Register functs as a service handler listening at busname and ServiceName
+ serv3_plain = Service("Except1Service", ExceptionFunc, busname=busname, numthreads=1, startonwith=True)
+ serv3_minimal_class = Service("Except2Service", OnlyMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ExceptionFunc})
+ serv3_full_class = Service("Except3Service", FullMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ExceptionFunc})
+ serv3_failing_class = Service("Except4Service", FailingMessageHandling, busname=busname, numthreads=1, startonwith=True,
+ handler_args={"function" : ExceptionFunc})
+
+ # 'with' sets up the connection context and defines the scope of the service.
+ with serv3_plain, serv3_minimal_class, serv3_full_class, serv3_failing_class:
+ # Redo exception tests via RPC
+ with RPC("Except1Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except IndexError as e:
+ print "Except1Service is OK"
+
+ with RPC("Except2Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except IndexError as e:
+ print "Except2Service is OK"
+
+ with RPC("Except3Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except IndexError as e:
+ print "Except3Service is OK"
+
+ with RPC("Except4Service", ForwardExceptions=True, busname=busname) as rpc:
+ try:
+ result = rpc("aap noot mies")
+ except IndexError as e:
+ print "Except4Service is OK"
+
+ print "Functions tested with RPC: All OK"
diff --git a/LCS/Messaging/python/messaging/test/t_service_message_handler.run b/LCS/Messaging/python/messaging/test/t_service_message_handler.run
new file mode 100755
index 0000000000000000000000000000000000000000..67ad6c38702db4a0169108c309572f22434f4109
--- /dev/null
+++ b/LCS/Messaging/python/messaging/test/t_service_message_handler.run
@@ -0,0 +1,38 @@
+#!/bin/sh -e
+
+#cleanup on normal exit and on SIGHUP, SIGINT, SIGQUIT, and SIGTERM
+trap 'qpid-config del exchange --force $queue' 0 1 2 3 15
+
+# Generate randome queue name
+queue=$(< /dev/urandom tr -dc [:alnum:] | head -c16)
+
+# Create the queue
+qpid-config add exchange topic $queue
+
+# Run the unit test
+# either with or without code coverage measurements,
+# depending wheter coverage has been installed
+
+if type "coverage" > /dev/null; then
+ #run test using python coverage tool
+
+ #erase previous results
+ coverage erase
+
+ #setup coverage config file
+ printf "[report]\nexclude_lines = \n if __name__ == .__main__.\n def main\n" > .coveragerc
+
+ coverage run --branch --include=*Messaging/python* t_service_message_handler.py $queue
+ RESULT=$?
+ if [ $RESULT -eq 0 ]; then
+ echo " *** Code coverage results *** "
+ coverage report -m
+ echo " *** End coverage results *** "
+ fi
+ exit $RESULT
+else
+ #coverage not available
+ echo "Please run: 'pip install coverage' to enable code coverage reporting of the unit tests"
+ #run plain test script
+ python t_service_message_handler.py $queue
+fi
diff --git a/LCS/Messaging/python/messaging/test/t_service_message_handler.sh b/LCS/Messaging/python/messaging/test/t_service_message_handler.sh
new file mode 100755
index 0000000000000000000000000000000000000000..ef9fd1a7376f292a74ae2168e7861c05dd8fcf97
--- /dev/null
+++ b/LCS/Messaging/python/messaging/test/t_service_message_handler.sh
@@ -0,0 +1,4 @@
+#!/bin/sh
+./runctest.sh t_service_message_handler
+
+
diff --git a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.cc b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.cc
index 9a81480094e613fa4e1318cce2aa003475d01a4f..268a664f815bdd046c72c783ba3a28f8f5019560 100644
--- a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.cc
+++ b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.cc
@@ -99,6 +99,14 @@ MACScheduler::MACScheduler() :
itsFinishedPeriod= globalParameterSet()->getTime("finishedPeriod", 86400) / 60; // in minutes
itsMaxPlanned = globalParameterSet()->getTime("maxPlannedList", 30);
itsMaxFinished = globalParameterSet()->getTime("maxFinishedList", 40);
+ itsExclPLcluster = globalParameterSet()->getString("excludePipelinesOnThisCluster", "");
+ if (itsExclPLcluster.length() > 0) {
+ LOG_INFO_STR("NOT running the pipelines on cluster: " << itsExclPLcluster);
+ // We need to exclude this cluster for pipelines so make sure the name begins with the 'not'-sign.
+ if (itsExclPLcluster[0] != '!') {
+ itsExclPLcluster.insert(0, 1, '!');
+ }
+ }
ASSERTSTR(itsMaxPlanned + itsMaxFinished < MAX_CONCURRENT_OBSERVATIONS,
"maxPlannedList + maxFinishedList should be less than " << MAX_CONCURRENT_OBSERVATIONS);
@@ -623,8 +631,8 @@ void MACScheduler::_updatePlannedList()
ptime currentTime = from_time_t(now);
ASSERTSTR (currentTime != not_a_date_time, "Can't determine systemtime, bailing out");
- // get new list (list is ordered on starttime)
- vector<OTDBtree> plannedDBlist = itsOTDBconnection->getTreeGroup(1, itsPlannedPeriod); // planned observations
+ // get new list (list is ordered on starttime) of planned observations
+ vector<OTDBtree> plannedDBlist = itsOTDBconnection->getTreeGroup(1, itsPlannedPeriod, itsExclPLcluster);
if (!plannedDBlist.empty()) {
LOG_DEBUG(formatString("OTDBCheck:First planned observation (%d) is at %s (active over %d seconds)",
@@ -754,7 +762,7 @@ void MACScheduler::_updateActiveList()
LOG_DEBUG("_updateActiveList()");
// get new list (list is ordered on starttime)
- vector<OTDBtree> activeDBlist = itsOTDBconnection->getTreeGroup(2, 0);
+ vector<OTDBtree> activeDBlist = itsOTDBconnection->getTreeGroup(2, 0, itsExclPLcluster);
if (activeDBlist.empty()) {
LOG_DEBUG ("No active Observations");
// NOTE: do not exit routine on emptylist: we need to write an empty list to clear the DB
@@ -797,7 +805,7 @@ void MACScheduler::_updateFinishedList()
LOG_DEBUG("_updateFinishedList()");
// get new list (list is ordered on starttime)
- vector<OTDBtree> finishedDBlist = itsOTDBconnection->getTreeGroup(3, itsFinishedPeriod);
+ vector<OTDBtree> finishedDBlist = itsOTDBconnection->getTreeGroup(3, itsFinishedPeriod, itsExclPLcluster);
if (finishedDBlist.empty()) {
LOG_DEBUG ("No finished Observations");
// NOTE: do not exit routine on emptylist: we need to write an empty list to clear the DB
diff --git a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.conf.in b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.conf.in
index 911f75463162324556fcb8d03141d9aece3381b6..ed9c486a187ff5e142c65ff6fd5a39ac561f72e8 100644
--- a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.conf.in
+++ b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.conf.in
@@ -30,3 +30,7 @@ maxFinishedList = 40 # Never show more finished observations
#ChildControl.MaxStartupRetry = 5
ParsetQueuename = lofar.task.specification.system
+
+# Pipelines on cluster X can be ignored by the MACScheduler with this key.
+# use e.g. 'CEP2' or 'CEP4'
+excludePipelinesOnThisCluster = ''
diff --git a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.h b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.h
index 5654ea1434ea58adf785ddd095daf398e81aac09..188d01364883ed3649aa4a234e3233eda771c34c 100644
--- a/MAC/APL/MainCU/src/MACScheduler/MACScheduler.h
+++ b/MAC/APL/MainCU/src/MACScheduler/MACScheduler.h
@@ -168,6 +168,9 @@ private:
// OTDB related variables.
OTDB::OTDBconnection* itsOTDBconnection; // connection to the database
+ // Cluster to exclude for pipelines. Key is used in the getTreeGroup stored-procedure in OTDB.
+ string itsExclPLcluster; // like !CEP2 or !CEP4
+
// Messagebus related variables
ToBus* itsMsgQueue; // Bus used for sending
};
diff --git a/MAC/APL/MainCU/src/MACScheduler/MACSchedulerMain.cc b/MAC/APL/MainCU/src/MACScheduler/MACSchedulerMain.cc
index 79ede7d94fe484fececff88c7e8db6692cf2d601..6d8caa91d240823ee307ba586b453bf28ae0091a 100644
--- a/MAC/APL/MainCU/src/MACScheduler/MACSchedulerMain.cc
+++ b/MAC/APL/MainCU/src/MACScheduler/MACSchedulerMain.cc
@@ -39,7 +39,7 @@ int main(int argc, char* argv[])
{
GCFScheduler::instance()->init(argc, argv, "MACScheduler");
- MessageBus::init();
+ MessageBus::init();
ChildControl* cc = ChildControl::instance();
cc->start(); // make initial transition
diff --git a/SAS/OTDB/include/OTDB/OTDBconnection.h b/SAS/OTDB/include/OTDB/OTDBconnection.h
index f47b0e6c3d42e2fb9bc46701e9aa66a7013d29db..878ef8ce06621f79186dd96edfeb6d71afe15e5c 100644
--- a/SAS/OTDB/include/OTDB/OTDBconnection.h
+++ b/SAS/OTDB/include/OTDB/OTDBconnection.h
@@ -100,7 +100,7 @@ public:
// groupType = 1: observations that are scheduled to start the next 'period' minutes
// 2: active observations ; period is ignored
// 3: observations that were finished during the last 'period' minutes
- vector<OTDBtree> getTreeGroup(uint32 groupType, uint32 periodInMinutes);
+ vector<OTDBtree> getTreeGroup(uint32 groupType, uint32 periodInMinutes, const string& cluster="");
// Get a list of all trees that are scheduled in the given period (partially).
vector<OTDBtree> getTreesInPeriod(treeType aTreeType,
diff --git a/SAS/OTDB/sql/README! b/SAS/OTDB/sql/README!
new file mode 100644
index 0000000000000000000000000000000000000000..8e587dde500f09a82e96ebb2fe47a549ff9a2a60
--- /dev/null
+++ b/SAS/OTDB/sql/README!
@@ -0,0 +1,3 @@
+This version needs a modification on the statehistory table.
+
+please execute (with \i) the upgrade_OTDB.sql script to upgrade an existing database to the new format.
diff --git a/SAS/OTDB/sql/create_base_tables.sql b/SAS/OTDB/sql/create_base_tables.sql
index ffdd15974233d62a206bed8e5ad2ef6aa1656d78..455c46062a91dc1db193430cb0c6113498c77d91 100644
--- a/SAS/OTDB/sql/create_base_tables.sql
+++ b/SAS/OTDB/sql/create_base_tables.sql
@@ -297,6 +297,17 @@ CREATE TABLE operator (
INSERT INTO operator VALUES (1, 'eucalypta', '0612345678');
INSERT INTO operator VALUES (2, 'gargamel', '0123456789');
+--
+-- otdb_admin table
+--
+-- Internal administration. Tables always has 1 record!
+-- NEVER DROP THIS TABLE!
+--
+CREATE TABLE otdb_admin (
+ treestatusevent timestamp(6)
+) WITHOUT OIDS;
+INSERT INTO otdb_admin VALUES(now());
+
--
-- ProcessType table
--
diff --git a/SAS/OTDB/sql/create_tree_table.sql b/SAS/OTDB/sql/create_tree_table.sql
index cd314a101bc809348499a6c9a56d2f4c2e613580..fefee8dd14a371bf4f7009e5792848e9057015e5 100644
--- a/SAS/OTDB/sql/create_tree_table.sql
+++ b/SAS/OTDB/sql/create_tree_table.sql
@@ -37,6 +37,7 @@ DROP SEQUENCE IF EXISTS OTDBtreeID;
DROP SEQUENCE IF EXISTS OTDBgroupID;
DROP TABLE IF EXISTS StateHistory CASCADE;
DROP INDEX IF EXISTS otdbtree_treeid_indx;
+DROP INDEX IF EXISTS statehist_creation_idx;
CREATE SEQUENCE OTDBtreeID START 1;
-- Create a new start number based on current time. To prevent overlap in treeid's
@@ -98,6 +99,9 @@ CREATE TABLE StateHistory (
momID INT4 NOT NULL,
state INT2 NOT NULL,
userID INT4 NOT NULL REFERENCES operator(ID),
- timestamp TIMESTAMP(0) DEFAULT now()
+ timestamp TIMESTAMP(0) DEFAULT now(),
+ creation TIMESTAMP(6) DEFAULT now()
) WITHOUT OIDS;
+CREATE INDEX statehist_creation_idx ON statehistory(creation);
+
diff --git a/SAS/OTDB/sql/exportTree_func.sql b/SAS/OTDB/sql/exportTree_func.sql
index 0d525c743bdda86d78378d7f0fb6c8f6cc7aab4d..461721ee6798ac6d482a20511d815d621a7be294 100644
--- a/SAS/OTDB/sql/exportTree_func.sql
+++ b/SAS/OTDB/sql/exportTree_func.sql
@@ -228,30 +228,27 @@ CREATE OR REPLACE FUNCTION exportPICSubTree(INT4, INT4, INT4)
DECLARE
vResult TEXT := '';
vRow RECORD;
+ vName PIChierarchy.name%TYPE;
BEGIN
- -- first dump own parameters
- FOR vRow IN
- SELECT name --, value
- FROM PIChierarchy
- WHERE treeID = $1
- AND parentID = $2
- AND leaf = true
- ORDER BY name
- LOOP
- vResult := vResult || substr(vRow.name,$3) || chr(10);
- END LOOP;
+ -- first get name of top node
+ SELECT name
+ INTO vName
+ FROM PIChierarchy
+ WHERE treeID = $1
+ AND nodeID = $2;
+ vName := vName || '%';
-- call myself for all the children
FOR vRow IN
- SELECT nodeID, name
+ SELECT name
FROM PIChierarchy
WHERE treeID = $1
- AND parentID = $2
- AND leaf = false
- ORDER BY name
+ AND name like vName
+ AND leaf = true
+ ORDER BY nodeid
LOOP
- vResult := vResult || exportPICSubTree($1, vRow.nodeID, $3);
+ vResult := vResult || substr(vRow.name,$3) || chr(10);
END LOOP;
RETURN vResult;
diff --git a/SAS/OTDB/sql/getStateChanges_func.sql b/SAS/OTDB/sql/getStateChanges_func.sql
new file mode 100644
index 0000000000000000000000000000000000000000..c4c71ae43207df8db9d034ec2dcad29c394a9990
--- /dev/null
+++ b/SAS/OTDB/sql/getStateChanges_func.sql
@@ -0,0 +1,88 @@
+--
+-- getStateChanges.sql: function for getting state changes in a given period
+--
+-- Copyright (C) 2015
+-- ASTRON (Netherlands Foundation for Research in Astronomy)
+-- P.O.Box 2, 7990 AA Dwingeloo, The Netherlands, softwaresupport@astron.nl
+--
+-- This program 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 2 of the License, or
+-- (at your option) any later version.
+--
+-- This program 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 this program; if not, write to the Free Software
+-- Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+--
+-- $Id: getStateList_func.sql 30919 2015-02-05 15:26:22Z amesfoort $
+--
+
+--
+-- getStateChanges(begindate, enddate)
+--
+-- Get a list of statechanges.
+--
+-- Authorisation: none
+--
+-- Tables: otdbtree read
+-- otdbuser read
+--
+-- Types: treeState
+--
+DROP TYPE IF EXISTS stateChange CASCADE;
+CREATE TYPE stateChange AS (
+ -- $Id: create_types.sql 30919 2015-02-05 15:26:22Z amesfoort $
+ treeID INT4, -- OTDBtree.treeID%TYPE,
+ momID INT4, -- OTDBtree.momID%TYPE,
+ state INT2, -- treestate.ID%TYPE,
+ username VARCHAR(20), -- OTDBuser.username%TYPE,
+ modtime timestamp(0),
+ creation timestamp(6)
+);
+
+CREATE OR REPLACE FUNCTION getStateChanges(TIMESTAMP(6), TIMESTAMP(6))
+ RETURNS SETOF stateChange AS '
+ -- $Id: getStateList_func.sql 30919 2015-02-05 15:26:22Z amesfoort $
+ DECLARE
+ vRecord RECORD;
+ vQuery TEXT;
+ vKeyField TEXT;
+
+ BEGIN
+ -- create query for treeID (when filled)
+ vQuery := \'WHERE \';
+ IF $1 IS NOT NULL THEN
+ IF $2 IS NULL THEN
+ vQuery := vQuery || \'s.creation >\' || chr(39) || $1 || chr(39);
+ ELSE
+ vQuery := vQuery || \'s.creation >\' || chr(39) || $1 || chr(39)
+ || \' AND s.creation <=\' || chr(39) || $2 || chr(39);
+ END IF;
+ END IF;
+
+ -- do selection
+ FOR vRecord IN EXECUTE \'
+ SELECT s.treeID,
+ s.momID,
+ s.state,
+ u.username,
+ s.timestamp,
+ s.creation
+ FROM StateHistory s
+ INNER JOIN OTDBuser u ON s.userid = u.userid
+ \' || vQuery || \'
+ ORDER BY s.creation ASC\'
+ LOOP
+ RETURN NEXT vRecord;
+ END LOOP;
+ RETURN;
+ END
+' LANGUAGE plpgsql;
+
+
+
diff --git a/SAS/OTDB/sql/getTreeGroup_func.sql b/SAS/OTDB/sql/getTreeGroup_func.sql
index 013d505bae4f48b8df2b338c8c621f3c9e8bdfab..b1b1dc95e23e548a1f9c26d4c5f90e945732165f 100644
--- a/SAS/OTDB/sql/getTreeGroup_func.sql
+++ b/SAS/OTDB/sql/getTreeGroup_func.sql
@@ -1,5 +1,5 @@
--
--- getTreesInPeriod.sql: function for getting treeinfo from the OTDB
+-- getTreeGroup.sql: function for getting treeinfo from the OTDB
--
-- Copyright (C) 2005
-- ASTRON (Netherlands Foundation for Research in Astronomy)
@@ -23,7 +23,7 @@
--
--
--- getTreesInPeriod (groupType, periodInMinutes)
+-- getTreeGroup (groupType, periodInMinutes, cluster)
--
-- groupType = 0: get all trees that have to be scheduled
-- 1: get all trees that are scheduled to start in the period: now till now+period
@@ -31,6 +31,11 @@
-- 3: get all trees with stoptime in the period: now-period till now
-- 4: get all trees with stoptime < now and have state >= APPROVED
--
+-- cluster : '' Does not include limitation to the selection.
+-- 'CEP2' Returns all PIPELINES that belong to the groupType and that are assigned to CEP2
+-- '!CEP2' Returns all trees that belongs to the groupType EXCEPT the PIPELINES that are assigned to CEP2
+-- Other values are also allowed like CEP4 en !CEP4
+--
-- With this function we can get the planned, active or finished trees from the database.
--
-- Authorisation: none
@@ -41,7 +46,7 @@
--
-- Types: treeInfo
--
-CREATE OR REPLACE FUNCTION getTreeGroup(INT, INT)
+CREATE OR REPLACE FUNCTION getTreeGroup(INT, INT, VARCHAR(20))
RETURNS SETOF treeInfo AS $$
-- $Id: addComponentToVT_func.sql 19935 2012-01-25 09:06:14Z mol $
DECLARE
@@ -53,31 +58,36 @@ CREATE OR REPLACE FUNCTION getTreeGroup(INT, INT)
TSfinished CONSTANT INT2 := 1000;
TThierarchy CONSTANT INT2 := 30;
TCoperational CONSTANT INT2 := 3;
+ vWhere TEXT;
vQuery TEXT;
vSortOrder TEXT;
+ vExcept TEXT;
+ vCluster TEXT;
BEGIN
- vQuery := '';
+ vWhere := '';
+ vQuery := 'SELECT v.* FROM VICtrees v';
vSortOrder := 't.starttime, t.treeID';
+ vExcept := '';
IF $1 = 0 THEN
- vQuery := ' AND (t.stoptime > now() OR t.stoptime IS NULL) ';
+ vWhere := ' AND (t.stoptime > now() OR t.stoptime IS NULL) ';
ELSE
IF $1 = 1 THEN
- vQuery := ' AND (t.state = ' || TSscheduled || ' OR t.state = ' || TSqueued || ') ';
- vQuery := vQuery || ' AND t.starttime >= now() AND t.starttime < now()+interval ' || chr(39) || $2 || ' minutes' || chr(39);
+ vWhere := ' AND (t.state = ' || TSscheduled || ' OR t.state = ' || TSqueued || ') ';
+ vWhere := vWhere || ' AND t.starttime >= now() AND t.starttime < now()+interval ' || chr(39) || $2 || ' minutes' || chr(39);
ELSE
IF $1 = 2 THEN
- vQuery := ' AND t.state > ' || TSscheduled || ' AND t.state < ' || TScompleting;
- vQuery := vQuery || ' AND t.stoptime>now()-interval ' || chr(39) || $2 || ' minutes' || chr(39);
+ vWhere := ' AND t.state > ' || TSscheduled || ' AND t.state < ' || TScompleting;
+ vWhere := vWhere || ' AND t.stoptime>now()-interval ' || chr(39) || $2 || ' minutes' || chr(39);
ELSE
IF $1 = 3 THEN
- vQuery := ' AND t.state >= ' || TScompleting;
- vQuery := vQuery || ' AND t.stoptime > now()-interval ' || chr(39) || $2 || ' minutes' || chr(39);
+ vWhere := ' AND t.state >= ' || TScompleting;
+ vWhere := vWhere || ' AND t.stoptime > now()-interval ' || chr(39) || $2 || ' minutes' || chr(39);
vSortOrder := 't.stoptime, t.treeID';
ELSE
IF $1 = 4 THEN
- vQuery := ' AND t.state >= ' || TSapproved;
- vQuery := vQuery || ' AND t.stoptime < now() ';
+ vWhere := ' AND t.state >= ' || TSapproved;
+ vWhere := vWhere || ' AND t.stoptime < now() ';
vSortOrder := 't.treeID';
ELSE
RAISE EXCEPTION 'groupType must be 0,1,2,3 or 4 not %', $1;
@@ -86,33 +96,51 @@ CREATE OR REPLACE FUNCTION getTreeGroup(INT, INT)
END IF;
END IF;
END IF;
+ IF $3 != '' THEN
+ vExcept := '
+ SELECT x.* FROM VICtrees x INNER JOIN VIChierarchy h USING(treeid)
+ WHERE x.processtype = \'Pipeline\'
+ AND h.name = \'LOFAR.ObsSW.Observation.Cluster.ProcessingCluster.clusterName\'
+ AND h.value='|| chr(39);
+
+ IF LEFT($3,1) = '!' THEN
+ vCluster := substring($3 from 2);
+ vExcept := ' EXCEPT ' || vExcept || vCluster || chr(39);
+ ELSE
+ vCluster := $3;
+ vQuery := vExcept || vCluster || chr(39);
+ vExcept := '';
+ END IF;
+ END IF;
-- do selection
FOR vRecord IN EXECUTE '
- SELECT t.treeID,
- t.momID,
- t.groupID,
- t.classif,
- u.username,
- t.d_creation,
- t.modificationdate,
- t.treetype,
- t.state,
- t.originID,
- c.name,
- t.starttime,
- t.stoptime,
- t.processType,
- t.processSubtype,
- t.strategy,
- t.description
- FROM OTDBtree t
- INNER JOIN OTDBuser u ON t.creator = u.userid
- INNER JOIN campaign c ON c.ID = t.campaign
- WHERE t.treetype = 30
- AND t.classif = 3
- ' || vQuery || '
- ORDER BY ' || vSortOrder
+ WITH VICtrees AS (
+ SELECT t.treeID,
+ t.momID,
+ t.groupID,
+ t.classif,
+ u.username,
+ t.d_creation,
+ t.modificationdate,
+ t.treetype,
+ t.state,
+ t.originID,
+ c.name,
+ t.starttime,
+ t.stoptime,
+ t.processType,
+ t.processSubtype,
+ t.strategy,
+ t.description
+ FROM OTDBtree t
+ INNER JOIN OTDBuser u ON t.creator = u.userid
+ INNER JOIN campaign c ON c.ID = t.campaign
+ WHERE t.treetype = 30
+ AND t.classif = 3
+ ' || vWhere || '
+ ORDER BY ' || vSortOrder || ')
+ ' || vQuery || vExcept
LOOP
RETURN NEXT vRecord;
END LOOP;
diff --git a/SAS/OTDB/sql/upgradeOTDB.sql b/SAS/OTDB/sql/upgradeOTDB.sql
index abe89337101bd290b54995a2933fb0a143308575..74b24c91e0fcb9893b22c18390add0af949b6e62 100644
--- a/SAS/OTDB/sql/upgradeOTDB.sql
+++ b/SAS/OTDB/sql/upgradeOTDB.sql
@@ -1,40 +1,17 @@
-- add new columns to the tree metadata table
-ALTER TABLE OTDBtree
- ADD COLUMN modificationDate timestamp(0) DEFAULT now();
+ALTER TABLE statehistory
+ ADD COLUMN creation timestamp(6) DEFAULT now();
--- Change treeInfo structure by adding 5 fields
-DROP TYPE IF EXISTS treeInfo CASCADE;
-CREATE TYPE treeInfo AS (
- treeID INT4, -- OTDBtree.treeID%TYPE,
- momID INT4,
- groupID INT4,
- classification INT2, -- classification.ID%TYPE,
- creator VARCHAR(20), -- OTDBuser.username%TYPE,
- creationDate timestamp(0),
- modificationDate timestamp(0),
- type INT2, -- treetype.ID%TYPE,
- state INT2, -- treestate.ID%TYPE,
- originalTree INT4, -- OTDBtree.treeID%TYPE,
- campaign VARCHAR(30), -- campaign.name%TYPE,
- starttime timestamp(0),
- stoptime timestamp(0),
- processType VARCHAR(20),
- processSubtype VARCHAR(50),
- strategy VARCHAR(30),
- description TEXT
-);
+CREATE INDEX statehist_creation_idx ON statehistory(creation);
--- Reload the functions that where dropped.
-\i getBrokenHardware_func.sql
-\i getDefaultTemplates_func.sql
-
--- Reload modified functions
-\i getTreeGroup_func.sql
-\i getTreesInPeriod_func.sql
-\i getTreeList_func.sql
-\i getTreeInfo_func.sql
-\i getExecutableTrees_func.sql
+CREATE TABLE otdb_admin (
+ treestatusevent timestamp(6)
+) WITHOUT OIDS;
+INSERT INTO otdb_admin VALUES(now());
-- Load new functions
-\i create_rules.sql
-\i getModifiedTrees_func.sql
+\i getStateChanges.sql
+
+-- Load modified/improved functions
+\i exportTree_func.sql
+\i getTreeGroup_func.sql
diff --git a/SAS/OTDB/src/OTDBconnection.cc b/SAS/OTDB/src/OTDBconnection.cc
index 0c11207bed196f6f4b472079c11f03515735dc1d..7be8ec340074a8ef5758168cf7cf4badc0af954f 100644
--- a/SAS/OTDB/src/OTDBconnection.cc
+++ b/SAS/OTDB/src/OTDBconnection.cc
@@ -370,26 +370,27 @@ vector<OTDBtree> OTDBconnection::getExecutableTrees(classifType aClassification)
}
//
-// getTreeGroup(groupType, periodInMinutes)
+// getTreeGroup(groupType, periodInMinutes, cluster)
//
-// 1 = planned, 2 = active, 3 = finished
+// 0 = to be scheduled, 1 = planned, 2 = active, 3 = finished, 4 = APPROVED trees
//
// Note: this function will probably make getExecutableTrees obsolete.
//
-vector<OTDBtree> OTDBconnection::getTreeGroup(uint32 groupType, uint32 period)
+vector<OTDBtree> OTDBconnection::getTreeGroup(uint32 groupType, uint32 period, const string& cluster)
{
if (!itsIsConnected && !connect()) {
vector<OTDBtree> empty;
return (empty);
}
- LOG_TRACE_FLOW_STR ("OTDB:getTreeGroup(" << groupType << "," << period << ")");
+ LOG_TRACE_FLOW_STR ("OTDB:getTreeGroup(" << groupType << "," << period << "," << cluster << ")");
try {
// construct a query that calls a stored procedure.
work xAction(*itsConnection, "getTreeGroup");
string query("SELECT * from getTreeGroup('" +
toString(groupType) + "','" +
- toString(period) + "')");
+ toString(period) + "','" +
+ cluster + "')");
// execute query
result res = xAction.exec(query);
diff --git a/SAS/OTDB/test/CMakeLists.txt b/SAS/OTDB/test/CMakeLists.txt
index 688717095bdcbbba22f0a397d439d848546bedaf..b308bfd5c8e431685b78cde28c0859fe81a299b8 100644
--- a/SAS/OTDB/test/CMakeLists.txt
+++ b/SAS/OTDB/test/CMakeLists.txt
@@ -2,6 +2,7 @@
include(LofarCTest)
+lofar_add_test(t_getTreeGroup)
lofar_add_test(tCampaign tCampaign.cc)
lofar_add_test(tPICtree tPICtree.cc)
lofar_add_test(tPICvalue tPICvalue.cc)
diff --git a/SAS/OTDB/test/t_getTreeGroup.py b/SAS/OTDB/test/t_getTreeGroup.py
new file mode 100644
index 0000000000000000000000000000000000000000..ef62ab9edd86a5da75a72bba05ea640ccb241551
--- /dev/null
+++ b/SAS/OTDB/test/t_getTreeGroup.py
@@ -0,0 +1,93 @@
+#!/usr/bin/env python
+#coding: iso-8859-15
+#
+# 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: Backtrace.cc 31468 2015-04-13 23:26:52Z amesfoort $
+"""
+Test the modified getTreeGroup stored procedure in the database.
+The tests assume a 'unittest_db' that contains the following trees:
+
+treeid treetype LOFAR.ObsSW.Observation.Cluster.ProcessingCluster.clusterName
+-----------------------------------------------------------------------------------
+1099268 Observation CEP2
+1099270 Observation CEP4
+1099271 Observation <empty>
+1099272 Observation <no such field>
+1099266 Pipeline CEP2
+1099273 Pipeline CEP4
+1099274 Pipeline <empty>
+1099275 Pipeline <no such field>
+
+When cluster-argument is empty all trees are returned (=backwards compatability)
+When cluster argument is e.g. 'CEP2' all CEP2 PIPELINES are returned.
+When cluster argument is e.g. '!CEP2' all trees EXCEPT the CEP2 PIPELINES are returned
+"""
+
+import sys, pg
+import logging
+
+logging.basicConfig(stream=sys.stdout, level=logging.WARNING)
+logger = logging.getLogger(__name__)
+
+dbcontent = {
+ 1099268: "CEP2-Obs", 1099266 : "CEP2-PL",
+ 1099270: "CEP4-Obs", 1099273 : "CEP4-PL",
+ 1099271: "empty-Obs",1099274 : "empty-PL",
+ 1099272: "None-Obs", 1099275 : "None-PL" }
+
+def construct_answer(cluster):
+ """
+ Implement the same algorithm as the SQL query we call
+ """
+ if cluster == '':
+ return [ (x,) for x in dbcontent.keys() ]
+ if cluster[0] == "!":
+ return [ (key,) for (key,value) in dbcontent.iteritems() if value != "%s-PL"%cluster[1:] ]
+ else:
+ return [ (key,) for (key,value) in dbcontent.iteritems() if value == "%s-PL"%cluster ]
+
+# Execute the getTreeGroup query
+def getTreeGroup(dbconnection, grouptype, period, cluster):
+ """
+ Do the database call and sort the result.
+ """
+ # Try to get the specification information
+ return sorted(dbconnection.query("select treeid from getTreeGroup(%d, %d, '%s')" % (grouptype, period, cluster)).getresult())
+
+if __name__ == "__main__":
+ if len(sys.argv) != 4:
+ print "Syntax: %s username hostname database" % sys.argv[0]
+ sys.exit(1)
+
+ username = sys.argv[1]
+ hostname = sys.argv[2]
+ database = sys.argv[3]
+
+ otdb_connection = pg.connect(user=username, host=hostname, dbname=database)
+
+ try:
+ success = True
+ for cluster in ['', 'CEP2', '!CEP2', 'CEP4', '!CEP4' ]:
+ success = success & (construct_answer(cluster) == getTreeGroup(otdb_connection, 0, 0, cluster))
+ except Exception as e:
+ print e
+ success = False
+
+ sys.exit(not(success)) # return 0 on success.
diff --git a/SAS/OTDB/test/t_getTreeGroup.run b/SAS/OTDB/test/t_getTreeGroup.run
new file mode 100755
index 0000000000000000000000000000000000000000..957d73c952a8448bdc0ebcd22e5038a34df9343a
--- /dev/null
+++ b/SAS/OTDB/test/t_getTreeGroup.run
@@ -0,0 +1,36 @@
+#!/bin/sh -x
+# constants
+DBHOST=sas099.control.lofar
+
+# Setup a clean database with predefined content
+dropdb -U postgres -h ${DBHOST} unittest_db
+gzip -dc $srcdir/unittest_db.dump.gz | psql -U postgres -h ${DBHOST} -f -
+
+# Run the unit test
+# either with or without code coverage measurements,
+# depending wheter coverage has been installed
+
+if type "coverage" > /dev/null; then
+ #run test using python coverage tool
+
+ #erase previous results
+ coverage erase
+
+ #setup coverage config file
+ printf "[report]\nexclude_lines = \n if __name__ == .__main__.\n def main\n" > .coveragerc
+
+ coverage run --branch python t_getTreeGroup.py postgres ${DBHOST} unittest_db
+ RESULT=$?
+ if [ $RESULT -eq 0 ]; then
+ echo " *** Code coverage results *** "
+ coverage report -m
+ echo " *** End coverage results *** "
+ fi
+ exit $RESULT
+else
+ #coverage not available
+ echo "Please run: 'pip install coverage' to enable code coverage reporting of the unit tests"
+ #run plain test script
+ python t_getTreeGroup.py postgres ${DBHOST} unittest_db
+fi
+
diff --git a/SAS/OTDB/test/t_getTreeGroup.sh b/SAS/OTDB/test/t_getTreeGroup.sh
new file mode 100755
index 0000000000000000000000000000000000000000..c25b8f5bce223c02e863df8ce331099297b34a51
--- /dev/null
+++ b/SAS/OTDB/test/t_getTreeGroup.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+./runctest.sh t_getTreeGroup
diff --git a/SAS/OTDB/test/unittest_db.dump.gz b/SAS/OTDB/test/unittest_db.dump.gz
new file mode 100644
index 0000000000000000000000000000000000000000..921888c76eac230061c7b8d0ae346d0406ca55dc
Binary files /dev/null and b/SAS/OTDB/test/unittest_db.dump.gz differ
diff --git a/SAS/OTDB_Services/CMakeLists.txt b/SAS/OTDB_Services/CMakeLists.txt
new file mode 100644
index 0000000000000000000000000000000000000000..4c6ee650e826b7185f0c7b77ef3b3fe18a777ee5
--- /dev/null
+++ b/SAS/OTDB_Services/CMakeLists.txt
@@ -0,0 +1,14 @@
+# $Id$
+
+#lofar_package(OTDB_Services 1.0 DEPENDS PyMessaging OTDB)
+
+include(PythonInstall)
+
+lofar_add_bin_scripts(
+ TreeService.py
+ TreeStatusEvents.py
+)
+
+
+add_subdirectory(test)
+
diff --git a/SAS/OTDB_Services/TreeService.py b/SAS/OTDB_Services/TreeService.py
new file mode 100755
index 0000000000000000000000000000000000000000..89ea80d515f120de9b09d70122faa26d8ab26676
--- /dev/null
+++ b/SAS/OTDB_Services/TreeService.py
@@ -0,0 +1,311 @@
+#!/usr/bin/env python
+#coding: iso-8859-15
+#
+# 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: Backtrace.cc 31468 2015-04-13 23:26:52Z amesfoort $
+"""
+Daemon that sets-up a set of servicess for the OTDB database.
+
+RPC functions that allow access to (VIC) trees in OTDB.
+
+TaskSpecificationRequest: get the specification(parset) of a tree as dict.
+KeyUpdateCommand : function to update the value of multiple (existing) keys.
+StatusUpdateCommand : finction to update the status of a tree.
+"""
+
+import sys, time, pg
+import logging
+from optparse import OptionParser
+from lofar.messaging.Service import *
+
+QUERY_EXCEPTIONS = (TypeError, ValueError, MemoryError, pg.ProgrammingError, pg.InternalError)
+
+logging.basicConfig(stream=sys.stdout, level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+# Define our own exceptions
+class FunctionError(Exception):
+ "Something when wrong during the execution of the function"
+ pass
+class DatabaseError(Exception):
+ "Connection with the database could not be made"
+ pass
+
+# Task Specification Request
+def TaskSpecificationRequest(input_dict, db_connection):
+ """
+ RPC function that retrieves the task specification from a tree.
+
+ Input : OtdbID (integer) - ID of the tree to retrieve the specifications of
+ Output: (dict) - The 'parset' of the tree
+
+ Exceptions:
+ AttributeError: There is something wrong with the given input values.
+ FunctionError: An error occurred during the execution of the function.
+ The text of the exception explains what is wrong.
+ """
+ # Check the input
+ if not isinstance(input_dict, dict):
+ raise AttributeError("TaskSpecificationRequest: Expected a dict as input")
+ try:
+ tree_id = input_dict['OtdbID']
+ except KeyError, info:
+ raise AttributeError("TaskSpecificationRequest: Key %s is missing in the input" % info)
+
+ # Try to get the specification information
+ try:
+ logger.info("TaskSpecificationRequest:%s" % input_dict)
+ top_node = db_connection.query("select nodeid from getTopNode('%s')" % tree_id).getresult()[0][0]
+ treeinfo = db_connection.query("select exportTree(1, '%s', '%s')" % (tree_id, top_node)).getresult()[0][0]
+ except QUERY_EXCEPTIONS, exc_info:
+ raise FunctionError("Error while requesting specs of tree %d: %s"% (tree_id, exc_info))
+ # When the query was succesfull 'treeinfo' is now a string that contains many 'key = value' lines seperated
+ # with newlines. To make it more usable for the user we convert that into a dict...
+
+ # Note: a PIC tree is a list of keys while a Template tree and a VIC tree is a list of key-values.
+ # Since we don't know what kind of tree was requested we assume a Template/VIC tree (most likely)
+ # but if this ends in exceptions we fall back to a PIC tree.
+ answer_dict = {}
+ answer_list = []
+ for line in treeinfo.split('\n'): # make seperate lines of it.
+ try:
+ # assume a 'key = value' line
+ (key, value) = line.split("=", 1)
+ answer_dict[key] = value
+ except ValueError:
+ # oops, no '=' on the line, must be a PIC tree that was queried: make a list iso a dict
+ answer_list.append(line)
+ if len(answer_list) > 1: # there is always one empty line, ignore that one...
+ answer_dict["tree"] = answer_list
+ return answer_dict
+
+# Status Update Command
+def StatusUpdateCommand(input_dict, db_connection):
+ """
+ RPC function to update the status of a tree.
+
+ Input : OtdbID (integer) - ID of the tree to change the status of.
+ NewStatus (string) - The new status of the tree. The following values are allowed:
+ described, prepared, approved, on_hold, conflict, prescheduled, scheduled, queued,
+ active, completing, finished, aborted, error, obsolete
+ UpdateTimestamps (boolean) - Optional parameter to also update the timestamp of the metadata of the
+ tree when the status of the tree is changed into 'active', 'finished' or 'aborted'. Resp. starttime
+ or endtime. Default this option is ON.
+ Output: (boolean) - Reflects the successful update of the status.
+
+ Exceptions:
+ AttributeError: There is something wrong with the given input values.
+ FunctionError: An error occurred during the execution of the function.
+ The text of the exception explains what is wrong.
+ """
+ # Check input
+ if not isinstance(input_dict, dict):
+ raise AttributeError("StatusUpdateCommand: Expected a dict as input")
+ try:
+ tree_id = input_dict['OtdbID']
+ new_status = input_dict['NewStatus']
+ update_times = True
+ if input_dict.has_key("UpdateTimestamps"):
+ update_times = bool(input_dict["UpdateTimestamps"])
+ logger.info("StatusUpdateCommand(%s,%s,%s)" % (tree_id, new_status, update_times))
+ except KeyError, info:
+ raise AttributeError("StatusUpdateCommand: Key %s is missing in the input" % info)
+
+ # Get list of allowed tree states
+ allowed_states = {}
+ try:
+ for (state_nr, name) in db_connection.query("select id,name from treestate").getresult():
+ allowed_states[name] = state_nr
+ except QUERY_EXCEPTIONS, exc_info:
+ raise FunctionError("Error while getting allowed states of tree %d: %s" % (tree_id, exc_info))
+
+ # Check value of new_status argument
+ if not new_status in allowed_states:
+ raise FunctionError("The newstatus(=%s) for tree %d must have one of the following values:%s" %
+ (new_status, tree_id, allowed_states.keys()))
+
+ # Finally try to change the status
+ try:
+ success = (db_connection.query("select setTreeState(1, %d, %d::INT2,%s)" %
+ (tree_id, allowed_states[new_status], str(update_times))).getresult()[0][0] == 't')
+ except QUERY_EXCEPTIONS, exc_info:
+ raise FunctionError("Error while setting the status of tree %d: %s" % (tree_id, exc_info))
+ return str(success)
+
+
+# Key Update Command
+def KeyUpdateCommand(input_dict, db_connection):
+ """
+ RPC function to update the values of a tree.
+
+ Input : OtdbID (integer) - ID of the tree to change the status of.
+ Updates (dict) - The key-value pairs that must be updated.
+ Output: (dict)
+ 'Errors' (dict) Refects the problems that occured {'key':'problem'}
+ Field is empty if all fields could be updated.
+ Exceptions:
+ AttributeError: There is something wrong with the given input values.
+ FunctionError: An error occurred during the execution of the function.
+ The text of the exception explains what is wrong.
+ """
+ # Check input
+ if not isinstance(input_dict, dict):
+ raise AttributeError("Expected a dict as input")
+ try:
+ tree_id = input_dict['OtdbID']
+ update_list = input_dict['Updates']
+ except KeyError, info:
+ raise AttributeError("KeyUpdateCommand: Key %s is missing in the input" % info)
+ if not isinstance(tree_id, int):
+ raise AttributeError("KeyUpdateCommand (tree=%d): Field 'OtdbID' must be of type 'integer'" % tree_id)
+ if not isinstance(update_list, dict):
+ raise AttributeError("KeyUpdateCommand (tree=%d): Field 'Updates' must be of type 'dict'" % tree_id)
+ logger.info("KeyUpdateCommand for tree: %d", tree_id)
+
+ # Finally try to update all keys
+ errors = {}
+ for (key, value) in update_list.iteritems():
+ try:
+ record_list = (db_connection.query("select nodeid,instances,limits from getvhitemlist (%d, '%s')" %
+ (tree_id, key))).getresult()
+ if len(record_list) == 0:
+ errors[key] = "Not found for tree %d" % tree_id
+ continue
+ if len(record_list) > 1:
+ errors[key] = "Not a unique key, found %d occurrences for tree %d" % (len(record_list), tree_id)
+ continue
+ # When one record was found record_list is a list with a single tuple (nodeid, instances, current_value)
+ node_id = record_list[0][0]
+ instances = record_list[0][1]
+ db_connection.query("select updateVTnode(1,%d,%d,%d::INT2,'%s')" % (tree_id, node_id, instances, value))
+ print "%s: %s ==> %s" % (key, record_list[0][2], value)
+ except QUERY_EXCEPTIONS, exc:
+ errors[key] = str(exc)
+ if len(errors):
+ raise FunctionError(("Not all key were updated:", errors))
+ return errors
+
+class PostgressMessageHandlerInterface(MessageHandlerInterface):
+ """
+ Implements a generic message handlers for services that are tied to a postgres database.
+ kwargs must contain the keys:
+ database <string> Name of the database to connect to
+ db_user <string> Name of the user used for logging into the database
+ db_host <string> Name of the machine the database server is running
+ function <type> Function to call when a message is received on the message bus.
+ """
+ def __init__(self, **kwargs):
+ super(PostgressMessageHandlerInterface, self).__init__()
+ self.database = kwargs.pop("database")
+ self.db_user = kwargs.pop("db_user", "postgres")
+ self.db_host = kwargs.pop("db_host", "localhost")
+ if len(kwargs):
+ raise AttributeError("Unknown keys in arguments of 'DatabaseTiedMessageHandler: %s" % kwargs)
+ self.connection = None
+ self.connected = False
+
+ def prepare_receive(self):
+ "Called in main processing loop just before a blocking wait for messages is done."
+ "Make sure we are connected with the database."
+ self.connected = (self.connection and self.connection.status == 1)
+ while not self.connected:
+ try:
+ self.connection = pg.connect(user=self.db_user, host=self.db_host, dbname=self.database)
+ self.connected = True
+ logger.info("Connected to database %s on host %s" % (self.database, self.db_host))
+ except (TypeError, SyntaxError, pg.InternalError):
+ self.connected = False
+ logger.error("Not connected to database %s on host %s (anymore), retry in 5 seconds"
+ % (self.database, self.db_host))
+ time.sleep(5)
+
+class PostgressTaskSpecificationRequest(PostgressMessageHandlerInterface):
+ """
+ Embedding of the TaskSpecificationRequest function in the postgress service class.
+ """
+ def __init__(self, **kwargs):
+ super(PostgressTaskSpecificationRequest, self).__init__(**kwargs)
+
+ def handle_message(self, msg):
+ " Connect to the right function"
+ return TaskSpecificationRequest(msg, self.connection)
+
+
+class PostgressStatusUpdateCommand(PostgressMessageHandlerInterface):
+ """
+ Embedding of the TaskSpecificationRequest function in the postgress service class.
+ """
+ def __init__(self, **kwargs):
+ super(PostgressStatusUpdateCommand, self).__init__(**kwargs)
+
+ def handle_message(self, msg):
+ " Connect to the right function"
+ return StatusUpdateCommand(msg, self.connection)
+
+
+class PostgressKeyUpdateCommand(PostgressMessageHandlerInterface):
+ """
+ Embedding of the TaskSpecificationRequest function in the postgress service class.
+ """
+ def __init__(self, **kwargs):
+ super(PostgressKeyUpdateCommand, self).__init__(**kwargs)
+
+ def handle_message(self, msg):
+ " Connect to the right function"
+ return KeyUpdateCommand(msg, self.connection)
+
+
+if __name__ == "__main__":
+ # Check the invocation arguments
+ parser = OptionParser("%prog [options]")
+ parser.add_option("-D", "--database", dest="dbName", type="string", default="",
+ help="Name of the database")
+ parser.add_option("-H", "--hostname", dest="dbHost", type="string", default="sasdb",
+ help="Hostname of database server")
+ (options, args) = parser.parse_args()
+
+ if not options.dbName:
+ print "Missing database name"
+ parser.print_help()
+ sys.exit(0)
+
+ if not options.dbHost:
+ print "Missing database server name"
+ parser.print_help()
+ sys.exit(0)
+
+ busname = sys.argv[1] if len(sys.argv) > 1 else "simpletest"
+
+ serv1 = Service("TaskSpecification", PostgressTaskSpecificationRequest,
+ busname=busname, numthreads=1, startonwith=True,
+ handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
+ serv2 = Service("StatusUpdateCmd", PostgressStatusUpdateCommand,
+ busname=busname, numthreads=1, startonwith=True,
+ handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
+ serv3 = Service("KeyUpdateCmd", PostgressKeyUpdateCommand,
+ busname=busname, numthreads=1, startonwith=True,
+ handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
+
+ with serv1, serv2, serv3:
+ logger.info("Started the OTDB services")
+ serv3.wait_for_interrupt()
+
+ logger.info("Stopped the OTDB services")
+
diff --git a/SAS/OTDB_Services/TreeStatusEvents.py b/SAS/OTDB_Services/TreeStatusEvents.py
new file mode 100755
index 0000000000000000000000000000000000000000..a9264f7b5780df6148c16cfcf22a09626a3416a3
--- /dev/null
+++ b/SAS/OTDB_Services/TreeStatusEvents.py
@@ -0,0 +1,153 @@
+#!/usr/bin/env python
+#coding: iso-8859-15
+#
+# 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: Backtrace.cc 31468 2015-04-13 23:26:52Z amesfoort $
+"""
+Daemon that watches the OTDB database for status changes of trees and publishes those on the messagebus.
+"""
+
+import os, sys, time, pg, signal
+from optparse import OptionParser
+from lofar.messaging import EventMessage, ToBus
+
+QUERY_EXCEPTIONS = (TypeError, ValueError, MemoryError, pg.ProgrammingError, pg.InternalError)
+alive = False
+
+# Define our own exceptions
+class FunctionError(Exception):
+ "Something when wrong during the execution of the function"
+ pass
+class DatabaseError(Exception):
+ "Connection with the database could not be made"
+ pass
+
+# Task Specification Request
+def PollForStatusChanges(start_time, end_time, otdb_connection):
+ """
+ Function that asked the database for status changes in the given period
+
+ Input : start_time (string) - Oldest time of change to include in the selection.
+ end_time (string) - Most recent time of change to include in the selection
+ The times must be specified in the format YYYY-Mon-DD HH24:MI:SS.US.
+ The selection delivers changes the match: startime <= time_of_change < end_time
+
+ Output: (list of tuples) - All status changes between the last polltime and the current time
+ Tuple = ( tree_id, new_state, time_of_change )
+
+ Exceptions:
+ ArgumentError: There is something wrong with the given input values.
+ FunctionError: An error occurred during the execution of the function.
+ The text of the exception explains what is wrong.
+ """
+ # Try to get the specification information
+ record_list = []
+ try:
+ record_list = otdb_connection.query("select treeid,state,modtime,creation from getStateChanges('%s','%s')" %
+ (start_time, end_time)).getresult()
+ except QUERY_EXCEPTIONS, exc_info:
+ raise FunctionError("Error while polling for state changes: %s"% exc_info)
+ return record_list
+
+def signal_handler(signum, frame):
+ "Signal redirection to stop the daemon in a neat way."
+ print "Stopping program"
+ global alive
+ alive = False
+
+
+if __name__ == "__main__":
+ # Check the invocation arguments
+ parser = OptionParser("%prog [options]")
+ parser.add_option("-D", "--database", dest="dbName", type="string", default="",
+ help="Name of the database")
+ parser.add_option("-H", "--hostname", dest="dbHost", type="string", default="sasdb",
+ help="Hostname of database server")
+ parser.add_option("-B", "--busname", dest="busname", type="string", default="",
+ help="Busname or queue-name the status changes are published on")
+ (options, args) = parser.parse_args()
+
+ if not options.dbName:
+ print "Missing database name"
+ parser.print_help()
+ sys.exit(0)
+
+ if not options.dbHost:
+ print "Missing database server name"
+ parser.print_help()
+ sys.exit(0)
+
+ if not options.busname:
+ print "Missing busname"
+ parser.print_help()
+ sys.exit(0)
+
+ # Set signalhandler to stop the program in a neat way.
+ signal.signal(signal.SIGINT, signal_handler)
+
+ alive = True
+ connected = False
+ otdb_connection = None
+ with ToBus(options.busname) as send_bus:
+ while alive:
+ while alive and not connected:
+ # Connect to the database
+ try:
+ otdb_connection = pg.connect(user="postgres", host=options.dbHost, dbname=options.dbName)
+ connected = True
+ # Get list of allowed tree states
+ allowed_states = {}
+ for (state_nr, name) in otdb_connection.query("select id,name from treestate").getresult():
+ allowed_states[state_nr] = name
+ except (TypeError, SyntaxError, pg.InternalError):
+ connected = False
+ print "DatabaseError: Connection to database could not be made, reconnect attempt in 5 seconds"
+ time.sleep(5)
+
+ # When we are connected we can poll the database
+ if connected:
+ # Get start_time (= creation time of last retrieved record if any)
+ start_time = ''
+ try:
+ start_time = otdb_connection.query("select treestatusevent from otdb_admin").getresult()[0][0]
+ except IndexError, QUERY_EXCEPTIONS:
+ start_time = "2015-01-01 00:00:00.00"
+ print "start_time=", start_time
+
+ try:
+ record_list = PollForStatusChanges(start_time, "now", otdb_connection)
+ except FunctionError, exc_info:
+ print exc_info
+ else:
+ for (treeid, state, modtime, creation) in record_list:
+ content = { "treeID" : treeid, "state" : allowed_states.get(state, "unknwon_state"),
+ "time_of_change" : modtime }
+ msg = EventMessage(context="otdb.treestatus", content=content)
+ print treeid, allowed_states.get(state, "unknwon_state"), modtime, creation
+ send_bus.send(msg)
+ otdb_connection.query("update otdb_admin set treestatusevent = '%s'" % start_time)
+ start_time = creation
+ print "==="
+
+ # Redetermine the database status.
+ connected = (otdb_connection and otdb_connection.status == 1)
+
+ time.sleep(2)
+
diff --git a/SAS/OTDB_Services/test/CMakeLists.txt b/SAS/OTDB_Services/test/CMakeLists.txt
new file mode 100644
index 0000000000000000000000000000000000000000..f5796145f0c26c0dfdf4d61b0b7efa3ab4f7f524
--- /dev/null
+++ b/SAS/OTDB_Services/test/CMakeLists.txt
@@ -0,0 +1,7 @@
+# $Id: CMakeLists.txt 1576 2015-09-29 15:22:28Z loose $
+
+include(LofarCTest)
+
+lofar_add_test(t_TreeService)
+lofar_add_test(t_TreeStatusEvents)
+
diff --git a/SAS/OTDB_Services/test/t_TreeService.py b/SAS/OTDB_Services/test/t_TreeService.py
new file mode 100644
index 0000000000000000000000000000000000000000..ff208e810339388539f6d57149c83fbd55e330bb
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeService.py
@@ -0,0 +1,113 @@
+#!/usr/bin/env python
+#coding: iso-8859-15
+#
+# 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: Backtrace.cc 31468 2015-04-13 23:26:52Z amesfoort $
+"""
+RPC functions that allow access to (VIC) trees in OTDB.
+
+TaskSpecificationRequest: get the specification(parset) of a tree as dict.
+KeyUpdateCommand : function to update the value of multiple (existing) keys.
+StatusUpdateCommand : finction to update the status of a tree.
+"""
+
+import sys
+import logging
+from lofar.messaging.RPC import *
+
+logging.basicConfig(stream=sys.stdout, level=logging.WARNING)
+logger = logging.getLogger(__name__)
+
+def do_rpc(rpc_instance, arg_dict):
+# try:
+ (data, status) = (rpc_instance)(arg_dict)
+ if status != "OK":
+ raise Exception("Status returned is %s" % status)
+ for key in sorted(data):
+ print "%s ==> %s" % (key, data[key])
+# except OverflowError as e:
+# pass
+ print "======"
+
+if __name__ == "__main__":
+ busname = sys.argv[1] if len(sys.argv) > 1 else "simpletest"
+
+ with RPC("TaskSpecification", ForwardExceptions=True, busname=busname, timeout=10) as task_spec_request:
+ do_rpc(task_spec_request, {'OtdbID':1099269}) # PIC
+ do_rpc(task_spec_request, {'OtdbID':1099238}) # Template
+ do_rpc(task_spec_request, {'OtdbID':1099266}) # VIC
+
+ with RPC("StatusUpdateCmd", ForwardExceptions=True, busname=busname, timeout=5) as status_update_command:
+ # PIC
+ (data, status) = status_update_command({'OtdbID':1099269, 'NewStatus':'finished', 'UpdateTimestamps':True})
+ print status, data
+ # Template
+ (data, status) = status_update_command({'OtdbID':1099238, 'NewStatus':'finished', 'UpdateTimestamps':True})
+ print status, data
+ # VIC
+ (data, status) = status_update_command({'OtdbID':1099266, 'NewStatus':'finished', 'UpdateTimestamps':True})
+ print status, data
+
+ # Nonexisting tree
+ try:
+ (data, status) = status_update_command({'OtdbID':10, 'NewStatus':'finished', 'UpdateTimestamps':True})
+ except RPCException as e:
+ print "Caught expected exception on invalid treeID in status update"
+
+ # VIC tree: invalid status
+ try:
+ (data, status) = status_update_command({'OtdbID':1099266, 'NewStatus':'what_happend', 'UpdateTimestamps':True})
+ except RPCException as e:
+ print "Caught expected exception on invalid status in status update"
+
+
+ with RPC("KeyUpdateCmd", ForwardExceptions=True, busname=busname, timeout=5) as key_update:
+ # VIC tree: valid
+ (data, status) = key_update({'OtdbID':1099266,
+ 'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.PythonControl.pythonHost':'NameOfTestHost'}})
+ print status, data
+
+ # Template tree: not supported yet
+ try:
+ (data, status) = key_update({'OtdbID':1099238,
+ 'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.PythonControl.pythonHost':'NameOfTestHost'}})
+ except RPCException as e:
+ print "Caught expected exception on invalid treetype in key update"
+
+ # PIC tree: not supported yet
+ try:
+ (data, status) = key_update({'OtdbID':1099269, 'Updates':{'LOFAR.PIC.Core.CS001.status_state':'50'}})
+ except RPCException as e:
+ print "Caught expected exception on invalid treetype (PIC) in key update"
+
+ # Non exsisting tree
+ try:
+ (data, status) = key_update({'OtdbID':10,
+ 'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.PythonControl.pythonHost':'NameOfTestHost'}})
+ except RPCException as e:
+ print "Caught expected exception on invalid treeID in key update"
+
+ # VIC tree: wrong key
+ try:
+ (data, status) = key_update({'OtdbID':1099266,
+ 'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.PythonControl.NoSuchKey':'NameOfTestHost'}})
+ except RPCException as e:
+ print "Caught expected exception on invalid key in key update"
+
diff --git a/SAS/OTDB_Services/test/t_TreeService.run b/SAS/OTDB_Services/test/t_TreeService.run
new file mode 100755
index 0000000000000000000000000000000000000000..14cc7a3968eb5a1f300a595e24721b6109db21a5
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeService.run
@@ -0,0 +1,49 @@
+#!/bin/sh -x
+# constants
+DBHOST=sas099.control.lofar
+
+#cleanup on normal exit and on SIGHUP, SIGINT, SIGQUIT, and SIGTERM
+trap 'qpid-config del exchange --force $queue ; kill ${SERVER_PID}' 0 1 2 3 15
+
+# Generate randome queue name
+queue=$(< /dev/urandom tr -dc [:alnum:] | head -c16)
+
+# Create the queue
+qpid-config add exchange topic $queue
+
+# Setup a clean database with predefined content
+dropdb -U postgres -h ${DBHOST} unittest_db
+gzip -dc $srcdir/unittest_db.dump.gz | psql -U postgres -h ${DBHOST} -f -
+TreeService.py $queue -D unittest_db -H ${DBHOST} &
+SERVER_PID=$!
+# Starting up takes a while
+sleep 3
+
+# Run the unit test
+# either with or without code coverage measurements,
+# depending wheter coverage has been installed
+
+if type "coverage" > /dev/null; then
+ #run test using python coverage tool
+
+ #erase previous results
+ coverage erase
+
+ #setup coverage config file
+ printf "[report]\nexclude_lines = \n if __name__ == .__main__.\n def main\n" > .coveragerc
+
+ coverage run --branch --include=*Messaging/python* t_TreeService.py $queue
+ RESULT=$?
+ if [ $RESULT -eq 0 ]; then
+ echo " *** Code coverage results *** "
+ coverage report -m
+ echo " *** End coverage results *** "
+ fi
+ exit $RESULT
+else
+ #coverage not available
+ echo "Please run: 'pip install coverage' to enable code coverage reporting of the unit tests"
+ #run plain test script
+ python t_TreeService.py $queue
+fi
+
diff --git a/SAS/OTDB_Services/test/t_TreeService.sh b/SAS/OTDB_Services/test/t_TreeService.sh
new file mode 100755
index 0000000000000000000000000000000000000000..03631dea975fba47babe769b5643604c26dfc04f
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeService.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+./runctest.sh t_TreeService
diff --git a/SAS/OTDB_Services/test/t_TreeStatusEvents.py b/SAS/OTDB_Services/test/t_TreeStatusEvents.py
new file mode 100644
index 0000000000000000000000000000000000000000..89ba25f40103594daae937d66f800475546ce6c3
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeStatusEvents.py
@@ -0,0 +1,91 @@
+#!/usr/bin/env python
+#coding: iso-8859-15
+#
+# 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: Backtrace.cc 31468 2015-04-13 23:26:52Z amesfoort $
+"""
+RPC functions that allow access to (VIC) trees in OTDB.
+
+TaskSpecificationRequest: get the specification(parset) of a tree as dict.
+KeyUpdateCommand : function to update the value of multiple (existing) keys.
+StatusUpdateCommand : finction to update the status of a tree.
+"""
+
+import sys, pg
+import logging
+from optparse import OptionParser
+from lofar.messaging import FromBus
+
+logging.basicConfig(stream=sys.stdout, level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+if __name__ == "__main__":
+ # Check the invocation arguments
+ parser = OptionParser("%prog [options]")
+ parser.add_option("-D", "--database", dest="dbName", type="string", default="",
+ help="Name of the database")
+ parser.add_option("-H", "--hostname", dest="dbHost", type="string", default="sasdb",
+ help="Hostname of database server")
+ parser.add_option("-B", "--busname", dest="busname", type="string", default="",
+ help="Busname or queue-name the status changes are published on")
+ (options, args) = parser.parse_args()
+
+ if not options.dbName:
+ print "Missing database name"
+ parser.print_help()
+ sys.exit(1)
+
+ if not options.dbHost:
+ print "Missing database server name"
+ parser.print_help()
+ sys.exit(1)
+
+ if not options.busname:
+ print "Missing busname"
+ parser.print_help()
+ sys.exit(1)
+
+ try:
+ print "user=postgres, host=", options.dbHost, "dbname=", options.dbName
+ otdb_connection = pg.connect(user="postgres", host=options.dbHost, dbname=options.dbName)
+ except (TypeError, SyntaxError, pg.InternalError):
+ print "DatabaseError: Connection to database could not be made"
+ sys.exit(77)
+
+ with FromBus(options.busname) as frombus:
+ # First drain the queue
+ no_exception = True
+ while no_exception:
+ try:
+ msg = frombus.receive(timeout=1)
+ frombus.ack(msg)
+ except Exception:
+ no_exception = False
+
+ otdb_connection.query("select setTreeState(1, %d, %d::INT2,'%s')" % (1099266, 500, False))
+ msg = frombus.receive(timeout=5) # TreeStateEVent are send every 2 seconds
+ frombus.ack(msg)
+ msg.show()
+ try:
+ ok = (msg.content['treeID'] == 1099266 and msg.content['state'] == 'queued')
+ except IndexError:
+ ok = False
+
+ sys.exit(not ok) # 0 = success
diff --git a/SAS/OTDB_Services/test/t_TreeStatusEvents.run b/SAS/OTDB_Services/test/t_TreeStatusEvents.run
new file mode 100755
index 0000000000000000000000000000000000000000..22148b99324eeabc1c49ec8f3b56edb082a8df64
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeStatusEvents.run
@@ -0,0 +1,49 @@
+#!/bin/sh -x
+# constants
+DBHOST=sas099.control.lofar
+
+#cleanup on normal exit and on SIGHUP, SIGINT, SIGQUIT, and SIGTERM
+trap 'qpid-config del exchange --force $queue ; kill ${SERVICE_PID}' 0 1 2 3 15
+
+# Generate randome queue name
+queue=$(< /dev/urandom tr -dc [:alnum:] | head -c16)
+
+# Create the queue
+qpid-config add exchange topic $queue
+
+# Setup a clean database with predefined content
+dropdb -U postgres -h ${DBHOST} unittest_db
+gzip -dc $srcdir/unittest_db.dump.gz | psql -U postgres -h ${DBHOST} -f -
+TreeStatusEvents.py -B $queue -D unittest_db -H ${DBHOST} &
+SERVICE_PID=$!
+# Starting up takes a while
+sleep 3
+
+# Run the unit test
+# either with or without code coverage measurements,
+# depending wheter coverage has been installed
+
+if type "coverage" > /dev/null; then
+ #run test using python coverage tool
+
+ #erase previous results
+ coverage erase
+
+ #setup coverage config file
+ printf "[report]\nexclude_lines = \n if __name__ == .__main__.\n def main\n" > .coveragerc
+
+ coverage run --branch --include=*Messaging/python* t_TreeStatusEvents.py -D unittest_db -H ${DBHOST} -B $queue
+ RESULT=$?
+ if [ $RESULT -eq 0 ]; then
+ echo " *** Code coverage results *** "
+ coverage report -m
+ echo " *** End coverage results *** "
+ fi
+ exit $RESULT
+else
+ #coverage not available
+ echo "Please run: 'pip install coverage' to enable code coverage reporting of the unit tests"
+ #run plain test script
+ python t_TreeStatusEvents.py -D unittest_db -H ${DBHOST} -B $queue
+fi
+
diff --git a/SAS/OTDB_Services/test/t_TreeStatusEvents.sh b/SAS/OTDB_Services/test/t_TreeStatusEvents.sh
new file mode 100755
index 0000000000000000000000000000000000000000..63d2326f4d9a556cfd978b626fa5429f6a8bd597
--- /dev/null
+++ b/SAS/OTDB_Services/test/t_TreeStatusEvents.sh
@@ -0,0 +1,2 @@
+#!/bin/sh
+./runctest.sh t_TreeStatusEvents
diff --git a/SAS/OTDB_Services/test/unittest_db.dump.gz b/SAS/OTDB_Services/test/unittest_db.dump.gz
new file mode 100644
index 0000000000000000000000000000000000000000..2fcaba7c9fcdafb08643bf5256bb8fbe1b012a9c
Binary files /dev/null and b/SAS/OTDB_Services/test/unittest_db.dump.gz differ