Remove support RebuildCollection, always rebuild on bulk import. (#117326)

pkg/storage/unified/migrations/migrator.go

@@ -4,9 +4,10 @@ import (
 	"context"
 	"fmt"
 
+	"google.golang.org/grpc/metadata"
+
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/registry/apis/dashboard/legacy"
-	"google.golang.org/grpc/metadata"
 
 	authlib "github.com/grafana/authlib/types"
 
@@ -34,10 +35,7 @@ type streamProvider interface {
 }
 
 func buildCollectionSettings(opts legacy.MigrateOptions) resource.BulkSettings {
-	settings := resource.BulkSettings{
-		RebuildCollection: true,
-		SkipValidation:    true,
-	}
+	settings := resource.BulkSettings{SkipValidation: true}
 	for _, res := range opts.Resources {
 		key := buildResourceKey(res, opts.Namespace)
 		if key != nil {

pkg/storage/unified/resource/bulk.go

@@ -22,13 +22,11 @@ import (
 )
 
 const grpcMetaKeyCollection = "x-gf-batch-collection"
-const grpcMetaKeyRebuildCollection = "x-gf-batch-rebuild-collection"
 const grpcMetaKeySkipValidation = "x-gf-batch-skip-validation"
 
 // Logged in trace.
 var metadataKeys = []string{
 	grpcMetaKeyCollection,
-	grpcMetaKeyRebuildCollection,
 	grpcMetaKeySkipValidation,
 }
 
@@ -64,10 +62,6 @@ type BulkSettings struct {
 	// All requests will be within this namespace/group/resource
 	Collection []*resourcepb.ResourceKey
 
-	// The batch will include everything from the collection
-	// - all existing values will be removed/replaced if the batch completes successfully
-	RebuildCollection bool
-
 	// The byte[] payload and folder has already been validated - no need to decode and verify
 	SkipValidation bool
 }
@@ -79,9 +73,6 @@ func (x *BulkSettings) ToMD() metadata.MD {
 			md[grpcMetaKeyCollection] = append(md[grpcMetaKeyCollection], SearchID(v))
 		}
 	}
-	if x.RebuildCollection {
-		md[grpcMetaKeyRebuildCollection] = []string{"true"}
-	}
 	if x.SkipValidation {
 		md[grpcMetaKeySkipValidation] = []string{"true"}
 	}
@@ -101,8 +92,6 @@ func NewBulkSettings(md metadata.MD) (BulkSettings, error) {
 				}
 				settings.Collection = append(settings.Collection, key)
 			}
-		case grpcMetaKeyRebuildCollection:
-			settings.RebuildCollection = grpcMetaValueIsTrue(v)
 		case grpcMetaKeySkipValidation:
 			settings.SkipValidation = grpcMetaValueIsTrue(v)
 		}
@@ -187,48 +176,39 @@ func (s *server) BulkProcess(stream resourcepb.BulkStore_BulkProcessServer) erro
 		}
 	}
 
