mirror of
https://github.com/hanzoai/s3.git
synced 2026-10-30 14:59:09 +00:00
zeekay
937bee8c7f
Completes the ZAP-native migration on main's leaderless base (schain
pinned-writer is canonical — NO consensus engine introduced). Every
non-test s3 package is now gRPC-free (`google.golang.org/grpc` importers:
16 -> 0); all gRPC-importing test files converted too, so the root module
has ZERO direct grpc importers.
Source (ported from feat/grpc-final's grpc-free versions onto main, which
do NOT touch the leaderless rewrite):
- pb/grpc_client_server.go: dialoption param thread, ZAP pools only.
- mq/broker/*: status.Errorf(codes.X) -> fmt.Errorf("X: ...").
- server/volume_grpc_admin.go: keeps the v2026.3.1 authz-bypass fix
(fail-closed over peerless ZAP) + grpc-free.
- server/volume_grpc_erasure_coding.go, server/common.go (request_id),
svc/master/stream_server.go (drop grpc.ServerStream embed),
wdclient/masterclient.go, operation/assign_file_id.go (string-match
classification).
- security/tls.go (drop LoadServerTLS/advancedtls/SNIStripping),
security/tls_reload.go + certreload/certreload.go (native mtime/size
keypair watcher, no grpc certprovider/pemfile),
util/http/client/http_client.go.
command/master.go: kept main's leaderless schain coordinator; stripped
only the vestigial reflection-only gRPC server (no master RPC rode it).
Tests: codes/status -> fmt/strings; grpc.ServerStream/metadata mocks ->
grpc-free seams (Send/Recv/Context); integration tests dial ZAP
(transport.Dial + pb.NewZapFilerClient / master.New(...ToMasterZapAddress)).
volume_grpc_admin_zap_authz_test.go still PASSES and still asserts the
admin bypass is CLOSED. Deleted tls_sni_test.go (tested removed SNIStripping).
go-bip39: upstream github.com/tyler-smith/go-bip39 repo is deleted (fetch
fails under GOPROXY=direct), breaking `go mod tidy`. Redirect to the luxfi
fork via an in-tree drop-in module (third_party/go-bip39) under the original
path delegating to github.com/luxfi/go-bip39 v1.1.2 — a plain replace is
impossible (the fork renamed its module path and self-imports
.../wordlists, "used for two different module paths"). `go mod tidy` now
succeeds.
go.mod: google.golang.org/grpc/security/advancedtls DROPPED; grpc demoted
direct -> // indirect. grpc now survives ONLY as a transitive of Google
Cloud SDKs (cloud.google.com/go/{kms,storage,pubsub}) used by the GCP
KMS / GCS storage / Pub/Sub integrations — external API-client transport,
not s3 internal RPC. Fully removing it would mean ripping out all GCP
cloud support (out of scope).
214 lines
7.3 KiB
Go
214 lines
7.3 KiB
Go
package volume_server_grpc_test
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/hanzoai/s3/test/volume_server/framework"
|
|
"github.com/hanzoai/s3/test/volume_server/matrix"
|
|
"github.com/hanzoai/s3/s3/pb/volume_server_pb"
|
|
)
|
|
|
|
func TestVolumeAdminLifecycleRPCs(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
cluster := framework.StartVolumeCluster(t, matrix.P1())
|
|
conn, client := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress())
|
|
defer conn.Close()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
const volumeID = uint32(11)
|
|
framework.AllocateVolume(t, client, volumeID, "")
|
|
|
|
statusResp, err := client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
|
|
if err != nil {
|
|
t.Fatalf("VolumeStatus failed: %v", err)
|
|
}
|
|
if statusResp.GetFileCount() != 0 {
|
|
t.Fatalf("new volume should be empty, got file_count=%d", statusResp.GetFileCount())
|
|
}
|
|
|
|
if _, err = client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: volumeID}); err != nil {
|
|
t.Fatalf("VolumeUnmount failed: %v", err)
|
|
}
|
|
if _, err = client.VolumeMount(ctx, &volume_server_pb.VolumeMountRequest{VolumeId: volumeID}); err != nil {
|
|
t.Fatalf("VolumeMount failed: %v", err)
|
|
}
|
|
|
|
if _, err = client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{VolumeId: volumeID, OnlyEmpty: true}); err != nil {
|
|
t.Fatalf("VolumeDelete failed: %v", err)
|
|
}
|
|
|
|
_, err = client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
|
|
if err == nil {
|
|
t.Fatalf("VolumeStatus should fail after delete")
|
|
}
|
|
if err.Error() == "" {
|
|
t.Fatalf("VolumeStatus error should be a non-empty status error, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestVolumeDeleteOnlyEmptyVariants(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
clusterHarness := framework.StartVolumeCluster(t, matrix.P1())
|
|
conn, grpcClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress())
|
|
defer conn.Close()
|
|
|
|
const volumeID = uint32(13)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
client := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, 66001, 0x11223344)
|
|
uploadResp := framework.UploadBytes(t, client, clusterHarness.VolumeAdminURL(), fid, []byte("volume-delete-only-empty"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{VolumeId: volumeID, OnlyEmpty: true})
|
|
if err == nil || !strings.Contains(err.Error(), "volume not empty") {
|
|
t.Fatalf("VolumeDelete only_empty=true expected volume-not-empty error, got: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{VolumeId: volumeID, OnlyEmpty: false})
|
|
if err != nil {
|
|
t.Fatalf("VolumeDelete only_empty=false failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{VolumeId: volumeID})
|
|
if err == nil {
|
|
t.Fatalf("VolumeStatus should fail after non-empty delete with only_empty=false")
|
|
}
|
|
}
|
|
|
|
func TestMaintenanceModeRejectsAllocateVolume(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
cluster := framework.StartVolumeCluster(t, matrix.P1())
|
|
conn, client := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress())
|
|
defer conn.Close()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
stateResp, err := client.GetState(ctx, &volume_server_pb.GetStateRequest{})
|
|
if err != nil {
|
|
t.Fatalf("GetState failed: %v", err)
|
|
}
|
|
_, err = client.SetState(ctx, &volume_server_pb.SetStateRequest{
|
|
State: &volume_server_pb.VolumeServerState{Maintenance: true, Version: stateResp.GetState().GetVersion()},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("SetState maintenance=true failed: %v", err)
|
|
}
|
|
|
|
_, err = client.AllocateVolume(ctx, &volume_server_pb.AllocateVolumeRequest{VolumeId: 12, Replication: "000"})
|
|
if err == nil {
|
|
t.Fatalf("AllocateVolume should fail when maintenance mode is enabled")
|
|
}
|
|
if !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("expected maintenance mode error, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestAllocateDuplicateAndMountUnmountMissingVariants(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
cluster := framework.StartVolumeCluster(t, matrix.P1())
|
|
conn, client := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress())
|
|
defer conn.Close()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
const missingVolumeID = uint32(99331)
|
|
const volumeID = uint32(14)
|
|
|
|
if _, err := client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: missingVolumeID}); err != nil {
|
|
t.Fatalf("VolumeUnmount missing volume should be idempotent success, got: %v", err)
|
|
}
|
|
|
|
_, err := client.VolumeMount(ctx, &volume_server_pb.VolumeMountRequest{VolumeId: missingVolumeID})
|
|
if err == nil {
|
|
t.Fatalf("VolumeMount missing volume should fail")
|
|
}
|
|
if !strings.Contains(err.Error(), "not found on disk") {
|
|
t.Fatalf("VolumeMount missing volume error mismatch: %v", err)
|
|
}
|
|
|
|
framework.AllocateVolume(t, client, volumeID, "")
|
|
|
|
_, err = client.AllocateVolume(ctx, &volume_server_pb.AllocateVolumeRequest{
|
|
VolumeId: volumeID,
|
|
Replication: "000",
|
|
})
|
|
if err == nil {
|
|
t.Fatalf("AllocateVolume duplicate should fail")
|
|
}
|
|
if !strings.Contains(strings.ToLower(err.Error()), "already exists") {
|
|
t.Fatalf("AllocateVolume duplicate error mismatch: %v", err)
|
|
}
|
|
|
|
if _, err = client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: volumeID}); err != nil {
|
|
t.Fatalf("VolumeUnmount existing volume failed: %v", err)
|
|
}
|
|
if _, err = client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: volumeID}); err != nil {
|
|
t.Fatalf("VolumeUnmount already-unmounted volume should be idempotent success, got: %v", err)
|
|
}
|
|
if _, err = client.VolumeMount(ctx, &volume_server_pb.VolumeMountRequest{VolumeId: volumeID}); err != nil {
|
|
t.Fatalf("VolumeMount remount failed: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestMaintenanceModeRejectsVolumeDelete(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
cluster := framework.StartVolumeCluster(t, matrix.P1())
|
|
conn, client := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress())
|
|
defer conn.Close()
|
|
|
|
const volumeID = uint32(15)
|
|
framework.AllocateVolume(t, client, volumeID, "")
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
stateResp, err := client.GetState(ctx, &volume_server_pb.GetStateRequest{})
|
|
if err != nil {
|
|
t.Fatalf("GetState failed: %v", err)
|
|
}
|
|
_, err = client.SetState(ctx, &volume_server_pb.SetStateRequest{
|
|
State: &volume_server_pb.VolumeServerState{Maintenance: true, Version: stateResp.GetState().GetVersion()},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("SetState maintenance=true failed: %v", err)
|
|
}
|
|
|
|
_, err = client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{VolumeId: volumeID, OnlyEmpty: true})
|
|
if err == nil {
|
|
t.Fatalf("VolumeDelete should fail when maintenance mode is enabled")
|
|
}
|
|
if !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("expected maintenance mode error, got: %v", err)
|
|
}
|
|
}
|