improve config_changes lookup for existing ones

api/cache.go

@@ -16,8 +16,6 @@ type TempCache struct {
  ctx context.Context
  items map[string]models.ConfigItem
  aliases map[string]string
-
- changes map[string]struct{}
 }
 
 func (t TempCache) FindExternal(ext v1.ExternalID) (*models.ConfigItem, error) {
@@ -78,38 +76,6 @@ func (t TempCache) Insert(item models.ConfigItem) {
  t.items[strings.ToLower(item.ID)] = item
 }
 
-func (t TempCache) IsChangePersisted(configID, externalChangeID string) (bool, error) {
- if configID == "" || externalChangeID == "" {
- return false, nil
- }
-
- configID = strings.ToLower(configID)
- externalChangeID = strings.ToLower(externalChangeID)
-
- if t.changes == nil {
- t.changes = make(map[string]struct{})
- }
-
- if _, ok := t.changes[configID+externalChangeID]; ok {
- return true, nil
- }
-
- var result models.ConfigChange
- if err := t.ctx.DB().Select("id").Where("config_id = ?", configID).
- Where("external_change_id = ?", externalChangeID).
- Limit(1).
- Find(&result).Error; err != nil {
- return false, err
- }
-
- if result.ID != "" {
- t.changes[configID+externalChangeID] = struct{}{}
- return true, nil
- }
-
- return false, nil
-}
-
 func (t TempCache) Get(id string) (*models.ConfigItem, error) {
  id = strings.ToLower(id)
  if id == "" {

db/changes.go

@@ -5,6 +5,8 @@ import (
  "time"
 
  sw "github.com/RussellLuo/slidingwindow"
+ "github.com/patrickmn/go-cache"
+ "github.com/samber/lo"
 
  "github.com/flanksource/commons/collections"
  "github.com/flanksource/config-db/api"
@@ -19,6 +21,8 @@ const (
  ChangeTypeTooManyChanges = "TooManyChanges"
 )
 
+var configChangesCache = cache.New(time.Hour*24, time.Hour*24)
+
 func GetWorkflowRunCount(ctx api.ScrapeContext, workflowID string) (int64, error) {
  var count int64
  err := ctx.DB().Table("config_changes").
@@ -122,6 +126,7 @@ func syncCurrentlyRateLimitedConfigs(ctx api.ScrapeContext, window time.Duration
  if err != nil {
  return nil, err
  }
+ defer rows.Close()
 
  output := make(map[string]struct{})
  for rows.Next() {
@@ -156,6 +161,7 @@ func syncWindow(ctx api.ScrapeContext, max int, window time.Duration) error {
  if err != nil {
  return err
  }
+ defer rows.Close()
 
  for rows.Next() {
  var configID string
@@ -179,3 +185,54 @@ func syncWindow(ctx api.ScrapeContext, max int, window time.Duration) error {
 
  return rows.Err()
 }
+
+// filterOutPersistedChanges returns only those changes that weren't seen in the db.
+func filterOutPersistedChanges(ctx api.ScrapeContext, changes []*models.ConfigChange) ([]*models.ConfigChange, error) {
+ // use cache to filter out ones that we've already seen before
+ changes = lo.Filter(changes, func(c *models.ConfigChange, _ int) bool {
+ _, found := configChangesCache.Get(c.ConfigID + c.ExternalChangeId)
+ if found {
+ _ = found
+ }
+ return !found
+ })
+
+ if len(changes) == 0 {
+ return nil, nil
+ }
+
+ query := `SELECT config_id, external_change_id
+ FROM config_changes
+ WHERE (config_id, external_change_id) IN ?`
+ args := lo.Map(changes, func(c *models.ConfigChange, _ int) []string {
+ return []string{c.ConfigID, c.ExternalChangeId}
+ })
+
+ rows, err := ctx.DB().Raw(query, args).Rows()
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ existing := make(map[string]struct{})
+ for rows.Next() {
+ var configID, externalChangeID string
+ if err := rows.Scan(&configID, &externalChangeID); err != nil {
+ return nil, err
+ }
+
+ configChangesCache.SetDefault(configID+externalChangeID, struct{}{})
+ existing[configID+externalChangeID] = struct{}{}
+ }
+
+ newOnes := lo.Filter(changes, func(c *models.ConfigChange, _ int) bool {
+ _, found := existing[c.ConfigID+c.ExternalChangeId]
+ return !found
+ })
+
+ if len(newOnes) > 0 {
+ _ = query
+ }
+
+ return newOnes, nil
+}

db/update.go

@@ -216,8 +216,8 @@ func shouldExcludeChange(result *v1.ScrapeResult, changeResult v1.ChangeResult)
 
 func extractChanges(ctx api.ScrapeContext, result *v1.ScrapeResult, ci *models.ConfigItem) ([]*models.ConfigChange, []*models.ConfigChange, error) {
  var (
- newOnes = []*models.ConfigChange{}
- updates = []*models.ConfigChange{}
+ toInsert = []*models.ConfigChange{}
+ toUpdate = []*models.ConfigChange{}
  )
 
  changes.ProcessRules(result, result.BaseScraper.Transform.Change.Mapping...)
@@ -274,17 +274,19 @@ func extractChanges(ctx api.ScrapeContext, result *v1.ScrapeResult, ci *models.C
  }
 
  if changeResult.UpdateExisting {
- updates = append(updates, change)
+ toUpdate = append(toUpdate, change)
  } else {
- if ok, err := ctx.TempCache().IsChangePersisted(change.ConfigID, change.ExternalChangeId); err != nil {
- return nil, nil, fmt.Errorf("failed to check if change is persisted: %w", err)
- } else if !ok {
- newOnes = append(newOnes, change)
- }
+ toInsert = append(toInsert, change)
  }
  }
 
- return newOnes, updates, nil
+ // Remove the changes that have already been inserted.
+ newOnes, err := filterOutPersistedChanges(ctx, toInsert)
+ if err != nil {
+ return nil, nil, err
+ }
+
+ return newOnes, toUpdate, nil
 }
 
 func upsertAnalysis(ctx api.ScrapeContext, result *v1.ScrapeResult) error {