Subversion Repositories DevTools

Rev

Rev 7397 | Blame | Last modification | View Log | RSS feed 

#! /usr/bin/perl
########################################################################
# COPYRIGHT - VIX IP PTY LTD ("VIX"). ALL RIGHTS RESERVED.
#
# Module name   : blatS3Sync.pl
# Module type   :
# Compiler(s)   : Perl
# Environment(s):
#
# Description   :   This is a blat related task that will perform S3 SYNC
#                   transfers for configured releases
#
# Usage         :   ARGV[0] - Path to config file for this instance
#
#......................................................................#

require 5.008_002;
use strict;
use warnings;
use Getopt::Long;
use File::Basename;
use Data::Dumper;
use File::Spec::Functions;
use POSIX ":sys_wait_h";
use File::Temp qw/tempfile/;
use Digest::MD5 qw(md5_base64 md5_hex);
use File::Path qw( rmtree );
use Archive::Zip qw( :ERROR_CODES :CONSTANTS );
use JSON;

use FindBin;                                    # Determine the current directory
use lib "$FindBin::Bin/lib";                    # Allow local libraries

use Utils;
use StdLogger;                                  # Log to sdtout
use Logger;                                     # Log to file

#
#   Database interface
#   Pinched from jats and modified so that this software is not dependent on JATS
#
use IO::Handle;
use JatsRmApi;
use DBI;

#
#   Globals
#
my $logger = StdLogger->new();                  # Stdout logger. Only during config
$logger->err("No config file specified") unless (defined $ARGV[0]);
$logger->err("Config File does not exist: $ARGV[0]") unless (-f $ARGV[0]);
my $name = basename( $ARGV[0]);
   $name =~ s~.conf$~~;
my $now = 0;
my $startTime = 0;
my $tagDirTime = 0;
my $lastDirScan = 0;
my $lastS3Refresh =  0;
my $lastTagListUpdate = 0;
my $mtimeConfig = 0;
my $conf;
my $yday = -1;
my $linkUp = 1;
my $RM_DB;
my $activeReleases;
my $wedgedCount = 0;

#
#   Contain statisics maintained while operating
#       Can be dumped with a kill -USR2
#       List here for documentation
#  

my %statistics = (
    SeqNum => 0,                        # Bumped when $statistics are dumped
    timeStamp => 0,                     # DateTime when statistics are dumped
    upTime => 0,                        # Seconds since program start
    Cycle => 0,                         # Major process loop counter
    phase => 'Init',                    # Current phase of operation
    state => 'OK',                      # Nagios state
    wedged => 0,                        # Wedge indication - main loop not cycling
                                        # 
                                        # The following are reset each day
    dayStart => 0,                      # DateTime when daily data was reset
    txCount => 0,                       # Packages Transferred
    txBytes => 0,                       # Bytes Transferred
    delCount => 0,                      # Packages marked for deletion
    linkErrors => 0,                    # Transfer (S3) errors
                                        # 
                                        # Per Cycle Data - Calculated each processing Cycle
    total => 0,                         # Number targets
);

#
#   Describe configuration parameters
#
my %cdata = (
    'piddir'          => {'mandatory' => 1      , 'fmt' => 'dir'},
    'sleep'           => {'default'   => 5      , 'fmt' => 'period'},
    'sleepLinkDown'   => {'default'   => '1m'   , 'fmt' => 'period'},
    'dpkg_archive'    => {'mandatory' => 1      , 'fmt' => 'dir'},
    'logfile'         => {'mandatory' => 1      , 'fmt' => 'vfile'},
    'logfile.size'    => {'default'   => '1M'   , 'fmt' => 'size'},
    'logfile.count'   => {'default'   => 9      , 'fmt' => 'int'},

    'verbose'         => {'default'   => 0      , 'fmt' => 'int'},                  # Debug ...
    'active'          => {'default'   => 1      , 'fmt' => 'bool'},                 # Disable alltogether
    'debug'           => {'default'   => 0      , 'fmt' => 'bool'},                 # Log to screen
    'txdetail'        => {'default'   => 0      , 'fmt' => 'bool'},                 # Show transfer times
    'noTransfers'     => {'default'   => 0      , 'fmt' => 'bool'},                 # Debugging option to prevent transfers
    'transferDir'     => {'default'   => 'pkg/S3TRANSFER' , 'fmt' => 'text'},

    'tagdir'          => {'mandatory' => 1      , 'fmt' => 'mkdir'},
    'workdir'         => {'mandatory' => 1      , 'fmt' => 'mkdir'},
    'forcedirscan'    => {'default'   => 100    , 'fmt' => 'period'},
    'forces3update'   => {'default'   => '30m'  , 'fmt' => 'period'},
    'tagListUpdate'   => {'default'   => '1h'   , 'fmt' => 'period'},
    'S3Bucket'        => {'mandatory' => 1      , 'fmt' => 'text'},
    'S3Profile'       => {'mandatory' => 1      , 'fmt' => 'text'},
    'S3Region'        => {'default' => undef    , 'fmt' => 'text'},

);


