diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go index ffeedc968..de9f98a36 100644 --- a/weed/s3api/s3api_handlers.go +++ b/weed/s3api/s3api_handlers.go @@ -5,6 +5,7 @@ import ( "fmt" "net/http" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "google.golang.org/grpc" @@ -15,7 +16,13 @@ import ( var _ = filer_pb.FilerClient(&S3ApiServer{}) func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { - + // Use filerClient for proper connection management and failover + if s3a.filerClient != nil { + return s3a.withFilerClientFailover(streamingMode, fn) + } + + // Fallback to direct connection if filerClient not initialized + // This should only happen during initialization or testing return pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { client := filer_pb.NewSeaweedFilerClient(grpcConnection) return fn(client) @@ -23,6 +30,50 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea } +// withFilerClientFailover attempts to execute fn with automatic failover to other filers +func (s3a *S3ApiServer) withFilerClientFailover(streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { + // Get current filer as starting point + currentFiler := s3a.filerClient.GetCurrentFiler() + + // Try current filer first + err := pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + client := filer_pb.NewSeaweedFilerClient(grpcConnection) + return fn(client) + }, currentFiler.ToGrpcAddress(), false, s3a.option.GrpcDialOption) + + if err == nil { + return nil + } + + // Current filer failed - try all other filers + // Note: This is a simple failover implementation + // For production, consider implementing exponential backoff and circuit breakers + filers := s3a.filerClient.GetAllFilers() + + for _, filer := range filers { + if filer == currentFiler { + continue // Already tried this one + } + + err = pb.WithGrpcClient(streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + client := filer_pb.NewSeaweedFilerClient(grpcConnection) + return fn(client) + }, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption) + + if err == nil { + // Success! Update current filer for future requests + // This is a simple round-robin update - could be improved with health tracking + glog.V(1).Infof("WithFilerClient: failover from %s to %s succeeded", currentFiler, filer) + return nil + } + + glog.V(2).Infof("WithFilerClient: failover to %s failed: %v", filer, err) + } + + // All filers failed + return fmt.Errorf("all filers failed, last error: %w", err) +} + func (s3a *S3ApiServer) AdjustedUrl(location *filer_pb.Location) string { return location.Url } diff --git a/weed/wdclient/filer_client.go b/weed/wdclient/filer_client.go index 0c6719c6f..057cedf19 100644 --- a/weed/wdclient/filer_client.go +++ b/weed/wdclient/filer_client.go @@ -205,6 +205,18 @@ func (fc *FilerClient) GetCurrentFiler() pb.ServerAddress { return fc.filerAddresses[index] } +// GetAllFilers returns a snapshot of all filer addresses +// Returns a copy to avoid concurrent modification issues +func (fc *FilerClient) GetAllFilers() []pb.ServerAddress { + fc.filerAddressesMu.RLock() + defer fc.filerAddressesMu.RUnlock() + + // Return a copy to avoid concurrent modification + filers := make([]pb.ServerAddress, len(fc.filerAddresses)) + copy(filers, fc.filerAddresses) + return filers +} + // Close stops the filer discovery goroutine if running // Safe to call multiple times (idempotent) func (fc *FilerClient) Close() {