Task #8531: TreeService should work now. It's a lot of effort to make a decent testdatabase. :-(

TreeStatusEvent also runs and produces message. Here also the remaining work is in the test setup.

.gitattributes

@@ -4821,6 +4821,7 @@ SAS/OTDB/test/tMetadata.cc -text
 SAS/OTDB/test/tQueryPIC.cc -text
 SAS/OTDB_Services/TreeService.py -text
 SAS/OTDB_Services/TreeStatusEvents.py -text
+SAS/OTDB_Services/test/t_TreeService.py -text
 SAS/Scheduler/src/.default_settings.set -text
 SAS/Scheduler/src/LOFAR_libScheduler.pro -text
 SAS/Scheduler/src/conflictdialog.ui -text

SAS/OTDB_Services/TreeService.py

@@ -70,7 +70,7 @@ def TaskSpecificationRequest(input_dict, db_connection):
 
     # Try to get the specification information
     try:
-        logger.debug("TaskSpecificationRequest:%s" % input_dict)
+        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:
@@ -86,7 +86,7 @@ def TaskSpecificationRequest(input_dict, db_connection):
     return answer
 
 # Status Update Command
-def StatusUpdateCommand(input_dict):
+def StatusUpdateCommand(input_dict, db_connection):
     """
     RPC function to update the status of a tree.
 
@@ -104,7 +104,6 @@ def StatusUpdateCommand(input_dict):
     FunctionError: An error occurred during the execution of the function. 
                    The text of the exception explains what is wrong.
     """
-    global otdb_connection
     # Check input
     if not isinstance(input_dict, dict):
         raise AttributeError("StatusUpdateCommand: Expected a dict as input")
@@ -114,13 +113,14 @@ def StatusUpdateCommand(input_dict):
         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 (id,name) in otdb_connection.query("select id,name from treestate").getresult():
+        for (id,name) in db_connection.query("select id,name from treestate").getresult():
             allowed_states[name] = id
     except QUERY_EXCEPTIONS, exc_info:
         raise FunctionError("Error while getting allowed states of tree %d: %s" % (tree_id, exc_info))
@@ -132,7 +132,7 @@ def StatusUpdateCommand(input_dict):
 
     # Finally try to change the status
     try:
-        success = (otdb_connection.query("select setTreeState(1, %d, %d::INT2,%s)" % 
+        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))
@@ -140,7 +140,7 @@ def StatusUpdateCommand(input_dict):
 
 
 # Key Update Command
-def KeyUpdateCommand(input_dict):
+def KeyUpdateCommand(input_dict, db_connection):
     """
     RPC function to update the values of a tree.
 
@@ -154,7 +154,6 @@ def KeyUpdateCommand(input_dict):
     FunctionError: An error occurred during the execution of the function. 
                    The text of the exception explains what is wrong.
     """
-    global otdb_connection
     # Check input
     if not isinstance(input_dict, dict):
         raise AttributeError("Expected a dict as input")