#
#   Read in the configuration
#       Set up a logger
#       Write a pidfile - thats not used
$now = $startTime = time();
readConfig();
Utils::writepid($conf);
$logger->logmsg("Starting...");
readStatistics();
sighandlers($conf);

#
#   Main processing loop
#   Will exit when terminated by parent
#
while (1)
{
    $logger->verbose3("Processing");
    $statistics{Cycle}++;
    $wedgedCount = 0;
    $now = time();

    $statistics{phase} = 'ReadConfig';
    readConfig();
    if ( $conf->{'active'} )
    {
        $statistics{phase} = 'Refresh S3 Info';
        refreshS3Info();
        if( $linkUp )
        {
            $statistics{phase} = 'Monitor Requests';
            monitorRequests();

            $statistics{phase} = 'maintainTagList';
            maintainTagList();
        }
    }

    $statistics{phase} = 'Sleep';
    sleep( $linkUp ? $conf->{'sleep'} : $conf->{'sleepLinkDown'} );
    reapChildren();

    #   If my PID file ceases to be, then exit the daemon
    #   Used to force daemon to restart
    #
    unless ( -f $conf->{'pidfile'} )
    {
        $logger->logmsg("Terminate. Pid file removed");
        last;
    }
}
$statistics{phase} = 'Terminated';
$logger->logmsg("Child End");
exit 0;

#-------------------------------------------------------------------------------
# Function        : reapChildren 
#
# Description     : Reap any and all dead children
#                   Call in major loops to prevent zombies accumulating 
#
# Inputs          : None
#
# Returns         : 
#
sub reapChildren
{
    my $currentPhase = $statistics{phase};
    $statistics{phase} = 'Reaping';

    my $kid;
    do {
        $kid = waitpid(-1, WNOHANG);
    } while ( $kid > 0 );

    $statistics{phase} = $currentPhase;
}


#-------------------------------------------------------------------------------
# Function        : readConfig
#
# Description     : Re read the config file if it modification time has changed
#
# Inputs          : Nothing
#
# Returns         : 0       - Config not read
#                   1       - Config read
#                             Config file has changed
#
sub readConfig
{
    my ($mtime) = Utils::mtime($ARGV[0]);
    my $rv = 0;

    if ( $mtimeConfig != $mtime )
    {
        $logger->logmsg("Reading config file: $ARGV[0]");
        $mtimeConfig = $mtime;
        my $errors;
        ($conf, $errors) = Utils::readconf ( $ARGV[0], \%cdata );
        if ( scalar @{$errors} > 0 )
        {
            warn "$_\n" foreach (@{$errors});
            die ("Config contained errors\n");
        }

        #
        #   Reset some information
        #   Create a new logger
        #
        $logger = Logger->new($conf) unless $conf->{debug};
        $conf->{logger} = $logger;
        $conf->{'pidfile'} = $conf->{'piddir'} . '/' . $name . '.pid';
        $logger->setVerbose($conf->{verbose});
        $logger->verbose("Log Levl: $conf->{verbose}");

        #
        #   Setup statistics filename
        $conf->{'statsfile'} = $conf->{'piddir'} . '/' . $name . '.stats';
        $conf->{'statsfiletmp'} = $conf->{'piddir'} . '/' . $name . '.stats.tmp';

        #
        #   When config is read force some actions
        #       - Force tagList to be created
        #       - Force refresh from S3
        $lastTagListUpdate = 0;
        $lastS3Refresh = 0;
        $rv = 1;

        #
        #   When config is read force some actions
#Utils::DebugDumpData ("Config", $conf);

        $logger->warn("All Transfers disabled") if ( $conf->{'noTransfers'} );
        $logger->warn("S3Sync is inactive") unless ( $conf->{'active'} );
    }

    return $rv;
}

#-------------------------------------------------------------------------------
# Function        : refreshS3Info 
#
# Description     : At startup, and at time after startup examine the S3 bucket
#                   and recover information from it 
#
# Inputs          : 
#
# Returns         : 0 - Gross error ( Bucket access) 
#
sub refreshS3Info
{
    my $rv = 1;
    if ( !$linkUp || ($now > ($lastS3Refresh + $conf->{'forces3update'})) )
    {
        $logger->verbose2("refreshS3Info");
        $lastS3Refresh = $now;

        #
        #   Examine the s3 bucket and extract useful information
        #
        my $startTime = time;
        $rv =  examineS3Bucket();
         unless ($rv) {
            $statistics{linkErrors}++;
            $linkUp = 0;
         } else {
             $linkUp = 1;
         }

         #
         #   Display the duration of the refresh
         #       Diagnostic use
         #
         if ($conf->{txdetail}) {
             my $duration = time - $startTime;
             $logger->logmsg("refreshS3Info: Stats: $duration Secs");
         }

    }
    return $rv;
}



