[pve-manager.git] / PVE / Service / pvestatd.pm

package PVE::Service::pvestatd;

use strict;
use warnings;

use PVE::SafeSyslog;
use PVE::Daemon;

use Time::HiRes qw (gettimeofday);
use PVE::Tools qw(dir_glob_foreach file_read_firstline);
use PVE::ProcFSTools;
use PVE::CpuSet;
use Filesys::Df;
use PVE::INotify;
use PVE::Cluster qw(cfs_read_file);
use PVE::Storage;
use PVE::QemuServer;
use PVE::LXC;
use PVE::LXC::Config;
use PVE::RPCEnvironment;
use PVE::API2::Subscription;
use PVE::AutoBalloon;

use PVE::Status::Plugin;
use PVE::Status::Graphite;
use PVE::Status::InfluxDB;

PVE::Status::Graphite->register();
PVE::Status::InfluxDB->register();
PVE::Status::Plugin->init();

use base qw(PVE::Daemon);

my $opt_debug;
my $restart_request;

my $nodename = PVE::INotify::nodename();

my $cmdline = [$0, @ARGV];

my %daemon_options = (restart_on_error => 5, stop_wait_time => 5);
my $daemon = __PACKAGE__->new('pvestatd', $cmdline, %daemon_options);

sub init {
    my ($self) = @_;

    $opt_debug = $self->{debug};

    PVE::Cluster::cfs_update();
}

sub shutdown {
    my ($self) = @_;

    syslog('info' , "server closing");

    # wait for children
    1 while (waitpid(-1, POSIX::WNOHANG()) > 0);

    $self->exit_daemon(0);
}

sub hup {
    my ($self) = @_;

    $restart_request = 1;
}

sub update_node_status {
    my ($status_cfg) = @_;

    my ($avg1, $avg5, $avg15) = PVE::ProcFSTools::read_loadavg();

    my $stat = PVE::ProcFSTools::read_proc_stat();

    my $netdev = PVE::ProcFSTools::read_proc_net_dev();

    my ($uptime) = PVE::ProcFSTools::read_proc_uptime();

    my $cpuinfo = PVE::ProcFSTools::read_cpuinfo();

    my $maxcpu = $cpuinfo->{cpus}; 

    my $subinfo = PVE::INotify::read_file('subscription');
    my $sublevel = $subinfo->{level} || '';

    # traffic from/to physical interface cards
    my $netin = 0;
    my $netout = 0;
    foreach my $dev (keys %$netdev) {
	next if $dev !~ m/^eth\d+$/;
	$netin += $netdev->{$dev}->{receive};
	$netout += $netdev->{$dev}->{transmit};
    }
 
    my $meminfo = PVE::ProcFSTools::read_meminfo();

    my $dinfo = df('/', 1);     # output is bytes

    my $ctime = time();

    # everything not free is considered to be used
    my $dused = $dinfo->{blocks} - $dinfo->{bfree};

    my $data = "$uptime:$sublevel:$ctime:$avg1:$maxcpu:$stat->{cpu}:$stat->{wait}:" .
	"$meminfo->{memtotal}:$meminfo->{memused}:" .
	"$meminfo->{swaptotal}:$meminfo->{swapused}:" .
	"$dinfo->{blocks}:$dused:$netin:$netout";

    PVE::Cluster::broadcast_rrd("pve2-node/$nodename", $data);

    foreach my $id (keys %{$status_cfg->{ids}}) {
	my $plugin_config = $status_cfg->{ids}->{$id};
	next if $plugin_config->{disable};
	my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});

	my $d = {};
	$d->{uptime} = $uptime;
	$d->{cpustat} = $stat;
	$d->{cpustat}->{avg1} = $avg1;
	$d->{cpustat}->{avg5} = $avg5;
	$d->{cpustat}->{avg15} = $avg15;
	$d->{cpustat}->{cpus} = $maxcpu;
	$d->{memory} = $meminfo;
	$d->{blockstat} = $dinfo;
	$d->{nics} = $netdev;

	$plugin->update_node_status($plugin_config, $nodename, $d, $ctime);
    }
}

sub auto_balloning {
    my ($vmstatus) =  @_;

    my $log = sub {
       return if !$opt_debug;
       print @_;
    };

    my $hostmeminfo = PVE::ProcFSTools::read_meminfo();

    # to debug, run 'pvestatd -d' and set  memtotal here
    #$hostmeminfo->{memtotal} = int(2*1024*1024*1024/0.8); # you can set this to test

    my $hostfreemem = $hostmeminfo->{memtotal} - $hostmeminfo->{memused};

    # we try to use about 80% host memory
    # goal: we want to change memory usage by this amount (positive or negative)
    my $goal = int($hostmeminfo->{memtotal}*0.8 - $hostmeminfo->{memused});

    my $maxchange = 100*1024*1024;
    my $res = PVE::AutoBalloon::compute_alg1($vmstatus, $goal, $maxchange);
 
    &$log("host goal: $goal free: $hostfreemem total: $hostmeminfo->{memtotal}\n");

    foreach my $vmid (keys %$vmstatus) {
	next if !$res->{$vmid};
	my $d = $vmstatus->{$vmid};
	my $diff = int($res->{$vmid} - $d->{balloon});
	my $absdiff = $diff < 0 ? -$diff : $diff;
	if ($absdiff > 0) {
	    &$log("BALLOON $vmid to $res->{$vmid} ($diff)\n");
	    eval {
		PVE::QemuServer::vm_mon_cmd($vmid, "balloon", 
					    value => int($res->{$vmid}));
	    };
	    warn $@ if $@;
	}
    }
}

