func TestDynamoDBHABackend(t *testing.T) {
	if os.Getenv("AWS_ACCESS_KEY_ID") == "" || os.Getenv("AWS_SECRET_ACCESS_KEY") == "" {
		t.SkipNow()
	}

	creds, err := credentials.NewEnvCredentials().Get()
	if err != nil {
		t.Fatalf("err: %v", err)
	}

	// If the variable is empty or doesn't exist, the default
	// AWS endpoints will be used
	endpoint := os.Getenv("AWS_DYNAMODB_ENDPOINT")

	region := os.Getenv("AWS_DEFAULT_REGION")
	if region == "" {
		region = "us-east-1"
	}

	conn := dynamodb.New(session.New(&aws.Config{
		Credentials: credentials.NewEnvCredentials(),
		Endpoint:    aws.String(endpoint),
		Region:      aws.String(region),
	}))

	var randInt = rand.New(rand.NewSource(time.Now().UnixNano())).Int()
	table := fmt.Sprintf("vault-dynamodb-testacc-%d", randInt)

	defer func() {
		conn.DeleteTable(&dynamodb.DeleteTableInput{
			TableName: aws.String(table),
		})
	}()

	logger := logformat.NewVaultLogger(log.LevelTrace)
	b, err := NewBackend("dynamodb", logger, map[string]string{
		"access_key":    creds.AccessKeyID,
		"secret_key":    creds.SecretAccessKey,
		"session_token": creds.SessionToken,
		"table":         table,
	})
	if err != nil {
		t.Fatalf("err: %s", err)
	}

	ha, ok := b.(HABackend)
	if !ok {
		t.Fatalf("dynamodb does not implement HABackend")
	}
	testHABackend(t, ha, ha)
}

func elbDNSName(c *conf) string {
	//weirdness around how aws handles credentials
	os.Setenv("AWS_ACCESS_KEY_ID", c.awsAccessKey)
	os.Setenv("AWS_SECRET_ACCESS_KEY", c.awsSecretKey)

	svc := elb.New(
		session.New(
			&aws.Config{
				Region:      aws.String(c.awsRegion),
				Credentials: credentials.NewEnvCredentials(),
			},
		),
	)

	params := &elb.DescribeLoadBalancersInput{
		LoadBalancerNames: []*string{
			aws.String(c.elbName), // Required
		},
	}

	resp, err := svc.DescribeLoadBalancers(params)

	if err != nil {
		log.Fatal(err)
	}

	return *resp.LoadBalancerDescriptions[0].DNSName
}

func main() {
	flag.StringVar(&streamName, "stream", "mystream", "what stream to tail")
	flag.StringVar(&shardName, "shard", "shardId-000000000000", "what shard to tail")
	flag.IntVar(&delayTime, "delay", 5, "time to sleep between GetRecord calls")

	flag.Parse()

	println(fmt.Sprintf("Working on kinesis stream: %s:%s", streamName, shardName))

	creds := credentials.NewEnvCredentials()
	config = aws.Config{}
	config.Credentials = creds
	config.Region = aws.String("us-east-1")

	// TODO, ensure we can load the region from ENV variables.
	// Bypass the error now and just use us-east-1
	if *defaults.DefaultConfig.Region == "" {
		println("Could not find AWS Region in ENV. Please configure your ENV for AWS access")
		//os.Exit(1)
	}

	listStreams()
	describeStream(streamName)

	watchStream(streamName, shardName)
}

