test: added test for kubernetes scraper to test relationship

db/config.go

@@ -95,13 +95,12 @@ func UpdateConfigItem(ci *models.ConfigItem) error {
 	return nil
 }
 
-// ConfigItemsIDs returns the uuid of config items which matches the given type, name & namespace
-func ConfigItemsIDs(ctx context.Context, configType, name, namespace string) ([]uuid.UUID, error) {
+// FindConfigIDsByNamespaceName returns the uuid of config items which matches the given type, name & namespace
+func FindConfigIDsByNamespaceName(ctx context.Context, namespace, name string) ([]uuid.UUID, error) {
 	var ids []uuid.UUID
 	err := ctx.DB().
 		Model(&models.ConfigItem{}).
 		Select("id").
-		Where("type = ?", configType).
 		Where("name = ?", name).
 		Where("namespace = ?", namespace).
 		Find(&ids).Error

fixtures/kubernetes.yaml

@@ -22,11 +22,11 @@ spec:
         - orders.acme.cert-manager.io
       relationships:
         - kind:
-            expr: has(spec.claimRef) && spec.claimRef.kind
+            expr: "has(spec.claimRef) ? spec.claimRef.kind : ''"
           name:
-            expr: has(spec.claimRef) && spec.claimRef.name
+            expr: "has(spec.claimRef) ? spec.claimRef.name : ''"
           namespace:
-            expr: has(spec.claimRef) && spec.claimRef.namespace
+            expr: "has(spec.claimRef) ? spec.claimRef.namespace : ''"
         - kind:
             value: Kustomization
           name:

scrapers/kubernetes/kubernetes.go

@@ -9,6 +9,8 @@ import (
 	"github.com/Jeffail/gabs/v2"
 	"github.com/flanksource/commons/collections"
 	"github.com/flanksource/commons/logger"
+	"github.com/flanksource/duty/context"
+
 	"github.com/flanksource/config-db/api"
 	v1 "github.com/flanksource/config-db/api/v1"
 	"github.com/flanksource/config-db/db"
@@ -32,6 +34,8 @@ func (kubernetes KubernetesScraper) Scrape(ctx api.ScrapeContext) v1.ScrapeResul
 	var (
 		results       v1.ScrapeResults
 		changeResults v1.ScrapeResults
+
+		err error
 	)
 
 	for _, config := range ctx.ScrapeConfig().Spec.Kubernetes {
@@ -51,7 +55,10 @@ func (kubernetes KubernetesScraper) Scrape(ctx api.ScrapeContext) v1.ScrapeResul
 		})
 
 		opts := options.NewDefaultCmdOptions()
-		opts = updateOptions(opts, config)
+		opts, err = updateOptions(ctx.DutyContext(), opts, config)
+		if err != nil {
+			return results.Errorf(err, "error setting up kube config")
+		}
 		objs := ketall.KetAll(opts)
 
 		resourceIDMap := getResourceIDsFromObjs(objs)
@@ -131,6 +138,10 @@ func (kubernetes KubernetesScraper) Scrape(ctx api.ScrapeContext) v1.ScrapeResul
 					return results.Errorf(err, "failed to evaluate kind: %v for config relationship", f.Kind)
 				}
 
+				if kind != obj.GetKind() {
+					continue // Try matching another relationship
+				}
+
 				name, err := f.Name.Eval(obj.GetLabels(), env)
 				if err != nil {
 					return results.Errorf(err, "failed to evaluate name: %v for config relationship", f.Name)
@@ -141,9 +152,9 @@ func (kubernetes KubernetesScraper) Scrape(ctx api.ScrapeContext) v1.ScrapeResul
 					return results.Errorf(err, "failed to evaluate namespace: %v for config relationship", f.Namespace)
 				}
 
-				linkedConfigItemIDs, err := db.ConfigItemsIDs(ctx.DutyContext(), fmt.Sprintf("%s%s", ConfigTypePrefix, kind), name, namespace)
+				linkedConfigItemIDs, err := db.FindConfigIDsByNamespaceName(ctx.DutyContext(), namespace, name)
 				if err != nil {
-					return results.Errorf(err, "failed to get linked config items: kind=%s, name=%s, namespace=%s", kind, name, namespace)
+					return results.Errorf(err, "failed to get linked config items: name=%s, namespace=%s", name, namespace)
 				}
 
 				for _, id := range linkedConfigItemIDs {
@@ -274,7 +285,7 @@ func getKubernetesAlias(obj *unstructured.Unstructured) []string {
 	return []string{strings.Join([]string{"Kubernetes", obj.GetKind(), obj.GetNamespace(), obj.GetName()}, "/")}
 }
 
-func updateOptions(opts *options.KetallOptions, config v1.Kubernetes) *options.KetallOptions {
+func updateOptions(ctx context.Context, opts *options.KetallOptions, config v1.Kubernetes) (*options.KetallOptions, error) {
 	opts.AllowIncomplete = config.AllowIncomplete
 	opts.Namespace = config.Namespace
 	opts.Scope = config.Scope
@@ -284,11 +295,16 @@ func updateOptions(opts *options.KetallOptions, config v1.Kubernetes) *options.K
 	opts.MaxInflight = config.MaxInflight
 	opts.Exclusions = config.Exclusions
 	opts.Since = config.Since
-	//TODO: update kubeconfig reference if provided by user
-	// if config.Kubeconfig != nil {
-	// 	opts.Kubeconfig = config.Kubeconfig.GetValue()
-	// }
-	return opts
+	if config.Kubeconfig != nil {
+		val, err := ctx.GetEnvValueFromCache(*config.Kubeconfig)
+		if err != nil {
+			return nil, err
+		}
+
+		opts.GenericCliFlags.KubeConfig = &val
+	}
+
+	return opts, nil
 }
 
 func extractDeployNameFromReplicaSet(rs string) string {

scrapers/run.go

@@ -22,8 +22,11 @@ func RunScraper(ctx api.ScrapeContext) (v1.ScrapeResults, error) {
 
 	// If error in any of the scrape results, don't delete old items
 	if len(results) > 0 && !v1.ScrapeResults(results).HasErr() {
-		if err := DeleteStaleConfigItems(*ctx.ScrapeConfig().GetPersistedID()); err != nil {
-			return nil, fmt.Errorf("error deleting stale config items: %w", err)
+		persistedID := ctx.ScrapeConfig().GetPersistedID()
+		if persistedID != nil {
+			if err := DeleteStaleConfigItems(ctx.DutyContext(), *persistedID); err != nil {
+				return nil, fmt.Errorf("error deleting stale config items: %w", err)
+			}
 		}
 	}
 

scrapers/runscrapers_suite_test.go

@@ -2,29 +2,44 @@ package scrapers
 
 import (
 	"os"
+	"path/filepath"
 	"testing"
 
 	epg "github.com/fergusstrange/embedded-postgres"
 	"github.com/flanksource/commons/logger"
+	v1 "github.com/flanksource/config-db/api/v1"
 	"github.com/flanksource/config-db/db"
 	"github.com/flanksource/duty"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"gorm.io/gorm"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	// +kubebuilder:scaffold:imports
 )
 
 func TestRunScrapers(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Scrapers Suite")
 }
 
-var postgres *epg.EmbeddedPostgres
+var (
+	postgres *epg.EmbeddedPostgres
+	gormDB   *gorm.DB
+)
 
 const (
 	pgUrl  = "postgres://postgres:postgres@localhost:9876/test?sslmode=disable"
 	pgPort = 9876
 )
 
 var _ = BeforeSuite(func() {
+	var err error
+
 	postgres = epg.NewDatabase(epg.DefaultConfig().Database("test").Port(pgPort))
 	if err := postgres.Start(); err != nil {
 		Fail(err.Error())
@@ -38,14 +53,85 @@ var _ = BeforeSuite(func() {
 		Fail(err.Error())
 	}
 
+	gormDB, err = duty.NewGorm(pgUrl, duty.DefaultGormConfig())
+	Expect(err).ToNot(HaveOccurred())
+
 	if err := os.Chdir(".."); err != nil {
 		Fail(err.Error())
 	}
+
+	setupTestK8s()
 })
 
 var _ = AfterSuite(func() {
+	if err := testEnv.Stop(); err != nil {
+		logger.Errorf("Error stopping test environment: %v", err)
+	}
+
 	logger.Infof("Stopping postgres")
 	if err := postgres.Stop(); err != nil {
 		Fail(err.Error())
 	}
 })
+
+var (
+	cfg            *rest.Config
+	k8sClient      client.Client
+	testEnv        *envtest.Environment
+	kubeConfigPath string
+)
+
+func setupTestK8s() {
+	By("bootstrapping test environment")
+	testEnv = &envtest.Environment{
+		CRDDirectoryPaths: []string{filepath.Join("chart", "crds")},
+	}
+
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).ToNot(HaveOccurred())
+	Expect(cfg).ToNot(BeNil())
+
+	err = v1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+
+	// +kubebuilder:scaffold:scheme
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	Expect(err).ToNot(HaveOccurred())
+	Expect(k8sClient).ToNot(BeNil())
+
+	kubeConfigPath, err = createKubeconfigFileForRestConfig(*cfg)
+	Expect(err).ToNot(HaveOccurred())
+}
+
+func createKubeconfigFileForRestConfig(restConfig rest.Config) (string, error) {
+	clusters := make(map[string]*clientcmdapi.Cluster)
+	clusters["default-cluster"] = &clientcmdapi.Cluster{
+		Server:                   restConfig.Host,
+		CertificateAuthorityData: restConfig.CAData,
+	}
+	contexts := make(map[string]*clientcmdapi.Context)
+	contexts["default-context"] = &clientcmdapi.Context{
+		Cluster:  "default-cluster",
+		AuthInfo: "default-user",
+	}
+	authinfos := make(map[string]*clientcmdapi.AuthInfo)
+	authinfos["default-user"] = &clientcmdapi.AuthInfo{
+		ClientCertificateData: restConfig.CertData,
+		ClientKeyData:         restConfig.KeyData,
+	}
+	clientConfig := clientcmdapi.Config{
+		Kind:           "Config",
+		APIVersion:     "v1",
+		Clusters:       clusters,
+		Contexts:       contexts,
+		CurrentContext: "default-context",
+		AuthInfos:      authinfos,
+	}
+	kubeConfigFile, err := os.CreateTemp("", "kubeconfig-*")
+	if err != nil {
+		return "", err
+	}
+	_ = clientcmd.WriteToFile(clientConfig, kubeConfigFile.Name())
+	return kubeConfigFile.Name(), nil
+}