feat: add generic metric, extractor, weigher/filter

helm/library/cortex/templates/rbac/role.yaml

@@ -51,4 +51,11 @@ rules:
   - get
   - patch
   - update
+- apiGroups: [""]
+  resources:
+  - secrets
+  verbs:
+  - get
+  - list
+  - watch
 {{- end -}}

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/controller.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
-
 	"github.com/cobaltcore-dev/cortex/internal/knowledge/db"
 	"github.com/cobaltcore-dev/cortex/pkg/multicluster"
 	corev1 "k8s.io/api/core/v1"
@@ -76,6 +75,7 @@ func (r *KnowledgeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 
 	// Check if all datasources configured share the same database secret ref.
 	var databaseSecretRef *corev1.SecretReference
+	var dataSources []*v1alpha1.Datasource
 	for _, dsRef := range knowledge.Spec.Dependencies.Datasources {
 		ds := &v1alpha1.Datasource{}
 		if err := r.Get(ctx, client.ObjectKey{
@@ -116,7 +116,34 @@ func (r *KnowledgeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 			}
 			return ctrl.Result{}, nil
 		}
+		dataSources = append(dataSources, ds)
+	}
+
+	var knowledgeSources []*v1alpha1.Knowledge
+	for _, knRef := range knowledge.Spec.Dependencies.Knowledges {
+		kn := &v1alpha1.Knowledge{}
+		if err := r.Get(ctx, client.ObjectKey{
+			Namespace: req.Namespace,
+			Name:      knRef.Name,
+		}, kn); err != nil {
+			log.Error(err, "failed to get knowledge", "name", knRef.Name)
+			old := knowledge.DeepCopy()
+			meta.SetStatusCondition(&knowledge.Status.Conditions, metav1.Condition{
+				Type:    v1alpha1.KnowledgeConditionReady,
+				Status:  metav1.ConditionFalse,
+				Reason:  "KnowledgeFetchFailed",
+				Message: "failed to get knowledge: " + err.Error(),
+			})
+			patch := client.MergeFrom(old)
+			if err := r.Status().Patch(ctx, knowledge, patch); err != nil {
+				log.Error(err, "failed to patch knowledge status")
+				return ctrl.Result{}, err
+			}
+			return ctrl.Result{}, err
+		}
+		knowledgeSources = append(knowledgeSources, kn)
 	}
+
 	// When we have datasources reading from a database, connect to it.
 	var authenticatedDatasourceDB *db.DB
 	if databaseSecretRef != nil {
@@ -160,7 +187,7 @@ func (r *KnowledgeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (
 		return ctrl.Result{}, err
 	}
 
-	features, err := extractor.Extract()
+	features, err := extractor.Extract(dataSources, knowledgeSources)
 	if err != nil {
 		log.Error(err, "failed to extract features", "name", knowledge.Spec.Extractor.Name)
 		old := knowledge.DeepCopy()

internal/knowledge/extractor/monitor.go

@@ -94,10 +94,10 @@ func monitorFeatureExtractor[F plugins.FeatureExtractor](label string, f F, m Mo
 }
 
 // Run the wrapped feature extractor and measure the time it takes.
-func (m FeatureExtractorMonitor[F]) Extract() ([]plugins.Feature, error) {
+func (m FeatureExtractorMonitor[F]) Extract(d []*v1alpha1.Datasource, k []*v1alpha1.Knowledge) ([]plugins.Feature, error) {
 	slog.Info("features: extracting", "extractor", m.label)
 
-	features, err := m.FeatureExtractor.Extract()
+	features, err := m.FeatureExtractor.Extract(d, k)
 	if err != nil {
 		return nil, err
 	}

internal/knowledge/extractor/monitor_test.go

@@ -26,7 +26,7 @@ func (m *mockFeatureExtractor) Init(datasourceDB *db.DB, client client.Client, s
 	return m.initError
 }
 
-func (m *mockFeatureExtractor) Extract() ([]plugins.Feature, error) {
+func (m *mockFeatureExtractor) Extract(_ []*v1alpha1.Datasource, _ []*v1alpha1.Knowledge) ([]plugins.Feature, error) {
 	if m.extractError != nil {
 		return nil, m.extractError
 	}
@@ -118,7 +118,7 @@ func TestMonitorFeatureExtractor_Extract_Success(t *testing.T) {
 
 	wrappedExtractor := monitorFeatureExtractor("test-extractor", mockExtractor, monitor)
 
-	features, err := wrappedExtractor.Extract()
+	features, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Errorf("Expected no error, got %v", err)
 	}
@@ -154,7 +154,7 @@ func TestMonitorFeatureExtractor_Extract_Error(t *testing.T) {
 
 	wrappedExtractor := monitorFeatureExtractor("test-extractor", mockExtractor, monitor)
 
-	features, err := wrappedExtractor.Extract()
+	features, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if !errors.Is(err, expectedError) {
 		t.Errorf("Expected error %v, got %v", expectedError, err)
 	}
@@ -171,7 +171,7 @@ func TestMonitorFeatureExtractor_Extract_EmptyFeatures(t *testing.T) {
 
 	wrappedExtractor := monitorFeatureExtractor("test-extractor", mockExtractor, monitor)
 
-	features, err := wrappedExtractor.Extract()
+	features, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Errorf("Expected no error, got %v", err)
 	}
@@ -202,7 +202,7 @@ func TestMonitorFeatureExtractor_NilMonitor(t *testing.T) {
 	wrappedExtractor := monitorFeatureExtractor("test-extractor", mockExtractor, monitor)
 
 	// Should not panic even with nil metrics
-	features, err := wrappedExtractor.Extract()
+	features, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Errorf("Expected no error, got %v", err)
 	}
@@ -235,7 +235,7 @@ func TestMonitorFeatureExtractor_MultipleExtractions(t *testing.T) {
 	wrappedExtractor := monitorFeatureExtractor("test-extractor", mockExtractor, monitor)
 
 	// First extraction
-	features1, err := wrappedExtractor.Extract()
+	features1, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Errorf("First extraction failed: %v", err)
 	}
@@ -249,7 +249,7 @@ func TestMonitorFeatureExtractor_MultipleExtractions(t *testing.T) {
 	}
 
 	// Second extraction
-	features2, err := wrappedExtractor.Extract()
+	features2, err := wrappedExtractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Errorf("Second extraction failed: %v", err)
 	}
@@ -286,12 +286,12 @@ func TestMonitorFeatureExtractor_DifferentExtractorNames(t *testing.T) {
 	wrappedExtractor2 := monitorFeatureExtractor("extractor-2", mockExtractor2, monitor)
 
 	// Extract from both
-	_, err1 := wrappedExtractor1.Extract()
+	_, err1 := wrappedExtractor1.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err1 != nil {
 		t.Errorf("Extractor 1 failed: %v", err1)
 	}
 
-	_, err2 := wrappedExtractor2.Extract()
+	_, err2 := wrappedExtractor2.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err2 != nil {
 		t.Errorf("Extractor 2 failed: %v", err2)
 	}

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

@@ -6,6 +6,7 @@ package compute
 import (
 	_ "embed"
 
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
 	"github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins"
 )
 
@@ -28,6 +29,6 @@ type HostAZExtractor struct {
 var hostAZQuery string
 
 // Extract the traits of a compute host from the database.
-func (e *HostAZExtractor) Extract() ([]plugins.Feature, error) {
+func (e *HostAZExtractor) Extract(_ []*v1alpha1.Datasource, _ []*v1alpha1.Knowledge) ([]plugins.Feature, error) {
 	return e.ExtractSQL(hostAZQuery)
 }

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

@@ -70,7 +70,7 @@ func TestHostAZExtractor_Extract(t *testing.T) {
 	if err := extractor.Init(&testDB, nil, config); err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}
-	features, err := extractor.Extract()
+	features, err := extractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}

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

@@ -6,6 +6,7 @@ package compute
 import (
 	_ "embed"
 
+	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
 	"github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins"
 )
 
@@ -32,6 +33,6 @@ type HostCapabilitiesExtractor struct {
 var hostCapabilitiesQuery string
 
 // Extract the traits of a compute host from the database.
-func (e *HostCapabilitiesExtractor) Extract() ([]plugins.Feature, error) {
+func (e *HostCapabilitiesExtractor) Extract(_ []*v1alpha1.Datasource, _ []*v1alpha1.Knowledge) ([]plugins.Feature, error) {
 	return e.ExtractSQL(hostCapabilitiesQuery)
 }

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

@@ -55,7 +55,7 @@ func TestHostCapabilitiesExtractor_Extract(t *testing.T) {
 	if err := extractor.Init(&testDB, nil, config); err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}
-	features, err := extractor.Extract()
+	features, err := extractor.Extract([]*v1alpha1.Datasource{}, []*v1alpha1.Knowledge{})
 	if err != nil {
 		t.Fatalf("expected no error, got %v", err)
 	}