Загрузить объект в AWS S3 без создания файла с помощью aws-sdk-go

Question

Я пытаюсь загрузить объект в AWS S3 с помощью golang sdk без необходимости создавать файл в моей системе (пытаюсь загрузить только строку). Но мне трудно это сделать. Кто-нибудь может дать мне пример того, как я могу загрузить в AWS S3 без необходимости создания файла?

Пример загрузки файла AWS:

// Creates a S3 Bucket in the region configured in the shared config
// or AWS_REGION environment variable.
//
// Usage:
//    go run s3_upload_object.go BUCKET_NAME FILENAME
func main() {
    if len(os.Args) != 3 {
        exitErrorf("bucket and file name required\nUsage: %s bucket_name filename",
            os.Args[0])
    }

    bucket := os.Args[1]
    filename := os.Args[2]

    file, err := os.Open(filename)
    if err != nil {
        exitErrorf("Unable to open file %q, %v", err)
    }

    defer file.Close()

    // Initialize a session in us-west-2 that the SDK will use to load
    // credentials from the shared credentials file ~/.aws/credentials.
    sess, err := session.NewSession(&aws.Config{
        Region: aws.String("us-west-2")},
    )

    // Setup the S3 Upload Manager. Also see the SDK doc for the Upload Manager
    // for more information on configuring part size, and concurrency.
    //
    // http://docs.aws.amazon.com/sdk-for-go/api/service/s3/s3manager/#NewUploader
    uploader := s3manager.NewUploader(sess)

    // Upload the file's body to S3 bucket as an object with the key being the
    // same as the filename.
    _, err = uploader.Upload(&s3manager.UploadInput{
        Bucket: aws.String(bucket),

        // Can also use the `filepath` standard library package to modify the
        // filename as need for an S3 object key. Such as turning absolute path
        // to a relative path.
        Key: aws.String(filename),

        // The file to be uploaded. io.ReadSeeker is preferred as the Uploader
        // will be able to optimize memory when uploading large content. io.Reader
        // is supported, but will require buffering of the reader's bytes for
        // each part.
        Body: file,
    })
    if err != nil {
        // Print the error and exit.
        exitErrorf("Unable to upload %q to %q, %v", filename, bucket, err)
    }

    fmt.Printf("Successfully uploaded %q to %q\n", filename, bucket)
}

Я уже пытался создать файл программно, но он создает файл в моей системе, а затем загружает его на S3.

Answer 1

Body поле UploadInput структура просто io.Reader, Так что сдавай io.Reader Вы хотите - это не должен быть файл.

Answer 2

В этом ответе я опубликую все, что сработало для меня, что связано с этим вопросом. Большое спасибо @ThunderCat и @Flimzy, которые предупредили меня, что параметром body запроса на загрузку уже был io.Reader. Я опубликую некоторые примеры кодов, комментируя то, что я узнал из этого вопроса и как это помогло мне решить эту проблему. Возможно, это поможет другим, таким как я и @AlokKumarSingh.

Случай 1: у вас уже есть данные в памяти (например, получение данных из службы потоковой передачи / обмена сообщениями, такой как Kafka, Kinesis или SQS)

func main() {
    if len(os.Args) != 3 {
        fmt.Printf(
            "bucket and file name required\nUsage: %s bucket_name filename",
            os.Args[0],
        )
    }

    bucket := os.Args[1]
    filename := os.Args[2]

    // this is your data that you have in memory
    // in this example it is hard coded but it may come from very distinct
    // sources, like streaming services for example.
    data := "Hello, world!"

    // create a reader from data data in memory
    reader := strings.NewReader(data)

    sess, err := session.NewSession(&aws.Config{
        Region: aws.String("us-east-1")},
    )
    uploader := s3manager.NewUploader(sess)

    _, err = uploader.Upload(&s3manager.UploadInput{
        Bucket: aws.String(bucket),
        Key: aws.String(filename),
        // here you pass your reader
        // the aws sdk will manage all the memory and file reading for you
        Body: reader,
    })
    if err != nil {.
        fmt.Printf("Unable to upload %q to %q, %v", filename, bucket, err)
    }

    fmt.Printf("Successfully uploaded %q to %q\n", filename, bucket)
}

Случай 2. У вас уже есть постоянный файл, и вы хотите загрузить его, но не хотите сохранять весь файл в памяти:

