mirror of
https://github.com/hanzoai/s3-csi.git
synced 2026-09-30 23:52:32 +00:00
zeekayandHanzo Dev
69ddc9947a
Fork of seaweedfs/seaweedfs-csi-driver, renamed seaweedfs -> s3 throughout and ported off protobuf entirely. WHY a fork was unavoidable: hanzoai/s3 serves HanzoFiler over the native ZAP transport on the port still named "grpc" (18888) — s3/command/filer.go binds it with transport.ListenStream and says it "replaces the legacy gRPC HanzoFiler server: the whole filer (28 unary + 5 streaming RPCs) now answers over ZAP, no gRPC". RegisterHanzoFilerServer exists nowhere in that tree. So the upstream driver's gRPC client cannot connect at all; it fails as `Unimplemented: unknown service filer_pb.SeaweedFiler`, which LOOKS like a service-name mismatch but is a gRPC client hitting a non-gRPC port. Renaming the service in the client would not have helped. pkg/filer is the only place this driver touches storage, and it speaks ZAP: Mkdir -> CreateEntry (CreateVolume) Remove -> DeleteEntry (DeleteVolume) Exists -> LookupDirectoryEntry (ValidateVolumeCapabilities) One shared, concurrency-safe client multiplexes over a single ZAP connection, dials lazily so controller start does not depend on filer readiness, and redials exactly once on a transport fault (covers a filer restart closing an idle conn without becoming a retry storm against a struggling backend). pkg/driver/volume.go Quota() likewise moves to s3/wire/mount — the fork generates no mount_pb at all. Dead gRPC plumbing (WithFilerClient, AdjustedUrl, grpcDialOption) is deleted rather than left inert. The kubelet-facing side stays CSI gRPC: that is the Kubernetes contract between kubelet and any CSI driver, not a Hanzo choice. Zero protobuf call sites remain. Whole module compiles clean. Co-authored-by: Hanzo Dev <[email protected]>
107 lines
3.0 KiB
Go
107 lines
3.0 KiB
Go
//go:build linux
|
|
// +build linux
|
|
|
|
package driver
|
|
|
|
import (
|
|
"errors"
|
|
"path/filepath"
|
|
"testing"
|
|
)
|
|
|
|
// fakeMounter records Mount/Unmount calls and reports success without
|
|
// touching the real mount service.
|
|
type fakeMounter struct {
|
|
mountCalls int
|
|
unmountCalls int
|
|
lastTarget string
|
|
mountErr error
|
|
}
|
|
|
|
func (f *fakeMounter) Mount(target string) (Unmounter, error) {
|
|
f.mountCalls++
|
|
f.lastTarget = target
|
|
if f.mountErr != nil {
|
|
return nil, f.mountErr
|
|
}
|
|
return &fakeUnmounter{parent: f}, nil
|
|
}
|
|
|
|
type fakeUnmounter struct {
|
|
parent *fakeMounter
|
|
}
|
|
|
|
func (f *fakeUnmounter) Unmount() error {
|
|
f.parent.unmountCalls++
|
|
return nil
|
|
}
|
|
|
|
// newTestNodeServer returns a NodeServer with all mount/filesystem
|
|
// factories replaced so tests do not touch the mount service or k8s API.
|
|
// The health monitor is not started.
|
|
func newTestNodeServer(t *testing.T, fake *fakeMounter) *NodeServer {
|
|
t.Helper()
|
|
ns := &NodeServer{
|
|
Driver: &S3Driver{},
|
|
volumeMutexes: NewKeyMutex(),
|
|
stopCh: make(chan struct{}),
|
|
mounterFactory: func(volumeID string, readOnly bool, driver *S3Driver, volContext map[string]string) (Mounter, error) {
|
|
return fake, nil
|
|
},
|
|
capacityFn: func(volumeID string) (int64, error) {
|
|
// Skip quota application in tests.
|
|
return 0, errors.New("capacity not available in test")
|
|
},
|
|
isHealthyFn: func(path string) bool { return true },
|
|
cleanupStagingFn: func(path string) error { return nil },
|
|
unmountFn: func(path string) error { return nil },
|
|
bindMountFn: func(source, target string, readOnly bool) error {
|
|
// Create the target so follow-up checkMount calls see a directory.
|
|
return nil
|
|
},
|
|
}
|
|
return ns
|
|
}
|
|
|
|
func TestStageNewVolumeUsesInjectedFactories(t *testing.T) {
|
|
fake := &fakeMounter{}
|
|
ns := newTestNodeServer(t, fake)
|
|
|
|
// Volume.Stage's checkMount will try to create the staging directory if
|
|
// it does not exist, so point it at a tempdir.
|
|
stagingPath := filepath.Join(t.TempDir(), "staging")
|
|
|
|
vol, err := ns.stageNewVolume("vol-1", stagingPath, map[string]string{"collection": "c"}, false)
|
|
if err != nil {
|
|
t.Fatalf("stageNewVolume failed: %v", err)
|
|
}
|
|
if vol == nil {
|
|
t.Fatal("stageNewVolume returned nil volume")
|
|
}
|
|
if fake.mountCalls != 1 {
|
|
t.Errorf("expected 1 mount call, got %d", fake.mountCalls)
|
|
}
|
|
if fake.lastTarget != stagingPath {
|
|
t.Errorf("expected mount target %q, got %q", stagingPath, fake.lastTarget)
|
|
}
|
|
if vol.VolumeId != "vol-1" {
|
|
t.Errorf("expected volume id %q, got %q", "vol-1", vol.VolumeId)
|
|
}
|
|
if vol.StagedPath != stagingPath {
|
|
t.Errorf("expected staged path %q, got %q", stagingPath, vol.StagedPath)
|
|
}
|
|
}
|
|
|
|
func TestStageNewVolumePropagatesMounterError(t *testing.T) {
|
|
wantErr := errors.New("mount refused")
|
|
ns := newTestNodeServer(t, nil)
|
|
ns.mounterFactory = func(volumeID string, readOnly bool, driver *S3Driver, volContext map[string]string) (Mounter, error) {
|
|
return nil, wantErr
|
|
}
|
|
|
|
_, err := ns.stageNewVolume("vol-1", t.TempDir(), nil, false)
|
|
if !errors.Is(err, wantErr) {
|
|
t.Fatalf("expected error %v, got %v", wantErr, err)
|
|
}
|
|
}
|