SW-488: commisioning fixes from test with Joern Sander Thomas and me

LCS/PyCommon/subprocess_utils.py

@@ -71,7 +71,7 @@ def execute_in_parallel(cmd_lists, timeout=3600, max_parallel=32):
 
     # log results of commands
     for cmd_list, result in zip(cmd_lists, results):
-        logger.debug("Results for cmd: %s\n  returncode=%s\n  stdout=%s\n  stderr=%s",
+        logger.info("Results for cmd: %s\n  returncode=%s\n  stdout=%s\n  stderr=%s",
                      " ".join(cmd_list),
                      result.returncode, result.stdout.rstrip(), result.stderr.rstrip())
     return results

MAC/Services/TBB/TBBServer/lib/tbbservice.py

@@ -47,7 +47,7 @@ from lofar.mac.tbb.tbb_set_storage import set_tbb_storage, create_mapping
 from lofar.mac.tbb.tbb_start_recording import start_tbb
 from lofar.mac.tbb.tbb_upload_to_cep import upload_tbb_data
 from lofar.parameterset import parameterset
-from lofar.mac.tbb.tbb_util import parse_parset_from_voevent
+from lofar.mac.tbb.tbb_util import parse_parset_from_voevent, get_cpu_nodes_available_for_tbb_datawriters_sorted_by_load
 from lofar.common.lcu_utils import stationname2hostname
 from lofar.mac.tbb.tbb_caltables import add_station_calibration_tables_h5_files_in_directory
 from lofar.mac.tbb.tbb_cable_delays import add_dipole_cable_delays_h5_files_in_directory
@@ -222,16 +222,16 @@ class TBBControlService:
             'Observation.TBB.TBBsetting.observatoryCoordinates': 0,
             'Observation.TBB.TBBsetting.observatoryCoordinatesCoordinateSystem': 0,
             'Observation.TBB.TBBsetting.triggerId': 0,
-            'Observation.TBB.TBBsetting.additionalInfo': voevent,
+            #'Observation.TBB.TBBsetting.additionalInfo': voevent,
             # older keys we probably still want to fill here:
             'Observation.TBB.TBBsetting.triggerType': 'Unknown',
             'Observation.TBB.TBBsetting.triggerVersion': 0
         }
-        self._update_parset(parset, defaults_parset)
+        parset = self._update_parset(parset, defaults_parset)
 
         # update parset with values from received voevent
         voevent_parset = parse_parset_from_voevent(voevent)
-        self._update_parset(parset, voevent_parset)
+        parset = self._update_parset(parset, voevent_parset)
 
         return parset_path, parset
 
@@ -250,11 +250,28 @@ class TBBControlService:
         # TODO: make dynamic, depending on which stations are actually used
         ports = ','.join(str(x) for x in range(31664, 31670))
 
+        # TODO: do not start more dw's than there are stations.
         available_nodes = get_cpu_nodes_available_for_tbb_datawriters_sorted_by_load(min_nr_of_free_nodes=1)
 
         start = datetime.utcnow()
         timeout = timedelta(days=1) #TODO: make timeout configurable (or an argument)
 
+        #create output dir
+        cmd = ['mkdir', '-p', output_path]
+        cmd = wrap_command_in_cep4_head_node_ssh_call(cmd)
+        logger.info('creating output dir on cep4: %s, executing: %s', output_path, ' '.join(cmd))
+        proc = Popen(cmd)
+        proc.wait()
+
+        #write parset
+        parset_dict = parset.dict()
+        parset_str = '\n'.join('%s = %s' % (k,parset_dict[k]) for k in sorted(parset_dict.keys())).replace('"', '').replace('ObsSW.', '')
+        cmd = ['echo', '\"%s\"' % (parset_str,), '>', parset_path]
+        cmd = wrap_command_in_cep4_head_node_ssh_call(cmd)
+        logger.info('writing tbb datawriter parset, executing: %s', ' '.join(cmd))
+        proc = Popen(cmd)
+        proc.wait()
+
         # TODO: what should be do if datawriters are already running? kill/wait for them first?
         self.procs = {}
 
@@ -264,10 +281,13 @@ class TBBControlService:
                    '-p', ports,
                    '-k', '0',
                    '-t', '60',
+                   '-d', '3840', #TODO: compute instead of hardcoded
                    '-o', output_path]
 
             # wrap the command in a cep4 docker ssh call
-            cmd = wrap_command_for_docker(cmd, DATAWRITER_DOCKER_IMAGE_NAME, DATAWRITER_DOCKER_IMAGE_TAG, added_privileges=True)
+            cmd = wrap_command_for_docker(cmd, DATAWRITER_DOCKER_IMAGE_NAME, DATAWRITER_DOCKER_IMAGE_TAG,
+                                          mount_dirs=['/data/projects', '/data/scratch', '/data/share', '/data/parsets'],
+                                          added_privileges=True)
             cmd = wrap_command_in_cep4_cpu_node_ssh_call(cmd, node, via_head=True)
 
             logger.info('starting datawriter on node %s, executing: %s', node, ' '.join(cmd))