func main() {
    if len(os.Args) != 3 {
        fmt.Printf(
            "bucket and file name required\nUsage: %s bucket_name filename",
            os.Args[0],
        )
    }

    bucket := os.Args[1]
    filename := os.Args[2]

    // open your file
    // the trick here is that the method os.Open just returns for you a reader
    // for the desired file, so you will not maintain the whole file in memory.
    // I know this might sound obvious, but for a starter (as I was at the time
    // of the question) it is not.
    fileReader, err := os.Open(filename)
    if err != nil {
        fmt.Printf("Unable to open file %q, %v", err)
    }
    defer fileReader.Close()

    sess, err := session.NewSession(&aws.Config{
        Region: aws.String("us-east-1")},
    )
    uploader := s3manager.NewUploader(sess)

    _, err = uploader.Upload(&s3manager.UploadInput{
        Bucket: aws.String(bucket),
        Key:    aws.String(filename),
        // here you pass your reader
        // the aws sdk will manage all the memory and file reading for you
        Body: fileReader,
    })
    if err != nil {
        fmt.Printf("Unable to upload %q to %q, %v", filename, bucket, err)
    }

    fmt.Printf("Successfully uploaded %q to %q\n", filename, bucket)
}

Случай 3: так я реализовал это в окончательной версии моей системы, но чтобы понять, почему я это сделал, я должен дать вам некоторую предысторию.

Мой вариант использования немного изменился. Код загрузки должен был стать функцией в Lambda, и файлы оказались огромными. Что означают эти изменения: если я загрузил файл через точку входа в API Gateway, присоединенную к функции Lambda, мне пришлось бы ждать, пока весь файл завершит загрузку в Lambda. Поскольку лямбда определяется длительностью и использованием памяти при вызове, это может быть действительно большой проблемой.

Итак, для решения этой проблемы я использовал предварительно подписанный URL-адрес публикации для загрузки. Как это влияет на архитектуру / рабочий процесс?

Вместо загрузки на S3 из моего внутреннего кода, я просто создаю и аутентифицирую URL для размещения объекта на S3 в серверном и отправляю этот URL во внешний интерфейс. После этого я только что реализовал многочастную загрузку по этому URL. Я знаю, что это гораздо более конкретный вопрос, чем вопрос, но найти это решение было непросто, поэтому я думаю, что было бы неплохо документировать его здесь для других.

Вот пример того, как создать этот предварительно подписанный URL в nodejs.

const AWS = require('aws-sdk');

module.exports.upload = async (event, context, callback) => {

  const s3 = new AWS.S3({ signatureVersion: 'v4' });
  const body = JSON.parse(event.body);

  const params = {
    Bucket: process.env.FILES_BUCKET_NAME,
    Fields: {
      key: body.filename,
    },
    Expires: 60 * 60
  }

  let promise = new Promise((resolve, reject) => {
    s3.createPresignedPost(params, (err, data) => {
      if (err) {
        reject(err);
      } else {
        resolve(data);
      }
    });
  })

  return await promise
    .then((data) => {
      return {
        statusCode: 200,
        body: JSON.stringify({
          message: 'Successfully created a pre-signed post url.',
          data: data,
        })
      }
    })
    .catch((err) => {
      return {
        statusCode: 400,
        body: JSON.stringify({
          message: 'An error occurred while trying to create a pre-signed post url',
          error: err,
        })
      }
    });
};

Если вы хотите использовать go, это та же идея, вам просто нужно изменить de sdk.

Answer 3

Вот небольшая реализация, которую я написал, которая использует каналы и включает тайм-ауты.

package example

import (
    "context"
    "fmt"
    "io"
    "sync"
    "time"

    "github.com/aws/aws-sdk-go/service/s3/s3manager"
)

func FileWriter(ctx context.Context, uploader *s3manager.Uploader, wg *sync.WaitGroup, bucket string, key string, timeout time.Duration) (writer *io.PipeWriter) {
    // create a per-file flush timeout
    fileCtx, cancel := context.WithTimeout(ctx, timeout)

    // pipes are open until one end is closed
    pr, pw := io.Pipe()

    wg.Add(1)
    go func() {
        params := &s3manager.UploadInput{
            Bucket: aws.String(bucket),
            Key:    aws.String(key),
            Body:   pr,
        }

        // blocking
        _, err := uploader.Upload(params)
        if err != nil {
            fmt.Printf("Unable to upload, %v. Bucket: %s", err, bucket)
        }

        // always call context cancel functions!
        cancel()
        wg.Done()
    }()

    // when context is cancelled, close the pipe
    go func() {
        <-fileCtx.Done()
        // should check fileCtx.Err() here
        if err := pw.Close(); err != nil {
            fmt.Printf("Unable to close")
        }
    }()

    return pw
}

Answer 4

Вот что я в итоге написал

func (s *S3Sink) upload() {
    now := time.Now()
    key := s.getNewKey(now)

    _, err := s.uploader.Upload(&s3manager.UploadInput{
        Bucket: aws.String(s.bucket),
        Key:    aws.String(key),
        Body:   s.bodyBuf,
    })

    if err != nil {
        glog.Errorf("Error uploading %s to s3, %v", key, err)
    }
    glog.Infof("Uploaded at %s", key)
    s.lastUploadTimestamp = now.UnixNano()

    s.bodyBuf.Truncate(0)
}

Более подробная информация ниже: https://github.com/heptiolabs/eventrouter/blob/20edca33bc6e20465810d49bdb213119464eb440/sinks/s3sink.go#L185-L201