PVE/QMPClient.pm

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