// GetAWSCreds returns the appropriate value as the need arises.
//
// evaluated in the following order
// 1. input variable
// 2. Environment variable
// 3. IAM Role
//
// "/.aws/credentials" necessary item increased about that, so it isn't used.
func (c *Config) GetAWSCreds() (*credentials.Credentials, error) {
	var creds *credentials.Credentials
	var err error

	err = nil
	// 1. input variable used
	if c.Aws.Accesskey != "" && c.Aws.SecretKey != "" {
		creds = credentials.NewStaticCredentials(c.Aws.Accesskey, c.Aws.SecretKey, "")
		creds.Expire()
		_, err = creds.Get()
	}

	if err != nil {
		// 2. Environment variable used
		creds = credentials.NewEnvCredentials()
		creds.Expire()
		_, err = creds.Get()

		if err != nil {
			// 3. IAM Role used
			creds = credentials.NewCredentials(&ec2rolecreds.EC2RoleProvider{})
			creds.Expire()
			_, err = creds.Get()
		}
	}

	return creds, err
}

// NewSNSPublisher will initiate the SNS client.
// If no credentials are passed in with the config,
// the publisher is instantiated with the AWS_ACCESS_KEY
// and the AWS_SECRET_KEY environment variables.
func NewSNSPublisher(cfg *config.SNS) (*SNSPublisher, error) {
	p := &SNSPublisher{}

	if cfg.Topic == "" {
		return p, errors.New("SNS topic name is required")
	}
	p.topic = cfg.Topic

	if cfg.Region == "" {
		return p, errors.New("SNS region is required")
	}

	var creds *credentials.Credentials
	if cfg.AccessKey != "" {
		creds = credentials.NewStaticCredentials(cfg.AccessKey, cfg.SecretKey, "")
	} else {
		creds = credentials.NewEnvCredentials()
	}

	p.sns = sns.New(session.New(&aws.Config{
		Credentials: creds,
		Region:      &cfg.Region,
	}))
	return p, nil
}

// Svc configures the DynamoDB service to use
func Svc(opts config.Options) *dynamodb.DynamoDB {
	awsConfig := &aws.Config{Region: aws.String(opts.Storage.AWS.Region)}

	// If a session was passed... (AWS Lambda does this)
	if opts.Storage.AWS.SessionToken != "" {
		os.Setenv("AWS_SESSION_TOKEN", opts.Storage.AWS.SessionToken)
	}

	// Look in a variety of places for AWS credentials. First, try the credentials file set by AWS CLI tool.
	// Note the empty string instructs to look under default file path (different based on OS).
	// This file can have multiple profiles and a default profile will be used unless otherwise configured.
	// See: https://godoc.org/github.com/aws/aws-sdk-go/aws/credentials#SharedCredentialsProvider
	creds := credentials.NewSharedCredentials("", opts.Storage.AWS.CredProfile)
	_, err := creds.Get()
	// If that failed, try environment variables.
	if err != nil {
		// The following are checked:
		// Access Key ID: AWS_ACCESS_KEY_ID or AWS_ACCESS_KEY
		// Secret Access Key: AWS_SECRET_ACCESS_KEY or AWS_SECRET_KEY
		creds = credentials.NewEnvCredentials()
	}

	// If credentials were passed via config, then use those. They will take priority over other methods.
	if opts.Storage.AWS.AccessKeyID != "" && opts.Storage.AWS.SecretAccessKey != "" {
		creds = credentials.NewStaticCredentials(opts.Storage.AWS.AccessKeyID, opts.Storage.AWS.SecretAccessKey, "")
	}
	awsConfig.Credentials = creds

	return dynamodb.New(session.New(awsConfig))
}

func TestMigrateStreams(t *testing.T) {
	if os.Getenv("AWS_ACCESS_KEY_ID") == "" || os.Getenv("AWS_SECRET_ACCESS_KEY") == "" {
		Log.Printf("WARNING: NO AWS CREDS SPECIFIED, SKIPPING MIGRATIONS TEST")
		return
	}

	config := &aws.Config{
		Credentials: credentials.NewEnvCredentials(),
		Region:      aws.String("us-east-1"),
	}
	client := kinesis.New(session.New(config))

	sm := StreamMigrator{
		Streams: []kinesis.CreateStreamInput{
			kinesis.CreateStreamInput{
				StreamName: aws.String(testDomain),
				ShardCount: aws.Int64(1),
			},
		},
		Client:  client,
		Timeout: 30,
	}

	sm.Migrate()
	sm.Migrate()

}

