mongodb - 如何将 exec 函数的标准输出通过管道传递给另一个函数的读取器？

Question

我正在尝试将标准输出从 mongodump 流式传输到 s3。我已经正确掌握了 S3 任意长度流的语法，但我不明白如何将这两个函数结合起来。我不想在开始上传到 S3 之前处理整个 mongodump 命令。这是我目前所拥有的:

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
dumpCmd.Stdout = os.Stdout


uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   dumpCmd.Stdout,
    Bucket: aws.String("myBucket"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

log.Println("Successfully uploaded to", result.Location)

不幸的是，dumpCmd.Stdout 是一个写入器，而不是一个读取器，我不知道如何将写入器的输出通过管道传递给读取器。

在使用管道后，我现在收到一个新错误，这可能属于一个新问题:

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
body, err := dumpCmd.StdoutPipe()
if err != nil {
    // handle error
}

if err := dumpCmd.Start(); err != nil {
    // handle error
}

uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   body,
    Bucket: aws.String("myBucket"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

if err := dumpCmd.Wait(); err != nil {
    // handle error
}

log.Println("Successfully uploaded to", result.Location)

错误:

2016/03/10 12:39:18 Failed to upload MultipartUpload: upload multipart failed
    upload id: QOWW4jBHH4PKjs1Tloc8dlCTtFN94vDHIJIWJChsrjxLZggScZbRUhM4FU9V.xOnIg9uYnBWqOA1x1xqStfA1p8vdAOHNyUp4gOO5b1gbuXvUitQyLdfFhKg9MnyxsV1
caused by: RequestError: send request failed
caused by: Put https://myBucket/myKey?partNumber=1&uploadId=QOWW4jBHH4PKjs1Tloc8dlCTtFN94vDHIJIWJChsrjxLZggScZbRUhM4FU9V.xOnIg9uYnBWqOA1x1xqStfA1p8vdAOHNyUp4gOO5b1gbuXvUitQyLdfFhKg9MnyxsV1: read |0: illegal seek
exit status 1

Answer 1

使用 pipe将命令的输出连接到上传的输入。

这里棘手的问题是上传者 seeks on the body ，但管道不支持查找。为了避免这种情况，在管道周围创建了一个包装器，以对上传者隐藏 Seek 方法。

dumpCmd := exec.Command("mongodump", "--host", "<host>", "--port", "<port>", "--archive")
body, err := dumpCmd.StdoutPipe()
if err != nil {
    // handle error
}

if err := dumpCmd.Start(); err != nil {
    // handle error
}

// Wrap the pipe to hide the seek methods from the uploader
bodyWrap := struct {
    io.Reader
}{body}

uploader := s3manager.NewUploader(session.New(&aws.Config{Region: aws.String("us-east-1")}))
result, err := uploader.Upload(&s3manager.UploadInput{
    Body:   bodyWrap,
    Bucket: aws.String("net-openwhere-mongodb-snapshots-dev"),
    Key:    aws.String("myKey"),
})
if err != nil {
    log.Fatalln("Failed to upload", err)
}

if err := dumpCmd.Wait(); err != nil {
   // handle error
}

log.Println("Successfully uploaded to", result.Location)