#-------------------------------------------------------------------------------
# Function        : monitorRequests
#
# Description     : Monitor S3Sync requests
#                   This is simply done my polling Release Manager - at the moment
#
# Inputs          : None
#
# Returns         : Nothing
#
sub monitorRequests
{
    #
    #   Determine if new tags are present by examining the time
    #   that the directory was last modified.
    #
    #   Allow for a forced scan to catch packages that did not transfer
    #   on the first attempt
    #
    my $tagCount = 0;
    my ($mtime) = Utils::mtime($conf->{'tagdir'} );
    if ( ($mtime > $tagDirTime) || ($now > ($lastDirScan + $conf->{'forcedirscan'})) )
    {
        $logger->verbose2("monitorRequests: $conf->{'tagdir'}");
        #$logger->verbose2("monitorRequests: mtime:" . ($mtime > $tagDirTime));
        #$logger->verbose2("monitorRequests: last:" . ($now > ($lastDirScan + $conf->{'forcedirscan'})));

        #
        #   Package tags information is not really used
        #       Just delete all the tags
        #       Used to trigger the scan - rather than rely on the slow data
        #       base poll. Still need a change in release sequence number
        #   
        my $dh;
        unless (opendir($dh, $conf->{'tagdir'}))
        {
            $logger->warn ("can't opendir $conf->{'tagdir'}: $!");
            return;
        }

        #
        #   Process each entry
        #       Ignore those that start with a .
        #       Remove all files
        #
        while (my $tag = readdir($dh) )
        {
            next if ( $tag =~ m~^\.~ );
            my $file = "$conf->{'tagdir'}/$tag";
            $logger->verbose3("processTags: $file");

            next unless ( -f $file );
            unlink $file;
        }

        #
        #   Reset the scan time triggers
        #   
        $tagDirTime = $mtime;
        $lastDirScan = $now;

        #
        #   Examine Release Manager looking for active releases that have S3Sync support
        #   Purpose is to:
        #       Detect new Releases
        #       Detect dead Releases
        #       Detect changed Releases
        #
        connectRM(\$RM_DB, $conf->{verbose} > 3);

        foreach my $rtag_id (keys %{$activeReleases}) {
            $activeReleases->{$rtag_id}{exists} = 0;
        }
        my $m_sqlstr = "SELECT rt.rtag_id,rm.seqnum, rt.s3sync, rt.official, rm.timestamp " . 
                       "FROM RELEASE_MANAGER.release_tags rt, RELEASE_MANAGER.release_modified rm " .
                       "WHERE rt.s3sync = 'Y' AND rm.rtag_id = rt.rtag_id AND rt.official in ('N', 'R', 'C')";

        my $curData = getDataFromRm ('monitorRequests', $m_sqlstr, {data => 0} );
        $statistics{total} = scalar @{$curData};

        foreach my $entry (@{$curData}) {
            my ($rtag_id, $seqnum) = @{$entry};
            $logger->verbose3("rtagid: $rtag_id, seqnumm: $seqnum");

            if (! exists $activeReleases->{$rtag_id} || ! exists $activeReleases->{$rtag_id}{s3}  ) {
                $logger->logmsg("New Release Detected. rtag_id: $rtag_id, seq:$seqnum");
                processChangedRelease($rtag_id, $seqnum);
                $lastTagListUpdate = 0;

            } elsif (($activeReleases->{$rtag_id}{seqnum} || 0) ne ($seqnum || 0) ) {
                $logger->logmsg("Change Release Detected. rtag_id: $rtag_id, seq:$seqnum");
                processChangedRelease($rtag_id, $seqnum);
            }

            # Update activeReleases so that changes will be detected
            $activeReleases->{$rtag_id}{exists} = 1;
        }

        # Detect Releases that are no longer active
        foreach my $rtag_id (keys %{$activeReleases}) {
            unless ($activeReleases->{$rtag_id}{exists}) {
                $logger->logmsg("Dead Release Detected. rtag_id: $rtag_id");
                removeDeadRelease($rtag_id);
                delete $activeReleases->{$rtag_id};
                $lastTagListUpdate = 0;
            }
        }

        disconnectRM(\$RM_DB);
    }
}

