seaweedfs/weed/worker/tasks/s3_lifecycle/handler.go

package s3_lifecycle

import (
	"context"
	"fmt"
	"strconv"
	"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"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3lifecycle/dailyrun"
	"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"
	"golang.org/x/time/rate"
	"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: "Cluster-wide algorithm choice and delete-throughput cap.",
					Fields: []*plugin_pb.ConfigField{
						{
							Name:        "algorithm",
							Label:       "Algorithm",
							Description: "Daily Replay = bounded daily meta-log scan (Phase 2, replay-only — buckets using ExpirationDate, ExpiredDeleteMarker, NewerNoncurrent, or scan_only rules will fail the run until Phase 4 ships). Streaming = legacy reader+heap path, kept as a runtime escape hatch during the Phase 4 rollout; Phase 5 deletes it.",
							FieldType:   plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_ENUM,
							Widget:      plugin_pb.ConfigWidget_CONFIG_WIDGET_SELECT,
							Options: []*plugin_pb.ConfigOption{
								{Value: AlgorithmDailyReplay, Label: "Daily Replay (default)", Description: "Bounded daily meta-log scan. Replay-only in Phase 2; buckets with walker-bound rules fail the run."},
								{Value: AlgorithmStreaming, Label: "Streaming (legacy fallback)", Description: "Long-running reader + per-shard heap + tick dispatcher. Pre-cutover behavior; removed in Phase 5."},
							},
						},
						{
							Name:        ClusterDeletesPerSecondAdminKey,
							Label:       "Cluster Delete Rate (per second)",
							Description: "Cluster-wide ceiling on lifecycle delete RPCs per second, divided evenly across active s3_lifecycle workers at job-dispatch time. 0 = unlimited (legacy behavior). Only honored by the Daily Replay algorithm; streaming ignores it.",
							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: 0}},
						},
						{
							Name:        ClusterDeletesBurstAdminKey,
							Label:       "Cluster Delete Burst",
							Description: "Token-bucket burst capacity across the cluster (max simultaneous deletes). 0 = 2 × rate. Same allocation rule as the rate.",
							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: 0}},
						},
					},
				},
			},
			DefaultValues: map[string]*plugin_pb.ConfigValue{
				"algorithm":                     {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultAlgorithm}},
				ClusterDeletesPerSecondAdminKey: {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
				ClusterDeletesBurstAdminKey:     {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
			},
		},
		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: "How often the worker checks its schedule, saves progress, reloads bucket lifecycle configs, and how long each run may take.",
					Fields: []*plugin_pb.ConfigField{
						{
							Name:        "dispatch_tick_minutes",
							Label:       "Schedule Check Interval (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:       "Progress Save Interval (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:       "Lifecycle Config Reload (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:        "bootstrap_interval_minutes",
							Label:       "Full Bucket Rescan Interval (minutes)",
							Description: "How often each bucket is re-walked. scan_only rules — those whose retention horizon exceeds meta-log retention — only fire from the bootstrap walk, so a non-zero value is required to enforce them on a long-running worker. 0 keeps the legacy walk-once-per-process behavior.",
							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: 0}},
						},
						{
							Name:        "max_runtime_minutes",
							Label:       "Per-Run Time Limit (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}},
				"bootstrap_interval_minutes": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultBootstrapIntervalMinutes}},
				"max_runtime_minutes":        {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxRuntimeMinutes}},
			},
		},
		AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
			DetectionIntervalMinutes: 24 * 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)

	if cfg.Algorithm == AlgorithmDailyReplay {
		return h.executeDailyReplay(runCtx, request, bucketsPath, filerClient, rpc, cfg, sender)
	}

	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,
		BootstrapInterval: cfg.BootstrapInterval,
	}
	if err := sched.Run(runCtx); err != nil {
		glog.Warningf("s3 lifecycle execute: %v", err)
		return err
	}
	return nil
}

