Implement a basic operator to reconcile the folder hierarchy from Unistore to Zanzana (#109705)

2025-08-20 17:14:06 +00:00
parent f7d39204cd
commit c8b0fd685b
20 changed files with 3317 additions and 9 deletions
@@ -0,0 +1,16 @@
+FROM golang:1.24-alpine AS builder
+
+WORKDIR /build
+COPY go.mod go.sum ./
+COPY vendor* ./vendor
+RUN test -f vendor/modules.txt || go mod download
+
+COPY cmd cmd
+COPY pkg pkg
+
+RUN go build -o "target/operator" cmd/operator/*.go
+
+FROM alpine AS runtime
+COPY --from=builder /build/target/operator /usr/bin/operator
+
+ENTRYPOINT ["/usr/bin/operator"]
@@ -0,0 +1,31 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+
+	utilnet "k8s.io/apimachinery/pkg/util/net"
+
+	"github.com/grafana/authlib/authn"
+)
+
+type authRoundTripper struct {
+	tokenExchangeClient *authn.TokenExchangeClient
+	transport           http.RoundTripper
+}
+
+func (t *authRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	tokenResponse, err := t.tokenExchangeClient.Exchange(req.Context(), authn.TokenExchangeRequest{
+		Audiences: []string{"folder.grafana.app"},
+		Namespace: "*",
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to exchange token: %w", err)
+	}
+
+	// clone the request as RTs are not expected to mutate the passed request
+	req = utilnet.CloneRequest(req)
+
+	req.Header.Set("X-Access-Token", "Bearer "+tokenResponse.Token)
+	return t.transport.RoundTrip(req)
+}
@@ -0,0 +1,120 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/grafana/grafana-app-sdk/plugin/kubeconfig"
+	"github.com/grafana/grafana-app-sdk/simple"
+)
+
+const (
+	ConnTypeGRPC = "grpc"
+	ConnTypeHTTP = "http"
+)
+
+type Config struct {
+	OTelConfig       simple.OpenTelemetryConfig
+	WebhookServer    WebhookServerConfig
+	KubeConfig       *kubeconfig.NamespacedConfig
+	ZanzanaClient    ZanzanaClientConfig
+	FolderReconciler FolderReconcilerConfig
+}
+
+type WebhookServerConfig struct {
+	Port        int
+	TLSCertPath string
+	TLSKeyPath  string
+}
+
+type ZanzanaClientConfig struct {
+	Addr string
+}
+
+type FolderReconcilerConfig struct {
+	Namespace string
+}
+
+func LoadConfigFromEnv() (*Config, error) {
+	cfg := Config{}
+	cfg.OTelConfig.ServiceName = os.Getenv("OTEL_SERVICE_NAME")
+	switch strings.ToLower(os.Getenv("OTEL_CONN_TYPE")) {
+	case ConnTypeGRPC:
+		cfg.OTelConfig.ConnType = ConnTypeGRPC
+	case ConnTypeHTTP:
+		cfg.OTelConfig.ConnType = ConnTypeHTTP
+	case "":
+		// Default
+		cfg.OTelConfig.ConnType = ConnTypeHTTP
+	default:
+		return nil, fmt.Errorf("unknown OTEL_CONN_TYPE '%s'", os.Getenv("OTEL_CONN_TYPE"))
+	}
+	cfg.OTelConfig.Host = os.Getenv("OTEL_HOST")
+	portStr := os.Getenv("OTEL_PORT")
+	if portStr == "" {
+		if cfg.OTelConfig.ConnType == ConnTypeGRPC {
+			// Default OTel GRPC port
+			cfg.OTelConfig.Port = 4317
+		} else {
+			// Default OTel HTTP port
+			cfg.OTelConfig.Port = 4318
+		}
+	} else {
+		var err error
+		cfg.OTelConfig.Port, err = strconv.Atoi(portStr)
+		if err != nil {
+			return nil, fmt.Errorf("invalid OTEL_PORT '%s': %w", portStr, err)
+		}
+	}
+
+	whPortStr := os.Getenv("WEBHOOK_PORT")
+	if whPortStr == "" {
+		cfg.WebhookServer.Port = 8443
+	} else {
+		var err error
+		cfg.WebhookServer.Port, err = strconv.Atoi(whPortStr)
+		if err != nil {
+			return nil, fmt.Errorf("invalid WEBHOOK_PORT '%s': %w", whPortStr, err)
+		}
+	}
+
+	cfg.WebhookServer.TLSCertPath = os.Getenv("WEBHOOK_CERT_PATH")
+	cfg.WebhookServer.TLSKeyPath = os.Getenv("WEBHOOK_KEY_PATH")
+
+	// Load the kube config
+	kubeConfigFile := os.Getenv("KUBE_CONFIG_FILE")
+	if kubeConfigFile != "" {
+		kubeConfig, err := LoadKubeConfigFromFile(kubeConfigFile)
+		if err != nil {
+			return nil, fmt.Errorf("unable to load kubernetes configuration from file '%s': %w", kubeConfigFile, err)
+		}
+		cfg.KubeConfig = kubeConfig
+	} else if folderAppURL := os.Getenv("FOLDER_APP_URL"); folderAppURL != "" {
+		exchangeUrl := os.Getenv("AUTH_TOKEN_EXCHANGE_URL")
+		authToken := os.Getenv("AUTH_TOKEN")
+		namespace := os.Getenv("FOLDER_APP_NAMESPACE")
+		if exchangeUrl == "" || authToken == "" {
+			return nil, fmt.Errorf("AUTH_TOKEN_EXCHANGE_URL and AUTH_TOKEN must be set when FOLDER_APP_URL is set")
+		}
+
+		kubeConfig, err := LoadKubeConfigFromFolderAppURL(folderAppURL, exchangeUrl, authToken, namespace)
+		if err != nil {
+			return nil, fmt.Errorf("unable to load kubernetes configuration from folder app URL '%s': %w", folderAppURL, err)
+		}
+		cfg.KubeConfig = kubeConfig
+	} else {
+		kubeConfig, err := LoadInClusterConfig()
+		if err != nil {
+			return nil, fmt.Errorf("unable to load in-cluster kubernetes configuration: %w", err)
+		}
+		cfg.KubeConfig = kubeConfig
+	}
+
+	cfg.ZanzanaClient.Addr = os.Getenv("ZANZANA_ADDR")
+
+	cfg.FolderReconciler.Namespace = os.Getenv("FOLDER_RECONCILER_NAMESPACE")
+
+	return &cfg, nil
+}
@@ -0,0 +1,85 @@
+package main
+
+import (
+	"fmt"
+	"net/http"
+
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+	"k8s.io/client-go/transport"
+
+	"github.com/grafana/authlib/authn"
+	"github.com/grafana/grafana-app-sdk/plugin/kubeconfig"
+)
+
+// LoadInClusterConfig loads a kubernetes in-cluster config.
+// Since the in-cluster config doesn't have a namespace, it defaults to "default"
+func LoadInClusterConfig() (*kubeconfig.NamespacedConfig, error) {
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, err
+	}
+	cfg.APIPath = "/apis"
+	return &kubeconfig.NamespacedConfig{
+		RestConfig: *cfg,
+		Namespace:  "default",
+	}, nil
+}
+
+// LoadKubeConfigFromEnv loads a NamespacedConfig from the value of an environment variable
+func LoadKubeConfigFromFolderAppURL(folderAppURL, exchangeUrl, authToken, namespace string) (*kubeconfig.NamespacedConfig, error) {
+	tokenExchangeClient, err := authn.NewTokenExchangeClient(authn.TokenExchangeConfig{
+		TokenExchangeURL: exchangeUrl,
+		Token:            authToken,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create token exchange client: %w", err)
+	}
+
+	return &kubeconfig.NamespacedConfig{
+		RestConfig: rest.Config{
+			APIPath: "/apis",
+			Host:    folderAppURL,
+			WrapTransport: transport.WrapperFunc(func(rt http.RoundTripper) http.RoundTripper {
+				return &authRoundTripper{
+					tokenExchangeClient: tokenExchangeClient,
+					transport:           rt,
+				}
+			}),
+			TLSClientConfig: rest.TLSClientConfig{
+				Insecure: true,
+			},
+		},
+		Namespace: namespace,
+	}, nil
+}
+
+// LoadKubeConfigFromFile loads a NamespacedConfig from a file on-disk (such as a mounted secret)
+func LoadKubeConfigFromFile(configPath string) (*kubeconfig.NamespacedConfig, error) {
+	// Load the kubeconfig file
+	config, err := clientcmd.LoadFromFile(configPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load kubeconfig from %s: %w", configPath, err)
+	}
+
+	// Build the REST config from the kubeconfig
+	restConfig, err := clientcmd.NewDefaultClientConfig(*config, &clientcmd.ConfigOverrides{}).ClientConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to create REST config: %w", err)
+	}
+
+	// Get the namespace from the current context, default to "default" if not set
+	namespace := "default"
+	if config.CurrentContext != "" {
+		if context, exists := config.Contexts[config.CurrentContext]; exists && context.Namespace != "" {
+			namespace = context.Namespace
+		}
+	}
+
+	restConfig.APIPath = "/apis"
+
+	return &kubeconfig.NamespacedConfig{
+		RestConfig: *restConfig,
+		Namespace:  namespace,
+	}, nil
+}
@@ -0,0 +1,82 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"os/signal"
+
+	"github.com/grafana/grafana-app-sdk/k8s"
+	"github.com/grafana/grafana-app-sdk/logging"
+	"github.com/grafana/grafana-app-sdk/operator"
+	"github.com/grafana/grafana-app-sdk/simple"
+	"github.com/grafana/grafana/apps/iam/pkg/app"
+)
+
+func main() {
+	// Configure the default logger to use slog
+	logging.DefaultLogger = logging.NewSLogLogger(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
+		Level: slog.LevelDebug,
+	}))
+
+	//Load the config from the environment
+	cfg, err := LoadConfigFromEnv()
+	if err != nil {
+		logging.DefaultLogger.With("error", err).Error("Unable to load config from environment")
+		panic(err)
+	}
+
+	// Set up tracing
+	if cfg.OTelConfig.Host != "" {
+		err = simple.SetTraceProvider(simple.OpenTelemetryConfig{
+			Host:        cfg.OTelConfig.Host,
+			Port:        cfg.OTelConfig.Port,
+			ConnType:    cfg.OTelConfig.ConnType,
+			ServiceName: cfg.OTelConfig.ServiceName,
+		})
+		if err != nil {
+			logging.DefaultLogger.With("error", err).Error("Unable to set trace provider")
+			panic(err)
+		}
+	}
+
+	// Create the operator config and the runner
+	operatorConfig := operator.RunnerConfig{
+		KubeConfig: cfg.KubeConfig.RestConfig,
+		WebhookConfig: operator.RunnerWebhookConfig{
+			Port: cfg.WebhookServer.Port,
+			TLSConfig: k8s.TLSConfig{
+				CertPath: cfg.WebhookServer.TLSCertPath,
+				KeyPath:  cfg.WebhookServer.TLSKeyPath,
+			},
+		},
+		MetricsConfig: operator.RunnerMetricsConfig{
+			Enabled: true,
+		},
+	}
+
+	runner, err := operator.NewRunner(operatorConfig)
+	if err != nil {
+		logging.DefaultLogger.With("error", err).Error("Unable to create operator runner")
+		panic(err)
+	}
+
+	// Context and cancel for the operator's Run method
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, os.Kill)
+	defer cancel()
+
+	// Create app config from operator config
+	appCfg := app.AppConfig{
+		ZanzanaAddr:               cfg.ZanzanaClient.Addr,
+		FolderReconcilerNamespace: cfg.FolderReconciler.Namespace,
+	}
+
+	// Run
+	logging.DefaultLogger.Info("Starting operator")
+	err = runner.Run(ctx, app.Provider(appCfg))
+	if err != nil {
+		logging.DefaultLogger.With("error", err).Error("Operator exited with error")
+		panic(err)
+	}
+	logging.DefaultLogger.Info("Normal operator exit")
+}