New Subscription Features

Since the adoption of NGAS at ICRAR for MWA and GLEAM, we have added / improved several new / existing features. This page aims to explain those new features and enhancement from the perspective of an NGAS operator (but not necessarily an NGAS developer)

Update an existing subscription

We can use the new command USUBSCRIBE to update a running subscription "on the fly", for example:

Set (increase or reduce) the # of parallel streams to five

curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&concurrent_threads=5

Go back in time to re-subscribe

curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&start_date=2012-06-17T16:31:19.936

Change delivery url

 curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&url=http://ngashost-02:7777/QARCHIVE

Change priority

 curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&priority=2

Suspend file delivery temporarily 

(useful when, e.g. disks on target NGAS is full)

 curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&suspend=1

Resume file delivery

 (useful when, e.g. new empty disks on target NGAS are added)

 curl http://ngashost-01:7777/USUBSCRIBE?subscr_id=ngashost-01-to-ngashost-02\&suspend=0

Monitor subscriptions

through subscription queues

Subscription queue is a table (ngas_subscr_queue) in the NGAS database. We (or some Web GUI) can connect to the NGAS database as a READONLY user for the following:

-- to check files already in the queue (to be sent), but not yet delivered:
SELECT * FROM ngas_subscr_queue WHERE subscr_id = 'ngashost-01-to-ngashost-02' AND status = -2;
 
-- to check files currently being delivered:
SELECT * FROM ngas_subscr_queue WHERE subscr_id = 'ngashost-01-to-ngashost-02' AND status = -1;
 
-- to check files (and transfer rate) that have been successfully delivered:
SELECT file_id, file_version, comment FROM ngas_subscr_queue WHERE subscr_id = 'ngashost-01-to-ngashost-02' AND status = 0;
 
-- to check files that Subscription has failed to deliver and to show their error messages:
SELECT file_id, file_version, comment FROM ngas_subscr_queue WHERE subscr_id = 'ngashost-01-to-ngashost-02' AND status > 0;

The persistent subscription queue allows subscription to survive NGAS server restart or unexpected system crashes. NGAS will always resume from the last successfully-delivered file.

through NGAS logs

grep delivered LogFile.nglog # to see all successful file deliveries in the past 24 hours
grep Delivering LogFile.nglog # to see all file delivery attempts in the past 24 hours

Data Mover setup

As discussed in the NGAS paper, a data mover is a stripped-down version of a regular NGAS server optimised for dataflow management. A Data Mover is able to perform all dataflow-related tasks (such as data transfer, subscription management, etc.) on behalf of a set of regular-mode NGAS servers as long as it has read access to their archived files on some storage media. Setting up a data mover requires two simple steps:

Update NGAS configuration file

This file is normally located at /home/ngas/NGAS/cfg/ngamsServer.conf

<Server Id="Server-DataMover"
          ArchiveName="DataMover"
          BlockSize="65536"
          MaxSimReqs="10"
          RootDirectory="/Users/Chen/NGAS"
          PortNo="7779"
		  IpAddress="192.168.1.3"
          ProxyMode="1"
          Simulation="0"
          SwVersion=""/>
... 
<DataMoverOnly FromHostIds="192.168.1.3:7777, 192.168.1.3:7778"
                                 SuspensionTime="0T00:03:00"/>

The above example shows a Data Mover's configuration file. It sits on host 192.168.1.3 at port 7779. It will check, for every three minutes, if any new files have been ingested into two NGAS servers running on the same host but at the port 7777 and 7778 respectively. If so, the Data Mover will notify all its subscribers, each of whom will then either deliver or ignore these files. As shown in the above configuration file, in the "FromHostIds" attribute of the DataMoverOnly XML element, host ids of several NGAS servers are separated by commas.

Launch command

Launching a data mover is not that much different from starting an NGAS server, for example:

python /home/ngas/ngas_rt/src/ngamsServer/ngamsDataMoverServer.py -cfg /home/ngas/NGAS_DM/cfg/ngamsServer_dm.conf -autoOnline -force -multiplesrvs

Submit a "subscription" job

It is possible to use the Subscription framework to execute jobs on selected files in the NGAS archive. For example, "to extract the FITS header from all image that (1) were ingested since last Friday, and (2) appear only in the southern sky, and insert them into a relational database". In terms of file manipulation, this request is no different from "delivering all image files to MIT that (1) were ingested since last Friday (2) and appear only in the southern sky"

The SUBSCRIBE example above looks like a regular subscription command. But the protocol scheme in the url parameter, which becomes ngasjob:// instead of http://, declares that this is a job submission rather than a file delivery request. The URL path is then taken as the (python) file name of the job executable (in this case, ngamsFITS_JobPI.py) located at the NGAS plugin directory (i.e. src/ngamsPlugIns/). All existing parameters used in regular subscription requests (i.e. filter_plug_in, plug_in_pars) are the same. An additional parameter is redo_on_fail, which determines whether a failed job execution on a file will be re-tried ( = 1) or not ( = 0). Note that this behaviour is always true for a regular subscription request / command. It should be emphasised that the redo_on_fail parameter is inside the url paramter of the subscription command. Therefore, it should be URL encoded. For example:

ngasjob://ngamsFITS_JobPI?redo_on_fail=0  becomes:  ngasjob://ngamsFITS_JobPI%3Fredo_on_fail%3D0  in the final subscription command URL request.

Since a job subscription is a special type of subscription (or the other way around?), the monitoring tools described above are all applicable to the job subscription. It is also possible to launch a data mover to which job subscriptions are subsequently issued.

When to use job subscription

In both MWA and GLEAM, Job-based subscription has been extensively used for the following three scenarios:

Batch processing - Process all files satisfying some conditions currently in the archive

Incremental processing - Automatically, continuously, and selectively processing "newly" ingested files. 

Asynchronous processing - Process each newly ingested files without interfering with archiving. This is an advantage over the DAPI plugin, which will block the archiving progress until the job processing is done. 

TCP buffersize

For long-haul TCP-based data transfer, it is extremely important to fine tune the Linux kernel for optimal bandwidth utilisation. But in systems where such fine tuning is not possible or straightforward, NGAS parameters can be adjusted for changing the TCP buffer size on a per-connection basis.

Client-side

To set the TCP buffer size to one Megabytes using the NGAS C-Client:

ngamsCClient -host ngas-host01 -port 7777 -cmd QARCHIVE -mimeType application/octet-stream -fileUri <full_path_file_name> -sndbuf 1048576

Server-side (Sender NGAS)

To set NGAS send buffer size to one Megabytes for ALL subscribers in the NGAS configuration file (i.e. the ArchiveSndBufSize attribute of the Server XML element). Note that currently we cannot set separate buffer sizes for different subscribers. We envisage this feature will be included in future NGAS releases.

<Server Id="Server-DataMover"
          ArchiveName="DataMover"
          BlockSize="65536"
		  ArchiveSndBufSize="1048576"
          MaxSimReqs="10"
          RootDirectory="/Users/Chen/NGAS"
          PortNo="7779"
		  IpAddress="192.168.1.3"
          ProxyMode="1"
          Simulation="0"
          SwVersion=""/>

Server-side (Receiver NGAS)

Note that if you have set the server side "ArchiveSndBufSize", sometimes it is also useful to set the receiver side TCP window: ArchiveRcvBufSize, which appears at the same XML section as the ArchiveSndBufSize, but located on the receiver side NGAS server.