mirror of
https://github.com/hanzoai/s3.git
synced 2026-10-30 14:59:09 +00:00
zeekay
e1e0356e75
filer, master, iam, plugin and worker no longer use google.golang.org/grpc on
the client path.
- gRPC-free client interfaces: the generated *_grpc.pb.go client interfaces drop
"opts ...grpc.CallOption" (no caller ever passed one) and their streaming
methods return the shared seam types in s3/pb/rpc (ServerStream[T] /
BidiStream[Req,Resp]) instead of grpc.ServerStreamingClient/BidiStreamingClient.
The gRPC server scaffolding (Unimplemented*Server, Register*Server, ServiceDesc,
handler trampolines) is deleted; servers serve over the ZAP wire Dispatch.
- The ZAP adapters (filerzap/masterzap/...) implement the gRPC-free interfaces;
grpc.DialOption becomes pb.DialOption (the dial is the ZAP transport).
- Unary calls honor context deadlines: the wire client threads ctx to
transport.Conn.CallContext (v1.8.1), so a call to an unresponsive peer aborts
on the caller's deadline instead of blocking forever.
- Error semantics: the filer tags each failure with its code name ("NotFound: …",
"Internal: …") in the message, which crosses the ZAP wire and is classified by
mount/error_classifier.go — replacing grpc status/codes.
- master ZAP listener port: grpcPort+10000 could exceed 65535; made overflow-safe.
- test doubles updated to the gRPC-free interfaces (0 tests skipped).
- dead cmd/s3-db and cmd/s3-sql (postgres-wire / SQL-engine entry points whose
backends were already ripped) deleted; finished the backend-rip import cleanup.
go build ./... green; go test ./s3/... green (heavy stream benches run under
their existing -short gates).
Still on gRPC (next wave): volume_server + mq (reverted — servers/adapters not
ready), the remaining ~23 plumbing files, and the go.mod dependency itself.
290 lines
7.2 KiB
Go
290 lines
7.2 KiB
Go
package erasure_coding_test
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/stretchr/testify/require"
|
|
"google.golang.org/protobuf/proto"
|
|
|
|
"github.com/hanzoai/s3/s3/pb"
|
|
|
|
"github.com/hanzoai/s3/s3/pb/master_pb"
|
|
"github.com/hanzoai/s3/s3/pb/plugin_pb"
|
|
"github.com/hanzoai/s3/s3/pb/worker_pb"
|
|
pluginworker "github.com/hanzoai/s3/s3/plugin/worker"
|
|
ecstorage "github.com/hanzoai/s3/s3/storage/erasure_coding"
|
|
"github.com/hanzoai/s3/s3/worker/tasks/erasure_coding"
|
|
pluginworkers "github.com/hanzoai/s3/test/plugin_workers"
|
|
)
|
|
|
|
type topologySpec struct {
|
|
name string
|
|
dataCenters int
|
|
racksPerDC int
|
|
nodesPerRack int
|
|
diskTypes []string
|
|
replicas int
|
|
collection string
|
|
}
|
|
|
|
type detectionCase struct {
|
|
name string
|
|
topology topologySpec
|
|
adminCollectionFilter string
|
|
expectProposals bool
|
|
}
|
|
|
|
func TestErasureCodingDetectionAcrossTopologies(t *testing.T) {
|
|
cases := []detectionCase{
|
|
{
|
|
name: "single-dc-multi-rack",
|
|
topology: topologySpec{
|
|
name: "single-dc-multi-rack",
|
|
dataCenters: 1,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "multi-dc",
|
|
topology: topologySpec{
|
|
name: "multi-dc",
|
|
dataCenters: 2,
|
|
racksPerDC: 1,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "multi-dc-multi-rack",
|
|
topology: topologySpec{
|
|
name: "multi-dc-multi-rack",
|
|
dataCenters: 2,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 4,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "mixed-disk-types",
|
|
topology: topologySpec{
|
|
name: "mixed-disk-types",
|
|
dataCenters: 1,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd", "ssd"},
|
|
replicas: 1,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "multi-replica-volume",
|
|
topology: topologySpec{
|
|
name: "multi-replica-volume",
|
|
dataCenters: 1,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 3,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "collection-filter-match",
|
|
topology: topologySpec{
|
|
name: "collection-filter-match",
|
|
dataCenters: 1,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "filtered",
|
|
},
|
|
adminCollectionFilter: "filtered",
|
|
expectProposals: true,
|
|
},
|
|
{
|
|
name: "collection-filter-mismatch",
|
|
topology: topologySpec{
|
|
name: "collection-filter-mismatch",
|
|
dataCenters: 1,
|
|
racksPerDC: 2,
|
|
nodesPerRack: 7,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "filtered",
|
|
},
|
|
adminCollectionFilter: "other",
|
|
expectProposals: false,
|
|
},
|
|
{
|
|
name: "insufficient-disks",
|
|
topology: topologySpec{
|
|
name: "insufficient-disks",
|
|
dataCenters: 1,
|
|
racksPerDC: 1,
|
|
nodesPerRack: 2,
|
|
diskTypes: []string{"hdd"},
|
|
replicas: 1,
|
|
collection: "ec-test",
|
|
},
|
|
expectProposals: false,
|
|
},
|
|
}
|
|
|
|
for _, tc := range cases {
|
|
tc := tc
|
|
t.Run(tc.name, func(t *testing.T) {
|
|
volumeID := uint32(7)
|
|
response := buildVolumeListResponse(t, tc.topology, volumeID)
|
|
master := pluginworkers.NewMasterServer(t, response)
|
|
|
|
dialOption := pb.DialOption{}
|
|
handler := erasure_coding.NewErasureCodingHandler(dialOption, t.TempDir())
|
|
harness := pluginworkers.NewHarness(t, pluginworkers.HarnessConfig{
|
|
WorkerOptions: pluginworker.WorkerOptions{
|
|
GrpcDialOption: dialOption,
|
|
},
|
|
Handlers: []pluginworker.JobHandler{handler},
|
|
})
|
|
harness.WaitForJobType("erasure_coding")
|
|
|
|
if tc.adminCollectionFilter != "" {
|
|
err := harness.Plugin().SaveJobTypeConfig(&plugin_pb.PersistedJobTypeConfig{
|
|
JobType: "erasure_coding",
|
|
AdminConfigValues: map[string]*plugin_pb.ConfigValue{
|
|
"collection_filter": {
|
|
Kind: &plugin_pb.ConfigValue_StringValue{StringValue: tc.adminCollectionFilter},
|
|
},
|
|
},
|
|
})
|
|
require.NoError(t, err)
|
|
}
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
|
|
proposals, err := harness.Plugin().RunDetection(ctx, "erasure_coding", &plugin_pb.ClusterContext{
|
|
MasterGrpcAddresses: []string{master.Address()},
|
|
}, 10)
|
|
require.NoError(t, err)
|
|
|
|
if !tc.expectProposals {
|
|
require.Empty(t, proposals)
|
|
return
|
|
}
|
|
|
|
require.NotEmpty(t, proposals)
|
|
|
|
proposal := proposals[0]
|
|
require.Equal(t, "erasure_coding", proposal.JobType)
|
|
paramsValue := proposal.Parameters["task_params_pb"]
|
|
require.NotNil(t, paramsValue)
|
|
|
|
params := &worker_pb.TaskParams{}
|
|
require.NoError(t, proto.Unmarshal(paramsValue.GetBytesValue(), params))
|
|
require.NotEmpty(t, params.Sources)
|
|
require.Len(t, params.Targets, ecstorage.TotalShardsCount)
|
|
})
|
|
}
|
|
}
|
|
|
|
func buildVolumeListResponse(t *testing.T, spec topologySpec, volumeID uint32) *master_pb.VolumeListResponse {
|
|
t.Helper()
|
|
|
|
volumeSizeLimitMB := uint64(100)
|
|
// Exceed the default fullness (0.95) and quiet (1h) thresholds so volumes are
|
|
// EC-eligible.
|
|
volumeSize := uint64(96) * 1024 * 1024
|
|
volumeModifiedAt := time.Now().Add(-2 * time.Hour).Unix()
|
|
|
|
diskTypes := spec.diskTypes
|
|
if len(diskTypes) == 0 {
|
|
diskTypes = []string{"hdd"}
|
|
}
|
|
replicas := spec.replicas
|
|
if replicas <= 0 {
|
|
replicas = 1
|
|
}
|
|
collection := spec.collection
|
|
if collection == "" {
|
|
collection = "ec-test"
|
|
}
|
|
|
|
var dataCenters []*master_pb.DataCenterInfo
|
|
nodeIndex := 0
|
|
replicasPlaced := 0
|
|
|
|
for dc := 0; dc < spec.dataCenters; dc++ {
|
|
var racks []*master_pb.RackInfo
|
|
for rack := 0; rack < spec.racksPerDC; rack++ {
|
|
var nodes []*master_pb.DataNodeInfo
|
|
for n := 0; n < spec.nodesPerRack; n++ {
|
|
nodeIndex++
|
|
address := fmt.Sprintf("127.0.0.1:%d", 20000+nodeIndex)
|
|
diskType := diskTypes[(nodeIndex-1)%len(diskTypes)]
|
|
|
|
diskInfo := &master_pb.DiskInfo{
|
|
DiskId: 0,
|
|
MaxVolumeCount: 100,
|
|
VolumeCount: 0,
|
|
VolumeInfos: []*master_pb.VolumeInformationMessage{},
|
|
}
|
|
|
|
if replicasPlaced < replicas {
|
|
diskInfo.VolumeCount = 1
|
|
diskInfo.VolumeInfos = append(diskInfo.VolumeInfos, &master_pb.VolumeInformationMessage{
|
|
Id: volumeID,
|
|
Collection: collection,
|
|
DiskId: 0,
|
|
Size: volumeSize,
|
|
DeletedByteCount: 0,
|
|
ModifiedAtSecond: volumeModifiedAt,
|
|
ReplicaPlacement: 1,
|
|
ReadOnly: false,
|
|
})
|
|
replicasPlaced++
|
|
}
|
|
|
|
nodes = append(nodes, &master_pb.DataNodeInfo{
|
|
Id: address,
|
|
Address: address,
|
|
DiskInfos: map[string]*master_pb.DiskInfo{diskType: diskInfo},
|
|
})
|
|
}
|
|
|
|
racks = append(racks, &master_pb.RackInfo{
|
|
Id: fmt.Sprintf("rack-%d", rack+1),
|
|
DataNodeInfos: nodes,
|
|
})
|
|
}
|
|
|
|
dataCenters = append(dataCenters, &master_pb.DataCenterInfo{
|
|
Id: fmt.Sprintf("dc-%d", dc+1),
|
|
RackInfos: racks,
|
|
})
|
|
}
|
|
|
|
return &master_pb.VolumeListResponse{
|
|
VolumeSizeLimitMb: volumeSizeLimitMB,
|
|
TopologyInfo: &master_pb.TopologyInfo{
|
|
DataCenterInfos: dataCenters,
|
|
},
|
|
}
|
|
}
|