public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[v10.4.x] Alerting/Annotations: Prevent panics from composite store jobs from crashing Grafana (#84596)

Browse Source 
Alerting/Annotations: Prevent panics from composite store jobs from crashing Grafana (#83459)

* Don't directly use pointer to json

* Don't crash entire process if a store job panics

* Add debug logs when failing to parse/handle Loki entries

(cherry picked from commit af528d2f66)

Co-authored-by: William Wernert <william.wernert@grafana.com>
This commit is contained in:
grafana-delivery-bot[bot]
2024-03-15 18:59:14 +02:00
committed by GitHub
co-authored by William Wernert
parent de5b771270
commit 591cb82b90
7 changed files with 116 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/annotations/annotationsimpl/composite_store.go
+34 -3
View File
@@ -2,8 +2,11 @@ package annotationsimpl
import (
"context"
"fmt"
"sort"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/dskit/concurrency"
"github.com/grafana/grafana/pkg/services/annotations"
"github.com/grafana/grafana/pkg/services/annotations/accesscontrol"
@@ -11,20 +14,29 @@ import (
// CompositeStore is a read store that combines two or more read stores, and queries all stores in parallel.
type CompositeStore struct {
logger log.Logger
readers []readStore
}
func NewCompositeStore(readers ...readStore) *CompositeStore {
func NewCompositeStore(logger log.Logger, readers ...readStore) *CompositeStore {
return &CompositeStore{
logger: logger,
readers: readers,
}
}
// Satisfy the commonStore interface, in practice this is not used.
func (c *CompositeStore) Type() string {
return "composite"
}
// Get returns annotations from all stores, and combines the results.
func (c *CompositeStore) Get(ctx context.Context, query *annotations.ItemQuery, accessResources *accesscontrol.AccessResources) ([]*annotations.ItemDTO, error) {
itemCh := make(chan []*annotations.ItemDTO, len(c.readers))
err := concurrency.ForEachJob(ctx, len(c.readers), len(c.readers), func(ctx context.Context, i int) error {
err := concurrency.ForEachJob(ctx, len(c.readers), len(c.readers), func(ctx context.Context, i int) (err error) {
defer handleJobPanic(c.logger, c.readers[i].Type(), &err)
items, err := c.readers[i].Get(ctx, query, accessResources)
itemCh <- items
return err
@@ -47,7 +59,9 @@ func (c *CompositeStore) Get(ctx context.Context, query *annotations.ItemQuery,
func (c *CompositeStore) GetTags(ctx context.Context, query *annotations.TagsQuery) (annotations.FindTagsResult, error) {
resCh := make(chan annotations.FindTagsResult, len(c.readers))
err := concurrency.ForEachJob(ctx, len(c.readers), len(c.readers), func(ctx context.Context, i int) error {
err := concurrency.ForEachJob(ctx, len(c.readers), len(c.readers), func(ctx context.Context, i int) (err error) {
defer handleJobPanic(c.logger, c.readers[i].Type(), &err)
res, err := c.readers[i].GetTags(ctx, query)
resCh <- res
return err
@@ -65,3 +79,20 @@ func (c *CompositeStore) GetTags(ctx context.Context, query *annotations.TagsQue
return annotations.FindTagsResult{Tags: res}, nil
}
// handleJobPanic is a helper function that recovers from a panic in a concurrent job.,
// It will log the error and set the job error if it is not nil.
func handleJobPanic(logger log.Logger, storeType string, jobErr *error) {
if r := recover(); r != nil {
logger.Error("Annotation store panic", "error", r, "store", storeType, "stack", log.Stack(1))
errMsg := "concurrent job panic"
if jobErr != nil {
err := fmt.Errorf(errMsg)
if panicErr, ok := r.(error); ok {
err = fmt.Errorf("%s: %w", errMsg, panicErr)
}
*jobErr = err
}
}
}