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).
857 lines
28 KiB
Go
857 lines
28 KiB
Go
package volume_server_grpc_test
|
|
|
|
import (
|
|
"context"
|
|
"io"
|
|
"math"
|
|
"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"
|
|
"github.com/hanzoai/s3/s3/storage/erasure_coding"
|
|
"github.com/hanzoai/s3/s3/storage/needle"
|
|
)
|
|
|
|
func TestEcMaintenanceModeRejections(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()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
stateResp, err := grpcClient.GetState(ctx, &volume_server_pb.GetStateRequest{})
|
|
if err != nil {
|
|
t.Fatalf("GetState failed: %v", err)
|
|
}
|
|
_, err = grpcClient.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 = grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{VolumeId: 1, Collection: ""})
|
|
if err == nil || !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("VolumeEcShardsGenerate maintenance error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{
|
|
VolumeId: 1,
|
|
Collection: "",
|
|
SourceDataNode: "127.0.0.1:1",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("VolumeEcShardsCopy maintenance error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
|
|
VolumeId: 1,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("VolumeEcShardsDelete maintenance error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcBlobDelete(ctx, &volume_server_pb.VolumeEcBlobDeleteRequest{
|
|
VolumeId: 1,
|
|
Collection: "",
|
|
FileKey: 1,
|
|
Version: 3,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("VolumeEcBlobDelete maintenance error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsToVolume(ctx, &volume_server_pb.VolumeEcShardsToVolumeRequest{
|
|
VolumeId: 1,
|
|
Collection: "",
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "maintenance mode") {
|
|
t.Fatalf("VolumeEcShardsToVolume maintenance error mismatch: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcMissingInvalidAndNoopPaths(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()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: 98791,
|
|
Collection: "",
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "not found") {
|
|
t.Fatalf("VolumeEcShardsGenerate missing-volume error mismatch: %v", err)
|
|
}
|
|
|
|
rebuildResp, err := grpcClient.VolumeEcShardsRebuild(ctx, &volume_server_pb.VolumeEcShardsRebuildRequest{
|
|
VolumeId: 98792,
|
|
Collection: "ec-rebuild",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsRebuild missing-volume should return empty success, got: %v", err)
|
|
}
|
|
if len(rebuildResp.GetRebuiltShardIds()) != 0 {
|
|
t.Fatalf("VolumeEcShardsRebuild expected no rebuilt shards for missing volume, got %v", rebuildResp.GetRebuiltShardIds())
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{
|
|
VolumeId: 98793,
|
|
Collection: "ec-copy",
|
|
SourceDataNode: "127.0.0.1:1",
|
|
ShardIds: []uint32{0},
|
|
DiskId: 99,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "invalid disk_id") {
|
|
t.Fatalf("VolumeEcShardsCopy invalid-disk error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
|
|
VolumeId: 98794,
|
|
Collection: "ec-delete",
|
|
ShardIds: []uint32{0, 1},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsDelete missing-volume should be no-op success, got: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: 98795,
|
|
Collection: "ec-mount",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err == nil {
|
|
t.Fatalf("VolumeEcShardsMount should fail for missing EC shards")
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{
|
|
VolumeId: 98796,
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsUnmount missing shards should be no-op success, got: %v", err)
|
|
}
|
|
|
|
readStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{
|
|
VolumeId: 98797,
|
|
ShardId: 0,
|
|
Offset: 0,
|
|
Size: 1,
|
|
})
|
|
if err == nil {
|
|
_, err = readStream.Recv()
|
|
}
|
|
if err == nil || err == io.EOF {
|
|
t.Fatalf("VolumeEcShardRead should fail for missing EC volume")
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcBlobDelete(ctx, &volume_server_pb.VolumeEcBlobDeleteRequest{
|
|
VolumeId: 98798,
|
|
Collection: "ec-blob",
|
|
FileKey: 1,
|
|
Version: 3,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcBlobDelete missing local EC volume should be no-op success, got: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsToVolume(ctx, &volume_server_pb.VolumeEcShardsToVolumeRequest{
|
|
VolumeId: 98799,
|
|
Collection: "ec-to-volume",
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "not found") {
|
|
t.Fatalf("VolumeEcShardsToVolume missing-volume error mismatch: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{
|
|
VolumeId: 98800,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "not found") {
|
|
t.Fatalf("VolumeEcShardsInfo missing-volume error mismatch: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcGenerateMountInfoUnmountLifecycle(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(115)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, 990001, 0x1234ABCD)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-generate-lifecycle-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate success path failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount success path failed: %v", err)
|
|
}
|
|
|
|
infoResp, err := grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{
|
|
VolumeId: volumeID,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsInfo after mount failed: %v", err)
|
|
}
|
|
if len(infoResp.GetEcShardInfos()) == 0 {
|
|
t.Fatalf("VolumeEcShardsInfo expected non-empty shard infos after mount")
|
|
}
|
|
if infoResp.GetVolumeSize() == 0 {
|
|
t.Fatalf("VolumeEcShardsInfo expected non-zero volume size after mount")
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsUnmount(ctx, &volume_server_pb.VolumeEcShardsUnmountRequest{
|
|
VolumeId: volumeID,
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsUnmount success path failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsInfo(ctx, &volume_server_pb.VolumeEcShardsInfoRequest{
|
|
VolumeId: volumeID,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "not found") {
|
|
t.Fatalf("VolumeEcShardsInfo after unmount expected not-found error, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcShardReadAndBlobDeleteLifecycle(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(116)
|
|
const fileKey = uint64(990002)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, fileKey, 0x2233CCDD)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-shard-read-delete-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount failed: %v", err)
|
|
}
|
|
|
|
readStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{
|
|
VolumeId: volumeID,
|
|
ShardId: 0,
|
|
Offset: 0,
|
|
Size: 1,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardRead start failed: %v", err)
|
|
}
|
|
firstChunk, err := readStream.Recv()
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardRead recv failed: %v", err)
|
|
}
|
|
if len(firstChunk.GetData()) == 0 {
|
|
t.Fatalf("VolumeEcShardRead expected non-empty data chunk before deletion")
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcBlobDelete(ctx, &volume_server_pb.VolumeEcBlobDeleteRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
FileKey: fileKey,
|
|
Version: uint32(needle.GetCurrentVersion()),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcBlobDelete first delete failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcBlobDelete(ctx, &volume_server_pb.VolumeEcBlobDeleteRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
FileKey: fileKey,
|
|
Version: uint32(needle.GetCurrentVersion()),
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcBlobDelete second delete should be idempotent success, got: %v", err)
|
|
}
|
|
|
|
deletedStream, err := grpcClient.VolumeEcShardRead(ctx, &volume_server_pb.VolumeEcShardReadRequest{
|
|
VolumeId: volumeID,
|
|
ShardId: 0,
|
|
FileKey: fileKey,
|
|
Offset: 0,
|
|
Size: 1,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardRead deleted-check start failed: %v", err)
|
|
}
|
|
deletedMsg, err := deletedStream.Recv()
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardRead deleted-check recv failed: %v", err)
|
|
}
|
|
if !deletedMsg.GetIsDeleted() {
|
|
t.Fatalf("VolumeEcShardRead expected IsDeleted=true after blob delete")
|
|
}
|
|
_, err = deletedStream.Recv()
|
|
if err != io.EOF {
|
|
t.Fatalf("VolumeEcShardRead deleted-check expected EOF after deleted marker, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcRebuildMissingShardLifecycle(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(117)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, 990003, 0x3344DDEE)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-rebuild-shard-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsDelete shard 0 failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err == nil {
|
|
t.Fatalf("VolumeEcShardsMount should fail when shard 0 has been deleted")
|
|
}
|
|
|
|
rebuildResp, err := grpcClient.VolumeEcShardsRebuild(ctx, &volume_server_pb.VolumeEcShardsRebuildRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsRebuild failed: %v", err)
|
|
}
|
|
if len(rebuildResp.GetRebuiltShardIds()) == 0 {
|
|
t.Fatalf("VolumeEcShardsRebuild expected rebuilt shard ids")
|
|
}
|
|
foundShard0 := false
|
|
for _, shardID := range rebuildResp.GetRebuiltShardIds() {
|
|
if shardID == 0 {
|
|
foundShard0 = true
|
|
break
|
|
}
|
|
}
|
|
if !foundShard0 {
|
|
t.Fatalf("VolumeEcShardsRebuild expected shard 0 to be rebuilt, got %v", rebuildResp.GetRebuiltShardIds())
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount shard 0 after rebuild failed: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcShardsToVolumeMissingShardAndNoLiveEntries(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()
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
t.Run("missing shard returns error", func(t *testing.T) {
|
|
const volumeID = uint32(118)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
fid := framework.NewFileID(volumeID, 990004, 0x4455EEFF)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-to-volume-missing-shard-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsDelete shard 0 failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{1},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount shard 1 failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsToVolume(ctx, &volume_server_pb.VolumeEcShardsToVolumeRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "missing shard 0") {
|
|
t.Fatalf("VolumeEcShardsToVolume missing-shard error mismatch: %v", err)
|
|
}
|
|
})
|
|
|
|
t.Run("no live entries returns failed precondition", func(t *testing.T) {
|
|
const volumeID = uint32(119)
|
|
const needleID = uint64(990005)
|
|
const cookie = uint32(0x5566FF11)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
fid := framework.NewFileID(volumeID, needleID, cookie)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-no-live-entries-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
deleteResp := framework.DoRequest(t, httpClient, mustNewRequest(t, http.MethodDelete, clusterHarness.VolumeAdminURL()+"/"+fid))
|
|
_ = framework.ReadAllAndClose(t, deleteResp)
|
|
if deleteResp.StatusCode != http.StatusAccepted {
|
|
t.Fatalf("delete expected 202, got %d", deleteResp.StatusCode)
|
|
}
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount data shards failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsToVolume(ctx, &volume_server_pb.VolumeEcShardsToVolumeRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err == nil {
|
|
t.Fatalf("VolumeEcShardsToVolume expected failed-precondition error when no live entries")
|
|
}
|
|
if !strings.Contains(err.Error(), "FailedPrecondition") {
|
|
t.Fatalf("VolumeEcShardsToVolume no-live-entries expected FailedPrecondition, got: %v", err)
|
|
}
|
|
if !strings.Contains(err.Error(), erasure_coding.EcNoLiveEntriesSubstring) {
|
|
t.Fatalf("VolumeEcShardsToVolume no-live-entries error should mention %q, got %v", erasure_coding.EcNoLiveEntriesSubstring, err)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestEcShardsToVolumeSuccessRoundTrip(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(120)
|
|
const needleID = uint64(990006)
|
|
const cookie = uint32(0x66771122)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, needleID, cookie)
|
|
payload := []byte("ec-shards-to-volume-success-roundtrip-content")
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, payload)
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount data shards failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsToVolume(ctx, &volume_server_pb.VolumeEcShardsToVolumeRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsToVolume success path failed: %v", err)
|
|
}
|
|
|
|
readResp := framework.ReadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid)
|
|
readBody := framework.ReadAllAndClose(t, readResp)
|
|
if readResp.StatusCode != http.StatusOK {
|
|
t.Fatalf("post-conversion read expected 200, got %d", readResp.StatusCode)
|
|
}
|
|
if string(readBody) != string(payload) {
|
|
t.Fatalf("post-conversion payload mismatch: got %q want %q", string(readBody), string(payload))
|
|
}
|
|
}
|
|
|
|
func TestEcShardsDeleteLastShardRemovesEcx(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(121)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, 990007, 0x77882233)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), fid, []byte("ec-delete-all-shards-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
// Verify .ecx is present before deleting all shards.
|
|
ecxBeforeDelete, err := grpcClient.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
IsEcVolume: true,
|
|
Ext: ".ecx",
|
|
CompactionRevision: math.MaxUint32,
|
|
StopOffset: 1,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("CopyFile .ecx before shard deletion start failed: %v", err)
|
|
}
|
|
if _, err = ecxBeforeDelete.Recv(); err != nil {
|
|
t.Fatalf("CopyFile .ecx before shard deletion recv failed: %v", err)
|
|
}
|
|
|
|
_, err = grpcClient.VolumeEcShardsDelete(ctx, &volume_server_pb.VolumeEcShardsDeleteRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsDelete all shards failed: %v", err)
|
|
}
|
|
|
|
ecxAfterDelete, err := grpcClient.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
IsEcVolume: true,
|
|
Ext: ".ecx",
|
|
CompactionRevision: math.MaxUint32,
|
|
StopOffset: 1,
|
|
})
|
|
if err == nil {
|
|
_, err = ecxAfterDelete.Recv()
|
|
}
|
|
if err == nil || !strings.Contains(err.Error(), "not found ec volume id") {
|
|
t.Fatalf("CopyFile .ecx after deleting all shards should fail not-found, got: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestEcShardsCopyFromPeerSuccess(t *testing.T) {
|
|
if testing.Short() {
|
|
t.Skip("skipping integration test in short mode")
|
|
}
|
|
|
|
clusterHarness := framework.StartDualVolumeCluster(t, matrix.P1())
|
|
sourceConn, sourceClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress(0))
|
|
defer sourceConn.Close()
|
|
destConn, destClient := framework.DialVolumeServer(t, clusterHarness.VolumeGRPCAddress(1))
|
|
defer destConn.Close()
|
|
|
|
const volumeID = uint32(122)
|
|
framework.AllocateVolume(t, sourceClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
fid := framework.NewFileID(volumeID, 990008, 0x88993344)
|
|
uploadResp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(0), fid, []byte("ec-copy-from-peer-content"))
|
|
_ = framework.ReadAllAndClose(t, uploadResp)
|
|
if uploadResp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("source upload expected 201, got %d", uploadResp.StatusCode)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := sourceClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("source VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
sourceDataNode := clusterHarness.VolumeAdminAddress(0) + "." + strings.Split(clusterHarness.VolumeGRPCAddress(0), ":")[1]
|
|
_, err = destClient.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
SourceDataNode: sourceDataNode,
|
|
ShardIds: []uint32{0},
|
|
CopyEcxFile: true,
|
|
CopyVifFile: true,
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("destination VolumeEcShardsCopy success path failed: %v", err)
|
|
}
|
|
|
|
for _, ext := range []string{".ec00", ".ecx", ".vif"} {
|
|
copyStream, copyErr := destClient.CopyFile(ctx, &volume_server_pb.CopyFileRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
IsEcVolume: true,
|
|
Ext: ext,
|
|
CompactionRevision: math.MaxUint32,
|
|
StopOffset: 1,
|
|
})
|
|
if copyErr != nil {
|
|
t.Fatalf("destination CopyFile %s start failed: %v", ext, copyErr)
|
|
}
|
|
if _, copyErr = copyStream.Recv(); copyErr != nil {
|
|
t.Fatalf("destination CopyFile %s recv failed: %v", ext, copyErr)
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestEcIndexConsistencyAfterEncode verifies that every needle indexed in .ecx
|
|
// can be read back correctly from EC shards after VolumeEcShardsGenerate.
|
|
// This catches the race condition fixed in this PR where .ecx could reference
|
|
// data not present in EC shards.
|
|
func TestEcIndexConsistencyAfterEncode(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(130)
|
|
framework.AllocateVolume(t, grpcClient, volumeID, "")
|
|
|
|
httpClient := framework.NewHTTPClient()
|
|
|
|
// Upload multiple needles of varying sizes
|
|
type testNeedle struct {
|
|
fid string
|
|
payload []byte
|
|
}
|
|
needles := []testNeedle{
|
|
{framework.NewFileID(volumeID, 1001, 0xAABB0001), []byte("small-needle-1")},
|
|
{framework.NewFileID(volumeID, 1002, 0xAABB0002), make([]byte, 1024)}, // 1KB
|
|
{framework.NewFileID(volumeID, 1003, 0xAABB0003), make([]byte, 64*1024)}, // 64KB
|
|
{framework.NewFileID(volumeID, 1004, 0xAABB0004), make([]byte, 256*1024)}, // 256KB
|
|
{framework.NewFileID(volumeID, 1005, 0xAABB0005), []byte("small-needle-2")},
|
|
}
|
|
|
|
// Fill larger payloads with recognizable data
|
|
for i := range needles {
|
|
for j := range needles[i].payload {
|
|
needles[i].payload[j] = byte(i*37 + j%251)
|
|
}
|
|
}
|
|
|
|
for _, n := range needles {
|
|
resp := framework.UploadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid, n.payload)
|
|
_ = framework.ReadAllAndClose(t, resp)
|
|
if resp.StatusCode != http.StatusCreated {
|
|
t.Fatalf("upload %s expected 201, got %d", n.fid, resp.StatusCode)
|
|
}
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
// EC encode
|
|
_, err := grpcClient.VolumeEcShardsGenerate(ctx, &volume_server_pb.VolumeEcShardsGenerateRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsGenerate failed: %v", err)
|
|
}
|
|
|
|
// Mount all data shards so reads go through the EC path
|
|
_, err = grpcClient.VolumeEcShardsMount(ctx, &volume_server_pb.VolumeEcShardsMountRequest{
|
|
VolumeId: volumeID,
|
|
Collection: "",
|
|
ShardIds: []uint32{0, 1, 2, 3, 4, 5, 6, 7, 8, 9},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("VolumeEcShardsMount failed: %v", err)
|
|
}
|
|
|
|
// Read every needle back from EC shards and verify payload
|
|
for _, n := range needles {
|
|
readResp := framework.ReadBytes(t, httpClient, clusterHarness.VolumeAdminURL(), n.fid)
|
|
readBody := framework.ReadAllAndClose(t, readResp)
|
|
if readResp.StatusCode != http.StatusOK {
|
|
t.Fatalf("EC read %s expected 200, got %d", n.fid, readResp.StatusCode)
|
|
}
|
|
if string(readBody) != string(n.payload) {
|
|
t.Fatalf("EC read %s payload mismatch: got %d bytes, want %d bytes", n.fid, len(readBody), len(n.payload))
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestEcShardsCopyFailsWhenSourceUnavailable(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()
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
_, err := grpcClient.VolumeEcShardsCopy(ctx, &volume_server_pb.VolumeEcShardsCopyRequest{
|
|
VolumeId: 12345,
|
|
Collection: "",
|
|
SourceDataNode: "127.0.0.1:1.1",
|
|
ShardIds: []uint32{0},
|
|
CopyEcxFile: true,
|
|
})
|
|
if err == nil || !strings.Contains(err.Error(), "VolumeEcShardsCopy volume") {
|
|
t.Fatalf("VolumeEcShardsCopy source-unavailable error mismatch: %v", err)
|
|
}
|
|
}
|