sub update_qemu_status {
    my ($status_cfg) = @_;

    my $ctime = time();

    my $vmstatus = PVE::QemuServer::vmstatus(undef, 1);

    eval { auto_balloning($vmstatus); };
    syslog('err', "auto ballooning error: $@") if $@;

    foreach my $vmid (keys %$vmstatus) {
	my $d = $vmstatus->{$vmid};
	my $data;
	my $status = $d->{qmpstatus} || $d->{status} || 'stopped';
	my $template = $d->{template} ? $d->{template} : "0";
	if ($d->{pid}) { # running
	    $data = "$d->{uptime}:$d->{name}:$status:$template:" .
		"$ctime:$d->{cpus}:$d->{cpu}:" .
		"$d->{maxmem}:$d->{mem}:" .
		"$d->{maxdisk}:$d->{disk}:" .
		"$d->{netin}:$d->{netout}:" .
		"$d->{diskread}:$d->{diskwrite}";
	} else {
	    $data = "0:$d->{name}:$status:$template:$ctime:$d->{cpus}::" .
		"$d->{maxmem}::" .
		"$d->{maxdisk}:$d->{disk}:" .
		":::";
	}
	PVE::Cluster::broadcast_rrd("pve2.3-vm/$vmid", $data);

	foreach my $id (keys %{$status_cfg->{ids}}) {
	    my $plugin_config = $status_cfg->{ids}->{$id};
	    next if $plugin_config->{disable};
	    my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
	    $plugin->update_qemu_status($plugin_config, $vmid, $d, $ctime);
	}
    }
}

sub remove_stale_lxc_consoles {

    my $vmstatus = PVE::LXC::vmstatus();
    my $pidhash = PVE::LXC::find_lxc_console_pids();

    foreach my $vmid (keys %$pidhash) {
	next if defined($vmstatus->{$vmid});
	syslog('info', "remove stale lxc-console for CT $vmid");
	foreach my $pid (@{$pidhash->{$vmid}}) {
	    kill(9, $pid);
	}
    }
}

my $rebalance_error_count = {};

sub rebalance_lxc_containers {

    return if !-d '/sys/fs/cgroup/cpuset/lxc'; # nothing to do...

    my $all_cpus = PVE::CpuSet->new_from_cgroup('lxc', 'effective_cpus');
    my @allowed_cpus = $all_cpus->members();
    my $cpucount = scalar(@allowed_cpus);
    my $max_cpuid = PVE::CpuSet::max_cpuid();

    my @cpu_ctcount = (0) x $max_cpuid;
    my @balanced_cts;

    my $modify_cpuset = sub {
	my ($vmid, $cpuset, $newset) = @_;

	if (!$rebalance_error_count->{$vmid}) {
	    syslog('info', "modified cpu set for lxc/$vmid: " .
		   $newset->short_string());
	}

	eval {
	    # allow all, so that we can set new cpuset in /ns
	    $all_cpus->write_to_cgroup("lxc/$vmid");
	    eval {
		$newset->write_to_cgroup("lxc/$vmid/ns");
	    };
	    if (my $err = $@) {
		warn $err if !$rebalance_error_count->{$vmid}++;
		# restore original
		$cpuset->write_to_cgroup("lxc/$vmid");
	    } else {
		# also apply to container root cgroup
		$newset->write_to_cgroup("lxc/$vmid");
		$rebalance_error_count->{$vmid} = 0;
	    }
	};
	if (my $err = $@) {
	    warn $err if !$rebalance_error_count->{$vmid}++;
	}
    };

    my $ctlist = PVE::LXC::config_list();

    foreach my $vmid (sort keys %$ctlist) {
	next if ! -d "/sys/fs/cgroup/cpuset/lxc/$vmid";

	my ($conf, $cpuset);
	eval {

	    $conf = PVE::LXC::Config->load_config($vmid);

	    $cpuset = PVE::CpuSet->new_from_cgroup("lxc/$vmid");
	};
	if (my $err = $@) {
	    warn $err;
	    next;
	}

	my @cpuset_members = $cpuset->members();

	if (!PVE::LXC::Config->has_lxc_entry($conf, 'lxc.cgroup.cpuset.cpus')) {

	    my $cores = $conf->{cores} || $cpucount;
	    $cores = $cpucount if $cores > $cpucount;

	    # see if the number of cores was hot-reduced or
	    # hasn't been enacted at all yet
	    my $newset = PVE::CpuSet->new();
	    if ($cores <  scalar(@cpuset_members)) {
		for (my $i = 0; $i < $cores; $i++) {
		    $newset->insert($cpuset_members[$i]);
		}
	    } elsif ($cores > scalar(@cpuset_members)) {
		my $count = $newset->insert(@cpuset_members);
		foreach my $cpu (@allowed_cpus) {
		    $count += $newset->insert($cpu);
		    last if $count >= $cores;
		}
	    } else {
		$newset->insert(@cpuset_members);
	    }

	    # Apply hot-plugged changes if any:
	    if (!$newset->is_equal($cpuset)) {
		@cpuset_members = $newset->members();
		$modify_cpuset->($vmid, $cpuset, $newset);
	    }

	    # Note: no need to rebalance if we already use all cores
	    push @balanced_cts, [$vmid, $cores, $newset]
		if defined($conf->{cores}) && ($cores != $cpucount);
	}

	foreach my $cpu (@cpuset_members) {
	    $cpu_ctcount[$cpu]++ if $cpu <= $max_cpuid;
	}
    }

    my $find_best_cpu = sub {
	my ($cpulist, $cpu) = @_;

	my $cur_cost = $cpu_ctcount[$cpu];
	my $cur_cpu = $cpu;

	foreach my $candidate (@$cpulist) {
	    my $cost = $cpu_ctcount[$candidate];
	    if ($cost < ($cur_cost -1)) {
		$cur_cost = $cost;
		$cur_cpu = $candidate;
	    }
	}

	return $cur_cpu;
    };

    foreach my $bct (@balanced_cts) {
	my ($vmid, $cores, $cpuset) = @$bct;

	my $newset = PVE::CpuSet->new();

	my $rest = [];
	foreach my $cpu (@allowed_cpus) {
	    next if $cpuset->has($cpu);
	    push @$rest, $cpu;
	}

	my @members = $cpuset->members();
	foreach my $cpu (@members) {
	    my $best =  &$find_best_cpu($rest, $cpu);
	    if ($best != $cpu) {
		$cpu_ctcount[$best]++;
		$cpu_ctcount[$cpu]--;
	    }
	    $newset->insert($best);
	}

	if (!$newset->is_equal($cpuset)) {
	    $modify_cpuset->($vmid, $cpuset, $newset);
	}
    }
}