// executeDailyReplay runs the bounded daily-replay path. Reuses the
// streaming path's filer / s3 / engine setup but routes the per-shard
// loop through dailyrun.Run instead of scheduler.Scheduler. Phase 2:
// replay-only, refuses walker-bound action kinds with a typed error.
func (h *Handler) executeDailyReplay(ctx context.Context, request *plugin_pb.ExecuteJobRequest, bucketsPath string, filerClient filer_pb.SeaweedFilerClient, rpc s3_lifecycle_pb.SeaweedS3LifecycleInternalClient, cfg Config, sender pluginworker.ExecutionSender) error {
	eng := engine.New()
	inputs, parseErrors, err := scheduler.LoadCompileInputs(ctx, filerClient, bucketsPath)
	if err != nil {
		return fmt.Errorf("daily_replay: load lifecycle inputs: %w", err)
	}
	for _, pe := range parseErrors {
		glog.V(1).Infof("daily_replay: %s: %v", pe.Bucket, pe.Err)
	}
	eng.Compile(inputs, engine.CompileOptions{PriorStates: scheduler.AllActivePriorStates(inputs)})

	shards := make([]int, 0, s3lifecycle.ShardCount)
	for i := 0; i < s3lifecycle.ShardCount; i++ {
		shards = append(shards, i)
	}

	limiter, limiterDesc := buildLimiterFromClusterContext(request.GetClusterContext())

	_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
		JobId: request.Job.JobId, JobType: jobType,
		State: plugin_pb.JobState_JOB_STATE_RUNNING, Stage: "starting",
		Message: fmt.Sprintf("daily_replay shards=%d workers=%d runtime=%s rate=%s", len(shards), cfg.Workers, cfg.MaxRuntime, limiterDesc),
	})

	runErr := dailyrun.Run(ctx, dailyrun.Config{
		Shards:      shards,
		BucketsPath: bucketsPath,
		Engine:      eng,
		FilerClient: filerClient,
		Client:      lifecycleRPCAdapter{c: rpc},
		Persister:   &dailyrun.FilerCursorPersister{Store: dispatcher.NewFilerStoreClient(filerClient)},
		Lister:      dispatcher.NewFilerSiblingLister(filerClient, bucketsPath),
		Workers:     cfg.Workers,
		Limiter:     limiter,
		ClientName:  "worker-s3-lifecycle-daily",
	})
	if dailyrun.IsUnsupportedRule(runErr) {
		// Surface the typed error verbatim so admin marks the run as
		// failed with the user-facing reason in the activity log.
		glog.Warningf("daily_replay: %v", runErr)
		return runErr
	}
	if runErr != nil {
		glog.Warningf("daily_replay: %v", runErr)
		return runErr
	}
	return nil
}

// buildLimiterFromClusterContext parses the per-worker share the admin
// wrote into ClusterContext.Metadata (see weed/admin/plugin/plugin.go's
// s3_lifecycle injection) and returns a rate.Limiter, or nil when no
// rate cap applies. The description string is for the JobProgressUpdate
// so operators can see "rate=unlimited" / "rate=12.5/s burst=25" in
// the activity log.
//
// Tolerant of missing keys, empty strings, malformed numbers, and
// non-positive values — all treated as "no limit" rather than failing
// the run. The admin allocator is the single point that decides whether
// to populate these keys; the worker doesn't second-guess.
func buildLimiterFromClusterContext(cc *plugin_pb.ClusterContext) (*rate.Limiter, string) {
	if cc == nil || cc.Metadata == nil {
		return nil, "unlimited"
	}
	rps, ok := parsePositiveFloat(cc.Metadata[MetadataKeyDeletesPerSecond])
	if !ok {
		return nil, "unlimited"
	}
	burst, _ := parsePositiveInt(cc.Metadata[MetadataKeyDeletesBurst])
	if burst <= 0 {
		// Sensible default: enough headroom for one tick's worth of
		// throughput. Caller may also supply 0 to opt into this default.
		burst = int(rps * 2)
		if burst < 1 {
			burst = 1
		}
	}
	return rate.NewLimiter(rate.Limit(rps), burst), fmt.Sprintf("%.3g/s burst=%d", rps, burst)
}

func parsePositiveFloat(s string) (float64, bool) {
	if s == "" {
		return 0, false
	}
	v, err := strconv.ParseFloat(s, 64)
	if err != nil || v <= 0 {
		return 0, false
	}
	return v, true
}

func parsePositiveInt(s string) (int, bool) {
	if s == "" {
		return 0, false
	}
	v, err := strconv.Atoi(s)
	if err != nil || v <= 0 {
		return 0, false
	}
	return v, true
}

// 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)