ingestpipeline.py

#!/usr/bin/env python3
import logging
import os
import time
import subprocess
import random
import socket
import re
import getpass
from prometheus_client import Gauge, Counter, Histogram, INF

from lofar.lta.ingest.common.job import *
from lofar.lta.ingest.server.sip import validateSIPAgainstSchema, addIngestInfoToSIP
from lofar.lta.ingest.server.ltacp import *
from lofar.lta.ingest.server.unspecifiedSIP import makeSIP
from lofar.lta.ingest.server.ltaclient import *
from lofar.lta.ingest.client.rpc import IngestTMSSRPC
from lofar.common.util import humanreadablesize
from lofar.common import isProductionEnvironment
from lofar.common.metrics import metric_track_duration
from lofar.common.subprocess_utils import communicate_returning_strings
from lofar.messaging import EventMessage, ToBus, DEFAULT_BROKER, DEFAULT_BUSNAME
from lofar.lta.ingest.common.config import INGEST_NOTIFICATION_PREFIX
from lofar.lta.ingest.server.config import GLOBUS_TIMEOUT
from lofar.common.dbcredentials import DBCredentials

logger = logging.getLogger(__name__)

#---------------------- Prometheus Metrics ----------------------------------------

metric_nr_transfers_in_progress = Gauge("ingest_transfers_in_progress", "Count how many transfers are currently in progress", labelnames=["site"])
metric_transfer_durations = Histogram("ingest_transfer_durations", "How long transfers are taking", labelnames=["site"],
    buckets=(1.0, 10.0, 60.0, 5 * 60.0, 10 * 60.0, 30 * 60.0, 3600.0, 2 * 3600.0, 4 * 3600.0, 8 * 3600.0, 12 * 3600.0, 24 * 3600.0, INF))
metric_nr_transfer_exceptions = Counter("ingest_transfer_exceptions", "Number of exceptions raised during transfer", labelnames=["site", "reason"])
metric_nr_bytes_transferred = Counter("ingest_bytes_ingested", "Number of payload bytes ingested into the LTA (sum of ingested file sizes)", labelnames=["site"])

#---------------------- Custom Exception ----------------------------------------

PipelineJobFailedError      = 1
PipelineNoSourceError       = 2
PipelineAlreadyInLTAError   = 3
PipelineNoProjectInLTAError = 4

class PipelineError(Exception):
    def __init__(self, message, type = PipelineJobFailedError):
        Exception.__init__(self, message)
        self.type         = type

#---------------------- IngestPipeline ------------------------------------------