sub update_lxc_status {
    my ($status_cfg) = @_;

    my $ctime = time();

    my $vmstatus = PVE::LXC::vmstatus();

    foreach my $vmid (keys %$vmstatus) {
	my $d = $vmstatus->{$vmid};
	my $template = $d->{template} ? $d->{template} : "0";
	my $data;
	if ($d->{status} eq 'running') { # running
	    $data = "$d->{uptime}:$d->{name}:$d->{status}:$template:" .
		"$ctime:$d->{cpus}:$d->{cpu}:" .
		"$d->{maxmem}:$d->{mem}:" .
		"$d->{maxdisk}:$d->{disk}:" .
		"$d->{netin}:$d->{netout}:" .
		"$d->{diskread}:$d->{diskwrite}";
	} else {
	    $data = "0:$d->{name}:$d->{status}:$template:$ctime:$d->{cpus}::" .
		"$d->{maxmem}::" .
		"$d->{maxdisk}:$d->{disk}:" .
		":::";
	}
	PVE::Cluster::broadcast_rrd("pve2.3-vm/$vmid", $data);

	foreach my $id (keys %{$status_cfg->{ids}}) {
	    my $plugin_config = $status_cfg->{ids}->{$id};
	    next if $plugin_config->{disable};
	    my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
	    $plugin->update_lxc_status($plugin_config, $vmid, $d, $ctime);
	}
    }
}

sub update_storage_status {
    my ($status_cfg) = @_;

    my $cfg = PVE::Storage::config();

    my $ctime = time();

    my $info = PVE::Storage::storage_info($cfg);

    foreach my $storeid (keys %$info) {
	my $d = $info->{$storeid};
	next if !$d->{active};

	my $data = "$ctime:$d->{total}:$d->{used}";

	my $key = "pve2-storage/${nodename}/$storeid";
	PVE::Cluster::broadcast_rrd($key, $data);

	foreach my $id (keys %{$status_cfg->{ids}}) {
	    my $plugin_config = $status_cfg->{ids}->{$id};
	    next if $plugin_config->{disable};
	    my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
	    $plugin->update_storage_status($plugin_config, $nodename, $storeid, $d, $ctime);
	}
    }
}

sub update_status {

    # update worker list. This is not really required and
    # we just call this to make sure that we have a correct
    # list in case of an unexpected crash.
    eval {
	my $tlist = PVE::RPCEnvironment::active_workers();
	PVE::Cluster::broadcast_tasklist($tlist);
    };
    my $err = $@;
    syslog('err', $err) if $err;

    my $status_cfg = PVE::Cluster::cfs_read_file('status.cfg');

    eval {
	update_node_status($status_cfg);
    };
    $err = $@;
    syslog('err', "node status update error: $err") if $err;

    eval {
	update_qemu_status($status_cfg);
    };
    $err = $@;
    syslog('err', "qemu status update error: $err") if $err;

    eval {
	update_lxc_status($status_cfg);
    };
    $err = $@;
    syslog('err', "lxc status update error: $err") if $err;

    eval {
	rebalance_lxc_containers();
    };
    $err = $@;
    syslog('err', "lxc cpuset rebalance error: $err") if $err;

    eval {
	update_storage_status($status_cfg);
    };
    $err = $@;
    syslog('err', "storage status update error: $err") if $err;

    eval {
	remove_stale_lxc_consoles();
    };
    $err = $@;
    syslog('err', "lxc console cleanup error: $err") if $err;
}