@@ -167,12 +166,13 @@ def KeyUpdateCommand(input_dict):
         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 = (otdb_connection.query("select nodeid,instances,limits from getvhitemlist (%d, '%s')" % 
+            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
@@ -183,7 +183,7 @@ def KeyUpdateCommand(input_dict):
             # 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]
-            result = ((otdb_connection.query("select updateVTnode(1,%d,%d,%d::INT2,'%s')" % 
+            result = ((db_connection.query("select updateVTnode(1,%d,%d,%d::INT2,'%s')" % 
                        (tree_id, node_id, instances, value))).getresult()[0][0] == 't')
             print "%s: %s ==> %s" % (key, record_list[0][2], value)
         except QUERY_EXCEPTIONS, exc:
@@ -218,11 +218,11 @@ class PostgressMessageHandlerInterface(MessageHandlerInterface):
             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" % (dbname, host))
+                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" 
-                             % (dbname, host))
+                             % (self.database, self.db_host))
                 time.sleep(5)
 
 class PostgressTaskSpecificationRequest(PostgressMessageHandlerInterface):
@@ -280,37 +280,21 @@ if __name__ == "__main__":
         parser.print_help()
         sys.exit(0)
 
-    def do_specification_rpc(function, args):
-        try:
-            (data,status) = function(args)
-            if data:
-                for (key,value) in  data.iteritems():
-                    print "%s ==> %s" % (key, value)
-            else:
-                print status
-        except Exception as e:
-            print e
-        print "################"
-
     busname = sys.argv[1] if len(sys.argv) > 1 else "simpletest"
 
-    serv1 = Service(busname, "TaskSpecification", PostgressTaskSpecificationRequest, 
-                    numthreads=1, startonwith=True,
+    serv1 = Service("TaskSpecification", PostgressTaskSpecificationRequest, 
+                    busname=busname, numthreads=1, startonwith=True,
                     handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
-    serv2 = Service(busname, "StatusUpdateCmd",   PostgressStatusUpdateCommand,
-                    numthreads=1, startonwith=True,
+    serv2 = Service("StatusUpdateCmd",   PostgressStatusUpdateCommand,
+                    busname=busname, numthreads=1, startonwith=True,
                     handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
-    serv3 = Service(busname, "KeyUpdateCmd",      PostgressKeyUpdateCommand,
-                    numthreads=1, startonwith=True,
+    serv3 = Service("KeyUpdateCmd",      PostgressKeyUpdateCommand,
+                    busname=busname, numthreads=1, startonwith=True,
                     handler_args = {"database" : options.dbName, "db_host" : options.dbHost})
 
     with serv1,serv2,serv3:
-        with RPC(busname, "TaskSpecification", ForwardExceptions=True) as task_spec_request:
-            do_specification_rpc(task_spec_request, {'OtdbID':63370})
-            do_specification_rpc(task_spec_request, {'OtdbID':82111})
-            do_specification_rpc(task_spec_request, {'OtdbID':146300})
-
-#    print StatusUpdateCommand({'OtdbID':146300, 'NewStatus':'finished', 'UpdateTimestamps':True})
-#    print KeyUpdateCommand({'OtdbID':63370, 'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.OnlineControl._hostname':'CCU099ABC'}})
-   
+        logger.info("Started the OTDB services")
+        serv3.wait_for_interrupt()
+
+    logger.info("Stopped the OTDB services")
 

SAS/OTDB_Services/TreeStatusEvents.py

@@ -26,6 +26,7 @@ Daemon that watches the OTDB database for status changes of trees and publishes
 
 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
@@ -82,6 +83,8 @@ if __name__ == "__main__":
                       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:
@@ -94,46 +97,55 @@ if __name__ == "__main__":
         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
-    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
-            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 = open('time_save.txt', 'rb').read()
-            except IOError:
-                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:
-                    print treeid, state, modtime, creation
-                    open('time_save.txt', 'wb').write(creation)
-                    start_time = creation
-                print "==="
-
-            # Redetermine the database status.
-            connected = (otdb_connection and otdb_connection.status == 1)
-
-            time.sleep(2)
+    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
+                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 = open('time_save.txt', 'rb').read()
+                except IOError:
+                    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" : state, "time_of_change" : modtime }
+                        msg = EventMessage(content)
+                        print treeid, state, modtime, creation
+                        send_bus.send(msg)
+                        open('time_save.txt', 'wb').write(creation)
+                        start_time = creation
+                    print "==="
+
+                # Redetermine the database status.
+                connected = (otdb_connection and otdb_connection.status == 1)
+
+                time.sleep(2)
 

SAS/OTDB_Services/test/t_TreeService.py

+#!/usr/bin/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.INFO)
+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)
+        print type(data)
+        for (key,value) in data.iteritems():
+            print "%s ==> %s" % (key, value)
+    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=2) as task_spec_request:
+#        do_rpc(task_spec_request, {'OtdbID':63370})
+#        do_rpc(task_spec_request, {'OtdbID':146300})
+#        do_rpc(task_spec_request, {'OtdbID':82111})
+
+#    print StatusUpdateCmd({'OtdbID':146300, 'NewStatus':'finished', 'UpdateTimestamps':True})
+
+    with RPC("KeyUpdateCmd", ForwardExceptions=True, busname=busname, timeout=2) as key_update:
+        print key_update({'OtdbID':63370, 
+                          'Updates':{'LOFAR.ObsSW.Observation.ObservationControl.OnlineControl._hostname':'CCU099ABC'}})
+
+   
+