如何在Go语言中实现SFTP与Azure Blob Storage的无本地存储集成

可以实现客户端上传文件到SFTP后直接流式上传至Azure Blob Storage，无需本地存储；下载功能同样可以采用流式传输的方式，直接从Blob拉取数据返回给客户端，不落地到本地。以下是具体实现方法：

一、上传功能改造（无本地存储） #

核心思路是自定义SFTP文件系统，替换默认的本地文件系统实现，当客户端上传文件时，直接将数据流导向Azure Blob的上传接口，而非写入本地磁盘。

1. 自定义Blob文件系统实现 #

import (
    "context"
    "fmt"
    "io"
    "os"
    "time"

    "github.com/pkg/sftp"
    "github.com/Azure/azure-storage-blob-go/azblob"
    "golang.org/x/net/context"
)

// BlobFS 自定义SFTP文件系统，对接Azure Blob
type BlobFS struct {
    blobStorage *Storage // 你的BlobStorage实例
    ctx         context.Context
}

// Create 处理SFTP的文件创建请求，返回流式写入器
func (fs *BlobFS) Create(name string) (sftp.File, error) {
    // 创建管道，一边接收SFTP客户端的写入，一边流式上传到Blob
    pr, pw := io.Pipe()

    // 启动goroutine处理流式上传
    go func() {
        defer pr.Close()
        // 构造Blob URL
        u, _ := url.Parse(fmt.Sprintf("%s/%s", fs.blobStorage.blobURL, name))
        blockBlobURL := azblob.NewBlockBlobURL(*u, fs.blobStorage.pipeline)
        
        // 流式上传到Blob，无需缓存整个文件到内存
        options := azblob.UploadStreamToBlockBlobOptions{
            BlobHTTPHeaders: azblob.BlobHTTPHeaders{
                ContentType: "text/csv", // 根据实际文件类型调整
            },
        }
        _, err := azblob.UploadStreamToBlockBlob(fs.ctx, pr, blockBlobURL, options)
        if err != nil {
            slog.ErrorContext(fs.ctx, "流式上传Blob失败", slog.Any(constants.Error, err))
        }
    }()

    // 返回自定义的文件写入器
    return &BlobFileWriter{Writer: pw}, nil
}

// BlobFileWriter 实现sftp.File接口的写入器
type BlobFileWriter struct {
    io.Writer
    closed bool
}

func (fw *BlobFileWriter) Close() error {
    if fw.closed {
        return nil
    }
    fw.closed = true
    if closer, ok := fw.Writer.(io.Closer); ok {
        return closer.Close()
    }
    return nil
}

func (fw *BlobFileWriter) Read(p []byte) (n int, err error) {
    return 0, io.EOF // 上传文件无需读操作
}

func (fw *BlobFileWriter) Seek(offset int64, whence int) (int64, error) {
    return 0, fmt.Errorf("Blob上传不支持Seek操作")
}

// Open 暂时返回错误，下载功能后续实现
func (fs *BlobFS) Open(name string) (sftp.File, error) {
    return nil, fmt.Errorf("暂未实现下载功能")
}

// Stat 返回模拟的文件信息，可根据需求对接Blob API获取真实信息
func (fs *BlobFS) Stat(name string) (os.FileInfo, error) {
    return &BlobFileInfo{name: name}, nil
}

// BlobFileInfo 实现os.FileInfo接口
type BlobFileInfo struct {
    name string
}

func (fi *BlobFileInfo) Name() string    { return fi.name }
func (fi *BlobFileInfo) Size() int64     { return 0 }
func (fi *BlobFileInfo) Mode() os.FileMode { return 0644 }
func (fi *BlobFileInfo) ModTime() time.Time { return time.Now() }
func (fi *BlobFileInfo) IsDir() bool     { return false }
func (fi *BlobFileInfo) Sys() interface{} { return nil }

// 按需实现其他FileSystem接口方法（如Remove、Mkdir等）
func (fs *BlobFS) Remove(name string) error {
    return fmt.Errorf("暂未实现删除功能")
}

2. 修改SFTP服务器初始化代码 #

