func BenchmarkPutItem(b *testing.B) {
	cfg := aws.Config{
		DisableSSL:  aws.Bool(true),
		Credentials: credentials.NewStaticCredentials("AKID", "SECRET", ""),
	}
	server := successRespServer([]byte(`{}`))
	cfg.Endpoint = aws.String(server.URL)

	svc := dynamodb.New(&cfg)
	svc.Handlers.Send.Clear()
	svc.Handlers.Send.PushBack(func(r *service.Request) {
		r.HTTPResponse = &http.Response{
			StatusCode: http.StatusOK,
			Status:     http.StatusText(http.StatusOK),
			Body:       noopBody,
		}
	})

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		av, err := dynamodbattribute.ConvertToMap(dbItem{Key: "MyKey", Data: "MyData"})
		if err != nil {
			b.Fatal("benchPutItem, expect no ConvertToMap errors", err)
		}
		params := &dynamodb.PutItemInput{
			Item:      av,
			TableName: aws.String("tablename"),
		}
		_, err = svc.PutItem(params)
		if err != nil {
			b.Error("benchPutItem, expect no request errors", err)
		}
	}
}

func init() {
	var ecsconfig aws.Config
	if region := os.Getenv("AWS_REGION"); region != "" {
		ecsconfig.Region = &region
	}
	if region := os.Getenv("AWS_DEFAULT_REGION"); region != "" {
		ecsconfig.Region = &region
	}
	if ecsconfig.Region == nil {
		if iid, err := ec2.GetInstanceIdentityDocument(); err == nil {
			ecsconfig.Region = &iid.Region
		}
	}
	if envEndpoint := os.Getenv("ECS_BACKEND_HOST"); envEndpoint != "" {
		ecsconfig.Endpoint = &envEndpoint
	}

	ECS = ecs.New(session.New(&ecsconfig))
	Cluster = "ecs-functional-tests"
	if envCluster := os.Getenv("ECS_CLUSTER"); envCluster != "" {
		Cluster = envCluster
	}
	ECS.CreateCluster(&ecs.CreateClusterInput{
		ClusterName: aws.String(Cluster),
	})
}

// New initializes a new S3 client connection based on config.
func New() *S3Client {
	var (
		cfg *aws.Config
	)

	if config.S3.Endpoint != "" {
		cfg = &aws.Config{
			Endpoint:         aws.String(config.S3.Endpoint),
			DisableSSL:       aws.Bool(strings.HasPrefix(config.S3.Endpoint, "http://")),
			Region:           aws.String(config.S3.Region),
			S3ForcePathStyle: aws.Bool(config.S3.PathStyle),
		}
	} else {
		cfg = &aws.Config{
			Region:           aws.String(config.S3.Region),
			S3ForcePathStyle: aws.Bool(config.S3.PathStyle),
		}
	}

	if config.S3.Access != "" && config.S3.Secret != "" {
		cfg.Credentials = credentials.NewStaticCredentials(
			config.S3.Access,
			config.S3.Secret,
			"",
		)
	}

	return &S3Client{
		client: s3.New(
			session.New(),
			cfg,
		),
	}
}

