diff --git a/pkg/operators/provisioning/config.go b/pkg/operators/provisioning/config.go index 7882062be12..983107ef4df 100644 --- a/pkg/operators/provisioning/config.go +++ b/pkg/operators/provisioning/config.go @@ -8,11 +8,14 @@ import ( "time" "github.com/grafana/authlib/authn" + "github.com/prometheus/client_golang/prometheus" "k8s.io/client-go/rest" "k8s.io/client-go/transport" "github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/setting" + "github.com/grafana/grafana/pkg/storage/unified" + "github.com/grafana/grafana/pkg/storage/unified/resource" provisioning "github.com/grafana/grafana/apps/provisioning/pkg/apis/provisioning/v0alpha1" authrt "github.com/grafana/grafana/apps/provisioning/pkg/auth" @@ -20,6 +23,7 @@ import ( "github.com/grafana/grafana/apps/provisioning/pkg/repository" "github.com/grafana/grafana/apps/provisioning/pkg/repository/github" "github.com/grafana/grafana/apps/provisioning/pkg/repository/local" + "github.com/grafana/grafana/pkg/registry/apis/provisioning/resources" secretdecrypt "github.com/grafana/grafana/pkg/registry/apis/secret/decrypt" ) @@ -28,6 +32,7 @@ type provisioningControllerConfig struct { provisioningClient *client.Clientset resyncInterval time.Duration repoFactory repository.Factory + unified resources.ResourceStore } // expects: @@ -40,6 +45,11 @@ type provisioningControllerConfig struct { // grpc_server_use_tls = // grpc_server_tls_ca_file = // grpc_server_tls_skip_verify = +// [unified_storage] +// grpc_address = +// grpc_index_address = +// allow_insecure = +// audiences = // [operator] // provisioning_server_url = // tls_insecure = @@ -115,9 +125,23 @@ func setupFromConfig(cfg *setting.Cfg) (controllerCfg *provisioningControllerCon return nil, fmt.Errorf("failed to setup repository getter: %w", err) } + // HACK: This logic directly connects to unified storage. We are doing this for now as there is no global + // search endpoint. But controllers, in general, should not connect directly to unified storage and instead + // go through the api server. Once there is a global search endpoint, we will switch to that here as well. + resourceClientCfg := resource.RemoteResourceClientConfig{ + Token: token, + TokenExchangeURL: tokenExchangeURL, + Namespace: gRPCAuth.Key("token_namespace").String(), + } + unified, err := setupUnifiedStorageClient(cfg, tracer, resourceClientCfg) + if err != nil { + return nil, fmt.Errorf("failed to setup unified storage: %w", err) + } + return &provisioningControllerConfig{ provisioningClient: provisioningClient, repoFactory: repoFactory, + unified: unified, resyncInterval: operatorSec.Key("resync_interval").MustDuration(60 * time.Second), }, nil } @@ -238,3 +262,41 @@ func setupDecrypter(cfg *setting.Cfg, tracer tracing.Tracer, tokenExchangeClient return repository.ProvideDecrypter(decryptSvc), nil } + +// HACK: This logic directly connects to unified storage. We are doing this for now as there is no global +// search endpoint. But controllers, in general, should not connect directly to unified storage and instead +// go through the api server. Once there is a global search endpoint, we will switch to that here as well. +func setupUnifiedStorageClient(cfg *setting.Cfg, tracer tracing.Tracer, resourceClientCfg resource.RemoteResourceClientConfig) (resources.ResourceStore, error) { + unifiedStorageSec := cfg.SectionWithEnvOverrides("unified_storage") + // Connect to Server + address := unifiedStorageSec.Key("grpc_address").String() + if address == "" { + return nil, fmt.Errorf("grpc_address is required in [unified_storage] section") + } + registry := prometheus.NewPedanticRegistry() + conn, err := unified.GrpcConn(address, registry) + if err != nil { + return nil, fmt.Errorf("create unified storage gRPC connection: %w", err) + } + + // Connect to Index + indexConn := conn + indexAddress := unifiedStorageSec.Key("grpc_index_address").String() + if indexAddress == "" { + indexConn, err = unified.GrpcConn(indexAddress, registry) + if err != nil { + return nil, fmt.Errorf("create unified storage index gRPC connection: %w", err) + } + } + + // Create client + resourceClientCfg.AllowInsecure = unifiedStorageSec.Key("allow_insecure").MustBool(false) + resourceClientCfg.Audiences = unifiedStorageSec.Key("audiences").Strings("|") + + client, err := resource.NewRemoteResourceClient(tracer, conn, indexConn, resourceClientCfg) + if err != nil { + return nil, fmt.Errorf("create unified storage client: %w", err) + } + + return client, nil +} diff --git a/pkg/operators/provisioning/jobs_operator.go b/pkg/operators/provisioning/jobs_operator.go index 84265091d78..61a42852aad 100644 --- a/pkg/operators/provisioning/jobs_operator.go +++ b/pkg/operators/provisioning/jobs_operator.go @@ -13,8 +13,11 @@ import ( "github.com/urfave/cli/v2" "k8s.io/client-go/tools/cache" + "github.com/grafana/grafana/pkg/apimachinery/identity" + "github.com/grafana/grafana/pkg/apimachinery/utils" "github.com/grafana/grafana/pkg/services/apiserver/standalone" "github.com/grafana/grafana/pkg/setting" + "github.com/grafana/grafana/pkg/storage/unified/resourcepb" "github.com/grafana/grafana/apps/provisioning/pkg/controller" informer "github.com/grafana/grafana/apps/provisioning/pkg/generated/informers/externalversions" @@ -42,6 +45,11 @@ func RunJobController(opts standalone.BuildInfo, c *cli.Context, cfg *setting.Cf cancel() }() + // Use unified storage client for testing purposes. + // TODO: remove this once the processing logic is in place + // https://github.com/grafana/git-ui-sync-project/issues/467 + go temporaryPeriodicCountManagedObjects(ctx, logger, controllerCfg) + // Jobs informer and controller (resync ~60s like in register.go) jobInformerFactory := informer.NewSharedInformerFactoryWithOptions( controllerCfg.provisioningClient, @@ -116,3 +124,46 @@ func getJobsControllerConfig(cfg *setting.Cfg) (*jobsControllerConfig, error) { historyExpiration: cfg.SectionWithEnvOverrides("operator").Key("history_expiration").MustDuration(0), }, nil } + +// Use unified storage client for testing purposes. +// TODO: remove this once the processing logic is in place +// https://github.com/grafana/git-ui-sync-project/issues/467 +func temporaryPeriodicCountManagedObjects(ctx context.Context, logger logging.Logger, controllerCfg *jobsControllerConfig) { + tick := time.NewTicker(controllerCfg.resyncInterval) + logger.Info("starting periodic managed resource lister", "interval", controllerCfg.resyncInterval.String()) + fetchAndLog := func(ctx context.Context) { + ctx, _, err := identity.WithProvisioningIdentity(ctx, "*") // "*" grants us access to all namespaces. + if err != nil { + logger.Error("failed to set identity", "error", err) + return + } + + resp, err := controllerCfg.unified.CountManagedObjects(ctx, &resourcepb.CountManagedObjectsRequest{ + Kind: string(utils.ManagerKindRepo), + }) + if err != nil { + logger.Error("failed to list managed objects", "error", err) + } else { + if len(resp.Items) == 0 { + logger.Info("no managed objects found") + return + } + + for _, obj := range resp.Items { + logger.Info("manage object counts", "item", obj) + } + } + } + + fetchAndLog(ctx) // Initial fetch + for { + select { + case <-ctx.Done(): + tick.Stop() + return + case <-tick.C: + // Periodic fetch + fetchAndLog(ctx) + } + } +}