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