fix: add logs to ConsumeKubernetesWatchResourcesJob

flanksource · Jun 11, 2024 · 02840b6 · 02840b6
1 parent c243586
commit 02840b6
Show file tree

Hide file tree

Showing 4 changed files with 23 additions and 11 deletions.
diff --git a/db/config.go b/db/config.go
@@ -211,10 +211,10 @@ func FindConfigChangesByItemID(ctx api.ScrapeContext, configItemID string) ([]du
 	return ci, nil
 }
 
-func SoftDeleteConfigItems(ctx context.Context, ids ...string) error {
-	return ctx.DB().
+func SoftDeleteConfigItems(ctx context.Context, ids ...string) (int, error) {
+	tx := ctx.DB().
 		Model(&models.ConfigItem{}).
 		Where("id IN ?", ids).
-		Update("deleted_at", gorm.Expr("NOW()")).
-		Error
+		Update("deleted_at", gorm.Expr("NOW()"))
+	return int(tx.RowsAffected), tx.Error
 }
diff --git a/db/update.go b/db/update.go
@@ -327,6 +327,10 @@ func SavePartialResults(ctx api.ScrapeContext, results []v1.ScrapeResult) error
 }
 
 func saveResults(ctx api.ScrapeContext, isPartialResultSet bool, results []v1.ScrapeResult) error {
+	if len(results) == 0 {
+		return nil
+	}
+
 	startTime, err := GetCurrentDBTime(ctx)
 	if err != nil {
 		return fmt.Errorf("unable to get current db time: %w", err)

diff --git a/scrapers/cron.go b/scrapers/cron.go
@@ -14,6 +14,7 @@ import (
 	"github.com/robfig/cron/v3"
 	"github.com/samber/lo"
 	"github.com/sethvargo/go-retry"
+	"go.opentelemetry.io/otel/attribute"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/types"
 
@@ -321,13 +322,20 @@ func ConsumeKubernetesWatchResourcesJobFunc(sc api.ScrapeContext, config v1.Kube
 			if !ok {
 				return fmt.Errorf("no resource watcher channel found for config (scrapeconfig: %s)", config.Hash())
 			}
-			deletChan := _deleteCh.(chan string)
+			deleteChan := _deleteCh.(chan string)
 
-			if len(deletChan) > 0 {
-				deletedResourcesIDs, _, _, _ := lo.Buffer(deletChan, len(deletChan))
-				if err := db.SoftDeleteConfigItems(ctx.Context, deletedResourcesIDs...); err != nil {
+			if len(deleteChan) > 0 {
+				deletedResourcesIDs, _, _, _ := lo.Buffer(deleteChan, len(deleteChan))
+
+				total, err := db.SoftDeleteConfigItems(ctx.Context, deletedResourcesIDs...)
+				if err != nil {
 					return fmt.Errorf("failed to delete %d resources: %w", len(deletedResourcesIDs), err)
+				} else if total != len(deletedResourcesIDs) {
+					ctx.GetSpan().SetAttributes(attribute.StringSlice("deletedResourcesIDs", deletedResourcesIDs))
+					ctx.Logger.Warnf("attempted to delete %d resources but only deleted %d", len(deletedResourcesIDs), total)
 				}
+
+				ctx.History.SuccessCount += total
 			}
 
 			return nil

diff --git a/scrapers/kubernetes/informers.go b/scrapers/kubernetes/informers.go
@@ -59,7 +59,7 @@ func (t *SharedInformerManager) Register(ctx api.ScrapeContext, kubeconfig strin
 				return
 			}
 
-			ctx.Logger.V(2).Infof("added: %s %s", u.GetKind(), u.GetName())
+			ctx.Logger.V(2).Infof("added: %s %s %s", u.GetUID(), u.GetKind(), u.GetName())
 			buffer <- u
 		},
 		UpdateFunc: func(oldObj any, newObj any) {
@@ -69,7 +69,7 @@ func (t *SharedInformerManager) Register(ctx api.ScrapeContext, kubeconfig strin
 				return
 			}
 
-			ctx.Logger.V(2).Infof("updated: %s %s", u.GetKind(), u.GetName())
+			ctx.Logger.V(2).Infof("updated: %s %s %s", u.GetUID(), u.GetKind(), u.GetName())
 			buffer <- u
 		},
 		DeleteFunc: func(obj any) {
@@ -79,7 +79,7 @@ func (t *SharedInformerManager) Register(ctx api.ScrapeContext, kubeconfig strin
 				return
 			}
 
-			ctx.Logger.V(2).Infof("deleted: %s %s", u.GetKind(), u.GetName())
+			ctx.Logger.V(2).Infof("deleted: %s %s %s", u.GetUID(), u.GetKind(), u.GetName())
 			deleteBuffer <- string(u.GetUID())
 		},
 	})