feat: Add Iceberg REST Catalog server and admin UI (#8175)
* feat: Add Iceberg REST Catalog server Implement Iceberg REST Catalog API on a separate port (default 8181) that exposes S3 Tables metadata through the Apache Iceberg REST protocol. - Add new weed/s3api/iceberg package with REST handlers - Implement /v1/config endpoint returning catalog configuration - Implement namespace endpoints (list/create/get/head/delete) - Implement table endpoints (list/create/load/head/delete/update) - Add -port.iceberg flag to S3 standalone server (s3.go) - Add -s3.port.iceberg flag to combined server mode (server.go) - Add -s3.port.iceberg flag to mini cluster mode (mini.go) - Support prefix-based routing for multiple catalogs The Iceberg REST server reuses S3 Tables metadata storage under /table-buckets and enables DuckDB, Spark, and other Iceberg clients to connect to SeaweedFS as a catalog. * feat: Add Iceberg Catalog pages to admin UI Add admin UI pages to browse Iceberg catalogs, namespaces, and tables. - Add Iceberg Catalog menu item under Object Store navigation - Create iceberg_catalog.templ showing catalog overview with REST info - Create iceberg_namespaces.templ listing namespaces in a catalog - Create iceberg_tables.templ listing tables in a namespace - Add handlers and routes in admin_handlers.go - Add Iceberg data provider methods in s3tables_management.go - Add Iceberg data types in types.go The Iceberg Catalog pages provide visibility into the same S3 Tables data through an Iceberg-centric lens, including REST endpoint examples for DuckDB and PyIceberg. * test: Add Iceberg catalog integration tests and reorg s3tables tests - Reorganize existing s3tables tests to test/s3tables/table-buckets/ - Add new test/s3tables/catalog/ for Iceberg REST catalog tests - Add TestIcebergConfig to verify /v1/config endpoint - Add TestIcebergNamespaces to verify namespace listing - Add TestDuckDBIntegration for DuckDB connectivity (requires Docker) - Update CI workflow to use new test paths * fix: Generate proper random UUIDs for Iceberg tables Address code review feedback: - Replace placeholder UUID with crypto/rand-based UUID v4 generation - Add detailed TODO comments for handleUpdateTable stub explaining the required atomic metadata swap implementation * fix: Serve Iceberg on localhost listener when binding to different interface Address code review feedback: properly serve the localhost listener when the Iceberg server is bound to a non-localhost interface. * ci: Add Iceberg catalog integration tests to CI Add new job to run Iceberg catalog tests in CI, along with: - Iceberg package build verification - Iceberg unit tests - Iceberg go vet checks - Iceberg format checks * fix: Address code review feedback for Iceberg implementation - fix: Replace hardcoded account ID with s3_constants.AccountAdminId in buildTableBucketARN() - fix: Improve UUID generation error handling with deterministic fallback (timestamp + PID + counter) - fix: Update handleUpdateTable to return HTTP 501 Not Implemented instead of fake success - fix: Better error handling in handleNamespaceExists to distinguish 404 from 500 errors - fix: Use relative URL in template instead of hardcoded localhost:8181 - fix: Add HTTP timeout to test's waitForService function to avoid hangs - fix: Use dynamic ephemeral ports in integration tests to avoid flaky parallel failures - fix: Add Iceberg port to final port configuration logging in mini.go * fix: Address critical issues in Iceberg implementation - fix: Cache table UUIDs to ensure persistence across LoadTable calls The UUID now remains stable for the lifetime of the server session. TODO: For production, UUIDs should be persisted in S3 Tables metadata. - fix: Remove redundant URL-encoded namespace parsing mux router already decodes %1F to \x1F before passing to handlers. Redundant ReplaceAll call could cause bugs with literal %1F in namespace. * fix: Improve test robustness and reduce code duplication - fix: Make DuckDB test more robust by failing on unexpected errors Instead of silently logging errors, now explicitly check for expected conditions (extension not available) and skip the test appropriately. - fix: Extract username helper method to reduce duplication Created getUsername() helper in AdminHandlers to avoid duplicating the username retrieval logic across Iceberg page handlers. * fix: Add mutex protection to table UUID cache Protects concurrent access to the tableUUIDs map with sync.RWMutex. Uses read-lock for fast path when UUID already cached, and write-lock for generating new UUIDs. Includes double-check pattern to handle race condition between read-unlock and write-lock. * style: fix go fmt errors * feat(iceberg): persist table UUID in S3 Tables metadata * feat(admin): configure Iceberg port in Admin UI and commands * refactor: address review comments (flags, tests, handlers) - command/mini: fix tracking of explicit s3.port.iceberg flag - command/admin: add explicit -iceberg.port flag - admin/handlers: reuse getUsername helper - tests: use 127.0.0.1 for ephemeral ports and os.Stat for file size check * test: check error from FileStat in verify_gc_empty_test
This commit is contained in:
Chris Lu
431
test/s3tables/table-buckets/client.go
Normal file
431
test/s3tables/table-buckets/client.go
Normal file
@@ -0,0 +1,431 @@
|
||||
package s3tables
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/aws/aws-sdk-go-v2/aws"
|
||||
v4 "github.com/aws/aws-sdk-go-v2/aws/signer/v4"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
|
||||
)
|
||||
|
||||
func getFirstNamespace(namespace []string) (string, error) {
|
||||
if len(namespace) == 0 {
|
||||
return "", fmt.Errorf("namespace must not be empty")
|
||||
}
|
||||
return namespace[0], nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) doRestRequest(method, path string, body interface{}) (*http.Response, error) {
|
||||
var bodyBytes []byte
|
||||
var err error
|
||||
|
||||
if body != nil {
|
||||
bodyBytes, err = json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal request body: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(method, c.endpoint+path, bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/x-amz-json-1.1")
|
||||
}
|
||||
|
||||
if err := c.signRequest(req, bodyBytes); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return c.client.Do(req)
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) doTargetRequest(operation string, body interface{}) (*http.Response, error) {
|
||||
var bodyBytes []byte
|
||||
var err error
|
||||
|
||||
if body != nil {
|
||||
bodyBytes, err = json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to marshal request body: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(http.MethodPost, c.endpoint+"/", bytes.NewReader(bodyBytes))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
|
||||
req.URL.RawPath = "/"
|
||||
req.Header.Set("Content-Type", "application/x-amz-json-1.1")
|
||||
req.Header.Set("X-Amz-Target", "S3Tables."+operation)
|
||||
|
||||
if err := c.signRequest(req, bodyBytes); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return c.client.Do(req)
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) doTargetRequestAndDecode(operation string, reqBody interface{}, respBody interface{}) error {
|
||||
resp, err := c.doTargetRequest(operation, reqBody)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
bodyBytes, readErr := io.ReadAll(resp.Body)
|
||||
if readErr != nil {
|
||||
return fmt.Errorf("%s failed with status %d and could not read error response body: %v", operation, resp.StatusCode, readErr)
|
||||
}
|
||||
var errResp s3tables.S3TablesError
|
||||
if err := json.Unmarshal(bodyBytes, &errResp); err != nil {
|
||||
return fmt.Errorf("%s failed with status %d, could not decode error response: %v. Body: %s", operation, resp.StatusCode, err, string(bodyBytes))
|
||||
}
|
||||
return fmt.Errorf("%s failed: %s - %s", operation, errResp.Type, errResp.Message)
|
||||
}
|
||||
|
||||
if respBody != nil {
|
||||
if err := json.NewDecoder(resp.Body).Decode(respBody); err != nil {
|
||||
return fmt.Errorf("failed to decode %s response: %w", operation, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) signRequest(req *http.Request, body []byte) error {
|
||||
creds := aws.Credentials{
|
||||
AccessKeyID: c.accessKey,
|
||||
SecretAccessKey: c.secretKey,
|
||||
}
|
||||
if req.Host == "" {
|
||||
req.Host = req.URL.Host
|
||||
}
|
||||
req.Header.Set("Host", req.URL.Host)
|
||||
payloadHash := sha256.Sum256(body)
|
||||
return v4.NewSigner().SignHTTP(context.Background(), creds, req, hex.EncodeToString(payloadHash[:]), "s3tables", c.region, time.Now())
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) doRestRequestAndDecode(operation, method, path string, reqBody interface{}, respBody interface{}) error {
|
||||
resp, err := c.doRestRequest(method, path, reqBody)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
bodyBytes, readErr := io.ReadAll(resp.Body)
|
||||
if readErr != nil {
|
||||
return fmt.Errorf("%s failed with status %d and could not read error response body: %v", operation, resp.StatusCode, readErr)
|
||||
}
|
||||
var errResp s3tables.S3TablesError
|
||||
if err := json.Unmarshal(bodyBytes, &errResp); err != nil {
|
||||
return fmt.Errorf("%s failed with status %d, could not decode error response: %v. Body: %s", operation, resp.StatusCode, err, string(bodyBytes))
|
||||
}
|
||||
return fmt.Errorf("%s failed: %s - %s", operation, errResp.Type, errResp.Message)
|
||||
}
|
||||
|
||||
if respBody != nil {
|
||||
if err := json.NewDecoder(resp.Body).Decode(respBody); err != nil {
|
||||
return fmt.Errorf("failed to decode %s response: %w", operation, err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Table Bucket operations
|
||||
|
||||
func (c *S3TablesClient) CreateTableBucket(name string, tags map[string]string) (*s3tables.CreateTableBucketResponse, error) {
|
||||
req := &s3tables.CreateTableBucketRequest{
|
||||
Name: name,
|
||||
Tags: tags,
|
||||
}
|
||||
var result s3tables.CreateTableBucketResponse
|
||||
if err := c.doRestRequestAndDecode("CreateTableBucket", http.MethodPut, "/buckets", req, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) GetTableBucket(arn string) (*s3tables.GetTableBucketResponse, error) {
|
||||
path := "/buckets/" + url.PathEscape(arn)
|
||||
var result s3tables.GetTableBucketResponse
|
||||
if err := c.doRestRequestAndDecode("GetTableBucket", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) ListTableBuckets(prefix, continuationToken string, maxBuckets int) (*s3tables.ListTableBucketsResponse, error) {
|
||||
query := url.Values{}
|
||||
if prefix != "" {
|
||||
query.Set("prefix", prefix)
|
||||
}
|
||||
if continuationToken != "" {
|
||||
query.Set("continuationToken", continuationToken)
|
||||
}
|
||||
if maxBuckets > 0 {
|
||||
query.Set("maxBuckets", strconv.Itoa(maxBuckets))
|
||||
}
|
||||
path := "/buckets"
|
||||
if encoded := query.Encode(); encoded != "" {
|
||||
path = path + "?" + encoded
|
||||
}
|
||||
var result s3tables.ListTableBucketsResponse
|
||||
if err := c.doRestRequestAndDecode("ListTableBuckets", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) DeleteTableBucket(arn string) error {
|
||||
path := "/buckets/" + url.PathEscape(arn)
|
||||
return c.doRestRequestAndDecode("DeleteTableBucket", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
|
||||
// Namespace operations
|
||||
|
||||
func (c *S3TablesClient) CreateNamespace(bucketARN string, namespace []string) (*s3tables.CreateNamespaceResponse, error) {
|
||||
if len(namespace) == 0 {
|
||||
return nil, fmt.Errorf("CreateNamespace requires namespace")
|
||||
}
|
||||
req := &s3tables.CreateNamespaceRequest{
|
||||
Namespace: namespace,
|
||||
}
|
||||
path := "/namespaces/" + url.PathEscape(bucketARN)
|
||||
var result s3tables.CreateNamespaceResponse
|
||||
if err := c.doRestRequestAndDecode("CreateNamespace", http.MethodPut, path, req, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) GetNamespace(bucketARN string, namespace []string) (*s3tables.GetNamespaceResponse, error) {
|
||||
name, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetNamespace requires namespace: %w", err)
|
||||
}
|
||||
path := "/namespaces/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(name)
|
||||
var result s3tables.GetNamespaceResponse
|
||||
if err := c.doRestRequestAndDecode("GetNamespace", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) ListNamespaces(bucketARN, prefix, continuationToken string, maxNamespaces int) (*s3tables.ListNamespacesResponse, error) {
|
||||
query := url.Values{}
|
||||
if prefix != "" {
|
||||
query.Set("prefix", prefix)
|
||||
}
|
||||
if continuationToken != "" {
|
||||
query.Set("continuationToken", continuationToken)
|
||||
}
|
||||
if maxNamespaces > 0 {
|
||||
query.Set("maxNamespaces", strconv.Itoa(maxNamespaces))
|
||||
}
|
||||
path := "/namespaces/" + url.PathEscape(bucketARN)
|
||||
if encoded := query.Encode(); encoded != "" {
|
||||
path = path + "?" + encoded
|
||||
}
|
||||
var result s3tables.ListNamespacesResponse
|
||||
if err := c.doRestRequestAndDecode("ListNamespaces", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) DeleteNamespace(bucketARN string, namespace []string) error {
|
||||
name, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("DeleteNamespace requires namespace: %w", err)
|
||||
}
|
||||
path := "/namespaces/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(name)
|
||||
return c.doRestRequestAndDecode("DeleteNamespace", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
|
||||
// Table operations
|
||||
|
||||
func (c *S3TablesClient) CreateTable(bucketARN string, namespace []string, name, format string, metadata *s3tables.TableMetadata, tags map[string]string) (*s3tables.CreateTableResponse, error) {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("CreateTable requires namespace: %w", err)
|
||||
}
|
||||
req := &s3tables.CreateTableRequest{
|
||||
Name: name,
|
||||
Format: format,
|
||||
Metadata: metadata,
|
||||
Tags: tags,
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace)
|
||||
var result s3tables.CreateTableResponse
|
||||
if err := c.doRestRequestAndDecode("CreateTable", http.MethodPut, path, req, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) GetTable(bucketARN string, namespace []string, name string) (*s3tables.GetTableResponse, error) {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetTable requires namespace: %w", err)
|
||||
}
|
||||
query := url.Values{}
|
||||
query.Set("tableBucketARN", bucketARN)
|
||||
query.Set("namespace", nameSpace)
|
||||
query.Set("name", name)
|
||||
path := "/get-table?" + query.Encode()
|
||||
var result s3tables.GetTableResponse
|
||||
if err := c.doRestRequestAndDecode("GetTable", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) ListTables(bucketARN string, namespace []string, prefix, continuationToken string, maxTables int) (*s3tables.ListTablesResponse, error) {
|
||||
query := url.Values{}
|
||||
if len(namespace) > 0 {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("ListTables requires namespace: %w", err)
|
||||
}
|
||||
query.Set("namespace", nameSpace)
|
||||
}
|
||||
if prefix != "" {
|
||||
query.Set("prefix", prefix)
|
||||
}
|
||||
if continuationToken != "" {
|
||||
query.Set("continuationToken", continuationToken)
|
||||
}
|
||||
if maxTables > 0 {
|
||||
query.Set("maxTables", strconv.Itoa(maxTables))
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN)
|
||||
if encoded := query.Encode(); encoded != "" {
|
||||
path = path + "?" + encoded
|
||||
}
|
||||
var result s3tables.ListTablesResponse
|
||||
if err := c.doRestRequestAndDecode("ListTables", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) DeleteTable(bucketARN string, namespace []string, name string) error {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("DeleteTable requires namespace: %w", err)
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace) + "/" + url.PathEscape(name)
|
||||
return c.doRestRequestAndDecode("DeleteTable", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
|
||||
// Policy operations
|
||||
|
||||
func (c *S3TablesClient) PutTableBucketPolicy(bucketARN, policy string) error {
|
||||
req := &s3tables.PutTableBucketPolicyRequest{
|
||||
ResourcePolicy: policy,
|
||||
}
|
||||
path := "/buckets/" + url.PathEscape(bucketARN) + "/policy"
|
||||
return c.doRestRequestAndDecode("PutTableBucketPolicy", http.MethodPut, path, req, nil)
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) GetTableBucketPolicy(bucketARN string) (*s3tables.GetTableBucketPolicyResponse, error) {
|
||||
path := "/buckets/" + url.PathEscape(bucketARN) + "/policy"
|
||||
var result s3tables.GetTableBucketPolicyResponse
|
||||
if err := c.doRestRequestAndDecode("GetTableBucketPolicy", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) DeleteTableBucketPolicy(bucketARN string) error {
|
||||
path := "/buckets/" + url.PathEscape(bucketARN) + "/policy"
|
||||
return c.doRestRequestAndDecode("DeleteTableBucketPolicy", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
|
||||
// Table Policy operations
|
||||
|
||||
func (c *S3TablesClient) PutTablePolicy(bucketARN string, namespace []string, name, policy string) error {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("PutTablePolicy requires namespace: %w", err)
|
||||
}
|
||||
req := &s3tables.PutTablePolicyRequest{
|
||||
ResourcePolicy: policy,
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace) + "/" + url.PathEscape(name) + "/policy"
|
||||
return c.doRestRequestAndDecode("PutTablePolicy", http.MethodPut, path, req, nil)
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) GetTablePolicy(bucketARN string, namespace []string, name string) (*s3tables.GetTablePolicyResponse, error) {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("GetTablePolicy requires namespace: %w", err)
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace) + "/" + url.PathEscape(name) + "/policy"
|
||||
var result s3tables.GetTablePolicyResponse
|
||||
if err := c.doRestRequestAndDecode("GetTablePolicy", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) DeleteTablePolicy(bucketARN string, namespace []string, name string) error {
|
||||
nameSpace, err := getFirstNamespace(namespace)
|
||||
if err != nil {
|
||||
return fmt.Errorf("DeleteTablePolicy requires namespace: %w", err)
|
||||
}
|
||||
path := "/tables/" + url.PathEscape(bucketARN) + "/" + url.PathEscape(nameSpace) + "/" + url.PathEscape(name) + "/policy"
|
||||
return c.doRestRequestAndDecode("DeleteTablePolicy", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
|
||||
// Tagging operations
|
||||
|
||||
func (c *S3TablesClient) TagResource(resourceARN string, tags map[string]string) error {
|
||||
req := &s3tables.TagResourceRequest{
|
||||
Tags: tags,
|
||||
}
|
||||
path := "/tag/" + url.PathEscape(resourceARN)
|
||||
return c.doRestRequestAndDecode("TagResource", http.MethodPost, path, req, nil)
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) ListTagsForResource(resourceARN string) (*s3tables.ListTagsForResourceResponse, error) {
|
||||
path := "/tag/" + url.PathEscape(resourceARN)
|
||||
var result s3tables.ListTagsForResourceResponse
|
||||
if err := c.doRestRequestAndDecode("ListTagsForResource", http.MethodGet, path, nil, &result); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &result, nil
|
||||
}
|
||||
|
||||
func (c *S3TablesClient) UntagResource(resourceARN string, tagKeys []string) error {
|
||||
if len(tagKeys) == 0 {
|
||||
return fmt.Errorf("tagKeys cannot be empty")
|
||||
}
|
||||
query := url.Values{}
|
||||
for _, key := range tagKeys {
|
||||
query.Add("tagKeys", key)
|
||||
}
|
||||
path := "/tag/" + url.PathEscape(resourceARN)
|
||||
if encoded := query.Encode(); encoded != "" {
|
||||
path = path + "?" + encoded
|
||||
}
|
||||
return c.doRestRequestAndDecode("UntagResource", http.MethodDelete, path, nil, nil)
|
||||
}
|
||||
699
test/s3tables/table-buckets/s3tables_integration_test.go
Normal file
699
test/s3tables/table-buckets/s3tables_integration_test.go
Normal file
@@ -0,0 +1,699 @@
|
||||
package s3tables
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
cryptorand "crypto/rand"
|
||||
"sync"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/command"
|
||||
"github.com/seaweedfs/seaweedfs/weed/glog"
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
|
||||
flag "github.com/seaweedfs/seaweedfs/weed/util/fla9"
|
||||
)
|
||||
|
||||
var (
|
||||
miniClusterMutex sync.Mutex
|
||||
)
|
||||
|
||||
func TestS3TablesIntegration(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("Skipping integration test in short mode")
|
||||
}
|
||||
|
||||
// Create and start test cluster
|
||||
cluster, err := startMiniCluster(t)
|
||||
require.NoError(t, err)
|
||||
defer cluster.Stop()
|
||||
|
||||
// Create S3 Tables client
|
||||
client := NewS3TablesClient(cluster.s3Endpoint, testRegion, testAccessKey, testSecretKey)
|
||||
|
||||
// Run test suite
|
||||
t.Run("TableBucketLifecycle", func(t *testing.T) {
|
||||
testTableBucketLifecycle(t, client)
|
||||
})
|
||||
|
||||
t.Run("NamespaceLifecycle", func(t *testing.T) {
|
||||
testNamespaceLifecycle(t, client)
|
||||
})
|
||||
|
||||
t.Run("TableLifecycle", func(t *testing.T) {
|
||||
testTableLifecycle(t, client)
|
||||
})
|
||||
|
||||
t.Run("TableBucketPolicy", func(t *testing.T) {
|
||||
testTableBucketPolicy(t, client)
|
||||
})
|
||||
|
||||
t.Run("TablePolicy", func(t *testing.T) {
|
||||
testTablePolicy(t, client)
|
||||
})
|
||||
|
||||
t.Run("Tagging", func(t *testing.T) {
|
||||
testTagging(t, client)
|
||||
})
|
||||
|
||||
t.Run("TargetOperations", func(t *testing.T) {
|
||||
testTargetOperations(t, client)
|
||||
})
|
||||
}
|
||||
|
||||
func testTableBucketLifecycle(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-bucket-" + randomString(8)
|
||||
|
||||
// Create table bucket
|
||||
createResp, err := client.CreateTableBucket(bucketName, nil)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
assert.Contains(t, createResp.ARN, bucketName)
|
||||
t.Logf("✓ Created table bucket: %s", createResp.ARN)
|
||||
|
||||
// Get table bucket
|
||||
getResp, err := client.GetTableBucket(createResp.ARN)
|
||||
require.NoError(t, err, "Failed to get table bucket")
|
||||
assert.Equal(t, bucketName, getResp.Name)
|
||||
t.Logf("✓ Got table bucket: %s", getResp.Name)
|
||||
|
||||
// List table buckets
|
||||
listResp, err := client.ListTableBuckets("", "", 0)
|
||||
require.NoError(t, err, "Failed to list table buckets")
|
||||
found := false
|
||||
for _, b := range listResp.TableBuckets {
|
||||
if b.Name == bucketName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.True(t, found, "Created bucket should appear in list")
|
||||
t.Logf("✓ Listed table buckets, found %d buckets", len(listResp.TableBuckets))
|
||||
|
||||
// Delete table bucket
|
||||
err = client.DeleteTableBucket(createResp.ARN)
|
||||
require.NoError(t, err, "Failed to delete table bucket")
|
||||
t.Logf("✓ Deleted table bucket: %s", bucketName)
|
||||
|
||||
// Verify bucket is deleted
|
||||
_, err = client.GetTableBucket(createResp.ARN)
|
||||
assert.Error(t, err, "Bucket should not exist after deletion")
|
||||
}
|
||||
|
||||
func testNamespaceLifecycle(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-ns-bucket-" + randomString(8)
|
||||
namespaceName := "test_namespace"
|
||||
|
||||
// Create table bucket first
|
||||
createBucketResp, err := client.CreateTableBucket(bucketName, nil)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
defer client.DeleteTableBucket(createBucketResp.ARN)
|
||||
|
||||
bucketARN := createBucketResp.ARN
|
||||
|
||||
// Create namespace
|
||||
createNsResp, err := client.CreateNamespace(bucketARN, []string{namespaceName})
|
||||
require.NoError(t, err, "Failed to create namespace")
|
||||
assert.Equal(t, []string{namespaceName}, createNsResp.Namespace)
|
||||
t.Logf("✓ Created namespace: %s", namespaceName)
|
||||
|
||||
// Get namespace
|
||||
getNsResp, err := client.GetNamespace(bucketARN, []string{namespaceName})
|
||||
require.NoError(t, err, "Failed to get namespace")
|
||||
assert.Equal(t, []string{namespaceName}, getNsResp.Namespace)
|
||||
t.Logf("✓ Got namespace: %v", getNsResp.Namespace)
|
||||
|
||||
// List namespaces
|
||||
listNsResp, err := client.ListNamespaces(bucketARN, "", "", 0)
|
||||
require.NoError(t, err, "Failed to list namespaces")
|
||||
found := false
|
||||
for _, ns := range listNsResp.Namespaces {
|
||||
if len(ns.Namespace) > 0 && ns.Namespace[0] == namespaceName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.True(t, found, "Created namespace should appear in list")
|
||||
t.Logf("✓ Listed namespaces, found %d namespaces", len(listNsResp.Namespaces))
|
||||
|
||||
// Delete namespace
|
||||
err = client.DeleteNamespace(bucketARN, []string{namespaceName})
|
||||
require.NoError(t, err, "Failed to delete namespace")
|
||||
t.Logf("✓ Deleted namespace: %s", namespaceName)
|
||||
|
||||
// Verify namespace is deleted
|
||||
_, err = client.GetNamespace(bucketARN, []string{namespaceName})
|
||||
assert.Error(t, err, "Namespace should not exist after deletion")
|
||||
}
|
||||
|
||||
func testTableLifecycle(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-table-bucket-" + randomString(8)
|
||||
namespaceName := "test_ns"
|
||||
tableName := "test_table"
|
||||
|
||||
// Create table bucket
|
||||
createBucketResp, err := client.CreateTableBucket(bucketName, nil)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
defer client.DeleteTableBucket(createBucketResp.ARN)
|
||||
|
||||
bucketARN := createBucketResp.ARN
|
||||
|
||||
// Create namespace
|
||||
_, err = client.CreateNamespace(bucketARN, []string{namespaceName})
|
||||
require.NoError(t, err, "Failed to create namespace")
|
||||
defer client.DeleteNamespace(bucketARN, []string{namespaceName})
|
||||
|
||||
// Create table with Iceberg schema
|
||||
icebergMetadata := &s3tables.TableMetadata{
|
||||
Iceberg: &s3tables.IcebergMetadata{
|
||||
Schema: s3tables.IcebergSchema{
|
||||
Fields: []s3tables.IcebergSchemaField{
|
||||
{Name: "id", Type: "int", Required: true},
|
||||
{Name: "name", Type: "string"},
|
||||
{Name: "value", Type: "int"},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
createTableResp, err := client.CreateTable(bucketARN, []string{namespaceName}, tableName, "ICEBERG", icebergMetadata, nil)
|
||||
require.NoError(t, err, "Failed to create table")
|
||||
assert.NotEmpty(t, createTableResp.TableARN)
|
||||
assert.NotEmpty(t, createTableResp.VersionToken)
|
||||
t.Logf("✓ Created table: %s (version: %s)", createTableResp.TableARN, createTableResp.VersionToken)
|
||||
|
||||
// Get table
|
||||
getTableResp, err := client.GetTable(bucketARN, []string{namespaceName}, tableName)
|
||||
require.NoError(t, err, "Failed to get table")
|
||||
assert.Equal(t, tableName, getTableResp.Name)
|
||||
assert.Equal(t, "ICEBERG", getTableResp.Format)
|
||||
t.Logf("✓ Got table: %s (format: %s)", getTableResp.Name, getTableResp.Format)
|
||||
|
||||
// List tables
|
||||
listTablesResp, err := client.ListTables(bucketARN, []string{namespaceName}, "", "", 0)
|
||||
require.NoError(t, err, "Failed to list tables")
|
||||
found := false
|
||||
for _, tbl := range listTablesResp.Tables {
|
||||
if tbl.Name == tableName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.True(t, found, "Created table should appear in list")
|
||||
t.Logf("✓ Listed tables, found %d tables", len(listTablesResp.Tables))
|
||||
|
||||
// Delete table
|
||||
err = client.DeleteTable(bucketARN, []string{namespaceName}, tableName)
|
||||
require.NoError(t, err, "Failed to delete table")
|
||||
t.Logf("✓ Deleted table: %s", tableName)
|
||||
|
||||
// Verify table is deleted
|
||||
_, err = client.GetTable(bucketARN, []string{namespaceName}, tableName)
|
||||
assert.Error(t, err, "Table should not exist after deletion")
|
||||
}
|
||||
|
||||
func testTableBucketPolicy(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-policy-bucket-" + randomString(8)
|
||||
|
||||
// Create table bucket
|
||||
createBucketResp, err := client.CreateTableBucket(bucketName, nil)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
defer client.DeleteTableBucket(createBucketResp.ARN)
|
||||
|
||||
bucketARN := createBucketResp.ARN
|
||||
|
||||
// Put bucket policy
|
||||
policy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`
|
||||
err = client.PutTableBucketPolicy(bucketARN, policy)
|
||||
require.NoError(t, err, "Failed to put table bucket policy")
|
||||
t.Logf("✓ Put table bucket policy")
|
||||
|
||||
// Get bucket policy
|
||||
getPolicyResp, err := client.GetTableBucketPolicy(bucketARN)
|
||||
require.NoError(t, err, "Failed to get table bucket policy")
|
||||
assert.Equal(t, policy, getPolicyResp.ResourcePolicy)
|
||||
t.Logf("✓ Got table bucket policy")
|
||||
|
||||
// Delete bucket policy
|
||||
err = client.DeleteTableBucketPolicy(bucketARN)
|
||||
require.NoError(t, err, "Failed to delete table bucket policy")
|
||||
t.Logf("✓ Deleted table bucket policy")
|
||||
|
||||
// Verify policy is deleted
|
||||
_, err = client.GetTableBucketPolicy(bucketARN)
|
||||
assert.Error(t, err, "Policy should not exist after deletion")
|
||||
}
|
||||
|
||||
func testTablePolicy(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-table-policy-bucket-" + randomString(8)
|
||||
namespaceName := "test_ns"
|
||||
tableName := "test_table"
|
||||
|
||||
// Create table bucket
|
||||
createBucketResp, err := client.CreateTableBucket(bucketName, nil)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
defer client.DeleteTableBucket(createBucketResp.ARN)
|
||||
|
||||
bucketARN := createBucketResp.ARN
|
||||
|
||||
// Create namespace
|
||||
_, err = client.CreateNamespace(bucketARN, []string{namespaceName})
|
||||
require.NoError(t, err, "Failed to create namespace")
|
||||
defer client.DeleteNamespace(bucketARN, []string{namespaceName})
|
||||
|
||||
// Create table
|
||||
icebergMetadata := &s3tables.TableMetadata{
|
||||
Iceberg: &s3tables.IcebergMetadata{
|
||||
Schema: s3tables.IcebergSchema{
|
||||
Fields: []s3tables.IcebergSchemaField{
|
||||
{Name: "id", Type: "int", Required: true},
|
||||
{Name: "name", Type: "string"},
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
createTableResp, err := client.CreateTable(bucketARN, []string{namespaceName}, tableName, "ICEBERG", icebergMetadata, nil)
|
||||
require.NoError(t, err, "Failed to create table")
|
||||
defer client.DeleteTable(bucketARN, []string{namespaceName}, tableName)
|
||||
|
||||
t.Logf("✓ Created table: %s", createTableResp.TableARN)
|
||||
|
||||
// Verify no policy exists initially
|
||||
_, err = client.GetTablePolicy(bucketARN, []string{namespaceName}, tableName)
|
||||
assert.Error(t, err, "Policy should not exist initially")
|
||||
t.Logf("✓ Verified no policy exists initially")
|
||||
|
||||
// Put table policy
|
||||
policy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`
|
||||
err = client.PutTablePolicy(bucketARN, []string{namespaceName}, tableName, policy)
|
||||
require.NoError(t, err, "Failed to put table policy")
|
||||
t.Logf("✓ Put table policy")
|
||||
|
||||
// Get table policy
|
||||
getPolicyResp, err := client.GetTablePolicy(bucketARN, []string{namespaceName}, tableName)
|
||||
require.NoError(t, err, "Failed to get table policy")
|
||||
assert.Equal(t, policy, getPolicyResp.ResourcePolicy)
|
||||
t.Logf("✓ Got table policy")
|
||||
|
||||
// Delete table policy
|
||||
err = client.DeleteTablePolicy(bucketARN, []string{namespaceName}, tableName)
|
||||
require.NoError(t, err, "Failed to delete table policy")
|
||||
t.Logf("✓ Deleted table policy")
|
||||
|
||||
// Verify policy is deleted
|
||||
_, err = client.GetTablePolicy(bucketARN, []string{namespaceName}, tableName)
|
||||
assert.Error(t, err, "Policy should not exist after deletion")
|
||||
t.Logf("✓ Verified policy deletion")
|
||||
}
|
||||
|
||||
func testTagging(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-tag-bucket-" + randomString(8)
|
||||
|
||||
// Create table bucket with tags
|
||||
initialTags := map[string]string{"Environment": "test"}
|
||||
createBucketResp, err := client.CreateTableBucket(bucketName, initialTags)
|
||||
require.NoError(t, err, "Failed to create table bucket")
|
||||
defer client.DeleteTableBucket(createBucketResp.ARN)
|
||||
|
||||
bucketARN := createBucketResp.ARN
|
||||
|
||||
// List tags
|
||||
listTagsResp, err := client.ListTagsForResource(bucketARN)
|
||||
require.NoError(t, err, "Failed to list tags")
|
||||
assert.Equal(t, "test", listTagsResp.Tags["Environment"])
|
||||
t.Logf("✓ Listed tags: %v", listTagsResp.Tags)
|
||||
|
||||
// Add more tags
|
||||
newTags := map[string]string{"Department": "Engineering"}
|
||||
err = client.TagResource(bucketARN, newTags)
|
||||
require.NoError(t, err, "Failed to tag resource")
|
||||
t.Logf("✓ Added tags")
|
||||
|
||||
// Verify tags
|
||||
listTagsResp, err = client.ListTagsForResource(bucketARN)
|
||||
require.NoError(t, err, "Failed to list tags")
|
||||
assert.Equal(t, "test", listTagsResp.Tags["Environment"])
|
||||
assert.Equal(t, "Engineering", listTagsResp.Tags["Department"])
|
||||
t.Logf("✓ Verified tags: %v", listTagsResp.Tags)
|
||||
|
||||
// Remove a tag
|
||||
err = client.UntagResource(bucketARN, []string{"Environment"})
|
||||
require.NoError(t, err, "Failed to untag resource")
|
||||
t.Logf("✓ Removed tag")
|
||||
|
||||
// Verify tag is removed
|
||||
listTagsResp, err = client.ListTagsForResource(bucketARN)
|
||||
require.NoError(t, err, "Failed to list tags")
|
||||
_, hasEnvironment := listTagsResp.Tags["Environment"]
|
||||
assert.False(t, hasEnvironment, "Environment tag should be removed")
|
||||
assert.Equal(t, "Engineering", listTagsResp.Tags["Department"])
|
||||
t.Logf("✓ Verified tag removal")
|
||||
}
|
||||
|
||||
func testTargetOperations(t *testing.T, client *S3TablesClient) {
|
||||
bucketName := "test-target-bucket-" + randomString(8)
|
||||
|
||||
var createResp s3tables.CreateTableBucketResponse
|
||||
err := client.doTargetRequestAndDecode("CreateTableBucket", &s3tables.CreateTableBucketRequest{
|
||||
Name: bucketName,
|
||||
}, &createResp)
|
||||
require.NoError(t, err, "Failed to create table bucket via target")
|
||||
defer client.doTargetRequestAndDecode("DeleteTableBucket", &s3tables.DeleteTableBucketRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
}, nil)
|
||||
|
||||
var listResp s3tables.ListTableBucketsResponse
|
||||
err = client.doTargetRequestAndDecode("ListTableBuckets", &s3tables.ListTableBucketsRequest{}, &listResp)
|
||||
require.NoError(t, err, "Failed to list table buckets via target")
|
||||
found := false
|
||||
for _, b := range listResp.TableBuckets {
|
||||
if b.Name == bucketName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
assert.True(t, found, "Created bucket should appear in target list")
|
||||
|
||||
var getResp s3tables.GetTableBucketResponse
|
||||
err = client.doTargetRequestAndDecode("GetTableBucket", &s3tables.GetTableBucketRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
}, &getResp)
|
||||
require.NoError(t, err, "Failed to get table bucket via target")
|
||||
assert.Equal(t, bucketName, getResp.Name)
|
||||
|
||||
namespaceName := "target_ns"
|
||||
var createNsResp s3tables.CreateNamespaceResponse
|
||||
err = client.doTargetRequestAndDecode("CreateNamespace", &s3tables.CreateNamespaceRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
}, &createNsResp)
|
||||
require.NoError(t, err, "Failed to create namespace via target")
|
||||
defer client.doTargetRequestAndDecode("DeleteNamespace", &s3tables.DeleteNamespaceRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
}, nil)
|
||||
|
||||
var listNsResp s3tables.ListNamespacesResponse
|
||||
err = client.doTargetRequestAndDecode("ListNamespaces", &s3tables.ListNamespacesRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
}, &listNsResp)
|
||||
require.NoError(t, err, "Failed to list namespaces via target")
|
||||
|
||||
tableName := "target_table"
|
||||
var createTableResp s3tables.CreateTableResponse
|
||||
err = client.doTargetRequestAndDecode("CreateTable", &s3tables.CreateTableRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
Name: tableName,
|
||||
Format: "ICEBERG",
|
||||
}, &createTableResp)
|
||||
require.NoError(t, err, "Failed to create table via target")
|
||||
defer client.doTargetRequestAndDecode("DeleteTable", &s3tables.DeleteTableRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
Name: tableName,
|
||||
}, nil)
|
||||
|
||||
var listTablesResp s3tables.ListTablesResponse
|
||||
err = client.doTargetRequestAndDecode("ListTables", &s3tables.ListTablesRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
}, &listTablesResp)
|
||||
require.NoError(t, err, "Failed to list tables via target")
|
||||
|
||||
var getTableResp s3tables.GetTableResponse
|
||||
err = client.doTargetRequestAndDecode("GetTable", &s3tables.GetTableRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
Namespace: []string{namespaceName},
|
||||
Name: tableName,
|
||||
}, &getTableResp)
|
||||
require.NoError(t, err, "Failed to get table via target")
|
||||
assert.Equal(t, tableName, getTableResp.Name)
|
||||
|
||||
policy := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":"*","Action":"s3tables:*","Resource":"*"}]}`
|
||||
err = client.doTargetRequestAndDecode("PutTableBucketPolicy", &s3tables.PutTableBucketPolicyRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
ResourcePolicy: policy,
|
||||
}, nil)
|
||||
require.NoError(t, err, "Failed to put bucket policy via target")
|
||||
|
||||
var getPolicyResp s3tables.GetTableBucketPolicyResponse
|
||||
err = client.doTargetRequestAndDecode("GetTableBucketPolicy", &s3tables.GetTableBucketPolicyRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
}, &getPolicyResp)
|
||||
require.NoError(t, err, "Failed to get bucket policy via target")
|
||||
assert.Equal(t, policy, getPolicyResp.ResourcePolicy)
|
||||
|
||||
err = client.doTargetRequestAndDecode("DeleteTableBucketPolicy", &s3tables.DeleteTableBucketPolicyRequest{
|
||||
TableBucketARN: createResp.ARN,
|
||||
}, nil)
|
||||
require.NoError(t, err, "Failed to delete bucket policy via target")
|
||||
|
||||
err = client.doTargetRequestAndDecode("TagResource", &s3tables.TagResourceRequest{
|
||||
ResourceARN: createResp.ARN,
|
||||
Tags: map[string]string{"Environment": "test"},
|
||||
}, nil)
|
||||
require.NoError(t, err, "Failed to tag resource via target")
|
||||
|
||||
var listTagsResp s3tables.ListTagsForResourceResponse
|
||||
err = client.doTargetRequestAndDecode("ListTagsForResource", &s3tables.ListTagsForResourceRequest{
|
||||
ResourceARN: createResp.ARN,
|
||||
}, &listTagsResp)
|
||||
require.NoError(t, err, "Failed to list tags via target")
|
||||
assert.Equal(t, "test", listTagsResp.Tags["Environment"])
|
||||
|
||||
err = client.doTargetRequestAndDecode("UntagResource", &s3tables.UntagResourceRequest{
|
||||
ResourceARN: createResp.ARN,
|
||||
TagKeys: []string{"Environment"},
|
||||
}, nil)
|
||||
require.NoError(t, err, "Failed to untag resource via target")
|
||||
}
|
||||
|
||||
// Helper functions
|
||||
|
||||
// findAvailablePort finds an available port by binding to port 0
|
||||
func findAvailablePort() (int, error) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer listener.Close()
|
||||
|
||||
addr := listener.Addr().(*net.TCPAddr)
|
||||
return addr.Port, nil
|
||||
}
|
||||
|
||||
// startMiniCluster starts a weed mini instance directly without exec
|
||||
func startMiniCluster(t *testing.T) (*TestCluster, error) {
|
||||
// Find available ports
|
||||
masterPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find master port: %v", err)
|
||||
}
|
||||
masterGrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find master grpc port: %v", err)
|
||||
}
|
||||
volumePort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find volume port: %v", err)
|
||||
}
|
||||
volumeGrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find volume grpc port: %v", err)
|
||||
}
|
||||
filerPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find filer port: %v", err)
|
||||
}
|
||||
filerGrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find filer grpc port: %v", err)
|
||||
}
|
||||
s3Port, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find s3 port: %v", err)
|
||||
}
|
||||
s3GrpcPort, err := findAvailablePort()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to find s3 grpc port: %v", err)
|
||||
}
|
||||
// Create temporary directory for test data
|
||||
testDir := t.TempDir()
|
||||
|
||||
// Ensure no configuration file from previous runs
|
||||
configFile := filepath.Join(testDir, "mini.options")
|
||||
_ = os.Remove(configFile)
|
||||
|
||||
// Create context with timeout
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
|
||||
s3Endpoint := fmt.Sprintf("http://127.0.0.1:%d", s3Port)
|
||||
cluster := &TestCluster{
|
||||
t: t,
|
||||
dataDir: testDir,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
masterPort: masterPort,
|
||||
volumePort: volumePort,
|
||||
filerPort: filerPort,
|
||||
s3Port: s3Port,
|
||||
s3Endpoint: s3Endpoint,
|
||||
}
|
||||
|
||||
// Create empty security.toml to disable JWT authentication in tests
|
||||
securityToml := filepath.Join(testDir, "security.toml")
|
||||
err = os.WriteFile(securityToml, []byte("# Empty security config for testing\n"), 0644)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, fmt.Errorf("failed to create security.toml: %v", err)
|
||||
}
|
||||
|
||||
// Start weed mini in a goroutine by calling the command directly
|
||||
cluster.wg.Add(1)
|
||||
go func() {
|
||||
defer cluster.wg.Done()
|
||||
|
||||
// Protect global state mutation with a mutex
|
||||
miniClusterMutex.Lock()
|
||||
defer miniClusterMutex.Unlock()
|
||||
|
||||
// Save current directory and args
|
||||
oldDir, _ := os.Getwd()
|
||||
oldArgs := os.Args
|
||||
defer func() {
|
||||
os.Chdir(oldDir)
|
||||
os.Args = oldArgs
|
||||
}()
|
||||
|
||||
// Change to test directory so mini picks up security.toml
|
||||
os.Chdir(testDir)
|
||||
|
||||
// Configure args for mini command
|
||||
os.Args = []string{
|
||||
"weed",
|
||||
"-dir=" + testDir,
|
||||
"-master.port=" + strconv.Itoa(masterPort),
|
||||
"-master.port.grpc=" + strconv.Itoa(masterGrpcPort),
|
||||
"-volume.port=" + strconv.Itoa(volumePort),
|
||||
"-volume.port.grpc=" + strconv.Itoa(volumeGrpcPort),
|
||||
"-filer.port=" + strconv.Itoa(filerPort),
|
||||
"-filer.port.grpc=" + strconv.Itoa(filerGrpcPort),
|
||||
"-s3.port=" + strconv.Itoa(s3Port),
|
||||
"-s3.port.grpc=" + strconv.Itoa(s3GrpcPort),
|
||||
"-webdav.port=0", // Disable WebDAV
|
||||
"-admin.ui=false", // Disable admin UI
|
||||
"-master.volumeSizeLimitMB=32", // Small volumes for testing
|
||||
"-ip=127.0.0.1",
|
||||
"-master.peers=none", // Faster startup
|
||||
"-s3.iam.readOnly=false", // Enable IAM write operations for tests
|
||||
}
|
||||
|
||||
// Suppress most logging during tests
|
||||
glog.MaxSize = 1024 * 1024
|
||||
|
||||
// Find and run the mini command
|
||||
for _, cmd := range command.Commands {
|
||||
if cmd.Name() == "mini" && cmd.Run != nil {
|
||||
cmd.Flag.Parse(os.Args[1:])
|
||||
args := cmd.Flag.Args()
|
||||
command.MiniClusterCtx = ctx
|
||||
cmd.Run(cmd, args)
|
||||
command.MiniClusterCtx = nil
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
// Wait for S3 service to be ready
|
||||
err = waitForS3Ready(cluster.s3Endpoint, 30*time.Second)
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, fmt.Errorf("S3 service failed to start: %v", err)
|
||||
}
|
||||
|
||||
cluster.isRunning = true
|
||||
|
||||
t.Logf("Test cluster started successfully at %s", cluster.s3Endpoint)
|
||||
return cluster, nil
|
||||
}
|
||||
|
||||
// Stop stops the test cluster
|
||||
func (c *TestCluster) Stop() {
|
||||
if c.cancel != nil {
|
||||
c.cancel()
|
||||
}
|
||||
// Give services time to shut down gracefully
|
||||
if c.isRunning {
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
}
|
||||
// Wait for the mini goroutine to finish
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
c.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
timer := time.NewTimer(2 * time.Second)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-done:
|
||||
// Goroutine finished
|
||||
case <-timer.C:
|
||||
// Timeout - goroutine doesn't respond to context cancel
|
||||
// This may indicate the mini cluster didn't shut down cleanly
|
||||
c.t.Log("Warning: Test cluster shutdown timed out after 2 seconds")
|
||||
}
|
||||
|
||||
// Reset the global cmdMini flags to prevent state leakage to other tests
|
||||
for _, cmd := range command.Commands {
|
||||
if cmd.Name() == "mini" {
|
||||
// Reset flags to defaults
|
||||
cmd.Flag.VisitAll(func(f *flag.Flag) {
|
||||
// Reset to default value
|
||||
f.Value.Set(f.DefValue)
|
||||
})
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// waitForS3Ready waits for the S3 service to be ready
|
||||
func waitForS3Ready(endpoint string, timeout time.Duration) error {
|
||||
client := &http.Client{Timeout: 1 * time.Second}
|
||||
deadline := time.Now().Add(timeout)
|
||||
|
||||
for time.Now().Before(deadline) {
|
||||
resp, err := client.Get(endpoint)
|
||||
if err == nil {
|
||||
resp.Body.Close()
|
||||
// Wait a bit more to ensure service is fully ready
|
||||
time.Sleep(500 * time.Millisecond)
|
||||
return nil
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
}
|
||||
|
||||
return fmt.Errorf("timeout waiting for S3 service at %s", endpoint)
|
||||
}
|
||||
|
||||
// randomString generates a random string for unique naming
|
||||
func randomString(length int) string {
|
||||
const charset = "abcdefghijklmnopqrstuvwxyz0123456789"
|
||||
b := make([]byte, length)
|
||||
if _, err := cryptorand.Read(b); err != nil {
|
||||
panic("failed to generate random string: " + err.Error())
|
||||
}
|
||||
for i := range b {
|
||||
b[i] = charset[int(b[i])%len(charset)]
|
||||
}
|
||||
return string(b)
|
||||
}
|
||||
53
test/s3tables/table-buckets/setup.go
Normal file
53
test/s3tables/table-buckets/setup.go
Normal file
@@ -0,0 +1,53 @@
|
||||
package s3tables
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/http"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TestCluster manages the weed mini instance for integration testing
|
||||
type TestCluster struct {
|
||||
t *testing.T
|
||||
dataDir string
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
isRunning bool
|
||||
startOnce sync.Once
|
||||
wg sync.WaitGroup
|
||||
masterPort int
|
||||
volumePort int
|
||||
filerPort int
|
||||
s3Port int
|
||||
s3Endpoint string
|
||||
}
|
||||
|
||||
// S3TablesClient is a simple client for S3 Tables API
|
||||
type S3TablesClient struct {
|
||||
endpoint string
|
||||
region string
|
||||
accessKey string
|
||||
secretKey string
|
||||
client *http.Client
|
||||
}
|
||||
|
||||
// NewS3TablesClient creates a new S3 Tables client
|
||||
func NewS3TablesClient(endpoint, region, accessKey, secretKey string) *S3TablesClient {
|
||||
return &S3TablesClient{
|
||||
endpoint: endpoint,
|
||||
region: region,
|
||||
accessKey: accessKey,
|
||||
secretKey: secretKey,
|
||||
client: &http.Client{Timeout: 30 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
// Test configuration constants
|
||||
const (
|
||||
testRegion = "us-west-2"
|
||||
testAccessKey = "admin"
|
||||
testSecretKey = "admin"
|
||||
testAccountID = "111122223333"
|
||||
)
|
||||
Reference in New Issue
Block a user