func mergeConfigSrcs(cfg, userCfg *aws.Config, envCfg envConfig, sharedCfg sharedConfig, handlers request.Handlers) {
	// Merge in user provided configuration
	cfg.MergeIn(userCfg)

	// Region if not already set by user
	if len(aws.StringValue(cfg.Region)) == 0 {
		if len(envCfg.Region) > 0 {
			cfg.WithRegion(envCfg.Region)
		} else if envCfg.EnableSharedConfig && len(sharedCfg.Region) > 0 {
			cfg.WithRegion(sharedCfg.Region)
		}
	}

	// Configure credentials if not already set
	if cfg.Credentials == credentials.AnonymousCredentials && userCfg.Credentials == nil {
		if len(envCfg.Creds.AccessKeyID) > 0 {
			cfg.Credentials = credentials.NewStaticCredentialsFromCreds(
				envCfg.Creds,
			)
		} else if envCfg.EnableSharedConfig && len(sharedCfg.AssumeRole.RoleARN) > 0 && sharedCfg.AssumeRoleSource != nil {
			cfgCp := *cfg
			cfgCp.Credentials = credentials.NewStaticCredentialsFromCreds(
				sharedCfg.AssumeRoleSource.Creds,
			)
			cfg.Credentials = stscreds.NewCredentials(
				&Session{
					Config:   &cfgCp,
					Handlers: handlers.Copy(),
				},
				sharedCfg.AssumeRole.RoleARN,
				func(opt *stscreds.AssumeRoleProvider) {
					opt.RoleSessionName = sharedCfg.AssumeRole.RoleSessionName

					if len(sharedCfg.AssumeRole.ExternalID) > 0 {
						opt.ExternalID = aws.String(sharedCfg.AssumeRole.ExternalID)
					}

					// MFA not supported
				},
			)
		} else if len(sharedCfg.Creds.AccessKeyID) > 0 {
			cfg.Credentials = credentials.NewStaticCredentialsFromCreds(
				sharedCfg.Creds,
			)
		} else {
			// Fallback to default credentials provider, include mock errors
			// for the credential chain so user can identify why credentials
			// failed to be retrieved.
			cfg.Credentials = credentials.NewCredentials(&credentials.ChainProvider{
				VerboseErrors: aws.BoolValue(cfg.CredentialsChainVerboseErrors),
				Providers: []credentials.Provider{
					&credProviderError{Err: awserr.New("EnvAccessKeyNotFound", "failed to find credentials in the environment.", nil)},
					&credProviderError{Err: awserr.New("SharedCredsLoad", fmt.Sprintf("failed to load profile, %s.", envCfg.Profile), nil)},
					defaults.RemoteCredProvider(*cfg, handlers),
				},
			})
		}
	}
}

func (r *run) initIamClient() *iam.IAM {
	var awsconf aws.Config
	if r.c.AccessKey != "" && r.c.SecretKey != "" {
		awscreds := awscred.NewStaticCredentials(r.c.AccessKey, r.c.SecretKey, "")
		awsconf.Credentials = awscreds
	}
	return iam.New(session.New(), &awsconf)
}

func (factory *ecrFactory) newClient(region, endpointOverride string) ECRSDK {
	var ecrConfig aws.Config
	ecrConfig.Region = &region
	ecrConfig.HTTPClient = factory.httpClient
	if endpointOverride != "" {
		ecrConfig.Endpoint = &endpointOverride
	}
	return ecrapi.New(&ecrConfig)
}

func getService(debug bool) *route53.Route53 {
	config := aws.Config{}
	// ensures throttled requests are retried
	config.MaxRetries = aws.Int(100)
	if debug {
		config.LogLevel = aws.LogLevel(aws.LogDebug)
	}
	return route53.New(&config)
}

// newSubmitStateChangeClient returns a client intended to be used for
// Submit*StateChange APIs which has the behavior of retrying the call on
// retriable errors for an extended period of time (roughly 24 hours).
func newSubmitStateChangeClient(awsConfig *aws.Config) *ecs.ECS {
	sscConfig := awsConfig.Copy()
	sscConfig.MaxRetries = submitStateChangeMaxDelayRetries
	client := ecs.New(&sscConfig)
	client.Handlers.AfterRetry.Clear()
	client.Handlers.AfterRetry.PushBack(
		extendedRetryMaxDelayHandlerFactory(submitStateChangeExtraRetries))
	client.DefaultMaxRetries = submitStateChangeMaxDelayRetries
	return client
}

func (factory *ecrFactory) newClient(region, endpointOverride string) ECRClient {
	var ecrConfig aws.Config
	ecrConfig.Region = &region
	ecrConfig.HTTPClient = factory.httpClient
	if endpointOverride != "" {
		ecrConfig.Endpoint = &endpointOverride
	}
	sdkClient := ecrapi.New(session.New(&ecrConfig))
	tokenCache := async.NewLRUCache(tokenCacheSize, tokenCacheTTL)
	return NewECRClient(sdkClient, tokenCache)
}

func newMultiRegion(conf *awsclient.Config, regions []string) *multiRegion {
	m := &multiRegion{
		regions: make(map[string]*ec2.EC2, 0),
	}

	for _, region := range regions {
		m.regions[region] = ec2.New(conf.Merge(&awsclient.Config{Region: region}))
	}

	return m
}

func MakeS3Backend(bucket string, prefix string, opts *ConnectOptions) ArchiveBackend {
	cfg := aws.Config{}
	if opts != nil && opts.S3Region != "" {
		cfg.Region = aws.String(opts.S3Region)
	}
	sess := session.New(&cfg)
	return &S3ArchiveBackend{
		svc:    s3.New(sess),
		bucket: bucket,
		prefix: prefix,
	}
}