#-------------------------------------------------------------------------------
# Function        : examineS3Bucket 
#
# Description     : Scan the S3 bucket looking for Releases
#                   Used to pre-populate the process so that we:
#                       - Delete dead releases
#                       - Don't do excessive work on startup
#                       
# Inputs          : Nothing 
#
# Returns         : Updates global structure ($activeReleases) 
# Returns         : 0 - Gross error ( Bucket access) 
#
sub examineS3Bucket
{
    my $bucket;
    my $prefix;

    #
    #   Remove data collected from s3
    #
    foreach my $rtag_id (keys %{$activeReleases}) {
        delete $activeReleases->{$rtag_id}{s3}  ;
    }

    if ($conf->{'S3Bucket'} =~ m~(.*?)/(.*)~) {
        $bucket = $1;
        $prefix = $2;
    } else {
        $bucket = $conf->{'S3Bucket'};
    }

    my $s3_cmd = "aws --profile $conf->{'S3Profile'} --output json";
    $s3_cmd .= " --region $conf->{'S3Region'}" if (defined $conf->{'S3Region'});
    $s3_cmd .= " s3api list-objects --bucket $bucket";
    $s3_cmd .= " --prefix '$prefix'" if (defined $prefix);

    $logger->verbose2("examineS3Bucket:s3_cmd:$s3_cmd");

    my $ph;
    my $jsontxt = "";
    open ($ph, "$s3_cmd |");
    while ( <$ph> ) {
        chomp;
        $logger->verbose3("examineS3Bucket:Data: $_");
        $jsontxt .= $_;
    }
    close ($ph);
    my $cmdRv = $?;
    if ($cmdRv != 0) {
        $logger->warn("Cannot read S3 Bucket Data");
        return 0;
    }

    if ($jsontxt) {
        my $json = from_json ($jsontxt);
        #Utils::DebugDumpData("JSON",$json->{'Contents'});
        foreach my $item ( @{$json->{'Contents'}})
        {
            if ($item->{Key} =~ m~(?:^|/)Release-(.*)\.zip$~ ) {

                my $rtag_id = $1;
                my $metaData = gets3ObjectMetaData($item->{Key});

                #
                #   Update info in the global structure ($activeReleases)
                #   This data could be discarded - only needed for diagnostics   
                #
                $activeReleases->{$rtag_id}{s3}{seqnum} = $metaData->{'releaseseq'};   
                $activeReleases->{$rtag_id}{s3}{md5}    = $metaData->{'md5'};   
                $activeReleases->{$rtag_id}{s3}{depsig} = $metaData->{'depsig'};
                
                #
                #   Recover information from S3
                #   Should only be done on the first call after restart
                #   
                if (! exists $activeReleases->{$rtag_id}{md5} ) {
                    $activeReleases->{$rtag_id}{md5} = $activeReleases->{$rtag_id}{s3}{md5};
                    $activeReleases->{$rtag_id}{depsig} = $activeReleases->{$rtag_id}{s3}{depsig};
                }
                   

            } else {
                $logger->warn("Unknown item in bucket: $item->{Key}");
            }
        }
    }
#Utils::DebugDumpData("activeReleases",$activeReleases);
    return 1;
}

#-------------------------------------------------------------------------------
# Function        : gets3ObjectMetaData 
#
# Description     : Get Metadata about one object
#                   Must do object by object :( 
#
# Inputs          : $key    - Key 
#
# Returns         : 
#

sub gets3ObjectMetaData
{
    my ($key) = @_;
    my $bucket;
    my $prefix;

    if ($conf->{'S3Bucket'} =~ m~(.*?)/(.*)~) {
        $bucket = $1;
        $prefix = $2;
    } else {
        $bucket = $conf->{'S3Bucket'};
        $prefix = '';
    }

    my $s3_cmd = "aws --profile $conf->{'S3Profile'} --output json";
    $s3_cmd .= " --region $conf->{'S3Region'}" if (defined $conf->{'S3Region'});
    $s3_cmd .= " s3api head-object --bucket $bucket --key $key";

    $logger->verbose2("gets3ObjectMetaData:s3_cmd:$s3_cmd");

    my $ph;
    my $jsontxt = "";
    open ($ph, "$s3_cmd |");
    while ( <$ph> ) {
        chomp;
        $logger->verbose3("gets3ObjectMetaData:Data: $_");
        $jsontxt .= $_;
    }
    close ($ph);

    my $json;
    $json->{Metadata} = {};

    if ($jsontxt) {
        $json = from_json ($jsontxt);
        #Utils::DebugDumpData("JSON",$json);
    }
    return $json->{Metadata};
}