class IngestPipeline():
    STATUS_INITIALIZING = 1
    STATUS_TRANSFERRING = 2
    STATUS_FINALIZING   = 3
    STATUS_FINISHED     = 4

    def __init__(self, job, ltaClient,
                 exchange=DEFAULT_BUSNAME,
                 broker=DEFAULT_BROKER,
                 user=getpass.getuser(),
                 globus_timeout=GLOBUS_TIMEOUT,
                 minimal_SIP=False):
        self.status              = IngestPipeline.STATUS_INITIALIZING

        self.hostname            = socket.gethostname()
        self.job                 = job
        self.ltaClient           = ltaClient
        self.user                = user

        if not self.user:
            self.user=getpass.getuser()

        self.globus_timeout      = globus_timeout

        self.minimal_SIP         = minimal_SIP

        self.event_bus           = ToBus(exchange, broker=broker, connection_log_level=logging.DEBUG)
        self.ingest_tmss_rpc     = IngestTMSSRPC.create(exchange, broker, timeout=300)
        self.Project             = job['Project']
        self.DataProduct         = job['DataProduct']
        self.FileName            = job['FileName']
        self.JobId               = job['JobId']
        self.ArchiveId           = int(job['ArchiveId'])
        self.ObsId               = int(job['ObservationId'])
        self.ExportID            = job['ExportID']
        self.Type                = job["Type"]
        self.HostLocation        = job['Location'].partition(':')[0]
        self.Location            = job['Location'].partition(':')[2]
        self.ticket              = ''
        self.FileSize            = '0'
        self.MD5Checksum         = ''
        self.Adler32Checksum     = ''
        self.ChecksumResult      = False
        self.SIP                 = ''
        self.PrimaryUri          = ''
        self.SecondaryUri        = ''
        self.lta_site            = ''

    @metric_track_duration(prefix="ingest_")
    def GetStorageTicket(self):
        do_check_already_in_lta=isProductionEnvironment()
        result = self.ltaClient.GetStorageTicket(self.Project, self.FileName, self.FileSize, self.ArchiveId, self.JobId, self.ObsId, do_check_already_in_lta, self.Type)

        error = result.get('error')
        if error:
            if 'StorageTicket with ID "%i"' % (self.ArchiveId) in error:
                if 'existing_ticket_id' in result and 'existing_ticket_state' in result:
                    logger.warning("Got a Tier 1 GetStorageTicket error for an incomplete storage ticket %s with status %s" % (result['existing_ticket_id'],result['existing_ticket_state']))
                    if result['existing_ticket_state'] < IngestSuccessful:
                        try:
                            self.ticket                = result['existing_ticket_id']
                            logger.warning("trying to repair status of StorageTicket %s" % self.ticket)

                            self.SendStatusToLTA(IngestFailed)
                        except Exception as e:
                            logger.exception('ResettingStatus IngestFailed failed for %s' % self.ticket)
                        raise Exception ('Had to reset state for %s' % self.ticket)
                    else:
                        raise PipelineError('GetStorageTicket error: Dataproduct already in LTA for %s' % (self.JobId), PipelineAlreadyInLTAError)
                else:
                    raise Exception('GetStorageTicket error I can''t interpret: %s' % result)

            if 'no storage resources defined for project' in error or "project does not exists" in error:
                raise PipelineError('GetStorageTicket error for project not known in LTA: %s' % error, PipelineNoProjectInLTAError)

            raise Exception('GetStorageTicket error: %s' % error)
        else:
            self.ticket            = result.get('ticket')
            self.PrimaryUri        = result.get('primary_uri_rnd')
            self.SecondaryUri      = result.get('secondary_uri_rnd')

            if 'sara' in self.PrimaryUri:
                self.lta_site = 'sara'
            elif 'juelich' in self.PrimaryUri:
                self.lta_site = 'juelich'
            elif 'psnc' in self.PrimaryUri:
                self.lta_site = 'poznan'

    @metric_track_duration(prefix="ingest_")
    def TransferFile(self):
        try:
            logger.info('Starting file transfer for %s ' % self.JobId)
            start = time.time()
            self.status = IngestPipeline.STATUS_TRANSFERRING

            self.__sendNotification('JobProgress',
                                    message='transfer starting',
                                    percentage_done=0.0,
                                    total_bytes_transfered=0)

            if 'cep4' in self.HostLocation.lower() or 'cpu' in self.HostLocation.lower():
                self.HostLocation = 'localhost'

            def progress_callback(percentage_done, current_speed, total_bytes_transfered):
                self.__sendNotification('JobProgress',
                                        percentage_done=min(100.0, round(10.0*percentage_done)/10.0),
                                        current_speed=current_speed,
                                        total_bytes_transfered=total_bytes_transfered)

            if (os.path.splitext(self.Location)[-1] == '.h5' and
                os.path.splitext(os.path.basename(self.Location))[0].endswith('_bf')):
                logger.info('dataproduct is a beamformed h5 file. adding raw file to the transfer')
                self.Location = [self.Location, self.Location.replace('.h5', '.raw')]

            if self.DataProduct not in self.Location and 'Source' in self.job:
                # old hack, is needed to support dynspec / pulsar archiving scripts
                self.Location = os.path.join(self.Location, self.job['Source'])

            with metric_track_duration.labels(site=self.lta_site).time(),
                 metric_nr_transfers_in_progress.labels(site=self.lta_site).track_inprogress():
                cp = LtaCp(self.Location,
                           self.PrimaryUri,
                           globus_timeout=self.globus_timeout,
                           progress_callback=progress_callback)

                transfer_result = cp.transfer(force=True)

            self.status = IngestPipeline.STATUS_FINALIZING

            if not transfer_result:
                msg = 'error while transferring %s with ltacp' % (self.JobId)
                logger.error(msg)
                raise Exception(msg)

            self.MD5Checksum = transfer_result[0]
            self.Adler32Checksum = transfer_result[1]
            self.FileSize = transfer_result[2]

            if self.MD5Checksum and self.Adler32Checksum and self.FileSize:
                logger.debug('valid checksums found for %s with filesize %sB (%s). md5: %s adler32: %s', self.JobId,
                                                                                                         self.FileSize,
                                                                                                         humanreadablesize(int(self.FileSize), 'B'),
                                                                                                         self.MD5Checksum,
                                                                                                         self.Adler32Checksum)
            else:
                msg = 'no valid checksums found for %s with filesize %sB (%s). md5: %s adler32: %s' % (self.JobId,
                                                                                                       self.FileSize,
                                                                                                       humanreadablesize(int(self.FileSize), 'B'),
                                                                                                       self.MD5Checksum,
                                                                                                       self.Adler32Checksum)
                logger.error(msg)
                raise Exception(msg)

            try:
                self.__sendNotification('JobProgress',
                                        message='transfer finished',
                                        percentage_done=100.0,
                                        total_bytes_transfered=int(self.FileSize))

                metric_nr_bytes_transferred.inc(self.FileSize)
            except ValueError, TypeError:
                pass

            elapsed = time.time() - start

            try:
                if int(self.FileSize) > 0:
                    avgSpeed = float(self.FileSize) / elapsed
                logger.info("Finished file transfer for %s in %d sec with an average speed of %s for %s including ltacp overhead" % (self.JobId, elapsed, humanreadablesize(avgSpeed, 'Bps'), humanreadablesize(float(self.FileSize), 'B')))
            except Exception:
                logger.info('Finished file transfer of %s in %s' % (self.JobId, elapsed))

        except Exception as exp:
            if isinstance(exp, LtacpException):
                if '550 File not found' in exp.value:
                    logger.error('Destination directory does not exist. Creating %s in LTA for %s' % (self.PrimaryUri, self.JobId))
                    metric_nr_transfer_exceptions.labels(site=self.lta_site,reason="target_directory_not_found").inc()

                    if create_missing_directories(self.PrimaryUri) == 0:
                        logger.info('Created path %s in LTA for %s' % (self.PrimaryUri, self.JobId))
                elif 'source path' in exp.value and 'does not exist' in exp.value:
                    metric_nr_transfer_exceptions.labels(site=self.lta_site,reason="source_path_not_found").inc()
                    raise PipelineError(exp.value, PipelineNoSourceError)
                else:
                    metric_nr_transfer_exceptions.labels(site=self.lta_site,reason="other").inc()

            raise Exception('transfer failed for %s: %s' % (self.JobId, str(exp)))

    @metric_track_duration(prefix="ingest_")
    def SendChecksumsToLTA(self):
        result = self.ltaClient.SendChecksums(self.JobId, self.Project, self.ticket, self.FileSize, self.PrimaryUri, self.SecondaryUri, self.MD5Checksum, self.Adler32Checksum)
        if not result.get('error'):
            #store final uri's
            self.PrimaryUri   = result['primary_uri']
            self.SecondaryUri = result.get('secondary_uri')

    @metric_track_duration(prefix="ingest_")
    def SendStatusToLTA(self, lta_state_id):
        if self.ticket:
            self.ltaClient.UpdateUriState(self.JobId, self.Project, self.ticket, self.PrimaryUri, lta_state_id)

    def CheckForValidSIP(self):
        if self.Type == "TMSS":
            # no need to check for valid SIP's in TMSS. It adds a significant load to the server.
            # If TMSS is able to return a SIP, it's guaranteed to be valid as TMSS does the validation already.
            # If TMSS is not able to return a SIP, that is caused by invalid observation/pipeline feedback, and the SIP won't validate anyway.
            pass
        elif 'SIPLocation' in self.job: # job file might know where the sip is when it is not a MoM job
            try:
                sip_host = self.job['SIPLocation'].split(':')[0]
                sip_path = self.job['SIPLocation'].split(':')[1]

                cmd = ['ssh', '-tt', '-n', '-x', '-q', '%s@%s' % (self.user, sip_host), 'cat %s' % sip_path]
                logger.info("GetSIP for %s at SIPLocation %s - cmd %s" % (self.JobId, self.job['SIPLocation'], ' ' .join(cmd)))
                p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                out, err = communicate_returning_strings(p)
                if p.returncode != 0:
                    raise PipelineError('GetSIP error getting EoR SIP for %s: %s' % (self.JobId, out + err))

                tmp_SIP = out

                tmp_SIP = addIngestInfoToSIP(tmp_SIP, self.ticket, self.FileSize, self.MD5Checksum, self.Adler32Checksum)

                tmp_SIP = tmp_SIP.replace('<stationType>Europe</stationType>','<stationType>International</stationType>')

                #make sure the source in the SIP is the same as the type of the storageticket
                tmp_SIP = re.compile('<source>eor</source>', re.IGNORECASE).sub('<source>%s</source>' % (self.Type,), tmp_SIP)

                if not validateSIPAgainstSchema(tmp_SIP):
                    logger.error('CheckForValidSIP: Invalid SIP:\n%s', tmp_SIP)
                    raise Exception('SIP for %s does not validate against schema' % self.JobId)

            except:
                logger.exception('CheckForValidSIP: Getting SIP from SIPLocation %s failed', self.job['SIPLocation'])
                raise

        logger.info('SIP for %s is valid, can proceed with transfer' % (self.JobId,))

    @metric_track_duration(prefix="ingest_")
    def GetSIP(self):
        try:
            if self.Type == "TMSS":
                # TMSS works differently than MoM
                # an Ingest-Export is a subtask with input dataproducts which should be ingested...
                # and output dataproducts which are ingested
                # so, for this transfered input dataproduct get the corresponding output,
                # and store the archive/transfer info with the output dataproduct

                output_dataproduct_id = self.ingest_tmss_rpc.get_output_dataproduct_id(subtask_id=self.job.get('TMSSIngestSubtaskId'),
                                                                                       input_dataproduct_id=self.job['TMSSInputDataproductId'])
                self.ingest_tmss_rpc.store_archive_information(dataproduct_id=output_dataproduct_id,
                                                               storage_ticket=self.ticket,
                                                               size=int(self.FileSize),
                                                               filepath=self.PrimaryUri,
                                                               md5_hash=self.MD5Checksum,
                                                               adler32_hash=self.Adler32Checksum)

                # get the SIP for the output dataproduct which is/should_be the same as for the input,
                # but enriched with the archive information which is needed by the LTA.
                self.SIP = self.ingest_tmss_rpc.get_SIP(output_dataproduct_id)

            elif 'SIPLocation' in self.job: # job file might know where the sip is when it is not a MoM job
                try:
                    sip_host = self.job['SIPLocation'].split(':')[0]
                    sip_path = self.job['SIPLocation'].split(':')[1]

                    cmd = ['ssh', '-tt', '-n', '-x', '-q', '%s@%s' % (self.user, sip_host), 'cat %s' % sip_path]
                    logger.info("GetSIP for %s at SIPLocation %s - cmd %s" % (self.JobId, self.job['SIPLocation'], ' ' .join(cmd)))
                    p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
                    out, err = communicate_returning_strings(p)
                    if p.returncode != 0:
                        raise PipelineError('GetSIP error getting EoR SIP for %s: %s' % (self.JobId, out + err))

                    self.SIP = out

                    self.SIP = addIngestInfoToSIP(self.SIP, self.ticket, self.FileSize, self.MD5Checksum, self.Adler32Checksum)

                    self.SIP = self.SIP.replace('<stationType>Europe</stationType>','<stationType>International</stationType>')

                    #make sure the source in the SIP is the same as the type of the storageticket
                    self.SIP = re.compile('<source>eor</source>', re.IGNORECASE).sub('<source>%s</source>' % (self.Type,), self.SIP)

                    if not validateSIPAgainstSchema(self.SIP):
                        logger.error('Invalid SIP:\n%s', self.SIP)
                        raise Exception('SIP for %s does not validate against schema' % self.JobId)

                except:
                    logger.exception('Getting SIP from SIPLocation %s failed', self.job['SIPLocation'])
                    raise

                logger.info('SIP received for %s from SIPLocation %s with size %d (%s): \n%s' % (self.JobId,
                                                                                                self.job['SIPLocation'],
                                                                                                len(self.SIP),
                                                                                                humanreadablesize(len(self.SIP)),
                                                                                                self.SIP[0:1024]))
            else:
                self.SIP = makeSIP(self.Project, self.ObsId, self.ArchiveId, self.ticket, self.FileName, self.FileSize, self.MD5Checksum, self.Adler32Checksum, self.Type)
                self.FileType = FILE_TYPE_UNSPECIFIED
        except Exception as e:
            if self.minimal_SIP:
                logger.info('making minimal SIP for %s', self.JobId)
                self.SIP = makeSIP(self.Project, self.ObsId, self.ArchiveId, self.ticket, self.FileName, self.FileSize, self.MD5Checksum, self.Adler32Checksum, self.Type)
                logger.info('minimal SIP for %s: \n%s', self.JobId, self.SIP)
                self.FileType = FILE_TYPE_UNSPECIFIED
            else:
                raise

    @metric_track_duration(prefix="ingest_")
    def SendSIPToLTA(self):
        try:
            self.ltaClient.SendSIP(self.JobId, self.SIP, self.ticket)
        except Exception as e:
            logger.error('SendSIPToLTA exception: %s', e)
            raise PipelineError(str(e), PipelineJobFailedError)

    @metric_track_duration(prefix="ingest_")
    def RollBack(self):
        try:
            logger.info('rolling back file transfer for %s', self.JobId)
            start     = time.time()

            if self.PrimaryUri:
                srmrm(self.PrimaryUri, log_prefix=self.JobId, timeout=300)

            if self.SecondaryUri:
                srmrm(self.SecondaryUri, log_prefix=self.JobId, timeout=300)

            if self.Type == "TMSS":
                # delete archive info for the output_dataproduct in TMSS
                output_dataproduct_id = self.ingest_tmss_rpc.get_output_dataproduct_id(subtask_id=self.job.get('TMSSIngestSubtaskId'),
                                                                                       input_dataproduct_id=self.job['TMSSInputDataproductId'])
                self.ingest_tmss_rpc.delete_hashes_and_archive_information(output_dataproduct_id)

            logger.debug("rollBack for %s took %ds", self.JobId, time.time() - start)
        except Exception as e:
            logger.exception('rollback failed for %s: %s', self.JobId, e)

    def __sendNotification(self, subject, message='', **kwargs):
        try:
            contentDict = { 'job_id': self.JobId,
                            'export_id': self.job.get('job_group_id', self.job.get('TMSSIngestSubtaskId')),
                            'archive_id': self.ArchiveId,
                            'project': self.Project,
                            'type': self.Type,
                            'ingest_server': self.hostname,
                            'dataproduct': self.DataProduct,
                            'srm_url': self.PrimaryUri }
            if 'ObservationId' in self.job and self.Type.lower()=='tmss':
                contentDict['tmss_producing_subtask_id'] = self.job['ObservationId']

            if 'TMSSInputDataproductId' in self.job and self.Type.lower()=='tmss':
                contentDict['input_dataproduct_id'] = self.job['TMSSInputDataproductId']

            if self.lta_site:
                contentDict['lta_site'] = self.lta_site

            if message:
                contentDict['message'] = message

            for k,v in list(kwargs.items()):
                contentDict[k] = v

            msg = EventMessage(subject="%s.%s" % (INGEST_NOTIFICATION_PREFIX, subject), content=contentDict)
            msg.ttl = 48*3600 #remove message from queue's when not picked up within 48 hours
            logger.info('Sending notification %s: %s' % (subject, str(contentDict).replace('\n', ' ')))
            self.event_bus.send(msg)
        except Exception as e:
            logger.error(str(e))

    def run(self):
        with self.event_bus, self.ingest_tmss_rpc:
            try:
                logger.info("starting ingestpipeline for %s" % self.JobId)
                start = time.time()
                self.__sendNotification('JobStarted')

                self.GetStorageTicket()
                self.CheckForValidSIP()
                self.TransferFile()
                self.SendChecksumsToLTA()
                self.GetSIP()
                self.SendSIPToLTA()
                self.SendStatusToLTA(IngestSuccessful)

                avgSpeed = 0
                elapsed = time.time() - start
                try:
                    avgSpeed = float(self.FileSize) / elapsed
                    logger.info("Ingest Pipeline finished for %s in %d sec with average speed of %s for %s including all overhead",
                                self.JobId, elapsed, humanreadablesize(avgSpeed, 'Bps'), humanreadablesize(float(self.FileSize), 'B'))
                except Exception:
                    logger.info("Ingest Pipeline finished for %s in %d sec", self.JobId, elapsed)

                self.__sendNotification('JobFinished',
                                        average_speed=avgSpeed,
                                        total_bytes_transfered=int(self.FileSize))

            except PipelineError as pe:
                logger.log(logging.WARNING if pe.type == PipelineAlreadyInLTAError else logging.ERROR,
                           'Encountered PipelineError for %s : %s', self.JobId, str(pe))
                if pe.type == PipelineNoSourceError:
                    self.__sendNotification('JobTransferFailed', 'data not transfered because it was not on disk')
                elif pe.type == PipelineAlreadyInLTAError:
                    self.__sendNotification('JobFinished', 'data was already in the LTA',
                                            average_speed=0,
                                            total_bytes_transfered=0)
                else:
                    self.RollBack()

                    # by default the error_message for the notification is the exception
                    error_message = str(pe)
                    self.__sendNotification('JobTransferFailed', error_message)

                try:
                    if pe.type != PipelineAlreadyInLTAError:
                        self.SendStatusToLTA(IngestFailed)
                except Exception as e:
                    logger.error('SendStatusToLTA failed for %s: %s', self.JobId, e)

            except Exception as e:
                logger.error('Encountered unexpected error for %s: %s', self.JobId, e)

                # by default the error_message for the notification is the exception
                error_message = str(e)
                # for known messsages in the exception, make a nice readable error_message
                if 'ltacp' in error_message and ('file listing failed' in error_message or 'du failed' in error_message):
                    error_message = 'dataproduct %s not found at location %s:%s' % (self.DataProduct, self.HostLocation, self.Location)
                elif 'does not validate against schema' in error_message:
                    error_message = 'invalid SIP does not validate against schema'

                try:
                    self.RollBack()
                except Exception as rbe:
                    logger.error('RollBack failed for %s: %s', self.JobId, rbe)
                try:
                    self.SendStatusToLTA(IngestFailed)
                except Exception as sse:
                    logger.error('SendStatusToLTA failed for %s: %s', self.JobId, sse)
                try:
                    self.__sendNotification('JobTransferFailed', error_message)
                except Exception as sne:
                    logger.error('sendNotification failed for %s: %s', self.JobId, sne)
            finally:
                self.status = IngestPipeline.STATUS_FINISHED