my $next_update = 0;

# do not update directly after startup, because install scripts
# have a problem with that
my $cycle = 0; 
my $updatetime = 10;

my $initial_memory_usage;

sub run {
    my ($self) = @_;

    for (;;) { # forever

 	$next_update = time() + $updatetime;

	if ($cycle) {
	    my ($ccsec, $cusec) = gettimeofday ();
	    eval {
		# syslog('info', "start status update");
		PVE::Cluster::cfs_update();
		update_status();
	    };
	    my $err = $@;

	    if ($err) {
		syslog('err', "status update error: $err");
	    }

	    my ($ccsec_end, $cusec_end) = gettimeofday ();
	    my $cptime = ($ccsec_end-$ccsec) + ($cusec_end - $cusec)/1000000;

	    syslog('info', sprintf("status update time (%.3f seconds)", $cptime))
		if ($cptime > 5);
	}

	$cycle++;

	my $mem = PVE::ProcFSTools::read_memory_usage();

	if (!defined($initial_memory_usage) || ($cycle < 10)) {
	    $initial_memory_usage = $mem->{resident};
	} else {
	    my $diff = $mem->{resident} - $initial_memory_usage;
	    if ($diff > 5*1024*1024) {
		syslog ('info', "restarting server after $cycle cycles to " .
			"reduce memory usage (free $mem->{resident} ($diff) bytes)");
		$self->restart_daemon();
	    }
	}

	my $wcount = 0;
	while ((time() < $next_update) && 
	       ($wcount < $updatetime) && # protect against time wrap
	       !$restart_request) { $wcount++; sleep (1); };

	$self->restart_daemon() if $restart_request;
    }
}

$daemon->register_start_command();
$daemon->register_restart_command(1);
$daemon->register_stop_command();
$daemon->register_status_command();

our $cmddef = {
    start => [ __PACKAGE__, 'start', []],
    restart => [ __PACKAGE__, 'restart', []],
    stop => [ __PACKAGE__, 'stop', []],
    status => [ __PACKAGE__, 'status', [], undef, sub { print shift . "\n";} ],
};

#my $cmd = shift;
#PVE::CLIHandler::handle_cmd($cmddef, $0, $cmd, \@ARGV, undef, $0);
#exit (0);

1;

__END__

=head1 NAME
                                          
pvestatd - PVE Status Daemon

=head1 SYNOPSIS

=include synopsis

=head1 DESCRIPTION

This daemom queries the status of VMs, storages and containers at
regular intervals. The result is sent to all nodes in the cluster.

=include pve_copyright
Commit	Line	Data
efd04666 DM	1	package PVE::Service::pvestatd;
	2
	3	use strict;
	4	use warnings;
	5
	6	use PVE::SafeSyslog;
	7	use PVE::Daemon;
	8
	9	use Time::HiRes qw (gettimeofday);
	10	use PVE::Tools qw(dir_glob_foreach file_read_firstline);
	11	use PVE::ProcFSTools;
41db757b	12	use PVE::CpuSet;
efd04666 DM	13	use Filesys::Df;
	14	use PVE::INotify;
	15	use PVE::Cluster qw(cfs_read_file);
	16	use PVE::Storage;
	17	use PVE::QemuServer;
	18	use PVE::LXC;
41db757b	19	use PVE::LXC::Config;
efd04666 DM	20	use PVE::RPCEnvironment;
	21	use PVE::API2::Subscription;
	22	use PVE::AutoBalloon;
	23
	24	use PVE::Status::Plugin;
	25	use PVE::Status::Graphite;
