X-Git-Url: https://git.proxmox.com/?a=blobdiff_plain;f=PVE%2FQMPClient.pm;h=ef5c19b3302f2f65b28f66651d7ac9e3c61765d2;hb=5674d19810e93fcd091afaa5f7b5a284f0000102;hp=a06b2e0d2fdd1c640481198094748304998c4e82;hpb=30a3378acdd2a6c40998a455c26437888fd98196;p=qemu-server.git

diff --git a/PVE/QMPClient.pm b/PVE/QMPClient.pm
old mode 100755
new mode 100644
index a06b2e0d..ef5c19b3
--- a/PVE/QMPClient.pm
+++ b/PVE/QMPClient.pm
@@ -1,43 +1,74 @@
-#!/usr/bin/perl -w
-
 package PVE::QMPClient;
 
 use strict;
-#use PVE::SafeSyslog;
-use PVE::QemuServer;
+use warnings;
+
 use IO::Multiplex;
 use JSON;
-use Data::Dumper;
+use POSIX qw(EINTR EAGAIN);
+use Scalar::Util qw(weaken);
+use Time::HiRes qw(usleep gettimeofday tv_interval);
 
-# Qemu Monitor Protocol (QMP) client.
+use PVE::IPCC;
+use PVE::QemuServer::Helpers;
+
+# QEMU Monitor Protocol (QMP) client.
 #
 # This implementation uses IO::Multiplex (libio-multiplex-perl) and
-# allows you to issue qmp commands to different VMs in parallel.
+# allows you to issue qmp and qga commands to different VMs in parallel.
 
-# Note: kvm can onyl handle 1 connection, so we close connections asap
+# Note: qemu can only handle 1 connection, so we close connections asap
 
 sub new {
     my ($class, $eventcb) = @_;
 
-    my $mux = new IO::Multiplex;
+    my $mux = IO::Multiplex->new();
 
     my $self = bless {
 	mux => $mux,
-	fhs => {}, # $vmid => fh
-	fhs_lookup => {}, # $fh => $vmid
-	queue => {},
-	current => {},
-	errors => {},
+	queue_lookup => {}, # $fh => $queue_info
+	queue_info => {},
     }, $class;
 
     $self->{eventcb} = $eventcb if $eventcb;
 
     $mux->set_callback_object($self);
 
+    # make sure perl doesn't believe this is a circular reference as we
+    # delete mux in DESTROY
+    weaken($mux->{_object});
+
     return $self;
 }
 
