Mirror
/
seaweedfs
mirror of https://github.com/seaweedfs/seaweedfs.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

s3tables: propagate request context to filer operations

pull/8147/head
Chris Lu 4 days ago
parent
6cdd34da77
commit
b30631c3b5
6 changed files with 68 additions and 66 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 28
      weed/s3api/s3tables/filer_ops.go
  2. 9
      weed/s3api/s3tables/handler_bucket_create.go
  3. 9
      weed/s3api/s3tables/handler_bucket_get_list_delete.go
  4. 19
      weed/s3api/s3tables/handler_namespace.go
  5. 37
      weed/s3api/s3tables/handler_policy.go
  6. 32
      weed/s3api/s3tables/handler_table.go

28
weed/s3api/s3tables/filer_ops.go
View File

@ -11,9 +11,9 @@ import (
// Filer operations - Common functions for interacting with the filer // Filer operations - Common functions for interacting with the filer
// createDirectory creates a new directory at the specified path // createDirectory creates a new directory at the specified path
func (h *S3TablesHandler) createDirectory(client filer_pb.SeaweedFilerClient, path string) error {
func (h *S3TablesHandler) createDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
dir, name := splitPath(path) dir, name := splitPath(path)
_, err := client.CreateEntry(context.Background(), &filer_pb.CreateEntryRequest{
_, err := client.CreateEntry(ctx, &filer_pb.CreateEntryRequest{
Directory: dir, Directory: dir,
Entry: &filer_pb.Entry{ Entry: &filer_pb.Entry{
Name: name, Name: name,
@ -29,11 +29,11 @@ func (h *S3TablesHandler) createDirectory(client filer_pb.SeaweedFilerClient, pa
} }
// setExtendedAttribute sets an extended attribute on an existing entry // setExtendedAttribute sets an extended attribute on an existing entry
func (h *S3TablesHandler) setExtendedAttribute(client filer_pb.SeaweedFilerClient, path, key string, data []byte) error {
func (h *S3TablesHandler) setExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string, data []byte) error {
dir, name := splitPath(path) dir, name := splitPath(path)
// First, get the existing entry // First, get the existing entry
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
resp, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir, Directory: dir,
Name: name, Name: name,
}) })
@ -53,7 +53,7 @@ func (h *S3TablesHandler) setExtendedAttribute(client filer_pb.SeaweedFilerClien
entry.Extended[key] = data entry.Extended[key] = data
// Save the updated entry // Save the updated entry
_, err = client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
Directory: dir, Directory: dir,
Entry: entry, Entry: entry,
}) })
@ -61,9 +61,9 @@ func (h *S3TablesHandler) setExtendedAttribute(client filer_pb.SeaweedFilerClien
} }
// getExtendedAttribute gets an extended attribute from an entry // getExtendedAttribute gets an extended attribute from an entry
func (h *S3TablesHandler) getExtendedAttribute(client filer_pb.SeaweedFilerClient, path, key string) ([]byte, error) {
func (h *S3TablesHandler) getExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) ([]byte, error) {
dir, name := splitPath(path) dir, name := splitPath(path)
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
resp, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir, Directory: dir,
Name: name, Name: name,
}) })
@ -84,11 +84,11 @@ func (h *S3TablesHandler) getExtendedAttribute(client filer_pb.SeaweedFilerClien
} }
// deleteExtendedAttribute deletes an extended attribute from an entry // deleteExtendedAttribute deletes an extended attribute from an entry
func (h *S3TablesHandler) deleteExtendedAttribute(client filer_pb.SeaweedFilerClient, path, key string) error {
func (h *S3TablesHandler) deleteExtendedAttribute(ctx context.Context, client filer_pb.SeaweedFilerClient, path, key string) error {
dir, name := splitPath(path) dir, name := splitPath(path)
// Get the existing entry // Get the existing entry
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
resp, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir, Directory: dir,
Name: name, Name: name,
}) })
@ -107,7 +107,7 @@ func (h *S3TablesHandler) deleteExtendedAttribute(client filer_pb.SeaweedFilerCl
} }
// Save the updated entry // Save the updated entry
_, err = client.UpdateEntry(context.Background(), &filer_pb.UpdateEntryRequest{
_, err = client.UpdateEntry(ctx, &filer_pb.UpdateEntryRequest{
Directory: dir, Directory: dir,
Entry: entry, Entry: entry,
}) })
@ -115,9 +115,9 @@ func (h *S3TablesHandler) deleteExtendedAttribute(client filer_pb.SeaweedFilerCl
} }
// deleteDirectory deletes a directory and all its contents // deleteDirectory deletes a directory and all its contents
func (h *S3TablesHandler) deleteDirectory(client filer_pb.SeaweedFilerClient, path string) error {
func (h *S3TablesHandler) deleteDirectory(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) error {
dir, name := splitPath(path) dir, name := splitPath(path)
_, err := client.DeleteEntry(context.Background(), &filer_pb.DeleteEntryRequest{
_, err := client.DeleteEntry(ctx, &filer_pb.DeleteEntryRequest{
Directory: dir, Directory: dir,
Name: name, Name: name,
IsDeleteData: true, IsDeleteData: true,
@ -128,9 +128,9 @@ func (h *S3TablesHandler) deleteDirectory(client filer_pb.SeaweedFilerClient, pa
} }
// entryExists checks if an entry exists at the given path // entryExists checks if an entry exists at the given path
func (h *S3TablesHandler) entryExists(client filer_pb.SeaweedFilerClient, path string) bool {
func (h *S3TablesHandler) entryExists(ctx context.Context, client filer_pb.SeaweedFilerClient, path string) bool {
dir, name := splitPath(path) dir, name := splitPath(path)
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
resp, err := client.LookupDirectoryEntry(ctx, &filer_pb.LookupDirectoryEntryRequest{
Directory: dir, Directory: dir,
Name: name, Name: name,
}) })

9
weed/s3api/s3tables/handler_bucket_create.go
View File

@ -1,7 +1,6 @@
package s3tables package s3tables
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
@ -48,7 +47,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
// Check if bucket already exists // Check if bucket already exists
exists := false exists := false
err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.LookupDirectoryEntry(context.Background(), &filer_pb.LookupDirectoryEntryRequest{
resp, err := client.LookupDirectoryEntry(r.Context(), &filer_pb.LookupDirectoryEntryRequest{
Directory: TablesPath, Directory: TablesPath,
Name: req.Name, Name: req.Name,
}) })
@ -90,12 +89,12 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
// Create bucket directory // Create bucket directory
if err := h.createDirectory(client, bucketPath); err != nil {
if err := h.createDirectory(r.Context(), client, bucketPath); err != nil {
return err return err
} }
// Set metadata as extended attribute // Set metadata as extended attribute
if err := h.setExtendedAttribute(client, bucketPath, ExtendedKeyMetadata, metadataBytes); err != nil {
if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata, metadataBytes); err != nil {
return err return err
} }
@ -105,7 +104,7 @@ func (h *S3TablesHandler) handleCreateTableBucket(w http.ResponseWriter, r *http
if err != nil { if err != nil {
return fmt.Errorf("failed to marshal tags: %w", err) return fmt.Errorf("failed to marshal tags: %w", err)
} }
if err := h.setExtendedAttribute(client, bucketPath, ExtendedKeyTags, tagsBytes); err != nil {
if err := h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyTags, tagsBytes); err != nil {
return err return err
} }
} }

9
weed/s3api/s3tables/handler_bucket_get_list_delete.go
View File

@ -1,7 +1,6 @@
package s3tables package s3tables
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
@ -40,7 +39,7 @@ func (h *S3TablesHandler) handleGetTableBucket(w http.ResponseWriter, r *http.Re
var metadata tableBucketMetadata var metadata tableBucketMetadata
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, bucketPath, ExtendedKeyMetadata)
data, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
if err != nil { if err != nil {
return err return err
} }
@ -79,7 +78,7 @@ func (h *S3TablesHandler) handleListTableBuckets(w http.ResponseWriter, r *http.
var buckets []TableBucketSummary var buckets []TableBucketSummary
err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err := filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{
Directory: TablesPath, Directory: TablesPath,
Limit: uint32(maxBuckets), Limit: uint32(maxBuckets),
}) })
@ -178,7 +177,7 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
// Check if bucket exists and is empty // Check if bucket exists and is empty
hasChildren := false hasChildren := false
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{
Directory: bucketPath, Directory: bucketPath,
Limit: 10, Limit: 10,
}) })
@ -207,7 +206,7 @@ func (h *S3TablesHandler) handleDeleteTableBucket(w http.ResponseWriter, r *http
// Delete the bucket // Delete the bucket
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.deleteDirectory(client, bucketPath)
return h.deleteDirectory(r.Context(), client, bucketPath)
}) })
if err != nil { if err != nil {

19
weed/s3api/s3tables/handler_namespace.go
View File

@ -1,7 +1,6 @@
package s3tables package s3tables
import ( import (
"context"
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
@ -48,7 +47,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
bucketPath := getTableBucketPath(bucketName) bucketPath := getTableBucketPath(bucketName)
var bucketExists bool var bucketExists bool
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, bucketPath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
bucketExists = err == nil bucketExists = err == nil
return nil return nil
}) })
@ -63,7 +62,7 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
// Check if namespace already exists // Check if namespace already exists
exists := false exists := false
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, namespacePath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
exists = err == nil exists = err == nil
return nil return nil
}) })
@ -89,12 +88,12 @@ func (h *S3TablesHandler) handleCreateNamespace(w http.ResponseWriter, r *http.R
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
// Create namespace directory // Create namespace directory
if err := h.createDirectory(client, namespacePath); err != nil {
if err := h.createDirectory(r.Context(), client, namespacePath); err != nil {
return err return err
} }
// Set metadata as extended attribute // Set metadata as extended attribute
if err := h.setExtendedAttribute(client, namespacePath, ExtendedKeyMetadata, metadataBytes); err != nil {
if err := h.setExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata, metadataBytes); err != nil {
return err return err
} }
@ -136,9 +135,10 @@ func (h *S3TablesHandler) handleGetNamespace(w http.ResponseWriter, r *http.Requ
namespacePath := getNamespacePath(bucketName, req.Namespace) namespacePath := getNamespacePath(bucketName, req.Namespace)
// Get namespace
var metadata namespaceMetadata var metadata namespaceMetadata
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, namespacePath, ExtendedKeyMetadata)
data, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
if err != nil { if err != nil {
return err return err
} }
@ -187,8 +187,9 @@ func (h *S3TablesHandler) handleListNamespaces(w http.ResponseWriter, r *http.Re
bucketPath := getTableBucketPath(bucketName) bucketPath := getTableBucketPath(bucketName)
var namespaces []NamespaceSummary var namespaces []NamespaceSummary
// List namespaces
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{
Directory: bucketPath, Directory: bucketPath,
Limit: uint32(maxNamespaces), Limit: uint32(maxNamespaces),
}) })
@ -272,7 +273,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
// Check if namespace exists and is empty // Check if namespace exists and is empty
hasChildren := false hasChildren := false
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(r.Context(), &filer_pb.ListEntriesRequest{
Directory: namespacePath, Directory: namespacePath,
Limit: 10, Limit: 10,
}) })
@ -301,7 +302,7 @@ func (h *S3TablesHandler) handleDeleteNamespace(w http.ResponseWriter, r *http.R
// Delete the namespace // Delete the namespace
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.deleteDirectory(client, namespacePath)
return h.deleteDirectory(r.Context(), client, namespacePath)
}) })
if err != nil { if err != nil {

37
weed/s3api/s3tables/handler_policy.go
View File

@ -4,6 +4,7 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"net/http" "net/http"
"strings"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
) )
@ -36,7 +37,7 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
bucketPath := getTableBucketPath(bucketName) bucketPath := getTableBucketPath(bucketName)
var bucketExists bool var bucketExists bool
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, bucketPath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyMetadata)
bucketExists = err == nil bucketExists = err == nil
return nil return nil
}) })
@ -48,7 +49,7 @@ func (h *S3TablesHandler) handlePutTableBucketPolicy(w http.ResponseWriter, r *h
// Write policy // Write policy
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.setExtendedAttribute(client, bucketPath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
return h.setExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
}) })
if err != nil { if err != nil {
@ -83,7 +84,7 @@ func (h *S3TablesHandler) handleGetTableBucketPolicy(w http.ResponseWriter, r *h
var policy []byte var policy []byte
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
var readErr error var readErr error
policy, readErr = h.getExtendedAttribute(client, bucketPath, ExtendedKeyPolicy)
policy, readErr = h.getExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
return readErr return readErr
}) })
@ -121,11 +122,12 @@ func (h *S3TablesHandler) handleDeleteTableBucketPolicy(w http.ResponseWriter, r
bucketPath := getTableBucketPath(bucketName) bucketPath := getTableBucketPath(bucketName)
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.deleteExtendedAttribute(client, bucketPath, ExtendedKeyPolicy)
return h.deleteExtendedAttribute(r.Context(), client, bucketPath, ExtendedKeyPolicy)
}) })
if err != nil {
// Ignore error if policy doesn't exist
if err != nil && !strings.Contains(err.Error(), "not found") {
h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table bucket policy")
return err
} }
h.writeJSON(w, http.StatusOK, nil) h.writeJSON(w, http.StatusOK, nil)
@ -160,7 +162,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
tablePath := getTablePath(bucketName, req.Namespace, req.Name) tablePath := getTablePath(bucketName, req.Namespace, req.Name)
var tableExists bool var tableExists bool
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, tablePath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
tableExists = err == nil tableExists = err == nil
return nil return nil
}) })
@ -172,7 +174,7 @@ func (h *S3TablesHandler) handlePutTablePolicy(w http.ResponseWriter, r *http.Re
// Write policy // Write policy
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.setExtendedAttribute(client, tablePath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
return h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy, []byte(req.ResourcePolicy))
}) })
if err != nil { if err != nil {
@ -207,7 +209,7 @@ func (h *S3TablesHandler) handleGetTablePolicy(w http.ResponseWriter, r *http.Re
var policy []byte var policy []byte
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
var readErr error var readErr error
policy, readErr = h.getExtendedAttribute(client, tablePath, ExtendedKeyPolicy)
policy, readErr = h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy)
return readErr return readErr
}) })
@ -245,11 +247,12 @@ func (h *S3TablesHandler) handleDeleteTablePolicy(w http.ResponseWriter, r *http
tablePath := getTablePath(bucketName, req.Namespace, req.Name) tablePath := getTablePath(bucketName, req.Namespace, req.Name)
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.deleteExtendedAttribute(client, tablePath, ExtendedKeyPolicy)
return h.deleteExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyPolicy)
}) })
if err != nil {
// Ignore error if policy doesn't exist
if err != nil && !strings.Contains(err.Error(), "not found") {
h.writeError(w, http.StatusInternalServerError, ErrCodeInternalError, "failed to delete table policy")
return err
} }
h.writeJSON(w, http.StatusOK, nil) h.writeJSON(w, http.StatusOK, nil)
@ -284,7 +287,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
// Read existing tags and merge // Read existing tags and merge
existingTags := make(map[string]string) existingTags := make(map[string]string)
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, resourcePath, extendedKey)
data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
if err == nil { if err == nil {
json.Unmarshal(data, &existingTags) json.Unmarshal(data, &existingTags)
} }
@ -303,7 +306,7 @@ func (h *S3TablesHandler) handleTagResource(w http.ResponseWriter, r *http.Reque
return fmt.Errorf("failed to marshal tags: %w", err) return fmt.Errorf("failed to marshal tags: %w", err)
} }
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.setExtendedAttribute(client, resourcePath, extendedKey, tagsBytes)
return h.setExtendedAttribute(r.Context(), client, resourcePath, extendedKey, tagsBytes)
}) })
if err != nil { if err != nil {
@ -336,7 +339,7 @@ func (h *S3TablesHandler) handleListTagsForResource(w http.ResponseWriter, r *ht
tags := make(map[string]string) tags := make(map[string]string)
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, resourcePath, extendedKey)
data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
if err != nil { if err != nil {
return nil // Return empty tags if not found return nil // Return empty tags if not found
} }
@ -378,7 +381,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
// Read existing tags // Read existing tags
tags := make(map[string]string) tags := make(map[string]string)
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, resourcePath, extendedKey)
data, err := h.getExtendedAttribute(r.Context(), client, resourcePath, extendedKey)
if err != nil { if err != nil {
return nil return nil
} }
@ -397,7 +400,7 @@ func (h *S3TablesHandler) handleUntagResource(w http.ResponseWriter, r *http.Req
return fmt.Errorf("failed to marshal tags: %w", err) return fmt.Errorf("failed to marshal tags: %w", err)
} }
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.setExtendedAttribute(client, resourcePath, extendedKey, tagsBytes)
return h.setExtendedAttribute(r.Context(), client, resourcePath, extendedKey, tagsBytes)
}) })
if err != nil { if err != nil {

32
weed/s3api/s3tables/handler_table.go
View File

@ -61,7 +61,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
namespacePath := getNamespacePath(bucketName, req.Namespace) namespacePath := getNamespacePath(bucketName, req.Namespace)
var namespaceExists bool var namespaceExists bool
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, namespacePath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, namespacePath, ExtendedKeyMetadata)
namespaceExists = err == nil namespaceExists = err == nil
return nil return nil
}) })
@ -76,7 +76,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
// Check if table already exists // Check if table already exists
exists := false exists := false
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, tablePath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
exists = err == nil exists = err == nil
return nil return nil
}) })
@ -109,18 +109,18 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
// Create table directory // Create table directory
if err := h.createDirectory(client, tablePath); err != nil {
if err := h.createDirectory(r.Context(), client, tablePath); err != nil {
return err return err
} }
// Create data subdirectory for Iceberg files // Create data subdirectory for Iceberg files
dataPath := tablePath + "/data" dataPath := tablePath + "/data"
if err := h.createDirectory(client, dataPath); err != nil {
if err := h.createDirectory(r.Context(), client, dataPath); err != nil {
return err return err
} }
// Set metadata as extended attribute // Set metadata as extended attribute
if err := h.setExtendedAttribute(client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata, metadataBytes); err != nil {
return err return err
} }
@ -130,7 +130,7 @@ func (h *S3TablesHandler) handleCreateTable(w http.ResponseWriter, r *http.Reque
if err != nil { if err != nil {
return fmt.Errorf("failed to marshal tags: %w", err) return fmt.Errorf("failed to marshal tags: %w", err)
} }
if err := h.setExtendedAttribute(client, tablePath, ExtendedKeyTags, tagsBytes); err != nil {
if err := h.setExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyTags, tagsBytes); err != nil {
return err return err
} }
} }
@ -189,7 +189,7 @@ func (h *S3TablesHandler) handleGetTable(w http.ResponseWriter, r *http.Request,
var metadata tableMetadataInternal var metadata tableMetadataInternal
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := h.getExtendedAttribute(client, tablePath, ExtendedKeyMetadata)
data, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
if err != nil { if err != nil {
return err return err
} }
@ -247,10 +247,10 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
// If namespace is specified, list tables in that namespace only // If namespace is specified, list tables in that namespace only
if req.Namespace != "" { if req.Namespace != "" {
err = h.listTablesInNamespace(filerClient, bucketName, req.Namespace, req.Prefix, maxTables, &tables)
err = h.listTablesInNamespace(r.Context(), filerClient, bucketName, req.Namespace, req.Prefix, maxTables, &tables)
} else { } else {
// List tables in all namespaces // List tables in all namespaces
err = h.listTablesInAllNamespaces(filerClient, bucketName, req.Prefix, maxTables, &tables)
err = h.listTablesInAllNamespaces(r.Context(), filerClient, bucketName, req.Prefix, maxTables, &tables)
} }
if err != nil { if err != nil {
@ -265,11 +265,11 @@ func (h *S3TablesHandler) handleListTables(w http.ResponseWriter, r *http.Reques
return nil return nil
} }
func (h *S3TablesHandler) listTablesInNamespace(filerClient FilerClient, bucketName, namespace, prefix string, maxTables int, tables *[]TableSummary) error {
func (h *S3TablesHandler) listTablesInNamespace(ctx context.Context, filerClient FilerClient, bucketName, namespace, prefix string, maxTables int, tables *[]TableSummary) error {
namespacePath := getNamespacePath(bucketName, namespace) namespacePath := getNamespacePath(bucketName, namespace)
return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
Directory: namespacePath, Directory: namespacePath,
Limit: uint32(maxTables), Limit: uint32(maxTables),
}) })
@ -323,12 +323,12 @@ func (h *S3TablesHandler) listTablesInNamespace(filerClient FilerClient, bucketN
}) })
} }
func (h *S3TablesHandler) listTablesInAllNamespaces(filerClient FilerClient, bucketName, prefix string, maxTables int, tables *[]TableSummary) error {
func (h *S3TablesHandler) listTablesInAllNamespaces(ctx context.Context, filerClient FilerClient, bucketName, prefix string, maxTables int, tables *[]TableSummary) error {
bucketPath := getTableBucketPath(bucketName) bucketPath := getTableBucketPath(bucketName)
return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { return filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
// List all namespaces first // List all namespaces first
resp, err := client.ListEntries(context.Background(), &filer_pb.ListEntriesRequest{
resp, err := client.ListEntries(ctx, &filer_pb.ListEntriesRequest{
Directory: bucketPath, Directory: bucketPath,
Limit: 1000, Limit: 1000,
}) })
@ -354,7 +354,7 @@ func (h *S3TablesHandler) listTablesInAllNamespaces(filerClient FilerClient, buc
namespace := entry.Entry.Name namespace := entry.Entry.Name
// List tables in this namespace // List tables in this namespace
if err := h.listTablesInNamespace(filerClient, bucketName, namespace, prefix, maxTables-len(*tables), tables); err != nil {
if err := h.listTablesInNamespace(ctx, filerClient, bucketName, namespace, prefix, maxTables-len(*tables), tables); err != nil {
continue continue
} }
@ -391,7 +391,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
// Check if table exists // Check if table exists
var tableExists bool var tableExists bool
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, err := h.getExtendedAttribute(client, tablePath, ExtendedKeyMetadata)
_, err := h.getExtendedAttribute(r.Context(), client, tablePath, ExtendedKeyMetadata)
tableExists = err == nil tableExists = err == nil
return nil return nil
}) })
@ -403,7 +403,7 @@ func (h *S3TablesHandler) handleDeleteTable(w http.ResponseWriter, r *http.Reque
// Delete the table // Delete the table
err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { err = filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
return h.deleteDirectory(client, tablePath)
return h.deleteDirectory(r.Context(), client, tablePath)
}) })
if err != nil { if err != nil {

Write Preview
Loading…
Cancel
Save