#-------------------------------------------------------------------------------
# Function        : removeDeadRelease 
#
# Description     : Remove a Dead Release from the S3 bucket 
#
# Inputs          : $rtag_id    - Release identifier 
#
# Returns         : 0   - Nothing deleted
#                   1   - Something deleted 
#
sub removeDeadRelease {
    my ($rtag_id) = @_;
    my $cmdRv;
    my $rv = 0;

    #   Create the process pipe to delete the package

    my $targetPath = generateBucketZipName($rtag_id);

    my $s3_cmd = "aws --profile $conf->{'S3Profile'} --output json";
    $s3_cmd .= " --region $conf->{'S3Region'}" if (defined $conf->{'S3Region'});
    $s3_cmd .= " s3 rm s3://$targetPath";

    $logger->logmsg("removeDeadRelease:$targetPath");
    $logger->verbose2("removeDeadRelease:s3_cmd:$s3_cmd");

    my $ph;
    open ($ph, "$s3_cmd |");
    while ( <$ph> ) {
        chomp;
        $logger->verbose2("removeDeadRelease:Data: $_");
    }
    close ($ph);
    $cmdRv = $?;

    #
    #   Common code
    #
    $logger->verbose("removeDeadRelease:End: $cmdRv");
    if ( $cmdRv == 0 ) {
        $rv = 1;
        $statistics{delCount}++;

    } else {
        $logger->warn("removeDeadRelease:Error: $rtag_id, $?");
    }
    return $rv;
}

