feat: add generic metric, extractor, weigher/filter

Author: henrichter

internal/knowledge/datasources/plugins/prometheus/supported_syncers.go

@@ -26,4 +26,5 @@ var supportedMetricSyncers = map[string]func(
 	"netapp_node_metric":                    newTypedSyncer[NetAppNodeMetric],
 	"netapp_volume_aggregate_labels_metric": newTypedSyncer[NetAppVolumeAggrLabelsMetric],
 	"kvm_libvirt_domain_metric":             newTypedSyncer[KVMDomainMetric],
+	"generic":                               newTypedSyncer[GenericMetric],
 }

internal/knowledge/datasources/plugins/prometheus/types.go

@@ -292,3 +292,37 @@ func (m NetAppVolumeAggrLabelsMetric) With(n string, t time.Time, v float64) Pro
 	m.Value = v
 	return m
 }
+
+type GenericMetric struct {
+	Name      string    `json:"name" db:"name"`
+	Host      string    `json:"host" db:"host"`
+	Value     float64   `json:"value" db:"value"`
+	Timestamp time.Time `json:"timestamp" db:"timestamp"`
+}
+
+func (m GenericMetric) GetName() string {
+	return m.Name
+}
+
+func (m GenericMetric) GetValue() float64 {
+	return m.Value
+}
+
+func (m GenericMetric) GetTimestamp() time.Time {
+	return m.Timestamp
+}
+
+func (m GenericMetric) TableName() string {
+	return "generic"
+}
+
+func (m GenericMetric) Indexes() map[string][]string {
+	return nil
+}
+
+func (m GenericMetric) With(alias string, t time.Time, v float64) PrometheusMetric {
+	m.Name = alias
+	m.Timestamp = t
+	m.Value = v
+	return m
+}

internal/knowledge/extractor/plugins/base.go

@@ -42,13 +42,13 @@ func (e *BaseExtractor[Opts, Feature]) Init(
 }
 
 // Extract the features directly from an sql query.
