From eb0824a8816fdfa6105956b06b318afdb5391a86 Mon Sep 17 00:00:00 2001 From: Andres Martinez Gotor Date: Thu, 23 Oct 2025 09:22:31 +0200 Subject: [PATCH] Advisor: Add support for multiple orgs (#112577) --- apps/advisor/pkg/app/app.go | 12 +- .../pkg/app/checkregistry/checkregistry.go | 2 + .../pkg/app/checks/datasourcecheck/check.go | 8 +- .../app/checks/datasourcecheck/check_test.go | 2 +- apps/advisor/pkg/app/checks/utils.go | 29 ++- apps/advisor/pkg/app/checks/utils_test.go | 50 ++++-- .../pkg/app/checkscheduler/checkscheduler.go | 166 +++++++++++------- .../app/checkscheduler/checkscheduler_test.go | 16 +- .../checktyperegisterer.go | 79 +++++---- .../checktyperegisterer_test.go | 88 +++++++--- apps/advisor/pkg/app/utils.go | 10 ++ pkg/extensions/enterprise_imports.go | 1 + pkg/registry/apps/advisor/register.go | 3 + pkg/server/wire_gen.go | 4 +- 14 files changed, 323 insertions(+), 147 deletions(-) diff --git a/apps/advisor/pkg/app/app.go b/apps/advisor/pkg/app/app.go index b572f4b3866..0e423957b82 100644 --- a/apps/advisor/pkg/app/app.go +++ b/apps/advisor/pkg/app/app.go @@ -69,12 +69,12 @@ func New(cfg app.Config) (app.App, error) { go func() { logger := log.WithContext(ctx).With("check", check.ID()) logger.Debug("Processing check", "namespace", req.Object.GetNamespace()) - requester, err := identity.GetRequester(ctx) + orgID, err := getOrgIDFromNamespace(req.Object.GetNamespace()) if err != nil { - logger.Error("Error getting requester", "error", err) + logger.Error("Error getting org ID from namespace", "error", err) return } - ctx = identity.WithServiceIdentityContext(context.WithoutCancel(ctx), requester.GetOrgID()) + ctx = identity.WithServiceIdentityContext(context.WithoutCancel(ctx), orgID) err = processCheck(ctx, logger, client, typesClient, req.Object, check) if err != nil { logger.Error("Error processing check", "error", err) @@ -85,12 +85,12 @@ func New(cfg app.Config) (app.App, error) { go func() { logger := log.WithContext(ctx).With("check", check.ID()) logger.Debug("Updating check", "namespace", req.Object.GetNamespace(), "name", req.Object.GetName()) - requester, err := identity.GetRequester(ctx) + orgID, err := getOrgIDFromNamespace(req.Object.GetNamespace()) if err != nil { - logger.Error("Error getting requester", "error", err) + logger.Error("Error getting org ID from namespace", "error", err) return } - ctx = identity.WithServiceIdentityContext(context.WithoutCancel(ctx), requester.GetOrgID()) + ctx = identity.WithServiceIdentityContext(context.WithoutCancel(ctx), orgID) err = processCheckRetry(ctx, logger, client, typesClient, req.Object, check) if err != nil { logger.Error("Error processing check retry", "error", err) diff --git a/apps/advisor/pkg/app/checkregistry/checkregistry.go b/apps/advisor/pkg/app/checkregistry/checkregistry.go index 1ec5f075fc7..661473def2b 100644 --- a/apps/advisor/pkg/app/checkregistry/checkregistry.go +++ b/apps/advisor/pkg/app/checkregistry/checkregistry.go @@ -10,6 +10,7 @@ import ( "github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins/repo" "github.com/grafana/grafana/pkg/services/datasources" + "github.com/grafana/grafana/pkg/services/org" "github.com/grafana/grafana/pkg/services/pluginsintegration/managedplugins" "github.com/grafana/grafana/pkg/services/pluginsintegration/pluginchecker" "github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext" @@ -91,4 +92,5 @@ type AdvisorAppConfig struct { CheckRegistry CheckService PluginConfig map[string]string StackID string + OrgService org.Service } diff --git a/apps/advisor/pkg/app/checks/datasourcecheck/check.go b/apps/advisor/pkg/app/checks/datasourcecheck/check.go index ed8aa35349b..524dae04268 100644 --- a/apps/advisor/pkg/app/checks/datasourcecheck/check.go +++ b/apps/advisor/pkg/app/checks/datasourcecheck/check.go @@ -54,7 +54,13 @@ func New( } func (c *check) Items(ctx context.Context) ([]any, error) { - dss, err := c.DatasourceSvc.GetAllDataSources(ctx, &datasources.GetAllDataSourcesQuery{}) + requester, err := identity.GetRequester(ctx) + if err != nil { + return nil, err + } + dss, err := c.DatasourceSvc.GetDataSources(ctx, &datasources.GetDataSourcesQuery{ + OrgID: requester.GetOrgID(), + }) if err != nil { return nil, err } diff --git a/apps/advisor/pkg/app/checks/datasourcecheck/check_test.go b/apps/advisor/pkg/app/checks/datasourcecheck/check_test.go index cd1e608af5b..8778a520dd9 100644 --- a/apps/advisor/pkg/app/checks/datasourcecheck/check_test.go +++ b/apps/advisor/pkg/app/checks/datasourcecheck/check_test.go @@ -347,7 +347,7 @@ type MockDatasourceSvc struct { dss []*datasources.DataSource } -func (m *MockDatasourceSvc) GetAllDataSources(context.Context, *datasources.GetAllDataSourcesQuery) ([]*datasources.DataSource, error) { +func (m *MockDatasourceSvc) GetDataSources(context.Context, *datasources.GetDataSourcesQuery) ([]*datasources.DataSource, error) { return m.dss, nil } diff --git a/apps/advisor/pkg/app/checks/utils.go b/apps/advisor/pkg/app/checks/utils.go index e58375ee64b..27ba7d50984 100644 --- a/apps/advisor/pkg/app/checks/utils.go +++ b/apps/advisor/pkg/app/checks/utils.go @@ -9,7 +9,7 @@ import ( "github.com/grafana/authlib/types" "github.com/grafana/grafana-app-sdk/resource" advisor "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "github.com/grafana/grafana/pkg/services/org" ) const ( @@ -57,15 +57,26 @@ func NewCheckReportFailureWithMoreInfo( } } -func GetNamespace(stackID string) (string, error) { - if stackID == "" { - return metav1.NamespaceDefault, nil +func GetNamespaces(ctx context.Context, stackID string, orgService org.Service) ([]string, error) { + var namespaces []string + if stackID != "" { + // Single namespace for cloud stack + stackId, err := strconv.ParseInt(stackID, 10, 64) + if err != nil { + return nil, fmt.Errorf("invalid stack id: %s", stackID) + } + namespaces = []string{types.CloudNamespaceFormatter(stackId)} + } else { + // Multiple namespaces for each org + orgs, err := orgService.Search(ctx, &org.SearchOrgsQuery{}) + if err != nil { + return nil, fmt.Errorf("failed to fetch orgs: %w", err) + } + for _, o := range orgs { + namespaces = append(namespaces, types.OrgNamespaceFormatter(o.ID)) + } } - stackId, err := strconv.ParseInt(stackID, 10, 64) - if err != nil { - return "", fmt.Errorf("invalid stack id: %s", stackID) - } - return types.CloudNamespaceFormatter(stackId), nil + return namespaces, nil } func GetStatusAnnotation(obj resource.Object) string { diff --git a/apps/advisor/pkg/app/checks/utils_test.go b/apps/advisor/pkg/app/checks/utils_test.go index ad0fda89de8..2cc55b83faf 100644 --- a/apps/advisor/pkg/app/checks/utils_test.go +++ b/apps/advisor/pkg/app/checks/utils_test.go @@ -1,40 +1,61 @@ package checks import ( + "context" "testing" + "github.com/grafana/grafana/pkg/services/org" "github.com/stretchr/testify/assert" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -func TestGetNamespace(t *testing.T) { +func TestGetNamespaces(t *testing.T) { tests := []struct { name string - input string - expected string + stackID string + orgs []string + expected []string expectedErr string }{ { name: "empty stack ID", - input: "", - expected: metav1.NamespaceDefault, + stackID: "", + orgs: []string{"default"}, + expected: []string{metav1.NamespaceDefault}, }, { name: "valid stack ID", - input: "1234567890", - expected: "stacks-1234567890", + stackID: "1234567890", + orgs: []string{"default"}, + expected: []string{"stacks-1234567890"}, }, { name: "invalid stack ID", - input: "invalid", - expected: "", + stackID: "invalid", + orgs: []string{"default"}, + expected: nil, expectedErr: "invalid stack id: invalid", }, + { + name: "multiple orgs and no stack ID", + stackID: "", + orgs: []string{"default", "org-2"}, + expected: []string{"default", "org-2"}, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - result, err := GetNamespace(tt.input) + fakeOrgService := &mockOrgService{ + SearchFunc: func(ctx context.Context, query *org.SearchOrgsQuery) ([]*org.OrgDTO, error) { + orgs := make([]*org.OrgDTO, len(tt.orgs)) + for i, o := range tt.orgs { + orgs[i] = &org.OrgDTO{ID: int64(i + 1), Name: o} + } + return orgs, nil + }, + } + result, err := GetNamespaces(context.Background(), tt.stackID, fakeOrgService) if tt.expectedErr != "" { assert.EqualError(t, err, tt.expectedErr) } else { @@ -44,3 +65,12 @@ func TestGetNamespace(t *testing.T) { }) } } + +type mockOrgService struct { + org.Service + SearchFunc func(ctx context.Context, query *org.SearchOrgsQuery) ([]*org.OrgDTO, error) +} + +func (m *mockOrgService) Search(ctx context.Context, query *org.SearchOrgsQuery) ([]*org.OrgDTO, error) { + return m.SearchFunc(ctx, query) +} diff --git a/apps/advisor/pkg/app/checkscheduler/checkscheduler.go b/apps/advisor/pkg/app/checkscheduler/checkscheduler.go index 6860d79ece5..2cf0c488794 100644 --- a/apps/advisor/pkg/app/checkscheduler/checkscheduler.go +++ b/apps/advisor/pkg/app/checkscheduler/checkscheduler.go @@ -16,6 +16,7 @@ import ( advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" "github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry" "github.com/grafana/grafana/apps/advisor/pkg/app/checks" + "github.com/grafana/grafana/pkg/services/org" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -37,8 +38,9 @@ type Runner struct { typesClient resource.Client defaultEvalInterval time.Duration maxHistory int - namespace string log logging.Logger + orgService org.Service + stackID string } // NewRunner creates a new Runner. @@ -49,6 +51,7 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { return nil, fmt.Errorf("invalid config type") } checkRegistry := specificConfig.CheckRegistry + orgService := specificConfig.OrgService evalInterval, err := getEvaluationInterval(specificConfig.PluginConfig) if err != nil { return nil, err @@ -57,10 +60,6 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { if err != nil { return nil, err } - namespace, err := checks.GetNamespace(specificConfig.StackID) - if err != nil { - return nil, err - } // Prepare storage client clientGenerator := k8s.NewClientRegistry(cfg.KubeConfig, k8s.ClientConfig{}) @@ -79,8 +78,9 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { typesClient: typesClient, defaultEvalInterval: evalInterval, maxHistory: maxHistory, - namespace: namespace, log: log.With("runner", "advisor.checkscheduler"), + orgService: orgService, + stackID: specificConfig.StackID, }, nil } @@ -89,49 +89,82 @@ func (r *Runner) Run(ctx context.Context) error { // We still need the context to eventually be cancelled to exit this function // but we don't want the requests to fail because of it ctxWithoutCancel := context.WithoutCancel(ctx) - lastCreated, err := r.checkLastCreated(ctxWithoutCancel, logger) + + // Determine namespaces based on StackID or OrgID + namespaces, err := checks.GetNamespaces(ctxWithoutCancel, r.stackID, r.orgService) + if err != nil { + return fmt.Errorf("failed to get namespaces: %w", err) + } + + logger.Debug("Scheduling checks", "namespaces", len(namespaces)) + + // Get the last created time for this specific namespace + lastCreatedMap, err := r.checkLastCreated(ctx, logger, namespaces) if err != nil { logger.Error("Error getting last check creation time", "error", err) return err } - // If there are checks already created, run an initial cleanup to remove old checks - if !lastCreated.IsZero() { - err = r.cleanupChecks(ctxWithoutCancel, logger) - if err != nil { - logger.Error("Error cleaning up old check reports", "error", err) - return err + + // If there are checks already created, run an initial cleanup + for _, namespace := range namespaces { + logger = logger.With("namespace", namespace) + lastCreated := lastCreatedMap[namespace] + + if !lastCreated.IsZero() { + err = r.cleanupChecks(ctx, logger, namespace) + if err != nil { + logger.Error("Error cleaning up old check reports", "error", err) + return err + } + err = r.markUnprocessedChecks(ctx, logger, namespace) + if err != nil { + logger.Error("Error marking unprocessed checks", "error", err) + return err + } } } - nextEvalTime := r.getNextEvalTime(r.defaultEvalInterval, lastCreated) + nextEvalTime := r.getNextEvalTime(r.defaultEvalInterval, lastCreatedMap) ticker := time.NewTicker(nextEvalTime) defer ticker.Stop() for { select { case <-ticker.C: - lastCreated, err := r.checkLastCreated(ctxWithoutCancel, logger) + // Get the current last created time for this namespace + lastCreatedMap, err := r.checkLastCreated(ctx, logger, namespaces) if err != nil { logger.Error("Error getting last check creation time", "error", err) return err } - // If there are checks already created, then we can automatically create more - if !lastCreated.IsZero() { - err = r.createChecks(ctxWithoutCancel, logger) - if err != nil { - logger.Error("Error creating new check reports", "error", err) - } + for _, namespace := range namespaces { + logger = logger.With("namespace", namespace) + lastCreated := lastCreatedMap[namespace] - // Clean up old checks to avoid going over the limit - err = r.cleanupChecks(ctxWithoutCancel, logger) - if err != nil { - logger.Error("Error cleaning up old check reports", "error", err) + // If there are checks already created and they are older than the evaluation interval + // then we can automatically create more + if !lastCreated.IsZero() && lastCreated.Before(time.Now().Add(-r.defaultEvalInterval)) { + err = r.createChecks(ctx, logger, namespace) + if err != nil { + logger.Error("Error creating new check reports", "error", err) + return err + } + + // Clean up old checks to avoid going over the limit + err = r.cleanupChecks(ctx, logger, namespace) + if err != nil { + logger.Error("Error cleaning up old check reports", "error", err) + return err + } + + // Update the last created time with the new created checks + lastCreatedMap[namespace] = time.Now() } } // Reset the ticker to the next send interval - nextEvalTime = r.getNextEvalTime(r.defaultEvalInterval, lastCreated) + nextEvalTime = r.getNextEvalTime(r.defaultEvalInterval, lastCreatedMap) ticker.Reset(nextEvalTime) case <-ctx.Done(): return ctx.Err() @@ -139,8 +172,8 @@ func (r *Runner) Run(ctx context.Context) error { } } -func (r *Runner) listChecks(ctx context.Context, logger logging.Logger) ([]resource.Object, error) { - list, err := r.checksClient.List(ctx, r.namespace, resource.ListOptions{ +func (r *Runner) listChecks(ctx context.Context, logger logging.Logger, namespace string) ([]resource.Object, error) { + list, err := r.checksClient.List(ctx, namespace, resource.ListOptions{ Limit: 1000, // Avoid pagination for normal uses cases, which is a costly operation }) if err != nil { @@ -150,7 +183,7 @@ func (r *Runner) listChecks(ctx context.Context, logger logging.Logger) ([]resou checks := list.GetItems() for list.GetContinue() != "" { logger.Debug("List has continue token, listing next page", "continue", list.GetContinue()) - list, err = r.checksClient.List(ctx, r.namespace, resource.ListOptions{Continue: list.GetContinue(), Limit: 1000}) + list, err = r.checksClient.List(ctx, namespace, resource.ListOptions{Continue: list.GetContinue(), Limit: 1000}) if err != nil { return nil, err } @@ -159,38 +192,48 @@ func (r *Runner) listChecks(ctx context.Context, logger logging.Logger) ([]resou return checks, nil } -// checkLastCreated returns the creation time of the last check created -// regardless of its ID. This assumes that the checks are created in batches -// so a batch will have a similar creation time. +// checkLastCreated returns the creation time of the last check created for a specific namespace. +// This assumes that the checks are created in batches so a batch will have a similar creation time. // In case it finds an unprocessed check from a previous run, it will set it to error. -func (r *Runner) checkLastCreated(ctx context.Context, log logging.Logger) (time.Time, error) { - checkList, err := r.listChecks(ctx, log) - if err != nil { - return time.Time{}, err - } - lastCreated := time.Time{} - for _, item := range checkList { - itemCreated := item.GetCreationTimestamp().Time - if itemCreated.After(lastCreated) { - lastCreated = itemCreated +func (r *Runner) checkLastCreated(ctx context.Context, log logging.Logger, namespaces []string) (map[string]time.Time, error) { + lastCreated := map[string]time.Time{} + for _, namespace := range namespaces { + checkList, err := r.listChecks(ctx, log, namespace) + if err != nil { + return nil, err } - - // If the check is unprocessed, set it to error - if checks.GetStatusAnnotation(item) == "" { - log.Info("Check is unprocessed, marking as error", "check", item.GetStaticMetadata().Identifier()) - err := checks.SetStatusAnnotation(ctx, r.checksClient, item, checks.StatusAnnotationError) - if err != nil { - log.Error("Error setting check status to error", "error", err) + for _, item := range checkList { + itemCreated := item.GetCreationTimestamp().Time + if itemCreated.After(lastCreated[namespace]) { + lastCreated[namespace] = itemCreated } } } return lastCreated, nil } +func (r *Runner) markUnprocessedChecks(ctx context.Context, log logging.Logger, namespace string) error { + checkList, err := r.listChecks(ctx, log, namespace) + if err != nil { + return err + } + for _, item := range checkList { + if checks.GetStatusAnnotation(item) == "" { + log.Info("Check is unprocessed, marking as error", "check", item.GetStaticMetadata().Identifier()) + err := checks.SetStatusAnnotation(ctx, r.checksClient, item, checks.StatusAnnotationError) + if err != nil { + log.Error("Error setting check status to error", "error", err) + return err + } + } + } + return nil +} + // createChecks creates a new check for each check type in the registry. -func (r *Runner) createChecks(ctx context.Context, logger logging.Logger) error { +func (r *Runner) createChecks(ctx context.Context, logger logging.Logger, namespace string) error { // List existing CheckType objects - list, err := r.typesClient.List(ctx, r.namespace, resource.ListOptions{}) + list, err := r.typesClient.List(ctx, namespace, resource.ListOptions{}) if err != nil { return fmt.Errorf("error listing check types: %w", err) } @@ -200,7 +243,7 @@ func (r *Runner) createChecks(ctx context.Context, logger logging.Logger) error for !allChecksRegistered && retryCount < waitMaxRetries { logger.Info("Waiting for all check types to be registered", "retryCount", retryCount, "waitInterval", waitInterval) time.Sleep(waitInterval) - list, err = r.typesClient.List(ctx, r.namespace, resource.ListOptions{}) + list, err = r.typesClient.List(ctx, namespace, resource.ListOptions{}) if err != nil { return fmt.Errorf("error listing check types: %w", err) } @@ -218,7 +261,7 @@ func (r *Runner) createChecks(ctx context.Context, logger logging.Logger) error obj := &advisorv0alpha1.Check{ ObjectMeta: metav1.ObjectMeta{ GenerateName: "check-", - Namespace: r.namespace, + Namespace: namespace, Labels: map[string]string{ checks.TypeLabel: checkType.Spec.Name, }, @@ -235,13 +278,13 @@ func (r *Runner) createChecks(ctx context.Context, logger logging.Logger) error } // cleanupChecks deletes the olders checks if the number of checks exceeds the limit. -func (r *Runner) cleanupChecks(ctx context.Context, logger logging.Logger) error { - checkList, err := r.listChecks(ctx, logger) +func (r *Runner) cleanupChecks(ctx context.Context, logger logging.Logger, namespace string) error { + checkList, err := r.listChecks(ctx, logger, namespace) if err != nil { return err } - logger.Debug("Cleaning up checks", "numChecks", len(checkList)) + logger.Debug("Cleaning up checks", "namespace", namespace, "numChecks", len(checkList)) // organize checks by type checksByType := map[string][]resource.Object{} @@ -294,12 +337,15 @@ func getEvaluationInterval(pluginConfig map[string]string) (time.Duration, error return evaluationInterval, nil } -func (r *Runner) getNextEvalTime(defaultEvaluationInterval time.Duration, lastCreated time.Time) time.Duration { +func (r *Runner) getNextEvalTime(defaultEvaluationInterval time.Duration, lastCreated map[string]time.Time) time.Duration { nextEvalTime := defaultEvaluationInterval - baseTime := lastCreated - if lastCreated.IsZero() { - baseTime = time.Now() + // Get the oldest last created time + baseTime := time.Now() + for _, lastNamespacedCreated := range lastCreated { + if !lastNamespacedCreated.IsZero() && lastNamespacedCreated.Before(baseTime) { + baseTime = lastNamespacedCreated + } } // Calculate the next evaluation time and add random variation diff --git a/apps/advisor/pkg/app/checkscheduler/checkscheduler_test.go b/apps/advisor/pkg/app/checkscheduler/checkscheduler_test.go index ce44dc07c29..0fe0d5ea77d 100644 --- a/apps/advisor/pkg/app/checkscheduler/checkscheduler_test.go +++ b/apps/advisor/pkg/app/checkscheduler/checkscheduler_test.go @@ -12,6 +12,7 @@ import ( advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" "github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry" "github.com/grafana/grafana/apps/advisor/pkg/app/checks" + "github.com/grafana/grafana/pkg/services/org" "github.com/stretchr/testify/assert" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -212,7 +213,8 @@ func TestRunner_Run_UnprocessedChecks(t *testing.T) { Items: []advisorv0alpha1.Check{ { ObjectMeta: metav1.ObjectMeta{ - Name: "unprocessed-check", + Name: "unprocessed-check", + CreationTimestamp: metav1.NewTime(time.Now().Add(-1 * time.Hour)), // No status annotation - unprocessed }, }, @@ -331,8 +333,9 @@ func createTestRunnerWithRegistry(checkClient, typesClient *MockClient, checkReg typesClient: typesClient, defaultEvalInterval: 5 * time.Millisecond, maxHistory: defaultMaxHistory, - namespace: "test-namespace", log: &logging.NoOpLogger{}, + orgService: &mockOrgService{orgs: []*org.OrgDTO{{ID: 1}}}, + stackID: "", } } @@ -383,3 +386,12 @@ func (m *mockCheck) ID() string { func (m *mockCheck) Steps() []checks.Step { return m.steps } + +type mockOrgService struct { + org.Service + orgs []*org.OrgDTO +} + +func (m *mockOrgService) Search(ctx context.Context, query *org.SearchOrgsQuery) ([]*org.OrgDTO, error) { + return m.orgs, nil +} diff --git a/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer.go b/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer.go index 67e588ec2b7..453772a57bd 100644 --- a/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer.go +++ b/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer.go @@ -16,6 +16,7 @@ import ( advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" "github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry" "github.com/grafana/grafana/apps/advisor/pkg/app/checks" + "github.com/grafana/grafana/pkg/services/org" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -26,7 +27,8 @@ import ( type Runner struct { checkRegistry checkregistry.CheckService client resource.Client - namespace string + orgService org.Service + stackID string log logging.Logger retryAttempts int retryDelay time.Duration @@ -40,10 +42,7 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { return nil, fmt.Errorf("invalid config type") } checkRegistry := specificConfig.CheckRegistry - namespace, err := checks.GetNamespace(specificConfig.StackID) - if err != nil { - return nil, err - } + orgService := specificConfig.OrgService // Prepare storage client clientGenerator := k8s.NewClientRegistry(cfg.KubeConfig, k8s.ClientConfig{}) @@ -55,7 +54,8 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { return &Runner{ checkRegistry: checkRegistry, client: client, - namespace: namespace, + orgService: orgService, + stackID: specificConfig.StackID, log: log.With("runner", "advisor.checktyperegisterer"), retryAttempts: 5, retryDelay: time.Second * 10, @@ -64,36 +64,47 @@ func New(cfg app.Config, log logging.Logger) (app.Runnable, error) { func (r *Runner) Run(ctx context.Context) error { logger := r.log.WithContext(ctx) - for _, t := range r.checkRegistry.Checks() { - steps := t.Steps() - stepTypes := make([]advisorv0alpha1.CheckTypeStep, len(steps)) - for i, s := range steps { - stepTypes[i] = advisorv0alpha1.CheckTypeStep{ - Title: s.Title(), - Description: s.Description(), - StepID: s.ID(), - Resolution: s.Resolution(), + + // Determine namespaces based on StackID or OrgID + namespaces, err := checks.GetNamespaces(ctx, r.stackID, r.orgService) + if err != nil { + return fmt.Errorf("failed to get namespaces: %w", err) + } + logger.Debug("Registering check types", "namespaces", len(namespaces)) + + // Register check types in each namespace + for _, namespace := range namespaces { + for _, t := range r.checkRegistry.Checks() { + steps := t.Steps() + stepTypes := make([]advisorv0alpha1.CheckTypeStep, len(steps)) + for i, s := range steps { + stepTypes[i] = advisorv0alpha1.CheckTypeStep{ + Title: s.Title(), + Description: s.Description(), + StepID: s.ID(), + Resolution: s.Resolution(), + } } - } - obj := &advisorv0alpha1.CheckType{ - ObjectMeta: metav1.ObjectMeta{ - Name: t.ID(), - Namespace: r.namespace, - Annotations: map[string]string{ - checks.NameAnnotation: t.Name(), - // Flag to indicate feature availability - checks.RetryAnnotation: "1", - checks.IgnoreStepsAnnotation: "1", + obj := &advisorv0alpha1.CheckType{ + ObjectMeta: metav1.ObjectMeta{ + Name: t.ID(), + Namespace: namespace, + Annotations: map[string]string{ + checks.NameAnnotation: t.Name(), + // Flag to indicate feature availability + checks.RetryAnnotation: "1", + checks.IgnoreStepsAnnotation: "1", + }, }, - }, - Spec: advisorv0alpha1.CheckTypeSpec{ - Name: t.ID(), - Steps: stepTypes, - }, - } - err := r.registerCheckType(ctx, logger, t.ID(), obj) - if err != nil { - return err + Spec: advisorv0alpha1.CheckTypeSpec{ + Name: t.ID(), + Steps: stepTypes, + }, + } + err := r.registerCheckType(ctx, logger, t.ID(), obj) + if err != nil { + return err + } } } return nil diff --git a/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer_test.go b/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer_test.go index 0ac148989e5..99275d43eee 100644 --- a/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer_test.go +++ b/apps/advisor/pkg/app/checktyperegisterer/checktyperegisterer_test.go @@ -10,6 +10,7 @@ import ( "github.com/grafana/grafana-app-sdk/resource" advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" "github.com/grafana/grafana/apps/advisor/pkg/app/checks" + "github.com/grafana/grafana/pkg/services/org" k8sErrs "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" @@ -67,14 +68,17 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { tests := []struct { name string checks []checks.Check + stackID string + orgService org.Service getFunc func(ctx context.Context, id resource.Identifier) (resource.Object, error) createFunc func(ctx context.Context, id resource.Identifier, obj resource.Object, opts resource.CreateOptions) (resource.Object, error) updateFunc func(ctx context.Context, id resource.Identifier, obj resource.Object, opts resource.UpdateOptions) (resource.Object, error) expectedErr error }{ { - name: "successful create", - checks: []checks.Check{newMockCheck}, + name: "successful create", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return nil, k8sErrs.NewNotFound(schema.GroupResource{}, id.Name) }, @@ -85,8 +89,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "resource exists with different annotations, should update", - checks: []checks.Check{newMockCheck}, + name: "resource exists with different annotations, should update", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectDifferentAnnotations, nil }, @@ -96,8 +101,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "resource exists with different steps, should update", - checks: []checks.Check{newMockCheck}, + name: "resource exists with different steps, should update", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectDifferentSteps, nil }, @@ -107,8 +113,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "resource exists with same annotations and steps, should not update", - checks: []checks.Check{newMockCheck}, + name: "resource exists with same annotations and steps, should not update", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectSameContent, nil }, @@ -118,8 +125,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "resource exists, with custom annotations preserved", - checks: []checks.Check{newMockCheck}, + name: "resource exists, with custom annotations preserved", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectDifferentAnnotations, nil }, @@ -132,8 +140,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "create error", - checks: []checks.Check{newMockCheck}, + name: "create error", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return nil, k8sErrs.NewNotFound(schema.GroupResource{}, id.Name) }, @@ -144,8 +153,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: errors.New("create error"), }, { - name: "update error", - checks: []checks.Check{newMockCheck}, + name: "update error", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectDifferentAnnotations, nil }, @@ -155,8 +165,9 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: errors.New("update error"), }, { - name: "shutting down error", - checks: []checks.Check{newMockCheck}, + name: "shutting down error", + checks: []checks.Check{newMockCheck}, + stackID: "123", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { return existingObjectDifferentAnnotations, nil }, @@ -166,14 +177,33 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { expectedErr: nil, }, { - name: "custom namespace", - checks: []checks.Check{newMockCheck}, + name: "cloud stack namespace", + checks: []checks.Check{newMockCheck}, + stackID: "456", getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { - return existingObjectDifferentAnnotations, nil + return nil, k8sErrs.NewNotFound(schema.GroupResource{}, id.Name) }, createFunc: func(ctx context.Context, id resource.Identifier, obj resource.Object, opts resource.CreateOptions) (resource.Object, error) { - if obj.GetNamespace() != "custom-namespace" { - return nil, fmt.Errorf("expected namespace %s, got %s", "custom-namespace", obj.GetNamespace()) + if obj.GetNamespace() != "stack-456" { + return nil, fmt.Errorf("expected namespace %s, got %s", "stack-456", obj.GetNamespace()) + } + return obj, nil + }, + expectedErr: nil, + }, + { + name: "multiple orgs", + checks: []checks.Check{newMockCheck}, + stackID: "", + orgService: &mockOrgService{orgs: []*org.OrgDTO{{ID: 1, Name: "Org1"}, {ID: 2, Name: "Org2"}}}, + getFunc: func(ctx context.Context, id resource.Identifier) (resource.Object, error) { + return nil, k8sErrs.NewNotFound(schema.GroupResource{}, id.Name) + }, + createFunc: func(ctx context.Context, id resource.Identifier, obj resource.Object, opts resource.CreateOptions) (resource.Object, error) { + // Should create in both org-1 and org-2 namespaces + ns := obj.GetNamespace() + if ns != "org-1" && ns != "org-2" { + return nil, fmt.Errorf("expected namespace org-1 or org-2, got %s", ns) } return obj, nil }, @@ -183,6 +213,10 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { + orgSvc := tt.orgService + if orgSvc == nil { + orgSvc = &mockOrgService{orgs: []*org.OrgDTO{}} + } r := &Runner{ checkRegistry: &mockCheckRegistry{checks: tt.checks}, client: &mockClient{ @@ -190,7 +224,8 @@ func TestCheckTypesRegisterer_Run(t *testing.T) { createFunc: tt.createFunc, updateFunc: tt.updateFunc, }, - namespace: "custom-namespace", + orgService: orgSvc, + stackID: tt.stackID, log: logging.DefaultLogger, retryAttempts: 1, retryDelay: 0, @@ -298,3 +333,12 @@ func (m *mockClient) Update(ctx context.Context, id resource.Identifier, obj res } return nil, errors.New("not implemented") } + +type mockOrgService struct { + org.Service + orgs []*org.OrgDTO +} + +func (m *mockOrgService) Search(ctx context.Context, query *org.SearchOrgsQuery) ([]*org.OrgDTO, error) { + return m.orgs, nil +} diff --git a/apps/advisor/pkg/app/utils.go b/apps/advisor/pkg/app/utils.go index d2810ec3644..9e88f9b5b11 100644 --- a/apps/advisor/pkg/app/utils.go +++ b/apps/advisor/pkg/app/utils.go @@ -9,6 +9,7 @@ import ( "sync" "time" + "github.com/grafana/authlib/types" "github.com/grafana/grafana-app-sdk/logging" "github.com/grafana/grafana-app-sdk/resource" advisorv0alpha1 "github.com/grafana/grafana/apps/advisor/pkg/apis/advisor/v0alpha1" @@ -317,3 +318,12 @@ func waitForRetryAnnotation(ctx context.Context, log logging.Logger, client reso log.Debug("Retry annotation persisted", "check", obj.GetName(), "item", itemToRetry) return nil } + +// getOrgIDFromNamespace extracts the org ID from a namespace using the standard authlib parser. +func getOrgIDFromNamespace(namespace string) (int64, error) { + info, err := types.ParseNamespace(namespace) + if err != nil { + return 0, fmt.Errorf("failed to parse namespace %s: %w", namespace, err) + } + return info.OrgID, nil +} diff --git a/pkg/extensions/enterprise_imports.go b/pkg/extensions/enterprise_imports.go index 2dbdad4a9d6..9c052688d8f 100644 --- a/pkg/extensions/enterprise_imports.go +++ b/pkg/extensions/enterprise_imports.go @@ -57,4 +57,5 @@ import ( _ "github.com/grafana/tempo/pkg/traceql" _ "github.com/grafana/grafana/apps/alerting/alertenrichment/pkg/apis/alertenrichment/v1beta1" + _ "github.com/grafana/grafana/apps/scope/pkg/apis/scope/v0alpha1" ) diff --git a/pkg/registry/apps/advisor/register.go b/pkg/registry/apps/advisor/register.go index 19a94551dc5..b3a58e8ed9b 100644 --- a/pkg/registry/apps/advisor/register.go +++ b/pkg/registry/apps/advisor/register.go @@ -9,6 +9,7 @@ import ( "github.com/grafana/grafana/apps/advisor/pkg/app/checkregistry" "github.com/grafana/grafana/pkg/services/apiserver/builder" "github.com/grafana/grafana/pkg/services/apiserver/builder/runner" + "github.com/grafana/grafana/pkg/services/org" "github.com/grafana/grafana/pkg/setting" ) @@ -19,6 +20,7 @@ type AdvisorAppProvider struct { func RegisterApp( checkRegistry checkregistry.CheckService, cfg *setting.Cfg, + orgService org.Service, ) *AdvisorAppProvider { provider := &AdvisorAppProvider{} pluginConfig := cfg.PluginSettings["grafana-advisor-app"] @@ -26,6 +28,7 @@ func RegisterApp( CheckRegistry: checkRegistry, PluginConfig: pluginConfig, StackID: cfg.StackID, + OrgService: orgService, } appCfg := &runner.AppBuilderConfig{ OpenAPIDefGetter: advisorv0alpha1.GetOpenAPIDefinitions, diff --git a/pkg/server/wire_gen.go b/pkg/server/wire_gen.go index 330e980897f..57f731fdb54 100644 --- a/pkg/server/wire_gen.go +++ b/pkg/server/wire_gen.go @@ -794,7 +794,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api zanzanaReconciler := dualwrite2.ProvideZanzanaReconciler(cfg, featureToggles, zanzanaClient, sqlStore, serverLockService, folderimplService) investigationsAppProvider := investigations.RegisterApp(cfg) checkregistryService := checkregistry.ProvideService(service15, pluginstoreService, plugincontextProvider, middlewareHandler, plugincheckerService, repoManager, preinstallImpl, managedpluginsNoop, noop, ssosettingsimplService, cfg, pluginerrsStore) - advisorAppProvider := advisor2.RegisterApp(checkregistryService, cfg) + advisorAppProvider := advisor2.RegisterApp(checkregistryService, cfg, orgService) appregistryService, err := appregistry.ProvideBuilderRunners(apiserverService, eventualRestConfigProvider, featureToggles, investigationsAppProvider, advisorAppProvider, cfg) if err != nil { return nil, err @@ -1406,7 +1406,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac zanzanaReconciler := dualwrite2.ProvideZanzanaReconciler(cfg, featureToggles, zanzanaClient, sqlStore, serverLockService, folderimplService) investigationsAppProvider := investigations.RegisterApp(cfg) checkregistryService := checkregistry.ProvideService(service15, pluginstoreService, plugincontextProvider, middlewareHandler, plugincheckerService, repoManager, preinstallImpl, managedpluginsNoop, noop, ssosettingsimplService, cfg, pluginerrsStore) - advisorAppProvider := advisor2.RegisterApp(checkregistryService, cfg) + advisorAppProvider := advisor2.RegisterApp(checkregistryService, cfg, orgService) appregistryService, err := appregistry.ProvideBuilderRunners(apiserverService, eventualRestConfigProvider, featureToggles, investigationsAppProvider, advisorAppProvider, cfg) if err != nil { return nil, err