[PATCH e2e-tests 25/38] test framework: add Ceph test helpers

["Michael Köppl" <m.koeppl@proxmox.com>]

Add helpers to set up a hyper-converged Ceph cluster on the cluster
nodes and to wait until it is ready. Ceph needs the PVE cluster to
exist, but post-setup scripts run before it is created, and the
tests roll back to the state after setup. So every Ceph test sets up
its own Ceph cluster with these helpers. The setup is based mostly on
the documentation [0].

The cluster uses dedicated public and cluster networks on the second
NIC of the nodes, as the documentation recommends keeping Ceph traffic
off the network of the Proxmox VE cluster. Without a network, the
monitors fall back to the node's cluster address.

The OSD memory target is lowered to 1 GiB, as the test nodes only have
4 GiB of memory for two OSDs, a monitor, and a manager.

[0] https://pve.proxmox.com/wiki/Deploy_Hyper-Converged_Ceph_Cluster

Signed-off-by: Michael Köppl <m.koeppl@proxmox.com>
---
 Proxmox/Test/Ceph.pm | 338 +++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 338 insertions(+)
 create mode 100644 Proxmox/Test/Ceph.pm

diff --git a/Proxmox/Test/Ceph.pm b/Proxmox/Test/Ceph.pm
new file mode 100644
index 0000000..357e0ed
--- /dev/null
+++ b/Proxmox/Test/Ceph.pm
@@ -0,0 +1,338 @@
+package Proxmox::Test::Ceph;
+use strict;
+use warnings FATAL => 'all';
+
+use parent 'Exporter';
+our @EXPORT_OK = qw(
+    CEPH_PUBLIC_NETWORK
+    CEPH_CLUSTER_NETWORK
+    ceph_disks
+    ceph_osds
+    ceph_status
+    wait_for_ceph
+    check_storage_active
+    setup_ceph_network
+    init_ceph
+    create_monitors
+    create_managers
+    create_osds
+    create_ceph_cluster
+    create_pool
+);
+
+use JSON;
+use Test::More;
+
+use Proxmox::Test::Util qw(poll_until);
+
+# Lower the OSD memory target from the default of 4 GiB, which would otherwise leave the test nodes
+# without memory for anything else.
+use constant OSD_MEMORY_TARGET => 1024 * 1024 * 1024;
+
+# Ceph traffic is kept off the network of the Proxmox VE cluster, as recommended for hyper-converged
+# setups. The Ceph networks use the NIC that the plan attaches to the 'tt_test' bridge.
+use constant CEPH_NIC => 'ens19';
+use constant CEPH_PUBLIC_PREFIX => '10.250.240';
+use constant CEPH_CLUSTER_PREFIX => '10.250.241';
+use constant CEPH_PUBLIC_NETWORK => CEPH_PUBLIC_PREFIX . '.0/24';
+use constant CEPH_CLUSTER_NETWORK => CEPH_CLUSTER_PREFIX . '.0/24';
+
+# Disks with serials ceph0, ceph1, ... are supplied by the plan.
+sub ceph_disks {
+    my ($instance) = @_;
+
+    my @disks;
+    for my $disk (($instance->get_additional_disks() // [])->@*) {
+        next if ($disk->{serial} // '') !~ m/^ceph\d+$/;
+
+        my $path = $instance->get_disk_by_serial($disk->{serial})
+            // die "no disk with serial '$disk->{serial}' on " . $instance->get_nodename() . "\n";
+        push @disks, $path;
+    }
+
+    return \@disks;
+}
+
+# The OSDs on the node of $instance, as listed in the OSD tree of the API.
+sub ceph_osds {
+    my ($instance) = @_;
+
+    my $node = $instance->get_nodename();
+    my $tree = $instance->client()->get("/nodes/$node/ceph/osd", {});
+
+    my @buckets = ($tree->{root});
+    while (my $bucket = shift @buckets) {
+        my $children = $bucket->{children} // [];
+        if (($bucket->{type} // '') eq 'host' && $bucket->{name} eq $node) {
+            return [grep { $_->{type} eq 'osd' } $children->@*];
+        }
+        push @buckets, $children->@*;
+    }
+
+    return [];
+}
+
+# The output of 'ceph status' on $instance.
+sub ceph_status {
+    my ($instance) = @_;
+
+    my ($out) = $instance->run_command('ceph status --format json');
+    return decode_json($out);
+}
+
+# Wait until the Ceph cluster seen from $instance has $opts{monitors} monitors in quorum,
+# $opts{osds} OSDs that are all up and in, and all placement groups active+clean. If
+# $opts{health_ok} is set as well, also wait for HEALTH OK.
+# Returns the final status.
+sub wait_for_ceph {
+    my ($instance, %opts) = @_;
+
+    my $monitors = $opts{monitors} // 0;
+    my $osds = $opts{osds} // 0;
+    my $last;
+
+    my $ready = eval {
+        poll_until(
+            sub {
+                $last = ceph_status($instance);
+
+                my $health = $last->{health}->{status} // '';
+                return 0 if $health eq 'HEALTH_ERR';
+                return 0 if $opts{health_ok} && $health ne 'HEALTH_OK';
+
+                return 0 if scalar(($last->{quorum_names} // [])->@*) < $monitors;
+
+                my $osdmap = $last->{osdmap} // {};
+                return 0 if ($osdmap->{num_osds} // 0) < $osds;
+                return 0 if ($osdmap->{num_up_osds} // 0) != ($osdmap->{num_osds} // 0);
+                return 0 if ($osdmap->{num_in_osds} // 0) != ($osdmap->{num_osds} // 0);
+
+                my $pgmap = $last->{pgmap} // {};
+                my $clean = 0;
+                for my $state (($pgmap->{pgs_by_state} // [])->@*) {
+                    $clean += $state->{count} if $state->{state_name} eq 'active+clean';
+                }
+                return $clean == ($pgmap->{num_pgs} // 0) ? $last : 0;
+            },
+            timeout => $opts{timeout} // 600,
+            interval => 5,
+            description => $opts{description} // 'the Ceph cluster to become ready',
+        );
+    };
+
+    if (!$ready) {
+        my $err = $@;
+        diag('last Ceph health: ' . encode_json($last->{health} // {})) if $last;
+        die $err;
+    }
+
+    return $ready;
+}
+
+# Check that $storage is active on $instance. If it is not, print the result of `pvesm status`.
+sub check_storage_active {
+    my ($instance, $storage) = @_;
+
+    my $node = $instance->get_nodename();
+    my $status = $instance->client()->get("/nodes/$node/storage/$storage/status", {});
+    return 1 if ok($status->{active}, "storage '$storage' is active on node '$node'");
+
+    my ($out, $err) = eval { $instance->run_command("pvesm status --storage $storage") };
+    diag("'pvesm status --storage $storage' on '$node':\n"
+        . join('', grep { defined } $out, $err, $@));
+    return 0;
+}
+
+# Give every instance in $instances an address in the Ceph public and the Ceph cluster network,
+# numbered in the order of $instances, and wait until the instances reach each other on both.
+# Returns the addresses as { <nodename> => { public => ..., cluster => ... } }.
+sub setup_ceph_network {
+    my ($instances) = @_;
+
+    my $nic = CEPH_NIC;
+    my $addresses = {};
+    for my $i (0 .. $instances->$#*) {
+        my $instance = $instances->[$i];
+        my $public = CEPH_PUBLIC_PREFIX . '.' . ($i + 1);
+        my $cluster = CEPH_CLUSTER_PREFIX . '.' . ($i + 1);
+
+        $instance->run_command("ip link set $nic up && ip addr add $public/24 dev $nic"
+            . " && ip addr add $cluster/24 dev $nic");
+        $addresses->{ $instance->get_nodename() } = { public => $public, cluster => $cluster };
+    }
+
+    for my $instance ($instances->@*) {
+        my $node = $instance->get_nodename();
+        for my $other (grep { $_ ne $node } sort keys $addresses->%*) {
+            for my $address (sort values $addresses->{$other}->%*) {
+                poll_until(
+                    sub { $instance->node_can_ping($address) },
+                    timeout => 30,
+                    interval => 2,
+                    description => "'$node' to reach '$other' at $address",
+                );
+            }
+        }
+    }
+
+    return $addresses;
+}
+
+# Write the initial Ceph configuration through the first instance in $instances for the whole
+# cluster. %opts is passed on as additional parameters of the init call.
+sub init_ceph {
+    my ($instances, %opts) = @_;
+
+    my $first = $instances->[0];
+    my $node = $first->get_nodename();
+
+    $first->client()->post("/nodes/$node/ceph/init", \%opts);
+}
+
+# Prints the end of the monitor logs to show why creating a monitor or getting quorum failed.
+my $diag_monitor_logs = sub {
+    my ($instances) = @_;
+
+    for my $instance ($instances->@*) {
+        my $node = $instance->get_nodename();
+        my ($log) =
+            eval { $instance->run_command("tail -n 30 /var/log/ceph/ceph-mon.$node.log") };
+        diag("end of the monitor log on '$node':\n" . ($log // "not available: $@"));
+    }
+};
+
+# Create a monitor on every instance in $instances, one at a time, and wait until each new monitor
+# has joined the quorum before creating the next one. The first monitor also gets a manager.
+sub create_monitors {
+    my ($instances) = @_;
+
+    my $first = $instances->[0];
+    my @with_monitor;
+
+    for my $instance ($instances->@*) {
+        my $node = $instance->get_nodename();
+
+        eval {
+            # Before creating another monitor, ensure connectivity to the existing ones.
+            if (@with_monitor) {
+                poll_until(
+                    sub {
+                        eval {
+                            $instance->run_command(
+                                'ceph --conf /etc/pve/ceph.conf --connect-timeout 5 status'
+                                    . ' >/dev/null');
+                            1;
+                        };
+                    },
+                    timeout => 60,
+                    interval => 2,
+                    description => "'$node' to connect to the existing monitors",
+                );
+            }
+
+            my $upid = $instance->client()->post("/nodes/$node/ceph/mon/$node", {});
+            $instance->task_ok($upid, "created monitor on '$node'")
+                or die "creating the monitor on '$node' failed\n";
+
+            poll_until(
+                sub {
+                    grep { $_ eq $node } (ceph_status($first)->{quorum_names} // [])->@*;
+                },
+                timeout => 120,
+                interval => 2,
+                description => "monitor '$node' to join the quorum",
+            );
+        };
+        if (my $err = $@) {
+            $diag_monitor_logs->([@with_monitor, $instance]);
+            die $err;
+        }
+
+        push @with_monitor, $instance;
+    }
+}
+
+# Create a manager on every instance in $instances that does not have one yet.
+sub create_managers {
+    my ($instances) = @_;
+
+    for my $instance ($instances->@*) {
+        my $node = $instance->get_nodename();
+
+        my $managers = $instance->client()->get("/nodes/$node/ceph/mgr", {});
+        next if grep { ($_->{name} // '') eq $node } $managers->@*;
+
+        my $upid = $instance->client()->post("/nodes/$node/ceph/mgr/$node", {});
+        $instance->task_ok($upid, "created manager on '$node'")
+            or die "creating the manager on '$node' failed\n";
+    }
+}
+
+# Create an OSD on every Ceph disk of every instance in $instances. %opts is passed on as
+# additional parameters of the OSD create call. Returns the number of OSDs created.
+sub create_osds {
+    my ($instances, %opts) = @_;
+
+    $instances->[0]->run_command('ceph config set osd osd_memory_target ' . OSD_MEMORY_TARGET);
+
+    my $count = 0;
+    for my $instance ($instances->@*) {
+        my $node = $instance->get_nodename();
+
+        for my $dev (ceph_disks($instance)->@*) {
+            my $upid =
+                $instance->client()->post("/nodes/$node/ceph/osd", { dev => $dev, %opts });
+            $instance->task_ok($upid, "created OSD on '$node' for '$dev'")
+                or die "creating the OSD on '$node' for '$dev' failed\n";
+            $count++;
+        }
+    }
+
+    return $count;
+}
+
+# Set up a Ceph cluster on the Ceph public and cluster network with a monitor, a manager and OSDs on
+# every instance in $instances, and wait until it is ready. The Ceph packages must already be
+# installed. Returns the number of OSDs.
+sub create_ceph_cluster {
+    my ($instances) = @_;
+
+    setup_ceph_network($instances);
+    init_ceph(
+        $instances,
+        network => CEPH_PUBLIC_NETWORK,
+        'cluster-network' => CEPH_CLUSTER_NETWORK,
+    );
+    create_monitors($instances);
+    create_managers($instances);
+    my $osds = create_osds($instances);
+
+    wait_for_ceph($instances->[0], monitors => scalar($instances->@*), osds => $osds);
+
+    return $osds;
+}
+
+# Create the replicated pool $name through $instance and wait for its placement groups. With
+# $opts{add_storages}, an RBD storage of the same name is added for the pool.
+sub create_pool {
+    my ($instance, $name, %opts) = @_;
+
+    my $node = $instance->get_nodename();
+    my $upid = $instance->client()
+        ->post("/nodes/$node/ceph/pool", { name => $name, pg_autoscale_mode => "on", %opts });
+    $instance->task_ok($upid, "created pool '$name'") or die "creating pool '$name' failed\n";
+
+    poll_until(
+        sub {
+            my ($out) = $instance->run_command("ceph df --format json");
+            return grep { $_->{name} eq $name } (decode_json($out)->{pools} // [])->@*;
+        },
+        timeout => 60,
+        interval => 2,
+        description => "pool '$name' to show up in the usage stats",
+    );
+
+    wait_for_ceph($instance);
+}
+
+1;
-- 
2.47.3