-func (e *BaseExtractor[Opts, F]) ExtractSQL(query string) ([]Feature, error) {
+func (e *BaseExtractor[Opts, F]) ExtractSQL(query string, args ...interface{}) ([]Feature, error) {
 	// This can happen when no datasource is provided that connects to a database.
 	if e.DB == nil {
 		return nil, errors.New("database connection is not initialized")
 	}
 	var features []F
-	if _, err := e.DB.Select(&features, query); err != nil {
+	if _, err := e.DB.Select(&features, query, args...); err != nil {
 		return nil, err
 	}
 	return e.Extracted(features)

internal/knowledge/extractor/plugins/compute/flavor_groups.go

@@ -9,6 +9,7 @@ import (
 	"errors"
 	"sort"
 
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
 	"github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins"
 	ctrl "sigs.k8s.io/controller-runtime"
 )
@@ -76,7 +77,7 @@ var flavorGroupsQuery string
 var flavorGroupLog = ctrl.Log.WithName("flavor_group_extractor")
 
 // Extract flavor groups from the database.
-func (e *FlavorGroupExtractor) Extract() ([]plugins.Feature, error) {
+func (e *FlavorGroupExtractor) Extract(_ []*v1alpha1.Datasource, _ []*v1alpha1.Knowledge) ([]plugins.Feature, error) {
 	if e.DB == nil {
 		return nil, errors.New("database connection is not initialized")
 	}

internal/knowledge/extractor/plugins/compute/flavor_groups_test.go

@@ -125,7 +125,7 @@ func TestFlavorGroupExtractor_Extract(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	features, err := extractor.Extract()
+	features, err := extractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -360,7 +360,7 @@ func TestFlavorGroupExtractor_RamCoreRatio_FixedRatio(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	features, err := extractor.Extract()
+	features, err := extractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/knowledge/extractor/plugins/compute/vrops_hostsystem_contention_short_term.go

@@ -26,7 +26,7 @@ type VROpsHostsystemContentionShortTerm struct {
 type VROpsHostsystemContentionShortTermExtractor struct {
 	// Common base for all extractors that provides standard functionality.
 	plugins.BaseExtractor[
-		struct{},                           // No options passed through yaml config
+		struct{}, // No options passed through yaml config
 		VROpsHostsystemContentionShortTerm, // Feature model
 	]
 }

internal/knowledge/extractor/plugins/generic.go

@@ -0,0 +1,39 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package plugins
+
+import (
+	_ "embed"
+	"errors"
+
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
+)
+
+type Generic struct {
+	Host  string  `db:"host"`
+	Value float64 `db:"value"`
+}
+
+type GenericExtractor struct {
+	BaseExtractor[
+		struct{},
+		Generic,
+	]
+}
+
+func (e *GenericExtractor) Extract(d []*v1alpha1.Datasource, _ []*v1alpha1.Knowledge) ([]Feature, error) {
+	if len(d) != 1 {
+		return nil, errors.New("generic Knowledge requires exactly one datasource")
+	}
+	dsSpec := &d[0].Spec
+	if dsSpec.Type != v1alpha1.DatasourceTypePrometheus {
+		return nil, errors.New("unsupported datasource type: expected prometheus")
+	}
+	name := dsSpec.Prometheus.Alias
+	if name == "" {
+		return nil, errors.New("prometheus datasource alias cannot be empty")
+	}
+	query := "SELECT host, value FROM generic WHERE name = ?"
+	return e.ExtractSQL(query, name)
+}

internal/knowledge/extractor/plugins/generic_test.go

@@ -0,0 +1,77 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package plugins
+
+import (
+	"slices"
+	"testing"
+
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
+	"github.com/cobaltcore-dev/cortex/internal/knowledge/datasources/plugins/prometheus"
+	"github.com/cobaltcore-dev/cortex/internal/knowledge/db"
+	testlibDB "github.com/cobaltcore-dev/cortex/internal/knowledge/db/testing"
+)
+
+func TestGenericExtractor_Extract(t *testing.T) {
+	dbEnv := testlibDB.SetupDBEnv(t)
+	testDB := db.DB{DbMap: dbEnv.DbMap}
+	defer dbEnv.Close()
+	// Create dependency table
+	if err := testDB.CreateTable(
+		testDB.AddTable(prometheus.GenericMetric{}),
+	); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	metrics := []any{
+		&prometheus.GenericMetric{Name: "node_cpu_seconds_total", Host: "node-01", Value: 0.81},
+		&prometheus.GenericMetric{Name: "node_cpu_seconds_total", Host: "node-02", Value: 0.37},
+	}
+	if err := testDB.Insert(metrics...); err != nil {
+		t.Fatalf("failed to insert manila storage pools: %v", err)
+	}
+
+	extractor := &GenericExtractor{}
+	config := v1alpha1.KnowledgeSpec{}
+	if err := extractor.Init(&testDB, nil, config); err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+	datasources := []*v1alpha1.Datasource{
+		{
+			Spec: v1alpha1.DatasourceSpec{
+				Type: v1alpha1.DatasourceTypePrometheus,
+				Prometheus: v1alpha1.PrometheusDatasource{
+					Alias: "node_cpu_seconds_total",
+				},
+			},
+		},
+	}
+
+	features, err := extractor.Extract(datasources, []*v1alpha1.Knowledge{})
+	if err != nil {
+		t.Fatalf("expected no error, got %v", err)
+	}
+
+	var actual []Generic
+	for _, f := range features {
+		actual = append(actual, f.(Generic))
+	}
+
+	expected := []Generic{
+		{Host: "node-01", Value: 0.81},
+		{Host: "node-02", Value: 0.37},
+	}
+
+	if len(actual) != len(expected) {
+		t.Errorf("expected %d rows, got %d", len(expected), len(actual))
+	}
+
+	for _, exp := range expected {
+		if !slices.ContainsFunc(actual, func(m Generic) bool {
+			return m.Host == exp.Host && m.Value == exp.Value
+		}) {
+			t.Errorf("expected to find %+v in actual results %+v", exp, actual)
+		}
+	}
+}

internal/knowledge/extractor/supported_extractors.go

@@ -26,4 +26,6 @@ var supportedExtractors = map[string]plugins.FeatureExtractor{
 	"flavor_groups":                                    &compute.FlavorGroupExtractor{},
 
 	"netapp_storage_pool_cpu_usage_extractor": &storage.StoragePoolCPUUsageExtractor{},
+
+	"generic": &plugins.GenericExtractor{},
 }

internal/scheduling/pods/plugins/filters/filter_generic.go

@@ -0,0 +1,76 @@
+// Copyright SAP SE
+// SPDX-License-Identifier: Apache-2.0
+
+package filters
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+
+	"github.com/cobaltcore-dev/cortex/api/external/pods"
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
+	"github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins"
+	"github.com/cobaltcore-dev/cortex/internal/scheduling/lib"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+type GenericFilterStepOpts struct {
+	Knowledge string `json:"knowledge"`
+}
+
+func (o GenericFilterStepOpts) Validate() error {
+	if o.Knowledge == "" {
+		return fmt.Errorf("knowledge name must be provided")
+	}
+	return nil
+}
+
+type GenericFilterStep struct {
+	lib.BaseFilter[pods.PodPipelineRequest, GenericFilterStepOpts]
+	knowledgeRef types.NamespacedName
+}
+
+func (f *GenericFilterStep) Init(ctx context.Context, client client.Client, spec v1alpha1.FilterSpec) error {
+	if err := f.BaseFilter.Init(ctx, client, spec); err != nil {
+		return err
+	}
+	knowledgeRef := types.NamespacedName{Name: f.Options.Knowledge}
+	if err := f.CheckKnowledges(ctx, knowledgeRef); err != nil {
+		return err
+	}
+	f.knowledgeRef = knowledgeRef
+
+	return nil
+}
+
+func (f *GenericFilterStep) Run(ctx context.Context, _ *slog.Logger, req pods.PodPipelineRequest) (*lib.FilterWeigherPipelineStepResult, error) {
+	knowledge := v1alpha1.Knowledge{}
+	if err := f.Client.Get(ctx, f.knowledgeRef, &knowledge); err != nil {
+		return nil, err
+	}
+
+	nodeFeatures, err := v1alpha1.
+		UnboxFeatureList[plugins.Generic](knowledge.Status.Raw)
+	if err != nil {
+		return nil, err
+	}
+
+	result := f.IncludeAllHostsFromRequest(req)
+	for _, nodeFeature := range nodeFeatures {
+		host := nodeFeature.Host
+		if _, exists := result.Activations[host]; !exists {
+			continue
+		}
+		if nodeFeature.Value == 1.0 {
+			delete(result.Activations, host)
+		}
+	}
+
+	return result, nil
+}
+
+func init() {
+	Index["generic"] = func() PodFilter { return &GenericFilterStep{} }
+}