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