Alerting: Persist annotations from multidimensional rules in batches (#56575)

* Reduce piecemeal state fields

* Read data directly off state instead of rule

* Unify state and context into single struct

* Expose contextual information to layer above setNextState

* Work in terms of ContextualState and call historian in batches

* Call annotations service in batches

* Export format state and reason and remove workaround in unrelated test package

* Add new method to annotation service for batch inserting

* Fix loop variable aliasing bug caught by linter, didn't change behavior

* Incl timerange on annotation tests

* Insert one at a time if tags are present

* Point to rule from ContextualState rather than copy fields

* Build annotations and copy data prior to starting goroutine

* Rename to StateTransition

* Use new bulk-insert utility

* Remove rule from StateTransition and pass in directly to historian

* Simplify annotations logic since we have only one rule

* Fix logs and context, nilcheck, simplify method name

* Regenerate mock

pkg/services/annotations/annotationsimpl/annotations.go

@@ -30,6 +30,12 @@ func (r *RepositoryImpl) Save(ctx context.Context, item *annotations.Item) error
 	return r.store.Add(ctx, item)
 }
 
+// SaveMany inserts multiple annotations at once.
+// It does not return IDs associated with created annotations. If you need this functionality, use the single-item Save instead.
+func (r *RepositoryImpl) SaveMany(ctx context.Context, items []annotations.Item) error {
+	return r.store.AddMany(ctx, items)
+}
+
 func (r *RepositoryImpl) Update(ctx context.Context, item *annotations.Item) error {
 	return r.store.Update(ctx, item)
 }

pkg/services/annotations/annotationsimpl/store.go

@@ -8,7 +8,8 @@ import (
 )
 
 type store interface {
-	Add(ctx context.Context, item *annotations.Item) error
+	Add(ctx context.Context, items *annotations.Item) error
+	AddMany(ctx context.Context, items []annotations.Item) error
 	Update(ctx context.Context, item *annotations.Item) error
 	Get(ctx context.Context, query *annotations.ItemQuery) ([]*annotations.ItemDTO, error)
 	Delete(ctx context.Context, params *annotations.DeleteParams) error

pkg/services/annotations/annotationsimpl/xorm_store.go

@@ -13,6 +13,7 @@ import (
 	"github.com/grafana/grafana/pkg/models"
 	ac "github.com/grafana/grafana/pkg/services/accesscontrol"
 	"github.com/grafana/grafana/pkg/services/annotations"
+	"github.com/grafana/grafana/pkg/services/sqlstore"
 	"github.com/grafana/grafana/pkg/services/sqlstore/permissions"
 	"github.com/grafana/grafana/pkg/services/sqlstore/searchstore"
 	"github.com/grafana/grafana/pkg/services/tag"
@@ -63,9 +64,64 @@ func (r *xormRepositoryImpl) Add(ctx context.Context, item *annotations.Item) er
 		if _, err := sess.Table("annotation").Insert(item); err != nil {
 			return err
 		}
+		return r.synchronizeTags(ctx, item)
+	})
+}
 
+// AddMany inserts large batches of annotations at once.
+// It does not return IDs associated with created annotations, and it does not support annotations with tags. If you need this functionality, use the single-item Add instead.
+// This is due to a limitation with some supported databases:
+// We cannot correlate the IDs of batch-inserted records without acquiring a full table lock in MySQL.
+// Annotations have no other uniquifier field, so we also cannot re-query for them after the fact.
+// So, callers can only reliably use this endpoint if they don't care about returned IDs.
+func (r *xormRepositoryImpl) AddMany(ctx context.Context, items []annotations.Item) error {
+	hasTags := make([]annotations.Item, 0)
+	hasNoTags := make([]annotations.Item, 0)
+
+	for i, item := range items {
+		tags := tag.ParseTagPairs(item.Tags)
+		item.Tags = tag.JoinTagPairs(tags)
+		item.Created = timeNow().UnixNano() / int64(time.Millisecond)
+		item.Updated = item.Created
+		if item.Epoch == 0 {
+			item.Epoch = item.Created
+		}
+		if err := r.validateItem(&items[i]); err != nil {
+			return err
+		}
+
+		if len(item.Tags) > 0 {
+			hasTags = append(hasTags, item)
+		} else {
+			hasNoTags = append(hasNoTags, item)
+		}
+	}
+
+	return r.db.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
+		// We can batch-insert every annotation with no tags. If an annotation has tags, we need the ID.
+		opts := sqlstore.NativeSettingsForDialect(r.db.GetDialect())
+		if _, err := sess.BulkInsert("annotation", hasNoTags, opts); err != nil {
+			return err
+		}
+
+		for i, item := range hasTags {
+			if _, err := sess.Table("annotation").Insert(item); err != nil {
+				return err
+			}
+			if err := r.synchronizeTags(ctx, &hasTags[i]); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	})
+}
+
+func (r *xormRepositoryImpl) synchronizeTags(ctx context.Context, item *annotations.Item) error {
+	// Will re-use session if one has already been opened with the same ctx.
+	return r.db.WithDbSession(ctx, func(sess *sqlstore.DBSession) error {
 		if item.Tags != nil {
-			tags, err := r.tagService.EnsureTagsExist(ctx, tags)
+			tags, err := r.tagService.EnsureTagsExist(ctx, tag.ParseTagPairs(item.Tags))
 			if err != nil {
 				return err
 			}

pkg/services/annotations/annotationsimpl/xorm_store_test.go

@@ -163,6 +163,47 @@ func TestIntegrationAnnotations(t *testing.T) {
 		require.Error(t, err)
 		require.ErrorIs(t, err, annotations.ErrBaseTagLimitExceeded)
 
+		t.Run("Can batch-insert annotations", func(t *testing.T) {
+			count := 10
+			items := make([]annotations.Item, count)
+			for i := 0; i < count; i++ {
+				items[i] = annotations.Item{
+					OrgId: 100,
+					Type:  "batch",
+					Epoch: 12,
+				}
+			}
+
+			err := repo.AddMany(context.Background(), items)
+
+			require.NoError(t, err)
+			query := &annotations.ItemQuery{OrgId: 100, SignedInUser: testUser}
+			inserted, err := repo.Get(context.Background(), query)
+			require.NoError(t, err)
+			assert.Len(t, inserted, count)
+		})
+
+		t.Run("Can batch-insert annotations with tags", func(t *testing.T) {
+			count := 10
+			items := make([]annotations.Item, count)
+			for i := 0; i < count; i++ {
+				items[i] = annotations.Item{
+					OrgId: 101,
+					Type:  "batch",
+					Epoch: 12,
+				}
+			}
+			items[0].Tags = []string{"type:test"}
+
+			err := repo.AddMany(context.Background(), items)
+
+			require.NoError(t, err)
+			query := &annotations.ItemQuery{OrgId: 101, SignedInUser: testUser}
+			inserted, err := repo.Get(context.Background(), query)
+			require.NoError(t, err)
+			assert.Len(t, inserted, count)
+		})
+
 		t.Run("Can query for annotation by id", func(t *testing.T) {
 			items, err := repo.Get(context.Background(), &annotations.ItemQuery{
 				OrgId:        1,
@@ -448,6 +489,7 @@ func TestIntegrationAnnotationListingWithRBAC(t *testing.T) {
 		OrgId:       1,
 		DashboardId: 2,
 		Epoch:       10,
+		Tags:        []string{"foo:bar"},
 	}
 	err = repo.Add(context.Background(), dash2Annotation)
 	require.NoError(t, err)