db/pgbackrest
1
0
Fork 0
mirror of https://github.com/pgbackrest/pgbackrest.git synced 2025-10-30 23:37:45 +02:00
Code Issues Releases Activity

Finally a working copy() with new protocol.

Browse Source
This commit is contained in:
David Steele
2014-06-14 19:50:54 -04:00
parent 3a32363a56
commit 79f85fe6c4
4 changed files with 274 additions and 109 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
bin/pg_backrest_remote.pl
View File

@@ -18,6 +18,7 @@ use lib dirname($0) . '/../lib';
use BackRest::Utility;
use BackRest::File;
use BackRest::Remote;
use BackRest::Exception;
####################################################################################################################################
# Operation constants
@@ -57,12 +58,16 @@ use constant
# Turn off logging
log_level_set(OFF, OFF);
# Create the file object
my $oFile = BackRest::File->new();
# Create the remote object
my $oRemote = BackRest::Remote->new();
# Create the file object
my $oFile = BackRest::File->new
(
bCompress => false,
oRemote => $oRemote
);
# Write the greeting so remote process knows who we are
$oRemote->greeting_write();
@@ -95,7 +100,8 @@ while ($strCommand ne OP_EXIT)
confess "destination_file must be defined";
}
$oFile->copy(PIPE_STDOUT, undef, PATH_ABSOLUTE, $oParamHash{destination_file});
$oFile->copy(PIPE_STDIN, undef, PATH_ABSOLUTE, $oParamHash{destination_file});
$oRemote->output_write();
}
else
{
@@ -105,7 +111,7 @@ while ($strCommand ne OP_EXIT)
}
}
};
if ($@)
{
$oRemote->error_write($@);

90
lib/BackRest/File.pm
View File

@@ -127,7 +127,7 @@ use constant
OP_FILE_COMPRESS => "compress",
OP_FILE_MOVE => "move",
OP_FILE_COPY_OUT => "copy_out",
OP_FILE_COPY_IN => "copy_in",
OP_FILE_COPY_IN => "File->copy_in",
OP_FILE_PATH_CREATE => "path_create"
};
@@ -730,8 +730,8 @@ sub copy
# Set working variables
my $strErrorPrefix = "File->copy";
my $bSourceRemote = $self->is_remote($strSourcePathType);
my $bDestinationRemote = $self->is_remote($strDestinationPathType);
my $bSourceRemote = $self->is_remote($strSourcePathType) || $strSourcePathType eq PIPE_STDIN;
my $bDestinationRemote = $self->is_remote($strDestinationPathType) || $strDestinationPathType eq PIPE_STDOUT;
my $strSourceOp = $strSourcePathType eq PIPE_STDIN ?
$strSourcePathType : $self->path_get($strSourcePathType, $strSourceFile);
my $strDestinationOp = $strDestinationPathType eq PIPE_STDOUT ?
@@ -746,24 +746,17 @@ sub copy
}
# Output trace info
&log(TRACE, "${strErrorPrefix}:" . ($bSourceRemote ? " remote" : " local") . " ${strSourcePathType}:${strSourceFile}" .
" to" . ($bDestinationRemote ? " remote" : " local") . " ${strDestinationPathType}:${strDestinationFile}" .
", compress = " . ($bCompress ? "true" : "false"));
# &log(TRACE, "${strErrorPrefix}:" . ($bSourceRemote ? " remote" : " local") . " ${strSourcePathType}:${strSourceFile}" .
# " to" . ($bDestinationRemote ? " remote" : " local") . " ${strDestinationPathType}:${strDestinationFile}" .
# ", compress = " . ($bCompress ? "true" : "false"));
# Open the source file
my $hSourceFile;
if (!$bSourceRemote)
{
if ($strSourcePathType eq PIPE_STDIN)
{
$hSourceFile = *STDIN;
}
else
{
open($hSourceFile, "<", $strSourceOp)
or confess &log(ERROR, "cannot open ${strSourceOp}: " . $!);
}
open($hSourceFile, "<", $strSourceOp)
or confess &log(ERROR, "cannot open ${strSourceOp}: " . $!);
}
# Open the destination file
@@ -771,36 +764,42 @@ sub copy
if (!$bDestinationRemote)
{
if ($strSourcePathType eq PIPE_STDOUT)
{
$hDestinationFile = *STDOUT;
}
else
{
open($hDestinationFile, ">", $strDestinationTmpOp)
or confess &log(ERROR, "cannot open ${strDestinationTmpOp}: " . $!);
}
open($hDestinationFile, ">", $strDestinationTmpOp)
or confess &log(ERROR, "cannot open ${strDestinationTmpOp}: " . $!);
}
# If source or destination are remote
if ($bSourceRemote || $bDestinationRemote)
{
print "got outside\n";
# print "got outside\n";
# Build the command and open the local file
my $hFile;
my $strOperation;
my %oParamHash;
my $hIn,
my $hOut;
my $strRemote;
# If source is remote and destination is local
if ($bSourceRemote && !$bDestinationRemote)
{
return false;
$strRemote = 'in';
$hOut = $hDestinationFile;
if ($strSourcePathType eq PIPE_STDIN)
{
$hIn = *STDIN;
}
}
# Else if source is local and destination is remote
elsif (!$bSourceRemote && $bDestinationRemote)
{
$strRemote = 'out';
$hIn = $hSourceFile;
$strOperation = OP_FILE_COPY_IN;
$oParamHash{destination_file} = ${strDestinationOp};
$hOut = $self->{oRemote}->{hIn};
# Build debug string
# $strDebug = "${strOperation}: remote (" . $self->{oRemote}->command_param_string(\%oParamHash) . "): " . $strDebug;
@@ -818,24 +817,26 @@ sub copy
return false;
}
# Trace command
&log(TRACE, "${strErrorPrefix} operation:" . $strOperation);
# Execute the operation
$self->{oRemote}->command_write($strOperation, \%oParamHash);
# If source is remote and destination is local
if ($bSourceRemote && !$bDestinationRemote)
# If an operation is defined then write it
if (defined($strOperation))
{
#$self->pipe($hOut, $hDestinationFile, $bCompress, !$bCompress);
}
# Else if source is local and destination is remote
elsif (!$bSourceRemote && $bDestinationRemote)
{
$self->{oRemote}->binary_xfer($hSourceFile, $self->{hOut}, 'out');
# $self->pipe($hSourceFile, $hIn, $bCompress, !$bCompress);
}
# Trace command
&log(TRACE, "${strErrorPrefix} operation:" . $strOperation);
# Execute the operation
$self->{oRemote}->command_write($strOperation, \%oParamHash);
}
# Transfer the file
# print "binary xfer start\n";
$self->{oRemote}->binary_xfer($hIn, $hOut, $strRemote);
# print "binary xfer stop\n";
if ($strRemote eq 'out')
{
$self->{oRemote}->output_read(false, $strErrorPrefix);
}
# Wait for process exit (and error)
# $self->wait_pid($pId, $strCommand, $hIn, $hOut, $hErr);
}
@@ -876,8 +877,7 @@ sub copy
# Move the file from tmp to final destination
$self->move($self->path_type_get($strDestinationPathType) . ":absolute", $strDestinationTmpOp,
$self->path_type_get($strDestinationPathType) . ":absolute", $strDestinationOp, $bPathCreate);
$self->move(PATH_ABSOLUTE, $strDestinationTmpOp, PATH_ABSOLUTE, $strDestinationOp, $bPathCreate);
}
return true;
@@ -1130,7 +1130,7 @@ sub exists
&log(DEBUG, $strDebug);
# Execute the command
return $self->{oRemote}->command_execute($strOperation, \%oParamHash, $strDebug) eq "Y";
return $self->{oRemote}->command_execute($strOperation, \%oParamHash, true, $strDebug) eq "Y";
}
# Run locally
else

