package runtime import ( "cmp" "context" "errors" "fmt" "log/slog" "strings" "sync" "time" "github.com/go-logr/logr" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" kruntime "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/dynamic" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" "sigs.k8s.io/controller-runtime/pkg/config" ctrllog "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" "sigs.k8s.io/multicluster-runtime/providers/multi" "github.com/johnandersen777/socialweb-computer-kcp/api/kcpgvr" "github.com/kcp-dev/multicluster-provider/apiexport" apisv1alpha1 "github.com/kcp-dev/sdk/apis/apis/v1alpha1" ) const ( FieldManagerPrefix = "socialweb.computer/" DefaultExportWorkspace = "root:socialweb-provider" clusterSeparator = "#" pathAnnotation = "kcp.io/path" ) type Config struct { Rest *rest.Config ExportWorkspace string AddToScheme []func(*kruntime.Scheme) error Workspaces func(workspace string) bool Log *slog.Logger } type Manager struct { base *rest.Config exportWorkspace string workspaces func(workspace string) bool scheme *kruntime.Scheme log *slog.Logger logr logr.Logger providers *multi.Provider mgr mcmanager.Manager mu sync.Mutex exports map[string]bool ctx context.Context exportsWG sync.WaitGroup sourcesWG sync.WaitGroup paths sync.Map } var setLogger sync.Once func NewManager(cfg Config) (*Manager, error) { if cfg.Rest == nil { return nil, errors.New("runtime: a kcp rest config is required") } log := cfg.Log if log == nil { log = slog.New(slog.DiscardHandler) } log = log.With("component", "controller-manager") exportWorkspace := strings.TrimSpace(cfg.ExportWorkspace) exportWorkspace = cmp.Or(exportWorkspace, DefaultExportWorkspace) base := rest.CopyConfig(cfg.Rest) host, _, _ := strings.Cut(strings.TrimSuffix(base.Host, "/"), "/clusters/") base.Host = host if base.QPS == 0 { base.QPS, base.Burst = 50, 100 } scheme := kruntime.NewScheme() adders := append([]func(*kruntime.Scheme) error{clientgoscheme.AddToScheme, apisv1alpha1.AddToScheme}, cfg.AddToScheme...) for _, add := range adders { if add == nil { continue } if err := add(scheme); err != nil { return nil, fmt.Errorf("runtime: building the scheme: %w", err) } } logger := logr.FromSlogHandler(newRuntimeLogHandler(log.Handler())) setLogger.Do(func() { ctrllog.SetLogger(logger) }) providers := multi.New(multi.Options{Separator: clusterSeparator}) mgr, err := mcmanager.New(clusterConfig(base, exportWorkspace), providers, manager.Options{ Scheme: scheme, Logger: logger, Metrics: metricsserver.Options{BindAddress: "0"}, Controller: config.Controller{SkipNameValidation: new(true)}, }) if err != nil { return nil, fmt.Errorf("runtime: the multicluster manager: %w", err) } return &Manager{base: base, exportWorkspace: exportWorkspace, workspaces: cfg.Workspaces, scheme: scheme, log: log, logr: logger, providers: providers, mgr: mgr, exports: map[string]bool{}}, nil } func (m *Manager) AddToScheme(adders ...func(*kruntime.Scheme) error) error { m.mu.Lock() defer m.mu.Unlock() for _, add := range adders { if add == nil { continue } if err := add(m.scheme); err != nil { return fmt.Errorf("runtime: extending the scheme: %w", err) } } return nil } func (m *Manager) Start(ctx context.Context) error { exportCtx, cancel := context.WithCancel(ctx) defer cancel() m.log.Info("the controller manager starts", "exportWorkspace", m.exportWorkspace) m.mu.Lock() m.ctx = exportCtx for name := range m.exports { m.exportsWG.Go(func() { m.serveExport(exportCtx, name) }) } m.mu.Unlock() err := m.mgr.Start(ctx) m.mu.Lock() m.ctx = nil m.mu.Unlock() cancel() m.exportsWG.Wait() m.sourcesWG.Wait() return err } func (m *Manager) stopping(ctx context.Context) bool { if ctx.Err() != nil { return true } m.mu.Lock() run := m.ctx m.mu.Unlock() return run != nil && run.Err() != nil } func (m *Manager) addExport(name string) { m.mu.Lock() defer m.mu.Unlock() if m.exports[name] { return } m.exports[name] = true if ctx := m.ctx; ctx != nil { m.exportsWG.Go(func() { m.serveExport(ctx, name) }) } } func (m *Manager) serveExport(ctx context.Context, name string) { logger := m.logr.WithValues("export", name) retry := newBackoff(ExportRetryFirst, DefaultBackoffMax) for failures := 0; ctx.Err() == nil; failures++ { p, err := apiexport.New(clusterConfig(m.base, m.exportWorkspace), name, apiexport.Options{Scheme: m.scheme, Log: &logger}) if err == nil { err = m.providers.AddProvider(name, p) } if err == nil { m.log.Info("the controller manager serves an APIExport", "export", name, "endpointSlice", m.exportWorkspace+":"+name) return } retryIn := retry.delay(failures) m.log.Warn("the APIExport provider does not start yet; it retries with backoff", "export", name, "workspace", m.exportWorkspace, "failures", failures+1, "retryIn", retryIn.String(), "err", err.Error()) timer := time.NewTimer(retryIn) select { case <-ctx.Done(): timer.Stop() return case <-timer.C: } } } func (m *Manager) path(ctx context.Context, cluster string) (string, error) { if v, ok := m.paths.Load(cluster); ok { return v.(string), nil } dyn, err := dynamic.NewForConfig(clusterConfig(m.base, cluster)) if err != nil { return "", fmt.Errorf("runtime: the client of logical cluster %s: %w", cluster, err) } lc, err := dyn.Resource(kcpgvr.LogicalClusters).Get(ctx, "cluster", metav1.GetOptions{}) if err != nil { return "", fmt.Errorf("runtime: the workspace path of logical cluster %s: %w", cluster, err) } path := strings.TrimSpace(lc.GetAnnotations()[pathAnnotation]) if path == "" { return cluster, nil } m.paths.Store(cluster, path) return path, nil } func clusterConfig(base *rest.Config, cluster string) *rest.Config { out := rest.CopyConfig(base) out.Host = strings.TrimSuffix(base.Host, "/") + "/clusters/" + cluster return out } func splitCluster(name string) (string, string) { export, cluster, ok := strings.Cut(name, clusterSeparator) if !ok { return "", name } return export, cluster }