在创建sftp.Server时，传入自定义的文件系统：

// 初始化你的BlobStorage实例
blobStorage := &Storage{
    blobURL:    os.Getenv("AZURE_BLOB_URL"),
    pipeline:   yourPipeline, // 提前初始化好的Blob pipeline
    credential: yourCredential,
}

// 创建SFTP服务器时指定自定义文件系统
server, err := sftp.NewServer(channel, sftp.WithFileSystem(&BlobFS{
    blobStorage: blobStorage,
    ctx:         ctx,
}))
if err != nil {
    log.Fatal("创建SFTP服务器失败:", err)
}
if err := server.Serve(); err != nil {
    if err != io.EOF {
        log.Fatal("SFTP服务器运行出错:", err)
    }
}
server.Close()

二、下载功能改造（无本地存储） #

同样通过自定义文件系统的Open方法，直接从Azure Blob拉取数据流返回给SFTP客户端，无需本地缓存。

1. 完善BlobFS的Open和Stat方法 #

// Open 处理SFTP的文件读取请求，返回Blob的下载流
func (fs *BlobFS) Open(name string) (sftp.File, error) {
    u, _ := url.Parse(fmt.Sprintf("%s/%s", fs.blobStorage.blobURL, name))
    blockBlobURL := azblob.NewBlockBlobURL(*u, fs.blobStorage.pipeline)
    
    // 从Blob获取下载流
    getResp, err := blockBlobURL.Download(fs.ctx, 0, azblob.CountToEnd, azblob.BlobAccessConditions{}, false, "")
    if err != nil {
        return nil, err
    }
    
    // 返回自定义的文件读取器
    return &BlobFileReader{Reader: getResp.Body(azblob.RetryReaderOptions{})}, nil
}

// BlobFileReader 实现sftp.File接口的读取器
type BlobFileReader struct {
    io.Reader
    closed bool
}

func (fr *BlobFileReader) Close() error {
    if fr.closed {
        return nil
    }
    fr.closed = true
    if closer, ok := fr.Reader.(io.Closer); ok {
        return closer.Close()
    }
    return nil
}

func (fr *BlobFileReader) Write(p []byte) (n int, err error) {
    return 0, fmt.Errorf("Blob下载不支持写操作")
}

func (fr *BlobFileReader) Seek(offset int64, whence int) (int64, error) {
    // 如需支持Seek，需重新请求Blob的指定Range数据，此处简化返回不支持
    return 0, fmt.Errorf("Blob下载暂不支持Seek操作")
}

// Stat 获取Blob的真实文件信息
func (fs *BlobFS) Stat(name string) (os.FileInfo, error) {
    u, _ := url.Parse(fmt.Sprintf("%s/%s", fs.blobStorage.blobURL, name))
    blockBlobURL := azblob.NewBlockBlobURL(*u, fs.blobStorage.pipeline)
    
    props, err := blockBlobURL.GetProperties(fs.ctx, azblob.BlobAccessConditions{})
    if err != nil {
        return nil, err
    }
    
    return &BlobFileInfo{
        name:    name,
        size:    props.ContentLength(),
        modTime: props.LastModified(),
    }, nil
}

// 更新BlobFileInfo结构体
type BlobFileInfo struct {
    name    string
    size    int64
    modTime time.Time
}

func (fi *BlobFileInfo) Size() int64     { return fi.size }
func (fi *BlobFileInfo) ModTime() time.Time { return fi.modTime }

关键注意事项 #

1. 流式传输：通过io.Pipe和Azure Blob的流式上传/下载接口，避免将大文件加载到内存，降低资源占用。
2. 接口完整性：需根据SFTP客户端的实际需求，完善FileSystem接口的其他方法（如Remove、Mkdir、Rename等）。
3. 错误处理：需补充更完善的错误捕获和日志记录，确保上传/下载失败时能及时反馈给客户端。
4. 权限控制：当前代码关闭了客户端认证（NoClientAuth: true），生产环境需添加用户认证逻辑，避免非法访问。

内容的提问来源于stack exchange，提问作者Kaushal