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