mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-21 06:36:54 +00:00
admin: auto migrating master maintenance scripts to admin_script plugin config (#8509)
* admin: seed admin_script plugin config from master maintenance scripts
When the admin server starts, fetch the maintenance scripts configuration
from the master via GetMasterConfiguration. If the admin_script plugin
worker does not already have a saved config, use the master's scripts as
the default value. This enables seamless migration from master.toml
[master.maintenance] to the admin script plugin worker.
Changes:
- Add maintenance_scripts and maintenance_sleep_minutes fields to
GetMasterConfigurationResponse in master.proto
- Populate the new fields from viper config in master_grpc_server.go
- On admin server startup, fetch the master config and seed the
admin_script plugin config if no config exists yet
- Strip lock/unlock commands from the master scripts since the admin
script worker handles locking automatically
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* fix: address review comments on admin_script seeding
- Replace TOCTOU race (separate Load+Save) with atomic
SaveJobTypeConfigIfNotExists on ConfigStore and Plugin
- Replace ineffective polling loop with single GetMaster call using
30s context timeout, since GetMaster respects context cancellation
- Add unit tests for SaveJobTypeConfigIfNotExists (in-memory + on-disk)
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* fix: apply maintenance script defaults in gRPC handler
The gRPC handler for GetMasterConfiguration read maintenance scripts
from viper without calling SetDefault, relying on startAdminScripts
having run first. If the admin server calls GetMasterConfiguration
before startAdminScripts sets the defaults, viper returns empty
strings and the seeding is silently skipped.
Apply SetDefault in the gRPC handler itself so it is self-contained.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* Revert "fix: apply maintenance script defaults in gRPC handler"
This reverts commit 068a506330.
* fix: use atomic save in ensureJobTypeConfigFromDescriptor
ensureJobTypeConfigFromDescriptor used a separate Load + Save, racing
with seedAdminScriptFromMaster. If the descriptor defaults (empty
script) were saved first, SaveJobTypeConfigIfNotExists in the seeding
goroutine would see an existing config and skip, losing the master's
maintenance scripts.
Switch to SaveJobTypeConfigIfNotExists so both paths are atomic. Whichever
wins, the other is a safe no-op.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* fix: fetch master scripts inline during config bootstrap, not in goroutine
Replace the seedAdminScriptFromMaster goroutine with a
ConfigDefaultsProvider callback. When the plugin bootstraps
admin_script defaults from the worker descriptor, it calls the
provider which fetches maintenance scripts from the master
synchronously. This eliminates the race between the seeding
goroutine and the descriptor-based config bootstrap.
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
* skip commented lock unlock
Co-Authored-By: Copilot <223556219+Copilot@users.noreply.github.com>
* reduce grpc calls
---------
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
Copilot
parent
7799804200
commit
b3620c7e14
@@ -256,6 +256,53 @@ func (s *ConfigStore) SaveJobTypeConfig(config *plugin_pb.PersistedJobTypeConfig
|
||||
return nil
|
||||
}
|
||||
|
||||
// SaveJobTypeConfigIfNotExists atomically checks whether a config for the
|
||||
// given job type already exists and only persists config when none is found.
|
||||
// Returns true if the config was saved, false if a config already existed.
|
||||
func (s *ConfigStore) SaveJobTypeConfigIfNotExists(config *plugin_pb.PersistedJobTypeConfig) (bool, error) {
|
||||
if config == nil {
|
||||
return false, fmt.Errorf("job type config is nil")
|
||||
}
|
||||
if config.JobType == "" {
|
||||
return false, fmt.Errorf("job type config has empty job_type")
|
||||
}
|
||||
sanitizedJobType, err := sanitizeJobType(config.JobType)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
config.JobType = sanitizedJobType
|
||||
|
||||
clone := proto.Clone(config).(*plugin_pb.PersistedJobTypeConfig)
|
||||
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
if !s.configured {
|
||||
if _, exists := s.memConfigs[config.JobType]; exists {
|
||||
return false, nil
|
||||
}
|
||||
s.memConfigs[config.JobType] = clone
|
||||
return true, nil
|
||||
}
|
||||
|
||||
pbPath := filepath.Join(s.baseDir, jobTypesDirName, config.JobType, configPBFileName)
|
||||
if _, statErr := os.Stat(pbPath); statErr == nil {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
jobTypeDir, err := s.ensureJobTypeDir(config.JobType)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
jsonPath := filepath.Join(jobTypeDir, configJSONFileName)
|
||||
if err := writeProtoFiles(clone, filepath.Join(jobTypeDir, configPBFileName), jsonPath); err != nil {
|
||||
return false, fmt.Errorf("save job type config for %s: %w", config.JobType, err)
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (s *ConfigStore) LoadJobTypeConfig(jobType string) (*plugin_pb.PersistedJobTypeConfig, error) {
|
||||
if _, err := sanitizeJobType(jobType); err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -208,6 +208,81 @@ func TestConfigStoreMonitorStateRoundTrip(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigStoreSaveJobTypeConfigIfNotExists(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
t.Run("in-memory", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, err := NewConfigStore("")
|
||||
if err != nil {
|
||||
t.Fatalf("NewConfigStore: %v", err)
|
||||
}
|
||||
testSaveJobTypeConfigIfNotExists(t, store)
|
||||
})
|
||||
|
||||
t.Run("on-disk", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
store, err := NewConfigStore(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatalf("NewConfigStore: %v", err)
|
||||
}
|
||||
testSaveJobTypeConfigIfNotExists(t, store)
|
||||
})
|
||||
}
|
||||
|
||||
func testSaveJobTypeConfigIfNotExists(t *testing.T, store *ConfigStore) {
|
||||
t.Helper()
|
||||
|
||||
cfg := &plugin_pb.PersistedJobTypeConfig{
|
||||
JobType: "admin_script",
|
||||
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: true},
|
||||
}
|
||||
|
||||
// First call should save.
|
||||
saved, err := store.SaveJobTypeConfigIfNotExists(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("first SaveJobTypeConfigIfNotExists: %v", err)
|
||||
}
|
||||
if !saved {
|
||||
t.Fatal("expected first call to save the config")
|
||||
}
|
||||
|
||||
// Second call with same job type should not save.
|
||||
saved, err = store.SaveJobTypeConfigIfNotExists(&plugin_pb.PersistedJobTypeConfig{
|
||||
JobType: "admin_script",
|
||||
AdminRuntime: &plugin_pb.AdminRuntimeConfig{Enabled: false},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("second SaveJobTypeConfigIfNotExists: %v", err)
|
||||
}
|
||||
if saved {
|
||||
t.Fatal("expected second call to be a no-op")
|
||||
}
|
||||
|
||||
// Verify the original config was preserved.
|
||||
loaded, err := store.LoadJobTypeConfig("admin_script")
|
||||
if err != nil {
|
||||
t.Fatalf("LoadJobTypeConfig: %v", err)
|
||||
}
|
||||
if loaded == nil {
|
||||
t.Fatal("expected config to exist")
|
||||
}
|
||||
if !loaded.AdminRuntime.Enabled {
|
||||
t.Fatal("expected original config (Enabled=true) to be preserved")
|
||||
}
|
||||
|
||||
// Different job type should still save.
|
||||
saved, err = store.SaveJobTypeConfigIfNotExists(&plugin_pb.PersistedJobTypeConfig{
|
||||
JobType: "vacuum",
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("SaveJobTypeConfigIfNotExists for different type: %v", err)
|
||||
}
|
||||
if !saved {
|
||||
t.Fatal("expected save for a different job type")
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigStoreJobDetailRoundTrip(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -34,6 +34,11 @@ type Options struct {
|
||||
SchedulerTick time.Duration
|
||||
ClusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error)
|
||||
LockManager LockManager
|
||||
// ConfigDefaultsProvider is an optional callback invoked when a job type's
|
||||
// config is being bootstrapped from its descriptor defaults. It can enrich
|
||||
// or replace the default config before it is persisted. If nil, descriptor
|
||||
// defaults are used as-is.
|
||||
ConfigDefaultsProvider func(config *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig
|
||||
}
|
||||
|
||||
// JobTypeInfo contains metadata about a plugin job type.
|
||||
@@ -54,6 +59,7 @@ type Plugin struct {
|
||||
|
||||
schedulerTick time.Duration
|
||||
clusterContextProvider func(context.Context) (*plugin_pb.ClusterContext, error)
|
||||
configDefaultsProvider func(config *plugin_pb.PersistedJobTypeConfig) *plugin_pb.PersistedJobTypeConfig
|
||||
lockManager LockManager
|
||||
|
||||
schedulerMu sync.Mutex
|
||||
@@ -161,6 +167,7 @@ func New(options Options) (*Plugin, error) {
|
||||
sendTimeout: sendTimeout,
|
||||
schedulerTick: schedulerTick,
|
||||
clusterContextProvider: options.ClusterContextProvider,
|
||||
configDefaultsProvider: options.ConfigDefaultsProvider,
|
||||
lockManager: options.LockManager,
|
||||
sessions: make(map[string]*streamSession),
|
||||
pendingSchema: make(map[string]chan *plugin_pb.ConfigSchemaResponse),
|
||||
@@ -402,6 +409,7 @@ func (r *Plugin) SaveJobTypeConfig(config *plugin_pb.PersistedJobTypeConfig) err
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
func (r *Plugin) LoadDescriptor(jobType string) (*plugin_pb.JobTypeDescriptor, error) {
|
||||
return r.store.LoadDescriptor(jobType)
|
||||
}
|
||||
@@ -1035,14 +1043,6 @@ func (r *Plugin) ensureJobTypeConfigFromDescriptor(jobType string, descriptor *p
|
||||
return nil
|
||||
}
|
||||
|
||||
existing, err := r.store.LoadJobTypeConfig(jobType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if existing != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
workerDefaults := CloneConfigValueMap(descriptor.WorkerDefaultValues)
|
||||
if len(workerDefaults) == 0 && descriptor.WorkerConfigForm != nil {
|
||||
workerDefaults = CloneConfigValueMap(descriptor.WorkerConfigForm.DefaultValues)
|
||||
@@ -1079,7 +1079,22 @@ func (r *Plugin) ensureJobTypeConfigFromDescriptor(jobType string, descriptor *p
|
||||
UpdatedBy: "plugin",
|
||||
}
|
||||
|
||||
return r.store.SaveJobTypeConfig(cfg)
|
||||
// Check existence first to avoid calling configDefaultsProvider unnecessarily
|
||||
// (e.g., it may make a blocking gRPC call to fetch master config).
|
||||
existing, err := r.store.LoadJobTypeConfig(jobType)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if existing != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
if r.configDefaultsProvider != nil {
|
||||
cfg = r.configDefaultsProvider(cfg)
|
||||
}
|
||||
|
||||
_, err = r.store.SaveJobTypeConfigIfNotExists(cfg)
|
||||
return err
|
||||
}
|
||||
|
||||
func (r *Plugin) handleDetectionProposals(workerID string, message *plugin_pb.DetectionProposals) {
|
||||
|
||||
Reference in New Issue
Block a user