269
lib/BackRest/Remote.pm
View File

@@ -11,11 +11,23 @@ use Carp;
use Moose;
use Net::OpenSSH;
use File::Basename;
use POSIX ":sys_wait_h";
use lib dirname($0) . "/../lib";
use BackRest::Exception;
use BackRest::Utility;
####################################################################################################################################
# Remote xfer default block size constant
####################################################################################################################################
use constant
{
DEFAULT_BLOCK_SIZE => 8192
};
####################################################################################################################################
# Module variables
####################################################################################################################################
# Protocol strings
has strGreeting => (is => 'ro', default => 'PG_BACKREST_REMOTE');
@@ -33,13 +45,8 @@ has hIn => (is => 'bare'); # SSH object
has hOut => (is => 'bare'); # SSH object
has hErr => (is => 'bare'); # SSH object
####################################################################################################################################
# Remote xfer block size constant
####################################################################################################################################
use constant
{
BLOCK_SIZE => 8192
};
# Block size
has iBlockSize => (is => 'bare', default => DEFAULT_BLOCK_SIZE); # Set block size to default
####################################################################################################################################
# CONSTRUCTOR
@@ -86,6 +93,10 @@ sub BUILD
$self->greeting_read();
}
else
{
}
}
####################################################################################################################################
@@ -111,7 +122,7 @@ sub greeting_read
my $self = shift;
# Make sure that the remote is running the right version
if (readline($self->{hOut}) ne $self->{strGreeting} . "\n")
if ($self->read_line($self->{hOut}) ne $self->{strGreeting})
{
confess &log(ERROR, "remote version mismatch");
}
@@ -147,6 +158,23 @@ sub string_write
}
}
####################################################################################################################################
# PIPE_TO_STRING Function
#
# Copies data from a file handle into a string.
####################################################################################################################################
sub pipe_to_string
{
my $self = shift;
my $hOut = shift;
my $strBuffer;
my $hString = IO::String->new($strBuffer);
$self->binary_xfer($hOut, $hString);
return $strBuffer;
}
####################################################################################################################################
# ERROR_WRITE
####################################################################################################################################
@@ -167,12 +195,14 @@ sub error_write
}
else
{
$strMessage = 'unknown error object';
syswrite(*STDERR, 'unknown error object: ' . $oMessage);
exit 1;
}
}
else
{
$strMessage = $oMessage;
syswrite(*STDERR, $oMessage);
exit 1;
}
if (defined($strMessage))
@@ -186,6 +216,44 @@ sub error_write
}
}
####################################################################################################################################
# READ_LINE
####################################################################################################################################
sub read_line
{
my $self = shift;
my $hIn = shift;
my $bError = shift;
my $strLine;
my $strChar;
my $iByteIn;
while (1)
{
$iByteIn = sysread($hIn, $strChar, 1);
if (!defined($iByteIn) || $iByteIn != 1)
{
if (defined($bError) and !$bError)
{
return undef;
}
confess &log(ERROR, "unable to read 1 byte" . (defined($!) ? ": " . $! : ""));
}
if ($strChar eq "\n")
{
last;
}
$strLine .= $strChar;
}
return $strLine;
}
####################################################################################################################################
# BINARY_XFER
#
@@ -199,31 +267,45 @@ sub binary_xfer
my $strRemote = shift;
my $bCompress = shift;
my $bDone = false;
my $iBlockSize = BLOCK_SIZE;
$strRemote = defined($strRemote) ? $strRemote : 'none';
my $iBlockSize = $self->{iBlockSize};
my $iHeaderSize = 12;
my $iBlockIn;
my $iBlockOut;
my $strBlockHeader;
my $strBlock;
print "got to begin\n";
while (!$bDone)
while (1)
{
if ($strRemote eq 'in')
{
my $strBlockHeader = readline($hIn);
$strBlockHeader = $self->read_line($hIn);
if ($strBlockHeader !~ /^block [0-9]+$/)
{
confess "unable to read block header ${strBlockHeader}";
}
$iBlockSize = substr($strBlockHeader, index($strBlockHeader, " ") + 1);
$iBlockIn = sysread($hIn, $strBlock, $iBlockSize);
$iBlockSize = trim(substr($strBlockHeader, index($strBlockHeader, " ") + 1));
if (!defined($iBlockIn) || $iBlockIn != $iBlockSize)
if ($iBlockSize != 0)
{
confess "unable to read ${iBlockSize} bytes from remote" . (defined($!) ? ": " . $! : "");
$iBlockIn = sysread($hIn, $strBlock, $iBlockSize);
if (!defined($iBlockIn) || $iBlockIn != $iBlockSize)
{
confess "unable to read ${iBlockSize} bytes from remote" . (defined($!) ? ": " . $! : "");
}
}
else
{
#
# if ($strRemote eq 'in')
# {
# confess &log(ERROR, "block size $iBlockSize");
# }
$iBlockIn = 0;
}
}
else
@@ -239,20 +321,26 @@ sub binary_xfer
}
}
if ($strRemote eq 'out')
{
$strBlockHeader = "block ${iBlockIn}\n";
# print "wrote block header: ${strBlockHeader}"; # REMOVE!
$iBlockOut = syswrite($hOut, $strBlockHeader);
if (!defined($iBlockOut) || $iBlockOut != length($strBlockHeader))
{
confess "unable to write block header";
}
}
if ($iBlockIn > 0)
{
if ($strRemote eq 'out')
{
print "wrote block header\n";
if (!syswrite($hOut, "block ${iBlockIn}"))
{
confess "unable to write block header";
}
}
# Write to the output handle
my $iBlockOut = syswrite($hOut, $strBlock, $iBlockIn);
# print "writing: ${strBlock}\n"; # REMOVE!
$iBlockOut = syswrite($hOut, $strBlock, $iBlockIn);
if (!defined($iBlockOut) || $iBlockOut != $iBlockIn)
{
@@ -261,7 +349,7 @@ sub binary_xfer
}
else
{
$bDone = true;
last;
}
}
}
@@ -272,6 +360,8 @@ sub binary_xfer
sub output_read
{
my $self = shift;
my $bOutputRequired = shift;
my $strErrorPrefix = shift;
# &log(TRACE, "Remote->output_read");
@@ -279,27 +369,100 @@ sub output_read
my $strOutput;
my $bError = false;
my $iErrorCode;
my $strError;
while ($strLine = readline($self->{hOut}))
if (waitpid($self->{pId}, WNOHANG) != 0)
{
print "process exited\n";
my $strError = $self->pipe_to_string($self->{hErr});
if (defined($strError))
{
$bError = true;
$strOutput = $strError;
}
# Capture any errors
if ($bError)
{
print "error: " . $strOutput->message();
confess &log(ERROR, (defined($strErrorPrefix) ? "${strErrorPrefix}" : "") .
(defined($strOutput) ? ": ${strOutput}" : ""));
}
}
# print "error read wait\n";
#
# if (!eof($self->{hErr}))
# {
# $strError = $self->pipe_to_string($self->{hErr});
#
# if (defined($strError))
# {
# $bError = true;
# $strOutput = $strError;
# $iErrorCode = undef;
# }
# }
print "output read wait\n";
while ($strLine = $self->read_line($self->{hOut}, false))
{
print "read a line ${strLine}\n";
if ($strLine =~ /^ERROR.*/)
{
$bError = true;
$iErrorCode = (split(' ', trim($strLine)))[1];
$iErrorCode = (split(' ', $strLine))[1];
last;
}
if (trim($strLine) =~ /^OK$/)
if ($strLine =~ /^OK$/)
{
print "found OK\n";
last;
}
$strOutput .= (defined($strOutput) ? "\n" : "") . trim(substr($strLine, 1));
$strOutput .= (defined($strOutput) ? "\n" : "") . substr($strLine, 1);
}
return ($strOutput, $bError, $iErrorCode);
print "and here\n";
# Capture any errors
if ($bError)
{
confess &log(ERROR, (defined($strErrorPrefix) ? "${strErrorPrefix}" : "") .
(defined($strOutput) ? ": ${strOutput}" : ""), $iErrorCode);
}
if ($bOutputRequired && !defined($strOutput))
{
my $strError = $self->pipe_to_string($self->{hErr});
if (defined($strError))
{
$bError = true;
$strOutput = $strError;
}
# Capture any errors
if ($bError)
{
print "error: " . $strOutput->message();
confess &log(ERROR, (defined($strErrorPrefix) ? "${strErrorPrefix}" : "") .
(defined($strOutput) ? ": ${strOutput}" : ""));
}
confess &log(ERROR, (defined($strErrorPrefix) ? "${strErrorPrefix}: " : "") . "output is not defined");
}
return $strOutput;
}
####################################################################################################################################
@@ -310,9 +473,17 @@ sub output_write
my $self = shift;
my $strOutput = shift;
$self->string_write(*STDOUT, $strOutput);
if (defined($strOutput))
{
$self->string_write(*STDOUT, "${strOutput}");
if (!syswrite(*STDOUT, "\nOK\n"))
if (!syswrite(*STDOUT, "\n"))
{
confess "unable to write output";
}
}
if (!syswrite(*STDOUT, "OK\n"))
{
confess "unable to write output";
}
@@ -350,9 +521,9 @@ sub command_read
my $strLine;
my $strCommand;
while ($strLine = readline(*STDIN))
while ($strLine = $self->read_line(*STDIN))
{
$strLine = trim($strLine);
# $strLine = trim($strLine);
if (!defined($strCommand))
{
@@ -432,7 +603,7 @@ sub command_write
$strOutput .= "end";
}
&log(TRACE, "Remote->command_write:\n" . trim($strOutput));
&log(TRACE, "Remote->command_write:\n" . $strOutput);
if (!syswrite($self->{hIn}, "${strOutput}\n"))
{
@@ -448,20 +619,12 @@ sub command_execute
my $self = shift;
my $strCommand = shift;
my $oParamRef = shift;
my $bOutputRequired = shift;
my $strErrorPrefix = shift;
$self->command_write($strCommand, $oParamRef);
my ($strOutput, $bError, $iErrorCode) = $self->output_read();
# Capture any errors
if ($bError)
{
confess &log(ERROR, (defined($strErrorPrefix) ? "${strErrorPrefix}" : "") .
(defined($strOutput) ? ": ${strOutput}" : ""), $iErrorCode);
}
return $strOutput;
return $self->output_read($bOutputRequired, $strErrorPrefix);
}
no Moose;

8
test/lib/BackRestTest/FileTest.pm
View File

@@ -807,15 +807,11 @@ sub BackRestFileTest
my $oFile = BackRest::File->new
(
strStanza => "db",
strCommand => $strCommand,
bCompress => $bDestinationCompressed,
strRemote => $bBackupRemote ? 'backup' : $bDbRemote ? 'db' : undef,
strBackupClusterPath => undef,
strBackupPath => ${strTestPath},
strBackupHost => $bBackupRemote ? $strHost : undef,
strBackupUser => $bBackupRemote ? $strUser : undef,
strDbHost => $bDbRemote ? $strHost : undef,
strDbUser => $bDbRemote ? $strUser : undef
strRemote => $bBackupRemote ? 'backup' : $bDbRemote ? 'db' : undef,
oRemote => $bBackupRemote || $bDbRemote ? $oRemote : undef
);
for (my $bSourceCompressed = 0; $bSourceCompressed <= 0; $bSourceCompressed++)