#-------------------------------------------------------------------------------
# Function        : processChangedRelease 
#
# Description     : Create/Update a release to the S3 bucket
# 
#   Various attempts are made to reduce the work that needs to be done
#   There are three checks to skip a transfer
#       1) The Release sequence number - must be diff for processing to occur
#       2) Packages inserted into the image
#           Dependent package versions are used to generate a MD5
#           If this does not change then there is no need to do work
#       3) An MD5 over the zip Image
#          If this is the same as the one in S3, then don't upload
#          
#   These three pices of information are held as metadata along with the
#   package. These are read at start up.
#
#
# Inputs          : $rtag_id    - Release identifier
#                   $seqnum     - Release sequence number
#                                 Added as metadata to objects 
#
# Returns         : Nothing 
#
sub processChangedRelease {
    my ($rtag_id, $seqnum) = @_;

    #
    #   Cleanout previous zip files
    #   
    my @files = glob($conf->{'workdir'} . '/*.zip');
    $logger->verbose("Delete old zips: @files");
    unlink @files;

    #
    #   Create an image of the data to be transferred
    #   Based on packages that support S3Sync
    #   
    my $m_sqlstr = "SELECT p.pkg_name, pv.pkg_version, pv.pv_id " .
                   " FROM RELEASE_MANAGER.release_content rc, RELEASE_MANAGER.packages p, RELEASE_MANAGER.package_versions  pv " .
                    " WHERE rc.rtag_id = " . $rtag_id .
                    "      AND rc.s3sync = 'Y' " .
                    "      AND rc.pv_id = pv.pv_id " .
                    "      AND pv.pkg_id = p.pkg_id " .
                    " ORDER by pv.pv_id";

    my $curData = getDataFromRm ('s3Pkgs', $m_sqlstr, {data => 0, dump => 0} );
#Utils::DebugDumpData("activeReleases",$activeReleases);

    #
    #   Generate a md5 of the PVIDs of the packages that will go into the image
    #   Used to detect true changes - only of the packages we are interested in
    #   
    my $signature = Digest::MD5->new;
    foreach my $entry (@{$curData}) {
        $signature->add( $entry->[2] )
    }
    my $depsig = $signature->hexdigest();
    my $reason = "";
    if ( !exists $activeReleases->{$rtag_id}{s3} ) {
        $reason = 'NoS3Data';

    } elsif (! exists $activeReleases->{$rtag_id}{depsig}) {
        $reason = 'NoSavedData';

    } elsif ($activeReleases->{$rtag_id}{depsig} ne $depsig ) {
        $reason = "Mismatch: $activeReleases->{$rtag_id}{depsig} ne $depsig";

    } else {
        $logger->verbose("Dependencies unchanged - upload skipped");
        $activeReleases->{$rtag_id}{seqnum} = $seqnum;
        return;
    }
    $logger->verbose("Dependency Test: $reason");

    #
    #   Create a monifest to go into the zip
    #
    my $manifest;
    $manifest->{Packages} = [];
    $manifest->{rtag_id} = $rtag_id;

    #
    #   Generate the zip of the objects to be pushed to S3
    #       Add directories
    #       Update the manifest entries
    #
    my $startTime = time;
    my $zip = Archive::Zip->new();
    foreach my $entry (@{$curData}) {
        my $src = getPackageBase($entry->[0], $entry->[1]);
        if (defined $src) {
            $logger->verbose("Zip addTree Src: $src");

            my %data;
            $data{name} = $entry->[0];
            $data{version} = $entry->[1];
            $data{pvid} = $entry->[2];
            push @{$manifest->{Packages}}, \%data;

            if ( $zip->addTree( $src, '' ) != AZ_OK ) {
                $logger->warn("Zip addTree Error: $rtag_id");
                return;
            }
        }
    }

    #   Add the manifest into the zip
    my $jsonText = to_json( $manifest, { ascii => 1, pretty => 1 });
    $zip->addString( $jsonText, 'ReleaseManifest.json' );
    $logger->verbose("ManfestJson: $jsonText");

    #   Generate the zip file
    my $zipFile = catdir( $conf->{'workdir'} , 'Images-' . $rtag_id . '.zip');
    if ( $zip->writeToFileNamed($zipFile) != AZ_OK ) {
        $logger->warn("Zip write Error: $rtag_id");
        return;
    }
    $logger->verbose("Zip created: $zipFile");

    #
    #   Display the size of the package (zipped)
    #       Diagnostic use
    #
    if ($conf->{txdetail}) {
        my $tzfsize = -s $zipFile;
        my $size = sprintf "%.3f", $tzfsize / 1024 / 1024 / 1024 ;
        my $duration = time - $startTime;
        $logger->logmsg("zipImage: Stats: $rtag_id, $size Gb, $duration Secs");
    }

    #
    #   Have a ZIP file of the desired contents
    #   Could try to detect if it differs from the one already in the bucket
    #       Don't want to trigger CI/CD pipeline operations unless we need to
    #       
    my $digest = md5_hex($zipFile);
    $reason = "";
    if ( !exists $activeReleases->{$rtag_id}{s3} ) {
        $reason = 'NoS3Data';

    } elsif (! exists $activeReleases->{$rtag_id}{md5}) {
        $reason = 'NoSavedMd5';

    } elsif ($activeReleases->{$rtag_id}{md5} ne $digest ) {
        $reason = "Mismatch: $activeReleases->{$rtag_id}{md5} ne $digest";

    } else {
        $logger->verbose("Zip file has same md5 hash - upload skipped");
        #
        #   Update the known signature
        $activeReleases->{$rtag_id}{depsig} = $depsig;
        $activeReleases->{$rtag_id}{seqnum} = $seqnum;
        $activeReleases->{$rtag_id}{md5} = $digest;
        return;
    }
    $logger->verbose("ZipMd5 Test: $reason");

    #   Create a command to transfer the file to AWS use the cli tools
    #   Note: Ive seen problem with this when used from Perth to AWS (Sydney)
    #         If this is an issue use curl - see the savePkgToS3.sh for an implementation
    #
    $startTime = time;
    my $targetPath = generateBucketZipName($rtag_id);

    my $s3_cmd = "aws --profile $conf->{'S3Profile'} --output json";
    $s3_cmd .= " --region $conf->{'S3Region'}" if (defined $conf->{'S3Region'});
    $s3_cmd .= " s3 cp $zipFile s3://$targetPath --metadata releaseseq=$seqnum,md5=$digest,depsig=$depsig";

    $logger->logmsg("transferPackage:$targetPath");
    $logger->verbose2("transferPackage:s3_cmd:$s3_cmd");

    my $cmdRv;
    unless ($conf->{'noTransfers'}) {
        my $ph;
        open ($ph, "$s3_cmd |");
        while ( <$ph> )
        {
            chomp;
            $logger->verbose2("transferPackage:Data: $_");
        }
        close ($ph);
        $cmdRv = $?;
        $logger->verbose("transferPackage:End: $cmdRv");
    }
    #
    #   Display the size of the package (zipped)
    #       Diagnostic use
    #
    if ($conf->{txdetail}) {
        my $tzfsize = -s $zipFile;
        my $size = sprintf "%.3f", $tzfsize / 1024 / 1024 / 1024 ;
        my $duration = time - $startTime;
        $logger->logmsg("S3 Copy: Stats: $rtag_id, $size Gb, $duration Secs");
    }

    if ($cmdRv == 0) {
        $statistics{txCount}++;
        $statistics{txBytes} += -s $zipFile; 

        #
        #   Mark the current entry as having been processed
        #
        $activeReleases->{$rtag_id}{depsig} = $depsig;
        $activeReleases->{$rtag_id}{md5} = $digest;
        $activeReleases->{$rtag_id}{seqnum} = $seqnum;
        $activeReleases->{$rtag_id}{s3}{sent} = 1;
    }
}

