diff --git a/observability-lib/go.mod b/observability-lib/go.mod index 29aa2d6bd5..75b5cdab76 100644 --- a/observability-lib/go.mod +++ b/observability-lib/go.mod @@ -7,6 +7,7 @@ require ( github.com/grafana/grafana-foundation-sdk/go v0.0.18 github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 + golang.org/x/sync v0.22.0 gopkg.in/yaml.v3 v3.0.1 ) diff --git a/observability-lib/go.sum b/observability-lib/go.sum index 336daf8274..ee7ab80a35 100644 --- a/observability-lib/go.sum +++ b/observability-lib/go.sum @@ -32,6 +32,8 @@ github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/observability-lib/grafana/dashboard.go b/observability-lib/grafana/dashboard.go index 5da06faed0..a195cee29c 100644 --- a/observability-lib/grafana/dashboard.go +++ b/observability-lib/grafana/dashboard.go @@ -37,6 +37,10 @@ func (o *Observability) GenerateJSON() ([]byte, error) { return output, nil } +// defaultConcurrency is the default bound on in-flight HTTP calls for +// alert-rule writes when DeployOptions.Concurrency is unset. +const defaultConcurrency = 8 + type DeployOptions struct { GrafanaURL string GrafanaToken string @@ -45,10 +49,30 @@ type DeployOptions struct { EnableAlerts bool RuleGroupFromDashboard bool // if true, set the alert rule group to the dashboard title on all alerts NotificationTemplates string + // Concurrency bounds in-flight HTTP calls for alert-rule writes (each rule + // is addressed by UID, so rules deploy independently). 0 uses + // defaultConcurrency; 1 restores the previous serial behavior. + Concurrency int + // Cache, when set, memoizes folder resolution and the full alert-rule + // fetch across the DeployToGrafana calls sharing it, so composite deploys + // pay those lookups once instead of once per dashboard. See DeployCache + // for scoping rules. + Cache *DeployCache +} + +func (o *DeployOptions) concurrency() int { + if o.Concurrency <= 0 { + return defaultConcurrency + } + return o.Concurrency } -func resolveDeployFolder(client *api.Client, options *DeployOptions) (*api.Folder, error) { +func resolveDeployFolder(client *api.Client, cache deployCache, options *DeployOptions) (*api.Folder, error) { if options.FolderUID != "" { + key := "uid_" + options.FolderUID + if folder, ok := cache.folder(key); ok { + return folder, nil + } folder, err := client.GetFolderByUID(options.FolderUID) if err != nil { return nil, err @@ -56,10 +80,20 @@ func resolveDeployFolder(client *api.Client, options *DeployOptions) (*api.Folde if folder == nil { return nil, fmt.Errorf("folder with UID %q not found", options.FolderUID) } + cache.setFolder(key, folder) return folder, nil } if options.FolderName != "" { - return client.FindOrCreateFolder(options.FolderName) + key := "name_" + options.FolderName + if folder, ok := cache.folder(key); ok { + return folder, nil + } + folder, err := client.FindOrCreateFolder(options.FolderName) + if err != nil { + return nil, err + } + cache.setFolder(key, folder) + return folder, nil } return nil, nil } @@ -82,26 +116,42 @@ func getAlertRuleByTitle(alerts []alerting.Rule, title string) *alerting.Rule { return nil } -func getAlertRules(grafanaClient *api.Client, dashboardUID *string, folderUID string, alertGroups []alerting.RuleGroup) ([]alerting.Rule, error) { +func getAlertRules(grafanaClient *api.Client, cache deployCache, dashboardUID *string, folderUID string, alertGroups []alerting.RuleGroup) ([]alerting.Rule, error) { + // Fetch the full rule list exactly once to amortize the cost of + // fetching alert rules. + allRules, cached := cache.alertRules() + if !cached { + var errGetAlertRules error + allRules, _, errGetAlertRules = grafanaClient.GetAlertRules() + if errGetAlertRules != nil { + return nil, errGetAlertRules + } + cache.setAlertRules(allRules) + } + var alertsRule []alerting.Rule - var errGetAlertRules error // check for alert rules by dashboard UID if dashboardUID != nil { - alertsRule, errGetAlertRules = grafanaClient.GetAlertRulesByDashboardUID(*dashboardUID) - if errGetAlertRules != nil { - return nil, errGetAlertRules + for _, rule := range allRules { + if rule.Annotations["__dashboardUid__"] == *dashboardUID { + alertsRule = append(alertsRule, rule) + } } } // check for alert rules by folder UID and group name if len(alertGroups) > 0 { + groupNames := make(map[string]bool, len(alertGroups)) for _, alertGroup := range alertGroups { - alertsRulePerGroup, errGetAlertRulesPerGroup := grafanaClient.GetAlertRulesByFolderUIDAndGroupName(folderUID, *alertGroup.Title) - if errGetAlertRulesPerGroup != nil { - return nil, errGetAlertRulesPerGroup + if alertGroup.Title != nil { + groupNames[*alertGroup.Title] = true + } + } + for _, rule := range allRules { + if rule.FolderUID != "" && rule.FolderUID == folderUID && groupNames[rule.RuleGroup] { + alertsRule = append(alertsRule, rule) } - alertsRule = append(alertsRule, alertsRulePerGroup...) } } @@ -114,8 +164,15 @@ func (o *Observability) DeployToGrafana(options *DeployOptions) error { options.GrafanaToken, ) + // Substitute a no-op cache when the caller didn't configure one, so the + // rest of the deploy path never handles a nil cache. + cache := deployCache(noopDeployCache{}) + if options.Cache != nil { + cache = options.Cache + } + // Create or update folder - folder, errFolder := resolveDeployFolder(grafanaClient, options) + folder, errFolder := resolveDeployFolder(grafanaClient, cache, options) if errFolder != nil { return errFolder } @@ -152,7 +209,7 @@ func (o *Observability) DeployToGrafana(options *DeployOptions) error { // If disabling alerts delete alerts for the folder and alert groups scope if folder != nil && !options.EnableAlerts && o.Alerts != nil && len(o.Alerts) > 0 { - alertsRule, errGetAlertRules := getAlertRules(grafanaClient, newDashboard.UID, folder.UID, o.AlertGroups) + alertsRule, errGetAlertRules := getAlertRules(grafanaClient, cache, newDashboard.UID, folder.UID, o.AlertGroups) if errGetAlertRules != nil { return errGetAlertRules } @@ -167,7 +224,7 @@ func (o *Observability) DeployToGrafana(options *DeployOptions) error { // Create or update alerts if folder != nil && options.EnableAlerts && o.Alerts != nil && len(o.Alerts) > 0 { - alertsRule, errGetAlertRules := getAlertRules(grafanaClient, newDashboard.UID, folder.UID, o.AlertGroups) + alertsRule, errGetAlertRules := getAlertRules(grafanaClient, cache, newDashboard.UID, folder.UID, o.AlertGroups) if errGetAlertRules != nil { return errGetAlertRules } @@ -183,7 +240,7 @@ func (o *Observability) DeployToGrafana(options *DeployOptions) error { } // Create alert rules - for _, alert := range o.Alerts { + errUpsertAlerts := parallelFor(o.Alerts, options.concurrency(), func(alert alerting.Rule) error { if folder.UID != "" { alert.FolderUID = folder.UID } @@ -235,6 +292,10 @@ func (o *Observability) DeployToGrafana(options *DeployOptions) error { return errPostAlertRule } } + return nil + }) + if errUpsertAlerts != nil { + return errUpsertAlerts } } diff --git a/observability-lib/grafana/dashboard_deploy_test.go b/observability-lib/grafana/dashboard_deploy_test.go new file mode 100644 index 0000000000..ed094c862e --- /dev/null +++ b/observability-lib/grafana/dashboard_deploy_test.go @@ -0,0 +1,187 @@ +package grafana_test + +import ( + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/grafana/grafana-foundation-sdk/go/alerting" + "github.com/grafana/grafana-foundation-sdk/go/dashboard" + "github.com/stretchr/testify/require" + + "github.com/smartcontractkit/chainlink-common/observability-lib/grafana" +) + +// fakeGrafana stubs the Grafana endpoints used by DeployToGrafana and records +// how the alert-rule endpoints are called. +type fakeGrafana struct { + mu sync.Mutex + + folderGets int + alertRuleGets int + alertRulePosts int + inFlightPosts int + maxInFlight int +} + +func (f *fakeGrafana) handler(t *testing.T) http.Handler { + mux := http.NewServeMux() + + mux.HandleFunc("GET /api/folders", func(w http.ResponseWriter, _ *http.Request) { + f.mu.Lock() + f.folderGets++ + f.mu.Unlock() + writeJSON(t, w, []map[string]any{{"id": 1, "uid": "folder-uid", "title": "Folder"}}) + }) + mux.HandleFunc("GET /api/search", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(t, w, []map[string]any{}) + }) + mux.HandleFunc("POST /api/dashboards/db", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(t, w, map[string]any{"uid": "dash-uid", "url": "/d/dash-uid"}) + }) + mux.HandleFunc("GET /api/v1/provisioning/alert-rules", func(w http.ResponseWriter, _ *http.Request) { + f.mu.Lock() + f.alertRuleGets++ + f.mu.Unlock() + writeJSON(t, w, []map[string]any{}) + }) + mux.HandleFunc("POST /api/v1/provisioning/alert-rules", func(w http.ResponseWriter, _ *http.Request) { + f.mu.Lock() + f.alertRulePosts++ + f.inFlightPosts++ + if f.inFlightPosts > f.maxInFlight { + f.maxInFlight = f.inFlightPosts + } + f.mu.Unlock() + + // Hold the request so overlapping POSTs are observable. + time.Sleep(50 * time.Millisecond) + + f.mu.Lock() + f.inFlightPosts-- + f.mu.Unlock() + + w.WriteHeader(http.StatusCreated) + writeJSON(t, w, map[string]any{}) + }) + mux.HandleFunc("PUT /api/v1/provisioning/folder/{folderUID}/rule-groups/{group}", func(w http.ResponseWriter, _ *http.Request) { + writeJSON(t, w, map[string]any{}) + }) + + return mux +} + +func writeJSON(t *testing.T, w http.ResponseWriter, v any) { + t.Helper() + w.Header().Set("Content-Type", "application/json") + require.NoError(t, json.NewEncoder(w).Encode(v)) +} + +func TestDeployToGrafanaFetchesRulesOnceAndWritesConcurrently(t *testing.T) { + fake := &fakeGrafana{} + server := httptest.NewServer(fake.handler(t)) + t.Cleanup(server.Close) + + const numAlerts = 20 + + title := "Test Dashboard" + o := &grafana.Observability{ + Dashboard: &dashboard.Dashboard{ //nolint:staticcheck + Title: &title, + }, + Alerts: make([]alerting.Rule, numAlerts), + AlertGroups: nil, + } + for i := range o.Alerts { + o.Alerts[i] = alerting.Rule{ + Title: fmt.Sprintf("alert-%d", i), + RuleGroup: "group", + Condition: "A", + Data: []alerting.Query{}, + } + } + + // Alert groups exercise the folder+group lookup in getAlertRules; with the + // dashboard UID lookup it must still fetch the full rule list only once. + group, err := grafana.NewAlertGroup(&grafana.AlertGroupOptions{Title: "group", Interval: 60}).Build() + require.NoError(t, err) + o.AlertGroups = []alerting.RuleGroup{group} + + err = o.DeployToGrafana(&grafana.DeployOptions{ + GrafanaURL: server.URL, + GrafanaToken: "test-token", + FolderName: "Folder", + EnableAlerts: true, + RuleGroupFromDashboard: true, + }) + require.NoError(t, err) + + require.Equal(t, 1, fake.alertRuleGets, "full alert rule list must be fetched exactly once per deploy") + require.Equal(t, numAlerts, fake.alertRulePosts) + require.Greater(t, fake.maxInFlight, 1, "alert rule writes should overlap") + require.LessOrEqual(t, fake.maxInFlight, 8, "alert rule writes must respect the default concurrency bound") +} + +// deployOne deploys a single-dashboard observability with the given options, +// failing the test on any error. +func deployOne(t *testing.T, title string, opts *grafana.DeployOptions) { + t.Helper() + o := &grafana.Observability{ + Dashboard: &dashboard.Dashboard{Title: &title}, //nolint:staticcheck + Alerts: []alerting.Rule{{ + Title: "alert-" + title, + RuleGroup: "group", + Condition: "A", + Data: []alerting.Query{}, + }}, + } + group, err := grafana.NewAlertGroup(&grafana.AlertGroupOptions{Title: "group", Interval: 60}).Build() + require.NoError(t, err) + o.AlertGroups = []alerting.RuleGroup{group} + + require.NoError(t, o.DeployToGrafana(opts)) +} + +func TestDeployToGrafanaSharesCacheAcrossDeploys(t *testing.T) { + fake := &fakeGrafana{} + server := httptest.NewServer(fake.handler(t)) + t.Cleanup(server.Close) + + cache := &grafana.DeployCache{} + for _, title := range []string{"Dashboard A", "Dashboard B"} { + deployOne(t, title, &grafana.DeployOptions{ + GrafanaURL: server.URL, + GrafanaToken: "test-token", + FolderName: "Folder", + EnableAlerts: true, + RuleGroupFromDashboard: true, + Cache: cache, + }) + } + + require.Equal(t, 1, fake.folderGets, "shared cache must resolve the folder once across deploys") + require.Equal(t, 1, fake.alertRuleGets, "shared cache must fetch the full alert rule list once across deploys") +} + +func TestDeployToGrafanaWithoutCacheFetchesPerDeploy(t *testing.T) { + fake := &fakeGrafana{} + server := httptest.NewServer(fake.handler(t)) + t.Cleanup(server.Close) + + for _, title := range []string{"Dashboard A", "Dashboard B"} { + deployOne(t, title, &grafana.DeployOptions{ + GrafanaURL: server.URL, + GrafanaToken: "test-token", + FolderName: "Folder", + EnableAlerts: true, + RuleGroupFromDashboard: true, + }) + } + + require.Equal(t, 2, fake.folderGets, "without a cache each deploy resolves the folder") + require.Equal(t, 2, fake.alertRuleGets, "without a cache each deploy fetches the full alert rule list") +} diff --git a/observability-lib/grafana/deploy_cache.go b/observability-lib/grafana/deploy_cache.go new file mode 100644 index 0000000000..17b184c291 --- /dev/null +++ b/observability-lib/grafana/deploy_cache.go @@ -0,0 +1,82 @@ +package grafana + +import ( + "sync" + + "github.com/grafana/grafana-foundation-sdk/go/alerting" + + "github.com/smartcontractkit/chainlink-common/observability-lib/api" +) + +type deployCache interface { + folder(key string) (*api.Folder, bool) + setFolder(key string, f *api.Folder) + alertRules() ([]alerting.Rule, bool) + setAlertRules(rules []alerting.Rule) +} + +type noopDeployCache struct{} + +func (noopDeployCache) folder(string) (*api.Folder, bool) { return nil, false } +func (noopDeployCache) setFolder(string, *api.Folder) {} +func (noopDeployCache) alertRules() ([]alerting.Rule, bool) { return nil, false } +func (noopDeployCache) setAlertRules([]alerting.Rule) {} + +// DeployCache memoizes values that are invariant across the dashboard deploys +// of a single run — folder resolution and the full alert-rule list — so a +// composite deploy pays those Grafana lookups once instead of once per +// dashboard.// +// Scoping rules: +// - A DeployCache is bound to one Grafana instance: do not share it across +// DeployOptions with different GrafanaURL/GrafanaToken. +// - The alert-rule snapshot is taken on first use and is not invalidated by +// the deploys' own writes. That is safe when each deploy in the run touches +// a disjoint set of rules (e.g. one rule group per dashboard, the default +// with RuleGroupFromDashboard); do not reuse a cache for a repeated deploy +// of the same dashboard. +// +// Safe for concurrent use. +type DeployCache struct { + mu sync.Mutex + // folders by resolution key: "uid_"+FolderUID or "name_"+FolderName. + folders map[string]*api.Folder + // rules holds the full alert-rule list for the instance; rulesReady + // distinguishes "fetched, empty" from "not fetched yet". + rules []alerting.Rule + rulesReady bool +} + +// folder returns the cached folder for key, if any. +func (c *DeployCache) folder(key string) (*api.Folder, bool) { + c.mu.Lock() + defer c.mu.Unlock() + f, ok := c.folders[key] + return f, ok +} + +func (c *DeployCache) setFolder(key string, f *api.Folder) { + if f == nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + if c.folders == nil { + c.folders = make(map[string]*api.Folder) + } + c.folders[key] = f +} + +// alertRules returns the cached full alert-rule list; the second return value +// reports whether it has been fetched (an empty list is a valid snapshot). +func (c *DeployCache) alertRules() ([]alerting.Rule, bool) { + c.mu.Lock() + defer c.mu.Unlock() + return c.rules, c.rulesReady +} + +func (c *DeployCache) setAlertRules(rules []alerting.Rule) { + c.mu.Lock() + defer c.mu.Unlock() + c.rules = rules + c.rulesReady = true +} diff --git a/observability-lib/grafana/parallel.go b/observability-lib/grafana/parallel.go new file mode 100644 index 0000000000..5ea5f77810 --- /dev/null +++ b/observability-lib/grafana/parallel.go @@ -0,0 +1,29 @@ +package grafana + +import "golang.org/x/sync/errgroup" + +// parallelFor runs fn for each item with at most limit calls in flight and +// returns the first non-nil error returned by any invocation. In-flight calls are allowed to finish; +// items whose fn has not started yet may still run after an error occurs, which +// matches the pre-existing partial-apply behavior of the serial loops (a +// failure mid-loop leaves earlier items applied). +// +// Callers must ensure fn is safe for concurrent use: items must be independent +// (e.g. alert rules addressed by distinct UIDs). +func parallelFor[T any](items []T, limit int, fn func(T) error) error { + if limit < 1 { + limit = 1 + } + + var g errgroup.Group + g.SetLimit(limit) + + for _, item := range items { + i := item + g.Go(func() error { + return fn(i) + }) + } + + return g.Wait() +} diff --git a/observability-lib/grafana/parallel_internal_test.go b/observability-lib/grafana/parallel_internal_test.go new file mode 100644 index 0000000000..745f366aa2 --- /dev/null +++ b/observability-lib/grafana/parallel_internal_test.go @@ -0,0 +1,67 @@ +package grafana + +import ( + "errors" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestParallelFor(t *testing.T) { + t.Run("processes all items with bounded concurrency", func(t *testing.T) { + const ( + items = 50 + limit = 5 + ) + + var mu sync.Mutex + inFlight := 0 + maxInFlight := 0 + processed := 0 + + err := parallelFor(make([]int, items), limit, func(int) error { + mu.Lock() + inFlight++ + if inFlight > maxInFlight { + maxInFlight = inFlight + } + mu.Unlock() + + time.Sleep(10 * time.Millisecond) + + mu.Lock() + inFlight-- + processed++ + mu.Unlock() + return nil + }) + + require.NoError(t, err) + require.Equal(t, items, processed) + require.LessOrEqual(t, maxInFlight, limit) + require.Greater(t, maxInFlight, 1) + }) + + t.Run("returns the first error", func(t *testing.T) { + errBoom := errors.New("boom") + err := parallelFor([]int{1, 2, 3}, 3, func(i int) error { + if i == 2 { + return errBoom + } + return nil + }) + require.ErrorIs(t, err, errBoom) + }) + + t.Run("limit below 1 falls back to serial", func(t *testing.T) { + processed := 0 + err := parallelFor([]int{1, 2, 3}, 0, func(int) error { + processed++ + return nil + }) + require.NoError(t, err) + require.Equal(t, 3, processed) + }) +}