PVE/QMPClient.pm

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