58541b94	26	use PVE::Status::InfluxDB;
efd04666 DM	27
efd04666 DM	28	PVE::Status::Graphite->register();
58541b94	29	PVE::Status::InfluxDB->register();
efd04666 DM	30	PVE::Status::Plugin->init();
	31
	32	use base qw(PVE::Daemon);
	33
	34	my $opt_debug;
	35	my $restart_request;
	36
	37	my $nodename = PVE::INotify::nodename();
	38
	39	my $cmdline = [$0, @ARGV];
	40
	41	my %daemon_options = (restart_on_error => 5, stop_wait_time => 5);
	42	my $daemon = __PACKAGE__->new('pvestatd', $cmdline, %daemon_options);
	43
	44	sub init {
	45	my ($self) = @_;
	46
	47	$opt_debug = $self->{debug};
	48
	49	PVE::Cluster::cfs_update();
	50	}
	51
	52	sub shutdown {
	53	my ($self) = @_;
	54
	55	syslog('info' , "server closing");
	56
	57	# wait for children
	58	1 while (waitpid(-1, POSIX::WNOHANG()) > 0);
	59
	60	$self->exit_daemon(0);
	61	}
	62
	63	sub hup {
	64	my ($self) = @_;
	65
	66	$restart_request = 1;
	67	}
	68
	69	sub update_node_status {
	70	my ($status_cfg) = @_;
	71
	72	my ($avg1, $avg5, $avg15) = PVE::ProcFSTools::read_loadavg();
	73
	74	my $stat = PVE::ProcFSTools::read_proc_stat();
	75
	76	my $netdev = PVE::ProcFSTools::read_proc_net_dev();
	77
	78	my ($uptime) = PVE::ProcFSTools::read_proc_uptime();
	79
	80	my $cpuinfo = PVE::ProcFSTools::read_cpuinfo();
	81
	82	my $maxcpu = $cpuinfo->{cpus};
	83
	84	my $subinfo = PVE::INotify::read_file('subscription');
	85	my $sublevel = $subinfo->{level} \|\| '';
	86
	87	# traffic from/to physical interface cards
	88	my $netin = 0;
	89	my $netout = 0;
	90	foreach my $dev (keys %$netdev) {
	91	next if $dev !~ m/^eth\d+$/;
	92	$netin += $netdev->{$dev}->{receive};
	93	$netout += $netdev->{$dev}->{transmit};
94	}
95
96	my $meminfo = PVE::ProcFSTools::read_meminfo();
97
98	my $dinfo = df('/', 1); # output is bytes
99
100	my $ctime = time();
101
102	# everything not free is considered to be used
103	my $dused = $dinfo->{blocks} - $dinfo->{bfree};
104
105	my $data = "$uptime:$sublevel:$ctime:$avg1:$maxcpu:$stat->{cpu}:$stat->{wait}:" .
106	"$meminfo->{memtotal}:$meminfo->{memused}:" .
107	"$meminfo->{swaptotal}:$meminfo->{swapused}:" .
108	"$dinfo->{blocks}:$dused:$netin:$netout";
109
110	PVE::Cluster::broadcast_rrd("pve2-node/$nodename", $data);
111
112	foreach my $id (keys %{$status_cfg->{ids}}) {
113	my $plugin_config = $status_cfg->{ids}->{$id};
114	next if $plugin_config->{disable};
115	my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
116
117	my $d = {};
118	$d->{uptime} = $uptime;
119	$d->{cpustat} = $stat;
120	$d->{cpustat}->{avg1} = $avg1;
121	$d->{cpustat}->{avg5} = $avg5;
122	$d->{cpustat}->{avg15} = $avg15;
123	$d->{cpustat}->{cpus} = $maxcpu;
124	$d->{memory} = $meminfo;
125	$d->{blockstat} = $dinfo;
126	$d->{nics} = $netdev;
127
128	$plugin->update_node_status($plugin_config, $nodename, $d, $ctime);
129	}
130	}
131
132	sub auto_balloning {
133	my ($vmstatus) = @_;
134
135	my $log = sub {
136	return if !$opt_debug;
137	print @_;
138	};
139
140	my $hostmeminfo = PVE::ProcFSTools::read_meminfo();
141
142	# to debug, run 'pvestatd -d' and set memtotal here
143	#$hostmeminfo->{memtotal} = int(210241024*1024/0.8); # you can set this to test
144
145	my $hostfreemem = $hostmeminfo->{memtotal} - $hostmeminfo->{memused};
146
147	# we try to use about 80% host memory
148	# goal: we want to change memory usage by this amount (positive or negative)
149	my $goal = int($hostmeminfo->{memtotal}*0.8 - $hostmeminfo->{memused});
150
151	my $maxchange = 10010241024;
152	my $res = PVE::AutoBalloon::compute_alg1($vmstatus, $goal, $maxchange);
153
154	&$log("host goal: $goal free: $hostfreemem total: $hostmeminfo->{memtotal}\n");
155
156	foreach my $vmid (keys %$vmstatus) {
157	next if !$res->{$vmid};
158	my $d = $vmstatus->{$vmid};
159	my $diff = int($res->{$vmid} - $d->{balloon});
160	my $absdiff = $diff < 0 ? -$diff : $diff;
161	if ($absdiff > 0) {
162	&$log("BALLOON $vmid to $res->{$vmid} ($diff)\n");
163	eval {
164	PVE::QemuServer::vm_mon_cmd($vmid, "balloon",
165	value => int($res->{$vmid}));
166	};
167	warn $@ if $@;
168	}
169	}
170	}
171
172	sub update_qemu_status {
173	my ($status_cfg) = @_;
174
175	my $ctime = time();
176
177	my $vmstatus = PVE::QemuServer::vmstatus(undef, 1);
178
179	eval { auto_balloning($vmstatus); };
180	syslog('err', "auto ballooning error: $@") if $@;
181
182	foreach my $vmid (keys %$vmstatus) {
183	my $d = $vmstatus->{$vmid};
184	my $data;
185	my $status = $d->{qmpstatus} \|\| $d->{status} \|\| 'stopped';
186	my $template = $d->{template} ? $d->{template} : "0";
187	if ($d->{pid}) { # running
188	$data = "$d->{uptime}:$d->{name}:$status:$template:" .
189	"$ctime:$d->{cpus}:$d->{cpu}:" .
190	"$d->{maxmem}:$d->{mem}:" .
191	"$d->{maxdisk}:$d->{disk}:" .
192	"$d->{netin}:$d->{netout}:" .
193	"$d->{diskread}:$d->{diskwrite}";
194	} else {
195	$data = "0:$d->{name}:$status:$template:$ctime:$d->{cpus}::" .
196	"$d->{maxmem}::" .
197	"$d->{maxdisk}:$d->{disk}:" .
198	":::";
199	}
200	PVE::Cluster::broadcast_rrd("pve2.3-vm/$vmid", $data);
201
202	foreach my $id (keys %{$status_cfg->{ids}}) {
203	my $plugin_config = $status_cfg->{ids}->{$id};
204	next if $plugin_config->{disable};
205	my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
206	$plugin->update_qemu_status($plugin_config, $vmid, $d, $ctime);
207	}
208	}
209	}
210
211	sub remove_stale_lxc_consoles {
212
213	my $vmstatus = PVE::LXC::vmstatus();
214	my $pidhash = PVE::LXC::find_lxc_console_pids();
215
216	foreach my $vmid (keys %$pidhash) {
217	next if defined($vmstatus->{$vmid});
218	syslog('info', "remove stale lxc-console for CT $vmid");
219	foreach my $pid (@{$pidhash->{$vmid}}) {
220	kill(9, $pid);
221	}
222	}
223	}
224
b3f1adb2 DM	225	my $rebalance_error_count = {};
b3f1adb2 DM	226
41db757b	227	sub rebalance_lxc_containers {
41db757b DM	228
	229	return if !-d '/sys/fs/cgroup/cpuset/lxc'; # nothing to do...
	230
	231	my $all_cpus = PVE::CpuSet->new_from_cgroup('lxc', 'effective_cpus');
	232	my @allowed_cpus = $all_cpus->members();
	233	my $cpucount = scalar(@allowed_cpus);
ccfff920	234	my $max_cpuid = PVE::CpuSet::max_cpuid();
41db757b	235
ccfff920	236	my @cpu_ctcount = (0) x $max_cpuid;
41db757b DM	237	my @balanced_cts;
41db757b DM	238
0b959507 DM	239	my $modify_cpuset = sub {
	240	my ($vmid, $cpuset, $newset) = @_;
	241
b3f1adb2 DM	242	if (!$rebalance_error_count->{$vmid}) {
	243	syslog('info', "modified cpu set for lxc/$vmid: " .
	244	$newset->short_string());
	245	}
	246
0b959507 DM	247	eval {
	248	# allow all, so that we can set new cpuset in /ns
	249	$all_cpus->write_to_cgroup("lxc/$vmid");
	250	eval {
	251	$newset->write_to_cgroup("lxc/$vmid/ns");
	252	};
	253	if (my $err = $@) {
b3f1adb2	254	warn $err if !$rebalance_error_count->{$vmid}++;
0b959507 DM	255	# restore original
	256	$cpuset->write_to_cgroup("lxc/$vmid");
	257	} else {
	258	# also apply to container root cgroup
	259	$newset->write_to_cgroup("lxc/$vmid");
b3f1adb2	260	$rebalance_error_count->{$vmid} = 0;
0b959507 DM	261	}
0b959507 DM	262	};
b3f1adb2 DM	263	if (my $err = $@) {
	264	warn $err if !$rebalance_error_count->{$vmid}++;
	265	}
0b959507 DM	266	};
0b959507 DM	267
e0dc09ad DM	268	my $ctlist = PVE::LXC::config_list();
	269
	270	foreach my $vmid (sort keys %$ctlist) {
	271	next if ! -d "/sys/fs/cgroup/cpuset/lxc/$vmid";
41db757b DM	272
	273	my ($conf, $cpuset);
	274	eval {
	275
	276	$conf = PVE::LXC::Config->load_config($vmid);
	277
	278	$cpuset = PVE::CpuSet->new_from_cgroup("lxc/$vmid");
	279	};
	280	if (my $err = $@) {
	281	warn $err;
	282	next;
	283	}
	284
	285	my @cpuset_members = $cpuset->members();
	286
8b750abc	287	if (!PVE::LXC::Config->has_lxc_entry($conf, 'lxc.cgroup.cpuset.cpus')) {
2499255b	288
8b750abc DM	289	my $cores = $conf->{cores} \|\| $cpucount;
8b750abc DM	290	$cores = $cpucount if $cores > $cpucount;
41db757b	291
2499255b DM	292	# see if the number of cores was hot-reduced or
	293	# hasn't been enacted at all yet
	294	my $newset = PVE::CpuSet->new();
	295	if ($cores < scalar(@cpuset_members)) {
	296	for (my $i = 0; $i < $cores; $i++) {
	297	$newset->insert($cpuset_members[$i]);
	298	}
	299	} elsif ($cores > scalar(@cpuset_members)) {
	300	my $count = $newset->insert(@cpuset_members);
	301	foreach my $cpu (@allowed_cpus) {
	302	$count += $newset->insert($cpu);
	303	last if $count >= $cores;
	304	}
	305	} else {
	306	$newset->insert(@cpuset_members);
	307	}
07f9595f	308
2499255b DM	309	# Apply hot-plugged changes if any:
	310	if (!$newset->is_equal($cpuset)) {
	311	@cpuset_members = $newset->members();
0b959507	312	$modify_cpuset->($vmid, $cpuset, $newset);
2499255b	313	}
07f9595f	314
2499255b DM	315	# Note: no need to rebalance if we already use all cores
2499255b DM	316	push @balanced_cts, [$vmid, $cores, $newset]
8b750abc	317	if defined($conf->{cores}) && ($cores != $cpucount);
2499255b	318	}
07f9595f	319
2499255b	320	foreach my $cpu (@cpuset_members) {
ccfff920	321	$cpu_ctcount[$cpu]++ if $cpu <= $max_cpuid;
07f9595f	322	}
2499255b	323	}
07f9595f	324
2499255b DM	325	my $find_best_cpu = sub {
2499255b DM	326	my ($cpulist, $cpu) = @_;
07f9595f	327
2499255b DM	328	my $cur_cost = $cpu_ctcount[$cpu];
2499255b DM	329	my $cur_cpu = $cpu;
41db757b	330
2499255b DM	331	foreach my $candidate (@$cpulist) {
	332	my $cost = $cpu_ctcount[$candidate];
	333	if ($cost < ($cur_cost -1)) {
	334	$cur_cost = $cost;
	335	$cur_cpu = $candidate;
	336	}
07f9595f DM	337	}
07f9595f DM	338
2499255b DM	339	return $cur_cpu;
	340	};
	341
	342	foreach my $bct (@balanced_cts) {
	343	my ($vmid, $cores, $cpuset) = @$bct;
41db757b DM	344
	345	my $newset = PVE::CpuSet->new();
	346
2499255b DM	347	my $rest = [];
	348	foreach my $cpu (@allowed_cpus) {
	349	next if $cpuset->has($cpu);
	350	push @$rest, $cpu;
	351	}
	352
	353	my @members = $cpuset->members();
	354	foreach my $cpu (@members) {
	355	my $best = &$find_best_cpu($rest, $cpu);
	356	if ($best != $cpu) {
	357	$cpu_ctcount[$best]++;
	358	$cpu_ctcount[$cpu]--;
	359	}
	360	$newset->insert($best);
41db757b DM	361	}
	362
	363	if (!$newset->is_equal($cpuset)) {
0b959507	364	$modify_cpuset->($vmid, $cpuset, $newset);
41db757b DM	365	}
	366	}
	367	}
	368
