fix(lvg): align thin-pool and PV resize sizing with LVM

e2e/tests/cluster_config.yml

@@ -3,15 +3,15 @@ clusterDefinition:
   masters: # Master nodes configuration
     - hostname: "master-1"
       hostType: "vm"
-      osType: "Ubuntu 22.04 6.2.0-39-generic" # See internal/config/images.go
+      osType: "RedOS 8.0 6.6.26-1.red80.x86_64"
       cpu: 4
       coreFraction: 20
       ram: 8
       diskSize: 60
   workers: # Worker nodes configuration // TODO implement logic allowing to deploy different number of workes and masters with the same config.
     - hostname: "worker-1"
       hostType: "vm"
-      osType: "RedOS 8.0 6.6.26-1.red80.x86_64" # See internal/config/images.go
+      osType: "RedOS 8.0 6.6.26-1.red80.x86_64"
       cpu: 2
       coreFraction: 20
       ram: 8
@@ -39,7 +39,7 @@ clusterDefinition:
     #   diskSize: 20
     - hostname: "worker-5"
       hostType: "vm"
-      osType: "Ubuntu 24.04 6.8.0-53-generic" # See internal/config/images.go
+      osType: "Ubuntu 24.04 6.8.0-53-generic"
       cpu: 2
       coreFraction: 20
       ram: 8

e2e/tests/sds_node_configurator_helpers_cleanup_scheduler_test.go

@@ -77,6 +77,18 @@ func printLVMVolumeGroupInfo(lvg *v1alpha1.LVMVolumeGroup) {
 	GinkgoWriter.Println("=================================================\n")
 }
 
+func formatLVMVolumeGroupConditions(conditions []metav1.Condition) string {
+	if len(conditions) == 0 {
+		return "<none>"
+	}
+
+	parts := make([]string, 0, len(conditions))
+	for _, c := range conditions {
+		parts = append(parts, fmt.Sprintf("%s=%s/%s:%s", c.Type, c.Status, c.Reason, c.Message))
+	}
+	return strings.Join(parts, "; ")
+}
+
 func restartSDSNodeConfiguratorAgentOnNode(ctx context.Context, cl client.Client, nodeName string) {
 	const (
 		namespace = "d8-sds-node-configurator"

e2e/tests/sds_node_configurator_helpers_discovery_ssh_test.go

@@ -21,6 +21,7 @@ import (
 	"crypto/md5"
 	"crypto/sha1"
 	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"os"
 	"sort"
@@ -30,6 +31,9 @@ import (
 
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sclient "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
@@ -178,6 +182,98 @@ func e2eCountPVsOnNode(ctx context.Context, testKubeconfig *rest.Config, nodeNam
 	return n, out, nil
 }
 
+func getPVSizeViaDirectSSHWithRetry(ctx context.Context, testKubeconfig *rest.Config, nodeName, sshUser, pvPath string, maxRetries int, retryInterval time.Duration) (int64, error) {
+	var lastErr error
+	for attempt := 1; attempt <= maxRetries; attempt++ {
+		size, err := getPVSizeViaDirectSSH(ctx, testKubeconfig, nodeName, sshUser, pvPath)
+		if err == nil {
+			return size, nil
+		}
+		lastErr = err
+		if attempt < maxRetries {
+			GinkgoWriter.Printf("      pvs SSH to %s attempt %d/%d failed: %v; retry in %v\n", nodeName, attempt, maxRetries, err, retryInterval)
+			time.Sleep(retryInterval)
+		}
+	}
+
+	return 0, lastErr
+}
+
+func getPVSizeViaDirectSSH(ctx context.Context, testKubeconfig *rest.Config, nodeName, sshUser, pvPath string) (int64, error) {
+	type pvsReport struct {
+		Report []struct {
+			PV []struct {
+				PVName string `json:"pv_name"`
+				PVSize string `json:"pv_size"`
+			} `json:"pv"`
+		} `json:"report"`
+	}
+
+	out, err := e2eExecOnTestClusterNodeSSH(
+		ctx,
+		testKubeconfig,
+		nodeName,
+		sshUser,
+		fmt.Sprintf("sudo -n pvs --units B --nosuffix -o pv_name,pv_size --reportformat json %q 2>&1", pvPath),
+	)
+	if err != nil {
+		return 0, fmt.Errorf("run pvs for %s on node %s: %w; output:\n%s", pvPath, nodeName, err, out)
+	}
+
+	var report pvsReport
+	if err := json.Unmarshal([]byte(out), &report); err != nil {
+		return 0, fmt.Errorf("parse pvs output for %s on node %s: %w", pvPath, nodeName, err)
+	}
+	for _, r := range report.Report {
+		for _, pv := range r.PV {
+			if pv.PVName != pvPath {
+				continue
+			}
+
+			size, parseErr := strconv.ParseInt(strings.TrimSpace(pv.PVSize), 10, 64)
+			if parseErr != nil {
+				return 0, fmt.Errorf("parse pv_size %q for %s on node %s: %w", pv.PVSize, pvPath, nodeName, parseErr)
+			}
+
+			return size, nil
+		}
+	}
+
+	return 0, fmt.Errorf("PV %s not found in pvs output on node %s", pvPath, nodeName)
+}
+
+func countResizePVSuccessLogs(ctx context.Context, testKubeconfig *rest.Config, nodeName, pvPath string) (int, error) {
+	clientset, err := k8sclient.NewForConfig(testKubeconfig)
+	if err != nil {
+		return 0, fmt.Errorf("build kubernetes client for logs: %w", err)
+	}
+
+	pods, err := clientset.CoreV1().Pods("d8-sds-node-configurator").List(ctx, metav1.ListOptions{
+		LabelSelector: "app=sds-node-configurator",
+		FieldSelector: "spec.nodeName=" + nodeName,
+	})
+	if err != nil {
+		return 0, fmt.Errorf("list sds-node-configurator pods on node %s: %w", nodeName, err)
+	}
+	if len(pods.Items) == 0 {
+		return 0, fmt.Errorf("no sds-node-configurator pod found on node %s", nodeName)
+	}
+
+	sort.Slice(pods.Items, func(i, j int) bool {
+		return pods.Items[i].CreationTimestamp.Before(&pods.Items[j].CreationTimestamp)
+	})
+
+	podName := pods.Items[len(pods.Items)-1].Name
+	data, err := clientset.CoreV1().Pods("d8-sds-node-configurator").GetLogs(podName, &corev1.PodLogOptions{
+		Container: "sds-node-configurator-agent",
+	}).DoRaw(ctx)
+	if err != nil {
+		return 0, fmt.Errorf("read logs for pod %s: %w", podName, err)
+	}
+
+	return strings.Count(string(data), fmt.Sprintf("[ResizePVIfNeeded] successfully resized PV %s", pvPath)), nil
+}
+
 // e2eCountPVsInVGOnNode returns how many PVs belong to vgName according to pvs on the node.
 func e2eCountPVsInVGOnNode(ctx context.Context, testKubeconfig *rest.Config, nodeName, sshUser, vgName string) (int, string, error) {
 	quotedVG := strconv.Quote(vgName)