yvan/kra-oa
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

优化结构

Browse Source
This commit is contained in:
Yvan
2026-08-24 17:03:46 +08:00
parent acaee94858
commit 7643152c1e
15 changed files with 244 additions and 203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/data/config_store.go
+11
View File
@@ -377,6 +377,14 @@ func (d *Data) reloadConfig(ctx context.Context) error {
_ = candidateRedis.Close()
}
}()
useRedisList := useRedis && next.Admin.System.UseMultipoint
candidateRedisList := openRedisList(next.Data.RedisList, useRedisList, d.logger())
candidateRedisListAccepted := false
defer func() {
if !candidateRedisListAccepted {
closeRedisList(candidateRedisList)
}
}()
useMongo := next.Admin.System != nil && next.Admin.System.UseMongo
candidateMongo, mongoErr := openMongo(next.Data.Mongo, useMongo)
if mongoErr != nil {
@@ -410,6 +418,7 @@ func (d *Data) reloadConfig(ctx context.Context) error {
}
d.replaceDatabaseList(candidateDBList)
d.redis.replace(candidateRedis)
d.replaceRedisList(candidateRedisList)
if mongoErr == nil {
d.mongo.replace(candidateMongo)
mongoAccepted = true
@@ -417,6 +426,7 @@ func (d *Data) reloadConfig(ctx context.Context) error {
closeCandidate = false
candidateDBListAccepted = true
candidateRedisAccepted = true
candidateRedisListAccepted = true
d.runtime.Replace(next)
if d.integrations != nil {
d.integrations.Replace(integrationConfigs)
@@ -424,6 +434,7 @@ func (d *Data) reloadConfig(ctx context.Context) error {
if d.storage != nil {
d.storage.Replace(candidateStorage)
}
d.notifyResources()
return nil
}
internal/data/config_watch.go
-123
View File
@@ -1,123 +0,0 @@
package data
import (
"path/filepath"
"sync"
"time"
"github.com/fsnotify/fsnotify"
"kra/internal/config"
)
// watchConfig refreshes the in-memory config snapshot on file changes, while
// /system/reloadSystem remains responsible
// for rebuilding database, Redis, storage, and scheduled tasks.
func (d *Data) watchConfig() func() {
logger := d.logger()
configPath := d.runtime.ConfigPath()
if configPath == "" {
return func() {}
}
absolute, err := filepath.Abs(configPath)
if err != nil {
logger.Error("resolve config watch path", "mod", "system", "error", err)
return func() {}
}
watcher, err := fsnotify.NewWatcher()
if err != nil {
logger.Error("create config watcher", "mod", "system", "error", err)
return func() {}
}
if err = watcher.Add(filepath.Dir(absolute)); err != nil {
logger.Error("watch config directory", "mod", "system", "error", err)
_ = watcher.Close()
return func() {}
}
done := make(chan struct{})
finished := make(chan struct{})
var once sync.Once
go func() {
defer close(finished)
var timer *time.Timer
var timerC <-chan time.Time
events := watcher.Events
watchErrors := watcher.Errors
for {
select {
case event, ok := <-events:
if !ok {
events = nil
continue
}
if filepath.Clean(event.Name) != absolute || event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename) == 0 {
continue
}
if timer != nil {
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
}
if timer == nil {
timer = time.NewTimer(100 * time.Millisecond)
} else {
timer.Reset(100 * time.Millisecond)
}
timerC = timer.C
case <-timerC:
// The debounce callback runs on this goroutine so shutdown can
// wait for it to finish without racing runtime replacement.
timerC = nil
select {
case <-done:
continue
default:
}
next, loadErr := readBootstrap(absolute)
if loadErr != nil {
logger.Error("reload changed config", "mod", "system", "error", loadErr)
continue
}
if next.Data == nil {
logger.Error("reload changed config: data configuration is required", "mod", "system")
continue
}
if next.Admin == nil {
next.Admin = &config.Admin{}
}
if current := d.runtime.Admin(); current != nil {
next.Admin.Storage = current.Storage
next.Admin.Email = current.Email
}
next.Admin.ConfigPath = absolute
d.runtime.Replace(next)
logger.Info("config file changed", "mod", "system", "path", absolute)
case watchErr, ok := <-watchErrors:
if !ok {
watchErrors = nil
continue
}
logger.Error("config watcher error", "mod", "system", "error", watchErr)
case <-done:
if timer != nil {
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
}
return
}
}
}()
return func() {
once.Do(func() {
close(done)
_ = watcher.Close()
<-finished
})
}
}
internal/data/config_watch_test.go
-59
View File
@@ -1,59 +0,0 @@
package data
import (
"os"
"path/filepath"
"sync"
"testing"
"time"
"kra/internal/config"
)
func TestConfigWatcherStopWaitsForDebouncedReload(t *testing.T) {
root := t.TempDir()
configPath := filepath.Join(root, "config.yaml")
rawConfig := []byte("data:\n database:\n driver: sqlite\n path: " + filepath.ToSlash(root) + "\nadmin: {}\n")
if err := os.WriteFile(configPath, rawConfig, 0o600); err != nil {
t.Fatal(err)
}
runtime := config.NewStore(&config.Config{Data: &config.Data{}, Admin: &config.Admin{ConfigPath: configPath}})
started := make(chan struct{})
release := make(chan struct{})
var startOnce sync.Once
stopListener := runtime.Subscribe(func(*config.Config) {
startOnce.Do(func() { close(started) })
<-release
})
defer stopListener()
data := &Data{runtime: runtime}
stop := data.watchConfig()
// A write event schedules the 100ms debounce reload.
if err := os.WriteFile(configPath, rawConfig, 0o600); err != nil {
t.Fatal(err)
}
select {
case <-started:
case <-time.After(3 * time.Second):
stop()
t.Fatal("debounced reload did not start")
}
stopped := make(chan struct{})
go func() {
stop()
close(stopped)
}()
select {
case <-stopped:
t.Fatal("watcher stop returned before reload callback finished")
case <-time.After(100 * time.Millisecond):
}
close(release)
select {
case <-stopped:
case <-time.After(time.Second):
t.Fatal("watcher stop did not wait for callback completion")
}
}
internal/data/data.go
+120 -5
View File
@@ -10,6 +10,7 @@ import (
"github.com/google/wire"
"github.com/redis/go-redis/v9"
"go.mongodb.org/mongo-driver/mongo"
"gorm.io/gorm"
"kra/internal/config"
dataintegration "kra/internal/data/integration"
@@ -45,9 +46,12 @@ func NewIntegrationRuntime(data *Data) *runtimeconfig.Store {
type Data struct {
initMu sync.Mutex
configMu sync.Mutex
resourceMu sync.RWMutex
databaseReady atomic.Bool
gormDB *reloadableDB
redis *reloadableRedis
redisListMu sync.RWMutex
redisList map[string]redis.UniversalClient
mongo *reloadableMongo
runtime *config.Store
integrations *runtimeconfig.Store
@@ -57,6 +61,7 @@ type Data struct {
appLogger *slog.Logger
auditLog *dataScopeAuditWriter
catalog module.Catalog
resourceHook func()
}
// DB exposes the active primary database to narrowly scoped data submodules.
@@ -105,6 +110,79 @@ func (d *Data) RedisClient() redis.UniversalClient {
return d.redis.load()
}
// MongoClient returns the currently active Mongo client. The client remains
// owned by Data and is only exposed here for framework-level resource wiring.
func (d *Data) MongoClient() *mongo.Client {
if d == nil || d.mongo == nil {
return nil
}
return d.mongo.load()
}
// NamedDatabases returns a copy of configured secondary database handles for
// the process-wide resource registry. Data retains lifecycle ownership.
func (d *Data) NamedDatabases() map[string]*gorm.DB {
if d == nil {
return nil
}
d.dbListMu.RLock()
defer d.dbListMu.RUnlock()
if len(d.dbList) == 0 {
return nil
}
result := make(map[string]*gorm.DB, len(d.dbList))
for name, db := range d.dbList {
if name != "" && db != nil {
result[name] = db
}
}
return result
}
// NamedRedisClients returns a copy of configured secondary Redis handles for
// the process-wide resource registry. Data retains lifecycle ownership.
func (d *Data) NamedRedisClients() map[string]redis.UniversalClient {
if d == nil {
return nil
}
d.redisListMu.RLock()
defer d.redisListMu.RUnlock()
if len(d.redisList) == 0 {
return nil
}
result := make(map[string]redis.UniversalClient, len(d.redisList))
for name, client := range d.redisList {
if name != "" && client != nil {
result[name] = client
}
}
return result
}
// SetResourceHook registers a composition-root callback invoked after active
// database, Redis, or Mongo handles change. A callback keeps data independent
// from the global package while allowing the root to refresh shared handles.
func (d *Data) SetResourceHook(hook func()) {
if d == nil {
return
}
d.resourceMu.Lock()
d.resourceHook = hook
d.resourceMu.Unlock()
}
func (d *Data) notifyResources() {
if d == nil {
return
}
d.resourceMu.RLock()
hook := d.resourceHook
d.resourceMu.RUnlock()
if hook != nil {
hook()
}
}
func (d *Data) logger() *slog.Logger {
if d != nil && d.appLogger != nil {
return d.appLogger
@@ -176,13 +254,9 @@ func NewData(runtime *config.Store, appLogger *slog.Logger, storageManager *stor
c.Database = &config.Database{}
}
d := &Data{runtime: runtime, integrations: runtimeconfig.NewStore(), appLogger: appLogger, storage: storageManager, catalog: catalog}
var stopConfigWatcher func()
var cleanupOnce sync.Once
cleanup := func() {
cleanupOnce.Do(func() {
if stopConfigWatcher != nil {
stopConfigWatcher()
}
if d.auditLog != nil {
d.auditLog.Close()
}
@@ -193,6 +267,7 @@ func NewData(runtime *config.Store, appLogger *slog.Logger, storageManager *stor
if d.redis != nil {
d.redis.close()
}
closeRedisList(d.redisList)
if d.mongo != nil {
d.mongo.close()
}
@@ -275,6 +350,8 @@ func NewData(runtime *config.Store, appLogger *slog.Logger, storageManager *stor
}
useRedis := admin != nil && admin.System != nil && admin.System.UseRedis
d.redis = newReloadableRedis(openRedis(c.Redis, useRedis, appLogger))
useRedisList := useRedis && admin.System.UseMultipoint
d.redisList = openRedisList(c.RedisList, useRedisList, appLogger)
useMongo := admin != nil && admin.System != nil && admin.System.UseMongo
mongoClient, err := openMongo(c.Mongo, useMongo)
if err != nil {
@@ -282,7 +359,6 @@ func NewData(runtime *config.Store, appLogger *slog.Logger, storageManager *stor
mongoClient = nil
}
d.mongo = newReloadableMongo(mongoClient)
stopConfigWatcher = d.watchConfig()
initialized = true
return d, cleanup, nil
}
@@ -322,8 +398,47 @@ func openRedis(config *config.Redis, enabled bool, appLogger ...*slog.Logger) re
return candidate
}
func openRedisList(configs []*config.Redis, enabled bool, appLogger ...*slog.Logger) map[string]redis.UniversalClient {
if !enabled || len(configs) == 0 {
return nil
}
clients := make(map[string]redis.UniversalClient)
for _, item := range configs {
if item == nil || item.Name == "" {
continue
}
if client := openRedis(item, true, appLogger...); client != nil {
clients[item.Name] = client
}
}
if len(clients) == 0 {
return nil
}
return clients
}
func closeRedisList(clients map[string]redis.UniversalClient) {
for _, client := range clients {
if client != nil {
_ = client.Close()
}
}
}
func (d *Data) replaceRedisList(clients map[string]redis.UniversalClient) {
if d == nil {
return
}
d.redisListMu.Lock()
old := d.redisList
d.redisList = clients
d.redisListMu.Unlock()
closeRedisList(old)
}
func (d *Data) activateDatabase(db *gorm.DB, config *config.Database) {
d.gormDB.replace(db, d.enqueueDataScopeAudit)
d.runtime.UpdateDatabase(config)
d.databaseReady.Store(true)
d.notifyResources()
}
internal/data/initialization_backend.go
+2
View File
@@ -105,6 +105,7 @@ func (d *Data) PersistRuntimeConfig(ctx context.Context, value *configpkg.Config
if d.storage != nil {
d.storage.Replace(candidateStorage)
}
d.notifyResources()
return nil
}
func (d *Data) ReloadConfig(ctx context.Context) error {
@@ -212,5 +213,6 @@ func (d *Data) InitializeDatabase(ctx context.Context, input *system.DatabaseCon
if d.integrations != nil {
d.integrations.Replace(integrationConfigs)
}
d.notifyResources()
return nil
}
internal/data/runtime_clients.go
+11
View File
@@ -36,6 +36,17 @@ func (r *reloadableMongo) replace(client *mongo.Client) {
}
r.mu.Unlock()
}
func (r *reloadableMongo) load() *mongo.Client {
if r == nil {
return nil
}
r.mu.RLock()
client := r.current
r.mu.RUnlock()
return client
}
func (r *reloadableMongo) close() {
r.mu.Lock()
all := append([]*mongo.Client{r.current}, r.retired...)