//nolint:funlen,maintidx package container import ( "context" "os" "github.com/aws/aws-sdk-go/aws/session" console_api "github.com/tidbcloud/dataflow-service/app/controllers/console" "github.com/tidbcloud/dataflow-service/app/controllers/inner" open_api "github.com/tidbcloud/dataflow-service/app/controllers/open" changefeed_open "github.com/tidbcloud/dataflow-service/app/controllers/open/changefeed" open_api_v2 "github.com/tidbcloud/dataflow-service/app/controllers/v1beta1" "github.com/tidbcloud/dataflow-service/app/event_bus/publisher/center" event_center_hooks "github.com/tidbcloud/dataflow-service/app/event_bus/publisher/hook_impl" "github.com/tidbcloud/dataflow-service/app/internal/aws_facade" aws_facade_impl "github.com/tidbcloud/dataflow-service/app/internal/aws_facade/impl" "github.com/tidbcloud/dataflow-service/app/internal/cluster_job" "github.com/tidbcloud/dataflow-service/app/internal/cluster_proxy" "github.com/tidbcloud/dataflow-service/app/internal/feature_gate" feature_gate_impl "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/impl" "github.com/tidbcloud/dataflow-service/app/internal/gcp_facade" gcp_facade_impl "github.com/tidbcloud/dataflow-service/app/internal/gcp_facade/impl" "github.com/tidbcloud/dataflow-service/app/internal/infra_metrics" infra_metrics_impl "github.com/tidbcloud/dataflow-service/app/internal/infra_metrics/impl" backup_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/backup/hooks" backup_export_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/backup_export/hooks" compact_backup_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/compact_backup/hooks" crrf_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/crrf/hooks" restore_metric_hooks "github.com/tidbcloud/dataflow-service/app/metrics/restore/hooks" "github.com/tidbcloud/dataflow-service/app/models/backup" backup_impl "github.com/tidbcloud/dataflow-service/app/models/backup/impl" log_backup_size_calculator "github.com/tidbcloud/dataflow-service/app/models/backup/impl/log_backup_size_calculator" billing_model "github.com/tidbcloud/dataflow-service/app/models/billing" billing_model_impl "github.com/tidbcloud/dataflow-service/app/models/billing/impl" "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed" biz_changefeed_impl "github.com/tidbcloud/dataflow-service/app/models/biz_changefeed/impl" "github.com/tidbcloud/dataflow-service/app/models/bucket" bucket_impl "github.com/tidbcloud/dataflow-service/app/models/bucket/impl" restoreaccess "github.com/tidbcloud/dataflow-service/app/models/classic_restore_access" restoreaccess_impl "github.com/tidbcloud/dataflow-service/app/models/classic_restore_access/impl" "github.com/tidbcloud/dataflow-service/app/models/config_manager" config_manager_impl "github.com/tidbcloud/dataflow-service/app/models/config_manager/impl" crrf_model "github.com/tidbcloud/dataflow-service/app/models/crrf" crrf_model_impl "github.com/tidbcloud/dataflow-service/app/models/crrf/impl" regional_private_link "github.com/tidbcloud/dataflow-service/app/models/private_link" regional_private_link_impl "github.com/tidbcloud/dataflow-service/app/models/private_link/impl" replay_model "github.com/tidbcloud/dataflow-service/app/models/replay" replay_impl "github.com/tidbcloud/dataflow-service/app/models/replay/impl" "github.com/tidbcloud/dataflow-service/app/models/restore" restore_impl "github.com/tidbcloud/dataflow-service/app/models/restore/impl" "github.com/tidbcloud/dataflow-service/app/models/ticdc" ticdc_impl "github.com/tidbcloud/dataflow-service/app/models/ticdc/impl" "github.com/tidbcloud/dataflow-service/app/models/ticdc_spec" "github.com/tidbcloud/dataflow-service/app/models/tidb" tidb_cluster_impl "github.com/tidbcloud/dataflow-service/app/models/tidb/impl" "github.com/tidbcloud/dataflow-service/app/repo" backup_repo "github.com/tidbcloud/dataflow-service/app/repo/backup" billing_repo "github.com/tidbcloud/dataflow-service/app/repo/billing" biz_changefeed_repo "github.com/tidbcloud/dataflow-service/app/repo/biz_changefeed" crrf_repo "github.com/tidbcloud/dataflow-service/app/repo/crrf" crrf_repo_impl "github.com/tidbcloud/dataflow-service/app/repo/crrf/impl" "github.com/tidbcloud/dataflow-service/app/repo/entity" repo_impl "github.com/tidbcloud/dataflow-service/app/repo/impl" private_link_repo "github.com/tidbcloud/dataflow-service/app/repo/private_link" replay_repo "github.com/tidbcloud/dataflow-service/app/repo/replay" restore_repo "github.com/tidbcloud/dataflow-service/app/repo/restore" ticdc_repo "github.com/tidbcloud/dataflow-service/app/repo/ticdc" ticdc_spec_repo "github.com/tidbcloud/dataflow-service/app/repo/ticdc_spec" tidb_repo "github.com/tidbcloud/dataflow-service/app/repo/tidb" "github.com/tidbcloud/dataflow-service/app/workflow" "github.com/tidbcloud/dataflow-service/app/workflow/action/biz_changefeed/scheduler" workflow_impl "github.com/tidbcloud/dataflow-service/app/workflow/impl" changefeedopenbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/openapi/v1beta1" service_config "github.com/tidbcloud/dataflow-service/internal/config" "github.com/tidbcloud/dataflow-service/pkg/account/accountdb" dfs3 "github.com/tidbcloud/dataflow-service/pkg/aws" "github.com/tidbcloud/dataflow-service/pkg/billing" "github.com/tidbcloud/dataflow-service/pkg/blob" "github.com/tidbcloud/dataflow-service/pkg/clusteragentcli" encryptor_wrapper "github.com/tidbcloud/dataflow-service/pkg/encryptor-wrapper" "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli" "github.com/tidbcloud/dataflow-service/pkg/id_generator" infraapi "github.com/tidbcloud/dataflow-service/pkg/infra-api" "github.com/tidbcloud/dataflow-service/pkg/log" regionalapi "github.com/tidbcloud/dataflow-service/pkg/regional-api" "github.com/tidbcloud/pkgv2/db" "github.com/tidbcloud/tiinf/sdk/auth" "go.uber.org/zap" gcp_compute "google.golang.org/api/compute/v1" gcp_dns "google.golang.org/api/dns/v1" "gorm.io/gorm" ) var globalContainer = &GlobalContainer{} type GlobalContainer struct { // db DB *gorm.DB TransProvider db.TransProvider BucketRepo repo.BucketRepo ClassicRestoreAccessRepo repo.ClassicRestoreAccessRepo BackupRepo backup_repo.BackupRepo BackupExportRepo backup_repo.BackupExportRepo AWSBucketConfigurationRepo repo.AWSBucketConfigurationRepo AWSBucketReplicationRepo repo.AWSBucketReplicationRepo GCPBucketConfigurationRepo repo.GCPBucketConfigurationRepo ServerlessBackupRepo backup_repo.ServerlessBackupRepo ServerlessUserClusterBackupRepo backup_repo.ServerlessUserClusterBackupRepo RestoreRepo restore_repo.RestoreRepo ServerlessRestoreRepo restore_repo.ServerlessRestoreRepo ChangefeedInstanceTypeRepo repo.ChangefeedInstanceTypeRepo CDCInstanceTypeRepo repo.CDCInstanceTypeRepo CDCMachineTypeRepo repo.CDCMachineTypeRepo BackupScheduleRepo backup_repo.BackupScheduleRepo LogBackupRepo backup_repo.LogBackupRepo LogBackupSizeRepo backup_repo.LogBackupSizeRepo BackupSettingRepo backup_repo.BackupSettingRepo LogBackupTruncateRepo backup_repo.LogBackupTruncateRepo CompactBackupRepo backup_repo.CompactBackupRepo PitBackupSwitchRepo backup_repo.PitBackupSwitchRepo CRRFSettingRepo crrf_repo.CRRFSettingRepo CRRFModel crrf_model.CRRFModel DedicatedGlobalGrpcClient grpcli.DedicatedGlobalGrpcClient BillingRecordRepo repo.BillingRecordRepo TiCDCClusterRepo repo.TiCDCClusterRepo TiCDCCustomStorageRepo repo.TiCDCCustomStorageRepo TiCDCBizChangefeedResourceRepo repo.TiCDCBizChangefeedResourceRepo TiCDCChangefeedRepo repo.TiCDCChangefeedRepo TiDBClusterGCSwitchRepo repo.TiDBClusterGCSwitchRepo KernelChangefeedTrafficRecordRepo repo.KernelChangefeedTrafficRecordRepo BizChangefeedRepo repo.BizChangefeedRepo BizPrivateLinkEndpointRepo repo.BizPrivateLinkEndpointRepo BizPrivateLinkEndpointUsageRepo repo.BizPrivateLinkEndpointUsageRepo // grpc clients CoreGrpcClient grpcli.CoreGrpcClient ImportGrpcClient grpcli.ImportGrpcClient AccountSvcGrpcClient grpcli.AccountSvcGrpcClient ChangefeedGrpcClient grpcli.ChangefeedGrpcClient AccountAclGrpcClient grpcli.AccountAclGrpcClient TiDBClusterInfraClient grpcli.TiDBClusterInfraClient ConfigCenterGrpcClient grpcli.ConfigCenterGrpcClient AccountProjectWrapper accountdb.ProjectWrapper ClusterAgentClient clusteragentcli.Client InternalVPCPeeringGrpcClient grpcli.InternalVPCPeeringGrpcClient RegionGrpcClient grpcli.RegionGrpcClient MgmtSvcGrpcClient grpcli.MgmtSvcGrpcClient DedicatedClusterServiceClient grpcli.DedicatedClusterServiceClient // regional api client RegionalClient regionalapi.RegionalClient TiCDCClusterAPIClient regionalapi.TiCDCClusterClient ChangefeedAPIClient regionalapi.ChangefeedClient TiCDCPrivateLinkAPIClient regionalapi.TiCDCPrivateLinkClient TiCDCPreCheckAPIClient regionalapi.TiCDCPreCheckClient CompactBackupClient regionalapi.CompactBackupClient CompactRegionalFacade infraapi.CompactFacade CRRFFacade regionalapi.CRRFFacade // aws AWSBucketClientProxy dfs3.AWSBucketClientProxy AWSIAMClientProxy dfs3.AWSIAMClientProxy AWSEC2ClientProxy dfs3.AWSEC2ClientProxy AWSEKSClientProxy dfs3.AWSEKSClientProxy AWSRoute53ClientProxy dfs3.AWSRoute53ClientProxy AWSBucketManagerModel bucket.AWSBucketManagerModel // gcp GCPComputeService gcp_compute.Service GCPDNSService gcp_dns.Service // azure AzureBucketManagerModel bucket.AzureBucketManagerModel // backup models AutoBackupScheduler backup.AutoBackupScheduler CompactBackupScheduler backup.CompactBackupScheduler CompactBackupModel backup.CompactBackupModel BackupSettingModel backup.BackupSettingModel SnapshotBackupModel backup.SnapshotBackupModel BackupExportModel backup.BackupExportModel PitBackupModel backup.PitBackupModel BackupCreateRouter backup.BackupCreateRouter SyncLogBackupFromInfra backup.LogBackupLongTermSynchronizationFromInfra LogBackupSizeCalculator backup.LogBackupSizeCalculator BackupBillingRecordGenerator backup.BackupBillingRecordGenerator LogBackupBillingRecordGenerator backup.BackupBillingRecordGenerator BackupHealthChecker backup.BackupHealthChecker // restore models RestoreSnapShotModel restore.RestoreModel RestorePiTRModel restore.RestoreModel RestoreModelRouter restore.RestoreModel // bucket models eKSBucketModel bucket.BucketModelV2 gkeBucketModel bucket.BucketModelV2 aksBucketModel bucket.BucketModelV2 BucketModelProxy bucket.BucketModelV2 ClassicRestoreAccessModel restoreaccess.Model AWSBucketWorkerV2 bucket.BucketWorkerV2 AWSBucketReplicationModel bucket.BucketReplicationModel AWSBucketReplicationWorker bucket.BucketReplicationWorker GCPBucketManagerModel bucket.GCPBucketManagerModel GCPProjectBucketWorkerV2 bucket.BucketWorkerV2 GCPClusterBucketWorkerV2 bucket.BucketWorkerV2 // cdc models CDCInstanceTypeModel ticdc_spec.CDCInstanceTypeModel ClusterJobModel cluster_job.Model TiCDCMachineManager ticdc.MachineManager RegionalTiCDCClusterModel ticdc.ClusterModel RegionalTiCDCChangefeedModel ticdc.ChangefeedModel ChangefeedTrafficMeter ticdc.ChangefeedTrafficMeter RegionalTiCDCClusterWorker ticdc.ClusterWorker RegionalTiCDCChangefeedWorker ticdc.LongTermSynchronization SinkEndpointModel regional_private_link.SinkEndpointModel SinkEndpointWorker regional_private_link.SinkEndpointWorker // billing models SyncBillingRecord billing_model.SyncBillingRecord // config manager models ConfigCenterManager config_manager.ConfigCenterManager TiCDCClusterConfigManager config_manager.TiCDCClusterConfigManager ChangefeedConfigManager config_manager.ChangefeedConfigManager // tidb model TiDBClusterGCSwitchModel tidb.ClusterGCSwitchModel // biz_changefeed model BizChangefeedDownstreamFactory *biz_changefeed_impl.DownstreamAdaptorFactory BizChangefeedSchedulerFactory *scheduler.Factory BizChangefeedHooker biz_changefeed.BizChangefeedHooker BizChangefeedModel biz_changefeed.BizChangefeedModel BizChangefeedMeteringWorker biz_changefeed.BizChangefeedMeteringWorker BizChangefeedMetricsWorker biz_changefeed.BizChangefeedMetricsWorker // Private Link models PrivateLinkWorker regional_private_link.BizPrivateLinkWorker DownstreamFactory *console_api.BizChangefeedDownstreamAdaptorFactory // replay ReplayRepo repo.ReplayRepo ReplayModel replay_model.ReplayModel ReplayWorker replay_model.ReplayWorker ReplayOperator replay_model.ReplayOperator // console-api BizChangefeedSpecController console_api.ChangefeedInstanceTypeController BizChangefeedController console_api.ChangefeedController BackupController console_api.BackupController BackupControllerV2 console_api.BackupControllerV2 CRRFController console_api.CRRFController SinkEndpointController console_api.SinkEndpointController // open-api RestoreOpenController open_api.RestoreController BackupOpenControllerV2 open_api_v2.BackupController // internal helper BackupRegionalFacade infraapi.BackupFacade RestoreRegionalFacade infraapi.RestoreFacade GCPStorageTransferClient gcp_facade.GCPStorageTransferClient WorkflowModel workflow.Workflow FeatureGate feature_gate.FeatureGate FeatureGateV2 feature_gate.FeatureGate Encryptor encryptor_wrapper.Encryptor PitBackupSwitcher backup.PitBackupSwitcher ClusterProxyModel cluster_proxy.Model InfraMetrics infra_metrics.InfraMetrics EventCenterPublisher center.EventCenterPublisher DataflowEventHookerManager entity.DataflowEventHookerManager AWSBatchOpsClient aws_facade.BatchOpsClient // backup grpc inner service BackupInnerController inner.BackupInnerController // grpc inner service for ops portal OpsInnerController inner.OpsInnerController // replay grpc inner service ReplayInnerController inner.ReplayInnerController // changefeed open api ChangefeedOpenAPI changefeedopenbp.ChangefeedServiceServer } func Global() *GlobalContainer { return globalContainer } // SetUpContainers this func only called once in main.go func SetUpContainers(ctx context.Context, db *gorm.DB) error { logger := log.GetLogger(ctx) container := &GlobalContainer{} // init db container.DB = db container.TransProvider = newTransProvider(container.DB) // init facade container.GCPStorageTransferClient = gcp_facade_impl.NewGCPStorageTransferClient() container.AWSBatchOpsClient = aws_facade_impl.NewBatchOpsClient() // init event center eventCenterPublisher, err := center.NewEventCenterPublisher() if err != nil { return err } container.EventCenterPublisher = eventCenterPublisher // init s3 sdk initAWSResourceWrapper(container) // init gcp sdk err = initGCPResourceWrapper(ctx, container) if err != nil { log.GetLogger(ctx).Error("init gcp sdk", zap.Error(err)) return err } // init azure sdk err = initAzureResourceWrapper(ctx, container) if err != nil { log.GetLogger(ctx).Error("init azure sdk", zap.Error(err)) return err } // init workflow err = initWorkflow(container) if err != nil { log.GetLogger(ctx).Error("init workflow failed", zap.Error(err)) return err } // init repo initRepos(container) // init grpc initGrpcClient(ctx, container) // init regional api initRegionalAPIClient(container) // init KMS encryptor, err := encryptor_wrapper.NewEncryptor(service_config.Global().InfraConfig.CMKArn) if err != nil { return err } container.Encryptor = encryptor // init internal helper initInternalHelper(container) // init model initModels(container) // init console controller initController(container) // init hooks initDBHooks(container) globalContainer = container logger.Info("container setup success") return nil } func initAWSResourceWrapper(container *GlobalContainer) { container.AWSBucketClientProxy = dfs3.NewAWSBucketProxy() container.AWSIAMClientProxy = dfs3.NewAWSIAMClientProxy() container.AWSEC2ClientProxy = dfs3.NewAWSEC2ClientProxy() container.AWSEKSClientProxy = dfs3.NewAWSEKSClientProxy() container.AWSRoute53ClientProxy = dfs3.NewAWSRoute53ClientProxy() container.AWSBucketManagerModel = bucket_impl.NewAWSBucketManagerModel( container.AWSBucketClientProxy, ) } func initAzureResourceWrapper(ctx context.Context, container *GlobalContainer) error { container.AzureBucketManagerModel = bucket_impl.NewAzureBucketManagerModel( service_config.Global().BackupConfig.AzureStorageAccountResourceGroup, ) return nil } func initGCPResourceWrapper(ctx context.Context, container *GlobalContainer) error { compute, err := gcp_compute.NewService(ctx) if err != nil { return err } dns, err := gcp_dns.NewService(ctx) if err != nil { return err } container.GCPComputeService = *compute container.GCPDNSService = *dns return nil } func initInternalHelper(container *GlobalContainer) { container.BackupRegionalFacade = regionalapi.NewBackupFacade(container.RegionalClient) container.RestoreRegionalFacade = regionalapi.NewRestoreFacade(container.RegionalClient) container.CRRFFacade = regionalapi.NewCRRFFacade(container.RegionalClient) container.FeatureGate = feature_gate_impl.NewFeatureGate() container.FeatureGateV2 = feature_gate_impl.NewFeatureGateV2() container.InfraMetrics = infra_metrics_impl.NewInfraMetrics() container.ClusterProxyModel = cluster_proxy.NewClusterProxyModelImpl( container.Encryptor, container.CoreGrpcClient, container.AccountAclGrpcClient, ) } func initController(container *GlobalContainer) { container.BizChangefeedController = console_api.NewBizChangefeedController( container.TransProvider, container.BizChangefeedModel, container.RegionalTiCDCClusterModel, container.CoreGrpcClient, container.FeatureGate, container.DownstreamFactory) var rbacCli changefeed_open.RBACClientFunc = auth.CheckIdentityPermission container.ChangefeedOpenAPI = changefeed_open.NewService(container.BizChangefeedController, container.CDCInstanceTypeModel, container.BizChangefeedModel, container.BizChangefeedRepo, container.TiCDCClusterRepo, container.TiCDCClusterConfigManager, rbacCli, container.MgmtSvcGrpcClient) container.BizChangefeedSpecController = console_api.NewBizChangefeedSpecController( container.TransProvider, container.BizChangefeedModel) container.BackupController = console_api.NewBackupController( container.TransProvider, container.RestoreModelRouter, container.SnapshotBackupModel, container.BackupSettingModel, container.BackupCreateRouter, container.PitBackupModel, container.BackupRepo, container.CoreGrpcClient, container.MgmtSvcGrpcClient, billing.GetHandler(), container.FeatureGate, ) container.BackupControllerV2 = console_api.NewBackupControllerV2( container.TransProvider, container.RestoreModelRouter, container.SnapshotBackupModel, container.BackupSettingModel, container.BackupCreateRouter, container.BackupExportRepo, container.BackupExportModel, container.BackupRepo, ) container.SinkEndpointController = console_api.NewSinkEndpointController(container.SinkEndpointModel) container.CRRFController = console_api.NewCRRFController( container.TransProvider, container.CRRFModel, ) // init open controller container.RestoreOpenController = open_api.NewRestoreController( container.TransProvider, container.RestoreModelRouter, container.SnapshotBackupModel, container.MgmtSvcGrpcClient, ) container.BackupOpenControllerV2 = open_api_v2.NewBackupController( container.TransProvider, container.BackupControllerV2, container.BackupController, container.BackupExportRepo, container.BackupExportModel, container.BackupCreateRouter, container.BackupRepo, container.SnapshotBackupModel, container.BackupSettingModel, container.PitBackupModel, container.BucketRepo, container.CoreGrpcClient, container.MgmtSvcGrpcClient, billing.GetHandler(), ) container.BackupInnerController = inner.NewBackupInnerController( container.SnapshotBackupModel, container.BucketModelProxy, container.RestoreModelRouter, container.MgmtSvcGrpcClient, container.ClassicRestoreAccessModel, ) container.OpsInnerController = inner.NewOpsInnerController( container.ConfigCenterManager, container.BizChangefeedModel, container.RegionalTiCDCClusterModel, container.RegionalTiCDCChangefeedModel, container.TiCDCClusterConfigManager, container.ChangefeedConfigManager, container.TiCDCClusterAPIClient, ) container.ReplayInnerController = inner.NewReplayInnerController( container.ReplayModel, ) } func initGrpcClient(ctx context.Context, container *GlobalContainer) { container.CoreGrpcClient = grpcli.NewCoreGrpcClient( ctx, service_config.Global().GrpcConfig.CoreSvcAddr) container.AccountSvcGrpcClient = grpcli.NewAccountSvcGrpcClient( ctx, service_config.Global().GrpcConfig.AccountProviderAddr) container.AccountAclGrpcClient = grpcli.NewAccountAclGrpcClient( ctx, service_config.Global().GrpcConfig.AccountAclSvcAddr) container.ImportGrpcClient = grpcli.NewImportGrpcClient( ctx, service_config.Global().GrpcConfig.DataflowSvcAddr) container.ChangefeedGrpcClient = grpcli.NewChangefeedGrpcClient( ctx, service_config.Global().GrpcConfig.DataflowSvcAddr) container.TiDBClusterInfraClient = grpcli.NewTiDBClusterInfraClient( ctx, service_config.Global().GrpcConfig.DedicatedGlobalSvcAddr) container.ConfigCenterGrpcClient = grpcli.NewConfigCenterGrpcClient( ctx, service_config.Global().GrpcConfig.ConfigCenterAddr) container.AccountProjectWrapper = accountdb.NewProjectWrapper() container.ClusterAgentClient = clusteragentcli.NewClient( service_config.Global().GrpcConfig.MgmtSvcProxyAddr, service_config.Global().GrpcConfig.MgmtSvcProxyToken) container.InternalVPCPeeringGrpcClient = grpcli.NewInternalVPCPeeringGrpcClientImpl( ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr) container.RegionGrpcClient = grpcli.NewRegionGrpcClient(ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr) container.MgmtSvcGrpcClient = grpcli.NewMgmtSvcGrpcClient(ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr) container.DedicatedClusterServiceClient = grpcli.NewDedicatedClusterServiceClient( ctx, service_config.Global().GrpcConfig.TiDBMgmtSvcAddr, ) container.DedicatedGlobalGrpcClient = grpcli.NewDedicatedGlobalGrpcClient(ctx, service_config.Global().GrpcConfig.DedicatedGlobalSvcAddr) } func initRegionalAPIClient(container *GlobalContainer) { container.RegionalClient = regionalapi.NewRegionalClient(container.RegionGrpcClient) container.TiCDCClusterAPIClient = regionalapi.NewTiCDCClusterClient(container.RegionalClient) container.ChangefeedAPIClient = regionalapi.NewChangefeedClient(container.RegionalClient) container.TiCDCPrivateLinkAPIClient = regionalapi.NewTiCDCPrivateLinkClient(container.RegionalClient) container.TiCDCPreCheckAPIClient = regionalapi.NewTiCDCPreCheckClient(container.RegionalClient) container.CompactBackupClient = regionalapi.NewCompactBackupClient(container.RegionalClient) container.CompactRegionalFacade = regionalapi.NewCompactFacade(container.RegionalClient) } func initWorkflow(container *GlobalContainer) error { impl, err := workflow_impl.NewWorkflow(checkIsWorker(), container.DB) if err != nil { return err } container.WorkflowModel = impl return nil } func checkIsWorker() bool { podName := os.Getenv("POD_NAME") return podName != "" } func initDBHooks(container *GlobalContainer) { container.DataflowEventHookerManager = entity.NewDataflowEventHookerManager() container.DataflowEventHookerManager.Register(event_center_hooks.GetBackupHooker()) container.DataflowEventHookerManager.Register(event_center_hooks.NewBackupExportHooker(container.BackupRepo)) container.DataflowEventHookerManager.Register(event_center_hooks.GetImportHooker()) container.DataflowEventHookerManager.Register(backup_metric_hooks.NewBackupMetricsHooker()) container.DataflowEventHookerManager.Register(backup_export_metric_hooks.NewBackupExportMetricsHooker()) container.DataflowEventHookerManager.Register(restore_metric_hooks.NewRestoreMetricsHooker()) container.DataflowEventHookerManager.Register(compact_backup_metric_hooks.NewCompactBackupMetricsHooker()) container.DataflowEventHookerManager.Register(crrf_metric_hooks.NewCRRFMetricsHooker()) } func initRepos(container *GlobalContainer) { container.BackupRepo = backup_repo.NewBackupRepo(container.TransProvider) container.BackupExportRepo = backup_repo.NewBackupExportRepo(container.TransProvider) container.AWSBucketConfigurationRepo = repo_impl.NewAWSBucketConfigurationRepo(container.TransProvider) container.AWSBucketReplicationRepo = repo_impl.NewAWSBucketReplicationRepo(container.TransProvider) container.GCPBucketConfigurationRepo = repo_impl.NewGCPBucketConfigurationRepo(container.TransProvider) container.ServerlessBackupRepo = backup_repo.NewServerlessBackupRepo(container.TransProvider) container.ServerlessUserClusterBackupRepo = backup_repo.NewServerlessUserClusterBackupRepo( container.TransProvider) container.RestoreRepo = restore_repo.NewRestoreRepo(container.TransProvider) container.ServerlessRestoreRepo = restore_repo.NewServerlessRestoreRepo(container.TransProvider) container.BucketRepo = repo_impl.NewBucketRepo(container.TransProvider) container.ClassicRestoreAccessRepo = repo_impl.NewClassicRestoreAccessRepo(container.TransProvider) container.ChangefeedInstanceTypeRepo = ticdc_spec_repo.NewChangefeedInstanceTypeRepo(container.TransProvider) container.CDCInstanceTypeRepo = ticdc_spec_repo.NewCDCInstanceTypeRepo(container.TransProvider) container.CDCMachineTypeRepo = ticdc_spec_repo.NewCDCMachineTypeRepo(container.TransProvider) container.BackupScheduleRepo = backup_repo.NewBackupScheduleRepo(container.TransProvider) container.LogBackupRepo = backup_repo.NewLogBackupRepo(container.TransProvider) container.LogBackupSizeRepo = backup_repo.NewLogBackupSizeRepo(container.TransProvider) container.BackupSettingRepo = backup_repo.NewBackupSettingRepo(container.TransProvider) container.LogBackupTruncateRepo = backup_repo.NewLogBackupTruncateRepo(container.TransProvider) container.CompactBackupRepo = backup_repo.NewCompactBackupRepo(container.TransProvider) container.PitBackupSwitchRepo = backup_repo.NewPitBackupSwitchRepo(container.TransProvider) container.CRRFSettingRepo = crrf_repo_impl.NewCRRFSettingRepo(container.TransProvider) container.BillingRecordRepo = billing_repo.NewBillingRecordRepo(container.TransProvider) container.TiCDCClusterRepo = ticdc_repo.NewTiCDCClusterRepo(container.TransProvider) container.TiCDCCustomStorageRepo = ticdc_repo.NewTiCDCCustomStorageRepo(container.TransProvider) container.TiCDCBizChangefeedResourceRepo = ticdc_repo.NewTiCDCBizChangefeedResourceRepo(container.TransProvider) container.TiCDCChangefeedRepo = ticdc_repo.NewTiCDCChangefeedRepo(container.TransProvider) container.KernelChangefeedTrafficRecordRepo = ticdc_repo. NewKernelChangefeedTrafficRecordRepo(container.TransProvider) container.TiDBClusterGCSwitchRepo = tidb_repo.NewTiDBClusterGCSwitchRepo(container.TransProvider) container.BizChangefeedRepo = biz_changefeed_repo.NewBizChangefeedRepo( container.TransProvider, id_generator.MustNewSnowflakeIDGenerator(id_generator.NewHostSnowflakeMachineIDGenerator())) container.BizPrivateLinkEndpointRepo = private_link_repo.NewBizPrivateLinkEndpointRepo(container.TransProvider) container.BizPrivateLinkEndpointUsageRepo = private_link_repo.NewBizPrivateLinkEndpointUsageRepo( container.TransProvider) // replay repo container.ReplayRepo = replay_repo.NewReplayRepo(container.TransProvider) } func initModels(container *GlobalContainer) { clusterJobModel := cluster_job.NewClusterJobModelImpl(container.CoreGrpcClient, container.MgmtSvcGrpcClient) container.ConfigCenterManager = config_manager_impl.NewConfigCenterManager(container.ConfigCenterGrpcClient) container.TiCDCClusterConfigManager = config_manager_impl.NewTiCDCClusterConfigCenterManager( container.ConfigCenterGrpcClient) container.ChangefeedConfigManager = config_manager_impl.NewChangefeedConfigCenterManager( container.ConfigCenterGrpcClient) container.TiDBClusterGCSwitchModel = tidb_cluster_impl.NewTiDBClusterGCSwitchModel( container.TransProvider, container.TiDBClusterGCSwitchRepo, container.ClusterProxyModel, ) container.ClusterJobModel = clusterJobModel container.PitBackupSwitcher = backup_impl.NewPitBackupSwitcher( container.TransProvider, container.PitBackupSwitchRepo, container.LogBackupRepo, container.WorkflowModel, container.MgmtSvcGrpcClient, container.ClusterJobModel, ) container.CDCInstanceTypeModel = ticdc_spec.NewCDCInstanceTypeModel( container.ChangefeedInstanceTypeRepo, container.CDCInstanceTypeRepo, container.CDCMachineTypeRepo, ) container.TiCDCMachineManager = ticdc_impl.NewMachineManagerRouter( container.CDCInstanceTypeModel, ) container.RegionalTiCDCClusterModel = ticdc_impl.NewTiCDCClusterRegionalModel( container.TransProvider, container.CDCInstanceTypeModel, container.TiCDCClusterAPIClient, container.TiCDCClusterRepo, container.TiCDCCustomStorageRepo, container.TiCDCClusterConfigManager, container.RegionGrpcClient, container.FeatureGate, ) container.RegionalTiCDCChangefeedModel = ticdc_impl.NewChangefeedRegionalModel( container.TransProvider, container.ChangefeedAPIClient, container.TiCDCClusterAPIClient, container.TiCDCChangefeedRepo, container.TiCDCClusterRepo, container.ChangefeedConfigManager, container.TiCDCPreCheckAPIClient, container.CoreGrpcClient, container.Encryptor, ) container.RegionalTiCDCClusterWorker = ticdc_impl.NewTiCDCClusterRegionalWorker( container.TransProvider, container.CDCInstanceTypeModel, container.TiCDCClusterAPIClient, container.TiCDCClusterRepo, container.TiCDCBizChangefeedResourceRepo, ) container.RegionalTiCDCChangefeedWorker = ticdc_impl.NewChangefeedRegionalWorker( container.ChangefeedAPIClient, container.TiCDCClusterRepo, container.TiCDCChangefeedRepo, ) container.ChangefeedTrafficMeter = ticdc_impl.NewChangefeedTrafficMeter( container.TransProvider, container.InfraMetrics, container.KernelChangefeedTrafficRecordRepo, ) container.SinkEndpointModel = regional_private_link_impl.NewSinkEndpointImpl( container.TransProvider, container.TiCDCPrivateLinkAPIClient, container.BizPrivateLinkEndpointRepo, container.BizPrivateLinkEndpointUsageRepo, ) container.SinkEndpointWorker = regional_private_link_impl.NewSinkEndpointWorkerImpl( container.TiCDCPrivateLinkAPIClient, container.BizPrivateLinkEndpointRepo, container.BizPrivateLinkEndpointUsageRepo, container.SinkEndpointModel, ) container.GCPBucketManagerModel = bucket_impl.NewGCPBucketManagerModel() // model need sub model , plz take care of the order { container.eKSBucketModel = bucket_impl.NewEKSBucketModel( container.TransProvider, container.BucketRepo, container.AWSBucketConfigurationRepo, container.WorkflowModel, container.TiDBClusterInfraClient, container.CoreGrpcClient, container.AWSBucketManagerModel, container.AWSBucketReplicationRepo, ) container.gkeBucketModel = bucket_impl.NewGKEBucketModel( container.TransProvider, container.BucketRepo, container.GCPBucketConfigurationRepo, container.GCPBucketManagerModel, container.CoreGrpcClient, container.TiDBClusterInfraClient, container.WorkflowModel, ) container.aksBucketModel = bucket_impl.NewAKSBucketModel( container.TransProvider, container.BucketRepo, container.Encryptor, // container.AzureBucketConfigurationRepo, container.AzureBucketManagerModel, container.WorkflowModel, container.MgmtSvcGrpcClient, ) container.BucketModelProxy = bucket_impl.NewBucketModelProxy( container.eKSBucketModel, container.gkeBucketModel, container.aksBucketModel, ) container.AWSBucketReplicationModel = bucket_impl.NewEKSBucketModel( container.TransProvider, container.BucketRepo, container.AWSBucketConfigurationRepo, container.WorkflowModel, container.TiDBClusterInfraClient, container.CoreGrpcClient, container.AWSBucketManagerModel, container.AWSBucketReplicationRepo, ) container.AWSBucketWorkerV2 = bucket_impl.NewAWSBucketWorker( container.BucketRepo, container.AWSBucketReplicationRepo, container.BackupRepo, container.LogBackupRepo, container.BucketModelProxy, container.CoreGrpcClient, container.AWSBucketManagerModel, ) container.AWSBucketReplicationWorker = bucket_impl.NewAWSBucketWorker( container.BucketRepo, container.AWSBucketReplicationRepo, container.BackupRepo, container.LogBackupRepo, container.BucketModelProxy, container.CoreGrpcClient, container.AWSBucketManagerModel, ) container.GCPProjectBucketWorkerV2 = bucket_impl.NewGCPProjectBucketWorker( container.BucketRepo, container.BackupRepo, container.LogBackupRepo, container.gkeBucketModel, container.CoreGrpcClient, ) container.GCPClusterBucketWorkerV2 = bucket_impl.NewGCPClusterBucketWorker( container.BucketRepo, container.BackupRepo, container.LogBackupRepo, container.gkeBucketModel, container.CoreGrpcClient, ) container.BackupSettingModel = backup_impl.NewBackupSettingModel( container.TransProvider, container.BackupSettingRepo, container.LogBackupRepo, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.WorkflowModel, container.BucketModelProxy, container.AWSBucketReplicationModel, container.BucketRepo, container.PitBackupSwitcher, container.FeatureGate, container.BackupScheduleRepo, container.AccountSvcGrpcClient, container.TiDBClusterInfraClient, container.RegionGrpcClient, container.CRRFSettingRepo, container.AWSBucketManagerModel, container.AWSBucketConfigurationRepo, ) container.CRRFModel = crrf_model_impl.NewCRRFModel( container.BackupSettingRepo, container.LogBackupRepo, container.CRRFSettingRepo, container.WorkflowModel, ) container.SnapshotBackupModel = backup_impl.NewSnapshotBackupModel( container.TransProvider, container.ClusterJobModel, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.ImportGrpcClient, container.AccountSvcGrpcClient, container.WorkflowModel, container.BucketRepo, container.BackupRepo, container.BackupScheduleRepo, container.RestoreRepo, container.BackupExportRepo, container.BackupSettingModel, container.FeatureGateV2, ) container.ClassicRestoreAccessModel = restoreaccess_impl.NewModel( container.SnapshotBackupModel, container.BucketModelProxy, container.ClassicRestoreAccessRepo, container.AWSIAMClientProxy, ) container.BackupExportModel = backup_impl.NewBackupExportModel( container.BackupRepo, container.BucketRepo, container.BackupExportRepo, container.GCPStorageTransferClient, container.AWSBatchOpsClient, container.WorkflowModel, ) container.PitBackupModel = backup_impl.NewPitBackupModel( container.TransProvider, container.PitBackupSwitchRepo, container.LogBackupRepo, container.BackupRepo, container.RestoreRepo, container.BucketRepo, container.WorkflowModel, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.SnapshotBackupModel, container.BackupSettingModel, container.BackupScheduleRepo, container.FeatureGateV2, ) container.BackupCreateRouter = backup_impl.NewBackupCreateRouter( container.BackupSettingModel, container.SnapshotBackupModel, container.PitBackupModel, ) container.AutoBackupScheduler = backup_impl.NewAutoBackupSchedulerImpl( blob.DefualtGetStorageFunc, container.TransProvider, container.BackupScheduleRepo, container.BackupSettingRepo, container.BackupRepo, container.LogBackupRepo, container.BucketRepo, container.LogBackupTruncateRepo, container.BackupSettingModel, container.MgmtSvcGrpcClient, container.ImportGrpcClient, container.BackupCreateRouter, container.SnapshotBackupModel, container.PitBackupModel, ) // CompactBackupModel container.CompactBackupModel = backup_impl.NewCompactBackupModel( container.TransProvider, container.CompactBackupRepo, container.LogBackupRepo, container.ClusterJobModel, container.WorkflowModel, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.BucketRepo, container.Encryptor, ) // CompactBackupScheduler container.CompactBackupScheduler = backup_impl.NewCompactBackupScheduler( container.TransProvider, container.LogBackupRepo, container.CompactBackupRepo, container.CompactBackupModel, container.CompactRegionalFacade, container.MgmtSvcGrpcClient, container.FeatureGate, ) container.BackupHealthChecker = backup_impl.NewBackupHealthChecker( container.BackupRepo, container.BackupExportRepo, container.LogBackupRepo, container.BackupScheduleRepo, container.BackupSettingModel, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.BackupRegionalFacade, container.WorkflowModel, ) container.RestoreSnapShotModel = restore_impl.NewSnapshotRestore( container.TransProvider, container.BackupRepo, container.RestoreRepo, container.BucketRepo, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.DedicatedGlobalGrpcClient, container.ConfigCenterGrpcClient, container.WorkflowModel, container.FeatureGateV2, ) container.RestorePiTRModel = restore_impl.NewPiTRestore( container.TransProvider, container.BackupRepo, container.RestoreRepo, container.BucketRepo, container.LogBackupRepo, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.DedicatedGlobalGrpcClient, container.ConfigCenterGrpcClient, container.WorkflowModel, container.FeatureGateV2, ) container.RestoreModelRouter = restore_impl.NewRestoreModelRouter( container.TransProvider, container.RestoreRepo, container.BackupRepo, container.RestoreSnapShotModel, container.RestorePiTRModel, container.CoreGrpcClient, container.MgmtSvcGrpcClient, container.DedicatedGlobalGrpcClient, container.AccountSvcGrpcClient, container.ConfigCenterGrpcClient, ) container.SyncLogBackupFromInfra = backup_impl.NewLogBackupLongTermSynchronizationFromInfra( container.TransProvider, container.LogBackupRepo, container.LogBackupTruncateRepo, container.RestoreRepo, container.BucketRepo, container.CoreGrpcClient, container.BackupRegionalFacade, container.MgmtSvcGrpcClient, ) container.LogBackupSizeCalculator = log_backup_size_calculator.NewLogBackupSizeCalculator( container.TransProvider, container.BucketRepo, container.LogBackupRepo, container.LogBackupSizeRepo, container.LogBackupTruncateRepo, container.Encryptor, ) container.BackupBillingRecordGenerator = backup_impl.NewBackupBillingRecordGenerator( container.TransProvider, container.BucketRepo, container.BackupRepo, container.BillingRecordRepo, container.MgmtSvcGrpcClient, container.AccountSvcGrpcClient, ) container.LogBackupBillingRecordGenerator = backup_impl.NewLogBackupBillingRecordGenerator( container.TransProvider, container.BucketRepo, container.LogBackupSizeRepo, container.BillingRecordRepo, container.MgmtSvcGrpcClient, container.AccountSvcGrpcClient, ) container.BizChangefeedDownstreamFactory = biz_changefeed_impl.NewDownstreamAdaptorFactory( container.TiCDCPreCheckAPIClient, container.RegionalTiCDCClusterModel, container.Encryptor, container.BizPrivateLinkEndpointRepo, container.CoreGrpcClient, container.TiDBClusterInfraClient, ) container.BizChangefeedSchedulerFactory = scheduler.NewFactory( container.TransProvider, container.BizChangefeedRepo, container.TiCDCBizChangefeedResourceRepo, container.RegionalTiCDCClusterModel, container.CDCInstanceTypeModel, container.CoreGrpcClient, container.TiDBClusterInfraClient, ) container.BizChangefeedHooker = biz_changefeed_impl.NewBizChangefeedEventBusHooker( container.EventCenterPublisher, ) container.BizChangefeedModel = biz_changefeed_impl.NewBizChangefeedModel( container.TransProvider, container.BizChangefeedRepo, container.ChangefeedInstanceTypeRepo, container.RegionalTiCDCChangefeedModel, container.RegionalTiCDCClusterModel, container.TiCDCMachineManager, container.SinkEndpointModel, container.ImportGrpcClient, container.CoreGrpcClient, container.WorkflowModel, billing.GetHandler(), container.BizChangefeedDownstreamFactory, container.BizChangefeedHooker, container.ClusterJobModel, ) container.BizChangefeedMeteringWorker = biz_changefeed_impl.NewBizChangefeedMeteringWorker( container.TiCDCMachineManager, container.BizChangefeedRepo, container.BillingRecordRepo, container.CoreGrpcClient, container.AccountProjectWrapper) container.BizChangefeedMetricsWorker = biz_changefeed_impl.NewBizChangefeedMetricsWorker( container.RegionalTiCDCChangefeedModel, container.BizChangefeedRepo, ) container.PrivateLinkWorker = regional_private_link_impl.NewPrivateLinkWorker( container.TransProvider, container.BillingRecordRepo, container.CoreGrpcClient, container.KernelChangefeedTrafficRecordRepo, container.ChangefeedTrafficMeter, container.BizChangefeedRepo, ) } container.SyncBillingRecord = billing_model_impl.NewSyncBillingRecord( container.TransProvider, container.BillingRecordRepo, ) container.DownstreamFactory = console_api.NewBizChangefeedDownstreamAdaptorFactory( container.SinkEndpointModel, session.Must(session.NewSession())) // replay model container.ReplayOperator = replay_impl.NewReplayOperator( container.MgmtSvcGrpcClient, regionalapi.NewReplayClient(container.RegionalClient), container.Encryptor, ) container.ReplayModel = replay_impl.NewReplayModel( container.TransProvider, container.ReplayRepo, container.ReplayOperator, container.Encryptor, ) container.ReplayWorker = replay_impl.NewReplayWorker( container.MgmtSvcGrpcClient, container.ReplayRepo, container.ReplayOperator, ) } func newTransProvider(handle *gorm.DB) db.TransProvider { return db.NewProvider(handle) }