tiago/pve-storage
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

plugin restructure to add cycle with time

Browse source
This commit is contained in:
Tiago Sousa 2025-09-26 21:41:13 +01:00
parent c952cdf3be
commit b19c70d0ea
1 changed files with 110 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

163
src/PVE/Service/pvestord.pm
View file

@ -3,6 +3,7 @@ package PVE::Service::pvestord;
use strict;
use warnings;
use Time::HiRes qw (gettimeofday);
use PVE::SafeSyslog;
use PVE::Daemon;
use PVE::Cluster qw(cfs_read_file);
@ -43,77 +44,133 @@ my sub get_drive_id {
my sub dequeue {
my ($queue) = @_;
PVE::Storage::lock_extend_queue(sub {
# TODO: This will have to have some sort of mechanism
# to make sure that the element that is removed is the one
# that this node handled
shift @$queue;
PVE::Storage::lock_extend_queue(
sub {
# TODO: This will have to have some sort of mechanism
# to make sure that the element that is removed is the one
# that this node handled
shift @$queue;
syslog("info", "queue after dequeue $queue\n");
PVE::Storage::write_extend_queue($queue);
},"Could not lock extend queue file");
syslog("info", "queue after dequeue $queue\n");
PVE::Storage::write_extend_queue($queue);
},
"Could not lock extend queue file",
);
}
my sub get_wr_highest_offset {
my ($block_stats, $drive_id) = @_;
my $file_blockdev = $block_stats->{$drive_id}->{parent}->{parent};
return $file_blockdev->{stats}->{wr_highest_offset};
sub perform_extend {
my $storecfg = PVE::Storage::config();
my $queue = PVE::Storage::extend_queue();
my $first_extend_request = @$queue[0];
return if !$first_extend_request;
my ($vmid, $blockdev_nodename) = @$first_extend_request;
my $vmlist = PVE::Cluster::get_vmlist();
my $owner_nodename = $vmlist->{ids}->{$vmid}->{node};
if ($owner_nodename eq $nodename) {
my $running = PVE::QemuServer::Helpers::vm_running_locally($vmid);
# NOTE: the way that the blockdev nodename is generated (sha256),
# makes it impossible to dehash it and get the disk which it references.
# So it's needed to get the blockstats and retrieve which disk has this
# nodename. The problem with this is that blockstats are only
# available when the machine is on
# Could the get_node_name() function be changed from sha256 digest
# to a reversible encoding like base64?
my $extend_function = sub {
dequeue($queue);
syslog("info", "Processsing extend request $vmid: $blockdev_nodename\n");
my $block_stats = PVE::QemuServer::Blockdev::get_block_stats($vmid);
my $drive_id = get_drive_id($block_stats, $blockdev_nodename);
if (!$drive_id) {
syslog("err", "Couldn't find drive_id for blockdev $blockdev_nodename");
return;
}
my $vm_conf = PVE::QemuConfig->load_config($vmid);
my $drive = PVE::QemuServer::parse_drive($drive_id, $vm_conf->{$drive_id});
my $volid = $drive->{file};
PVE::QemuServer::Blockdev::underlay_resize(
$storecfg, $vmid, $drive_id, $volid
);
};
PVE::QemuConfig->lock_config($vmid, $extend_function);
}
}
my $next_update = 0;
my $cycle = 0;
my $restart_request = 0;
my $initial_memory_usage = 0;
# 1 second cycles
my $updatetime = 1;
sub run {
my ($self) = @_;
syslog("info", "Running on node $nodename\n");
my $storecfg = PVE::Storage::config();
for (;;) { # forever
# get next extend request
my $queue = PVE::Storage::extend_queue();
syslog("info", "Queue data: " . Dumper($queue) . "\n");
my $first_extend_request = @$queue[0];
if (!$first_extend_request) {
# wait for next cycle
sleep(60);
next;
}
my ($vmid, $blockdev_nodename) = @$first_extend_request;
syslog("info", "Extend request $vmid: $blockdev_nodename\n");
$next_update = time() + $updatetime;
# if it is for a local machine
my $vmlist = PVE::Cluster::get_vmlist();
my $running_nodename = $vmlist->{ids}->{$vmid}->{node};
# if not ignore it
if ($running_nodename eq $nodename) {
my $running = PVE::QemuServer::Helpers::vm_running_locally($vmid);
my $block_stats = PVE::QemuServer::Blockdev::get_block_stats($vmid);
# TODO: Lock VM
if ($cycle) {
my ($ccsec, $cusec) = gettimeofday();
eval {
# syslog('info', "start status update");
PVE::Cluster::cfs_update();
perform_extend();
};
my $err = $@;
my $drive_id = get_drive_id($block_stats, $blockdev_nodename);
if (!$drive_id) {
syslog("error", "Couldn't find drive_id for blockdev $blockdev_nodename");
# FIX: loop cycle delay
sleep(60);
next;
if ($err) {
syslog('err', "status update error: $err");
}
my $wr_highest_offset = get_wr_highest_offset($block_stats, $drive_id);
my $vm_conf = PVE::QemuConfig->load_config($vmid);
my $drive = PVE::QemuServer::parse_drive($drive_id, $vm_conf->{$drive_id});
syslog("info", Dumper($drive) . "\n");
my ($ccsec_end, $cusec_end) = gettimeofday();
my $cptime = ($ccsec_end - $ccsec) + ($cusec_end - $cusec) / 1000000;
my $volid = $drive->{file};
my ($storeid) = PVE::Storage::parse_volume_id($volid);
my $scfg = PVE::Storage::storage_config($storecfg, $storeid);
my $backing_snap = undef;
PVE::QemuServer::Blockdev::underlay_resize($storecfg, $vmid, $drive_id, $volid, 'snap');
syslog("info", "found $volid with config: " . Dumper($scfg) . "\n");
syslog('info', sprintf("extend process time (%.3f seconds)", $cptime))
if ($cptime > 1);
}
dequeue($queue);
# sleep(60);
die "everything acording to plan";
$cycle++;
my $mem = PVE::ProcFSTools::read_memory_usage();
my $resident_kb = $mem->{resident} / 1024;
if (!defined($initial_memory_usage) || ($cycle < 10)) {
$initial_memory_usage = $resident_kb;
} else {
my $diff = $resident_kb - $initial_memory_usage;
if ($diff > 15 * 1024) {
syslog(
'info',
"restarting server after $cycle cycles to "
. "reduce memory usage (free $resident_kb ($diff) KB)",
);
$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;
}
}