def main():
    import os.path
    from optparse import OptionParser, OptionGroup
    from lofar.common import dbcredentials

    # Check the invocation arguments
    parser = OptionParser("%prog [options] <path_to_jobfile.xml>",
                          description='Run the ingestpipeline on a single jobfile.')
    parser.add_option('-q', '--broker', dest='broker', type='string', default=DEFAULT_BROKER, help='Address of the qpid broker, default: %default')
    parser.add_option('--busname', dest='busname', type='string', default=DEFAULT_BUSNAME, help='Name of the bus exchange on the qpid broker on which the ingest notifications are published, default: %default')
    parser.add_option("-u", "--user", dest="user", type="string", default=getpass.getuser(), help="username for to login on <host>, [default: %default]")
    parser.add_option('-s', '--minimal-SIP', dest='minimal_SIP', action='store_true', help='create and upload a minimal SIP to the LTA catalogue when the normal SIP is not accepted.')
    parser.add_option('-V', '--verbose', dest='verbose', action='store_true', help='verbose logging')
    parser.add_option('-t', '--timeout', dest='globus_timeout', type='int', default=GLOBUS_TIMEOUT, help='number of seconds (default=%default) to wait for globus-url-copy to finish after the transfer is done (while lta-site is computing checksums)')
    parser.add_option("-l", "--lta_credentials", dest="lta_credentials", type="string",
                      default='LTA' if isProductionEnvironment() else 'LTA_test',
                      help="Name of lofar credentials for lta user/pass (see ~/.lofar/dbcredentials) [default=%default]")
    (options, args) = parser.parse_args()

    logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
                        level=logging.DEBUG if options.verbose else logging.INFO)

    if len(args) != 1:
        parser.print_help()
        sys.exit(1)

    try:
        path = args[0]

        if os.path.isfile(path):
            job = parseJobXmlFile(path)
            job['filename'] = path
            logger.info("Parsed jobfile %s: %s", path, job)

            ltacreds = dbcredentials.DBCredentials().get(options.lta_credentials)
            ltaClient = LTAClient(ltacreds.user, ltacreds.password)

            jobPipeline = IngestPipeline(job, ltaClient,
                                         busname=options.busname,
                                         broker=options.broker,
                                         user=options.user,
                                         globus_timeout=options.globus_timeout,
                                         minimal_SIP=options.minimal_SIP)
            jobPipeline.run()
            exit(0)
        else:
            logger.info("No such file %s", path)
            exit(1)
    except Exception as e:
        logger.error(e)
        exit(1)

if __name__ == '__main__':
    main()