From 35ff2f055aa6a247c8ee090f7f9402437e194f4d Mon Sep 17 00:00:00 2001 From: douxu Date: Wed, 24 Jun 2026 15:18:27 +0800 Subject: [PATCH] refactor: move ModelRT startup assembly into Wire app package - add app package with Wire providers, injector declaration, and generated InitializeApp implementation - move config loading, logger/tracer setup, database, redis, pool, RabbitMQ, task worker, router, and HTTP server construction out of main.go - preserve startup warmup, background publishers, task worker launch, and graceful shutdown behavior behind App methods - add github.com/google/wire dependency for generated dependency injection --- app/app.go | 135 +++++++++++++++++++++++ app/providers.go | 238 +++++++++++++++++++++++++++++++++++++++++ app/types.go | 39 +++++++ app/wire.go | 34 ++++++ app/wire_gen.go | 80 ++++++++++++++ go.mod | 1 + go.sum | 2 + main.go | 272 +++-------------------------------------------- 8 files changed, 543 insertions(+), 258 deletions(-) create mode 100644 app/app.go create mode 100644 app/providers.go create mode 100644 app/types.go create mode 100644 app/wire.go create mode 100644 app/wire_gen.go diff --git a/app/app.go b/app/app.go new file mode 100644 index 0000000..04cf5db --- /dev/null +++ b/app/app.go @@ -0,0 +1,135 @@ +package app + +import ( + "context" + "net/http" + "os" + "os/signal" + "syscall" + + "modelRT/config" + "modelRT/database" + "modelRT/diagram" + "modelRT/logger" + "modelRT/model" + "modelRT/mq" + "modelRT/real-time-data/alert" + "modelRT/task" + + "github.com/gomodule/redigo/redis" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "gorm.io/gorm" + + realtimedata "modelRT/real-time-data" +) + +// App owns the assembled service: the HTTP server plus every long-lived resource +// constructed during startup. Resource handles are retained so their lifecycles +// (and the Wire-aggregated cleanup) are tied to the App. +type App struct { + cfg config.ModelRTConfig + server *http.Server + db *gorm.DB + + tp *sdktrace.TracerProvider + parsePool ParsePool + searchPool *redis.Pool + storage StorageRedis + locker LockerRedis + anchorPool AnchorPool + rabbit *mq.RabbitMQProxy + taskWorker *task.TaskWorker + alertMgr *alert.EventManager +} + +// Warmup loads circuit-diagram, measurement, attribute and topology data from +// Postgres into the in-memory/redis caches. It mirrors the bootstrap transaction +// that used to run inline in main(); failures panic, as before. +func (a *App) Warmup(ctx context.Context) { + a.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + cacheMap, err := model.GetNSpathToIsLocalMap(ctx, a.db) + if err != nil { + logger.Error(ctx, "get nspath to is_local map failed", "error", err) + panic(err) + } + model.NSPathToIsLocalMap = cacheMap + + if err := model.CleanupRecommendRedisCache(ctx); err != nil { + logger.Error(ctx, "clean up component measurement and attribute group failed", "error", err) + panic(err) + } + + measurementSet, err := database.GetFullMeasurementSet(ctx, a.db) + if err != nil { + logger.Error(ctx, "generate component measurement group failed", "error", err) + panic(err) + } + fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet) + if err != nil { + logger.Error(ctx, "store component measurement group into redis failed", "error", err) + panic(err) + } + + compAttrSet, err := database.GenAllAttributeMap(tx) + if err != nil { + logger.Error(ctx, "generate component attribute group failed", "error", err) + panic(err) + } + + if err := model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet); err != nil { + logger.Error(ctx, "store component attribute group into redis failed", "error", err) + panic(err) + } + + allMeasurement, err := database.GetAllMeasurements(ctx, tx) + if err != nil { + logger.Error(ctx, "load topologic info from postgres failed", "error", err) + panic(err) + } + go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement) + + tree, _, err := database.QueryTopologicFromDB(ctx, tx) + if err != nil { + logger.Error(ctx, "load topologic info from postgres failed", "error", err) + panic(err) + } + diagram.GlobalTree = tree + return nil + }) +} + +// StartBackground launches the async publishers and the task worker. Each runs in +// its own goroutine, matching the original main() startup. +func (a *App) StartBackground(ctx context.Context) { + if a.taskWorker != nil { + go a.taskWorker.Start() + } + go mq.PushUpDownLimitEventToRabbitMQ(ctx, mq.MsgChan) + go mq.PushMessageToRabbitMQ(ctx, mq.MessageMsgChan) + go task.PushTaskToRabbitMQ(ctx, a.cfg.RabbitMQConfig, task.TaskMsgChan) +} + +// Run starts the HTTP server and blocks until it is shut down. A received +// interrupt/terminate signal triggers a graceful shutdown. +func (a *App) Run(ctx context.Context) { + done := make(chan os.Signal, 10) + signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) + go func() { + <-done + logger.Info(ctx, "shutdown signal received, cleaning up...") + if err := a.server.Shutdown(context.Background()); err != nil { + logger.Error(ctx, "shutdown serverError", "err", err) + } + logger.Info(ctx, "resources cleaned up, exiting") + }() + + logger.Info(ctx, "starting ModelRT server") + err := a.server.ListenAndServe() + if err != nil { + if err == http.ErrServerClosed { + logger.Info(ctx, "server closed under request") + } else { + logger.Error(ctx, "server closed unexpected", "err", err) + } + } +} diff --git a/app/providers.go b/app/providers.go new file mode 100644 index 0000000..8633f3f --- /dev/null +++ b/app/providers.go @@ -0,0 +1,238 @@ +package app + +import ( + "context" + "errors" + "fmt" + "log" + "net/http" + "os" + "path/filepath" + "time" + + "modelRT/config" + "modelRT/constants" + "modelRT/database" + "modelRT/diagram" + locker "modelRT/distributedlock" + "modelRT/logger" + "modelRT/middleware" + "modelRT/model" + "modelRT/mq" + "modelRT/pool" + "modelRT/real-time-data/alert" + "modelRT/router" + "modelRT/task" + "modelRT/util" + + "github.com/gin-contrib/cors" + "github.com/gin-gonic/gin" + "github.com/gomodule/redigo/redis" + "github.com/panjf2000/ants/v2" + swaggerFiles "github.com/swaggo/files" + ginSwagger "github.com/swaggo/gin-swagger" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "gorm.io/gorm" +) + +// provideConfig loads (and, if missing, bootstraps from the example file) the +// service configuration, then initializes the global logger. Because every other +// provider depends transitively on the config, initializing the logger here +// guarantees it is ready before any provider that logs. The returned cleanup +// flushes the logger on shutdown. +func provideConfig(cli CLIArgs) (config.ModelRTConfig, func(), error) { + configPath := filepath.Join(cli.ConfigDir, cli.ConfigName+"."+cli.ConfigType) + if _, err := os.Stat(configPath); os.IsNotExist(err) { + log.Println("configuration file not found, checking for example file") + + exampleConfigPath := filepath.Join(cli.ConfigDir, cli.ConfigName+".example."+cli.ConfigType) + configDir := filepath.Dir(configPath) + if err := os.MkdirAll(configDir, 0o755); err != nil { + return config.ModelRTConfig{}, nil, fmt.Errorf("failed to create config directory %s: %w", configDir, err) + } + if _, err := os.Stat(exampleConfigPath); err == nil { + if err := util.CopyFile(exampleConfigPath, configPath); err != nil { + return config.ModelRTConfig{}, nil, fmt.Errorf("failed to copy example config file: %w", err) + } + } else { + return config.ModelRTConfig{}, nil, errors.New("no config file and no config example file found") + } + } + + cfg := config.ReadAndInitConfig(cli.ConfigDir, cli.ConfigName, cli.ConfigType) + + logger.InitLoggerInstance(cfg.LoggerConfig) + cleanup := func() { _ = logger.GetLoggerInstance().Sync() } + + return cfg, cleanup, nil +} + +// provideTracerProvider initializes the OTel tracer provider. A failure here is +// non-fatal: tracing is simply disabled, matching the original startup behavior. +func provideTracerProvider(ctx context.Context, cfg config.ModelRTConfig) (*sdktrace.TracerProvider, func()) { + tp, err := middleware.InitTracerProvider(ctx, cfg) + if err != nil { + log.Printf("warn: OTLP tracer init failed, tracing disabled: %v", err) + return nil, func() {} + } + cleanup := func() { + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + _ = tp.Shutdown(shutdownCtx) + } + return tp, cleanup +} + +// provideServiceToken generates the signed client token bound to this host. +func provideServiceToken(cfg config.ModelRTConfig) (ServiceToken, error) { + hostName, err := os.Hostname() + if err != nil { + return "", fmt.Errorf("get host name failed: %w", err) + } + tok, err := util.GenerateClientToken(hostName, cfg.ServiceName, cfg.SecretKey) + if err != nil { + return "", fmt.Errorf("generate client token failed: %w", err) + } + return ServiceToken(tok), nil +} + +// providePostgres opens the global Postgres client and closes the underlying +// sql.DB on shutdown. +func providePostgres(ctx context.Context, cfg config.ModelRTConfig) (*gorm.DB, func(), error) { + db := database.InitPostgresDBInstance(ctx, cfg.PostgresDBURI) + cleanup := func() { + if sqlDB, err := db.DB(); err == nil { + _ = sqlDB.Close() + } + } + return db, cleanup, nil +} + +// provideParsePool builds the ants pool that runs model parse tasks. +func provideParsePool(cfg config.ModelRTConfig) (ParsePool, func(), error) { + p, err := ants.NewPoolWithFunc(cfg.ParseConcurrentQuantity, pool.ParseFunc) + if err != nil { + return nil, nil, fmt.Errorf("init concurrent parse task pool failed: %w", err) + } + return ParsePool(p), func() { p.Release() }, nil +} + +// provideSearchPool builds the redigo pool used for autocomplete/search and +// wires it into the global autocompleter. +func provideSearchPool(cfg config.ModelRTConfig) (*redis.Pool, func(), error) { + sp, err := util.NewRedigoPool(cfg.StorageRedisConfig) + if err != nil { + return nil, nil, fmt.Errorf("init redigo pool failed: %w", err) + } + model.InitAutocompleterWithPool(sp) + return sp, func() { _ = sp.Close() }, nil +} + +// provideStorageRedis opens the go-redis client backing diagram/storage data. +func provideStorageRedis(cfg config.ModelRTConfig) (StorageRedis, func()) { + c := diagram.InitRedisClientInstance(cfg.StorageRedisConfig, cfg.DeployEnv) + return StorageRedis(c), func() { _ = c.Close() } +} + +// provideLockerRedis opens the go-redis client backing the distributed lock. +func provideLockerRedis(cfg config.ModelRTConfig) (LockerRedis, func()) { + c := locker.InitClientInstance(cfg.LockerRedisConfig, cfg.DeployEnv) + return LockerRedis(c), func() { _ = c.Close() } +} + +// provideAnchorPool builds the ants pool that runs real-time anchor param tasks. +func provideAnchorPool(cfg config.ModelRTConfig) (AnchorPool, func(), error) { + p, err := pool.AnchorPoolInit(cfg.RTDReceiveConcurrentQuantity) + if err != nil { + return nil, nil, fmt.Errorf("init concurrent anchor param task pool failed: %w", err) + } + return AnchorPool(p), func() { p.Release() }, nil +} + +// provideRabbit opens the global RabbitMQ proxy. +func provideRabbit(ctx context.Context, cfg config.ModelRTConfig) (*mq.RabbitMQProxy, func()) { + proxy := mq.InitRabbitProxy(ctx, cfg.RabbitMQConfig) + return proxy, func() { mq.CloseRabbitProxy() } +} + +// provideTaskWorker initializes the async task worker. A failure is non-fatal: +// the service continues without a worker, matching the original startup behavior. +// The returned worker may be nil; App.StartBackground guards against that. +func provideTaskWorker(ctx context.Context, cfg config.ModelRTConfig, db *gorm.DB) (*task.TaskWorker, func()) { + tw, err := task.InitTaskWorker(ctx, cfg, db) + if err != nil { + logger.Error(ctx, "failed to initialize task worker", "error", err) + return nil, func() {} + } + return tw, func() { _ = tw.Stop() } +} + +// provideAlertManager initializes the global alert event manager. Its value is +// held by App purely so Wire keeps it in the dependency graph. +func provideAlertManager() *alert.EventManager { + return alert.InitAlertEventManager() +} + +// provideEngine builds the gin engine with CORS, routes and (outside production) +// the swagger UI. +func provideEngine(cfg config.ModelRTConfig, token ServiceToken) *gin.Engine { + if cfg.DeployEnv == constants.ProductionDeployMode { + gin.SetMode(gin.ReleaseMode) + } + engine := gin.New() + engine.Use(cors.New(cors.Config{ + AllowOrigins: []string{"*"}, + AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"}, + AllowHeaders: []string{"Origin", "Content-Type", "Authorization"}, + ExposeHeaders: []string{"Content-Length"}, + AllowCredentials: true, + MaxAge: 12 * time.Hour, + })) + router.RegisterRoutes(engine, string(token)) + + if cfg.DeployEnv != constants.ProductionDeployMode { + engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) + } + return engine +} + +// provideServer builds the HTTP server bound to the configured address. +func provideServer(cfg config.ModelRTConfig, engine *gin.Engine) *http.Server { + return &http.Server{ + Addr: cfg.ServiceAddr, + Handler: engine, + } +} + +// provideApp aggregates every constructed resource into the App. App holds the +// resource handles (pools, redis clients, etc.) so that Wire keeps their +// providers — and therefore their cleanups — in the dependency graph. +func provideApp( + cfg config.ModelRTConfig, + server *http.Server, + db *gorm.DB, + tp *sdktrace.TracerProvider, + parsePool ParsePool, + searchPool *redis.Pool, + storage StorageRedis, + lockerClient LockerRedis, + anchorPool AnchorPool, + rabbit *mq.RabbitMQProxy, + taskWorker *task.TaskWorker, + alertMgr *alert.EventManager, +) *App { + return &App{ + cfg: cfg, + server: server, + db: db, + tp: tp, + parsePool: parsePool, + searchPool: searchPool, + storage: storage, + locker: lockerClient, + anchorPool: anchorPool, + rabbit: rabbit, + taskWorker: taskWorker, + alertMgr: alertMgr, + } +} diff --git a/app/types.go b/app/types.go new file mode 100644 index 0000000..3297e6a --- /dev/null +++ b/app/types.go @@ -0,0 +1,39 @@ +// Package app holds the dependency-injection wiring for the ModelRT service. +// +// Phase 1 of the Wire migration centralizes the resource assembly that used to +// live inline in main(). Package-level singletons and their Get... accessors are +// intentionally left in place; only the startup assembly is moved here. +package app + +import ( + "github.com/panjf2000/ants/v2" + "github.com/redis/go-redis/v9" +) + +// CLIArgs carries the command-line configuration locators into the injector. +type CLIArgs struct { + ConfigDir string + ConfigName string + ConfigType string +} + +// The following named types disambiguate providers that would otherwise share +// the same underlying type. Wire resolves dependencies by type, so the two +// *redis.Client instances (storage vs locker) and the two *ants.PoolWithFunc +// instances (parse vs anchor) must be distinct named types. +type ( + // ServiceToken is the signed client token used to authenticate routes. + ServiceToken string + + // StorageRedis is the go-redis client backing diagram/storage data. + StorageRedis *redis.Client + + // LockerRedis is the go-redis client backing the distributed lock. + LockerRedis *redis.Client + + // ParsePool is the ants pool running model parse tasks. + ParsePool *ants.PoolWithFunc + + // AnchorPool is the ants pool running real-time anchor param tasks. + AnchorPool *ants.PoolWithFunc +) diff --git a/app/wire.go b/app/wire.go new file mode 100644 index 0000000..86f9aa3 --- /dev/null +++ b/app/wire.go @@ -0,0 +1,34 @@ +//go:build wireinject +// +build wireinject + +package app + +import ( + "context" + + "github.com/google/wire" +) + +// InitializeApp builds the fully assembled App and an aggregated cleanup function +// from the command-line config locators. The body is a Wire injector declaration; +// the real implementation is generated into wire_gen.go by `go generate`. +func InitializeApp(ctx context.Context, cli CLIArgs) (*App, func(), error) { + wire.Build( + provideConfig, + provideTracerProvider, + provideServiceToken, + providePostgres, + provideParsePool, + provideSearchPool, + provideStorageRedis, + provideLockerRedis, + provideAnchorPool, + provideRabbit, + provideTaskWorker, + provideAlertManager, + provideEngine, + provideServer, + provideApp, + ) + return nil, nil, nil +} diff --git a/app/wire_gen.go b/app/wire_gen.go new file mode 100644 index 0000000..ec4efe9 --- /dev/null +++ b/app/wire_gen.go @@ -0,0 +1,80 @@ +// Code generated by Wire. DO NOT EDIT. + +//go:generate go run -mod=mod github.com/google/wire/cmd/wire +//go:build !wireinject +// +build !wireinject + +package app + +import ( + "context" +) + +// Injectors from wire.go: + +// InitializeApp builds the fully assembled App and an aggregated cleanup function +// from the command-line config locators. The body is a Wire injector declaration; +// the real implementation is generated into wire_gen.go by `go generate`. +func InitializeApp(ctx context.Context, cli CLIArgs) (*App, func(), error) { + modelRTConfig, cleanup, err := provideConfig(cli) + if err != nil { + return nil, nil, err + } + serviceToken, err := provideServiceToken(modelRTConfig) + if err != nil { + cleanup() + return nil, nil, err + } + engine := provideEngine(modelRTConfig, serviceToken) + server := provideServer(modelRTConfig, engine) + db, cleanup2, err := providePostgres(ctx, modelRTConfig) + if err != nil { + cleanup() + return nil, nil, err + } + tracerProvider, cleanup3 := provideTracerProvider(ctx, modelRTConfig) + parsePool, cleanup4, err := provideParsePool(modelRTConfig) + if err != nil { + cleanup3() + cleanup2() + cleanup() + return nil, nil, err + } + pool, cleanup5, err := provideSearchPool(modelRTConfig) + if err != nil { + cleanup4() + cleanup3() + cleanup2() + cleanup() + return nil, nil, err + } + storageRedis, cleanup6 := provideStorageRedis(modelRTConfig) + lockerRedis, cleanup7 := provideLockerRedis(modelRTConfig) + anchorPool, cleanup8, err := provideAnchorPool(modelRTConfig) + if err != nil { + cleanup7() + cleanup6() + cleanup5() + cleanup4() + cleanup3() + cleanup2() + cleanup() + return nil, nil, err + } + rabbitMQProxy, cleanup9 := provideRabbit(ctx, modelRTConfig) + taskWorker, cleanup10 := provideTaskWorker(ctx, modelRTConfig, db) + eventManager := provideAlertManager() + app := provideApp(modelRTConfig, server, db, tracerProvider, parsePool, pool, storageRedis, lockerRedis, anchorPool, rabbitMQProxy, taskWorker, eventManager) + return app, func() { + cleanup10() + cleanup9() + cleanup8() + cleanup7() + cleanup6() + cleanup5() + cleanup4() + cleanup3() + cleanup2() + cleanup() + }, nil +} diff --git a/go.mod b/go.mod index 441ab18..dff199b 100644 --- a/go.mod +++ b/go.mod @@ -10,6 +10,7 @@ require ( github.com/gin-gonic/gin v1.10.1 github.com/gofrs/uuid v4.4.0+incompatible github.com/gomodule/redigo v1.8.9 + github.com/google/wire v0.7.0 github.com/gorilla/websocket v1.5.3 github.com/json-iterator/go v1.1.12 github.com/natefinch/lumberjack v2.0.0+incompatible diff --git a/go.sum b/go.sum index 19c0d18..19fc260 100644 --- a/go.sum +++ b/go.sum @@ -80,6 +80,8 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/google/wire v0.7.0 h1:JxUKI6+CVBgCO2WToKy/nQk0sS+amI9z9EjVmdaocj4= +github.com/google/wire v0.7.0/go.mod h1:n6YbUQD9cPKTnHXEBN2DXlOp/mVADhVErcMFb0v3J18= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= diff --git a/main.go b/main.go index 84fa153..6378073 100644 --- a/main.go +++ b/main.go @@ -3,44 +3,12 @@ package main import ( "context" - "errors" "flag" - "fmt" "log" - "net/http" - "os" - "os/signal" - "path/filepath" - "syscall" - "time" - "modelRT/config" - "modelRT/constants" - "modelRT/database" - "modelRT/diagram" - "modelRT/logger" - "modelRT/middleware" - "modelRT/model" - "modelRT/mq" - "modelRT/pool" - "modelRT/real-time-data/alert" - "modelRT/router" - "modelRT/task" - "modelRT/util" + "modelRT/app" - "github.com/gin-contrib/cors" - - locker "modelRT/distributedlock" _ "modelRT/docs" - - realtimedata "modelRT/real-time-data" - - "github.com/gin-gonic/gin" - "github.com/panjf2000/ants/v2" - swaggerFiles "github.com/swaggo/files" - ginSwagger "github.com/swaggo/gin-swagger" - "go.opentelemetry.io/otel" - "gorm.io/gorm" ) var ( @@ -49,13 +17,6 @@ var ( modelRTConfigType = flag.String("modelRT_config_type", "yaml", "config file type of model runtime service") ) -var ( - modelRTConfig config.ModelRTConfig - postgresDBClient *gorm.DB - // alertManager *alert.EventManager -) - -// TODO 使用 wire 依赖注入管理 DVIE 面板注册的 panel // @title ModelRT 实时模型服务 API 文档 // @version 1.0 // @description 实时数据计算和模型运行服务的 API 服务 @@ -74,229 +35,24 @@ var ( func main() { flag.Parse() - configPath := filepath.Join(*modelRTConfigDir, *modelRTConfigName+"."+*modelRTConfigType) - if _, err := os.Stat(configPath); os.IsNotExist(err) { - log.Println("configuration file not found,checking for example file") + ctx := context.Background() - exampleConfigPath := filepath.Join(*modelRTConfigDir, *modelRTConfigName+".example."+*modelRTConfigType) - configDir := filepath.Dir(configPath) - if err := os.MkdirAll(configDir, 0o755); err != nil { - panic(fmt.Errorf("failed to create config directory %s:%w", configDir, err)) - } - if _, err := os.Stat(exampleConfigPath); err == nil { - if err := util.CopyFile(exampleConfigPath, configPath); err != nil { - panic(fmt.Errorf("failed to copy example config file:%w", err)) - } - } else { - panic(errors.New("no config file and no config example file found")) - } + cli := app.CLIArgs{ + ConfigDir: *modelRTConfigDir, + ConfigName: *modelRTConfigName, + ConfigType: *modelRTConfigType, } - modelRTConfig = config.ReadAndInitConfig(*modelRTConfigDir, *modelRTConfigName, *modelRTConfigType) - - // init logger - logger.InitLoggerInstance(modelRTConfig.LoggerConfig) - defer logger.GetLoggerInstance().Sync() - - // init OTel TracerProvider - tp, tpErr := middleware.InitTracerProvider(context.Background(), modelRTConfig) - if tpErr != nil { - log.Printf("warn: OTLP tracer init failed, tracing disabled: %v", tpErr) - } - if tp != nil { - defer func() { - shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - tp.Shutdown(shutdownCtx) - }() - } - - ctx, startupSpan := otel.Tracer("modelRT/main").Start(context.Background(), "startup") - defer startupSpan.End() - - hostName, err := os.Hostname() + application, cleanup, err := app.InitializeApp(ctx, cli) if err != nil { - logger.Error(ctx, "get host name failed", "error", err) - panic(err) + log.Fatalf("failed to initialize app: %v", err) } + defer cleanup() - serviceToken, err := util.GenerateClientToken(hostName, modelRTConfig.ServiceName, modelRTConfig.SecretKey) - if err != nil { - logger.Error(ctx, "generate client token failed", "error", err) - panic(err) - } + // warm caches from postgres, then start background publishers/worker + application.Warmup(ctx) + application.StartBackground(ctx) - // init postgresDBClient - postgresDBClient = database.InitPostgresDBInstance(ctx, modelRTConfig.PostgresDBURI) - - defer func() { - sqlDB, err := postgresDBClient.DB() - if err != nil { - panic(err) - } - sqlDB.Close() - }() - - // init alert manager - _ = alert.InitAlertEventManager() - - // init model parse ants pool - parsePool, err := ants.NewPoolWithFunc(modelRTConfig.ParseConcurrentQuantity, pool.ParseFunc) - if err != nil { - logger.Error(ctx, "init concurrent parse task pool failed", "error", err) - panic(err) - } - defer parsePool.Release() - - searchPool, err := util.NewRedigoPool(modelRTConfig.StorageRedisConfig) - if err != nil { - logger.Error(ctx, "init redigo pool failed", "error", err) - panic(err) - } - defer searchPool.Close() - model.InitAutocompleterWithPool(searchPool) - - storageClient := diagram.InitRedisClientInstance(modelRTConfig.StorageRedisConfig, modelRTConfig.DeployEnv) - defer storageClient.Close() - - lockerClient := locker.InitClientInstance(modelRTConfig.LockerRedisConfig, modelRTConfig.DeployEnv) - defer lockerClient.Close() - - // init anchor param ants pool - anchorRealTimePool, err := pool.AnchorPoolInit(modelRTConfig.RTDReceiveConcurrentQuantity) - if err != nil { - logger.Error(ctx, "init concurrent anchor param task pool failed", "error", err) - panic(err) - } - defer anchorRealTimePool.Release() - - // init rabbitmq connection - mq.InitRabbitProxy(ctx, modelRTConfig.RabbitMQConfig) - defer mq.CloseRabbitProxy() - - // init async task worker - taskWorker, err := task.InitTaskWorker(ctx, modelRTConfig, postgresDBClient) - if err != nil { - logger.Error(ctx, "failed to initialize task worker", "error", err) - // continue without task worker, but log warning - } else { - go taskWorker.Start() - defer taskWorker.Stop() - } - - // async push event to rabbitMQ - go mq.PushUpDownLimitEventToRabbitMQ(ctx, mq.MsgChan) - // async push message (task state changes etc.) to rabbitMQ - go mq.PushMessageToRabbitMQ(ctx, mq.MessageMsgChan) - // async push task message to rabbitMQ - go task.PushTaskToRabbitMQ(ctx, modelRTConfig.RabbitMQConfig, task.TaskMsgChan) - - postgresDBClient.WithContext(ctx).Transaction(func(tx *gorm.DB) error { - // load circuit diagram from postgres - // componentTypeMap, err := database.QueryCircuitDiagramComponentFromDB(cancelCtx, tx, parsePool) - // if err != nil { - // logger.Error(ctx, "load circuit diagrams from postgres failed", "error", err) - // panic(err) - // } - - cacheMap, err := model.GetNSpathToIsLocalMap(ctx, postgresDBClient) - if err != nil { - logger.Error(ctx, "get nspath to is_local map failed", "error", err) - panic(err) - } - model.NSPathToIsLocalMap = cacheMap - - err = model.CleanupRecommendRedisCache(ctx) - if err != nil { - logger.Error(ctx, "clean up component measurement and attribute group failed", "error", err) - panic(err) - } - - measurementSet, err := database.GetFullMeasurementSet(ctx, postgresDBClient) - if err != nil { - logger.Error(ctx, "generate component measurement group failed", "error", err) - panic(err) - } - fullParentPath, isLocalParentPath, err := model.TraverseMeasurementGroupTables(ctx, *measurementSet) - if err != nil { - logger.Error(ctx, "store component measurement group into redis failed", "error", err) - panic(err) - } - - compAttrSet, err := database.GenAllAttributeMap(tx) - if err != nil { - logger.Error(ctx, "generate component attribute group failed", "error", err) - panic(err) - } - - err = model.TraverseAttributeGroupTables(ctx, tx, fullParentPath, isLocalParentPath, compAttrSet) - if err != nil { - logger.Error(ctx, "store component attribute group into redis failed", "error", err) - panic(err) - } - - allMeasurement, err := database.GetAllMeasurements(ctx, tx) - if err != nil { - logger.Error(ctx, "load topologic info from postgres failed", "error", err) - panic(err) - } - go realtimedata.StartComputingRealTimeDataLimit(ctx, allMeasurement) - - tree, _, err := database.QueryTopologicFromDB(ctx, tx) - if err != nil { - logger.Error(ctx, "load topologic info from postgres failed", "error", err) - panic(err) - } - diagram.GlobalTree = tree - return nil - }) - - // use release mode in production - if modelRTConfig.DeployEnv == constants.ProductionDeployMode { - gin.SetMode(gin.ReleaseMode) - } - engine := gin.New() - // add CORS middleware - engine.Use(cors.New(cors.Config{ - AllowOrigins: []string{"*"}, // 或指定具体域名 - AllowMethods: []string{"GET", "POST", "PUT", "DELETE", "OPTIONS"}, - AllowHeaders: []string{"Origin", "Content-Type", "Authorization"}, - ExposeHeaders: []string{"Content-Length"}, - AllowCredentials: true, - MaxAge: 12 * time.Hour, - })) - router.RegisterRoutes(engine, serviceToken) - - if modelRTConfig.DeployEnv != constants.ProductionDeployMode { - engine.GET("/swagger/*any", ginSwagger.WrapHandler(swaggerFiles.Handler)) - } - - server := http.Server{ - Addr: modelRTConfig.ServiceAddr, - Handler: engine, - } - - // creating a system signal receiver - done := make(chan os.Signal, 10) - signal.Notify(done, os.Interrupt, syscall.SIGINT, syscall.SIGTERM) - go func() { - <-done - logger.Info(ctx, "shutdown signal received, cleaning up...") - if err := server.Shutdown(context.Background()); err != nil { - logger.Error(ctx, "shutdown serverError", "err", err) - } - logger.Info(ctx, "resources cleaned up, exiting") - }() - - logger.Info(ctx, "starting ModelRT server") - err = server.ListenAndServe() - if err != nil { - if err == http.ErrServerClosed { - // the service receives the shutdown signal normally and then closes - logger.Info(ctx, "server closed under request") - } else { - // abnormal shutdown of service - logger.Error(ctx, "server closed unexpected", "err", err) - } - } + // blocks until the server is shut down + application.Run(ctx) }