// MustClient will use the cache cluster ID to describe
// the cache cluster and instantiate a memcache.Client
// with the cache nodes returned from AWS.
func (e *ElastiCache) MustClient() *memcache.Client {
	var creds *credentials.Credentials
	if e.AccessKey != "" {
		creds = credentials.NewStaticCredentials(e.AccessKey, e.SecretKey, "")
	} else {
		creds = credentials.NewEnvCredentials()
	}

	ecclient := elasticache.New(session.New(&aws.Config{
		Credentials: creds,
		Region:      &e.Region,
	}))

	resp, err := ecclient.DescribeCacheClusters(&elasticache.DescribeCacheClustersInput{
		CacheClusterId:    &e.ClusterID,
		ShowCacheNodeInfo: aws.Bool(true),
	})
	if err != nil {
		log.Fatalf("unable to describe cache cluster: %s", err)
	}

	var nodes []string
	for _, cluster := range resp.CacheClusters {
		for _, cnode := range cluster.CacheNodes {
			addr := fmt.Sprintf("%s:%d", *cnode.Endpoint.Address, *cnode.Endpoint.Port)
			nodes = append(nodes, addr)
		}
	}

	return memcache.New(nodes...)
}

func main() {
	var err error

	// connect to the database
	dbconn, err = sql.Open("postgres", PACKAGEBUG_DB)
	if err != nil {
		log.Fatal(err)
	}

	// make sure the database up
	err = dbconn.Ping()
	if err != nil {
		log.Fatal(err)
	}

	// set up aws SDK credentials & config
	cred := credentials.NewEnvCredentials()
	_, err = cred.Get()
	if err != nil {
		log.Fatal(err)
	}
	config := aws.NewConfig()
	config.Credentials = cred
	config.Endpoint = &PACKAGEBUG_SQS_ENDPOINT
	config.Region = &PACKAGEBUG_SQS_REGION

	sqsconn = sqs.New(config)
	// dispatch jobs once a day
	for {
		<-time.After(24 * time.Hour)
		go dispatchJobs()
	}
}

func addToLambda(dir string) error {
	desc, err := util.ReadTestDescription(dir)
	if err != nil {
		return err
	}

	var zipContents []byte
	if desc.Runtime == "java8" {
		zipContents, err = ioutil.ReadFile(filepath.Join(dir, "test-build.jar"))
		if err != nil {
			return err
		}
	} else {
		zipContents, err = makeZip(dir)
		if err != nil {
			return err
		}
	}

	s := session.New(&aws.Config{Region: aws.String("us-east-1"), Credentials: credentials.NewEnvCredentials()})

	l := lambda.New(s)

	err = createLambdaFunction(l, zipContents, desc.Runtime, lambdaRole, desc.Name, desc.Handler, desc.Timeout)
	return err
}

func (s *S3) Init() error {
	if s.Bucket == "" {
		return errors.New("S3 bucket not set")
	} else if s.Key == "" {
		return errors.New("S3 key not set")
	}
	if s.Region == "" {
		s.Region = "ap-southeast-2"
	}
	creds := credentials.AnonymousCredentials
	if s.Access != "" {
		creds = credentials.NewStaticCredentials(s.Access, s.Secret, "")
	} else if os.Getenv("AWS_ACCESS_KEY") != "" {
		creds = credentials.NewEnvCredentials()
	}
	config := &aws.Config{
		Credentials: creds,
		Region:      &s.Region,
	}
	s.client = s3.New(session.New(config))

	//TODO include this? maybe given access to bucket after init
	// resp, err := s.client.HeadBucketRequest(&s3.HeadBucketInput{Bucket: &s.Bucket})
	// if err != nil {}

	//apply defaults
	if s.Interval == 0 {
		s.Interval = 5 * time.Minute
	}
	return nil
}

