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]>
95 lines
3.1 KiB
Go
95 lines
3.1 KiB
Go
package k8s
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"path"
|
|
"time"
|
|
|
|
corev1 "k8s.io/api/core/v1"
|
|
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
|
"k8s.io/apimachinery/pkg/util/validation"
|
|
"k8s.io/client-go/kubernetes"
|
|
"k8s.io/client-go/rest"
|
|
)
|
|
|
|
func newInCluster() (*kubernetes.Clientset, error) {
|
|
//creates the in-cluster config
|
|
config, err := rest.InClusterConfig()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to get in-cluster config: %v", err)
|
|
}
|
|
|
|
// creates the clientset
|
|
clientset, err := kubernetes.NewForConfig(config)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to create in-cluster client: %v", err)
|
|
}
|
|
return clientset, nil
|
|
}
|
|
|
|
func GetVolumeCapacity(driverName, volumeId string) (int64, error) {
|
|
client, err := newInCluster()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
return getVolumeCapacity(ctx, client, driverName, volumeId)
|
|
}
|
|
|
|
func getVolumeCapacity(ctx context.Context, client kubernetes.Interface, driverName, volumeId string) (int64, error) {
|
|
// Fast path: avoid listing every PersistentVolume in the cluster on each
|
|
// stage. Legacy dynamic volumes used the PV name directly as the CSI volume
|
|
// handle, while newer ones use a full filer path (e.g. "/buckets/pvc-xxxx")
|
|
// whose last component is still the PV name. In both cases a direct Get by
|
|
// that name resolves the volume, and we confirm the match before trusting it.
|
|
pvName := path.Base(volumeId)
|
|
if len(validation.IsDNS1123Subdomain(pvName)) == 0 {
|
|
if volume, err := client.CoreV1().PersistentVolumes().Get(ctx, pvName, metav1.GetOptions{}); err == nil &&
|
|
volume.Spec.CSI != nil && volume.Spec.CSI.Driver == driverName &&
|
|
(volume.Spec.CSI.VolumeHandle == volumeId || volume.Name == volumeId) {
|
|
return persistentVolumeCapacity(volume)
|
|
}
|
|
}
|
|
|
|
// Fallback: the handle does not map to a PV name (e.g. a static volume whose
|
|
// handle is an arbitrary filer path), so match by CSI volume handle instead.
|
|
volumes, err := client.CoreV1().PersistentVolumes().List(ctx, metav1.ListOptions{})
|
|
if err != nil {
|
|
return 0, fmt.Errorf("list persistent volumes for CSI volume handle %q: %w", volumeId, err)
|
|
}
|
|
|
|
var matched *corev1.PersistentVolume
|
|
for i := range volumes.Items {
|
|
volume := &volumes.Items[i]
|
|
if volume.Spec.CSI == nil ||
|
|
volume.Spec.CSI.Driver != driverName ||
|
|
volume.Spec.CSI.VolumeHandle != volumeId {
|
|
continue
|
|
}
|
|
if matched != nil {
|
|
return 0, fmt.Errorf("multiple persistent volumes use CSI volume handle %q", volumeId)
|
|
}
|
|
matched = volume
|
|
}
|
|
if matched == nil {
|
|
return 0, fmt.Errorf("persistent volume with name or CSI volume handle %q not found", volumeId)
|
|
}
|
|
|
|
return persistentVolumeCapacity(matched)
|
|
}
|
|
|
|
func persistentVolumeCapacity(volume *corev1.PersistentVolume) (int64, error) {
|
|
storage := volume.Spec.Capacity.Storage()
|
|
if storage == nil {
|
|
return 0, fmt.Errorf("persistent volume %q has no storage capacity", volume.Name)
|
|
}
|
|
capacity, ok := storage.AsInt64()
|
|
if !ok {
|
|
return 0, fmt.Errorf("persistent volume %q storage capacity does not fit in int64", volume.Name)
|
|
}
|
|
return capacity, nil
|
|
}
|