Alerting: delete orphaned records from kvstore (#40337) (#40450)

(cherry picked from commit 153c356993)

Co-authored-by: Jean-Philippe Quéméner <JohnnyQQQQ@users.noreply.github.com>

pkg/services/ngalert/notifier/multiorg_alertmanager.go

@@ -219,13 +219,14 @@ func (moa *MultiOrgAlertmanager) SyncAlertmanagersForOrgs(ctx context.Context, o
 	// We look for orphan directories and remove them. Orphan directories can
 	// occur when an organization is deleted and the node running Grafana is
 	// shutdown before the next sync is executed.
-	moa.cleanupOrphanLocalOrgState(orgsFound)
+	moa.cleanupOrphanLocalOrgState(ctx, orgsFound)
 }
 
 // cleanupOrphanLocalOrgState will check if there is any organization on
 // disk that is not part of the active organizations. If this is the case
 // it will delete the local state from disk.
-func (moa *MultiOrgAlertmanager) cleanupOrphanLocalOrgState(activeOrganizations map[int64]struct{}) {
+func (moa *MultiOrgAlertmanager) cleanupOrphanLocalOrgState(ctx context.Context,
+	activeOrganizations map[int64]struct{}) {
 	dataDir := filepath.Join(moa.settings.DataPath, workingDir)
 	files, err := ioutil.ReadDir(dataDir)
 	if err != nil {
@@ -251,6 +252,27 @@ func (moa *MultiOrgAlertmanager) cleanupOrphanLocalOrgState(activeOrganizations
 			fileStore.CleanUp()
 		}
 	}
+	// Remove all orphaned items from kvstore by listing all existing items
+	// in our used namespace and comparing them to the currently active
+	// organizations.
+	storedFiles := []string{notificationLogFilename, silencesFilename}
+	for _, fileName := range storedFiles {
+		keys, err := moa.kvStore.Keys(ctx, kvstore.AllOrganizations, KVNamespace, fileName)
+		if err != nil {
+			moa.logger.Error("failed to fetch items from kvstore", "err", err,
+				"namespace", KVNamespace, "key", fileName)
+		}
+		for _, key := range keys {
+			if _, exists := activeOrganizations[key.OrgId]; exists {
+				continue
+			}
+			err = moa.kvStore.Del(ctx, key.OrgId, key.Namespace, key.Key)
+			if err != nil {
+				moa.logger.Error("failed to delete item from kvstore", "err", err,
+					"orgID", key.OrgId, "namespace", KVNamespace, "key", key.Key)
+			}
+		}
+	}
 }
 
 func (moa *MultiOrgAlertmanager) StopAndWait() {

pkg/services/ngalert/notifier/multiorg_alertmanager_test.go

@@ -96,8 +96,9 @@ grafana_alerting_discovered_configurations 4
 		require.Len(t, mam.alertmanagers, 4)
 	}
 
-	// Orphaned local state should be removed.
+	// Orphaned state should be removed.
 	{
+		orgID := int64(6)
 		// First we create a directory and two files for an ograniztation that
 		// is not existing in the current state.
 		orphanDir := filepath.Join(tmpDir, "alerting", "6")
@@ -120,12 +121,26 @@ grafana_alerting_discovered_configurations 4
 		require.NoError(t, err)
 		require.Equal(t, info.Name(), notificationLogFilename)
 
+		// We also populate the kvstore with orphaned records.
+		err = kvStore.Set(ctx, orgID, KVNamespace, silencesFilename, "file_1")
+		require.NoError(t, err)
+
+		err = kvStore.Set(ctx, orgID, KVNamespace, notificationLogFilename, "file_1")
+		require.NoError(t, err)
+
 		// Now re run the sync job once.
 		require.NoError(t, mam.LoadAndSyncAlertmanagersForOrgs(ctx))
 
 		// The organization directory should be gone by now.
 		_, err = os.Stat(orphanDir)
 		require.True(t, errors.Is(err, fs.ErrNotExist))
+
+		// The organization kvstore records should be gone by now.
+		_, exists, _ := kvStore.Get(ctx, orgID, KVNamespace, silencesFilename)
+		require.False(t, exists)
+
+		_, exists, _ = kvStore.Get(ctx, orgID, KVNamespace, notificationLogFilename)
+		require.False(t, exists)
 	}
 }
 

pkg/services/ngalert/notifier/testing.go

@@ -2,9 +2,11 @@ package notifier
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"testing"
 
+	"github.com/grafana/grafana/pkg/infra/kvstore"
 	"github.com/grafana/grafana/pkg/services/ngalert/models"
 	"github.com/grafana/grafana/pkg/services/ngalert/store"
 )
@@ -130,6 +132,29 @@ func (fkv *FakeKVStore) Del(_ context.Context, orgId int64, namespace string, ke
 	return nil
 }
 
+func (fkv *FakeKVStore) Keys(ctx context.Context, orgID int64, namespace string, keyPrefix string) ([]kvstore.Key, error) {
+	fkv.mtx.Lock()
+	defer fkv.mtx.Unlock()
+	var keys []kvstore.Key
+	for orgIDFromStore, namespaceMap := range fkv.store {
+		if orgID != kvstore.AllOrganizations && orgID != orgIDFromStore {
+			continue
+		}
+		if keyMap, exists := namespaceMap[namespace]; exists {
+			for k := range keyMap {
+				if strings.HasPrefix(k, keyPrefix) {
+					keys = append(keys, kvstore.Key{
+						OrgId:     orgIDFromStore,
+						Namespace: namespace,
+						Key:       keyPrefix,
+					})
+				}
+			}
+		}
+	}
+	return keys, nil
+}
+
 type fakeState struct {
 	data string
 }