func getService(debug bool, profile string) *route53.Route53 {
	config := aws.Config{}
	if profile != "" {
		config.Credentials = credentials.NewSharedCredentials("", profile)
	}
	// ensures throttled requests are retried
	config.MaxRetries = aws.Int(100)
	if debug {
		config.LogLevel = aws.LogLevel(aws.LogDebug)
	}
	return route53.New(session.New(), &config)
}

func mergeConfigSrcs(cfg, userCfg *aws.Config, envCfg envConfig, sharedCfg sharedConfig, handlers request.Handlers) {
	// Merge in user provided configuration
	cfg.MergeIn(userCfg)

	// Region if not already set by user
	if len(aws.StringValue(cfg.Region)) == 0 {
		if len(envCfg.Region) > 0 {
			cfg.WithRegion(envCfg.Region)
		} else if envCfg.EnableSharedConfig && len(sharedCfg.Region) > 0 {
			cfg.WithRegion(sharedCfg.Region)
		}
	}

	// Configure credentials if not already set
	if cfg.Credentials == credentials.AnonymousCredentials && userCfg.Credentials == nil {
		if len(envCfg.Creds.AccessKeyID) > 0 {
			cfg.Credentials = credentials.NewStaticCredentialsFromCreds(
				envCfg.Creds,
			)
		} else if envCfg.EnableSharedConfig && len(sharedCfg.AssumeRole.RoleARN) > 0 && sharedCfg.AssumeRoleSource != nil {
			cfgCp := *cfg
			cfgCp.Credentials = credentials.NewStaticCredentialsFromCreds(
				sharedCfg.AssumeRoleSource.Creds,
			)
			cfg.Credentials = stscreds.NewCredentials(
				&Session{
					Config:   &cfgCp,
					Handlers: handlers.Copy(),
				},
				sharedCfg.AssumeRole.RoleARN,
				func(opt *stscreds.AssumeRoleProvider) {
					opt.RoleSessionName = sharedCfg.AssumeRole.RoleSessionName

					if len(sharedCfg.AssumeRole.ExternalID) > 0 {
						opt.ExternalID = aws.String(sharedCfg.AssumeRole.ExternalID)
					}

					// MFA not supported
				},
			)
		} else if len(sharedCfg.Creds.AccessKeyID) > 0 {
			cfg.Credentials = credentials.NewStaticCredentialsFromCreds(
				sharedCfg.Creds,
			)
		} else {
			// Fallback to default credentials provider
			cfg.Credentials = credentials.NewCredentials(
				defaults.RemoteCredProvider(*cfg, handlers),
			)
		}
	}
}

func getConfig(c *cli.Context) *aws.Config {
	debug := c.Bool("debug")
	profile := c.String("profile")
	config := aws.Config{}
	if profile != "" {
		config.Credentials = credentials.NewSharedCredentials("", profile)
	}
	// ensures throttled requests are retried
	config.MaxRetries = aws.Int(100)
	if debug {
		config.LogLevel = aws.LogLevel(aws.LogDebug)
	}
	return &config
}

func (t *Test) invokeLambdas(awsConfig *aws.Config, sqsURL string) {
	lambdas := numberOfLambdas(t.config.Concurrency, len(t.config.Regions))

	for i := 0; i < lambdas; i++ {
		region := t.config.Regions[i%len(t.config.Regions)]
		requests, requestsRemainder := divide(t.config.TotalRequests, lambdas)
		concurrency, _ := divide(t.config.Concurrency, lambdas)

		if requestsRemainder > 0 && i == lambdas-1 {
			requests += requestsRemainder
		}

		c := t.config
		args := []string{
			"-u",
			fmt.Sprintf("%s", c.URL),
			"-c",
			fmt.Sprintf("%s", strconv.Itoa(int(concurrency))),
			"-n",
			fmt.Sprintf("%s", strconv.Itoa(int(requests))),
			"-s",
			fmt.Sprintf("%s", sqsURL),
			"-q",
			fmt.Sprintf("%s", c.Regions[0]),
			"-t",
			fmt.Sprintf("%s", c.RequestTimeout.String()),
			"-f",
			fmt.Sprintf("%s", reportingFrequency(lambdas).String()),
			"-r",
			fmt.Sprintf("%s", region),
			"-m",
			fmt.Sprintf("%s", c.Method),
			"-b",
			fmt.Sprintf("%s", c.Body),
		}

		for _, v := range t.config.Headers {
			args = append(args, "-H", fmt.Sprintf("%s", v))
		}

		invokeargs := invokeArgs{
			File: "./goad-lambda",
			Args: args,
		}

		config := awsConfig.WithRegion(region)
		go t.invokeLambda(config, invokeargs)
	}
}

