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 'query-block-jobs' ||
  87                  $cmd->{execute} eq 'backup-cancel' ||
  88                  $cmd->{execute} eq 'query-savevm' ||
  89                  $cmd->{execute} eq 'delete-drive-snapshot' ||
  90                  $cmd->{execute} eq 'snapshot-drive'  ) {
  91             $timeout = 10*60; # 10 mins ?
  92         } else {
  93             $timeout = 3; # default
  94         }
  95     }
  96
  97     $self->queue_execute($timeout);
  98
  99     my $cmdstr = $cmd->{execute} || '';
 100     die "VM $vmid qmp command '$cmdstr' failed - $self->{errors}->{$vmid}"
 101         if defined($self->{errors}->{$vmid});
 102
 103     return $result;
 104 };
 105
 106 my $cmdid_seq = 0;
 107 my $next_cmdid = sub {
 108     $cmdid_seq++;
 109     return "$$:$cmdid_seq";
 110 };
 111
 112 my $close_connection = sub {
 113     my ($self, $vmid) = @_;
 114
 115     my $fh = $self->{fhs}->{$vmid};
 116     return if !$fh;
 117
 118     delete $self->{fhs}->{$vmid};
 119     delete $self->{fhs_lookup}->{$fh};
 120
 121     $self->{mux}->close($fh);
 122 };
 123
 124 my $open_connection = sub {
 125     my ($self, $vmid, $timeout) = @_;
 126
 127     my $sname = PVE::QemuServer::qmp_socket($vmid);
 128
 129     $timeout = 1 if !$timeout;
 130
 131     my $fh;
 132     my $starttime = [gettimeofday];
 133     my $count = 0;
 134     for (;;) {
 135         $count++;
 136         $fh = IO::Socket::UNIX->new(Peer => $sname, Blocking => 0, Timeout => 1);
 137         last if $fh;
 138         if ($! != EINTR && $! != EAGAIN) {
 139             die "unable to connect to VM $vmid socket - $!\n";
 140         }
 141         my $elapsed = tv_interval($starttime, [gettimeofday]);
 142         if ($elapsed >= $timeout) {
 143             die "unable to connect to VM $vmid socket - timeout after $count retries\n";
 144         }
 145         usleep(100000);
 146     }
 147
 148     $self->{fhs}->{$vmid} = $fh;
 149     $self->{fhs_lookup}->{$fh} = $vmid;
 150     $self->{mux}->add($fh);
 151
 152     return $fh;
 153 };
 154
 155 my $check_queue = sub {
 156     my ($self) = @_;
 157
 158     my $running = 0;
 159
 160     foreach my $vmid (keys %{$self->{queue}}) {
 161         my $fh = $self->{fhs}->{$vmid};
 162         next if !$fh;
 163
 164         if ($self->{errors}->{$vmid}) {
 165             &$close_connection($self, $vmid);
 166             next;
 167         }
 168
 169         if ($self->{current}->{$vmid}) { # command running, waiting for response
 170             $running++;
 171             next;
 172         }
 173
 174         if (!scalar(@{$self->{queue}->{$vmid}})) { # no more commands for the VM
 175             &$close_connection($self, $vmid);
 176             next;
 177         }
 178
 179         eval {
 180
 181             my $cmd = $self->{current}->{$vmid} = shift @{$self->{queue}->{$vmid}};
 182             $cmd->{id} = &$next_cmdid();
 183
 184             my $fd = -1;
 185             if ($cmd->{execute} eq 'add-fd' || $cmd->{execute} eq 'getfd') {
 186                 $fd = $cmd->{arguments}->{fd};
 187                 delete $cmd->{arguments}->{fd};
 188             }
 189
 190             my $qmpcmd = to_json({
 191                 execute => $cmd->{execute},
 192                 arguments => $cmd->{arguments},
 193                 id => $cmd->{id}});
 194
 195             if ($fd >= 0) {
 196                 my $ret = PVE::IPCC::sendfd(fileno($fh), $fd, $qmpcmd);
 197                 die "sendfd failed" if $ret < 0;
 198             } else {
 199                 $self->{mux}->write($fh, $qmpcmd);
 200             }
 201         };
 202         if (my $err = $@) {
 203             $self->{errors}->{$vmid} = $err;
 204         } else {
 205             $running++;
 206         }
 207     }
 208
 209     $self->{mux}->endloop() if !$running;
 210
 211     return $running;
 212 };
 213
 214 # execute all queued command
 215 sub queue_execute {
 216     my ($self, $timeout) = @_;
 217
 218     $timeout = 3 if !$timeout;
 219
 220     $self->{current} = {};
 221     $self->{errors} = {};
 222
 223     # open all necessary connections
 224     foreach my $vmid (keys %{$self->{queue}}) {
 225         next if !scalar(@{$self->{queue}->{$vmid}}); # no commands for the VM
 226
 227         eval {
 228             my $fh = &$open_connection($self, $vmid, $timeout);
 229             my $cmd = { execute => 'qmp_capabilities', arguments => {} };
 230             unshift @{$self->{queue}->{$vmid}}, $cmd;
 231             $self->{mux}->set_timeout($fh, $timeout);
 232         };
 233         if (my $err = $@) {
 234             warn $err;
 235             $self->{errors}->{$vmid} = $err;
 236         }
 237     }
 238
 239     my $running;
 240
 241     for (;;) {
 242
 243         $running = &$check_queue($self);
 244
 245         last if !$running;
 246
 247         $self->{mux}->loop;
 248     }
 249
 250     # make sure we close everything
 251     foreach my $vmid (keys %{$self->{fhs}}) {
 252         &$close_connection($self, $vmid);
 253     }
 254
 255     $self->{queue} = $self->{current} = $self->{fhs} = $self->{fhs_lookup} = {};
 256 }
 257
 258 sub mux_close {
 259     my ($self, $mux, $fh) = @_;
 260
 261     my $vmid = $self->{fhs_lookup}->{$fh} || 'undef';
 262     return if !defined($vmid);
 263
 264     $self->{errors}->{$vmid} = "client closed connection\n" if !$self->{errors}->{$vmid};
 265 }
 266
 267 # mux_input is called when input is available on one of
 268 # the descriptors.
 269 sub mux_input {
 270     my ($self, $mux, $fh, $input) = @_;
 271
 272     return if $$input !~ s/^(.*})\r\n(.*)$/$2/so;
 273
 274     my $raw = $1;
 275
 276     my $vmid = $self->{fhs_lookup}->{$fh};
 277     if (!$vmid) {
 278         warn "internal error - unable to lookup vmid";
 279         return;
 280     }
 281
 282     eval {
 283         my @jsons = split("\n", $raw);
 284
 285         foreach my $json (@jsons) {
 286             my $obj = from_json($json);
 287             next if defined($obj->{QMP}); # skip monitor greeting
 288
 289             if (exists($obj->{error}->{desc})) {
 290                 my $desc = $obj->{error}->{desc};
 291                 chomp $desc;
 292                 die "$desc\n" if $desc !~ m/Connection can not be completed immediately/;
 293                 next;
 294             }
 295
 296             if (defined($obj->{event})) {
 297                 if (my $eventcb = $self->{eventcb}) {
 298                     &$eventcb($obj);
 299                 }
 300                 next;
 301             }
 302
 303             my $cmdid = $obj->{id};
 304             die "received responsed without command id\n" if !$cmdid;
 305
 306             my $curcmd = $self->{current}->{$vmid};
 307             die "unable to lookup current command for VM $vmid\n" if !$curcmd;
 308
 309             delete $self->{current}->{$vmid};
 310
 311             if ($curcmd->{id} ne $cmdid) {
 312                 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
 313             }
 314
 315             if (my $callback = $curcmd->{callback}) {
 316                 &$callback($vmid, $obj);
 317             }
 318         }
 319     };
 320     if (my $err = $@) {
 321         $self->{errors}->{$vmid} = $err;
 322     }
 323
 324     &$check_queue($self);
 325 }
 326
 327 # This gets called every second to update player info, etc...
 328 sub mux_timeout {
 329     my ($self, $mux, $fh) = @_;
 330
 331     if (my $vmid = $self->{fhs_lookup}->{$fh}) {
 332         $self->{errors}->{$vmid} = "got timeout\n";
 333     }
 334
 335     &$check_queue($self);
 336 }
 337
 338 1;