-	if settings.RebuildCollection {
-		for _, k := range settings.Collection {
-			// Can we delete the whole collection
-			rsp, err := s.access.Check(ctx, user, authlib.CheckRequest{
-				Namespace: k.Namespace,
-				Group:     k.Group,
-				Resource:  k.Resource,
-				Verb:      utils.VerbDeleteCollection,
-			}, "")
-			if err != nil || !rsp.Allowed {
-				return sendAndClose(&resourcepb.BulkResponse{
-					Error: &resourcepb.ErrorResult{
-						Message: fmt.Sprintf("Requester must be able to: %s", utils.VerbDeleteCollection),
-						Code:    http.StatusForbidden,
-					},
-				})
-			}
-
-			// This will be called for each request -- with the folder ID
-			//nolint:staticcheck // SA1019: Compile is deprecated but BatchCheck is not yet fully implemented
-			runner.checker[NSGR(k)], _, err = s.access.Compile(ctx, user, authlib.ListRequest{
-				Namespace: k.Namespace,
-				Group:     k.Group,
-				Resource:  k.Resource,
-				Verb:      utils.VerbCreate,
+	for _, k := range settings.Collection {
+		// Can we delete the whole collection
+		rsp, err := s.access.Check(ctx, user, authlib.CheckRequest{
+			Namespace: k.Namespace,
+			Group:     k.Group,
+			Resource:  k.Resource,
+			Verb:      utils.VerbDeleteCollection,
+		}, "")
+		if err != nil || !rsp.Allowed {
+			return sendAndClose(&resourcepb.BulkResponse{
+				Error: &resourcepb.ErrorResult{
+					Message: fmt.Sprintf("Requester must be able to: %s", utils.VerbDeleteCollection),
+					Code:    http.StatusForbidden,
+				},
 			})
-			if err != nil {
-				return sendAndClose(&resourcepb.BulkResponse{
-					Error: &resourcepb.ErrorResult{
-						Message: "Unable to check `create` permission",
-						Code:    http.StatusForbidden,
-					},
-				})
-			}
 		}
-	} else {
-		return sendAndClose(&resourcepb.BulkResponse{
-			Error: &resourcepb.ErrorResult{
-				Message: "Bulk currently only supports RebuildCollection",
-				Code:    http.StatusBadRequest,
-			},
+
+		// This will be called for each request -- with the folder ID
+		//nolint:staticcheck // SA1019: Compile is deprecated but BatchCheck is not yet fully implemented
+		runner.checker[NSGR(k)], _, err = s.access.Compile(ctx, user, authlib.ListRequest{
+			Namespace: k.Namespace,
+			Group:     k.Group,
+			Resource:  k.Resource,
+			Verb:      utils.VerbCreate,
 		})
+		if err != nil {
+			return sendAndClose(&resourcepb.BulkResponse{
+				Error: &resourcepb.ErrorResult{
+					Message: "Unable to check `create` permission",
+					Code:    http.StatusForbidden,
+				},
+			})
+		}
 	}
 
 	backend, ok := s.backend.(BulkProcessingBackend)

pkg/storage/unified/resource/storage_backend.go

@@ -1322,67 +1322,57 @@ func (b *kvStorageBackend) ProcessBulk(ctx context.Context, setting BulkSettings
 	summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection))
 	rsp := &resourcepb.BulkResponse{}
 
-	if setting.RebuildCollection {
-		for _, key := range setting.Collection {
-			events := make([]string, 0)
-			for evtKeyStr, err := range b.eventStore.ListKeysSince(ctx, 1, SortOrderAsc) {
-				if err != nil {
-					b.log.Error("failed to list event: %s", err)
-					return rsp
-				}
-
-				evtKey, err := ParseEventKey(evtKeyStr)
-				if err != nil {
-					b.log.Error("error parsing event key: %s", err)
-					return rsp
-				}
-
-				if evtKey.Group != key.Group || evtKey.Resource != key.Resource || evtKey.Namespace != key.Namespace {
-					continue
-				}
-
-				events = append(events, evtKeyStr)
-			}
-
-			if err := b.eventStore.batchDelete(ctx, events); err != nil {
-				b.log.Error("failed to delete events: %s", err)
+	for _, key := range setting.Collection {
+		events := make([]string, 0)
+		for evtKeyStr, err := range b.eventStore.ListKeysSince(ctx, 1, SortOrderAsc) {
+			if err != nil {
+				b.log.Error("failed to list event: %s", err)
 				return rsp
 			}
 
-			historyKeys := make([]DataKey, 0)
-
-			for dataKey, err := range b.dataStore.Keys(ctx, ListRequestKey{
-				Namespace: key.Namespace,
-				Group:     key.Group,
-				Resource:  key.Resource,
-			}, SortOrderAsc) {
-				if err != nil {
-					b.log.Error("failed to list collection before delete: %s", err)
-					return rsp
-				}
-
-				historyKeys = append(historyKeys, dataKey)
-			}
-
-			previousCount := int64(len(historyKeys))
-			if err := b.dataStore.batchDelete(ctx, historyKeys); err != nil {
-				b.log.Error("failed to delete collection: %s", err)
+			evtKey, err := ParseEventKey(evtKeyStr)
+			if err != nil {
+				b.log.Error("error parsing event key: %s", err)
 				return rsp
 			}
-			summaries[NSGR(key)] = &resourcepb.BulkResponse_Summary{
-				Namespace:     key.Namespace,
-				Group:         key.Group,
-				Resource:      key.Resource,
-				PreviousCount: previousCount,
+
+			if evtKey.Group != key.Group || evtKey.Resource != key.Resource || evtKey.Namespace != key.Namespace {
+				continue
 			}
+
+			events = append(events, evtKeyStr)
 		}
-	} else {
-		for _, key := range setting.Collection {
-			summaries[NSGR(key)] = &resourcepb.BulkResponse_Summary{
-				Namespace: key.Namespace,
-				Group:     key.Group,
-				Resource:  key.Resource,
+
+		if err := b.eventStore.batchDelete(ctx, events); err != nil {
+			b.log.Error("failed to delete events: %s", err)
+			return rsp
+		}
+
+		historyKeys := make([]DataKey, 0)
+
+		for dataKey, err := range b.dataStore.Keys(ctx, ListRequestKey{
+			Namespace: key.Namespace,
+			Group:     key.Group,
+			Resource:  key.Resource,
+		}, SortOrderAsc) {
+			if err != nil {
+				b.log.Error("failed to list collection before delete: %s", err)
+				return rsp
 			}
+
+			historyKeys = append(historyKeys, dataKey)
+		}
+
+		previousCount := int64(len(historyKeys))
+		if err := b.dataStore.batchDelete(ctx, historyKeys); err != nil {
+			b.log.Error("failed to delete collection: %s", err)
+			return rsp
+		}
+		summaries[NSGR(key)] = &resourcepb.BulkResponse_Summary{
+			Namespace:     key.Namespace,
+			Group:         key.Group,
+			Resource:      key.Resource,
+			PreviousCount: previousCount,
 		}
 	}
 

pkg/storage/unified/sql/backend_bulk.go

@@ -225,23 +225,13 @@ func (b *backend) processBulkWithTx(ctx context.Context, tx db.Tx, setting resou
 	summaries := make(map[string]*resourcepb.BulkResponse_Summary, len(setting.Collection))
 
 	// First clear everything in the transaction
-	if setting.RebuildCollection {
-		for _, key := range setting.Collection {
-			summary, err := bulk.deleteCollection(key)
-			if err != nil {
-				return rollbackWithError(err)
-			}
-			summaries[resource.NSGR(key)] = summary
-			rsp.Summary = append(rsp.Summary, summary)
-		}
-	} else {
-		for _, key := range setting.Collection {
-			summaries[resource.NSGR(key)] = &resourcepb.BulkResponse_Summary{
-				Namespace: key.Namespace,
-				Group:     key.Group,
-				Resource:  key.Resource,
-			}
+	for _, key := range setting.Collection {
+		summary, err := bulk.deleteCollection(key)
+		if err != nil {
+			return rollbackWithError(err)
 		}
+		summaries[resource.NSGR(key)] = summary
+		rsp.Summary = append(rsp.Summary, summary)
 	}
 
 	obj := &unstructured.Unstructured{}

pkg/storage/unified/testing/storage_backend.go

@@ -1501,10 +1501,7 @@ func runTestIntegrationGetResourceLastImportTime(t *testing.T, backend resource.
 			},
 		}
 
-		resp := bulk.ProcessBulk(ctx, resource.BulkSettings{
-			Collection:        collections,
-			RebuildCollection: true,
-		}, toBulkIterator(bulkRequests))
+		resp := bulk.ProcessBulk(ctx, resource.BulkSettings{Collection: collections}, toBulkIterator(bulkRequests))
 		require.Nil(t, resp.Error)
 
 		result := collectLastImportedTimes(t, backend, ctx)
@@ -1540,10 +1537,7 @@ func runTestIntegrationGetResourceLastImportTime(t *testing.T, backend resource.
 			Value:  nil,
 		}}
 
-		resp1 := bulk.ProcessBulk(ctx, resource.BulkSettings{
-			Collection:        collections1,
-			RebuildCollection: true,
-		}, toBulkIterator(bulkRequests1))
+		resp1 := bulk.ProcessBulk(ctx, resource.BulkSettings{Collection: collections1}, toBulkIterator(bulkRequests1))
 		require.Nil(t, resp1.Error)
 
 		firstImport := time.Now()
@@ -1571,10 +1565,7 @@ func runTestIntegrationGetResourceLastImportTime(t *testing.T, backend resource.
 			Value:  nil,
 		}}
 
-		resp2 := bulk.ProcessBulk(ctx, resource.BulkSettings{
-			Collection:        collections2,
-			RebuildCollection: false,
-		}, toBulkIterator(bulkRequests2))
+		resp2 := bulk.ProcessBulk(ctx, resource.BulkSettings{Collection: collections2}, toBulkIterator(bulkRequests2))
 		require.Nil(t, resp2.Error)
 
 		secondImport := time.Now()