@@ -283,7 +303,7 @@ class TBBControlService:
         Monitor started procs until they are all done or timeout seconds have
         passed.
         :param timeout:  Timeout in seconds until data writers will be
-        forcefully killed.  The default is 24 hours. 
+        forcefully killed.  The default is 24 hours.
         '''
         start = datetime.utcnow()
         while self.procs:
@@ -496,10 +516,10 @@ class TBBControlService:
         datawriter_nodes = self.start_datawriters(output_path=output_path, voevent=voevent_xml)
         if nodes is None:
             logger.info('Nodes to use are not configured, using all with datawriter')
-            nodes = datawriter_nodes
+            nodes = ['cpu%02d' % (node,) for node in datawriter_nodes]
         else:
             logger.info('Filtering node list for those who actually have a running datawriter')
-            nodes = [node for node in nodes if node in datawriter_nodes]
+            nodes = ['cpu%02d' % (node,) for node in nodes if node in datawriter_nodes]
 
         # create mapping for storage nodes
         try:

MAC/TBB/lib/tbb_upload_to_cep.py

@@ -12,7 +12,7 @@ import subprocess
 import logging
 from lofar.mac.tbb.tbb_config import *
 from lofar.mac.tbb.tbb_util import split_stations_by_boardnumber, expand_list, calculate_adjusted_start_time, wrap_remote_composite_command
-
+from lofar.common.lcu_utils import execute_in_parallel_over_stations
 
 def upload_tbb_data(stations, dm, start_time, duration, sub_bands, wait_time, boards):
     """
@@ -40,33 +40,47 @@ def upload_tbb_data(stations, dm, start_time, duration, sub_bands, wait_time, bo
         # batch handle all stations with same number of boards through lcurun
         for num_boards in stationlists.keys():
             logging.debug("Creating TBB commands for stations with %s boards..." % num_boards)
-            relay = lcurun_command + [",".join(stationlists[num_boards])]
+            #relay = lcurun_command + [",".join(stationlists[num_boards])]
+            stations_with_num_boards = stationlists[num_boards]
 
             # iterate over tbb boards
-            board_list = ""
+            board_list = []
             for board in boards:
                 if int(board) <= num_boards:
-                    board_list += " %s" % (board)
+                    board_list.append(board)
                 else:
                     logging.error("Stations \"%s\" do not have a board #%s!  The stations have only %s boards.  Continuing without this board." % (stationlists[num_boards], board, num_boards))
                     continue
-            # Now build the two loops that execute tbbctl for every channel and
-            # for every board.
-            # Why first loop over channels and then over boards?  That allows
-            # to add only one sleep after a channel is uploaded for all boards.
-            # Otherwise each board would have the sleep and we would sleep
-            # # of boards times for every channel instead. 
-            cmd = "for((channel = 0; channel <= 15; ++channel)); do for board in"
-            cmd += board_list
-            cmd += "; do "
-            cmd += tbb_cmd % (sub_band, adjusted_start_time, slice_nr, duration)
-            cmd += "; done; sleep "
-            cmd += str(wait_time)
-            cmd += "; done"
-            cmd = wrap_remote_composite_command(cmd)
-            command = relay + [cmd]
-            logging.info('Executing %s', ' '.join(command))
-            subprocess.check_call(command)
+
+            cmd_list = []
+            for channel in range(15):
+                for board in board_list:
+                    tbb_cmd = "tbbctl --readband=%s,%s,%s,%s,%s,%s" % (board, channel, sub_band, adjusted_start_time, slice_nr, duration)
+                    cmd_list.append(tbb_cmd)
+                cmd_list.append("sleep %s" % (wait_time,))
+
+            full_cmd = '\"%s\"' % (' ; '.join(cmd_list),)
+
+            execute_in_parallel_over_stations(full_cmd, stations_with_num_boards)
+
+
+            # # Now build the two loops that execute tbbctl for every channel and
+            # # for every board.
+            # # Why first loop over channels and then over boards?  That allows
+            # # to add only one sleep after a channel is uploaded for all boards.
+            # # Otherwise each board would have the sleep and we would sleep
+            # # # of boards times for every channel instead.
+            # cmd = "for((channel = 0; channel <= 15; ++channel)); do for board in"
+            # cmd += board_list
+            # cmd += "; do "
+            # cmd += tbb_cmd % (sub_band, adjusted_start_time, slice_nr, duration)
+            # cmd += "; done; sleep "
+            # cmd += str(wait_time)
+            # cmd += "; done"
+            # cmd = wrap_remote_composite_command(cmd)
+            # command = relay + [cmd]
+            # logging.info('Executing %s', ' '.join(command))
+            # subprocess.check_call(command)
         time.sleep(wait_time)
     logging.info("Uploading TBB data done.")
 
@@ -94,7 +108,7 @@ def parse_args():
     if args.wait_time_between_sub_bands is not None and args.wait_time_between_sub_bands > 0.0:
         args.wait_time = float(args.wait_time_between_sub_bands)
     else:
-        args.wait_time = float(args.duration) * 0.00012
+        args.wait_time = float(args.duration/1000.0) * 0.00012
 
     return args