-# add a single command to the queue for later execution 
+# Note: List of special QGA command. Those commands can close the connection
+# without sending a response.
+
+my $qga_allow_close_cmds = {
+    'guest-shutdown' => 1,
+    'guest-suspend-ram' => 1,
+    'guest-suspend-disk' => 1,
+    'guest-suspend-hybrid' => 1,
+};
+
+my $push_cmd_to_queue = sub {
+    my ($self, $vmid, $cmd) = @_;
+
+    my $execute = $cmd->{execute} || die "no command name specified";
+
+    my $qga = ($execute =~ /^guest\-+/) ? 1 : 0;
+
+    my $sname = PVE::QemuServer::Helpers::qmp_socket($vmid, $qga);
+
+    $self->{queue_info}->{$sname} = { qga => $qga, vmid => $vmid, sname => $sname, cmds => [] }
+        if !$self->{queue_info}->{$sname};
+
+    push @{$self->{queue_info}->{$sname}->{cmds}}, $cmd;
+
+    return $self->{queue_info}->{$sname};
+};
+
+# add a single command to the queue for later execution
 # with queue_execute()
 sub queue_cmd {
     my ($self, $vmid, $callback, $execute, %params) = @_;
@@ -47,69 +78,149 @@ sub queue_cmd {
     $cmd->{arguments} = \%params;
     $cmd->{callback} = $callback;
 
-    push @{$self->{queue}->{$vmid}}, $cmd;
+    &$push_cmd_to_queue($self, $vmid, $cmd);
+
+    return;
 }
 
 # execute a single command
 sub cmd {
-    my ($self, $vmid, $cmd) = @_;
+    my ($self, $vmid, $cmd, $timeout) = @_;
 
     my $result;
 
     my $callback = sub {
 	my ($vmid, $resp) = @_;
 	$result = $resp->{'return'};
+	$result = { error => $resp->{'error'} } if !defined($result) && $resp->{'error'};
     };
 
+    die "no command specified" if !($cmd && $cmd->{execute});
+
     $cmd->{callback} = $callback;
     $cmd->{arguments} = {} if !defined($cmd->{arguments});
 
-    $self->{queue}->{$vmid} = [ $cmd ];
+    my $queue_info = &$push_cmd_to_queue($self, $vmid, $cmd);
+
+    if (!$timeout) {
+	# hack: monitor sometime blocks
+	if ($cmd->{execute} eq 'query-migrate') {
+	    $timeout = 60*60; # 1 hour
+	} elsif ($cmd->{execute} =~ m/^(eject|change)/) {
+	    $timeout = 60; # note: cdrom mount command is slow
+	} elsif ($cmd->{execute} eq 'guest-fsfreeze-freeze') {
+	    # freeze syncs all guest FS, if we kill it it stays in an unfreezable
+	    # locked state with high probability, so use an generous timeout
+	    $timeout = 60*60; # 1 hour
+	} elsif ($cmd->{execute} eq 'guest-fsfreeze-thaw') {
+	    # While it should return instantly or never (dead locked) for Linux guests,
+	    # the variance for Windows guests can be big. And there might be hook scripts
+	    # that are executed upon thaw, so use 3 minutes to be on the safe side.
+	    $timeout = 3 * 60;
+	} elsif (
+	    $cmd->{execute} eq 'savevm-start' ||
+	    $cmd->{execute} eq 'savevm-end' ||
+	    $cmd->{execute} eq 'query-backup' ||
+	    $cmd->{execute} eq 'query-block-jobs' ||
+	    $cmd->{execute} eq 'block-job-cancel' ||
+	    $cmd->{execute} eq 'block-job-complete' ||
+	    $cmd->{execute} eq 'backup-cancel' ||
+	    $cmd->{execute} eq 'query-savevm' ||
+	    $cmd->{execute} eq 'guest-fstrim' ||
+	    $cmd->{execute} eq 'guest-shutdown' ||
+	    $cmd->{execute} eq 'blockdev-snapshot-internal-sync' ||
+	    $cmd->{execute} eq 'blockdev-snapshot-delete-internal-sync'
+	 ) {
+	    $timeout = 10*60; # 10 mins ?
+	} else {
+	    #  NOTE: if you came here as user and want to change this, try using IO-Threads first
+	    # which move out quite some processing of the main thread, leaving more time for QMP
+	    $timeout = 5; # default
+	}
+    }
 
-    $self->queue_execute();
+    $self->queue_execute($timeout, 2);
 
-    my $cmdstr = $cmd->{execute} || '';
-    die "VM $vmid qmp command '$cmdstr' failed - $self->{errors}->{$vmid}"
-	if defined($self->{errors}->{$vmid});    
+    die "VM $vmid qmp command '$cmd->{execute}' failed - $queue_info->{error}"
+	if defined($queue_info->{error});
 
     return $result;
 };
 
 my $cmdid_seq = 0;
+my $cmdid_seq_qga = 0;
+
 my $next_cmdid = sub {
-    $cmdid_seq++;
-    return "$$:$cmdid_seq";
+    my ($qga) = @_;
+
+    if($qga){
+	$cmdid_seq_qga++;
+	return "$$"."0".$cmdid_seq_qga;
+    } else {
+	$cmdid_seq++;
+	return "$$:$cmdid_seq";
+    }
 };
 
-my $close_connection = sub {
-    my ($self, $vmid) = @_;
-	    
-    my $fh = $self->{fhs}->{$vmid};
-    return if !$fh;
- 
-    delete $self->{fhs}->{$vmid};
-    delete $self->{fhs_lookup}->{$fh};
+my $lookup_queue_info = sub {
+    my ($self, $fh, $noerr) = @_;
 
-    $self->{mux}->close($fh);
+    my $queue_info = $self->{queue_lookup}->{$fh};
+    if (!$queue_info) {
+	warn "internal error - unable to lookup queue info" if !$noerr;
+	return;
+    }
+    return $queue_info;
+};
 
-    print "CLOSE SOCKET to $vmid\n";
+my $close_connection = sub {
+    my ($self, $queue_info) = @_;
 
+    if (my $fh = delete $queue_info->{fh}) {
+	delete $self->{queue_lookup}->{$fh};
+	$self->{mux}->close($fh);
+    }
 };
 
 my $open_connection = sub {
-    my ($self, $vmid) = @_;
+    my ($self, $queue_info, $timeout) = @_;
+
+    die "duplicate call to open" if defined($queue_info->{fh});
+
+    my $vmid = $queue_info->{vmid};
+    my $qga = $queue_info->{qga};
 
-    my $sname = PVE::QemuServer::qmp_socket($vmid);
+    my $sname = PVE::QemuServer::Helpers::qmp_socket($vmid, $qga);
 
-    my $fh = IO::Socket::UNIX->new(Peer => $sname, Blocking => 0, Timeout => 1) ||
-	die "unable to connect to VM $vmid socket - $!\n";
+    $timeout = 1 if !$timeout;
 
-    print "OPEN SOCKET to $vmid \n";
+    my $fh;
+    my $starttime = [gettimeofday];
+    my $count = 0;
+
+    my $sotype = $qga ? 'qga' : 'qmp';
+
+    for (;;) {
+	$count++;
+	$fh = IO::Socket::UNIX->new(Peer => $sname, Blocking => 0, Timeout => 1);
+	last if $fh;
+	if ($! != EINTR && $! != EAGAIN) {
+	    die "unable to connect to VM $vmid $sotype socket - $!\n";
+	}
+	my $elapsed = tv_interval($starttime, [gettimeofday]);
+	if ($elapsed >= $timeout) {
+	    die "unable to connect to VM $vmid $sotype socket - timeout after $count retries\n";
+	}
+	usleep(100000);
+    }
+
+    $queue_info->{fh} = $fh;
+
+    $self->{queue_lookup}->{$fh} = $queue_info;
 
-    $self->{fhs}->{$vmid} = $fh;
-    $self->{fhs_lookup}->{$fh} = $vmid;
     $self->{mux}->add($fh);
- 
+    $self->{mux}->set_timeout($fh, $timeout);
+
     return $fh;
 };
 
@@ -117,42 +228,65 @@ my $check_queue = sub {
     my ($self) = @_;
 
     my $running = 0;
-	
-    foreach my $vmid (keys %{$self->{queue}}) {
-	my $fh = $self->{fhs}->{$vmid};
+
+    foreach my $sname (keys %{$self->{queue_info}}) {
+	my $queue_info = $self->{queue_info}->{$sname};
+	my $fh = $queue_info->{fh};
 	next if !$fh;
 
-	if ($self->{errors}->{$vmid}) {
-	    &$close_connection($self, $vmid);
+	my $qga = $queue_info->{qga};
+
+	if ($queue_info->{error}) {
+	    &$close_connection($self, $queue_info);
 	    next;
 	}
 
-	if ($self->{current}->{$vmid}) { # command running, waiting for response
+	if ($queue_info->{current}) { # command running, waiting for response
 	    $running++;
 	    next;
 	}
 
-	if (!scalar(@{$self->{queue}->{$vmid}})) { # no more commands for the VM
-	    &$close_connection($self, $vmid);
+	if (!scalar(@{$queue_info->{cmds}})) { # no more commands
+	    &$close_connection($self, $queue_info);
 	    next;
 	}
 
 	eval {
 
-	    my $cmd = $self->{current}->{$vmid} = shift @{$self->{queue}->{$vmid}};
-	    $cmd->{id} = &$next_cmdid();
+	    my $cmd = $queue_info->{current} = shift @{$queue_info->{cmds}};
+	    $cmd->{id} = &$next_cmdid($qga);
+
+	    my $fd = -1;
+	    if ($cmd->{execute} eq 'add-fd' || $cmd->{execute} eq 'getfd') {
+		$fd = $cmd->{arguments}->{fd};
+		delete $cmd->{arguments}->{fd};
+	    }
+
+	    my $qmpcmd;
+
+	    if ($qga) {
+
+		$qmpcmd = to_json({ execute => 'guest-sync-delimited',
+				    arguments => { id => int($cmd->{id})}}) . "\n" .
+		    to_json({ execute => $cmd->{execute}, arguments => $cmd->{arguments}}) . "\n";
 
-	    my $qmpcmd = to_json({
-		execute => $cmd->{execute},
-		arguments => $cmd->{arguments},
-		id => $cmd->{id}});
+	    } else {
 
-	    print "WRITECMD:$vmid: $qmpcmd\n";
-	    $self->{mux}->write($fh, $qmpcmd);
+		$qmpcmd = to_json({
+		    execute => $cmd->{execute},
+		    arguments => $cmd->{arguments},
+		    id => $cmd->{id}});
+	    }
+
+	    if ($fd >= 0) {
+		my $ret = PVE::IPCC::sendfd(fileno($fh), $fd, $qmpcmd);
+		die "sendfd failed" if $ret < 0;
+	    } else {
+		$self->{mux}->write($fh, $qmpcmd);
+	    }
 	};
 	if (my $err = $@) {
-	    $self->{errors}->{$vmid} = $err;
-	    # fixme: close fh?
+	    $queue_info->{error} = $err;
 	} else {
 	    $running++;
 	}
@@ -164,29 +298,30 @@ my $check_queue = sub {
 };
 
 # execute all queued command
+
 sub queue_execute {
-    my ($self, $timeout) = @_;
+    my ($self, $timeout, $noerr) = @_;
 
     $timeout = 3 if !$timeout;
 
-    print "start exec queue\n";
-
-    $self->{current} = {};
-    $self->{errors} = {};
-
     # open all necessary connections
-    foreach my $vmid (keys %{$self->{queue}}) {
-	next if !scalar(@{$self->{queue}->{$vmid}}); # no commands for the VM
+    foreach my $sname (keys %{$self->{queue_info}}) {
+	my $queue_info = $self->{queue_info}->{$sname};
+	next if !scalar(@{$queue_info->{cmds}}); # no commands
+
+	$queue_info->{error} = undef;
+	$queue_info->{current} = undef;
 
 	eval {
-	    my $fh = &$open_connection($self, $vmid);
-	    my $cmd = { execute => 'qmp_capabilities', arguments => {} };
-	    unshift @{$self->{queue}->{$vmid}}, $cmd;
-	    $self->{mux}->set_timeout($fh, $timeout);
+	    &$open_connection($self, $queue_info, $timeout);
+
+	    if (!$queue_info->{qga}) {
+		my $cap_cmd = { execute => 'qmp_capabilities', arguments => {} };
+		unshift @{$queue_info->{cmds}}, $cap_cmd;
+	    }
 	};
 	if (my $err = $@) {
-	    warn $err;
-	    $self->{errors}->{$vmid} = $err;
+	    $queue_info->{error} = $err;
 	}
     }
 
@@ -202,39 +337,88 @@ sub queue_execute {
     }
 
     # make sure we close everything
-    foreach my $vmid (keys %{$self->{fhs}}) {
-	&$close_connection($self, $vmid);
+    my $errors = '';
+    foreach my $sname (keys %{$self->{queue_info}}) {
+	my $queue_info = $self->{queue_info}->{$sname};
+	&$close_connection($self, $queue_info);
+	if ($queue_info->{error}) {
+	    if ($noerr) {
+		warn $queue_info->{error} if $noerr < 2;
+	    } else {
+		$errors .= $queue_info->{error}
+	    }
+	}
     }
 
-    $self->{queue} = $self->{current} = $self->{fhs} = $self->{fhs_lookup} = {};
+    $self->{queue_info} = $self->{queue_lookup} = {};
+
+    die $errors if $errors;
+}
+
+sub mux_close {
+    my ($self, $mux, $fh) = @_;
 
-    print "end exec queue $running\n";
+    my $queue_info = &$lookup_queue_info($self, $fh, 1);
+    return if !$queue_info;
 
+    $queue_info->{error} = "client closed connection\n"
+	if !$queue_info->{error};
 }
 
-# mux_input is called when input is available on one of
-# the descriptors.
+# mux_input is called when input is available on one of the descriptors.
 sub mux_input {
     my ($self, $mux, $fh, $input) = @_;
 
-    print "GOT: $$input\n";
- 
-    return if $$input !~ m/}\r\n$/;
+    my $queue_info = &$lookup_queue_info($self, $fh);
+    return if !$queue_info;
 
-    my $raw = $$input;
+    my $sname = $queue_info->{sname};
+    my $vmid = $queue_info->{vmid};
+    my $qga = $queue_info->{qga};
 
-    # Remove the input from the input buffer.
-    $$input = '';
+    my $curcmd = $queue_info->{current};
+    die "unable to lookup current command for VM $vmid ($sname)\n" if !$curcmd;
 
-    my $vmid = $self->{fhs_lookup}->{$fh};
-    if (!$vmid) {
-	warn "internal error - unable to lookup vmid";
-	return;
+    my $raw;
+
+    if ($qga) {
+	return if $$input !~ s/^.*\xff([^\n]+}\r?\n[^\n]+})\r?\n(.*)$/$2/so;
+	$raw = $1;
+    } else {
+	return if $$input !~ s/^(.*})\r?\n(.*)$/$2/so;
+	$raw = $1;
     }
 
     eval {
 	my @jsons = split("\n", $raw);
 
+	if ($qga) {
+
+	    die "response is not complete" if @jsons != 2 ;
+
+	    my $obj = from_json($jsons[0]);
+
+	    my $cmdid = $obj->{'return'};
+	    die "received responsed without command id\n" if !$cmdid;
+
+	    # skip results fro previous commands
+	    return if $cmdid < $curcmd->{id};
+
+	    if ($curcmd->{id} ne $cmdid) {
+		die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
+	    }
+
+	    delete $queue_info->{current};
+
+	    $obj = from_json($jsons[1]);
+
+	    if (my $callback = $curcmd->{callback}) {
+		&$callback($vmid, $obj);
+	    }
+
+	    return;
+	}
+
 	foreach my $json (@jsons) {
 	    my $obj = from_json($json);
 	    next if defined($obj->{QMP}); # skip monitor greeting
@@ -246,11 +430,6 @@ sub mux_input {
 		next;
 	    }
 
-	    # die $obj->{error}->{desc} if defined($obj->{error}->{desc});
- 
-	    #print "GOTOBJ: " . Dumper($obj);
-
-	    # we do not need events for now
 	    if (defined($obj->{event})) {
 		if (my $eventcb = $self->{eventcb}) {
 		    &$eventcb($obj);
@@ -261,22 +440,19 @@ sub mux_input {
 	    my $cmdid = $obj->{id};
 	    die "received responsed without command id\n" if !$cmdid;
 
-	    my $curcmd = $self->{current}->{$vmid};
-	    die "unable to lookup current command for VM $vmid\n" if !$curcmd;
-
-	    delete $self->{current}->{$vmid};
-	    
 	    if ($curcmd->{id} ne $cmdid) {
 		die "got wrong command id '$cmdid' (expected $curcmd->{id})\n";
 	    }
 
+	    delete $queue_info->{current};
+
 	    if (my $callback = $curcmd->{callback}) {
 		&$callback($vmid, $obj);
 	    }
 	}
     };
     if (my $err = $@) {
-	$self->{errors}->{$vmid} = $err;
+	$queue_info->{error} = $err;
     }
 
     &$check_queue($self);
@@ -286,122 +462,64 @@ sub mux_input {
 sub mux_timeout {
     my ($self, $mux, $fh) = @_;
 
-    if (my $vmid = $self->{fhs_lookup}->{$fh}) {
-
-	print "GOT timeout for $vmid\n";
-
-	$self->{errors}->{$vmid} = "got timeout\n";
+    if (my $queue_info = &$lookup_queue_info($self, $fh)) {
+	$queue_info->{error} = "got timeout\n";
+	$self->{mux}->inbuffer($fh, ''); # clear to avoid warnings
     }
 
     &$check_queue($self);
 }
 
+sub mux_eof {
+    my ($self, $mux, $fh, $input) = @_;
 
+    my $queue_info = &$lookup_queue_info($self, $fh);
+    return if !$queue_info;
 
-package test;
-
-use strict;
-use PVE::SafeSyslog;
-use PVE::INotify;
-use PVE::QemuServer;
-use PVE::Cluster;
-use Data::Dumper;
-
-initlog($0);
-
-$ENV{'PATH'} = '/sbin:/bin:/usr/sbin:/usr/bin';
-
-die "please run as root\n" if $> != 0;
-
-PVE::INotify::inotify_init();
-
-my $nodename = PVE::INotify::nodename();
-
-sub vm_qmp_command {
-    my ($vmid, $cmd, $nocheck) = @_;
-
-    my $res;
-
-    eval {
-	die "VM $vmid not running\n" if !PVE::QemuServer::check_running($vmid, $nocheck);
-
-	my $qmpclient = PVE::QMPClient->new();
-
-	$res = $qmpclient->cmd($vmid, $cmd);
-
-    };
-    if (my $err = $@) {
-	syslog("err", "VM $vmid qmp command failed - $err");
-	die $err;
-    }
+    my $sname = $queue_info->{sname};
+    my $vmid = $queue_info->{vmid};
+    my $qga = $queue_info->{qga};
 
-    return $res;
-}
+    my $curcmd = $queue_info->{current};
+    die "unable to lookup current command for VM $vmid ($sname)\n" if !$curcmd;
 
-# print Dumper(vm_qmp_command(100, { execute => 'query-status' }));
+    if ($qga && $qga_allow_close_cmds->{$curcmd->{execute}}) {
 
-sub update_qemu_stats {
-    
-    print "start update\n";
+	return if $$input !~ s/^.*\xff([^\n]+})\r?\n(.*)$/$2/so;
 
-    my $ctime = time();
+	my $raw = $1;
 
-    my $vmstatus = PVE::QemuServer::vmstatus();
+	eval {
+	    my $obj = from_json($raw);
 
-    my $qmpclient = PVE::QMPClient->new();
+	    my $cmdid = $obj->{'return'};
+	    die "received responsed without command id\n" if !$cmdid;
 
-    my $res = {};
+	    delete $queue_info->{current};
 
-    my $blockstatscb = sub {
-	my ($vmid, $resp) = @_;
-	my $data = $resp->{'return'} || [];
-	my $totalrdbytes = 0;
-	my $totalwrbytes = 0;
-	for my $blockstat (@$data) {
-	    $totalrdbytes = $totalrdbytes + $blockstat->{stats}->{rd_bytes};
-	    $totalwrbytes = $totalwrbytes + $blockstat->{stats}->{wr_bytes};
+	    if (my $callback = $curcmd->{callback}) {
+		&$callback($vmid, undef);
+	    }
+	};
+	if (my $err = $@) {
+	    $queue_info->{error} = $err;
 	}
-	$res->{$vmid}->{diskread} = $totalrdbytes;
-	$res->{$vmid}->{diskwrite} = $totalwrbytes;
-    };
 
-    my $statuscb = sub {
-	my ($vmid, $resp) = @_;
-	$qmpclient->queue_cmd($vmid, $blockstatscb, 'query-blockstats');
+	&$close_connection($self, $queue_info);
 
-	my $status = 'unknown';
-	if (!defined($status = $resp->{'return'}->{status})) {
-	    warn "unable to get VM status\n";
-	    return;
+	if (scalar(@{$queue_info->{cmds}}) && !$queue_info->{error}) {
+	    $queue_info->{error} = "Got EOF but command queue is not empty.\n";
 	}
+    }
+}
 
-	$res->{$vmid}->{status} = $resp->{'return'}->{status};
-    };
-
-    foreach my $vmid (keys %$vmstatus) {
-	my $d = $vmstatus->{$vmid};
-	my $data;
-	if ($d->{pid}) { # running
-
-	    $qmpclient->queue_cmd($vmid, $statuscb, 'query-status');
+sub DESTROY {
+    my ($self) = @_;
 
-	}
-    }
-    print "start loop\n";
-    $qmpclient->queue_execute();
-    print "end loop\n";
-    print Dumper($res);
-    foreach my $vmid (keys %{$qmpclient->{errors}}) {
-	my $msg = "qmp error on VM $vmid: $qmpclient->{errors}->{$vmid}";
-	chomp $msg;
-	warn "$msg\n";
+    foreach my $sname (keys %{$self->{queue_info}}) {
+	my $queue_info = $self->{queue_info}->{$sname};
+	$close_connection->($self, $queue_info);
     }
-
-    print "end update\n";
 }
 
-for(;;) {
-    PVE::Cluster::cfs_update();
-    update_qemu_stats();
-    sleep(3);
-}
+1;