PVE/QMPClient.pm

   1 package PVE::QMPClient;
   2
   3 use strict;
   4 use PVE::QemuServer;
   5 use IO::Multiplex;
   6 use POSIX qw(EINTR EAGAIN);
   7 use JSON;
   8 use Time::HiRes qw(usleep gettimeofday tv_interval);
   9 use Scalar::Util qw(weaken);
  10 use PVE::IPCC;
  11
  12 use Data::Dumper;
  13
  14 # Qemu Monitor Protocol (QMP) client.
  15 #
  16 # This implementation uses IO::Multiplex (libio-multiplex-perl) and
  17 # allows you to issue qmp commands to different VMs in parallel.
  18
  19 # Note: kvm can onyl handle 1 connection, so we close connections asap
  20
  21 sub new {
  22     my ($class, $eventcb) = @_;
  23
  24     my $mux = new IO::Multiplex;
  25
  26     my $self = bless {
  27         mux => $mux,
  28         fhs => {}, # $vmid => fh
  29         fhs_lookup => {}, # $fh => $vmid
  30         queue => {},
  31         current => {},
  32         errors => {},
  33     }, $class;
  34
  35     $self->{eventcb} = $eventcb if $eventcb;
  36
  37     $mux->set_callback_object($self);
  38
  39     # make sure perl doesn't believe this is a circular reference as we
  40     # delete mux in DESTROY
  41     weaken($mux->{_object});
  42
  43     return $self;
  44 }
  45
  46 # add a single command to the queue for later execution
  47 # with queue_execute()
  48 sub queue_cmd {
  49     my ($self, $vmid, $callback, $execute, %params) = @_;
  50
  51     my $cmd = {};
  52     $cmd->{execute} = $execute;
  53     $cmd->{arguments} = \%params;
  54     $cmd->{callback} = $callback;
  55
  56     push @{$self->{queue}->{$vmid}}, $cmd;
  57 }
  58
  59 # execute a single command
  60 sub cmd {
  61     my ($self, $vmid, $cmd, $timeout) = @_;
  62
  63     my $result;
  64
  65     my $callback = sub {
  66         my ($vmid, $resp) = @_;
  67         $result = $resp->{'return'};
  68     };
  69
  70     die "no command specified" if !($cmd &&  $cmd->{execute});
  71
  72     $cmd->{callback} = $callback;
  73     $cmd->{arguments} = {} if !defined($cmd->{arguments});
  74
  75     $self->{queue}->{$vmid} = [ $cmd ];
  76
  77     if (!$timeout) {
  78         # hack: monitor sometime blocks
  79         if ($cmd->{execute} eq 'query-migrate') {
  80             $timeout = 60*60; # 1 hour
  81         } elsif ($cmd->{execute} =~ m/^(eject|change)/) {
  82             $timeout = 60; # note: cdrom mount command is slow
  83         } elsif ($cmd->{execute} eq 'savevm-start' ||
  84                  $cmd->{execute} eq 'savevm-end' ||
  85                  $cmd->{execute} eq 'query-backup' ||
  86                  $cmd->{execute} eq 'backup_cancel' ||
  87                  $cmd->{execute} eq 'query-savevm' ||
  88                  $cmd->{execute} eq 'delete-drive-snapshot' ||
  89                  $cmd->{execute} eq 'snapshot-drive'  ) {
  90             $timeout = 10*60; # 10 mins ?
  91         } else {
  92             $timeout = 3; # default
  93         }
  94     }
  95
  96     $self->queue_execute($timeout);
  97
  98     my $cmdstr = $cmd->{execute} || '';
  99     die "VM $vmid qmp command '$cmdstr' failed - $self->{errors}->{$vmid}"
 100         if defined($self->{errors}->{$vmid});
 101
 102     return $result;
 103 };
 104
 105 my $cmdid_seq = 0;
 106 my $next_cmdid = sub {
 107     $cmdid_seq++;
 108     return "$$:$cmdid_seq";
 109 };
 110
 111 my $close_connection = sub {
 112     my ($self, $vmid) = @_;
 113
 114     my $fh = $self->{fhs}->{$vmid};
 115     return if !$fh;
 116
 117     delete $self->{fhs}->{$vmid};
 118     delete $self->{fhs_lookup}->{$fh};
 119
 120     $self->{mux}->close($fh);
 121 };
 122
 123 my $open_connection = sub {
 124     my ($self, $vmid, $timeout) = @_;
 125
 126     my $sname = PVE::QemuServer::qmp_socket($vmid);
 127
 128     $timeout = 1 if !$timeout;
 129
 130     my $fh;
 131     my $starttime = [gettimeofday];
 132     my $count = 0;
 133     for (;;) {
 134         $count++;
 135         $fh = IO::Socket::UNIX->new(Peer => $sname, Blocking => 0, Timeout => 1);
 136         last if $fh;
 137         if ($! != EINTR && $! != EAGAIN) {
 138             die "unable to connect to VM $vmid socket - $!\n";
 139         }
 140         my $elapsed = tv_interval($starttime, [gettimeofday]);
 141         if ($elapsed >= $timeout) {
 142             die "unable to connect to VM $vmid socket - timeout after $count retries\n";
 143         }
 144         usleep(100000);
 145     }
 146
 147     $self->{fhs}->{$vmid} = $fh;
 148     $self->{fhs_lookup}->{$fh} = $vmid;
 149     $self->{mux}->add($fh);
 150
 151     return $fh;
 152 };
 153
 154 my $check_queue = sub {
 155     my ($self) = @_;
 156
 157     my $running = 0;
 158
 159     foreach my $vmid (keys %{$self->{queue}}) {
 160         my $fh = $self->{fhs}->{$vmid};
 161         next if !$fh;
 162
 163         if ($self->{errors}->{$vmid}) {
 164             &$close_connection($self, $vmid);
 165             next;
 166         }
 167
 168         if ($self->{current}->{$vmid}) { # command running, waiting for response
 169             $running++;
 170             next;
 171         }
 172
 173         if (!scalar(@{$self->{queue}->{$vmid}})) { # no more commands for the VM
 174             &$close_connection($self, $vmid);
 175             next;
 176         }
 177
 178         eval {
 179
 180             my $cmd = $self->{current}->{$vmid} = shift @{$self->{queue}->{$vmid}};
 181             $cmd->{id} = &$next_cmdid();
 182
 183             my $fd = -1;
 184             if ($cmd->{execute} eq 'add-fd' || $cmd->{execute} eq 'getfd') {
 185                 $fd = $cmd->{arguments}->{fd};
 186                 delete $cmd->{arguments}->{fd};
 187             }
 188
 189             my $qmpcmd = to_json({
 190                 execute => $cmd->{execute},
 191                 arguments => $cmd->{arguments},
 192                 id => $cmd->{id}});
 193
 194             if ($fd >= 0) {
 195                 my $ret = PVE::IPCC::sendfd(fileno($fh), $fd, $qmpcmd);
 196                 die "sendfd failed" if $ret < 0;
 197             } else {
 198                 $self->{mux}->write($fh, $qmpcmd);
 199             }
 200         };
 201         if (my $err = $@) {
 202             $self->{errors}->{$vmid} = $err;
 203         } else {
 204             $running++;
 205         }
 206     }
 207
 208     $self->{mux}->endloop() if !$running;
 209
 210     return $running;
 211 };
 212
 213 # execute all queued command
 214 sub queue_execute {
 215     my ($self, $timeout) = @_;
 216
 217     $timeout = 3 if !$timeout;
 218
 219     $self->{current} = {};
 220     $self->{errors} = {};
 221
 222     # open all necessary connections
 223     foreach my $vmid (keys %{$self->{queue}}) {
 224         next if !scalar(@{$self->{queue}->{$vmid}}); # no commands for the VM
 225
 226         eval {
 227             my $fh = &$open_connection($self, $vmid, $timeout);
 228             my $cmd = { execute => 'qmp_capabilities', arguments => {} };
 229             unshift @{$self->{queue}->{$vmid}}, $cmd;
 230             $self->{mux}->set_timeout($fh, $timeout);
 231         };
 232         if (my $err = $@) {
 233             warn $err;
 234             $self->{errors}->{$vmid} = $err;
 235         }
 236     }
 237
 238     my $running;
 239
 240     for (;;) {
 241
 242         $running = &$check_queue($self);
 243
 244         last if !$running;
 245
 246         $self->{mux}->loop;
 247     }
 248
 249     # make sure we close everything
 250     foreach my $vmid (keys %{$self->{fhs}}) {
 251         &$close_connection($self, $vmid);
 252     }
 253
 254     $self->{queue} = $self->{current} = $self->{fhs} = $self->{fhs_lookup} = {};
 255 }
 256
 257 # mux_input is called when input is available on one of
 258 # the descriptors.
 259 sub mux_input {
 260     my ($self, $mux, $fh, $input) = @_;
 261
 262     return if $$input !~ m/}\r\n$/;
 263
 264     my $raw = $$input;
 265
 266     # Remove the input from the input buffer.
 267     $$input = '';
 268
 269     my $vmid = $self->{fhs_lookup}->{$fh};
 270     if (!$vmid) {
 271         warn "internal error - unable to lookup vmid";
 272         return;
 273     }
 274
 275     eval {
 276         my @jsons = split("\n", $raw);
 277
 278         foreach my $json (@jsons) {
 279             my $obj = from_json($json);
 280             next if defined($obj->{QMP}); # skip monitor greeting
 281
 282             if (exists($obj->{error}->{desc})) {
 283                 my $desc = $obj->{error}->{desc};
 284                 chomp $desc;
 285                 die "$desc\n" if $desc !~ m/Connection can not be completed immediately/;
 286                 next;
 287             }
 288
 289             if (defined($obj->{event})) {
 290                 if (my $eventcb = $self->{eventcb}) {
 291                     &$eventcb($obj);
 292                 }
 293                 next;
 294             }
 295
 296             my $cmdid = $obj->{id};
 297             die "received responsed without command id\n" if !$cmdid;
 298
 299             my $curcmd = $self->{current}->{$vmid};
 300             die "unable to lookup current command for VM $vmid\n" if !$curcmd;
 301
 302             delete $self->{current}->{$vmid};
 303
 304             if ($curcmd->{id} ne $cmdid) {
 305                 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
 306             }
 307
 308             if (my $callback = $curcmd->{callback}) {
 309                 &$callback($vmid, $obj);
 310             }
 311         }
 312     };
 313     if (my $err = $@) {
 314         $self->{errors}->{$vmid} = $err;
 315     }
 316
 317     &$check_queue($self);
 318 }
 319
 320 # This gets called every second to update player info, etc...
 321 sub mux_timeout {
 322     my ($self, $mux, $fh) = @_;
 323
 324     if (my $vmid = $self->{fhs_lookup}->{$fh}) {
 325         $self->{errors}->{$vmid} = "got timeout\n";
 326     }
 327
 328     &$check_queue($self);
 329 }
 330
 331 1;