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]>
105 lines
2.5 KiB
Go
105 lines
2.5 KiB
Go
package mountmanager
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"time"
|
|
)
|
|
|
|
// Client talks to the mount service over a Unix domain socket.
|
|
type Client struct {
|
|
httpClient *http.Client
|
|
baseURL string
|
|
}
|
|
|
|
// NewClient builds a new Client for the given endpoint.
|
|
func NewClient(endpoint string) (*Client, error) {
|
|
scheme, address, err := ParseEndpoint(endpoint)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if scheme != "unix" {
|
|
return nil, fmt.Errorf("unsupported endpoint scheme: %s", scheme)
|
|
}
|
|
|
|
dialer := &net.Dialer{}
|
|
transport := &http.Transport{
|
|
DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
|
|
return dialer.DialContext(ctx, "unix", address)
|
|
},
|
|
}
|
|
|
|
return &Client{
|
|
httpClient: &http.Client{
|
|
Timeout: 30 * time.Second,
|
|
Transport: transport,
|
|
},
|
|
baseURL: "http://unix",
|
|
}, nil
|
|
}
|
|
|
|
// Mount mounts a volume using the mount service.
|
|
func (c *Client) Mount(req *MountRequest) (*MountResponse, error) {
|
|
var resp MountResponse
|
|
if err := c.doPost("/mount", req, &resp); err != nil {
|
|
return nil, err
|
|
}
|
|
return &resp, nil
|
|
}
|
|
|
|
// Unmount unmounts a volume using the mount service.
|
|
func (c *Client) Unmount(req *UnmountRequest) (*UnmountResponse, error) {
|
|
var resp UnmountResponse
|
|
if err := c.doPost("/unmount", req, &resp); err != nil {
|
|
return nil, err
|
|
}
|
|
return &resp, nil
|
|
}
|
|
|
|
func (c *Client) doPost(path string, payload any, out any) error {
|
|
body := &bytes.Buffer{}
|
|
if err := json.NewEncoder(body).Encode(payload); err != nil {
|
|
return fmt.Errorf("encode request: %w", err)
|
|
}
|
|
|
|
req, err := http.NewRequest(http.MethodPost, c.baseURL+path, body)
|
|
if err != nil {
|
|
return fmt.Errorf("build request: %w", err)
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
|
|
resp, err := c.httpClient.Do(req)
|
|
if err != nil {
|
|
return fmt.Errorf("call mount service: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode >= 400 {
|
|
var errResp ErrorResponse
|
|
if err := json.NewDecoder(resp.Body).Decode(&errResp); err == nil && errResp.Error != "" {
|
|
return errors.New(errResp.Error)
|
|
}
|
|
data, readErr := io.ReadAll(resp.Body)
|
|
if readErr != nil {
|
|
return fmt.Errorf("mount service error: %s (failed to read body: %v)", resp.Status, readErr)
|
|
}
|
|
return fmt.Errorf("mount service error: %s (%s)", resp.Status, string(data))
|
|
}
|
|
|
|
if out == nil {
|
|
_, _ = io.Copy(io.Discard, resp.Body)
|
|
return nil
|
|
}
|
|
|
|
if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
|
|
return fmt.Errorf("decode response: %w", err)
|
|
}
|
|
return nil
|
|
}
|