Provisioning: Build unified storage client for operators (#110671)

* Build unified storage client in config
* Testing ticker to count managed objects

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
+}

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)
+		}
+	}
+}