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) = @_;
  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
  38     $mux->set_callback_object($self);
  39
  40     # make sure perl doesn't believe this is a circular reference as we
  41     # delete mux in DESTROY
  42     weaken($mux->{_object});
  43
  44     return $self;
  45 }
  46
  47 # add a single command to the queue for later execution
  48 # with queue_execute()
  49 sub queue_cmd {
  50     my ($self, $vmid, $callback, $execute, %params) = @_;
  51
  52     my $cmd = {};
  53     $cmd->{execute} = $execute;
  54     $cmd->{arguments} = \%params;
  55     $cmd->{callback} = $callback;
  56
  57     push @{$self->{queue}->{$vmid}}, $cmd;
  58 }
  59
  60 # execute a single command
  61 sub cmd {
  62     my ($self, $vmid, $cmd, $timeout) = @_;
  63
  64     my $result;
  65
  66     my $callback = sub {
  67         my ($vmid, $resp) = @_;
  68         $result = $resp->{'return'};
  69     };
  70
  71     die "no command specified" if !($cmd &&  $cmd->{execute});
  72
  73     $cmd->{callback} = $callback;
  74     $cmd->{arguments} = {} if !defined($cmd->{arguments});
  75
  76     $self->{queue}->{$vmid} = [ $cmd ];
  77
  78     if (!$timeout) {
  79         # hack: monitor sometime blocks
  80         if ($cmd->{execute} eq 'query-migrate') {
  81             $timeout = 60*60; # 1 hour
  82         } elsif ($cmd->{execute} =~ m/^(eject|change)/) {
  83             $timeout = 60; # note: cdrom mount command is slow
  84         } elsif ($cmd->{execute} eq 'savevm-start' ||
  85                  $cmd->{execute} eq 'savevm-end' ||
  86                  $cmd->{execute} eq 'query-backup' ||
  87                  $cmd->{execute} eq 'query-block-jobs' ||
  88                  $cmd->{execute} eq 'backup-cancel' ||
  89                  $cmd->{execute} eq 'query-savevm' ||
  90                  $cmd->{execute} eq 'delete-drive-snapshot' ||
  91                  $cmd->{execute} eq 'guest-shutdown' ||
  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 $cmdid_seq_qga = 0;
 110 my $next_cmdid = sub {
 111     my ($qga) = @_;
 112
 113     if($qga){
 114         $cmdid_seq_qga++;
 115         return "$$"."0".$cmdid_seq_qga;
 116     } else {
 117         $cmdid_seq++;
 118         return "$$:$cmdid_seq";
 119     }
 120 };
 121
 122 my $close_connection = sub {
 123     my ($self, $vmid) = @_;
 124
 125     my $fh = $self->{fhs}->{$vmid};
 126     return if !$fh;
 127
 128     delete $self->{fhs}->{$vmid};
 129     delete $self->{fhs_lookup}->{$fh};
 130
 131     $self->{mux}->close($fh);
 132 };
 133
 134 my $open_connection = sub {
 135     my ($self, $vmid, $timeout, $qga) = @_;
 136
 137     my $sname = PVE::QemuServer::qmp_socket($vmid, $qga);
 138
 139     $timeout = 1 if !$timeout;
 140
 141     my $fh;
 142     my $starttime = [gettimeofday];
 143     my $count = 0;
 144     for (;;) {
 145         $count++;
 146         $fh = IO::Socket::UNIX->new(Peer => $sname, Blocking => 0, Timeout => 1);
 147         last if $fh;
 148         if ($! != EINTR && $! != EAGAIN) {
 149             die "unable to connect to VM $vmid socket - $!\n";
 150         }
 151         my $elapsed = tv_interval($starttime, [gettimeofday]);
 152         if ($elapsed >= $timeout) {
 153             die "unable to connect to VM $vmid socket - timeout after $count retries\n";
 154         }
 155         usleep(100000);
 156     }
 157
 158     $self->{fhs}->{$vmid} = $fh;
 159     $self->{fhs_lookup}->{$fh} = $vmid;
 160     $self->{mux}->add($fh);
 161
 162     return $fh;
 163 };
 164
 165 my $check_queue = sub {
 166     my ($self) = @_;
 167
 168     my $running = 0;
 169
 170     foreach my $vmid (keys %{$self->{queue}}) {
 171         my $fh = $self->{fhs}->{$vmid};
 172         next if !$fh;
 173
 174         if ($self->{errors}->{$vmid}) {
 175             &$close_connection($self, $vmid);
 176             next;
 177         }
 178
 179         if ($self->{current}->{$vmid}) { # command running, waiting for response
 180             $running++;
 181             next;
 182         }
 183
 184         if (!scalar(@{$self->{queue}->{$vmid}})) { # no more commands for the VM
 185             &$close_connection($self, $vmid);
 186             next;
 187         }
 188
 189         eval {
 190
 191             my $cmd = $self->{current}->{$vmid} = shift @{$self->{queue}->{$vmid}};
 192             $cmd->{id} = &$next_cmdid($cmd->{qga});
 193
 194             my $fd = -1;
 195             if ($cmd->{execute} eq 'add-fd' || $cmd->{execute} eq 'getfd') {
 196                 $fd = $cmd->{arguments}->{fd};
 197                 delete $cmd->{arguments}->{fd};
 198             }
 199
 200             my $qmpcmd = undef;
 201
 202             if($self->{current}->{$vmid}->{qga}){
 203
 204                 my $qmpcmdid =to_json({
 205                     execute => 'guest-sync',
 206                     arguments => { id => int($cmd->{id})}});
 207
 208                 $qmpcmd = to_json({
 209                     execute => $cmd->{execute},
 210                     arguments => $cmd->{arguments}});
 211
 212                 $qmpcmd = $qmpcmdid.$qmpcmd;
 213
 214             }else{
 215
 216                 $qmpcmd = to_json({
 217                     execute => $cmd->{execute},
 218                     arguments => $cmd->{arguments},
 219                     id => $cmd->{id}});
 220             }
 221
 222             if ($fd >= 0) {
 223                 my $ret = PVE::IPCC::sendfd(fileno($fh), $fd, $qmpcmd);
 224                 die "sendfd failed" if $ret < 0;
 225             } else {
 226                 $self->{mux}->write($fh, $qmpcmd);
 227             }
 228         };
 229         if (my $err = $@) {
 230             $self->{errors}->{$vmid} = $err;
 231         } else {
 232             $running++;
 233         }
 234     }
 235
 236     $self->{mux}->endloop() if !$running;
 237
 238     return $running;
 239 };
 240
 241 # execute all queued command
 242 sub queue_execute {
 243     my ($self, $timeout) = @_;
 244
 245     $timeout = 3 if !$timeout;
 246
 247     $self->{current} = {};
 248     $self->{errors} = {};
 249
 250     # open all necessary connections
 251     foreach my $vmid (keys %{$self->{queue}}) {
 252         next if !scalar(@{$self->{queue}->{$vmid}}); # no commands for the VM
 253
 254         if ($self->{queue}->{$vmid}[0]->{execute} =~ /^guest\-+/){
 255             $self->{queue}->{$vmid}[0]->{qga} = "1";
 256         }
 257
 258         eval {
 259             my $fh = &$open_connection($self, $vmid, $timeout, $self->{queue}->{$vmid}[0]->{qga});
 260
 261             if(!$self->{queue}->{$vmid}[0]->{qga}){
 262                 my $cmd = { execute => 'qmp_capabilities', arguments => {} };
 263                 unshift @{$self->{queue}->{$vmid}}, $cmd;
 264             }
 265             $self->{mux}->set_timeout($fh, $timeout);
 266         };
 267         if (my $err = $@) {
 268             warn $err;
 269             $self->{errors}->{$vmid} = $err;
 270         }
 271     }
 272
 273     my $running;
 274
 275     for (;;) {
 276
 277         $running = &$check_queue($self);
 278
 279         last if !$running;
 280
 281         $self->{mux}->loop;
 282     }
 283
 284     # make sure we close everything
 285     foreach my $vmid (keys %{$self->{fhs}}) {
 286         &$close_connection($self, $vmid);
 287     }
 288
 289     $self->{queue} = $self->{current} = $self->{fhs} = $self->{fhs_lookup} = {};
 290 }
 291
 292 sub mux_close {
 293     my ($self, $mux, $fh) = @_;
 294
 295     my $vmid = $self->{fhs_lookup}->{$fh} || 'undef';
 296     return if !defined($vmid);
 297
 298     if(!$self->{no_answer}){
 299         $self->{errors}->{$vmid} = "client closed connection\n" if !$self->{errors}->{$vmid};
 300     } else {
 301         delete $self->{no_anwser};
 302     }
 303 }
 304
 305 # mux_input is called when input is available on one of
 306 # the descriptors.
 307 sub mux_input {
 308     my ($self, $mux, $fh, $input) = @_;
 309
 310     my $vmid = $self->{fhs_lookup}->{$fh};
 311     if (!$vmid) {
 312         warn "internal error - unable to lookup vmid";
 313         return;
 314     }
 315
 316     my $curcmd = $self->{current}->{$vmid};
 317     die "unable to lookup current command for VM $vmid\n" if !$curcmd;
 318
 319     my $raw;
 320
 321     if ($curcmd->{qga}) {
 322         return if $$input !~ s/^([^\n]+}\n[^\n]+})\n(.*)$/$2/so;
 323         $raw = $1;
 324     } else {
 325         return if $$input !~ s/^([^\n]+})\r?\n(.*)$/$2/so;
 326         $raw = $1;
 327     }
 328
 329     eval {
 330         my @jsons = split("\n", $raw);
 331
 332         if ($curcmd->{qga}) {
 333
 334             die "response is not complete" if @jsons != 2 ;
 335
 336             my $obj = from_json($jsons[0]);
 337             my $cmdid = $obj->{return};
 338             die "received responsed without command id\n" if !$cmdid;
 339
 340             delete $self->{current}->{$vmid};
 341
 342             if ($curcmd->{id} ne $cmdid) {
 343                 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
 344             }
 345
 346             $obj = from_json($jsons[1]);
 347
 348             if (my $callback = $curcmd->{callback}) {
 349                 &$callback($vmid, $obj);
 350             }
 351
 352             return;
 353         }
 354
 355         foreach my $json (@jsons) {
 356             my $obj = from_json($json);
 357             next if defined($obj->{QMP}); # skip monitor greeting
 358
 359             if (exists($obj->{error}->{desc})) {
 360                 my $desc = $obj->{error}->{desc};
 361                 chomp $desc;
 362                 die "$desc\n" if $desc !~ m/Connection can not be completed immediately/;
 363                 next;
 364             }
 365
 366             if (defined($obj->{event})) {
 367                 if (my $eventcb = $self->{eventcb}) {
 368                     &$eventcb($obj);
 369                 }
 370                 next;
 371             }
 372
 373             my $cmdid = $obj->{id};
 374             die "received responsed without command id\n" if !$cmdid;
 375
 376             my $curcmd = $self->{current}->{$vmid};
 377             die "unable to lookup current command for VM $vmid\n" if !$curcmd;
 378
 379             delete $self->{current}->{$vmid};
 380
 381             if ($curcmd->{id} ne $cmdid) {
 382                 die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
 383             }
 384
 385             if (my $callback = $curcmd->{callback}) {
 386                 &$callback($vmid, $obj);
 387             }
 388         }
 389     };
 390     if (my $err = $@) {
 391         $self->{errors}->{$vmid} = $err;
 392     }
 393
 394     &$check_queue($self);
 395 }
 396
 397 # This gets called every second to update player info, etc...
 398 sub mux_timeout {
 399     my ($self, $mux, $fh) = @_;
 400
 401     if (my $vmid = $self->{fhs_lookup}->{$fh}) {
 402         $self->{errors}->{$vmid} = "got timeout\n";
 403     }
 404
 405     &$check_queue($self);
 406 }
 407
 408 sub mux_eof {
 409     my ($self, $mux, $fh, $input) = @_;
 410
 411     my $vmid = $self->{fhs_lookup}->{$fh};
 412     if(check_no_answer($self->{current}->{$vmid}->{execute})){
 413         my @jsons = split("\n", $$input);
 414
 415         my $obj = from_json($jsons[0]);
 416
 417         my $cmdid = $obj->{return};
 418         die "received responsed without command id\n" if !$cmdid;
 419
 420         my $curcmd = $self->{current}->{$vmid};
 421         die "unable to lookup current command for VM $vmid\n" if !$curcmd;
 422
 423         delete $self->{current}->{$vmid};
 424
 425         $self->{no_answer} = 1;
 426       }
 427 }
 428
 429 sub check_no_answer {
 430     my($cmd) = @_;
 431
 432     if ($cmd eq 'guest-shutdown'){
 433         return 1;
 434     } elsif ($cmd eq 'guest-suspend-ram'){
 435         return 1;
 436     } elsif ($cmd eq 'guest-suspend-disk'){
 437         return 1;
 438     } elsif ($cmd eq 'guest-suspend-hybrid'){
 439         return 1;
 440     }
 441     return 0;
 442 }
 443
 444 1;