#-------------------------------------------------------------------------------
# Function        : getPackageBase 
#
# Description     : Calculate the base of a package in dpkg_archive
#                   With errors and wanings
#
# Inputs          : $pname      - Package name
#                   $pver       - Package version
#
#
# Returns         : undef - bad
#                   Path to the S3TRANSFER section within the archive
sub getPackageBase {
    my ($pname, $pver) = @_;

    #
    #   Locate package
    #
    unless ( -d $conf->{'dpkg_archive'}) {
        $logger->warn("addPartsToImage: dpkg_archive not found");
        return undef;
    }
    
    my $src = catdir($conf->{'dpkg_archive'}, $pname, $pver);
    unless ( -d $src ) {
        $logger->warn("addPartsToImage: Package not found: $pname, $pver");
        return undef;
    }

    $src = catdir( $src, $conf->{'transferDir'});
    unless ( -d $src ) {
        $logger->verbose("addPartsToImage: Package has no $conf->{'transferDir'}: $pname, $pver");
        return undef;
    }
    return $src;
}

#-------------------------------------------------------------------------------
# Function        : generateBucketZipName 
#
# Description     : Generate the name of the zipfile created within the bucket  
#
# Inputs          : $rtag_id 
#
# Returns         : Full name - including bucket name
#
sub generateBucketZipName
{
    my ($rtag_id) = @_;
    my $targetName = 'Release-' . $rtag_id . '.zip';
    my $targetPath = catdir ($conf->{'S3Bucket'}, $targetName );
    return $targetPath;
}


#-------------------------------------------------------------------------------
# Function        : getDataFromRm 
#
# Description     : Get an array of data from RM
#                   Normally an array of arrays 
#
# Inputs          : $name           - Query Name
#                   $m_sqlstr       - Query
#                   $options        - Ref to a hash of options
#                                       sql     - show sql
#                                       data    - show data
#                                       dump    - show results
#                                       oneRow  - Only fetch one row
#                                       error   - Must find data
#                                       
# Returns         : ref to array of data
#
sub getDataFromRm
{
    my ($name,$m_sqlstr, $options ) = @_;
    my @row;
    my $data;

    if (ref $options ne 'HASH') {
        $options = {}; 
    }

    if ($options->{sql}) {
        $logger->logmsg("$name: $m_sqlstr")
    }
    my $sth = $RM_DB->prepare($m_sqlstr);
    if ( defined($sth) )
    {
        if ( $sth->execute( ) ) {
            if ( $sth->rows ) {
                while ( @row = $sth->fetchrow_array ) {
                    if ($options->{data}) {
                        $logger->warn ("$name: @row");
                    }
                    #Debug0("$name: @row");
                    push @{$data}, [@row];

                    last if $options->{oneRow};
                }
            }
            $sth->finish();
        } else {
            $logger->warn("Execute failure:$name: $m_sqlstr", $sth->errstr() );
        }
    } else {
        $logger->warn("Prepare failure:$name" );
    }

    if (!$data && $options->{error}) {
        $logger->warn( $options->{error} );
    }

    if ($data && $options->{oneRow}) {
        $data = $data->[0];
    }
 
    if ($options->{dump}) {
        Utils::DebugDumpData("$name", $data);
    }
    return $data;
}

#-------------------------------------------------------------------------------
# Function        : maintainTagList
#
# Description     : Maintain a data structure for the maintenance of the
#                   tags directory
#
# Inputs          : None
#
# Returns         : Nothing
#
sub maintainTagList
{
    #
    #   Time to perform the scan
    #   Will do at startup and every time period there after
    #
    return unless ( $now > ($lastTagListUpdate + $conf->{tagListUpdate} ));
    $logger->verbose("maintainTagList");
    $lastTagListUpdate = $now;

    #
    #   Generate new configuration
    #
    my %config;
    $config{s3Sync} = 1;                # Indicate that it may be special

    %{$config{releases}} = map { $_ => 1 } keys %{$activeReleases};

    #
    #   Save data
    #
    my $dump =  Data::Dumper->new([\%config], [qw(*config)]);
#print $dump->Dump;
#$dump->Reset;

    #
    #   Save config data
    #
    my $conf_file = catfile( $conf->{'tagdir'},'.config' );
    $logger->verbose3("maintainTagList: Writting $conf_file");

    my $fh;
    open ( $fh, '>', $conf_file ) or $logger->err("Can't create $conf_file: $!");
    print $fh $dump->Dump;
    close $fh;
}