// NewKinesisRelay gets the relay started.
func NewKinesisRelay(l Logger, c ConfigValues) *KinesisRelay {
	l.Info.Println("Beginning to start kinesis")

	// New KinesisRelay setup.
	kr := new(KinesisRelay)
	kr.logger = l
	kr.config = c

	// Setup the AWS config.
	creds := credentials.NewEnvCredentials()
	kr.awsConfig = aws.Config{}
	kr.awsConfig.Credentials = creds

	// TODO - handle this via ENV variables as well.
	kr.awsConfig.Region = aws.String(kr.config.Kinesis.Region)

	// TODO configure the channel better in the future.
	kr.Pipe = make(chan string, 100)

	// Prep the stream.
	l.Trace.Printf("We want to use stream: %s", kr.config.Kinesis.StreamName)
	kr.createStreamIfNotExists(kr.config.Kinesis.StreamName)

	l.Info.Println("Done starting kinesis relay")
	return kr
}

func main() {
	if len(os.Args) != 3 {
		log.Fatal("wrong number of arguments")
	}
	src, _ := os.Args[1], os.Args[2]

	creds := credentials.NewEnvCredentials()
	if _, err := creds.Get(); err != nil {
		log.Fatal(err)
	}

	svc := s3.New(&aws.Config{
		Credentials:      creds,
		Region:           "us-east-2",
		Endpoint:         "s3.amazonaws.com",
		S3ForcePathStyle: true,
	})

	uploader := s3manager.NewUploader(&s3manager.UploadOptions{S3: svc})

	watcher, err := fsnotify.NewWatcher()
	if err != nil {
		log.Fatal(err)
	}
	defer watcher.Close()

	done := make(chan bool)
	go func() {
		for {
			select {
			case event := <-watcher.Events:
				log.Println("event:", event)
				switch event.Op {
				case fsnotify.Create:
					handleCreate(watcher, uploader, event.Name)
				case fsnotify.Write:
					log.Println("modified file:", event.Name)
				}

			case err := <-watcher.Errors:
				log.Println("error:", err)
			}
		}
	}()

	err = filepath.Walk(src, func(path string, info os.FileInfo, err error) error {
		if info.IsDir() {
			log.Println(path)
			return watcher.Add(path)
		}
		return nil
	})
	if err != nil {
		log.Fatal(err)
	}

	<-done

}

func loadConfig() *aws.Config {
	region := os.Getenv("AWS_REGION")
	if region == "" {
		region = "eu-west-1"
	}
	creds := credentials.NewEnvCredentials()
	return aws.NewConfig().WithCredentials(creds).WithRegion(region)
}

func createConfig() *aws.Config {
	config := aws.NewConfig()
	config.WithCredentials(credentials.NewEnvCredentials())
	config.WithRegion(*region)
	if *logging {
		config.WithLogLevel(aws.LogDebugWithHTTPBody)
	}
	return config
}

func main() {
	flag.Parse()
	log.Println("Total:", *total)
	log.Println("Gophers:", *gophers)
	log.Println("ID Top:", *idTop)
	if *debug {
		*verbose = true
	}

	db := dynamodb.New(session.New(
		aws.NewConfig().
			WithEndpoint(*endpoint).
			WithRegion(*region).
			WithCredentials(credentials.NewEnvCredentials()),
	))

	jobChans := make(chan struct{}, *gophers)
	start := time.Now()
	var totalDuration int64
	var wg sync.WaitGroup

	rand.Seed(time.Now().Unix())
	for i := 0; i <= *total; i++ {
		jobChans <- struct{}{}
		wg.Add(1)
		if i%((*total)/100) == 0 {
			fmt.Printf("\r%s Queried %d%%", time.Now().Format("2006-01-02 15:04:05"), i/((*total)/100))
		}
		go func() {
			start := time.Now()
			if resp, err := db.GetItem(&dynamodb.GetItemInput{
				TableName: aws.String(*table),
				Key: map[string]*dynamodb.AttributeValue{
					// "bench_area": &dynamodb.AttributeValue{S: aws.String("KingsLanding")},
					"id": &dynamodb.AttributeValue{S: aws.String(fmt.Sprintf("%s%d", *idPrefix, rand.Intn(*idTop)))},
				},
			}); err != nil {
				fmt.Println(err)
			} else {
				if *debug {
					log.Println(resp)
				}
			}

			atomic.AddInt64(&totalDuration, int64(time.Now().Sub(start)))
			wg.Done()
			<-jobChans
		}()
	}

	wg.Wait()

	fmt.Println("")
	log.Println("Took:", time.Now().Sub(start))
	log.Println("Total Duration:", time.Duration(totalDuration))
	log.Println("TPQ:", time.Duration(totalDuration/int64(*total)))
}