efd04666 DM	369	sub update_lxc_status {
	370	my ($status_cfg) = @_;
	371
	372	my $ctime = time();
	373
	374	my $vmstatus = PVE::LXC::vmstatus();
	375
	376	foreach my $vmid (keys %$vmstatus) {
	377	my $d = $vmstatus->{$vmid};
	378	my $template = $d->{template} ? $d->{template} : "0";
	379	my $data;
	380	if ($d->{status} eq 'running') { # running
	381	$data = "$d->{uptime}:$d->{name}:$d->{status}:$template:" .
	382	"$ctime:$d->{cpus}:$d->{cpu}:" .
	383	"$d->{maxmem}:$d->{mem}:" .
	384	"$d->{maxdisk}:$d->{disk}:" .
	385	"$d->{netin}:$d->{netout}:" .
	386	"$d->{diskread}:$d->{diskwrite}";
	387	} else {
	388	$data = "0:$d->{name}:$d->{status}:$template:$ctime:$d->{cpus}::" .
	389	"$d->{maxmem}::" .
	390	"$d->{maxdisk}:$d->{disk}:" .
	391	":::";
	392	}
	393	PVE::Cluster::broadcast_rrd("pve2.3-vm/$vmid", $data);
	394
	395	foreach my $id (keys %{$status_cfg->{ids}}) {
	396	my $plugin_config = $status_cfg->{ids}->{$id};
	397	next if $plugin_config->{disable};
	398	my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
	399	$plugin->update_lxc_status($plugin_config, $vmid, $d, $ctime);
	400	}
	401	}
	402	}
	403
	404	sub update_storage_status {
	405	my ($status_cfg) = @_;
	406
bbcfdc08	407	my $cfg = PVE::Storage::config();
efd04666 DM	408
	409	my $ctime = time();
	410
	411	my $info = PVE::Storage::storage_info($cfg);
	412
	413	foreach my $storeid (keys %$info) {
	414	my $d = $info->{$storeid};
	415	next if !$d->{active};
	416
	417	my $data = "$ctime:$d->{total}:$d->{used}";
	418
	419	my $key = "pve2-storage/${nodename}/$storeid";
	420	PVE::Cluster::broadcast_rrd($key, $data);
	421
	422	foreach my $id (keys %{$status_cfg->{ids}}) {
	423	my $plugin_config = $status_cfg->{ids}->{$id};
	424	next if $plugin_config->{disable};
	425	my $plugin = PVE::Status::Plugin->lookup($plugin_config->{type});
	426	$plugin->update_storage_status($plugin_config, $nodename, $storeid, $d, $ctime);
	427	}
	428	}
	429	}
	430
	431	sub update_status {
	432
	433	# update worker list. This is not really required and
	434	# we just call this to make sure that we have a correct
	435	# list in case of an unexpected crash.
	436	eval {
	437	my $tlist = PVE::RPCEnvironment::active_workers();
	438	PVE::Cluster::broadcast_tasklist($tlist);
	439	};
	440	my $err = $@;
	441	syslog('err', $err) if $err;
	442
	443	my $status_cfg = PVE::Cluster::cfs_read_file('status.cfg');
	444
	445	eval {
	446	update_node_status($status_cfg);
	447	};
	448	$err = $@;
	449	syslog('err', "node status update error: $err") if $err;
	450
	451	eval {
	452	update_qemu_status($status_cfg);
	453	};
	454	$err = $@;
	455	syslog('err', "qemu status update error: $err") if $err;
	456
	457	eval {
	458	update_lxc_status($status_cfg);
	459	};
	460	$err = $@;
	461	syslog('err', "lxc status update error: $err") if $err;
	462
e0dc09ad DM	463	eval {
	464	rebalance_lxc_containers();
	465	};
	466	$err = $@;
	467	syslog('err', "lxc cpuset rebalance error: $err") if $err;
	468
efd04666 DM	469	eval {
	470	update_storage_status($status_cfg);
	471	};
	472	$err = $@;
	473	syslog('err', "storage status update error: $err") if $err;
	474
	475	eval {
	476	remove_stale_lxc_consoles();
	477	};
	478	$err = $@;
	479	syslog('err', "lxc console cleanup error: $err") if $err;
	480	}
	481
	482	my $next_update = 0;
	483
	484	# do not update directly after startup, because install scripts
	485	# have a problem with that
	486	my $cycle = 0;
	487	my $updatetime = 10;
	488
	489	my $initial_memory_usage;
	490
	491	sub run {
	492	my ($self) = @_;
	493
	494	for (;;) { # forever
	495
	496	$next_update = time() + $updatetime;
	497
	498	if ($cycle) {
	499	my ($ccsec, $cusec) = gettimeofday ();
	500	eval {
	501	# syslog('info', "start status update");
	502	PVE::Cluster::cfs_update();
	503	update_status();
	504	};
	505	my $err = $@;
	506
	507	if ($err) {
	508	syslog('err', "status update error: $err");
	509	}
	510
	511	my ($ccsec_end, $cusec_end) = gettimeofday ();
	512	my $cptime = ($ccsec_end-$ccsec) + ($cusec_end - $cusec)/1000000;
	513
	514	syslog('info', sprintf("status update time (%.3f seconds)", $cptime))
	515	if ($cptime > 5);
	516	}
	517
	518	$cycle++;
	519
	520	my $mem = PVE::ProcFSTools::read_memory_usage();
	521
	522	if (!defined($initial_memory_usage) \|\| ($cycle < 10)) {
	523	$initial_memory_usage = $mem->{resident};
	524	} else {
	525	my $diff = $mem->{resident} - $initial_memory_usage;
	526	if ($diff > 510241024) {
	527	syslog ('info', "restarting server after $cycle cycles to " .
	528	"reduce memory usage (free $mem->{resident} ($diff) bytes)");
	529	$self->restart_daemon();
	530	}
	531	}
	532
533	my $wcount = 0;
534	while ((time() < $next_update) &&
535	($wcount < $updatetime) && # protect against time wrap
536	!$restart_request) { $wcount++; sleep (1); };
537
538	$self->restart_daemon() if $restart_request;
539	}
540	}
541
542	$daemon->register_start_command();
543	$daemon->register_restart_command(1);
544	$daemon->register_stop_command();
545	$daemon->register_status_command();
546
547	our $cmddef = {
548	start => [ __PACKAGE__, 'start', []],
549	restart => [ __PACKAGE__, 'restart', []],
550	stop => [ __PACKAGE__, 'stop', []],
551	status => [ __PACKAGE__, 'status', [], undef, sub { print shift . "\n";} ],
552	};
553
554	#my $cmd = shift;
555	#PVE::CLIHandler::handle_cmd($cmddef, $0, $cmd, \@ARGV, undef, $0);
556	#exit (0);
557
558	1;
559
560	__END__
561
562	=head1 NAME
563
564	pvestatd - PVE Status Daemon
565
566	=head1 SYNOPSIS
567
568	=include synopsis
569
570	=head1 DESCRIPTION
571
572	This daemom queries the status of VMs, storages and containers at
573	regular intervals. The result is sent to all nodes in the cluster.
574
575	=include pve_copyright
576
577
578
579
580