diff --git a/weed/s3api/iceberg/handlers_metrics.go b/weed/s3api/iceberg/handlers_metrics.go new file mode 100644 index 000000000..87dd3a813 --- /dev/null +++ b/weed/s3api/iceberg/handlers_metrics.go @@ -0,0 +1,58 @@ +package iceberg + +import ( + "encoding/json" + "io" + "net/http" + + "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/glog" +) + +// Reports carry per-file scan metrics and grow with the table. The content is +// discarded either way, so anything past this is accepted without being read +// into memory rather than rejected. +const maxMetricsReportBytes = 1 << 20 + +// handleReportMetrics accepts the scan and commit reports engines send after +// planning or committing. The catalog keeps no metrics store, but the endpoint +// has to exist: clients that get a 404 here log an error per query, and some +// treat repeated failures as a broken catalog. +func (s *Server) handleReportMetrics(w http.ResponseWriter, r *http.Request) { + vars := mux.Vars(r) + namespace := parseNamespace(vars["namespace"]) + tableName := vars["table"] + + if len(namespace) == 0 || tableName == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "Namespace and table name are required") + return + } + + // One byte past the limit distinguishes a report that fits from one that + // was cut short, which would otherwise fail to parse and look malformed. + body, err := io.ReadAll(io.LimitReader(r.Body, maxMetricsReportBytes+1)) + if err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body") + return + } + if len(body) > maxMetricsReportBytes { + glog.V(3).Infof("Iceberg: discarding oversized metrics report for %s.%s", flattenNamespacePath(namespace), tableName) + w.WriteHeader(http.StatusNoContent) + return + } + + var report struct { + ReportType string `json:"report-type"` + } + if err := json.Unmarshal(body, &report); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body: "+err.Error()) + return + } + if report.ReportType == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "report-type is required") + return + } + + glog.V(3).Infof("Iceberg: metrics report %q for %s.%s", report.ReportType, flattenNamespacePath(namespace), tableName) + w.WriteHeader(http.StatusNoContent) +} diff --git a/weed/s3api/iceberg/handlers_metrics_test.go b/weed/s3api/iceberg/handlers_metrics_test.go new file mode 100644 index 000000000..41659c003 --- /dev/null +++ b/weed/s3api/iceberg/handlers_metrics_test.go @@ -0,0 +1,63 @@ +package iceberg + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/gorilla/mux" +) + +func postMetrics(t *testing.T, body string) *httptest.ResponseRecorder { + t.Helper() + r := httptest.NewRequest(http.MethodPost, "/v1/namespaces/ns/tables/t/metrics", strings.NewReader(body)) + r = mux.SetURLVars(r, map[string]string{"namespace": "ns", "table": "t"}) + w := httptest.NewRecorder() + (&Server{}).handleReportMetrics(w, r) + return w +} + +func TestHandleReportMetricsAcceptsScanReport(t *testing.T) { + w := postMetrics(t, `{"report-type":"scan-report","table-name":"t","snapshot-id":1}`) + if w.Code != http.StatusNoContent { + t.Errorf("status = %d, want %d: %s", w.Code, http.StatusNoContent, w.Body.String()) + } +} + +func TestHandleReportMetricsRejectsGarbage(t *testing.T) { + w := postMetrics(t, `not json`) + if w.Code != http.StatusBadRequest { + t.Errorf("status = %d, want %d", w.Code, http.StatusBadRequest) + } +} + +// The spec makes the body and its report-type required. +func TestHandleReportMetricsRequiresReportType(t *testing.T) { + for _, body := range []string{``, `{}`, `{"report-type":""}`} { + if w := postMetrics(t, body); w.Code != http.StatusBadRequest { + t.Errorf("body %q: status = %d, want %d", body, w.Code, http.StatusBadRequest) + } + } +} + +// A report larger than the read limit is still a valid report; truncating it +// and then failing to parse would hand the client an error for a query that +// worked. +func TestHandleReportMetricsAcceptsOversizedReport(t *testing.T) { + padding := strings.Repeat("x", maxMetricsReportBytes) + w := postMetrics(t, `{"report-type":"scan-report","padding":"`+padding+`"}`) + if w.Code != http.StatusNoContent { + t.Errorf("status = %d, want %d: %s", w.Code, http.StatusNoContent, w.Body.String()) + } +} + +func TestHandleReportMetricsRequiresTable(t *testing.T) { + r := httptest.NewRequest(http.MethodPost, "/v1/namespaces/ns/tables//metrics", strings.NewReader(`{}`)) + r = mux.SetURLVars(r, map[string]string{"namespace": "ns"}) + w := httptest.NewRecorder() + (&Server{}).handleReportMetrics(w, r) + if w.Code != http.StatusBadRequest { + t.Errorf("status = %d, want %d", w.Code, http.StatusBadRequest) + } +} diff --git a/weed/s3api/iceberg/handlers_table.go b/weed/s3api/iceberg/handlers_table.go index 4e130bfa1..8929693dc 100644 --- a/weed/s3api/iceberg/handlers_table.go +++ b/weed/s3api/iceberg/handlers_table.go @@ -516,6 +516,11 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) { } result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName) + if buildErr == nil { + // Only LoadTable defines ?snapshots=; a create response always carries + // the metadata it just wrote. + result.Metadata, buildErr = applySnapshotsParam(r, result.Metadata) + } if buildErr != nil { glog.Errorf("Iceberg: LoadTable %s: %v", tableName, buildErr) writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata") @@ -567,6 +572,46 @@ func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTable }, nil } +// applySnapshotsParam honours ?snapshots=refs, which asks for only the +// snapshots that branches and tags point at. Clients use it to avoid pulling a +// long snapshot history they will not read. Anything else, including the +// default, returns the metadata untouched. +func applySnapshotsParam(r *http.Request, metadata table.Metadata) (table.Metadata, error) { + if !strings.EqualFold(strings.TrimSpace(r.URL.Query().Get("snapshots")), "refs") { + return metadata, nil + } + if metadata == nil { + return metadata, nil + } + + referenced := make(map[int64]struct{}) + for _, ref := range metadata.Refs() { + referenced[ref.SnapshotID] = struct{}{} + } + if current := metadata.CurrentSnapshot(); current != nil { + referenced[current.SnapshotID] = struct{}{} + } + + var unreferenced []int64 + for _, snapshot := range metadata.Snapshots() { + if _, keep := referenced[snapshot.SnapshotID]; !keep { + unreferenced = append(unreferenced, snapshot.SnapshotID) + } + } + if len(unreferenced) == 0 { + return metadata, nil + } + + builder, err := table.MetadataBuilderFromBase(metadata, "") + if err != nil { + return nil, err + } + if err := builder.RemoveSnapshots(unreferenced, false); err != nil { + return nil, err + } + return builder.Build() +} + // buildFileIOConfig returns the FileIO properties to advertise to catalog // clients so they can read the table's data files directly from S3 without // separately discovering the endpoint. The region defaults to the same diff --git a/weed/s3api/iceberg/handlers_view.go b/weed/s3api/iceberg/handlers_view.go index 2a08b21d8..108ca5226 100644 --- a/weed/s3api/iceberg/handlers_view.go +++ b/weed/s3api/iceberg/handlers_view.go @@ -387,3 +387,61 @@ func isViewAlreadyExists(err error) bool { } return strings.Contains(strings.ToLower(err.Error()), "already exists") } + +// handleRenameView moves a view's catalog pointer to a new namespace/name, +// the view counterpart of POST /v1/{prefix}/tables/rename. +func (s *Server) handleRenameView(w http.ResponseWriter, r *http.Request) { + var req RenameTableRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeError(w, http.StatusBadRequest, "BadRequestException", "Invalid request body") + return + } + + source := parseNamespace(encodeNamespace(req.Source.Namespace)) + dest := parseNamespace(encodeNamespace(req.Destination.Namespace)) + if len(source) == 0 || req.Source.Name == "" || len(dest) == 0 || req.Destination.Name == "" { + writeError(w, http.StatusBadRequest, "BadRequestException", "source and destination namespace and name are required") + return + } + + bucketName := getBucketFromPrefix(r) + identityName := s3_constants.GetIdentityNameFromContext(r) + + renameReq := &s3tables.RenameTableRequest{ + TableBucketARN: buildTableBucketARN(bucketName), + SourceNamespace: source, + SourceName: req.Source.Name, + DestNamespace: dest, + DestName: req.Destination.Name, + } + + err := s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + mgrClient := s3tables.NewManagerClient(client) + return s.tablesManager.Execute(r.Context(), mgrClient, "RenameView", renameReq, nil, identityName) + }) + + if err != nil { + var viewErr *s3tables.S3TablesError + if errors.As(err, &viewErr) { + switch viewErr.Type { + case s3tables.ErrCodeNoSuchView: + writeError(w, http.StatusNotFound, "NoSuchViewException", fmt.Sprintf("View does not exist: %s", req.Source.Name)) + return + case s3tables.ErrCodeNoSuchNamespace: + writeError(w, http.StatusNotFound, "NoSuchNamespaceException", fmt.Sprintf("Namespace does not exist: %v", dest)) + return + case s3tables.ErrCodeViewAlreadyExists: + writeError(w, http.StatusConflict, "AlreadyExistsException", fmt.Sprintf("View already exists: %s", req.Destination.Name)) + return + case s3tables.ErrCodeInvalidRequest: + writeError(w, http.StatusBadRequest, "BadRequestException", viewErr.Message) + return + } + } + glog.V(1).Infof("Iceberg: RenameView error: %v", err) + writeManagerError(w, err) + return + } + + w.WriteHeader(http.StatusNoContent) +} diff --git a/weed/s3api/iceberg/iceberg_snapshots_param_test.go b/weed/s3api/iceberg/iceberg_snapshots_param_test.go new file mode 100644 index 000000000..85c1c02d0 --- /dev/null +++ b/weed/s3api/iceberg/iceberg_snapshots_param_test.go @@ -0,0 +1,93 @@ +package iceberg + +import ( + "net/http/httptest" + "testing" + "time" + + "github.com/apache/iceberg-go/table" + "github.com/google/uuid" +) + +// metadataWithSnapshots builds metadata whose main branch is on the last +// snapshot, optionally tagging one of the earlier ones. +func metadataWithSnapshots(t *testing.T, ids []int64, tagged int64) table.Metadata { + t.Helper() + + base, err := newTableMetadata(uuid.New(), "s3://bkt/ns/t", nil, nil, nil, nil) + if err != nil { + t.Fatalf("newTableMetadata() error = %v", err) + } + builder, err := table.MetadataBuilderFromBase(base, "") + if err != nil { + t.Fatalf("MetadataBuilderFromBase() error = %v", err) + } + + now := time.Now().UnixMilli() + for i, id := range ids { + snapshot := table.Snapshot{ + SnapshotID: id, + TimestampMs: now + int64(i), + ManifestList: "metadata/snap-" + string(rune('0'+i)) + ".avro", + } + if err := builder.AddSnapshot(&snapshot); err != nil { + t.Fatalf("AddSnapshot(%d) error = %v", id, err) + } + } + if err := builder.SetSnapshotRef(table.MainBranch, ids[len(ids)-1], table.BranchRef); err != nil { + t.Fatalf("SetSnapshotRef(main) error = %v", err) + } + if tagged != 0 { + if err := builder.SetSnapshotRef("release", tagged, table.TagRef); err != nil { + t.Fatalf("SetSnapshotRef(release) error = %v", err) + } + } + + metadata, err := builder.Build() + if err != nil { + t.Fatalf("Build() error = %v", err) + } + return metadata +} + +func snapshotIDs(metadata table.Metadata) map[int64]bool { + ids := map[int64]bool{} + for _, snapshot := range metadata.Snapshots() { + ids[snapshot.SnapshotID] = true + } + return ids +} + +func TestApplySnapshotsParamRefsKeepsOnlyReferenced(t *testing.T) { + metadata := metadataWithSnapshots(t, []int64{1, 2, 3}, 1) + + got, err := applySnapshotsParam(httptest.NewRequest("GET", "/v1/namespaces/ns/tables/t?snapshots=refs", nil), metadata) + if err != nil { + t.Fatalf("applySnapshotsParam() error = %v", err) + } + + ids := snapshotIDs(got) + if !ids[1] { + t.Error("tagged snapshot 1 was dropped") + } + if !ids[3] { + t.Error("current snapshot 3 was dropped") + } + if ids[2] { + t.Error("unreferenced snapshot 2 was returned") + } +} + +func TestApplySnapshotsParamDefaultsToAll(t *testing.T) { + metadata := metadataWithSnapshots(t, []int64{1, 2, 3}, 1) + + for _, target := range []string{"/v1/namespaces/ns/tables/t", "/v1/namespaces/ns/tables/t?snapshots=all"} { + got, err := applySnapshotsParam(httptest.NewRequest("GET", target, nil), metadata) + if err != nil { + t.Fatalf("applySnapshotsParam(%s) error = %v", target, err) + } + if len(got.Snapshots()) != 3 { + t.Errorf("applySnapshotsParam(%s) kept %d snapshots, want 3", target, len(got.Snapshots())) + } + } +} diff --git a/weed/s3api/iceberg/server.go b/weed/s3api/iceberg/server.go index 366853bf1..9a3709f59 100644 --- a/weed/s3api/iceberg/server.go +++ b/weed/s3api/iceberg/server.go @@ -116,6 +116,8 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleViewExists)).Methods(http.MethodHead) router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleDropView)).Methods(http.MethodDelete) router.HandleFunc("/v1/namespaces/{namespace}/views/{view}", s.Auth(s.handleUpdateView)).Methods(http.MethodPost) + router.HandleFunc("/v1/views/rename", s.Auth(s.handleRenameView)).Methods(http.MethodPost) + router.HandleFunc("/v1/namespaces/{namespace}/tables/{table}/metrics", s.Auth(s.handleReportMetrics)).Methods(http.MethodPost) // Multi-table transaction commit - wrapped with Auth middleware router.HandleFunc("/v1/transactions/commit", s.Auth(s.handleCommitTransaction)).Methods(http.MethodPost) @@ -141,6 +143,8 @@ func (s *Server) RegisterRoutes(router *mux.Router) { router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleViewExists)).Methods(http.MethodHead) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleDropView)).Methods(http.MethodDelete) router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/views/{view}", s.Auth(s.handleUpdateView)).Methods(http.MethodPost) + router.HandleFunc("/v1/{prefix}/views/rename", s.Auth(s.handleRenameView)).Methods(http.MethodPost) + router.HandleFunc("/v1/{prefix}/namespaces/{namespace}/tables/{table}/metrics", s.Auth(s.handleReportMetrics)).Methods(http.MethodPost) router.HandleFunc("/v1/{prefix}/transactions/commit", s.Auth(s.handleCommitTransaction)).Methods(http.MethodPost) // Catch-all for debugging diff --git a/weed/s3api/iceberg/server_routes_test.go b/weed/s3api/iceberg/server_routes_test.go new file mode 100644 index 000000000..67417afda --- /dev/null +++ b/weed/s3api/iceberg/server_routes_test.go @@ -0,0 +1,41 @@ +package iceberg + +import ( + "net/http" + "net/http/httptest" + "testing" + + "github.com/gorilla/mux" +) + +// The catch-all answers anything unrouted, so a path that reaches it is a +// route that does not exist. +func TestRegisteredRoutesReachAHandler(t *testing.T) { + router := mux.NewRouter().SkipClean(true) + (&Server{}).RegisterRoutes(router) + + cases := []struct { + method string + target string + }{ + {http.MethodPost, "/v1/views/rename"}, + {http.MethodPost, "/v1/warehouse/views/rename"}, + {http.MethodPost, "/v1/namespaces/ns/tables/t/metrics"}, + {http.MethodPost, "/v1/warehouse/namespaces/ns/tables/t/metrics"}, + } + + for _, tc := range cases { + var match mux.RouteMatch + if !router.Match(httptest.NewRequest(tc.method, tc.target, nil), &match) { + t.Errorf("%s %s matched no route", tc.method, tc.target) + continue + } + if match.MatchErr != nil { + t.Errorf("%s %s: %v", tc.method, tc.target, match.MatchErr) + continue + } + if tmpl, err := match.Route.GetPathTemplate(); err == nil && tmpl == "/" { + t.Errorf("%s %s fell through to the catch-all", tc.method, tc.target) + } + } +} diff --git a/weed/s3api/s3tables/handler.go b/weed/s3api/s3tables/handler.go index 6b42c21cb..9e813660e 100644 --- a/weed/s3api/s3tables/handler.go +++ b/weed/s3api/s3tables/handler.go @@ -178,6 +178,8 @@ func (h *S3TablesHandler) HandleRequest(w http.ResponseWriter, r *http.Request, err = h.handleUpdateView(w, r, filerClient) case "DeleteView": err = h.handleDeleteView(w, r, filerClient) + case "RenameView": + err = h.handleRenameView(w, r, filerClient) // Table Policy operations case "PutTablePolicy": diff --git a/weed/s3api/s3tables/handler_table.go b/weed/s3api/s3tables/handler_table.go index 04014ac2a..4e5a357a9 100644 --- a/weed/s3api/s3tables/handler_table.go +++ b/weed/s3api/s3tables/handler_table.go @@ -1234,13 +1234,63 @@ var renamedTableAttributes = []string{ ExtendedKeyTags, ExtendedKeyMaintenance, ExtendedKeyMaintenanceStatus, + ExtendedKeyEntryType, } +// catalogEntryKind describes the entry a rename operates on, so tables and +// views share one implementation of the catalog-only move. +type catalogEntryKind struct { + entryType string + noun string + renameOp string + createOp string + notFoundCode string + existsCode string + // resourceARN builds the ARN a policy scoped to this entry would name, so a + // view is authorized against its view ARN and not a table ARN. + resourceARN func(h *S3TablesHandler, ownerAccountID, bucketName, id string) string +} + +var ( + tableEntryKind = catalogEntryKind{ + entryType: EntryTypeTable, + noun: "table", + renameOp: "RenameTable", + createOp: "CreateTable", + notFoundCode: ErrCodeNoSuchTable, + existsCode: ErrCodeTableAlreadyExists, + resourceARN: func(h *S3TablesHandler, ownerAccountID, bucketName, id string) string { + return h.generateTableARN(ownerAccountID, bucketName, id) + }, + } + viewEntryKind = catalogEntryKind{ + entryType: EntryTypeView, + noun: "view", + renameOp: "RenameView", + createOp: "CreateView", + notFoundCode: ErrCodeNoSuchView, + existsCode: ErrCodeViewAlreadyExists, + resourceARN: func(h *S3TablesHandler, ownerAccountID, bucketName, id string) string { + return h.generateViewARN(ownerAccountID, bucketName, id) + }, + } +) + // handleRenameTable moves a table's catalog entry to a new namespace/name within // the same bucket. It is catalog-only: the metadata.json and data files stay put, // the destination keeps the source's MetadataLocation, and the source name is // soft-deleted in place (its catalog xattrs are dropped, its data is left intact). func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + return h.renameCatalogEntry(w, r, filerClient, tableEntryKind) +} + +// handleRenameView is handleRenameTable for views, which live in the same +// namespace directory under the same name rules. +func (h *S3TablesHandler) handleRenameView(w http.ResponseWriter, r *http.Request, filerClient FilerClient) error { + return h.renameCatalogEntry(w, r, filerClient, viewEntryKind) +} + +func (h *S3TablesHandler) renameCatalogEntry(w http.ResponseWriter, r *http.Request, filerClient FilerClient, kind catalogEntryKind) error { var req RenameTableRequest if err := h.readRequestBody(r, &req); err != nil { h.writeError(w, http.StatusBadRequest, ErrCodeInvalidRequest, err.Error()) @@ -1295,6 +1345,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque var bucketTags map[string]string var tableTags map[string]string var bucketMetadata tableBucketMetadata + var srcExtended map[string][]byte err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { data, err := h.getExtendedAttribute(r.Context(), client, srcPath, ExtendedKeyMetadata) if err != nil { @@ -1325,6 +1376,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque if err != nil { return err } + srcExtended = srcEntry.Extended copiedFromSource = make(map[string][]byte, len(renamedTableAttributes)) for _, key := range renamedTableAttributes { copiedFromSource[key] = srcEntry.Extended[key] @@ -1373,19 +1425,26 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque }) if err != nil { - if errors.Is(err, filer_pb.ErrNotFound) { - h.writeError(w, http.StatusNotFound, ErrCodeNoSuchTable, fmt.Sprintf("table %s not found", srcName)) + if errors.Is(err, filer_pb.ErrNotFound) || errors.Is(err, ErrAttributeNotFound) { + h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName)) } else { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check table: %v", err)) + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, fmt.Sprintf("failed to check %s: %v", kind.noun, err)) } return err } - tableARN := h.generateTableARN(metadata.OwnerAccountID, bucketName, srcNamespace+"/"+srcName) + // Tables and views share the namespace directory, so a rename must not pick + // up the other kind under the same name. + if entryType(srcExtended) != kind.entryType { + h.writeError(w, http.StatusNotFound, kind.notFoundCode, fmt.Sprintf("%s %s not found", kind.noun, srcName)) + return fmt.Errorf("%s %s not found", kind.noun, srcName) + } + + tableARN := kind.resourceARN(h, metadata.OwnerAccountID, bucketName, srcNamespace+"/"+srcName) bucketARN := h.generateTableBucketARN(bucketMetadata.OwnerAccountID, bucketName) principal := h.getAccountID(r) identityActions := getIdentityActions(r) - tableAllowed := CheckPermissionWithContext("RenameTable", principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{ + tableAllowed := CheckPermissionWithContext(kind.renameOp, principal, metadata.OwnerAccountID, tablePolicy, tableARN, &PolicyContext{ TableBucketName: bucketName, Namespace: srcNamespace, TableName: srcName, @@ -1394,7 +1453,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque IdentityActions: identityActions, DefaultAllow: h.defaultAllowFor(r), }) - bucketAllowed := CheckPermissionWithContext("RenameTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ + bucketAllowed := CheckPermissionWithContext(kind.renameOp, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ TableBucketName: bucketName, Namespace: srcNamespace, TableName: srcName, @@ -1404,8 +1463,8 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque DefaultAllow: h.defaultAllowFor(r), }) if !tableAllowed && !bucketAllowed { - h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to rename table") - return NewAuthError("RenameTable", principal, "not authorized to rename table") + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to rename "+kind.noun) + return NewAuthError(kind.renameOp, principal, "not authorized to rename "+kind.noun) } // Require the destination namespace to exist and the destination table to be free. @@ -1436,7 +1495,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque if err != nil { if errors.Is(err, ErrTableAlreadyExists) { - h.writeError(w, http.StatusConflict, ErrCodeTableAlreadyExists, fmt.Sprintf("table %s already exists", destName)) + h.writeError(w, http.StatusConflict, kind.existsCode, fmt.Sprintf("%s %s already exists", kind.noun, destName)) } else if errors.Is(err, filer_pb.ErrNotFound) { h.writeError(w, http.StatusNotFound, ErrCodeNoSuchNamespace, fmt.Sprintf("namespace %s not found", destNamespace)) } else { @@ -1448,7 +1507,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque // Renaming places the table into the destination namespace, so the principal // must also be allowed to create a table there (the source check alone lets a // caller move tables into namespaces they don't control). - destNamespaceAllowed := CheckPermissionWithContext("CreateTable", principal, destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN, &PolicyContext{ + destNamespaceAllowed := CheckPermissionWithContext(kind.createOp, principal, destNamespaceMetadata.OwnerAccountID, destNamespacePolicy, bucketARN, &PolicyContext{ TableBucketName: bucketName, Namespace: destNamespace, TableName: destName, @@ -1456,7 +1515,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque IdentityActions: identityActions, DefaultAllow: h.defaultAllowFor(r), }) - destBucketAllowed := CheckPermissionWithContext("CreateTable", principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ + destBucketAllowed := CheckPermissionWithContext(kind.createOp, principal, bucketMetadata.OwnerAccountID, bucketPolicy, bucketARN, &PolicyContext{ TableBucketName: bucketName, Namespace: destNamespace, TableName: destName, @@ -1465,8 +1524,8 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque DefaultAllow: h.defaultAllowFor(r), }) if !destNamespaceAllowed && !destBucketAllowed { - h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create table in the destination namespace") - return NewAuthError("RenameTable", principal, "not authorized to create table in the destination namespace") + h.writeError(w, http.StatusForbidden, ErrCodeAccessDenied, "not authorized to create "+kind.noun+" in the destination namespace") + return NewAuthError(kind.renameOp, principal, "not authorized to create "+kind.noun+" in the destination namespace") } metadata.Name = destName @@ -1475,7 +1534,7 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque metadataBytes, err := json.Marshal(&metadata) if err != nil { - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal table metadata") + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to marshal "+kind.noun+" metadata") return fmt.Errorf("failed to marshal metadata: %w", err) } @@ -1488,6 +1547,9 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadata, metadataBytes); err != nil { return err } + if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyEntryType, []byte(kind.entryType)); err != nil { + return err + } if len(metadataVersionXattr) > 0 { if err := h.setExtendedAttribute(r.Context(), client, destPath, ExtendedKeyMetadataVersion, metadataVersionXattr); err != nil { return err @@ -1526,15 +1588,15 @@ func (h *S3TablesHandler) handleRenameTable(w http.ResponseWriter, r *http.Reque if err != nil { if errors.Is(err, ErrConcurrentUpdate) { - h.writeError(w, http.StatusConflict, ErrCodeConflict, "table changed during rename, retry the request") + h.writeError(w, http.StatusConflict, ErrCodeConflict, kind.noun+" changed during rename, retry the request") return err } - h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to rename table") + h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to rename "+kind.noun) return err } h.writeJSON(w, http.StatusOK, &RenameTableResponse{ - TableARN: h.generateTableARN(metadata.OwnerAccountID, bucketName, destNamespace+"/"+destName), + TableARN: kind.resourceARN(h, metadata.OwnerAccountID, bucketName, destNamespace+"/"+destName), MetadataLocation: metadata.MetadataLocation, }) return nil diff --git a/weed/s3api/s3tables/handler_view_rename_test.go b/weed/s3api/s3tables/handler_view_rename_test.go new file mode 100644 index 000000000..237f2f141 --- /dev/null +++ b/weed/s3api/s3tables/handler_view_rename_test.go @@ -0,0 +1,134 @@ +package s3tables + +import ( + "context" + "encoding/json" + "path" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// seedView adds a view alongside the table the rename harness already creates. +func seedView(t *testing.T, fs *memFilerServer, name string) { + t.Helper() + + viewMeta, err := json.Marshal(tableMetadataInternal{ + Name: name, + Namespace: "ns", + Format: "ICEBERG", + OwnerAccountID: DefaultAccountID, + MetadataVersion: 1, + MetadataLocation: "s3://" + renameTestBucket + "/ns/" + name + "/metadata/v1.metadata.json", + }) + require.NoError(t, err) + + fs.putEntry(GetNamespacePath(renameTestBucket, "ns"), name, map[string][]byte{ + ExtendedKeyMetadata: viewMeta, + ExtendedKeyMetadataVersion: []byte("1"), + ExtendedKeyEntryType: []byte(EntryTypeView), + }) + viewPath := GetTablePath(renameTestBucket, "ns", name) + fs.putEntry(viewPath, "metadata", nil) + fs.putEntry(path.Join(viewPath, "metadata"), "v1.metadata.json", nil) +} + +func runRenameView(t *testing.T, m *Manager, fs *memFilerServer, sourceName, destName string) error { + t.Helper() + return m.Execute(context.Background(), NewManagerClient(fs.client), "RenameView", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: sourceName, + DestNamespace: []string{"ns"}, + DestName: destName, + }, nil, "") +} + +func TestRenameViewMovesCatalogPointer(t *testing.T) { + fs, m := startRenameManager(t) + seedView(t, fs, "v") + + require.NoError(t, runRenameView(t, m, fs, "v", "v2")) + + dest := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "v2") + require.NotNil(t, dest) + assert.Equal(t, EntryTypeView, entryType(dest.Extended), "destination must stay a view") + + var moved tableMetadataInternal + require.NoError(t, json.Unmarshal(dest.Extended[ExtendedKeyMetadata], &moved)) + assert.Equal(t, "v2", moved.Name) + assert.Equal(t, "s3://"+renameTestBucket+"/ns/v/metadata/v1.metadata.json", moved.MetadataLocation, + "rename is catalog-only, the metadata stays where it was written") + + src := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "v") + require.NotNil(t, src) + _, stillListed := src.Extended[ExtendedKeyMetadata] + assert.False(t, stillListed, "source name must stop resolving") + assert.NotNil(t, fs.getEntry(path.Join(GetTablePath(renameTestBucket, "ns", "v"), "metadata"), "v1.metadata.json"), + "the view's metadata file must survive") +} + +// Tables and views share one namespace directory, so each rename must refuse +// the other kind rather than moving it. +func TestRenameViewRejectsTable(t *testing.T) { + fs, m := startRenameManager(t) + + err := runRenameView(t, m, fs, "t", "t2") + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchView, s3Err.Type) +} + +func TestRenameTableRejectsView(t *testing.T) { + fs, m := startRenameManager(t) + seedView(t, fs, "v") + + err := runRename(t, m, fs, &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "v", + DestNamespace: []string{"ns"}, + DestName: "v2", + }) + require.Error(t, err) + var s3Err *S3TablesError + require.ErrorAs(t, err, &s3Err) + assert.Equal(t, ErrCodeNoSuchTable, s3Err.Type) +} + +// A policy scoped to the view ARN must authorize renaming that view. Checking +// the source against a table ARN would silently ignore it. +func TestRenameViewAuthorizesAgainstTheViewARN(t *testing.T) { + fs, m := startRenameManager(t) + m.SetTrusted(false) + m.SetDefaultAllow(false) + seedView(t, fs, "v") + + const principal = "analyst" + viewARN := "arn:aws:s3tables:" + DefaultRegion + ":" + DefaultAccountID + ":bucket/" + renameTestBucket + "/view/ns/v" + viewPolicy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"` + principal + + `","Action":"s3tables:RenameView","Resource":"` + viewARN + `"}]}` + view := fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "v") + require.NotNil(t, view) + view.Extended[ExtendedKeyPolicy] = []byte(viewPolicy) + + // Landing in the namespace needs create permission there as well. + namespacePolicy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"` + principal + + `","Action":"s3tables:CreateView","Resource":"` + mustBucketARN(t) + `"}]}` + namespace := fs.getEntry(GetTableBucketPath(renameTestBucket), "ns") + require.NotNil(t, namespace) + namespace.Extended[ExtendedKeyPolicy] = []byte(namespacePolicy) + + err := m.Execute(context.Background(), NewManagerClient(fs.client), "RenameView", &RenameTableRequest{ + TableBucketARN: mustBucketARN(t), + SourceNamespace: []string{"ns"}, + SourceName: "v", + DestNamespace: []string{"ns"}, + DestName: "v2", + }, nil, principal) + require.NoError(t, err) + + assert.NotNil(t, fs.getEntry(GetNamespacePath(renameTestBucket, "ns"), "v2")) +}