如何使用 go 将 UploadPart S3 操作从传入请求流式传输到 AWS S3？

Question

上下文与我的团队一起，我们正在构建一个反向代理来拦截对S3的所有传出请求，以便审核和控制来自不同应用程序的访问。我们通过流式传输文件内容，成功实现了几乎所有操作。例如，为了使用单个操作上传，我们使用将传入请求的正文（这是一个）流式传输到S3并下载（单部分和多部分风格），我们使用原语从（这是一个）写入响应。s3manager.Uploaderio.Readerio.Copys3.GetObjectOutput.Bodyio.ReadCloser问题：我们仍然无法通过流式处理实现的唯一操作是上传部分（在分段上传的上下文中）。问题是需要一个 和 来传递传入请求的正文，你需要在某个位置（例如，在内存中）缓冲它。s3.UploadPartInputaws.ReadSeekCloser这就是我们到目前为止所拥有的：func (ph *VaultProxyHandler) HandleUploadPart(w http.ResponseWriter, r *http.Request, s3api s3iface.S3API, bucket string, key string, uploadID string, part int64) {    buf := bytes.NewBuffer(nil)        // here loads the entire body to memory    if _, err := io.Copy(buf, r.Body); err != nil {        http.Error(w, err.Error(), http.StatusInternalServerError)        return    }    payload := buf.Bytes()    input := &s3.UploadPartInput{        Bucket:     aws.String(bucket),        Key:        aws.String(key),        UploadId:   aws.String(uploadID),        PartNumber: aws.Int64(part),        Body:       aws.ReadSeekCloser(bytes.NewReader(payload)),    }    output, err := s3api.UploadPart(input)    // and so on...}问题：有没有办法将 的传入请求流式传输到 S3？（我的意思是不要将整个身体存储在内存中）。UploadPart

Answer 1

最后，我得到了一种方法，通过使用 AWS 开发工具包构建请求并使用未签名的有效负载对其进行签名，通过流处理反向代理传入的 UploadPart。

下面是一个基本示例：

type AwsService struct {

    Region   string

    S3Client s3iface.S3API

    Signer   *v4.Signer

}

func NewAwsService(region string, accessKey string, secretKey string, sessionToken string) (*AwsService, error) {

    creds := credentials.NewStaticCredentials(accessKey, secretKey, sessionToken)

    awsConfig := aws.NewConfig().

        WithRegion(region).

        WithCredentials(creds).

        WithCredentialsChainVerboseErrors(true)

    sess, err := session.NewSession(awsConfig)

    if err != nil {

        return nil, err

    }

    svc := s3.New(sess)

    signer := v4.NewSigner(creds)

    v4.WithUnsignedPayload(signer)

    return &AwsService{

        Region:   region,

        S3Client: svc,

        Signer:   signer,

    }, nil

}

func (s *AwsService) UploadPart(bucket string, key string, part int, uploadID string, payloadReader io.Reader, contentLength int64) (string, error) {

    input := &s3.UploadPartInput{

        Bucket:        aws.String(bucket),

        Key:           aws.String(key),

        UploadId:      aws.String(uploadID),

        PartNumber:    aws.Int64(int64(part)),

        ContentLength: aws.Int64(contentLength),

        Body:          aws.ReadSeekCloser(payloadReader),

    }

    req, output := s.S3Client.UploadPartRequest(input)

    _, err := s.Signer.Sign(req.HTTPRequest, req.Body, s3.ServiceName, s.Region, time.Now())

    err = req.Send()

    if err != nil {

        return "", err

    }

    return *output.ETag, nil

}

然后，可以从处理程序调用它：

func HandleUploadPart(w http.ResponseWriter, r *http.Request) {

    query := r.URL.Query()

    region := query.Get("region")

    bucket := query.Get("bucket")

    key := query.Get("key")

    part, err := strconv.Atoi(query.Get("part"))

    if err != nil {

        http.Error(w, err.Error(), http.StatusInternalServerError)

        return

    }

    uploadID := query.Get("upload-id")

    payloadReader := r.Body

    contentLength, err := strconv.ParseInt(r.Header.Get("Content-Length"), 10, 64)

    if err != nil {

        http.Error(w, err.Error(), http.StatusInternalServerError)

        return

    }

    etag, err := awsService.UploadPart(region, bucket, key, part, uploadID, payloadReader, contentLength)

    if err != nil {

        http.Error(w, err.Error(), http.StatusInternalServerError)

        return

    }

    w.Header().Set("ETag", etag)

}

缺点：

客户端必须提前知道内容长度并发送。

无法对有效负载进行签名。