func New(region, keyID string) (*KMS, error) {
	config := aws.NewConfig().WithCredentials(credentials.NewEnvCredentials()).WithRegion(region)
	session := session.New(config)
	kms := &KMS{
		kms:   kms.New(session),
		keyID: keyID,
	}
	return kms, nil
}

func main() {

	bucketFlag := flag.String("bucket", "my-s3-bucket", "s3 bucket to upload to")
	subfolderFlag := flag.String("subfolder", "", "subfolder in s3 bucket, can be blank")
	num_workersFlag := flag.Int("workers", 100, "number of upload workers to use")
	regionFlag := flag.String("region", "eu-west-1", "aws region")
	aclFlag := flag.String("acl", "private", "s3 upload acl - use either private or public")
	sourceDirFlag := flag.String("sourcedir", "files/", "source directory")
	destDirFlag := flag.String("destdir", "files-uploaded/", "dest dir for uploaded files (on local box)")

	flag.Parse()

	bucket := *bucketFlag
	subfolder := *subfolderFlag
	num_workers := *num_workersFlag
	region := *regionFlag
	acl := *aclFlag
	sourceDir := *sourceDirFlag
	destDir := *destDirFlag

	fmt.Println("Using options:")
	fmt.Println("bucket:", bucket)
	fmt.Println("subfolder:", subfolder)
	fmt.Println("num_workers:", num_workers)
	fmt.Println("region:", region)
	fmt.Println("acl:", acl)
	fmt.Println("sourceDir:", sourceDir)
	fmt.Println("destDir:", destDir)

	var wg sync.WaitGroup
	wg.Add(num_workers + 1) // add 1 to account for the get_file_list thread!

	// file channel and thread to get the files
	file_channel := make(chan string, 0)
	go get_file_list(sourceDir, file_channel, num_workers, &wg)

	// set up s3 credentials from environment variables
	// these are shared to every worker
	creds := credentials.NewEnvCredentials()

	fmt.Println("Starting " + strconv.Itoa(num_workers) + " workers...")

	// create the desired number of workers
	for i := 1; i <= num_workers; i++ {
		// make a new worker
		sess := session.New(&aws.Config{Region: aws.String(region), Credentials: creds, LogLevel: aws.LogLevel(1)})
		svc := s3.New(sess)
		worker := &Worker{Acl: acl, Bucket: bucket, Subfolder: subfolder, Svc: svc, File_channel: file_channel, Wg: &wg, SourceDir: sourceDir, DestDir: destDir, Id: i}
		go worker.doUploads()
	}

	// wait for all workers to finish
	// (1x file worker and all uploader workers)
	wg.Wait()
}

func checkCreds() (err error) {
	creds := credentials.NewEnvCredentials()
	_, err = creds.Get()
	if err == nil {
		// If ENV credentials are present, I don't need to check shared credentials on fs
		return
	}
	creds = credentials.NewSharedCredentials("", "")
	_, err = creds.Get()
	return
}

func New(region, keyID string) (*KMS, error) {
	config := &aws.Config{
		Credentials: credentials.NewEnvCredentials(),
		Region:      &region,
	}
	kms := &KMS{
		kms:   kms.New(config),
		keyID: keyID,
	}
	return kms, nil
}