From 7f254e158e49a9b80160e0c9574c6741036d0d7d Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 8 May 2026 10:30:02 -0700 Subject: [PATCH] feat(worker/s3_lifecycle): plugin handler with admin UI config (#9362) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(s3/lifecycle): scheduler — N pipelines over an even shard split Scheduler.Run spawns Workers Pipeline goroutines plus one engine-refresh ticker. Each worker owns a contiguous AssignShards(idx, total) slice of [0, ShardCount) and runs Pipeline.Run with EventBudget bounding each iteration; brief RetryBackoff between iterations avoids hot-loop on errors. The refresh ticker rebuilds the engine snapshot from the filer's bucket configs every RefreshInterval. LoadCompileInputs / IsBucketVersioned / AllActivePriorStates are exported from a configload.go sibling so the shell command can move to this shared implementation in a follow-up. * refactor(shell): reuse scheduler.LoadCompileInputs in run-shard Drop the local copies of loadLifecycleCompileInputs / isBucketVersioned / allActivePriorStates / lifecycleParseError that the new scheduler package now exports. Same behavior, one source of truth. * feat(worker/s3_lifecycle): plugin handler with admin UI config Registers a JobHandler for s3_lifecycle via pluginworker.RegisterHandler. Admin pulls the descriptor over the worker plugin gRPC and renders the AdminConfigForm + WorkerConfigForm in the existing UI: Admin form (cluster shape): - workers (1..16, default 1) - s3_grpc_endpoints (comma list) Worker form (operational tuning): - dispatch_tick_ms (default 5000) - checkpoint_tick_ms (default 30000) - refresh_interval_ms (default 300000) - event_budget (default 0 = unbounded) Detect emits a single proposal whenever S3 endpoints + filer addresses are configured. MaxExecutionConcurrency=1 so admin only ever runs one lifecycle daemon per worker; a fresh proposal next cycle restarts it if the prior Execute exits. Execute dials the configured S3 endpoint + filer, builds a scheduler.Scheduler with the parsed config, and runs it until ctx cancellation. Reuses the existing scheduler / dispatcher / reader / engine packages — the handler is the thin glue that parses descriptor values and wires the long-running daemon. * proto(plugin): add s3_grpc_addresses to ClusterContext So workers can dial s3 servers discovered by the master rather than a hand-typed list in the admin form. * feat(admin): populate ClusterContext.s3_grpc_addresses from master ListClusterNodes(S3Type) returns the live S3 servers; the plugin scheduler now hands these to job handlers alongside filer/volume addresses. * feat(worker/s3_lifecycle): discover s3 endpoints from cluster context Drop the s3_grpc_endpoints admin form field and read the master-supplied ClusterContext.S3GrpcAddresses instead. Operators no longer maintain a hand-typed list, and a stale entry self-heals when the master's view updates. * feat(worker/s3_lifecycle): time-based runtime cap, friendlier cadence units - dispatch_tick_minutes (was *_ms): minutes is the natural granularity for a daily batch; default 1 minute. - checkpoint_tick_seconds: seconds for the durable cursor write; default 30 seconds. - refresh_interval_minutes: minutes for the engine snapshot rebuild. - max_runtime_minutes replaces event_budget. Each daily run is bounded by wall clock — typical run wraps in well under an hour because the cursor persists and the meta-log streams fast. Default 60 minutes. - AdminRuntimeDefaults.DetectionIntervalSeconds = 86400 so the admin schedules one job per day. --- weed/admin/dash/plugin_api.go | 31 ++ weed/pb/plugin.proto | 1 + weed/pb/plugin_pb/plugin.pb.go | 13 +- weed/plugin/worker/handlers/handlers.go | 1 + .../s3api/s3lifecycle/scheduler/configload.go | 103 ++++++ weed/s3api/s3lifecycle/scheduler/scheduler.go | 166 ++++++++++ .../s3lifecycle/scheduler/scheduler_test.go | 86 +++++ weed/shell/command_s3_lifecycle_run_shard.go | 101 +----- weed/worker/tasks/s3_lifecycle/config.go | 70 ++++ weed/worker/tasks/s3_lifecycle/config_test.go | 77 +++++ weed/worker/tasks/s3_lifecycle/handler.go | 308 ++++++++++++++++++ 11 files changed, 858 insertions(+), 99 deletions(-) create mode 100644 weed/s3api/s3lifecycle/scheduler/configload.go create mode 100644 weed/s3api/s3lifecycle/scheduler/scheduler.go create mode 100644 weed/s3api/s3lifecycle/scheduler/scheduler_test.go create mode 100644 weed/worker/tasks/s3_lifecycle/config.go create mode 100644 weed/worker/tasks/s3_lifecycle/config_test.go create mode 100644 weed/worker/tasks/s3_lifecycle/handler.go diff --git a/weed/admin/dash/plugin_api.go b/weed/admin/dash/plugin_api.go index 53a30508a..c3f936723 100644 --- a/weed/admin/dash/plugin_api.go +++ b/weed/admin/dash/plugin_api.go @@ -14,7 +14,9 @@ import ( "github.com/gorilla/mux" "github.com/seaweedfs/seaweedfs/weed/admin/plugin" + "github.com/seaweedfs/seaweedfs/weed/cluster" "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" @@ -738,6 +740,9 @@ func (s *AdminServer) parseOrBuildClusterContext(raw json.RawMessage) (*plugin_p if len(contextMessage.VolumeGrpcAddresses) == 0 { contextMessage.VolumeGrpcAddresses = append(contextMessage.VolumeGrpcAddresses, fallback.VolumeGrpcAddresses...) } + if len(contextMessage.S3GrpcAddresses) == 0 { + contextMessage.S3GrpcAddresses = append(contextMessage.S3GrpcAddresses, fallback.S3GrpcAddresses...) + } if contextMessage.Metadata == nil { contextMessage.Metadata = map[string]string{} } @@ -751,6 +756,7 @@ func (s *AdminServer) buildDefaultPluginClusterContext() *plugin_pb.ClusterConte MasterGrpcAddresses: make([]string, 0), FilerGrpcAddresses: make([]string, 0), VolumeGrpcAddresses: make([]string, 0), + S3GrpcAddresses: make([]string, 0), Metadata: map[string]string{ "source": "admin", }, @@ -794,9 +800,34 @@ func (s *AdminServer) buildDefaultPluginClusterContext() *plugin_pb.ClusterConte glog.V(1).Infof("failed to build default plugin volume context: %v", err) } + s3Seen := map[string]struct{}{} + if err := s.WithMasterClient(func(client master_pb.SeaweedClient) error { + resp, err := client.ListClusterNodes(context.Background(), &master_pb.ListClusterNodesRequest{ + ClientType: cluster.S3Type, + }) + if err != nil { + return err + } + for _, node := range resp.GetClusterNodes() { + address := strings.TrimSpace(node.GetAddress()) + if address == "" { + continue + } + if _, exists := s3Seen[address]; exists { + continue + } + s3Seen[address] = struct{}{} + clusterContext.S3GrpcAddresses = append(clusterContext.S3GrpcAddresses, address) + } + return nil + }); err != nil { + glog.V(1).Infof("failed to list cluster S3 nodes: %v", err) + } + sort.Strings(clusterContext.MasterGrpcAddresses) sort.Strings(clusterContext.FilerGrpcAddresses) sort.Strings(clusterContext.VolumeGrpcAddresses) + sort.Strings(clusterContext.S3GrpcAddresses) return clusterContext } diff --git a/weed/pb/plugin.proto b/weed/pb/plugin.proto index 1647a6dcf..8bf5f10c4 100644 --- a/weed/pb/plugin.proto +++ b/weed/pb/plugin.proto @@ -351,6 +351,7 @@ message ClusterContext { repeated string filer_grpc_addresses = 2; repeated string volume_grpc_addresses = 3; map metadata = 4; + repeated string s3_grpc_addresses = 5; } message ActivityEvent { diff --git a/weed/pb/plugin_pb/plugin.pb.go b/weed/pb/plugin_pb/plugin.pb.go index 555837e71..e0f08bc84 100644 --- a/weed/pb/plugin_pb/plugin.pb.go +++ b/weed/pb/plugin_pb/plugin.pb.go @@ -3567,6 +3567,7 @@ type ClusterContext struct { FilerGrpcAddresses []string `protobuf:"bytes,2,rep,name=filer_grpc_addresses,json=filerGrpcAddresses,proto3" json:"filer_grpc_addresses,omitempty"` VolumeGrpcAddresses []string `protobuf:"bytes,3,rep,name=volume_grpc_addresses,json=volumeGrpcAddresses,proto3" json:"volume_grpc_addresses,omitempty"` Metadata map[string]string `protobuf:"bytes,4,rep,name=metadata,proto3" json:"metadata,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + S3GrpcAddresses []string `protobuf:"bytes,5,rep,name=s3_grpc_addresses,json=s3GrpcAddresses,proto3" json:"s3_grpc_addresses,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -3629,6 +3630,13 @@ func (x *ClusterContext) GetMetadata() map[string]string { return nil } +func (x *ClusterContext) GetS3GrpcAddresses() []string { + if x != nil { + return x.S3GrpcAddresses + } + return nil +} + type ActivityEvent struct { state protoimpl.MessageState `protogen:"open.v1"` Source ActivitySource `protobuf:"varint,1,opt,name=source,proto3,enum=plugin.ActivitySource" json:"source,omitempty"` @@ -4265,12 +4273,13 @@ const file_plugin_proto_rawDesc = "" + "\asummary\x18\x02 \x01(\tR\asummary\x1aT\n" + "\x11OutputValuesEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12)\n" + - "\x05value\x18\x02 \x01(\v2\x13.plugin.ConfigValueR\x05value:\x028\x01\"\xa9\x02\n" + + "\x05value\x18\x02 \x01(\v2\x13.plugin.ConfigValueR\x05value:\x028\x01\"\xd5\x02\n" + "\x0eClusterContext\x122\n" + "\x15master_grpc_addresses\x18\x01 \x03(\tR\x13masterGrpcAddresses\x120\n" + "\x14filer_grpc_addresses\x18\x02 \x03(\tR\x12filerGrpcAddresses\x122\n" + "\x15volume_grpc_addresses\x18\x03 \x03(\tR\x13volumeGrpcAddresses\x12@\n" + - "\bmetadata\x18\x04 \x03(\v2$.plugin.ClusterContext.MetadataEntryR\bmetadata\x1a;\n" + + "\bmetadata\x18\x04 \x03(\v2$.plugin.ClusterContext.MetadataEntryR\bmetadata\x12*\n" + + "\x11s3_grpc_addresses\x18\x05 \x03(\tR\x0fs3GrpcAddresses\x1a;\n" + "\rMetadataEntry\x12\x10\n" + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xb9\x02\n" + diff --git a/weed/plugin/worker/handlers/handlers.go b/weed/plugin/worker/handlers/handlers.go index c100b4898..d96792e22 100644 --- a/weed/plugin/worker/handlers/handlers.go +++ b/weed/plugin/worker/handlers/handlers.go @@ -9,5 +9,6 @@ import ( _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/ec_balance" // register ec_balance handler _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/erasure_coding" // register erasure_coding handler _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/iceberg" // register iceberg_maintenance handler + _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/s3_lifecycle" // register s3_lifecycle handler _ "github.com/seaweedfs/seaweedfs/weed/worker/tasks/vacuum" // register vacuum handler ) diff --git a/weed/s3api/s3lifecycle/scheduler/configload.go b/weed/s3api/s3lifecycle/scheduler/configload.go new file mode 100644 index 000000000..571b87b5b --- /dev/null +++ b/weed/s3api/s3lifecycle/scheduler/configload.go @@ -0,0 +1,103 @@ +package scheduler + +import ( + "context" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" +) + +const BucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml" + +// ParseError is returned alongside successfully-loaded inputs so callers can +// surface malformed bucket configs rather than silently dropping them. +type ParseError struct { + Bucket string + Err error +} + +// LoadCompileInputs walks the buckets directory and returns one +// engine.CompileInput per bucket that carries a non-empty lifecycle +// configuration. Pagination loops with startFrom so clusters with more +// than one page of buckets don't drop the tail. +func LoadCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []ParseError, error) { + var ( + inputs []engine.CompileInput + parseErrors []ParseError + startFrom string + ) + const pageSize uint32 = 1024 + for { + pageCount := 0 + var lastName string + err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error { + pageCount++ + lastName = entry.Name + if !entry.IsDirectory { + return nil + } + xmlBytes, ok := entry.Extended[BucketLifecycleConfigurationXMLKey] + if !ok || len(xmlBytes) == 0 { + return nil + } + rules, err := lifecycle_xml.ParseCanonical(xmlBytes) + if err != nil { + parseErrors = append(parseErrors, ParseError{Bucket: entry.Name, Err: err}) + return nil + } + if len(rules) == 0 { + return nil + } + inputs = append(inputs, engine.CompileInput{ + Bucket: entry.Name, + Rules: rules, + Versioned: IsBucketVersioned(entry), + }) + return nil + }, startFrom, false, pageSize) + if err != nil { + return nil, nil, err + } + if uint32(pageCount) < pageSize { + break + } + startFrom = lastName + } + return inputs, parseErrors, nil +} + +// IsBucketVersioned returns true when the bucket's filer entry has the +// versioning extended attribute set to Enabled or Suspended. +func IsBucketVersioned(entry *filer_pb.Entry) bool { + v, ok := entry.Extended[s3_constants.ExtVersioningKey] + if !ok { + return false + } + s := strings.ToLower(strings.TrimSpace(string(v))) + return s == "enabled" || s == "suspended" +} + +// AllActivePriorStates seeds every compiled action as bootstrap-complete + +// event-driven so the scheduler dispatches whatever fires immediately. +// Production bootstrap walks set this incrementally per bucket; the +// scheduler runs out-of-band of that flow for now. +func AllActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState { + prior := map[s3lifecycle.ActionKey]engine.PriorState{} + for _, in := range inputs { + for _, rule := range in.Rules { + hash := s3lifecycle.RuleHash(rule) + for _, kind := range s3lifecycle.RuleActionKinds(rule) { + key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind} + prior[key] = engine.PriorState{ + BootstrapComplete: true, + Mode: engine.ModeEventDriven, + } + } + } + } + return prior +} diff --git a/weed/s3api/s3lifecycle/scheduler/scheduler.go b/weed/s3api/s3lifecycle/scheduler/scheduler.go new file mode 100644 index 000000000..803a75c1c --- /dev/null +++ b/weed/s3api/s3lifecycle/scheduler/scheduler.go @@ -0,0 +1,166 @@ +package scheduler + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/reader" +) + +const ( + defaultRefreshInterval = 5 * time.Minute + defaultRetryBackoff = 5 * time.Second +) + +// Scheduler runs N pipeline goroutines, one per worker, each owning a +// contiguous shard slice of [0, ShardCount). It periodically rebuilds the +// engine snapshot from the filer's bucket configs. +type Scheduler struct { + BucketsPath string + Engine *engine.Engine + Persister reader.Persister + Client dispatcher.LifecycleClient + FilerClient filer_pb.SeaweedFilerClient + ClientID int32 + ClientName string + + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + RetryBackoff time.Duration +} + +// Run blocks until ctx is canceled. Spawns Workers + 1 goroutines: one +// engine-refresh ticker plus one Pipeline runner per worker. +func (s *Scheduler) Run(ctx context.Context) error { + if s.Engine == nil || s.Persister == nil || s.Client == nil || s.FilerClient == nil { + return errors.New("scheduler: missing required dependency") + } + if s.BucketsPath == "" { + return errors.New("scheduler: BucketsPath required") + } + + workers := s.Workers + if workers <= 0 { + workers = 1 + } + refresh := s.RefreshInterval + if refresh <= 0 { + refresh = defaultRefreshInterval + } + retry := s.RetryBackoff + if retry <= 0 { + retry = defaultRetryBackoff + } + + s.refreshEngine(ctx) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + t := time.NewTicker(refresh) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + s.refreshEngine(ctx) + } + } + }() + + for i := 0; i < workers; i++ { + shards := AssignShards(i, workers) + if len(shards) == 0 { + continue + } + wg.Add(1) + go func(idx int, shardSet []int) { + defer wg.Done() + for { + select { + case <-ctx.Done(): + return + default: + } + p := &dispatcher.Pipeline{ + Shards: shardSet, + BucketsPath: s.BucketsPath, + Engine: s.Engine, + Persister: s.Persister, + Client: s.Client, + FilerClient: s.FilerClient, + ClientID: s.ClientID, + ClientName: fmt.Sprintf("%s-w%02d", s.ClientName, idx), + DispatchTick: s.DispatchTick, + CheckpointTick: s.CheckpointTick, + } + if err := p.Run(ctx); err != nil && !errors.Is(err, context.Canceled) { + glog.Warningf("lifecycle scheduler: worker=%d shards=%v: %v", idx, shardSet, err) + } + select { + case <-ctx.Done(): + return + case <-time.After(retry): + } + } + }(i, shards) + } + + wg.Wait() + return nil +} + +func (s *Scheduler) refreshEngine(ctx context.Context) { + inputs, parseErrors, err := LoadCompileInputs(ctx, s.FilerClient, s.BucketsPath) + if err != nil { + glog.Warningf("lifecycle scheduler: refresh engine: %v", err) + return + } + for _, pe := range parseErrors { + glog.Warningf("lifecycle scheduler: malformed config in bucket %s: %v", pe.Bucket, pe.Err) + } + s.Engine.Compile(inputs, engine.CompileOptions{PriorStates: AllActivePriorStates(inputs)}) +} + +// AssignShards returns the contiguous shard slice for worker idx of total. +// Shards distribute as evenly as possible: with ShardCount=16 and total=3, +// workers receive [0..5], [6..10], [11..15]. +func AssignShards(idx, total int) []int { + if total <= 0 || idx < 0 || idx >= total { + return nil + } + shardCount := s3lifecycle.ShardCount + base := shardCount / total + extra := shardCount % total + var lo, hi int + if idx < extra { + lo = idx * (base + 1) + hi = lo + base + 1 + } else { + lo = idx*base + extra + hi = lo + base + } + if hi > shardCount { + hi = shardCount + } + if lo >= hi { + return nil + } + out := make([]int, 0, hi-lo) + for i := lo; i < hi; i++ { + out = append(out, i) + } + return out +} diff --git a/weed/s3api/s3lifecycle/scheduler/scheduler_test.go b/weed/s3api/s3lifecycle/scheduler/scheduler_test.go new file mode 100644 index 000000000..d973f52b3 --- /dev/null +++ b/weed/s3api/s3lifecycle/scheduler/scheduler_test.go @@ -0,0 +1,86 @@ +package scheduler + +import ( + "reflect" + "testing" +) + +func TestAssignShardsSingleWorker(t *testing.T) { + got := AssignShards(0, 1) + want := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15} + if !reflect.DeepEqual(got, want) { + t.Fatalf("got %v, want %v", got, want) + } +} + +func TestAssignShardsEvenSplit(t *testing.T) { + a := AssignShards(0, 2) + b := AssignShards(1, 2) + wantA := []int{0, 1, 2, 3, 4, 5, 6, 7} + wantB := []int{8, 9, 10, 11, 12, 13, 14, 15} + if !reflect.DeepEqual(a, wantA) { + t.Fatalf("worker 0: got %v, want %v", a, wantA) + } + if !reflect.DeepEqual(b, wantB) { + t.Fatalf("worker 1: got %v, want %v", b, wantB) + } +} + +func TestAssignShardsUnevenSplit(t *testing.T) { + // 16 shards / 3 workers = 5,5,5 + 1 remainder. Earlier workers absorb + // the remainder so the slices stay contiguous: 6,5,5. + w0 := AssignShards(0, 3) + w1 := AssignShards(1, 3) + w2 := AssignShards(2, 3) + if len(w0) != 6 || w0[0] != 0 || w0[len(w0)-1] != 5 { + t.Fatalf("w0 expected [0..5], got %v", w0) + } + if len(w1) != 5 || w1[0] != 6 || w1[len(w1)-1] != 10 { + t.Fatalf("w1 expected [6..10], got %v", w1) + } + if len(w2) != 5 || w2[0] != 11 || w2[len(w2)-1] != 15 { + t.Fatalf("w2 expected [11..15], got %v", w2) + } + // All shards covered, exactly once. + covered := map[int]int{} + for _, w := range [][]int{w0, w1, w2} { + for _, s := range w { + covered[s]++ + } + } + if len(covered) != 16 { + t.Fatalf("expected 16 unique shards, got %d", len(covered)) + } + for s, c := range covered { + if c != 1 { + t.Fatalf("shard %d covered %d times, expected 1", s, c) + } + } +} + +func TestAssignShardsMoreWorkersThanShards(t *testing.T) { + // 20 workers, 16 shards: first 16 get one shard each, last 4 get nothing. + for i := 0; i < 16; i++ { + got := AssignShards(i, 20) + if len(got) != 1 || got[0] != i { + t.Fatalf("worker %d: got %v, want [%d]", i, got, i) + } + } + for i := 16; i < 20; i++ { + if got := AssignShards(i, 20); got != nil { + t.Fatalf("worker %d should get nil, got %v", i, got) + } + } +} + +func TestAssignShardsBounds(t *testing.T) { + if got := AssignShards(-1, 4); got != nil { + t.Fatalf("idx -1: got %v, want nil", got) + } + if got := AssignShards(4, 4); got != nil { + t.Fatalf("idx==total: got %v, want nil", got) + } + if got := AssignShards(0, 0); got != nil { + t.Fatalf("total 0: got %v, want nil", got) + } +} diff --git a/weed/shell/command_s3_lifecycle_run_shard.go b/weed/shell/command_s3_lifecycle_run_shard.go index 334491eb0..a39eb61e7 100644 --- a/weed/shell/command_s3_lifecycle_run_shard.go +++ b/weed/shell/command_s3_lifecycle_run_shard.go @@ -13,11 +13,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" - "github.com/seaweedfs/seaweedfs/weed/s3api/lifecycle_xml" - "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" "github.com/seaweedfs/seaweedfs/weed/util" ) @@ -102,16 +101,13 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i // Run the whole pipeline inside one WithFilerClient so the reader's // SubscribeMetadata stream and the persister share a single connection. return env.WithFilerClient(true, func(filerClient filer_pb.SeaweedFilerClient) error { - inputs, parseErrors, err := loadLifecycleCompileInputs(context.Background(), filerClient, bucketsPath) + inputs, parseErrors, err := scheduler.LoadCompileInputs(context.Background(), filerClient, bucketsPath) if err != nil { return fmt.Errorf("load lifecycle configs: %w", err) } for i, pe := range parseErrors { - // Surface up to the first three parse errors so the operator - // can chase malformed configs; cap the rest with a count so - // the output stays readable on large clusters. if i < 3 { - fmt.Fprintf(writer, "warning: %s: %v\n", pe.bucket, pe.err) + fmt.Fprintf(writer, "warning: %s: %v\n", pe.Bucket, pe.Err) } } if extra := len(parseErrors) - 3; extra > 0 { @@ -123,11 +119,8 @@ func (c *commandS3LifecycleRunShard) Do(args []string, env *CommandEnv, writer i } fmt.Fprintf(writer, "loaded lifecycle for %d bucket(s)\n", len(inputs)) - // Activate every action so this manual run dispatches whatever fires. - // The production bootstrap walker promotes actions only after a clean - // walk; this shell entrypoint runs out-of-band of that flow. eng := engine.New() - eng.Compile(inputs, engine.CompileOptions{PriorStates: allActivePriorStates(inputs)}) + eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)}) pipeline := &dispatcher.Pipeline{ Shards: shards, @@ -288,89 +281,3 @@ func resolveBucketsPath(env *CommandEnv) (string, error) { return path, nil } -type lifecycleParseError struct { - bucket string - err error -} - -// loadLifecycleCompileInputs walks the buckets directory and reads each -// bucket entry's lifecycle XML from its Extended attributes. Pagination -// loops with startFrom so clusters with more than one page of buckets -// don't drop the tail. Parse errors are collected per bucket and returned -// alongside the successful inputs so the caller can surface them. -func loadLifecycleCompileInputs(ctx context.Context, client filer_pb.SeaweedFilerClient, bucketsPath string) ([]engine.CompileInput, []lifecycleParseError, error) { - var ( - inputs []engine.CompileInput - parseErrors []lifecycleParseError - startFrom string - ) - const pageSize uint32 = 1024 - for { - pageCount := 0 - var lastName string - err := filer_pb.SeaweedList(ctx, client, bucketsPath, "", func(entry *filer_pb.Entry, isLast bool) error { - pageCount++ - lastName = entry.Name - if !entry.IsDirectory { - return nil - } - xmlBytes, ok := entry.Extended[bucketLifecycleConfigurationXMLKey] - if !ok || len(xmlBytes) == 0 { - return nil - } - rules, err := lifecycle_xml.ParseCanonical(xmlBytes) - if err != nil { - parseErrors = append(parseErrors, lifecycleParseError{bucket: entry.Name, err: err}) - return nil - } - if len(rules) == 0 { - return nil - } - inputs = append(inputs, engine.CompileInput{ - Bucket: entry.Name, - Rules: rules, - Versioned: isBucketVersioned(entry), - }) - return nil - }, startFrom, false, pageSize) - if err != nil { - return nil, nil, err - } - if uint32(pageCount) < pageSize { - break - } - startFrom = lastName - } - return inputs, parseErrors, nil -} - -const bucketLifecycleConfigurationXMLKey = "s3-bucket-lifecycle-configuration-xml" - -func isBucketVersioned(entry *filer_pb.Entry) bool { - v, ok := entry.Extended[s3_constants.ExtVersioningKey] - if !ok { - return false - } - s := strings.ToLower(strings.TrimSpace(string(v))) - return s == "enabled" || s == "suspended" -} - -// allActivePriorStates seeds every compiled action as bootstrap-complete + -// event-driven so the run dispatches whatever fires. Production bootstrap -// walks set this incrementally per bucket; this manual run skips the walk. -func allActivePriorStates(inputs []engine.CompileInput) map[s3lifecycle.ActionKey]engine.PriorState { - prior := map[s3lifecycle.ActionKey]engine.PriorState{} - for _, in := range inputs { - for _, rule := range in.Rules { - hash := s3lifecycle.RuleHash(rule) - for _, kind := range s3lifecycle.RuleActionKinds(rule) { - key := s3lifecycle.ActionKey{Bucket: in.Bucket, RuleHash: hash, ActionKind: kind} - prior[key] = engine.PriorState{ - BootstrapComplete: true, - Mode: engine.ModeEventDriven, - } - } - } - } - return prior -} diff --git a/weed/worker/tasks/s3_lifecycle/config.go b/weed/worker/tasks/s3_lifecycle/config.go new file mode 100644 index 000000000..07d8288f4 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/config.go @@ -0,0 +1,70 @@ +package s3_lifecycle + +import ( + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +const ( + jobType = "s3_lifecycle" + + defaultWorkers = 1 + defaultDispatchTickMinutes = int64(1) + defaultCheckpointTickSeconds = int64(30) + defaultRefreshIntervalMinutes = int64(5) + defaultMaxRuntimeMinutes = int64(60) +) + +// Config is the parsed AdminConfigForm + WorkerConfigForm view. +type Config struct { + Workers int + DispatchTick time.Duration + CheckpointTick time.Duration + RefreshInterval time.Duration + MaxRuntime time.Duration +} + +// ParseConfig pulls the lifecycle Handler config from the merged +// admin+worker config values. Missing fields fall back to defaults. +func ParseConfig(adminValues, workerValues map[string]*plugin_pb.ConfigValue) Config { + cfg := Config{ + Workers: int(readInt64(adminValues, "workers", defaultWorkers)), + DispatchTick: time.Duration(readInt64(workerValues, "dispatch_tick_minutes", defaultDispatchTickMinutes)) * time.Minute, + CheckpointTick: time.Duration(readInt64(workerValues, "checkpoint_tick_seconds", defaultCheckpointTickSeconds)) * time.Second, + RefreshInterval: time.Duration(readInt64(workerValues, "refresh_interval_minutes", defaultRefreshIntervalMinutes)) * time.Minute, + MaxRuntime: time.Duration(readInt64(workerValues, "max_runtime_minutes", defaultMaxRuntimeMinutes)) * time.Minute, + } + if cfg.Workers <= 0 { + cfg.Workers = defaultWorkers + } + if cfg.DispatchTick <= 0 { + cfg.DispatchTick = time.Duration(defaultDispatchTickMinutes) * time.Minute + } + if cfg.CheckpointTick <= 0 { + cfg.CheckpointTick = time.Duration(defaultCheckpointTickSeconds) * time.Second + } + if cfg.RefreshInterval <= 0 { + cfg.RefreshInterval = time.Duration(defaultRefreshIntervalMinutes) * time.Minute + } + if cfg.MaxRuntime <= 0 { + cfg.MaxRuntime = time.Duration(defaultMaxRuntimeMinutes) * time.Minute + } + return cfg +} + +func readInt64(values map[string]*plugin_pb.ConfigValue, field string, fallback int64) int64 { + v, ok := values[field] + if !ok || v == nil { + return fallback + } + switch k := v.Kind.(type) { + case *plugin_pb.ConfigValue_Int64Value: + return k.Int64Value + case *plugin_pb.ConfigValue_DoubleValue: + return int64(k.DoubleValue) + case *plugin_pb.ConfigValue_StringValue: + return fallback + } + return fallback +} diff --git a/weed/worker/tasks/s3_lifecycle/config_test.go b/weed/worker/tasks/s3_lifecycle/config_test.go new file mode 100644 index 000000000..d2afae189 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/config_test.go @@ -0,0 +1,77 @@ +package s3_lifecycle + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" +) + +func TestParseConfigDefaults(t *testing.T) { + cfg := ParseConfig(nil, nil) + if cfg.Workers != defaultWorkers { + t.Errorf("Workers default=%d, want %d", cfg.Workers, defaultWorkers) + } + if cfg.DispatchTick != 1*time.Minute { + t.Errorf("DispatchTick default=%v, want 1m", cfg.DispatchTick) + } + if cfg.CheckpointTick != 30*time.Second { + t.Errorf("CheckpointTick default=%v, want 30s", cfg.CheckpointTick) + } + if cfg.RefreshInterval != 5*time.Minute { + t.Errorf("RefreshInterval default=%v, want 5m", cfg.RefreshInterval) + } + if cfg.MaxRuntime != 60*time.Minute { + t.Errorf("MaxRuntime default=%v, want 60m", cfg.MaxRuntime) + } +} + +func TestParseConfigOverrides(t *testing.T) { + admin := map[string]*plugin_pb.ConfigValue{ + "workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 4}}, + } + worker := map[string]*plugin_pb.ConfigValue{ + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 2}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 15}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 10}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 120}}, + } + cfg := ParseConfig(admin, worker) + if cfg.Workers != 4 { + t.Errorf("Workers=%d, want 4", cfg.Workers) + } + if cfg.DispatchTick != 2*time.Minute { + t.Errorf("DispatchTick=%v, want 2m", cfg.DispatchTick) + } + if cfg.CheckpointTick != 15*time.Second { + t.Errorf("CheckpointTick=%v, want 15s", cfg.CheckpointTick) + } + if cfg.RefreshInterval != 10*time.Minute { + t.Errorf("RefreshInterval=%v, want 10m", cfg.RefreshInterval) + } + if cfg.MaxRuntime != 120*time.Minute { + t.Errorf("MaxRuntime=%v, want 120m", cfg.MaxRuntime) + } +} + +func TestParseConfigClampsZeroAndNegative(t *testing.T) { + worker := map[string]*plugin_pb.ConfigValue{ + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: -5}}, + } + cfg := ParseConfig(nil, worker) + if cfg.DispatchTick != 1*time.Minute { + t.Errorf("zero DispatchTick should clamp to default, got %v", cfg.DispatchTick) + } + if cfg.CheckpointTick != 30*time.Second { + t.Errorf("zero CheckpointTick should clamp to default, got %v", cfg.CheckpointTick) + } + if cfg.RefreshInterval != 5*time.Minute { + t.Errorf("zero RefreshInterval should clamp to default, got %v", cfg.RefreshInterval) + } + if cfg.MaxRuntime != 60*time.Minute { + t.Errorf("negative MaxRuntime should clamp to default, got %v", cfg.MaxRuntime) + } +} diff --git a/weed/worker/tasks/s3_lifecycle/handler.go b/weed/worker/tasks/s3_lifecycle/handler.go new file mode 100644 index 000000000..4e6f81e60 --- /dev/null +++ b/weed/worker/tasks/s3_lifecycle/handler.go @@ -0,0 +1,308 @@ +package s3_lifecycle + +import ( + "context" + "fmt" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/s3_lifecycle_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dispatcher" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/engine" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/scheduler" + "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc" +) + +func init() { + pluginworker.RegisterHandler(pluginworker.HandlerFactory{ + JobType: jobType, + Category: pluginworker.CategoryDefault, + Aliases: []string{"s3-lifecycle", "s3.lifecycle", "lifecycle"}, + Build: func(opts pluginworker.HandlerBuildOptions) (pluginworker.JobHandler, error) { + return NewHandler(opts.GrpcDialOption), nil + }, + }) +} + +// Handler is the worker-side runner for S3 object lifecycle expiration. +// One Execute call drives a long-running scheduler.Scheduler against the +// S3 endpoints discovered from the master; admin caps concurrency at one +// job per worker so a fresh proposal only spawns a new run after the +// prior one exits. +type Handler struct { + grpcDialOption grpc.DialOption +} + +func NewHandler(grpcDialOption grpc.DialOption) *Handler { + return &Handler{grpcDialOption: grpcDialOption} +} + +func (h *Handler) Capability() *plugin_pb.JobTypeCapability { + return &plugin_pb.JobTypeCapability{ + JobType: jobType, + CanDetect: true, + CanExecute: true, + MaxDetectionConcurrency: 1, + MaxExecutionConcurrency: 1, + DisplayName: "S3 Lifecycle", + Description: "Daily batch: scan the filer meta-log and delete objects whose lifecycle rule has fired.", + Weight: 20, + } +} + +func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor { + return &plugin_pb.JobTypeDescriptor{ + JobType: jobType, + DisplayName: "S3 Lifecycle", + Description: "Daily S3 object expiration scan with per-shard cursors and bounded retry.", + Icon: "fas fa-recycle", + DescriptorVersion: 1, + AdminConfigForm: &plugin_pb.ConfigForm{ + FormId: "s3-lifecycle-admin", + Title: "S3 Lifecycle", + Description: "Cluster-wide controls for the lifecycle scheduler.", + Sections: []*plugin_pb.ConfigSection{ + { + SectionId: "scope", + Title: "Scope", + Description: "How many pipeline goroutines split the 16-shard space.", + Fields: []*plugin_pb.ConfigField{ + { + Name: "workers", + Label: "Worker Count", + Description: "Number of pipeline goroutines per executing worker. Each owns a contiguous slice of [0, 16) shards. Default 1 = one goroutine handles all 16 shards.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, + MaxValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 16}}, + }, + }, + }, + }, + DefaultValues: map[string]*plugin_pb.ConfigValue{ + "workers": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultWorkers}}, + }, + }, + WorkerConfigForm: &plugin_pb.ConfigForm{ + FormId: "s3-lifecycle-worker", + Title: "S3 Lifecycle Worker", + Description: "Operational tuning for the lifecycle pipeline.", + Sections: []*plugin_pb.ConfigSection{ + { + SectionId: "cadence", + Title: "Cadence", + Description: "Tick intervals for the dispatch loop, durable checkpoint, and engine refresh; plus the wall-clock cap on each daily run.", + Fields: []*plugin_pb.ConfigField{ + { + Name: "dispatch_tick_minutes", + Label: "Dispatch Tick (minutes)", + Description: "How often each pipeline drains its schedule.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, + }, + { + Name: "checkpoint_tick_seconds", + Label: "Cursor Checkpoint Tick (seconds)", + Description: "How often each pipeline persists its cursor map to the filer.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, + }, + { + Name: "refresh_interval_minutes", + Label: "Engine Refresh Interval (minutes)", + Description: "How often the scheduler rebuilds the engine snapshot from bucket lifecycle configs.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, + }, + { + Name: "max_runtime_minutes", + Label: "Max Runtime (minutes)", + Description: "Wall-clock cap on each run. Each daily run processes events for one day; the cursor persists across runs.", + FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_INT64, + Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_NUMBER, + MinValue: &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 1}}, + }, + }, + }, + }, + DefaultValues: map[string]*plugin_pb.ConfigValue{ + "dispatch_tick_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDispatchTickMinutes}}, + "checkpoint_tick_seconds": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultCheckpointTickSeconds}}, + "refresh_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultRefreshIntervalMinutes}}, + "max_runtime_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}}, + }, + }, + AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{ + DetectionIntervalSeconds: 24 * 60 * 60, // daily + DetectionTimeoutSeconds: 60, + MaxJobsPerDetection: 1, + }, + } +} + +func (h *Handler) Detect(ctx context.Context, request *plugin_pb.RunDetectionRequest, sender pluginworker.DetectionSender) error { + if request == nil || sender == nil { + return fmt.Errorf("detect: nil request or sender") + } + if request.JobType != "" && request.JobType != jobType { + return fmt.Errorf("job type %q is not handled by s3 lifecycle handler", request.JobType) + } + s3Endpoints := clusterS3Endpoints(request.ClusterContext) + if len(s3Endpoints) == 0 { + _ = sender.SendActivity(pluginworker.BuildDetectorActivity("skipped", "no s3 servers registered with master", nil)) + return sender.SendComplete(&plugin_pb.DetectionComplete{JobType: jobType, Success: true}) + } + filerAddresses := []string{} + if request.ClusterContext != nil { + filerAddresses = append(filerAddresses, request.ClusterContext.FilerGrpcAddresses...) + } + if len(filerAddresses) == 0 { + _ = sender.SendActivity(pluginworker.BuildDetectorActivity("skipped", "no filer addresses in cluster context", nil)) + return sender.SendComplete(&plugin_pb.DetectionComplete{JobType: jobType, Success: true}) + } + + proposal := &plugin_pb.JobProposal{ + JobType: jobType, + ProposalId: fmt.Sprintf("s3-lifecycle-%d", time.Now().UnixNano()), + Priority: plugin_pb.JobPriority_JOB_PRIORITY_NORMAL, + Parameters: map[string]*plugin_pb.ConfigValue{ + "filer_grpc_address": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: filerAddresses[0]}}, + }, + } + if err := sender.SendProposals(&plugin_pb.DetectionProposals{ + JobType: jobType, + Proposals: []*plugin_pb.JobProposal{proposal}, + }); err != nil { + return err + } + return sender.SendComplete(&plugin_pb.DetectionComplete{ + JobType: jobType, + Success: true, + TotalProposals: 1, + }) +} + +func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender) error { + if request == nil || request.Job == nil || sender == nil { + return fmt.Errorf("execute: nil request/job/sender") + } + if request.Job.JobType != "" && request.Job.JobType != jobType { + return fmt.Errorf("job type %q is not handled by s3 lifecycle handler", request.Job.JobType) + } + cfg := ParseConfig(request.GetAdminConfigValues(), request.GetWorkerConfigValues()) + s3Endpoints := clusterS3Endpoints(request.ClusterContext) + if len(s3Endpoints) == 0 { + return fmt.Errorf("execute: no s3 servers registered with master") + } + filerAddress := readString(request.Job.Parameters, "filer_grpc_address", "") + if filerAddress == "" { + return fmt.Errorf("execute: missing filer_grpc_address in job parameters") + } + + runCtx, cancel := context.WithTimeout(ctx, cfg.MaxRuntime) + defer cancel() + + _ = sender.SendProgress(&plugin_pb.JobProgressUpdate{ + JobId: request.Job.JobId, JobType: jobType, + State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting", + Message: fmt.Sprintf("scheduler workers=%d s3=%v runtime=%s", cfg.Workers, s3Endpoints, cfg.MaxRuntime), + }) + + dialCtx, dialCancel := context.WithTimeout(runCtx, 30*time.Second) + filerConn, err := pb.GrpcDial(dialCtx, filerAddress, false, h.grpcDialOption) + dialCancel() + if err != nil { + return fmt.Errorf("dial filer %s: %w", filerAddress, err) + } + defer filerConn.Close() + filerClient := filer_pb.NewSeaweedFilerClient(filerConn) + + bucketsPath, err := lookupBucketsPath(runCtx, filerClient) + if err != nil { + return fmt.Errorf("buckets path: %w", err) + } + + dialCtx, dialCancel = context.WithTimeout(runCtx, 30*time.Second) + s3Conn, err := pb.GrpcDial(dialCtx, s3Endpoints[0], false, h.grpcDialOption) + dialCancel() + if err != nil { + return fmt.Errorf("dial s3 %s: %w", s3Endpoints[0], err) + } + defer s3Conn.Close() + rpc := s3_lifecycle_pb.NewSeaweedS3LifecycleInternalClient(s3Conn) + + sched := &scheduler.Scheduler{ + BucketsPath: bucketsPath, + Engine: engine.New(), + Persister: &dispatcher.FilerPersister{Store: dispatcher.NewFilerStoreClient(filerClient)}, + Client: lifecycleRPCAdapter{c: rpc}, + FilerClient: filerClient, + ClientID: util.RandomInt32(), + ClientName: "worker-s3-lifecycle", + Workers: cfg.Workers, + DispatchTick: cfg.DispatchTick, + CheckpointTick: cfg.CheckpointTick, + RefreshInterval: cfg.RefreshInterval, + } + if err := sched.Run(runCtx); err != nil { + glog.Warningf("s3 lifecycle execute: %v", err) + return err + } + return nil +} + +// clusterS3Endpoints returns the master-discovered S3 gRPC addresses for the +// cluster. The handler dials the first reachable one; the master refreshes +// the list on KeepConnected so a stale entry self-heals on the next run. +func clusterS3Endpoints(cc *plugin_pb.ClusterContext) []string { + if cc == nil { + return nil + } + out := make([]string, 0, len(cc.S3GrpcAddresses)) + for _, addr := range cc.S3GrpcAddresses { + if addr != "" { + out = append(out, addr) + } + } + return out +} + +type lifecycleRPCAdapter struct { + c s3_lifecycle_pb.SeaweedS3LifecycleInternalClient +} + +func (a lifecycleRPCAdapter) LifecycleDelete(ctx context.Context, req *s3_lifecycle_pb.LifecycleDeleteRequest) (*s3_lifecycle_pb.LifecycleDeleteResponse, error) { + return a.c.LifecycleDelete(ctx, req) +} + +func lookupBucketsPath(ctx context.Context, client filer_pb.SeaweedFilerClient) (string, error) { + resp, err := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{}) + if err != nil { + return "", err + } + if path := resp.GetDirBuckets(); path != "" { + return path, nil + } + return "/buckets", nil +} + +func readString(values map[string]*plugin_pb.ConfigValue, field, fallback string) string { + v, ok := values[field] + if !ok || v == nil { + return fallback + } + if k, ok := v.Kind.(*plugin_pb.ConfigValue_StringValue); ok { + return k.StringValue + } + return fallback +} + +var _ pluginworker.JobHandler = (*Handler)(nil)