pipedream

README

Pipe Dream

Easy multipart uploads for Amazon S3, DigitalOcean Spaces and S3-compatible services. Available as a CLI and Go library.

CLI

Usage

# Set your secrets, region and endpoint in the environment
export ACCESS_KEY="..."
export SECRET_KEY="..."
export ENDPOINT="sfo2.digitaloceanspaces.com" # for AWS set REGION

# Pipe in data or redirect in a file
pipedream --bucket images --path pets/puppy.jpg < puppy.jpg

# Get fancy
export now=$(date +"%Y-%m-%d_%H:%M:%S_%Z")
cat /data/dump.rdb | gzip | pipedream --bucket backups --path dump-$now.rdb.gz

# For more info
pipedream -h

Installation

Download a build from the releases page. macOS, Linux and Windows builds are available for a variety of architectures.

macOS users can also use Homebrew:

brew tap meowgorithm/tap && brew install meowgorithm/tap/pipedream

Or you can just use go get:

go get github.com/meowgorithm/pipedream/pipedream

Library

The library uses an event based model, sending events through a channel.

import "github.com/meowgorithm/pipedream"

// Create a new multipart upload object
m := pipedream.MultipartUpload{
    AccessKey: os.Getenv("ACCESS_KEY"),
    SecretKey: os.Getenv("SECRET_KEY"),
    Endpoint:  "sfo2.digitaloceanspaces.com", // you could use Region for AWS
    Bucket:    "my-fave-bucket",
}

// Get an io.Reader, like an *os.File or os.Stdout
f, err := os.Open("big-redis-dump.rdb")
if err != nil {
    fmt.Printf("Rats: %v\n", err)
    os.Exit(1)
}
defer f.Close()

// Send up the data. Pipdream returns a channel where you can listen for events
ch := m.Send(f, "backups/dump.rdb")
done := make(chan struct{})

// Listen for activity. For more detailed reporting, see the docs
go func() {
    for {
        e := <-ch
        switch e.(type) {
        case pipedream.Complete:
            fmt.Println("It worked!")
            close(done)
            return
        case pipedream.Error:
            fmt.Println("Rats, it didn't work.")
            close(done)
            return
        }
    }
}()

<-done

Full source of this example. For an example with more detailed reporting, see the source code in the CLI.

Awknowledgements

Thanks to to Apoorva Manjunath‘s S3 multipart upload example for the S3 implementation details.

License

MIT

Documentation

Overview

Package pipedream provides a simple interface to multipart Amazon S3 uploads. It also works with other S3 compatible services such as DigitalOcean's spaces.

The general workflow is to create MultipartUpload struct, run Send on the struct, and then listen for events on the returned channel.

Example usage:

package main

import (
    "fmt"
    "os"

    "github.com/meowgorithm/pipedream"
)

func main() {

    // Prep the multipart upload
    m := pipedream.MultipartUpload{
         AccessKey: os.Getenv("ACCESS_KEY"),
         SecretKey: os.Getenv("SECRET_KEY"),
         Endpoint:  "sfo2.digitaloceanspaces.com", // you could use Region for AWS
         Bucket:    "my-fave-bucket",
    }

    // Get an io.Reader
    f, err := os.Open("big-redis-dump.rdb")
    if err != nil {
         fmt.Printf("Rats: %v\n", err)
         os.Exit(1)
    }
    defer f.Close()

    // Send it up! Pipdream returns a channel where you can listen for events.
    ch := m.Send(f, "backups/dump.rdb")
    done := make(chan struct{})

    // Listen for activity. For more detailed reporting, see the docs below.
    go func() {
        for {
            e := <-ch
            switch e.(type) {
            case pipedream.Complete:
                fmt.Println("It worked!")
                close(done)
                return
            case pipedream.Error:
                fmt.Println("Rats, it didn't work.")
                close(done)
                return
           }
       }
    }()

    <-done
}

There's also a command line interface available at https://github.com/meowgorithm/pipedream/pipedream

Index

• Constants
• func EnglishJoin(words []string, oxfordComma bool) string
• type Complete
• type Error
  • func (e Error) Error() string
• type Event
• type MultipartUpload
  • func (m MultipartUpload) Abort() error
  • func (m *MultipartUpload) Send(reader io.Reader, path string) chan Event
• type Progress
• type Retry

Constants

const (
	// Kilobyte is a convenience measurement useful when setting upload part
	// sizes.
	Kilobyte int64 = 1024

	// Megabyte is a convenience measurement useful when setting upload part
	// sizes.
	Megabyte int64 = Kilobyte * 1024

	// DefaultRegion is the region to use as a default. This should be used for
	// services that don't use regions, like DigitalOcean spaces.
	DefaultRegion = "us-east-1"
)

Functions

func EnglishJoin

func EnglishJoin(words []string, oxfordComma bool) string

EnglishJoin joins a slice of strings with commas and the word "and" like one would in English. Oxford comma optional.

Types

type Complete

type Complete struct {
	Bytes  int
	Result *s3.CompleteMultipartUploadOutput
}

Complete is an Event sent when an upload has completed successfully. When a Complete is received there will be no further activity send on the channel, so you can confidently move on.

type Error

type Error struct {
	Err error
}

Error is an event indicating that an Error occurred during the upload. When an Error is received the operation has failed and no further activity will be send, so you can confidently move on.

func (Error) Error

func (e Error) Error() string

Error returns the a string representation of the error. It satisfies the Error interface.

type Event

type Event interface {
	// contains filtered or unexported methods
}

Event represents activity that occurred during the upload. Events are sent through the channel returned by MultipartUpload.Send(). To figure out which event was received use a type switch or type assertion.

type MultipartUpload

type MultipartUpload struct {
	Endpoint    string
	Region      string
	Bucket      string
	AccessKey   string
	SecretKey   string
	MaxRetries  int
	MaxPartSize int64
	// contains filtered or unexported fields
}

MultipartUpload handles multipart uploads to S3 and S3-compatible systems.

func (MultipartUpload) Abort

func (m MultipartUpload) Abort() error

Abort cancels the upload.

func (*MultipartUpload) Send

func (m *MultipartUpload) Send(reader io.Reader, path string) chan Event

Send uploads data from a given io.Reader (such as an *os.File or os.Stdin) to a given path in a bucket.

type Progress

type Progress struct {
	PartNumber int
	Bytes      int
}

Progress is an Event indicating upload progress. It's sent when a part has successfully uploaded.

type Retry

type Retry struct {
	PartNumber  int
	RetryNumber int
	MaxRetries  int
}

Retry is an Event indicating there was an error uploading a part and the part is being retried. An Error will be send if the retries are exhaused and the upload fails.