PVE/RADOS.pm

   1 package PVE::RADOS;
   2
   3 use 5.014002;
   4 use strict;
   5 use warnings;
   6 use Carp;
   7 use JSON;
   8 use Socket;
   9 use PVE::Tools;
  10 use PVE::INotify;
  11 use PVE::RPCEnvironment;
  12
  13 require Exporter;
  14
  15 my $rados_default_timeout = 5;
  16
  17
  18 our @ISA = qw(Exporter);
  19
  20 # Items to export into callers namespace by default. Note: do not export
  21 # names by default without a very good reason. Use EXPORT_OK instead.
  22 # Do not simply export all your public functions/methods/constants.
  23
  24 # This allows declaration       use PVE::RADOS ':all';
  25 # If you do not need this, moving things directly into @EXPORT or @EXPORT_OK
  26 # will save memory.
  27 our %EXPORT_TAGS = ( 'all' => [ qw(
  28
  29 ) ] );
  30
  31 our @EXPORT_OK = ( @{ $EXPORT_TAGS{'all'} } );
  32
  33 our @EXPORT = qw(
  34
  35 );
  36
  37 our $VERSION = '1.0';
  38
  39 require XSLoader;
  40 XSLoader::load('PVE::RADOS', $VERSION);
  41
  42 my $writedata = sub {
  43     my ($fh, $cmd, $data) = @_;
  44
  45     local $SIG{PIPE} = 'IGNORE';
  46
  47     my $bin = pack "a L/a*", $cmd, $data || '';
  48     my $res = syswrite $fh, $bin;
  49
  50     die "write data failed - $!\n" if !defined($res);
  51 };
  52
  53 my $readdata = sub {
  54     my ($fh, $allow_eof) = @_;
  55
  56     my $head = '';
  57
  58     local $SIG{PIPE} = 'IGNORE';
  59
  60     while (length($head) < 5) {
  61         last if !sysread $fh, $head, 5 - length($head), length($head);
  62     }
  63     return undef if $allow_eof && length($head) == 0;
  64
  65     die "partial read\n" if length($head) < 5;
  66
  67     my ($cmd, $len) = unpack "a L", $head;
  68
  69     my $data = '';
  70     while (length($data) < $len) {
  71         last if !sysread $fh, $data, $len - length($data), length($data);
  72     }
  73     die "partial data read\n" if length($data) < $len;
  74
  75     return wantarray ? ($cmd, $data) : $data;
  76 };
  77
  78 my $kill_worker = sub {
  79     my ($self) = @_;
  80
  81     return if !$self->{cpid};
  82     return if  $self->{__already_killed};
  83
  84     $self->{__already_killed} = 1;
  85
  86     close($self->{child}) if defined($self->{child});
  87
  88     # only kill if we created the process
  89     return if $self->{pid} != $$;
  90
  91     kill(9, $self->{cpid});
  92     waitpid($self->{cpid}, 0);
  93 };
  94
  95 my $sendcmd = sub {
  96     my ($self, $cmd, $data, $expect_tag) = @_;
  97
  98     $expect_tag = '>' if !$expect_tag;
  99
 100     die "detected forked connection" if $self->{pid} != $$;
 101
 102     my ($restag, $raw);
 103     my $code = sub {
 104         &$writedata($self->{child}, $cmd, $data) if $expect_tag ne 'S';
 105         ($restag, $raw) = &$readdata($self->{child});
 106     };
 107     eval { PVE::Tools::run_with_timeout($self->{timeout}, $code); };
 108     if (my $err = $@) {
 109         &$kill_worker($self);
 110         die $err;
 111     }
 112     if ($restag eq 'E') {
 113         die $raw if $raw;
 114         die "unknown error\n";
 115     }
 116
 117     die "got unexpected result\n" if $restag ne $expect_tag;
 118
 119     return $raw;
 120 };
 121
 122 sub new {
 123     my ($class, %params) = @_;
 124
 125     my $rpcenv = PVE::RPCEnvironment::get();
 126
 127     socketpair(my $child, my $parent, AF_UNIX, SOCK_STREAM, PF_UNSPEC)
 128         ||  die "socketpair: $!";
 129
 130     my $cpid = fork();
 131
 132     die "unable to fork - $!\n" if !defined($cpid);
 133
 134     my $self = bless {};
 135
 136     my $timeout = delete $params{timeout} || $rados_default_timeout;
 137
 138     $self->{timeout} = $timeout;
 139     $self->{pid} = $$;
 140
 141     if ($cpid) { # parent
 142         close $parent;
 143
 144         $self->{cpid} = $cpid;
 145         $self->{child} = $child;
 146
 147         &$sendcmd($self, undef, undef, 'S'); # wait for sync
 148
 149     } else { # child
 150         $0 = 'pverados';
 151
 152         PVE::INotify::inotify_close();
 153
 154         if (my $atfork = $rpcenv->{atfork}) {
 155             &$atfork();
 156         }
 157
 158         # fixme: timeout?
 159
 160         close $child;
 161
 162         my $conn;
 163         eval {
 164             $conn = pve_rados_create() ||
 165                 die "unable to create RADOS object\n";
 166
 167             pve_rados_conf_set($conn, 'client_mount_timeout', $timeout);
 168
 169             foreach my $k (keys %params) {
 170                 pve_rados_conf_set($conn, $k, $params{$k});
 171             }
 172
 173             pve_rados_connect($conn);
 174         };
 175         if (my $err = $@) {
 176             &$writedata($parent, 'E', $err);
 177             die $err;
 178         }
 179         &$writedata($parent, 'S');
 180
 181         $self->{conn} = $conn;
 182
 183         for (;;) {
 184             my ($cmd, $data) = &$readdata($parent, 1);
 185
 186             last if !$cmd || $cmd eq 'Q';
 187
 188             my $res;
 189             eval {
 190                 if ($cmd eq 'M') { # rados monitor commands
 191                     $res = pve_rados_mon_command($self->{conn}, [ $data ]);
 192                 } elsif ($cmd eq 'C') { # class methods
 193                     my $aref = decode_json($data);
 194                     my $method = shift @$aref;
 195                     $res = encode_json($self->$method(@$aref));
 196                 } else {
 197                     die "invalid command\n";
 198                 }
 199             };
 200             if (my $err = $@) {
 201                 &$writedata($parent, 'E', $err);
 202                 die $err;
 203             }
 204             &$writedata($parent, '>', $res);
 205         }
 206
 207         exit(0);
 208     }
 209
 210     return $self;
 211 }
 212
 213 sub timeout {
 214     my ($self, $new_timeout) = @_;
 215
 216     $self->{timeout} = $new_timeout if $new_timeout;
 217
 218     return $self->{timeout};
 219 }
 220
 221 sub DESTROY {
 222     my ($self) = @_;
 223
 224     if ($self->{cpid}) {
 225         #print "$$: DESTROY WAIT0\n";
 226         &$kill_worker($self);
 227         #print "$$: DESTROY WAIT\n";
 228     } else {
 229         #print "$$: DESTROY SHUTDOWN\n";
 230         pve_rados_shutdown($self->{conn}) if $self->{conn};
 231     }
 232 }
 233
 234 sub cluster_stat {
 235     my ($self, @args) = @_;
 236
 237     if ($self->{cpid}) {
 238         my $data = encode_json(['cluster_stat', @args]);
 239         my $raw = &$sendcmd($self, 'C', $data);
 240         return decode_json($raw);
 241     } else {
 242         return pve_rados_cluster_stat($self->{conn});
 243     }
 244 }
 245
 246 # example1: { prefix => 'get_command_descriptions'})
 247 # example2: { prefix => 'mon dump', format => 'json' }
 248 sub mon_command {
 249     my ($self, $cmd) = @_;
 250
 251     $cmd->{format} = 'json' if !$cmd->{format};
 252
 253     my $json = encode_json($cmd);
 254
 255     my $raw = &$sendcmd($self, 'M', $json);
 256
 257     if ($cmd->{format} && $cmd->{format} eq 'json') {
 258         return length($raw) ? decode_json($raw) : undef;
 259     }
 260     return $raw;
 261 }
 262
 263
 264 1;
 265 __END__
 266
 267 =head1 NAME
 268
 269 PVE::RADOS - Perl bindings for librados
 270
 271 =head1 SYNOPSIS
 272
 273   use PVE::RADOS;
 274
 275   my $rados = PVE::RADOS::new();
 276   my $stat = $rados->cluster_stat();
 277   my $res = $rados->mon_command({ prefix => 'mon dump', format => 'json' });
 278
 279 =head1 DESCRIPTION
 280
 281 Perl bindings for librados.
 282
 283 =head2 EXPORT
 284
 285 None by default.
 286
 287 =head1 AUTHOR
 288
 289 Dietmar Maurer, E<lt>dietmar@proxmox.com<gt>
 290
 291 =cut