package main import ( "context" "flag" compute "cloud.google.com/go/compute/apiv1" container "cloud.google.com/go/container/apiv1" "github.com/aws/aws-sdk-go-v2/config" pcav1beta1 "github.com/cert-manager/aws-privateca-issuer/pkg/api/v1beta1" ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1" ec2v1beta1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1beta1" "github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1" eksv1alpha1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1alpha1" eksv1beta1 "github.com/crossplane-contrib/provider-aws/apis/eks/v1beta1" awsiamv1alpha1 "github.com/crossplane-contrib/provider-aws/apis/iam/v1alpha1" iamv1beta1 "github.com/crossplane-contrib/provider-aws/apis/iam/v1beta1" route53v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/route53/v1alpha1" xpawsv1beta1 "github.com/crossplane-contrib/provider-aws/apis/v1beta1" computev1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/compute/v1alpha1" computev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/compute/v1beta1" containerv1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1" containerv1beta2 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta2" gcpiamv1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/iam/v1alpha1" gcpstoragev1alpha1 "github.com/crossplane-contrib/provider-gcp/apis/storage/v1alpha1" gcpv1beta1 "github.com/crossplane-contrib/provider-gcp/apis/v1beta1" gcpvpcpeeringv1beta1 "github.com/crossplane/provider-gcp/apis/vpcpeering/v1beta1" certmanagerv1 "github.com/jetstack/cert-manager/pkg/apis/certmanager/v1" "github.com/pingcap/log" operatorv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1" tiflowv1alpha1 "github.com/pingcap/tiflow-operator/api/v1alpha1" cloudv1 "github.com/tidbcloud/infra-api/api/v1" "github.com/tidbcloud/infra-provider/pkg/cloud/aws" "github.com/tidbcloud/infra-provider/pkg/cloud/gcp" "github.com/tidbcloud/infra-provider/pkg/consts" asts "github.com/tidbcloud/infra-provider/pkg/pingcap/apis/v1" infrascheme "github.com/tidbcloud/infra-provider/pkg/scheme" awsupboundeksv1beta1 "github.com/upbound/provider-aws/apis/eks/v1beta1" awsupboundiamv1beta1 "github.com/upbound/provider-aws/apis/iam/v1beta1" awsupboundv1beta1 "github.com/upbound/provider-aws/apis/v1beta1" azureauthv1beta1 "github.com/upbound/provider-azure/apis/authorization/v1beta1" azurev1beta1 "github.com/upbound/provider-azure/apis/azure/v1beta1" azurecontainerv1beta1 "github.com/upbound/provider-azure/apis/containerservice/v1beta1" azuremanagedidentityv1beta1 "github.com/upbound/provider-azure/apis/managedidentity/v1beta1" azurenetworkv1beta1 "github.com/upbound/provider-azure/apis/network/v1beta1" upboundcontainerv1beta1 "github.com/upbound/provider-gcp/apis/container/v1beta1" "go.uber.org/zap" "go.uber.org/zap/zapcore" apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/serializer" utilruntime "k8s.io/apimachinery/pkg/util/runtime" clientgoscheme "k8s.io/client-go/kubernetes/scheme" "k8s.io/client-go/rest" _ "k8s.io/client-go/tools/clientcmd" _ "k8s.io/client-go/util/homedir" "k8s.io/klog/v2" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/cache" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/healthz" sigszap "sigs.k8s.io/controller-runtime/pkg/log/zap" metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" "sigs.k8s.io/controller-runtime/pkg/predicate" "github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/acscli" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/clustercli" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/models" o11yapi "github.com/tidbcloud/infra-provider/regional-server/pkg/common/o11ycli/api" "github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils" rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/appbox" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/approute" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/asyncappbox" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/internet" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/o11yroute" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservice" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/privatelinkservicemigration" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/upgradeoperation" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpc" "github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpcpeering" "github.com/tidbcloud/infra-provider/regional-server/pkg/gc" "github.com/tidbcloud/infra-provider/regional-server/pkg/gc/nodepoolgc" "github.com/tidbcloud/infra-provider/regional-server/pkg/metrics" "github.com/tidbcloud/infra-provider/regional-server/pkg/proxy" "github.com/tidbcloud/infra-provider/regional-server/pkg/router" appboxrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/appbox" k8sresourcerouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/k8sresource" "github.com/tidbcloud/infra-provider/regional-server/pkg/router/k8supgrade" migrationxpawsrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/migration/crossplane/aws" migrationxpgcprouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/migration/crossplane/gcp" migrationrsrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/migration/dedicatedregional" networkrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/network" plsrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/privatelinkservice" vpcpeeringrouter "github.com/tidbcloud/infra-provider/regional-server/pkg/router/vpcpeering" appboxsvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/appbox" k8sresourcesvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/k8sresource" k8supgradesvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/k8supgrade" migrationxpawssvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/migration/crossplane/aws" migrationxpgcpsvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/migration/crossplane/gcp" migrationrssvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/migration/dedicatedregional" networksvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/network" vpcpeeringsvc "github.com/tidbcloud/infra-provider/regional-server/pkg/service/vpcpeering" agentv2 "github.com/tidbcloud/observability/agent-control-server/api/v2" agentcontrolapi "github.com/tidbcloud/observability/agent-control-server/pkg/agentcontrol/api" agentcontrolimpl "github.com/tidbcloud/observability/agent-control-server/pkg/agentcontrol/implement" ) var ( clientMode bool logLevel string port int rootConfigPath string addonTmplConfigPath string infraTmplConfigPath string dedicatedTmplConfigPath string dpTmplConfigPath string probeAddr string metricsAddr string enableLeaderElection bool disableControllers bool ) var ( Scheme = runtime.NewScheme() ) // +kubebuilder:rbac:groups="dedicated.regionalserver.tidbcloud.com",resources="*",verbs="*" // +kubebuilder:rbac:groups="",resources="*",verbs="*" // +kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters,verbs=get;list;watch func init() { utilruntime.Must(clientgoscheme.AddToScheme(Scheme)) utilruntime.Must(cloudv1.AddToScheme(Scheme)) utilruntime.Must(v1alpha1.AddToScheme(Scheme)) loadAWSScheme() loadAWSUpboundScheme() loadGCPScheme() loadAzureScheme() loadDataplaneScheme() // we need to replace the infrascheme.Scheme and infrascheme.Codecs with our Scheme and CodecFactory infrascheme.Scheme = Scheme infrascheme.Codecs = serializer.NewCodecFactory(Scheme) //+kubebuilder:scaffold:scheme } func init_logger() { utils.SetDefaultSlog(logLevel) // init logger, use pingcap/log now, this logger will be used in apiserver logger, props, err := log.InitLogger(&log.Config{Level: logLevel}) if err != nil { panic(err) } log.ReplaceGlobals(logger, props) // init controller-runtime logger level, err := zapcore.ParseLevel(logLevel) if err != nil { level = zapcore.InfoLevel } opts := sigszap.Options{ Development: false, Level: level, TimeEncoder: zapcore.RFC3339TimeEncoder, } zlog := sigszap.New(sigszap.UseFlagOptions(&opts)) ctrl.SetLogger(zlog) // This helps to standardize the log layout of client-go. klog.SetLogger(zlog) } func main() { flag.IntVar(&port, "port", 2333, "api service listening port") flag.BoolVar(&clientMode, "client-mode", true, "client mode (true: in-cluster, false: out-of-cluster)") flag.StringVar(&metricsAddr, "metrics-bind-address", ":8082", "The address the metric endpoint binds to.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") flag.StringVar(&logLevel, "log-level", "info", "log level. default: info") flag.StringVar(&rootConfigPath, "root-config", "", "The path to the root config.") flag.BoolVar(&enableLeaderElection, "leader-elect", false, "Enable leader election for controller manager.") flag.StringVar(&addonTmplConfigPath, "addon-template-config", "", "The path to the k8s addons templates.") flag.StringVar(&infraTmplConfigPath, "infra-template-config", "", "The path to the k8s infra components templates.") flag.StringVar(&dedicatedTmplConfigPath, "dedicated-template-config", "", "The path to the dedicated infra components templates.") flag.StringVar(&dpTmplConfigPath, "dp-template-config", "", "The path to the data platform infra components templates.") flag.BoolVar(&disableControllers, "disable-controllers", false, "Disable all controllers, only run the api server.") flag.Parse() ctx := utils.SetupSignalHandler() init_logger() //nolint defer log.Sync() log.Info("Version", zap.Any("version", utils.GetVersionInfo())) // build kube config var kubeConfig *rest.Config var err error if clientMode { // creates the in-cluster config kubeConfig, err = rest.InClusterConfig() } else { // creates the out-of-cluster config kubeConfig, err = ctrl.GetConfig() // kubeConfig, err = clientcmd.BuildConfigFromFlags("", kubeConfigPath) } if err != nil { log.Error("init kubeconfig failed", zap.Error(err)) } // avoid client-side throttling kubeConfig.Burst = 50 kubeConfig.QPS = 20 config, err := rsconfig.LoadRootConfig(rootConfigPath) if err != nil { log.Fatal("unable to load root config", zap.Error(err)) } rsconfig.SetGlobalConfig(config) config.SelfSignedCAConfig.InitTiDBServerSelfSignedCAData() tmplConfig, err := rsconfig.LoadTemplatesConfig(addonTmplConfigPath, infraTmplConfigPath, dedicatedTmplConfigPath, dpTmplConfigPath) if err != nil { log.Fatal("unable to load template config", zap.Error(err)) } log.Info("root config", zap.Any("config", config)) cs, err := buildCacheSelector(config) if err != nil { log.Fatal("unable to build cache selector", zap.Error(err)) } mgr, err := ctrl.NewManager(kubeConfig, ctrl.Options{ Scheme: Scheme, HealthProbeBindAddress: probeAddr, Metrics: metricsserver.Options{BindAddress: metricsAddr}, LeaderElection: enableLeaderElection, LeaderElectionID: "fc66da15.dedicated.regionalserver.tidbcloud.com", // Only load objects to local cache if it matches the label selector. Cache: cache.Options{ByObject: cs}, }) if err != nil { log.Fatal("unable to start manager", zap.Error(err)) } rsmanager := controllers.NewManager(mgr.GetClient(), Scheme) go func() { err := <-rsmanager.Errors() log.Fatal("init mutli-k8s client manager error", zap.Error(err)) }() agentControlAPI := agentcontrolimpl.NewAgentControlImplWithConfig( mgr.GetClient(), config.Env, o11yapi.GetO11yVendor(config.Provider), config.RegionConfig.Name, agentcontrolimpl.AgentControlConfig{ Namespace: config.O11yAgentNamespace, BizType: config.O11yBizType, }) // start api server startAPIServer(ctx, mgr, rsmanager, config, tmplConfig, agentControlAPI) if !config.GCConfig.DryRun { log.Info("gc is enabled") go func() { // TODO(cwen0): add leader election for gc err := gc.NewGC(config.GCConfig, mgr.GetClient(), ctrl.Log.WithName("gc")).Run(ctx) if err != nil { log.Error("gc run failed", zap.Error(err)) } }() } if config.GCConfig.NodePoolGC.Enabled { nodePoolGCConfig := config.GCConfig.NodePoolGC nodePoolGCConfig.DryRun = config.GCConfig.DryRun runner, err := nodepoolgc.New( mgr.GetClient(), ctrl.Log.WithName("gc").WithName("nodepool"), nodePoolGCConfig, config.Provider, rsmanager, ) if err != nil { log.Fatal("unable to create NodePoolGC runnable", zap.Error(err)) } if err := mgr.Add(runner); err != nil { log.Fatal("unable to add NodePoolGC runnable", zap.Error(err)) } } if !disableControllers { log.Info("starting controllers") startControllers(ctx, mgr, config, tmplConfig, rsmanager, agentControlAPI) } if err = mgr.Start(ctx); err != nil { log.Fatal("unable to start manager", zap.Error(err)) } } func startAPIServer( ctx context.Context, mgr ctrl.Manager, rsmanager *controllers.Manager, config rsconfig.RootConfig, tmplConfig rsconfig.TemplatesConfig, agentControlAPI agentcontrolapi.AgentControlAPI, ) { // FIXME: casual naming, just to make mounting service and handler work vpcpeeringsvc, err := vpcpeeringsvc.NewService( mgr.GetClient(), mgr.GetAPIReader(), vpcpeeringsvc.WithRegion(config.RegionConfig.Name), vpcpeeringsvc.WithProvider(config.Provider), vpcpeeringsvc.WithConfig(config), ) if err != nil { log.Fatal("new vpcpeering service failed", zap.Error(err)) } rp, err := proxy.NewReverseProxy(config.ClusterControlServiceEndpoint) if err != nil { log.Fatal("new reverse proxy for cluster service failed", zap.Error(err)) } rp = rp.WithPathPerfixRewrite("/api/dedicated/v1", "/api/v1") chp, err := proxy.NewClusterServiceHijackProxy( rp, mgr.GetClient(), &config, agentControlAPI) if err != nil { log.Fatal("new hijack proxy for cluster service failed", zap.Error(err)) } chp.WithLogger(ctrl.Log.WithName("router").WithName("gin").WithName("cluster-service-hijack-proxy")) clusterAgentProxy := proxy.NewClusterAgentProxy(mgr.GetClient()) var gcpResourceManager gcp.ResourceManagerClient migrrssvc, err := migrationrssvc.NewService( mgr.GetClient(), mgr.GetAPIReader(), clustercli.NewClusterCli(config.ClusterControlServiceEndpoint), config, tmplConfig, rsmanager) if err != nil { log.Fatal("new migration rs service failed", zap.Error(err)) } // k8s upgrade service need to create index for kubernetes version. if err = mgr.GetFieldIndexer().IndexField(ctx, &v1alpha1.APPBox{}, controllers.KubernetesVersionField, func(o client.Object) []string { c := o.(*v1alpha1.APPBox) return []string{c.Spec.KubernetesConfig.Version} }); err != nil { log.Fatal("failed to create index for kubernetes version", zap.Error(err)) } upgradeSvc := k8supgradesvc.NewService(mgr.GetClient(), config) appboxSvc := appboxsvc.NewAPPBoxService(mgr.GetClient(), config, tmplConfig) networkSvc := networksvc.NewNetworkService(mgr.GetClient(), config) if config.Provider == consts.ProviderTypeGCP || config.Provider == consts.ProviderTypeGKE { { cli, err := compute.NewSubnetworksRESTClient(ctx) if err != nil { log.Fatal("unable to create gcp subenetwork rest client", zap.Error(err)) } migrrssvc.WithGCPSubnentworkClient(cli) } { cli, err := compute.NewServiceAttachmentsRESTClient(ctx) if err != nil { log.Fatal("unable to create gcp service attchment rest client", zap.Error(err)) } migrrssvc.WithGCPServiceAttchementClient(cli) } { cli, err := gcp.NewClientSet() if err != nil { log.Fatal("failed to init gcp client", zap.Error(err)) } gcpResourceManager = cli.ResourceManager() upgradeSvc.WithGCPClient(cli) } { gkecli, err := container.NewClusterManagerClient(ctx) if err != nil { log.Fatal("unable to create gcp cluster manager client", zap.Error(err)) } appboxSvc.WithGKEClient(gkecli) networkSvc.WithGKEClient(gkecli) } } clusterAgentProxy.WithGKEPolicyControlResourceManager(gcpResourceManager) o11yrp := &proxy.ReverseProxyWrapper{} if config.IsEnableO11yService() { o11yrp, err = proxy.NewReverseProxy(config.O11yRegionalServiceEndpoint) if err != nil { log.Fatal("new reverse proxy for o11y regional server failed", zap.Error(err)) } o11yrp = o11yrp.WithPathPerfixRewrite("/api/dedicated/v1/migration/o11yregional", "/api/v1") } k8sResourceSvc := k8sresourcesvc.NewService( mgr.GetClient(), mgr.GetAPIReader(), mgr.GetConfig(), rsmanager, ) mountFns := []router.MountFn{ router.MountVPCPeeringHandler(vpcpeeringrouter.NewVPCPeeringHandler(vpcpeeringsvc)), router.MountClusterServiceProxyHandler(rp, clusterAgentProxy), router.MountClusterServiceHijackProxyHandler(chp), router.MountDedicatedRegionalMigrationHandler(migrationrsrouter.NewHandler(migrrssvc)), router.MountK8sUpgradeServiceHandler(k8supgrade.NewHandler(upgradeSvc)), router.MountK8sResourceServiceHandler(k8sresourcerouter.NewHandler(k8sResourceSvc)), router.MountAPPBoxServiceHandler(appboxrouter.NewAPPBoxHandler(appboxSvc)), router.MountCompatNeworkServiceHandler(networkrouter.NewNetworkHandler(networkSvc)), } switch config.Provider { case consts.ProviderTypeAWS, consts.ProviderTypeEKS: svc, err := migrationxpawssvc.NewService(mgr.GetClient(), mgr.GetAPIReader()) if err != nil { log.Fatal("new migration xpaws service failed", zap.Error(err)) } cli, err := initAWSClientSet() if err != nil { log.Fatal("failed to init aws client", zap.Error(err)) } mountFns = append(mountFns, router.MountCrossplaneAWSMigrationHandler(migrationxpawsrouter.NewHandler(svc))) plsHandler := plsrouter.NewPrivateLinkServiceHandler(mgr.GetClient(), config).WithAWSClient(cli) mountFns = append(mountFns, router.PrivateLinkServiceMounts(config.Provider, plsHandler)...) if config.IsEnableO11yService() { o11yawshp, err := proxy.NewO11yRegionlSericeAWSHijackProxy(o11yrp, cli) if err != nil { log.Fatal("new o11y regional aws hijack proxy failed", zap.Error(err)) } o11yawshp.WithLogger(ctrl.Log.WithName("router").WithName("gin").WithName("o11y-regional-hijack-proxy")) mountFns = append(mountFns, router.MountO11yAgentAWSMigrationHandler(o11yawshp)) } case consts.ProviderTypeGCP, consts.ProviderTypeGKE: svc, err := migrationxpgcpsvc.NewService(mgr.GetClient(), mgr.GetAPIReader()) if err != nil { log.Fatal("new migration xp aws service failed", zap.Error(err)) } cli, err := container.NewClusterManagerClient(ctx) if err != nil { log.Fatal("unable to create gcp cluster manager client", zap.Error(err)) } mountFns = append(mountFns, router.MountCrossplaneGCPMigrationHandler(migrationxpgcprouter.NewHandler(svc))) if config.IsEnableO11yService() { o11ygcphp, err := proxy.NewO11yRegionlSericeGCPHijackProxy(o11yrp, cli) if err != nil { log.Fatal("new o11y regional gcp hijack proxy failed", zap.Error(err)) } o11ygcphp.WithLogger(ctrl.Log.WithName("router").WithName("gin").WithName("o11y-regional-hijack-proxy")) mountFns = append(mountFns, router.MountO11yAgentGCPMigrationHandler(o11ygcphp)) } } handler := router.Route(mgr.GetClient(), clustercli.NewClusterCli(config.ClusterControlServiceEndpoint), config, tmplConfig, mountFns..., ) log.Info("starting api server") go router.StartService(ctx, ctrl.Log.WithName("router").WithName("gin"), models.ServiceInfo{ Port: port, }, handler) } func startControllers(ctx context.Context, mgr ctrl.Manager, config rsconfig.RootConfig, tmplConfig rsconfig.TemplatesConfig, rsmanager *controllers.Manager, agentControlAPI agentcontrolapi.AgentControlAPI) { provider := config.Provider var awsCli aws.Client var gcpCli gcp.Client var err error switch provider { case consts.ProviderTypeEKS: awsCli, err = initAWSClientSet() if err != nil { log.Fatal("failed to init aws client", zap.Error(err)) } case consts.ProviderTypeGKE: gcpCli, err = gcp.NewClientSet() if err != nil { log.Fatal("failed to init gcp client", zap.Error(err)) } } p, err := predicate.LabelSelectorPredicate(config.Predicate) if err != nil { log.Fatal("unable to create predicate", zap.Error(err)) } if err := vpc.NewVPCReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("vpc"), mgr.GetScheme(), mgr.GetEventRecorderFor("vpc-controller"), config, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create vpc controller", zap.Error(err)) } if err := appbox.NewAPPBoxReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("appbox"), mgr.GetScheme(), config, tmplConfig, rsmanager, mgr.GetEventRecorderFor("appbox-controller"), provider, awsCli, gcpCli, agentControlAPI, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create appbox controller", zap.Error(err)) } if err := asyncappbox.NewAsyncAPPBoxReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("asyncappbox"), mgr.GetScheme(), config, rsmanager, mgr.GetEventRecorderFor("asyncappbox-controller"), provider, gcpCli, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create asyncappbox controller", zap.Error(err)) } clusterCli := clustercli.NewClusterCli(config.ClusterControlServiceEndpoint) r := approute.NewAPPRouteReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("approute"), mgr.GetScheme(), mgr.GetEventRecorderFor("approute-controller"), config, agentControlAPI, ).WithClusterClient(clusterCli) if provider == consts.ProviderTypeGKE { cli, err := container.NewClusterManagerClient(ctx) if err != nil { log.Fatal("unable to create gcp cluster manager client", zap.Error(err)) } r.WithGCPContainerClient(cli) } if err := r.SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create approute controller", zap.Error(err)) } // register vpcpeering reconciler { r := vpcpeering.NewVPCPeeringReconciler( mgr.GetClient(), mgr.GetAPIReader(), ctrl.Log.WithName("controllers").WithName("vpcpeering"), mgr.GetScheme(), mgr.GetEventRecorderFor("vpcpeering-controller"), ) if provider == consts.ProviderTypeAWS || provider == consts.ProviderTypeEKS { // we use the service sa r = r.WithAWSClient(awsCli) } if err := r.SetupWithManager(mgr); err != nil { log.Fatal("unable to create vpcpeering controller", zap.Error(err)) } } if err := internet.NewInternetReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("internet"), mgr.GetScheme(), mgr.GetEventRecorderFor("internet-controller"), config, gcpCli, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create internet controller", zap.Error(err)) } if err := privatelinkservice.NewPrivateLinkServiceReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("privatelinkservice"), mgr.GetScheme(), mgr.GetEventRecorderFor("privatelinkservice-controller"), config, awsCli, clusterCli, gcpCli, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create privatelinkservice controller", zap.Error(err)) } if err := o11yroute.NewO11yRouteReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("o11yRoute"), mgr.GetScheme(), mgr.GetEventRecorderFor("o11yRoute-controller"), config, ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create o11yRoute controller", zap.Error(err)) } if err := upgradeoperation.NewReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("UpgradeOperation"), mgr.GetEventRecorderFor("UpgradeOperation-controller"), provider, clusterCli, gcpCli, awsCli, rsmanager, tmplConfig, config.APPBoxConfig, acscli.NewClient(config.GetACSServiceEndpoint()), ).SetupWithManager(mgr); err != nil { log.Fatal("unable to create UpgradeOperation controller", zap.Error(err)) } if err := privatelinkservicemigration.NewPrivateLinkServiceMigrationReconciler( mgr.GetClient(), ctrl.Log.WithName("controllers").WithName("privatelinkservicemigration"), mgr.GetScheme(), awsCli, clusterCli, mgr.GetEventRecorderFor("privatelinkservicemigration-controller"), ).SetupWithManager(mgr, p); err != nil { log.Fatal("unable to create privatelinkservicemigration controller", zap.Error(err)) } metrics.SetupCollectors(mgr.GetClient(), ctrl.Log.WithName("metrics-collector")) if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil { log.Fatal("unable to set up health check", zap.Error(err)) } if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil { log.Fatal("unable to set up ready check", zap.Error(err)) } } func buildCacheSelector(cfg rsconfig.RootConfig) (map[client.Object]cache.ByObject, error) { predicates := cfg.Predicate // parse cache selector from root config selector, err := metav1.LabelSelectorAsSelector(&predicates) if err != nil { return nil, err } return map[client.Object]cache.ByObject{ &v1alpha1.APPBox{}: { Label: selector, }, &v1alpha1.VPC{}: { Label: selector, }, &v1alpha1.VPCPeering{}: { Label: selector, }, &v1alpha1.Internet{}: { Label: selector, }, &v1alpha1.APPRoute{}: { Label: selector, }, &v1alpha1.O11yRoute{}: { Label: selector, }, &v1alpha1.APPController{}: { Label: selector, }, &v1alpha1.UpgradeOperation{}: { Label: selector, }, }, nil } func initAWSClientSet() (aws.Client, error) { cfg, err := config.LoadDefaultConfig(context.TODO()) if err != nil { return nil, err } return aws.NewClientSet(cfg), nil } // +kubebuilder:rbac:groups=ec2.aws.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=eks.aws.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=iam.aws.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=aws.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=route53.aws.crossplane.io,resources="*",verbs="*" func loadAWSScheme() { utilruntime.Must(ec2v1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(ec2v1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(manualv1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(awsiamv1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(iamv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(eksv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(eksv1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(xpawsv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(route53v1alpha1.SchemeBuilder.AddToScheme(Scheme)) } // +kubebuilder:rbac:groups=iam.aws.upbound.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=aws.upbound.io,resources="*",verbs="*" func loadAWSUpboundScheme() { utilruntime.Must(awsupboundiamv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(awsupboundv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(awsupboundeksv1beta1.SchemeBuilder.AddToScheme(Scheme)) } // +kubebuilder:rbac:groups=container.gcp.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=iam.gcp.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=gcp.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=storage.gcp.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=vpcpeering.gcp.crossplane.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=compute.gcp.crossplane.io,resources="*",verbs="*" func loadGCPScheme() { utilruntime.Must(containerv1beta1.SchemeBuilder.AddToScheme(Scheme)) // nodepool is v1beta1 utilruntime.Must(containerv1beta2.SchemeBuilder.AddToScheme(Scheme)) // cluster is v1beta2 utilruntime.Must(computev1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(gcpiamv1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(computev1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(gcpvpcpeeringv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(gcpstoragev1alpha1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(gcpv1beta1.SchemeBuilder.AddToScheme(Scheme)) utilruntime.Must(upboundcontainerv1beta1.SchemeBuilder.AddToScheme(Scheme)) } // +kubebuilder:rbac:groups=azure.upbound.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=network.azure.upbound.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=containerservice.azure.upbound.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=authorization.azure.upbound.io,resources="*",verbs="*" // +kubebuilder:rbac:groups=managedidentity.azure.upbound.io,resources="*",verbs="*" func loadAzureScheme() { utilruntime.Must(azurev1beta1.AddToScheme(Scheme)) utilruntime.Must(azurenetworkv1beta1.AddToScheme(Scheme)) utilruntime.Must(azurecontainerv1beta1.AddToScheme(Scheme)) utilruntime.Must(azureauthv1beta1.AddToScheme(Scheme)) utilruntime.Must(azuremanagedidentityv1beta1.AddToScheme(Scheme)) } func loadDataplaneScheme() { utilruntime.Must(certmanagerv1.AddToScheme(Scheme)) utilruntime.Must(pcav1beta1.AddToScheme(Scheme)) utilruntime.Must(asts.AddToScheme(Scheme)) utilruntime.Must(operatorv1alpha1.AddToScheme(Scheme)) utilruntime.Must(tiflowv1alpha1.AddToScheme(Scheme)) utilruntime.Must(apiextensionsv1.AddToScheme(Scheme)) utilruntime.Must(agentv2.AddToScheme(Scheme)) }