#-------------------------------------------------------------------------------
# Function        : resetDailyStatistics 
#
# Description     : Called periodically to reset the daily statistics
#
# Inputs          : $time       - Current time
#
# Returns         : 
#
sub resetDailyStatistics
{
    my ($time) = @_;

    #
    #   Detect a new day
    #
    my $today = (localtime($time))[7];
    if ($yday != $today)
    {
        $yday = $today;
        $logger->logmsg('Resetting daily statistics' );

        # Note: Must match @recoverTags in readStatistics
        $statistics{dayStart} = $time;
        $statistics{txCount} = 0;
        $statistics{txBytes} = 0;
        $statistics{delCount} = 0;
        $statistics{linkErrors} = 0;
    }
}

#-------------------------------------------------------------------------------
# Function        : readStatistics 
#
# Description     : Read in the last set of stats
#                   Used after a restart to recover daily statistics
#
# Inputs          : 
#
# Returns         : 
#
sub readStatistics
{
    my @recoverTags = qw(dayStart txCount txBytes delCount linkErrors);

    if ($conf->{'statsfile'} and -f $conf->{'statsfile'})
    {
        if (open my $fh, $conf->{'statsfile'})
        {
            while (<$fh>)
            {
                m~(.*):(.*)~;
                if ( grep( /^$1$/, @recoverTags ) ) 
                {
                    $statistics{$1} = $2;
                    $logger->verbose("readStatistics $1, $2");
                }
            }
            close $fh;
            $yday = (localtime($statistics{dayStart}))[7];
        }
    }
}


#-------------------------------------------------------------------------------
# Function        : periodicStatistics 
#
# Description     : Called on a regular basis to write out statistics
#                   Used to feed information into Nagios
#                   
#                   This function is called via an alarm and may be outside the normal
#                   processing loop. Don't make assumptions on the value of $now
#
# Inputs          : 
#
# Returns         : 
#
sub periodicStatistics
{
    #
    #   A few local stats
    #
    $statistics{SeqNum}++;
    $statistics{timeStamp} = time();
    $statistics{upTime} = $statistics{timeStamp} - $startTime;
    $statistics{wedged} = $wedgedCount++ > 30  ? 1 : 0;

    if ( $statistics{wedged}) {
         $statistics{state} = 'Wedged';
    } elsif(!$linkUp){
        $statistics{state} = 'S3 Bucket Read Error';
    } else {
        $statistics{state} = 'OK';
    }

    #   Reset daily accumulations - on first use each day
    resetDailyStatistics($statistics{timeStamp});
    
    #
    #   Write statistics to a file
    #       Write to a tmp file, then rename.
    #       Attempt to make the operation atomic - so that the file consumer
    #       doesn't get a badly formed file.
    #   
    if ($conf->{'statsfiletmp'})
    {
        my $fh;
        unless (open ($fh, '>', $conf->{'statsfiletmp'}))
        {
            $fh = undef;
            $logger->warn("Cannot create temp stats file: $!");
        }
        else
        {
            foreach my $key ( sort { lc($a) cmp lc($b) } keys %statistics)
            {
                print $fh $key . ':' . $statistics{$key} . "\n";
                $logger->verbose2('Statistics:'. $key . ':' . $statistics{$key});
            }
            close $fh;

            # Rename temp to real file
            rename  $conf->{'statsfiletmp'},  $conf->{'statsfile'} ;
        }
    }
}

#-------------------------------------------------------------------------------
# Function        : sighandlers
#
# Description     : Install signal handlers
#
# Inputs          : $conf           - System config
#
# Returns         : Nothing
#
sub sighandlers
{
    my $conf = shift;
    my $logger = $conf->{logger};

    $SIG{TERM} = sub {
        # On shutdown
        $logger->logmsg('Received SIGTERM. Shutting down....' );
        unlink $conf->{'pidfile'} if (-f $conf->{'pidfile'});
        exit 0;
    };

    $SIG{HUP} = sub {
        # On logrotate
        $logger->logmsg('Received SIGHUP.');
        $logger->rotatelog();
    };

    $SIG{USR1} = sub {
        # On Force Rescans
        $logger->logmsg('Received SIGUSR1.');
        $lastTagListUpdate = 0;
        $lastS3Refresh = 0;
    };

    alarm 60;
    $SIG{ALRM} = sub {
        # On Dump Statistics
        $logger->verbose2('Received SIGUSR2.');
        periodicStatistics();
        alarm 60;
    };

    $SIG{__WARN__} = sub { $logger->warn("@_") };
    $SIG{__DIE__} = sub { $logger->err("@_") };
}


#-------------------------------------------------------------------------------
# Function        : Error, Verbose, Warning
#
# Description     : Support for JatsRmApi
#
# Inputs          : Message
#
# Returns         : Nothing
#
sub Error
{
    $logger->err("@_");
}

sub Verbose
{
    $logger->verbose2("@_");
}

sub Warning
{
    $logger->warn("@_");
}