mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-08-28 11:56:07 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
seaweedfs/weed/worker/tasks/iceberg/handler.go
T
Chris LuandGitHub eef6f3d1e6 s3tables: add the maintenance configuration APIs (#10773) 
* s3tables: add the maintenance configuration APIs

Stores the configuration verbatim as the wire shape under a new
s3tables.maintenance extended attribute, so Get hands back what Put took
and no translation layer can drift from the AWS model.

Nothing reads the configuration yet.

Put merges a single type into the stored map so configuring compaction
does not drop snapshot management, and asserts the attribute's prior value
so two concurrent Puts cannot silently clobber each other.

* iceberg: apply the maintenance configuration in the worker

The worker now reads the per-table and per-bucket maintenance
configuration written by the control plane, so the wildcard plugin config
is a default rather than the only setting a table can have.

Table properties still win by default, since a table declaring its own
layout is what every engine honours and the compactor has to agree with
whoever writes the files. Clearing table_properties_override makes the
maintenance configuration authoritative instead.

Status is not part of that contest: a disabled type drops its operations
and no property can re-enable them, so the operator's kill switch always
holds. Manifest and delete-file rewrites have no AWS equivalent and ride
with compaction.

Detection reads both attributes from entries it already lists.

* s3tables: report maintenance job status

The worker records the outcome of each run in its own extended attribute,
separate from the configuration so operator and worker writes do not
contend, and GetTableMaintenanceJobStatus reads it back.

Only the types a run touched are written, so a partial run cannot erase
what an earlier one recorded. The reader fills in the rest: Disabled when
the configuration switched a type off, Not_Yet_Run otherwise.

Status is advisory, so a lost race is logged rather than failing a job
whose work already committed.

* s3tables: route the maintenance APIs over REST

The five actions were only reachable by X-Amz-Target dispatch, which the
AWS CLI and SDK do not use for this service. They address the operations
by path, so the APIs were unreachable from any official client.

* s3tables: fix the table bucket ARN field name

GetTableBucketMaintenanceConfiguration emitted tableBucketArn where the
wire field is tableBucketARN, as every other response in this package
already spells it. Official SDK deserializers ignore the unknown key, so
the required field came back unset.

* s3tables: carry the compaction strategy through to the worker

IcebergCompactionSettings modelled only targetFileSizeMB, so a request
naming a strategy was accepted and then dropped on the way to storage.
The worker now maps binpack and sort onto its own rewrite strategy and
lets auto defer to the worker configuration.

z-order is rejected rather than accepted and quietly binpacked.

* s3tables: report bucket-level maintenance status

GetTableMaintenanceJobStatus read only the table's configuration, so
unreferenced file removal — which is configured on the bucket — reported
Not_Yet_Run or a stale success after an operator disabled it.

The merge helper now lives in this package and the worker shares it.

* iceberg: delete orphans only after the non-current window

AWS marks a file non-current once it has been unreferenced for
unreferencedDays, then deletes it a further nonCurrentDays later.
The cutoff was taken from unreferencedDays alone, so a 3/10 configuration
hard-deleted on day three and threw away the ten day recovery window.

remove_orphans deletes in one step rather than marking, so the cutoff is
now the sum of the two.

* s3tables: assert every attribute when rewriting an entry

UpdateEntry writes the whole entry back from the snapshot the caller
read, and its precondition only covers the keys the caller names. Both
maintenance writers named one key, so a job status write could revert a
maintenance configuration an operator had just disabled, turning an
advisory write into a silent re-enable.

Both now assert the entry's full attribute set, including the target key
when absent so a concurrent create also fails the precondition.

* s3tables: assert absent attributes when rewriting an entry

The precondition covered the attributes present when the writer read the
entry, so an attribute created between that read and the write was absent
from it. A first-time PutTableMaintenanceConfiguration disabling a type
therefore lands, passes the per-key checks, and is then deleted by the
stale whole-entry write.

Every attribute this package stores is now asserted, absent ones
included. The metadata commit and planning index writers rewrite the same
entries and had the same exposure, so both use the shared snapshot too.

* iceberg: implement the auto compaction strategy

auto was accepted, stored and read back, but left the worker on its own
default, so a sorted table configured as auto was compacted with binpack.

AWS defines auto as sorting tables that declare a sort order and
bin-packing the rest. That needs the table metadata, so the choice is made
where the rewrite plan is resolved: an unsorted table falls back to
binpack rather than failing the way an explicit sort request does.

* s3tables: validate the maintenance setting ranges

PUT accepted zero, negative and oversized values for every numeric
setting. The worker then ignores a non-positive value and saturates an
oversized one, so the configuration read back was not the one that ran.

AWS bounds all five to 1..2147483647, which is now enforced. The fields
are pointers so an explicit zero is distinguishable from an omitted one
and can be rejected rather than silently ignored.

* s3tables: give every entry writer the same compare-and-swap

updateExtendedAttribute asserted the entry's attributes, but the helpers
behind the metadata, policy and tag handlers still wrote the whole entry
unconditionally. Any of them could land on a stale snapshot and delete a
maintenance configuration an operator had just written.

They all share one read-modify-write loop now, so the precondition and
the bounded retry apply wherever an entry is rewritten.

* s3tables: move the maintenance configuration with a renamed table

RenameTable carried the metadata, version, policy and tags to the new
name but left the maintenance configuration and job status behind. A
table with snapshot management disabled came back enabled under its new
name, and the stale configuration stayed on the old name where a table
created there would inherit it.

The decoupled-delete cleanup left the same two attributes behind.

* s3tables: accept every AWS partition in ARNs

The route regexes and the ARN patterns both hardcoded arn:aws, so valid
aws-cn and aws-us-gov ARNs never reached a handler. The router now shares
the partition-tolerant prefix with the parser, and a generated ARN uses
the partition its region belongs to so it parses back.

* s3tables: generate ARNs in the region's partition

The handler's own ARN generators still formatted arn:aws directly rather
than going through the partition-aware builder, so a China or GovCloud
deployment routed the request but then returned a commercial ARN and
matched IAM policies against it.

The round-trip test missed this because parsing accepts any partition, so
it now asserts the prefix the region implies.

* s3tables: complete the ARN partition table

aws-iso-e, aws-iso-f and aws-eusc were missing, so eu-isoe-*, us-isof-*
and eusc-* regions fell through to the commercial partition.

* s3tables: do not let a rename swallow a concurrent maintenance write

Rename copied the source attributes early and cleared the source at the
end, so a Put landing in between missed the copy to the destination and
was then deleted by the cleanup. It succeeded and vanished.

The cleanup now clears the source only while it still holds exactly what
was copied, and returns a conflict otherwise. Put checks the catalog
identity inside the same conditional mutation, so it also cannot write to
a name that a rename or delete has already soft-deleted.
2026-08-16 10:36:59 -07:00

721 lines
31 KiB
Go
Raw Permalink Blame History

package iceberg
import (
"context"
"fmt"
"path"
"strings"
"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"
pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/timestamppb"
)
func init() {
pluginworker.RegisterHandler(pluginworker.HandlerFactory{
JobType: jobType,
Category: pluginworker.CategoryHeavy,
Aliases: []string{"iceberg-maintenance", "iceberg.maintenance", "iceberg"},
Build: func(opts pluginworker.HandlerBuildOptions) (pluginworker.JobHandler, error) {
return NewHandler(opts.GrpcDialOption), nil
},
})
}
// Handler implements the JobHandler interface for Iceberg table maintenance:
// snapshot expiration, orphan file removal, and manifest rewriting.
type Handler struct {
grpcDialOption grpc.DialOption
}
const filerConnectTimeout = 5 * time.Second
// NewHandler creates a new handler for iceberg table maintenance.
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: 4,
DisplayName: "Iceberg Maintenance",
Description: "Compacts data, rewrites delete files, expires snapshots, removes orphans, and rewrites manifests for Iceberg tables in S3 table buckets",
Weight: 50,
}
}
func (h *Handler) Descriptor() *plugin_pb.JobTypeDescriptor {
return &plugin_pb.JobTypeDescriptor{
JobType: jobType,
DisplayName: "Iceberg Maintenance",
Description: "Automated maintenance for Iceberg tables: data compaction, delete-file rewrite, snapshot expiration, orphan removal, and manifest rewriting",
Icon: "fas fa-snowflake",
DescriptorVersion: 1,
AdminConfigForm: &plugin_pb.ConfigForm{
FormId: "iceberg-maintenance-admin",
Title: "Iceberg Maintenance Admin Config",
Description: "Admin-side controls for Iceberg table maintenance scope.",
Sections: []*plugin_pb.ConfigSection{
{
SectionId: "scope",
Title: "Scope",
Description: "Filters to restrict which tables are scanned for maintenance.",
Fields: []*plugin_pb.ConfigField{
{
Name: "bucket_filter",
Label: "Bucket Filter",
Description: "Comma-separated wildcard patterns for table buckets (* and ? supported). Blank = all.",
Placeholder: "prod-*, staging-*",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
{
Name: "namespace_filter",
Label: "Namespace Filter",
Description: "Comma-separated wildcard patterns for namespaces (* and ? supported). Blank = all.",
Placeholder: "analytics, events-*",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
{
Name: "table_filter",
Label: "Table Filter",
Description: "Comma-separated wildcard patterns for table names (* and ? supported). Blank = all.",
Placeholder: "clicks, orders-*",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
},
},
{
SectionId: "resources",
Title: "Resource Groups",
Description: "Controls for fair proposal distribution across buckets or namespaces.",
Fields: []*plugin_pb.ConfigField{
{
Name: "resource_group_by",
Label: "Group Proposals By",
Description: "When set, detection emits proposals in round-robin order across the selected resource group.",
Placeholder: "none, bucket, namespace, or bucket_namespace",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
{
Name: "max_tables_per_resource_group",
Label: "Max Tables Per Group",
Description: "Optional cap on how many proposals a single resource group can receive in one detection run. Zero disables the cap.",
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{
"bucket_filter": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}},
"namespace_filter": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}},
"table_filter": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}},
"resource_group_by": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: resourceGroupNone}},
"max_tables_per_resource_group": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
},
},
WorkerConfigForm: &plugin_pb.ConfigForm{
FormId: "iceberg-maintenance-worker",
Title: "Iceberg Maintenance Worker Config",
Description: "Worker-side thresholds for maintenance operations.",
Sections: []*plugin_pb.ConfigSection{
{
SectionId: "snapshots",
Title: "Snapshot Expiration",
Description: "Controls for automatic snapshot cleanup.",
Fields: []*plugin_pb.ConfigField{
{
Name: "snapshot_retention_hours",
Label: "Retention (hours)",
Description: "Expire snapshots older than this many hours.",
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_snapshots_to_keep",
Label: "Max Snapshots",
Description: "Always keep at least this many most recent snapshots.",
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}},
},
},
},
{
SectionId: "compaction",
Title: "Data Compaction",
Description: "Controls for bin-packing or sorting small Parquet data files.",
Fields: []*plugin_pb.ConfigField{
{
Name: "target_file_size_mb",
Label: "Target File Size (MB)",
Description: "Files smaller than this (in megabytes) are candidates for compaction.",
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: "min_input_files",
Label: "Min Input Files",
Description: "Minimum number of small files in a partition to trigger compaction.",
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: 2}},
},
{
Name: "apply_deletes",
Label: "Apply Deletes",
Description: "When true, compaction applies position and equality deletes to data files. When false, tables with delete manifests are skipped.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
},
{
Name: "rewrite_strategy",
Label: "Rewrite Strategy",
Description: "binpack keeps the existing row order; sort rewrites each compaction bin using the Iceberg table sort order; auto sorts tables that declare one and bin-packs the rest.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
Placeholder: "binpack or sort",
},
{
Name: "sort_max_input_mb",
Label: "Sort Max Input (MB)",
Description: "Optional hard cap for the total bytes in a sorted compaction bin. Zero = no extra cap beyond binning.",
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}},
},
},
},
{
SectionId: "delete_rewrite",
Title: "Delete Rewrite",
Description: "Controls for rewriting small position-delete files into fewer larger files.",
Fields: []*plugin_pb.ConfigField{
{
Name: "delete_target_file_size_mb",
Label: "Delete Target File Size (MB)",
Description: "Target size for rewritten position-delete files.",
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: "delete_min_input_files",
Label: "Delete Min Input Files",
Description: "Minimum number of position-delete files in a group before rewrite is triggered.",
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: 2}},
},
{
Name: "delete_max_file_group_size_mb",
Label: "Delete Max Group Size (MB)",
Description: "Skip rewriting delete groups larger than this bound.",
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: "delete_max_output_files",
Label: "Delete Max Output Files",
Description: "Maximum number of rewritten delete files a single group may produce.",
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}},
},
},
},
{
SectionId: "orphans",
Title: "Orphan Removal",
Description: "Controls for orphan file cleanup.",
Fields: []*plugin_pb.ConfigField{
{
Name: "orphan_older_than_hours",
Label: "Safety Window (hours)",
Description: "Only remove orphan files older than this many hours.",
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}},
},
},
},
{
SectionId: "manifests",
Title: "Manifest Rewriting",
Description: "Controls for merging small manifests.",
Fields: []*plugin_pb.ConfigField{
{
Name: "min_manifests_to_rewrite",
Label: "Min Manifests",
Description: "Minimum number of manifests before rewriting is triggered.",
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: 2}},
},
},
},
{
SectionId: "general",
Title: "General",
Description: "General maintenance settings.",
Fields: []*plugin_pb.ConfigField{
{
Name: "max_commit_retries",
Label: "Max Commit Retries",
Description: "Maximum number of commit retries on version conflict.",
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: 20}},
},
{
Name: "operations",
Label: "Operations",
Description: "Comma-separated list of operations to run: compact, rewrite_position_delete_files, expire_snapshots, remove_orphans, rewrite_manifests, or 'all'.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
},
{
Name: "table_properties_override",
Label: "Table Properties Win",
Description: "Let a table's own Iceberg properties override these settings and its maintenance configuration. Clear to make the maintenance configuration authoritative.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_BOOL,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TOGGLE,
},
{
Name: "where",
Label: "Where Filter",
Description: "Optional partition filter for compact, rewrite_position_delete_files, and rewrite_manifests. Supports field = literal, field IN (...), and AND.",
FieldType: plugin_pb.ConfigFieldType_CONFIG_FIELD_TYPE_STRING,
Widget: plugin_pb.ConfigWidget_CONFIG_WIDGET_TEXT,
Placeholder: "region = 'us' AND dt IN ('2026-03-15')",
},
},
},
},
DefaultValues: map[string]*plugin_pb.ConfigValue{
"target_file_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultTargetFileSizeMB}},
"min_input_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMinInputFiles}},
"delete_target_file_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteTargetFileSizeMB}},
"delete_min_input_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMinInputFiles}},
"delete_max_file_group_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMaxGroupSizeMB}},
"delete_max_output_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMaxOutputFiles}},
"min_manifests_to_rewrite": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMinManifestsToRewrite}},
"snapshot_retention_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultSnapshotRetentionHours}},
"max_snapshots_to_keep": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxSnapshotsToKeep}},
"orphan_older_than_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultOrphanOlderThanHours}},
"max_commit_retries": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
"operations": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
"apply_deletes": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
"table_properties_override": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
"rewrite_strategy": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultRewriteStrategy}},
"sort_max_input_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
"where": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}},
},
},
AdminRuntimeDefaults: &plugin_pb.AdminRuntimeDefaults{
Enabled: false, // disabled by default
DetectionIntervalMinutes: 60, // 1 hour
DetectionTimeoutSeconds: 300,
MaxJobsPerDetection: 100,
GlobalExecutionConcurrency: 4,
PerWorkerExecutionConcurrency: 2,
RetryLimit: 1,
RetryBackoffSeconds: 60,
JobTypeMaxRuntimeSeconds: 3600, // 1 hour max
ExecutionTimeoutSeconds: 3600,
},
WorkerDefaultValues: map[string]*plugin_pb.ConfigValue{
"target_file_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultTargetFileSizeMB}},
"min_input_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMinInputFiles}},
"delete_target_file_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteTargetFileSizeMB}},
"delete_min_input_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMinInputFiles}},
"delete_max_file_group_size_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMaxGroupSizeMB}},
"delete_max_output_files": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultDeleteMaxOutputFiles}},
"snapshot_retention_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultSnapshotRetentionHours}},
"max_snapshots_to_keep": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxSnapshotsToKeep}},
"orphan_older_than_hours": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultOrphanOlderThanHours}},
"max_commit_retries": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: defaultMaxCommitRetries}},
"operations": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultOperations}},
"apply_deletes": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
"table_properties_override": {Kind: &plugin_pb.ConfigValue_BoolValue{BoolValue: true}},
"rewrite_strategy": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: defaultRewriteStrategy}},
"sort_max_input_mb": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: 0}},
"where": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: ""}},
},
}
}
func (h *Handler) Detect(ctx context.Context, request *plugin_pb.RunDetectionRequest, sender pluginworker.DetectionSender) error {
if request == nil {
return fmt.Errorf("run detection request is nil")
}
if sender == nil {
return fmt.Errorf("detection sender is nil")
}
if request.JobType != "" && request.JobType != jobType {
return fmt.Errorf("job type %q is not handled by iceberg maintenance handler", request.JobType)
}
workerConfig := ParseConfig(request.GetWorkerConfigValues())
ops, err := parseOperations(workerConfig.Operations)
if err != nil {
return fmt.Errorf("invalid operations config: %w", err)
}
if err := validateWhereOperations(workerConfig.Where, ops); err != nil {
return fmt.Errorf("invalid where config: %w", err)
}
// Detection interval is managed by the scheduler via AdminRuntimeDefaults.DetectionIntervalMinutes.
// ClusterContext.FilerAddresses are pb.ServerAddress strings
// (host:httpPort.grpcPort); collapse each to a dialable gRPC address.
filerGrpcAddresses := make([]string, 0)
if request.ClusterContext != nil {
for _, filer := range request.ClusterContext.FilerAddresses {
filerGrpcAddresses = append(filerGrpcAddresses, pb.ServerAddress(filer).ToGrpcAddress())
}
}
if len(filerGrpcAddresses) == 0 {
_ = sender.SendActivity(pluginworker.BuildDetectorActivity("skipped", "no filer addresses in cluster context", nil))
return h.sendEmptyDetection(sender)
}
// Read scope filters
bucketFilter := strings.TrimSpace(readStringConfig(request.GetAdminConfigValues(), "bucket_filter", ""))
namespaceFilter := strings.TrimSpace(readStringConfig(request.GetAdminConfigValues(), "namespace_filter", ""))
tableFilter := strings.TrimSpace(readStringConfig(request.GetAdminConfigValues(), "table_filter", ""))
resourceGroups, err := readResourceGroupConfig(request.GetAdminConfigValues())
if err != nil {
return fmt.Errorf("invalid admin resource group config: %w", err)
}
// Connect to filer — try each address until one succeeds.
filerGrpcAddress, conn, err := h.connectToFiler(ctx, filerGrpcAddresses)
if err != nil {
return fmt.Errorf("connect to filer: %w", err)
}
defer conn.Close()
filerClient := filer_pb.NewSeaweedFilerClient(conn)
maxResults := int(request.MaxResults)
scanLimit := maxResults
if resourceGroups.enabled() {
scanLimit = 0
}
tables, err := h.scanTablesForMaintenance(ctx, filerClient, workerConfig, bucketFilter, namespaceFilter, tableFilter, scanLimit)
if err != nil {
_ = sender.SendActivity(pluginworker.BuildDetectorActivity("scan_error", fmt.Sprintf("error scanning tables: %v", err), nil))
return fmt.Errorf("scan tables: %w", err)
}
_ = sender.SendActivity(pluginworker.BuildDetectorActivity("scan_complete",
fmt.Sprintf("found %d table(s) needing maintenance", len(tables)),
map[string]*plugin_pb.ConfigValue{
"tables_found": {Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: int64(len(tables))}},
}))
tables, hasMore := selectTablesByResourceGroup(tables, resourceGroups, maxResults)
proposals := make([]*plugin_pb.JobProposal, 0, len(tables))
for _, t := range tables {
proposal := h.buildMaintenanceProposal(t, filerGrpcAddress, resourceGroupKey(t, resourceGroups.GroupBy))
proposals = append(proposals, proposal)
}
if err := sender.SendProposals(&plugin_pb.DetectionProposals{
JobType: jobType,
Proposals: proposals,
HasMore: hasMore,
}); err != nil {
return err
}
return sender.SendComplete(&plugin_pb.DetectionComplete{
JobType: jobType,
Success: true,
TotalProposals: int32(len(proposals)),
})
}
func (h *Handler) Execute(ctx context.Context, request *plugin_pb.ExecuteJobRequest, sender pluginworker.ExecutionSender) error {
if request == nil || request.Job == nil {
return fmt.Errorf("execute request/job is nil")
}
if sender == nil {
return fmt.Errorf("execution sender is nil")
}
if request.Job.JobType != "" && request.Job.JobType != jobType {
return fmt.Errorf("job type %q is not handled by iceberg maintenance handler", request.Job.JobType)
}
canonicalJobType := request.Job.JobType
if canonicalJobType == "" {
canonicalJobType = jobType
}
params := request.Job.Parameters
bucketName := readStringConfig(params, "bucket_name", "")
namespace := readStringConfig(params, "namespace", "")
tableName := readStringConfig(params, "table_name", "")
tablePath := readStringConfig(params, "table_path", "")
filerGrpcAddress := readStringConfig(params, "filer_address", "")
if bucketName == "" || namespace == "" || tableName == "" || filerGrpcAddress == "" {
return fmt.Errorf("missing required parameters: bucket_name=%q, namespace=%q, table_name=%q, filer_address=%q", bucketName, namespace, tableName, filerGrpcAddress)
}
// Reject path traversal in bucket/namespace/table names.
for _, name := range []string{bucketName, namespace, tableName} {
if strings.Contains(name, "..") || strings.ContainsAny(name, "/\\") {
return fmt.Errorf("invalid name %q: must not contain path separators or '..'", name)
}
}
if tablePath == "" {
tablePath = path.Join(namespace, tableName)
}
// Sanitize tablePath to prevent directory traversal.
tablePath = path.Clean(tablePath)
expected := path.Join(namespace, tableName)
if tablePath != expected && !strings.HasPrefix(tablePath, expected+"/") {
return fmt.Errorf("invalid table_path %q: must be %q or a subpath", tablePath, expected)
}
workerConfig := ParseConfig(request.GetWorkerConfigValues())
ops, opsErr := parseOperations(workerConfig.Operations)
if opsErr != nil {
return fmt.Errorf("invalid operations config: %w", opsErr)
}
if err := validateWhereOperations(workerConfig.Where, ops); err != nil {
return fmt.Errorf("invalid where config: %w", err)
}
// Send initial progress
if err := sender.SendProgress(&plugin_pb.JobProgressUpdate{
JobId: request.Job.JobId,
JobType: canonicalJobType,
State: plugin_pb.JobState_JOB_STATE_ASSIGNED,
ProgressPercent: 0,
Stage: "assigned",
Message: fmt.Sprintf("maintenance job accepted for %s/%s/%s", bucketName, namespace, tableName),
Activities: []*plugin_pb.ActivityEvent{
pluginworker.BuildExecutorActivity("assigned", fmt.Sprintf("maintenance job accepted for %s/%s/%s", bucketName, namespace, tableName)),
},
}); err != nil {
return err
}
// Connect to filer
conn, err := h.dialFiler(ctx, filerGrpcAddress)
if err != nil {
return fmt.Errorf("connect to filer %s: %w", filerGrpcAddress, err)
}
defer conn.Close()
filerClient := filer_pb.NewSeaweedFilerClient(conn)
// Resolve once for the whole job: compaction commits new metadata as it
// runs, and a job should not change settings halfway through. Both reads
// fail the job rather than fall back, since without the configuration a
// disabled operation would run anyway, and without the properties the
// operations would rewrite the table to a size it did not ask for.
maintenance, err := loadMaintenanceConfiguration(ctx, filerClient, bucketName, tablePath)
if err != nil {
return fmt.Errorf("read maintenance configuration for %s/%s: %w", bucketName, tablePath, err)
}
ops = filterDisabledOperations(ops, maintenance)
state, err := loadCurrentMetadata(ctx, filerClient, bucketName, tablePath)
if err != nil {
return fmt.Errorf("read table properties for %s/%s: %w", bucketName, tablePath, err)
}
workerConfig = resolveTableConfig(workerConfig, state.Metadata.Properties(), maintenance)
var results []string
if len(ops) == 0 {
results = append(results, "all maintenance operations are disabled for this table")
}
var lastErr error
totalOps := len(ops)
completedOps := 0
allMetrics := make(map[string]int64)
opErrors := make(map[string]error, totalOps)
// Execute operations in canonical maintenance order as defined by
// parseOperations.
for _, op := range ops {
select {
case <-ctx.Done():
return ctx.Err()
default:
}
progress := float64(completedOps) / float64(totalOps) * 100
if err := sender.SendProgress(&plugin_pb.JobProgressUpdate{
JobId: request.Job.JobId,
JobType: canonicalJobType,
State: plugin_pb.JobState_JOB_STATE_RUNNING,
ProgressPercent: progress,
Stage: op,
Message: fmt.Sprintf("running %s", op),
Activities: []*plugin_pb.ActivityEvent{
pluginworker.BuildExecutorActivity(op, fmt.Sprintf("starting %s for %s/%s/%s", op, bucketName, namespace, tableName)),
},
}); err != nil {
return err
}
var opResult string
var opErr error
var opMetrics map[string]int64
switch op {
case "compact":
opResult, opMetrics, opErr = h.compactDataFiles(ctx, filerClient, bucketName, tablePath, workerConfig, func(binIdx, totalBins int) {
binProgress := progress + float64(binIdx+1)/float64(totalBins)*(100.0/float64(totalOps))
_ = sender.SendProgress(&plugin_pb.JobProgressUpdate{
JobId: request.Job.JobId,
JobType: canonicalJobType,
State: plugin_pb.JobState_JOB_STATE_RUNNING,
ProgressPercent: binProgress,
Stage: fmt.Sprintf("compact bin %d/%d", binIdx+1, totalBins),
Message: fmt.Sprintf("compacting bin %d of %d", binIdx+1, totalBins),
})
})
case "rewrite_position_delete_files":
opResult, opMetrics, opErr = h.rewritePositionDeleteFiles(ctx, filerClient, bucketName, tablePath, workerConfig)
case "expire_snapshots":
opResult, opMetrics, opErr = h.expireSnapshots(ctx, filerClient, bucketName, tablePath, workerConfig)
case "remove_orphans":
opResult, opMetrics, opErr = h.removeOrphans(ctx, filerClient, bucketName, tablePath, workerConfig)
case "rewrite_manifests":
opResult, opMetrics, opErr = h.rewriteManifests(ctx, filerClient, bucketName, tablePath, workerConfig)
default:
glog.Warningf("unknown maintenance operation: %s", op)
continue
}
// Accumulate per-operation metrics with dot-prefixed keys
for k, v := range opMetrics {
allMetrics[op+"."+k] = v
}
completedOps++
opErrors[op] = opErr
if opErr != nil {
glog.Warningf("iceberg maintenance %s failed for %s/%s/%s: %v", op, bucketName, namespace, tableName, opErr)
results = append(results, fmt.Sprintf("%s: error: %v", op, opErr))
lastErr = opErr
} else {
results = append(results, fmt.Sprintf("%s: %s", op, opResult))
}
}
recordJobStatus(ctx, filerClient, bucketName, tablePath, buildJobStatus(opErrors, time.Now().UTC()))
resultSummary := strings.Join(results, "; ")
success := lastErr == nil
// Build OutputValues with base table info + per-operation metrics
outputValues := map[string]*plugin_pb.ConfigValue{
"bucket": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: bucketName}},
"namespace": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: namespace}},
"table": {Kind: &plugin_pb.ConfigValue_StringValue{StringValue: tableName}},
}
for k, v := range allMetrics {
outputValues[k] = &plugin_pb.ConfigValue{Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: v}}
}
return sender.SendCompleted(&plugin_pb.JobCompleted{
JobId: request.Job.JobId,
JobType: canonicalJobType,
Success: success,
ErrorMessage: func() string {
if lastErr != nil {
return lastErr.Error()
}
return ""
}(),
Result: &plugin_pb.JobResult{
Summary: resultSummary,
OutputValues: outputValues,
},
Activities: []*plugin_pb.ActivityEvent{
pluginworker.BuildExecutorActivity("completed", resultSummary),
},
CompletedAt: timestamppb.Now(),
})
}
func (h *Handler) sendEmptyDetection(sender pluginworker.DetectionSender) error {
if err := sender.SendProposals(&plugin_pb.DetectionProposals{
JobType: jobType,
Proposals: []*plugin_pb.JobProposal{},
HasMore: false,
}); err != nil {
return err
}
return sender.SendComplete(&plugin_pb.DetectionComplete{
JobType: jobType,
Success: true,
TotalProposals: 0,
})
}
// dialFiler connects to a filer at the given gRPC address. The address must
// already be a dialable host:grpcPort: Detect resolves it from
// ClusterContext.FilerAddresses via ToGrpcAddress and stores that resolved form
// in the job proposal parameter, so dialFiler dials it verbatim.
func (h *Handler) dialFiler(ctx context.Context, grpcAddress string) (*grpc.ClientConn, error) {
opCtx, opCancel := context.WithTimeout(ctx, filerConnectTimeout)
defer opCancel()
conn, err := pb.GrpcDial(opCtx, grpcAddress, false, h.grpcDialOption)
if err != nil {
return nil, err
}
client := filer_pb.NewSeaweedFilerClient(conn)
if _, err := client.Ping(opCtx, &filer_pb.PingRequest{}); err != nil {
_ = conn.Close()
return nil, err
}
return conn, nil
}
// connectToFiler tries each filer gRPC address in order and returns the
// first address whose gRPC connection and Ping request succeed.
func (h *Handler) connectToFiler(ctx context.Context, grpcAddresses []string) (string, *grpc.ClientConn, error) {
var lastErr error
for _, grpcAddr := range grpcAddresses {
conn, err := h.dialFiler(ctx, grpcAddr)
if err != nil {
lastErr = fmt.Errorf("filer %s: %w", grpcAddr, err)
continue
}
return grpcAddr, conn, nil
}
if lastErr == nil {
lastErr = fmt.Errorf("no filer addresses provided")
}
return "", nil, lastErr
}
// Ensure Handler implements JobHandler.
var _ pluginworker.JobHandler = (*Handler)(nil)
Reference in New Issue View Git Blame Copy Permalink