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