First KVM feature extractor and scheduler (#31)

PhilippMatthes · web-flow · commit a82018e58daf · 2025-03-05T06:56:49.000Z
diff --git a/helm/cortex/values.yaml b/helm/cortex/values.yaml
@@ -241,6 +241,13 @@ conf:
             prometheus:
               metrics:
                 - vrops_hostsystem_cpu_contention_percentage
+      # KVM-specific extractors.
+      - name: node_exporter_host_cpu_usage_extractor
+        dependencies:
+          sync:
+            prometheus:
+              metrics:
+                - node_exporter_cpu_usage_pct
 
   scheduler:
     # The port to use for the scheduler API.
@@ -274,3 +281,13 @@ conf:
           features:
             extractors:
               - vrops_hostsystem_contention_extractor
+      # KVM specific scheduler steps.
+      - name: kvm_avoid_overloaded_hosts
+        options:
+          avgCPUUsageThreshold: 10
+          maxCPUUsageThreshold: 20
+          activationOnHit: -1.0
+        dependencies:
+          features:
+            extractors:
+              - node_exporter_host_cpu_usage_extractor
diff --git a/internal/features/pipeline.go b/internal/features/pipeline.go
@@ -10,16 +10,20 @@ import (
 	"github.com/cobaltcore-dev/cortex/internal/conf"
 	"github.com/cobaltcore-dev/cortex/internal/db"
 	"github.com/cobaltcore-dev/cortex/internal/features/plugins"
+	"github.com/cobaltcore-dev/cortex/internal/features/plugins/kvm"
 	"github.com/cobaltcore-dev/cortex/internal/features/plugins/vmware"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
 // Configuration of feature extractors supported by the scheduler.
 // The actual features to extract are defined in the configuration file.
 var supportedExtractors = []plugins.FeatureExtractor{
+	// VMware-specific extractors
 	&vmware.VROpsHostsystemResolver{},
 	&vmware.VROpsProjectNoisinessExtractor{},
 	&vmware.VROpsHostsystemContentionExtractor{},
+	// KVM-specific extractors
+	&kvm.NodeExporterHostCPUUsageExtractor{},
 }
 
 // Pipeline that contains multiple feature extractors and executes them.
diff --git a/internal/features/plugins/kvm/node_exporter_host_cpu_usage.go b/internal/features/plugins/kvm/node_exporter_host_cpu_usage.go
@@ -0,0 +1,53 @@
+// Copyright 2025 SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package kvm
+
+import (
+	"github.com/cobaltcore-dev/cortex/internal/features/plugins"
+)
+
+// Feature that maps CPU usage of kvm hosts.
+type NodeExporterHostCPUUsage struct {
+	ComputeHost string  `db:"compute_host"`
+	AvgCPUUsage float64 `db:"avg_cpu_usage"`
+	MaxCPUUsage float64 `db:"max_cpu_usage"`
+}
+
+// Table under which the feature is stored.
+func (NodeExporterHostCPUUsage) TableName() string {
+	return "feature_host_cpu_usage"
+}
+
+// Extractor that extracts CPU usage of kvm hosts and stores
+// it as a feature into the database.
+type NodeExporterHostCPUUsageExtractor struct {
+	// Common base for all extractors that provides standard functionality.
+	plugins.BaseExtractor[
+		struct{},                 // No options passed through yaml config
+		NodeExporterHostCPUUsage, // Feature model
+	]
+}
+
+// Name of this feature extractor that is used in the yaml config, for logging etc.
+func (*NodeExporterHostCPUUsageExtractor) GetName() string {
+	return "node_exporter_host_cpu_usage_extractor"
+}
+
+// Extract CPU usage of kvm hosts.
+// Depends on resolved kvm hosts (feature_resolved_host).
+func (e *NodeExporterHostCPUUsageExtractor) Extract() ([]plugins.Feature, error) {
+	var features []NodeExporterHostCPUUsage
+	if _, err := e.DB.Select(&features, `
+		SELECT
+			node AS compute_host,
+			AVG(value) AS avg_cpu_usage,
+			MAX(value) AS max_cpu_usage
+		FROM node_exporter_metrics
+		WHERE name = 'node_exporter_cpu_usage_pct'
+		GROUP BY node;
+	`); err != nil {
+		return nil, err
+	}
+	return e.Extracted(features)
+}
diff --git a/internal/features/plugins/kvm/node_exporter_host_cpu_usage_test.go b/internal/features/plugins/kvm/node_exporter_host_cpu_usage_test.go
@@ -0,0 +1,95 @@
+// Copyright 2025 SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package kvm
+
+import (
+	"testing"
+
+	"github.com/cobaltcore-dev/cortex/internal/conf"
+	"github.com/cobaltcore-dev/cortex/internal/db"
+	"github.com/cobaltcore-dev/cortex/internal/sync/prometheus"
+	testlibDB "github.com/cobaltcore-dev/cortex/testlib/db"
+)
+
+func TestNodeExporterHostCPUUsageExtractor_Init(t *testing.T) {
+	dbEnv := testlibDB.SetupDBEnv(t)
+	testDB := db.DB{DbMap: dbEnv.DbMap}
+	defer testDB.Close()
+	defer dbEnv.Close()
+
+	extractor := &NodeExporterHostCPUUsageExtractor{}
+	if err := extractor.Init(testDB, conf.NewRawOpts("")); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	if !testDB.TableExists(NodeExporterHostCPUUsage{}) {
+		t.Error("expected table to be created")
+	}
+}
+
+func TestNodeExporterHostCPUUsageExtractor_Extract(t *testing.T) {
+	dbEnv := testlibDB.SetupDBEnv(t)
+	testDB := db.DB{DbMap: dbEnv.DbMap}
+	defer testDB.Close()
+	defer dbEnv.Close()
+
+	// Create dependency tables
+	if err := testDB.CreateTable(
+		testDB.AddTable(prometheus.NodeExporterMetric{}),
+	); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// Insert mock data into the node_exporter_metrics table
+	_, err := testDB.Exec(`
+        INSERT INTO node_exporter_metrics (node, name, value)
+        VALUES
+            ('node1', 'node_exporter_cpu_usage_pct', 20.0),
+            ('node2', 'node_exporter_cpu_usage_pct', 30.0),
+            ('node1', 'node_exporter_cpu_usage_pct', 40.0)
+    `)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	extractor := &NodeExporterHostCPUUsageExtractor{}
+	if err := extractor.Init(testDB, conf.NewRawOpts("")); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	if _, err = extractor.Extract(); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// Verify the data was inserted into the feature_host_cpu_usage table
+	var usages []NodeExporterHostCPUUsage
+	_, err = testDB.Select(&usages, "SELECT * FROM feature_host_cpu_usage")
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	if len(usages) != 2 {
+		t.Errorf("expected 2 rows, got %d", len(usages))
+	}
+	expected := map[string]struct {
+		AvgCPUUsage float64
+		MaxCPUUsage float64
+	}{
+		"node1": {AvgCPUUsage: 30.0, MaxCPUUsage: 40.0}, // Average of 20.0 and 40.0, Max of 40.0
+		"node2": {AvgCPUUsage: 30.0, MaxCPUUsage: 30.0}, // Single value of 30.0
+	}
+	for _, u := range usages {
+		if expected[u.ComputeHost].AvgCPUUsage != u.AvgCPUUsage {
+			t.Errorf(
+				"expected avg_cpu_usage for compute_host %s to be %f, got %f",
+				u.ComputeHost, expected[u.ComputeHost].AvgCPUUsage, u.AvgCPUUsage,
+			)
+		}
+		if expected[u.ComputeHost].MaxCPUUsage != u.MaxCPUUsage {
+			t.Errorf(
+				"expected max_cpu_usage for compute_host %s to be %f, got %f",
+				u.ComputeHost, expected[u.ComputeHost].MaxCPUUsage, u.MaxCPUUsage,
+			)
+		}
+	}
+}
diff --git a/internal/scheduler/pipeline.go b/internal/scheduler/pipeline.go
@@ -14,14 +14,18 @@ import (
 	"github.com/cobaltcore-dev/cortex/internal/conf"
 	"github.com/cobaltcore-dev/cortex/internal/db"
 	"github.com/cobaltcore-dev/cortex/internal/scheduler/plugins"
+	"github.com/cobaltcore-dev/cortex/internal/scheduler/plugins/kvm"
 	"github.com/cobaltcore-dev/cortex/internal/scheduler/plugins/vmware"
 )
 
 // Configuration of steps supported by the scheduler.
 // The steps actually used by the scheduler are defined through the configuration file.
 var supportedSteps = []plugins.Step{
+	// VMware-specific steps
 	&vmware.AntiAffinityNoisyProjectsStep{},
 	&vmware.AvoidContendedHostsStep{},
+	// KVM-specific steps
+	&kvm.AvoidOverloadedHostsStep{},
 }
 
 // Sequence of scheduler steps that are executed in parallel.
diff --git a/internal/scheduler/plugins/kvm/avoid_overloaded_hosts.go b/internal/scheduler/plugins/kvm/avoid_overloaded_hosts.go
@@ -0,0 +1,58 @@
+// Copyright 2025 SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package kvm
+
+import (
+	"github.com/cobaltcore-dev/cortex/internal/features/plugins/kvm"
+	"github.com/cobaltcore-dev/cortex/internal/scheduler/plugins"
+)
+
+// Options for the scheduling step, given through the
+// step config in the service yaml file.
+type AvoidOverloadedHostsStepOpts struct {
+	AvgCPUUsageThreshold float64 `yaml:"avgCPUUsageThreshold"`
+	MaxCPUUsageThreshold float64 `yaml:"maxCPUUsageThreshold"`
+	ActivationOnHit      float64 `yaml:"activationOnHit"`
+}
+
+// Step to avoid high cpu hosts by downvoting them.
+type AvoidOverloadedHostsStep struct {
+	// BaseStep is a helper struct that provides common functionality for all steps.
+	plugins.BaseStep[AvoidOverloadedHostsStepOpts]
+}
+
+// Get the name of this step, used for identification in config, logs, metrics, etc.
+func (s *AvoidOverloadedHostsStep) GetName() string {
+	return "kvm_avoid_overloaded_hosts"
+}
+
+// Downvote hosts that have high cpu load.
+func (s *AvoidOverloadedHostsStep) Run(scenario plugins.Scenario) (map[string]float64, error) {
+	activations := s.BaseStep.BaseActivations(scenario)
+	if scenario.GetVMware() {
+		// Don't run this step for VMware VMs.
+		return activations, nil
+	}
+
+	var highlyUsedHosts []kvm.NodeExporterHostCPUUsage
+	if _, err := s.DB.Select(&highlyUsedHosts, `
+		SELECT * FROM feature_host_cpu_usage
+		WHERE avg_cpu_usage > :avg_cpu_usage_threshold
+		OR max_cpu_usage > :max_cpu_usage_threshold
+	`, map[string]any{
+		"avg_cpu_usage_threshold": s.Options.AvgCPUUsageThreshold,
+		"max_cpu_usage_threshold": s.Options.MaxCPUUsageThreshold,
+	}); err != nil {
+		return nil, err
+	}
+
+	// Push the VM away from highly used hosts.
+	for _, host := range highlyUsedHosts {
+		// Only modify the weight if the host is in the scenario.
+		if _, ok := activations[host.ComputeHost]; ok {
+			activations[host.ComputeHost] = s.Options.ActivationOnHit
+		}
+	}
+	return activations, nil
+}
diff --git a/internal/scheduler/plugins/kvm/avoid_overloaded_hosts_test.go b/internal/scheduler/plugins/kvm/avoid_overloaded_hosts_test.go
@@ -0,0 +1,120 @@
+// Copyright 2025 SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package kvm
+
+import (
+	"testing"
+
+	"github.com/cobaltcore-dev/cortex/internal/conf"
+	"github.com/cobaltcore-dev/cortex/internal/db"
+	"github.com/cobaltcore-dev/cortex/internal/features/plugins/kvm"
+	"github.com/cobaltcore-dev/cortex/internal/scheduler/plugins"
+	testlibDB "github.com/cobaltcore-dev/cortex/testlib/db"
+	testlibPlugins "github.com/cobaltcore-dev/cortex/testlib/scheduler/plugins"
+)
+
+func TestAvoidOverloadedHostsStep_Run(t *testing.T) {
+	dbEnv := testlibDB.SetupDBEnv(t)
+	testDB := db.DB{DbMap: dbEnv.DbMap}
+	defer testDB.Close()
+	defer dbEnv.Close()
+
+	// Create dependency tables
+	err := testDB.CreateTable(testDB.AddTable(kvm.NodeExporterHostCPUUsage{}))
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// Insert mock data into the feature_host_cpu_usage table
+	_, err = testDB.Exec(`
+        INSERT INTO feature_host_cpu_usage (compute_host, avg_cpu_usage, max_cpu_usage)
+        VALUES
+            ('host1', 15.0, 25.0),
+            ('host2', 5.0, 10.0),
+            ('host3', 20.0, 30.0)
+    `)
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	// Create an instance of the step
+	opts := conf.NewRawOpts(`
+        avgCPUUsageThreshold: 10.0
+        maxCPUUsageThreshold: 20.0
+        activationOnHit: -1.0
+    `)
+	step := &AvoidOverloadedHostsStep{}
+	if err := step.Init(testDB, opts); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	tests := []struct {
+		name           string
+		scenario       plugins.Scenario
+		downvotedHosts map[string]struct{}
+	}{
+		{
+			name: "Non-vmware vm",
+			scenario: &testlibPlugins.MockScenario{
+				VMware: false,
+				Hosts: []testlibPlugins.MockScenarioHost{
+					{ComputeHost: "host1", HypervisorHostname: "hypervisor1"},
+					{ComputeHost: "host2", HypervisorHostname: "hypervisor2"},
+					{ComputeHost: "host3", HypervisorHostname: "hypervisor3"},
+				},
+			},
+			// Should downvote hosts with high CPU usage
+			downvotedHosts: map[string]struct{}{
+				"host1": {},
+				"host3": {},
+			},
+		},
+		{
+			name: "VMware vm",
+			scenario: &testlibPlugins.MockScenario{
+				VMware: true,
+				Hosts: []testlibPlugins.MockScenarioHost{
+					{ComputeHost: "host1", HypervisorHostname: "hypervisor1"},
+					{ComputeHost: "host2", HypervisorHostname: "hypervisor2"},
+					{ComputeHost: "host3", HypervisorHostname: "hypervisor3"},
+				},
+			},
+			// Should not do anything for VMware VMs
+			downvotedHosts: map[string]struct{}{},
+		},
+		{
+			name: "No overloaded hosts",
+			scenario: &testlibPlugins.MockScenario{
+				VMware: false,
+				Hosts: []testlibPlugins.MockScenarioHost{
+					{ComputeHost: "host4", HypervisorHostname: "hypervisor4"},
+					{ComputeHost: "host5", HypervisorHostname: "hypervisor5"},
+				},
+			},
+			// Should not downvote any hosts
+			downvotedHosts: map[string]struct{}{},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			weights, err := step.Run(tt.scenario)
+			if err != nil {
+				t.Fatalf("expected no error, got %v", err)
+			}
+			// Check that the weights have decreased
+			for host, weight := range weights {
+				if _, ok := tt.downvotedHosts[host]; ok {
+					if weight >= 0 {
+						t.Errorf("expected weight for host %s to be less than 0, got %f", host, weight)
+					}
+				} else {
+					if weight != 0 {
+						t.Errorf("expected weight for host %s to be 0, got %f", host, weight)
+					}
+				}
+			}
+		})
+	}
+}