// New creates an instance of the DynamoDB struct for us.
func New(access string, secret string, region string, tableName string) (DynamoDB, error) {
	creds := credentials.NewStaticCredentials(access, secret, "")

	cfg := aws.Config{
		Region: aws.String(region),
	}

	cfg.WithCredentials(creds)

	ddb := DynamoDB{
		Table: tableName,
		Conn:  dynamodb.New(session.New(), &cfg),
	}

	return ddb, nil
}

// getSessionWithConfig grabs the region and appends to the current config
func getSessionWithConfig(config *aws.Config) (*session.Session, error) {
	region, err := getAWSRegion()

	if profileName != "" {
		fmt.Println("Profile: ", *profile)
	} else {
		fmt.Println("Profile: default")
	}

	if region != "" {
		fmt.Println("Region: ", region)
		config = config.WithRegion(region)
	}

	fmt.Println()
	return session.New(config), err
}

func NewECSClient(credentialProvider *credentials.Credentials, config *config.Config, httpClient *http.Client, ec2MetadataClient ec2.EC2MetadataClient) ECSClient {
	var ecsConfig aws.Config
	ecsConfig.Credentials = credentialProvider
	ecsConfig.Region = &config.AWSRegion
	ecsConfig.HTTPClient = httpClient
	if config.APIEndpoint != "" {
		ecsConfig.Endpoint = &config.APIEndpoint
	}
	standardClient := ecs.New(&ecsConfig)
	submitStateChangeClient := newSubmitStateChangeClient(&ecsConfig)
	return &ApiECSClient{
		credentialProvider:      credentialProvider,
		config:                  config,
		standardClient:          standardClient,
		submitStateChangeClient: submitStateChangeClient,
		ec2metadata:             ec2MetadataClient,
	}
}

func setOptionalEndpoint(cfg *aws.Config) string {
	endpoint := os.Getenv("AWS_METADATA_URL")
	if endpoint != "" {
		log.Printf("[INFO] Setting custom metadata endpoint: %q", endpoint)
		cfg.Endpoint = aws.String(endpoint)
		return endpoint
	}
	return ""
}

func NewGoofys(bucket string, awsConfig *aws.Config, flags *FlagStorage) *Goofys {
	// Set up the basic struct.
	fs := &Goofys{
		bucket: bucket,
		flags:  flags,
		umask:  0122,
	}

	if flags.DebugS3 {
		awsConfig.LogLevel = aws.LogLevel(aws.LogDebug | aws.LogDebugWithRequestErrors)
		s3Log.Level = logrus.DebugLevel
	}

	fs.awsConfig = awsConfig
	fs.sess = session.New(awsConfig)
	fs.s3 = fs.newS3()

	err := fs.detectBucketLocation()
	if err != nil {
		return nil
	}

	now := time.Now()
	fs.rootAttrs = fuseops.InodeAttributes{
		Size:   4096,
		Nlink:  2,
		Mode:   flags.DirMode | os.ModeDir,
		Atime:  now,
		Mtime:  now,
		Ctime:  now,
		Crtime: now,
		Uid:    fs.flags.Uid,
		Gid:    fs.flags.Gid,
	}

	fs.bufferPool = BufferPool{}.Init()

	fs.nextInodeID = fuseops.RootInodeID + 1
	fs.inodes = make(map[fuseops.InodeID]*Inode)
	root := NewInode(aws.String(""), aws.String(""), flags)
	root.Id = fuseops.RootInodeID
	root.Attributes = &fs.rootAttrs

	fs.inodes[fuseops.RootInodeID] = root
	fs.inodesCache = make(map[string]*Inode)

	fs.nextHandleID = 1
	fs.dirHandles = make(map[fuseops.HandleID]*DirHandle)

	fs.fileHandles = make(map[fuseops.HandleID]*FileHandle)

	return fs
}