// findLatestSnap finds the source snapshot to copy
func (c *config) findLatestSnap() error {
	cli := rds.New(session.New(), &aws.Config{Region: aws.String(c.src)})
	c.debug(fmt.Sprintf("Searching for snapshots for: %s", c.dbid))
	q := rds.DescribeDBSnapshotsInput{}
	q.DBInstanceIdentifier = aws.String(c.dbid)
	resp, err := cli.DescribeDBSnapshots(&q)
	if err != nil {
		return err
	}
	newest := time.Unix(0, 0)
	newestId := ""
	if len(resp.DBSnapshots) < 1 {
		return fmt.Errorf("No snapshots found")
	}
	c.debug(fmt.Sprintf("Found %d snapshots for: %s", len(resp.DBSnapshots), c.dbid))
	for _, r := range resp.DBSnapshots {
		if r.SnapshotCreateTime.After(newest) {
			newestId = *r.DBSnapshotIdentifier
			newest = *r.SnapshotCreateTime
		}
	}
	if len(newestId) < 1 {
		return fmt.Errorf("No usable snapshot found")
	}
	c.arn = fmt.Sprintf("arn:aws:rds:%s:%s:snapshot:%s", c.src, c.awsAcctId, newestId)
	c.debug(fmt.Sprintf("Found latest snapshot: %s: %s", newestId, newest.String()))
	return nil
}

// checkSnapCopied returns true if the source snapshot has already been copied to the destination region
func (c *config) checkSnapCopied() bool {
	cli := rds.New(session.New(), &aws.Config{Region: aws.String(c.dst)})
	q := rds.DescribeDBSnapshotsInput{}
	q.DBInstanceIdentifier = aws.String(c.dbid)
	resp, err := cli.DescribeDBSnapshots(&q)
	if err != nil {
		return false
	}
	for _, s := range resp.DBSnapshots {
		q := rds.ListTagsForResourceInput{ResourceName: aws.String(fmt.Sprintf("arn:aws:rds:%s:%s:snapshot:%s", c.dst, c.awsAcctId, *s.DBSnapshotIdentifier))}
		tags, err := cli.ListTagsForResource(&q)
		if err != nil {
			continue
		}
		managedByUs := false
		matchedSource := false
		for _, t := range tags.TagList {
			if *t.Key == "managedby" && *t.Value == "rdsbackup" {
				managedByUs = true
			} else if *t.Key == "sourcearn" && *t.Value == c.arn {
				matchedSource = true
			}
		}
		if managedByUs && matchedSource {
			return true
		}
	}
	return false
}

func main() {
	flag.Parse()

	config, err := LoadConfig(configFilePath)
	if err != nil {
		log.Fatalf("Error loading config file: %s", err)
	}

	logger := buildLogger(config.LogLevel)

	awsConfig := aws.NewConfig().WithRegion(config.RDSConfig.Region)
	awsSession := session.New(awsConfig)

	iamsvc := iam.New(awsSession)
	rdssvc := rds.New(awsSession)
	dbInstance := awsrds.NewRDSDBInstance(config.RDSConfig.Region, iamsvc, rdssvc, logger)
	dbCluster := awsrds.NewRDSDBCluster(config.RDSConfig.Region, iamsvc, rdssvc, logger)

	sqlProvider := sqlengine.NewProviderService(logger)

	serviceBroker := rdsbroker.New(config.RDSConfig, dbInstance, dbCluster, sqlProvider, logger)

	credentials := brokerapi.BrokerCredentials{
		Username: config.Username,
		Password: config.Password,
	}

	brokerAPI := brokerapi.New(serviceBroker, logger, credentials)
	http.Handle("/", brokerAPI)

	fmt.Println("RDS Service Broker started on port " + port + "...")
	http.ListenAndServe(":"+port, nil)
}

// return "httptest.Server" need call close !!
// and Command into OutConfig.Root need call remove !!
func getTestClient(code int, body string) (*httptest.Server, *Command) {
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.WriteHeader(code)
		w.Header().Set("Content-Type", "application/xml")
		fmt.Fprintln(w, body)
	}))

	transport := &http.Transport{
		Proxy: func(req *http.Request) (*url.URL, error) {
			return url.Parse(server.URL)
		},
	}

	httpClient := &http.Client{Transport: transport}

	// Override endpoints
	testRegion := "rds-try-test-1"
	awsConf := aws.NewConfig()
	awsConf = awsConf.WithCredentials(credentials.NewStaticCredentials("awsAccesskey1", "awsSecretKey2", ""))
	awsConf = awsConf.WithRegion(testRegion)
	awsConf = awsConf.WithEndpoint(server.URL)
	awsConf = awsConf.WithHTTPClient(httpClient)

	awsRds := rds.New(awsConf)

	testName := utils.GetAppName() + "-test"
	tempDir, _ := ioutil.TempDir("", testName)
	defer os.RemoveAll(tempDir)

	out := config.OutConfig{
		Root: tempDir,
		File: true,
		Bom:  true,
	}

	rds := config.RDSConfig{
		MultiAz: false,
		DBId:    utils.GetFormatedDBDisplayName(testName),
		Region:  testRegion,
		User:    "test-admin",
		Pass:    "pass-pass",
		Type:    "db.m3.medium",
	}

	cmdTest := &Command{
		OutConfig: out,
		RDSConfig: rds,
		RDSClient: awsRds,
		ARNPrefix: "arn:aws:rds:" + testRegion + ":" + "123456789" + ":",
	}

	return server, cmdTest
}

func NewRDSClient(region string) (*RDSClient, error) {
	sess, err := session.NewSession(&aws.Config{Region: aws.String(region)})
	if err != nil {
		fmt.Println("Failed to create AWS session,", err)
		return nil, err
	}

	rdssvc := rds.New(sess)
	return &RDSClient{
		region: region,
		rdssvc: rdssvc,
	}, nil
}

func getCommandStruct(conf *config.Config) (*command.Command, int) {
	// aws Credentials
	creds, err := conf.GetAWSCreds()
	if err != nil {
		log.Errorf("%s", err.Error())
		return nil, 1
	}
	// aws config init
	awsConfig := aws.NewConfig()
	awsConfig = awsConfig.WithCredentials(creds)
	awsConfig = awsConfig.WithRegion(conf.Rds[nameFlag].Region)

	// new iam
	awsIam := iam.New(awsConfig)

	// IAM info
	iamUsers, err := awsIam.ListUsers(&iam.ListUsersInput{})
	if err != nil {
		log.Errorf("%s", err.Error())
		return nil, 1
	}
	if len(iamUsers.Users) <= 0 {
		log.Errorf("iam user not found")
		return nil, 1
	}

	// edit IAM ARN
	// arn:aws:iam::<account>:user/<username> to arn:aws:rds:<region>:<account>:
	// see also
	// Tagging Amazon RDS Resources - Amazon Relational Database Service
	// http://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_Tagging.html#USER_Tagging.ARN
	iamSplit := strings.SplitAfter(*iamUsers.Users[0].Arn, "::")
	iamAccount := iamSplit[len(iamSplit)-1]
	iamSplit = strings.Split(iamAccount, ":")
	iamAccount = iamSplit[0]

	// new rds
	awsRds := rds.New(awsConfig)

	commandStruct := &command.Command{
		OutConfig: conf.Out,
		RDSConfig: conf.Rds[nameFlag],
		RDSClient: awsRds,
		ARNPrefix: "arn:aws:rds:" + conf.Rds[nameFlag].Region + ":" + iamAccount + ":",
	}
	log.Debugf("Command: %+v", commandStruct)

	return commandStruct, 0
}

// cleanupSnaps
func (c *config) cleanupSnaps() error {
	if c.purge <= 0 {
		return nil
	}
	c.debug(fmt.Sprintf("Cleaning up old snapshots in dest region %s...", c.dst))
	cli := rds.New(session.New(), &aws.Config{Region: aws.String(c.dst)})
	q := rds.DescribeDBSnapshotsInput{}
	q.DBInstanceIdentifier = aws.String(c.dbid)
	resp, err := cli.DescribeDBSnapshots(&q)
	if err != nil {
		return err
	}
	snaps := map[int64]string{}
	keys := int64arr{}
	for _, s := range resp.DBSnapshots {
		q := rds.ListTagsForResourceInput{ResourceName: aws.String(fmt.Sprintf("arn:aws:rds:%s:%s:snapshot:%s", c.dst, c.awsAcctId, *s.DBSnapshotIdentifier))}
		tags, err := cli.ListTagsForResource(&q)
		if err != nil {
			continue
		}
		for _, t := range tags.TagList {
			if *t.Key == "managedby" && *t.Value == "rdsbackup" {
				if s.SnapshotCreateTime.Unix() > 0 {
					snaps[s.SnapshotCreateTime.Unix()] = *s.DBSnapshotIdentifier
					keys = append(keys, s.SnapshotCreateTime.Unix())
				}
			}
		}
	}
	if len(snaps) <= c.purge {
		c.debug(fmt.Sprintf("Found %d snapshots. Purge flag is %d, so nothing will be purged.", len(snaps), c.purge))
	} else {
		c.debug(fmt.Sprintf("Found %d snapshots. Purge flag is %d, so the oldest %d snapshots will be purged.", len(snaps), c.purge, len(snaps)-c.purge))
		sort.Sort(keys)
		for i := 0; i < len(snaps)-c.purge; i++ {
			c.debug(fmt.Sprintf("Purging snapshot %s.", snaps[keys[i]]))
			q := rds.DeleteDBSnapshotInput{DBSnapshotIdentifier: aws.String(snaps[keys[i]])}
			resp, err := cli.DeleteDBSnapshot(&q)
			if err != nil {
				return err
			}
			if *resp.DBSnapshot.Status != "deleted" {
				c.debug(fmt.Sprintf("Warning: snapshot was not deleted successfully: %s", snaps[keys[i]]))
			}
		}
		c.debug("Done purging shapshots.")
	}
	return nil
}

func (tp *TypeMetrics) rdsInstanceTypes(regionName string, out chan string) {

	service := rds.New(tp.newSession(&regionName))
	result, err := service.DescribeDBInstances(&rds.DescribeDBInstancesInput{})

	if err != nil {
		log.Printf("%s\n", err)
		return
	}

	instTypes := make(map[string]int, 0)
	for _, instance := range result.DBInstances {
		instTypes[*instance.DBInstanceClass]++
	}

	for instType, instCount := range instTypes {
		out <- fmt.Sprintf("aws.%s.instance_types.%s %d %d", regionName, instType, instCount, int32(time.Now().Unix()))
	}
}

func Fetch(dbid, path *string) (io.ReadCloser, error) {
	svc := rds.New(&aws.Config{
		Region: aws.String("ap-northeast-1"),
		//LogLevel: aws.LogLevel(aws.LogDebugWithHTTPBody),
	})
	// RDSのHandlerはQuery APIになっているのでをREST APIに変更
	svc.Handlers.Build.Clear()
	svc.Handlers.Build.PushBack(rest.Build)
	svc.Handlers.Unmarshal.Clear()
	svc.Handlers.Unmarshal.PushBack(rest.Unmarshal)

	out, err := restrds.DownloadCompleteDBLogFile(svc, &restrds.DownloadCompleteDBLogFileInput{
		DBInstanceIdentifier: dbid,
		LogFileName:          path,
	})
	if err != nil {
		return nil, err
	}
	return out.Body, nil
}

func (s *AWSSession) fetchRDSList() []*AWSElement {
	svc := rds.New(s.Sess)

	params := &rds.DescribeDBInstancesInput{}
	resp, err := svc.DescribeDBInstances(params)

	if err != nil {
		fmt.Println("fetchLoadBalancerList: ", err.Error())
		return []*AWSElement{}
	}

	//fmt.Println(resp)
	var rdss []*AWSElement
	for _, rds := range resp.DBInstances {
		rdss = append(rdss, &AWSElement{
			Name:    *rds.DBInstanceIdentifier,
			DNSName: *rds.Endpoint.Address,
		})
	}
	return rdss
}

// waitForCopy waits for the RDS snapshot copy to finish
func (c *config) waitForCopy() error {
	c.debug(fmt.Sprintf("Waiting for copy %s...", c.copyId))
	cli := rds.New(session.New(), &aws.Config{Region: aws.String(c.dst)})
	q := rds.DescribeDBSnapshotsInput{}
	q.DBSnapshotIdentifier = aws.String(c.copyId)
	for {
		resp, err := cli.DescribeDBSnapshots(&q)
		if err != nil {
			return err
		}
		if len(resp.DBSnapshots) != 1 {
			return fmt.Errorf("New snapshot missing!")
		}
		s := resp.DBSnapshots[0]
		if *s.Status != "creating" {
			break
		}
		c.debug(fmt.Sprintf("Waiting %s (%d%% complete)", *s.Status, *s.PercentProgress))
		time.Sleep(10 * time.Second)
	}
	return nil
}

// copySnap starts the RDS snapshot copy
func (c *config) copySnap() error {
	cli := rds.New(session.New(), &aws.Config{Region: aws.String(c.dst)})
	t := time.Now()
	c.copyId = fmt.Sprintf("%s-%s", c.dbid, t.Format("2006-01-02at15-04MST"))
	m := rds.CopyDBSnapshotInput{
		SourceDBSnapshotIdentifier: aws.String(c.arn),
		Tags: []*rds.Tag{
			&rds.Tag{Key: aws.String("time"), Value: aws.String(t.Format("2006-01-02 15:04:05 -0700"))},
			&rds.Tag{Key: aws.String("timestamp"), Value: aws.String(fmt.Sprintf("%d", t.Unix()))},
			&rds.Tag{Key: aws.String("source"), Value: aws.String(c.src)},
			&rds.Tag{Key: aws.String("sourceid"), Value: aws.String(c.dbid)},
			&rds.Tag{Key: aws.String("sourcearn"), Value: aws.String(c.arn)},
			&rds.Tag{Key: aws.String("managedby"), Value: aws.String("rdsbackup")},
		},
		TargetDBSnapshotIdentifier: aws.String(c.copyId),
	}
	resp, err := cli.CopyDBSnapshot(&m)
	if err != nil {
		return err
	} else if *resp.DBSnapshot.Status != "creating" || *resp.DBSnapshot.Status != "pending" {
		return fmt.Errorf("Error creating snapshot - unexpected status: %s", *resp.DBSnapshot.Status)
	}
	return nil
}

func getJsonSawsInfo(config *Config) []byte {
	svc := ec2.New(session.New())
	rdsc := rds.New(session.New())
	instancelist := make([]*ec2.Instance, 0)
	for i := range config.EC2 {
		//fmt.Println(config.EC2[i].Name)
		instances := getInstancesByName(svc, config.EC2[i].Name)
		for k := range instances {
			instancelist = append(instancelist, instances[k])
		}
	}

	rdslist := make([]*rds.DBInstance, 0)
	for i := range config.RDS {
		//fmt.Println(config.RDS[i].DBInstanceIdentifier)
		dbinstance, err := getRDSInstanceById(rdsc, &config.RDS[i].DBInstanceIdentifier)
		if err != nil {
			//fmt.Println("Failed to find db instance", config.RDS[i].DBInstanceIdentifier)
		} else {
			rdslist = append(rdslist, dbinstance)
		}
	}

	sawsinfo := SawsInfo{}
	sawsinfo.EC2 = instancelist
	sawsinfo.RDS = rdslist

	marsh, err := json.Marshal(&sawsinfo)
	if err != nil {
		fmt.Println("Failed to unmarshal", err)
		panic(err)
	}

	return marsh

}

// Client configures and returns a fully initialized AWSClient
func (c *Config) Client() (interface{}, error) {
	var client AWSClient

	// Get the auth and region. This can fail if keys/regions were not
	// specified and we're attempting to use the environment.
	var errs []error

	log.Println("[INFO] Building AWS region structure")
	err := c.ValidateRegion()
	if err != nil {
		errs = append(errs, err)
	}

	if len(errs) == 0 {
		// store AWS region in client struct, for region specific operations such as
		// bucket storage in S3
		client.region = c.Region

		log.Println("[INFO] Building AWS auth structure")
		// We fetched all credential sources in Provider. If they are
		// available, they'll already be in c. See Provider definition.
		creds := credentials.NewStaticCredentials(c.AccessKey, c.SecretKey, c.Token)
		awsConfig := &aws.Config{
			Credentials: creds,
			Region:      aws.String(c.Region),
			MaxRetries:  aws.Int(c.MaxRetries),
		}

		log.Println("[INFO] Initializing IAM Connection")
		client.iamconn = iam.New(awsConfig)

		err := c.ValidateCredentials(client.iamconn)
		if err != nil {
			errs = append(errs, err)
		}

		awsDynamoDBConfig := &aws.Config{
			Credentials: creds,
			Region:      aws.String(c.Region),
			MaxRetries:  aws.Int(c.MaxRetries),
			Endpoint:    aws.String(c.DynamoDBEndpoint),
		}

		log.Println("[INFO] Initializing DynamoDB connection")
		client.dynamodbconn = dynamodb.New(awsDynamoDBConfig)

		log.Println("[INFO] Initializing ELB connection")
		client.elbconn = elb.New(awsConfig)

		log.Println("[INFO] Initializing S3 connection")
		client.s3conn = s3.New(awsConfig)

		log.Println("[INFO] Initializing SQS connection")
		client.sqsconn = sqs.New(awsConfig)

		log.Println("[INFO] Initializing SNS connection")
		client.snsconn = sns.New(awsConfig)

		log.Println("[INFO] Initializing RDS Connection")
		client.rdsconn = rds.New(awsConfig)

		log.Println("[INFO] Initializing Kinesis Connection")
		client.kinesisconn = kinesis.New(awsConfig)

		authErr := c.ValidateAccountId(client.iamconn)
		if authErr != nil {
			errs = append(errs, authErr)
		}

		log.Println("[INFO] Initializing AutoScaling connection")
		client.autoscalingconn = autoscaling.New(awsConfig)

		log.Println("[INFO] Initializing EC2 Connection")
		client.ec2conn = ec2.New(awsConfig)

		log.Println("[INFO] Initializing ECS Connection")
		client.ecsconn = ecs.New(awsConfig)

		log.Println("[INFO] Initializing EFS Connection")
		client.efsconn = efs.New(awsConfig)

		// aws-sdk-go uses v4 for signing requests, which requires all global
		// endpoints to use 'us-east-1'.
		// See http://docs.aws.amazon.com/general/latest/gr/sigv4_changes.html
		log.Println("[INFO] Initializing Route 53 connection")
		client.r53conn = route53.New(&aws.Config{
			Credentials: creds,
			Region:      aws.String("us-east-1"),
			MaxRetries:  aws.Int(c.MaxRetries),
		})

		log.Println("[INFO] Initializing Elasticache Connection")
		client.elasticacheconn = elasticache.New(awsConfig)

		log.Println("[INFO] Initializing Lambda Connection")
		client.lambdaconn = lambda.New(awsConfig)

		log.Println("[INFO] Initializing CloudWatch SDK connection")
		client.cloudwatchconn = cloudwatch.New(awsConfig)
//.........这里部分代码省略.........

// Client configures and returns a fully initialized AWSClient
func (c *Config) Client() (interface{}, error) {
	// Get the auth and region. This can fail if keys/regions were not
	// specified and we're attempting to use the environment.
	log.Println("[INFO] Building AWS region structure")
	err := c.ValidateRegion()
	if err != nil {
		return nil, err
	}

	var client AWSClient
	// store AWS region in client struct, for region specific operations such as
	// bucket storage in S3
	client.region = c.Region

	log.Println("[INFO] Building AWS auth structure")
	creds, err := GetCredentials(c)
	if err != nil {
		return nil, err
	}
	// Call Get to check for credential provider. If nothing found, we'll get an
	// error, and we can present it nicely to the user
	cp, err := creds.Get()
	if err != nil {
		if awsErr, ok := err.(awserr.Error); ok && awsErr.Code() == "NoCredentialProviders" {
			return nil, errors.New(`No valid credential sources found for AWS Provider.
  Please see https://terraform.io/docs/providers/aws/index.html for more information on
  providing credentials for the AWS Provider`)
		}

		return nil, fmt.Errorf("Error loading credentials for AWS Provider: %s", err)
	}

	log.Printf("[INFO] AWS Auth provider used: %q", cp.ProviderName)

	awsConfig := &aws.Config{
		Credentials:      creds,
		Region:           aws.String(c.Region),
		MaxRetries:       aws.Int(c.MaxRetries),
		HTTPClient:       cleanhttp.DefaultClient(),
		S3ForcePathStyle: aws.Bool(c.S3ForcePathStyle),
	}

	if logging.IsDebugOrHigher() {
		awsConfig.LogLevel = aws.LogLevel(aws.LogDebugWithHTTPBody)
		awsConfig.Logger = awsLogger{}
	}

	if c.Insecure {
		transport := awsConfig.HTTPClient.Transport.(*http.Transport)
		transport.TLSClientConfig = &tls.Config{
			InsecureSkipVerify: true,
		}
	}

	// Set up base session
	sess, err := session.NewSession(awsConfig)
	if err != nil {
		return nil, errwrap.Wrapf("Error creating AWS session: {{err}}", err)
	}

	// Removes the SDK Version handler, so we only have the provider User-Agent
	// Ex: "User-Agent: APN/1.0 HashiCorp/1.0 Terraform/0.7.9-dev"
	sess.Handlers.Build.Remove(request.NamedHandler{Name: "core.SDKVersionUserAgentHandler"})
	sess.Handlers.Build.PushFrontNamed(addTerraformVersionToUserAgent)

	if extraDebug := os.Getenv("TERRAFORM_AWS_AUTHFAILURE_DEBUG"); extraDebug != "" {
		sess.Handlers.UnmarshalError.PushFrontNamed(debugAuthFailure)
	}

	// Some services exist only in us-east-1, e.g. because they manage
	// resources that can span across multiple regions, or because
	// signature format v4 requires region to be us-east-1 for global
	// endpoints:
	// http://docs.aws.amazon.com/general/latest/gr/sigv4_changes.html
	usEast1Sess := sess.Copy(&aws.Config{Region: aws.String("us-east-1")})

	// Some services have user-configurable endpoints
	awsEc2Sess := sess.Copy(&aws.Config{Endpoint: aws.String(c.Ec2Endpoint)})
	awsElbSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.ElbEndpoint)})
	awsIamSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.IamEndpoint)})
	awsS3Sess := sess.Copy(&aws.Config{Endpoint: aws.String(c.S3Endpoint)})
	dynamoSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.DynamoDBEndpoint)})
	kinesisSess := sess.Copy(&aws.Config{Endpoint: aws.String(c.KinesisEndpoint)})

	// These two services need to be set up early so we can check on AccountID
	client.iamconn = iam.New(awsIamSess)
	client.stsconn = sts.New(sess)

	if !c.SkipCredsValidation {
		err = c.ValidateCredentials(client.stsconn)
		if err != nil {
			return nil, err
		}
	}

	if !c.SkipRequestingAccountId {
		partition, accountId, err := GetAccountInfo(client.iamconn, client.stsconn, cp.ProviderName)
		if err == nil {
			client.partition = partition
//.........这里部分代码省略.........

func RDS() *rds.RDS {
	return rds.New(awsConfig())
}

		testSink *lagertest.TestSink
		logger   lager.Logger

		rdsDBCluster DBCluster
	)

	BeforeEach(func() {
		region = "rds-region"
		dbClusterIdentifier = "cf-cluster-id"
	})

	JustBeforeEach(func() {
		awsSession = session.New(nil)

		iamsvc = iam.New(awsSession)
		rdssvc = rds.New(awsSession)

		logger = lager.NewLogger("rdsdbcluster_test")
		testSink = lagertest.NewTestSink()
		logger.RegisterSink(testSink)

		rdsDBCluster = NewRDSDBCluster(region, iamsvc, rdssvc, logger)
	})

	var _ = Describe("Describe", func() {
		var (
			properDBClusterDetails DBClusterDetails

			describeDBClusters []*rds.DBCluster
			describeDBCluster  *rds.DBCluster

func newClient() *rds.RDS {
	cfg := loadConfig()
	return rds.New(session.New(cfg))
}

func setupRDS(c *cli.Context) *rds.RDS {
	region := c.GlobalString("region")
	maxRetries := c.GlobalInt("max-retries")
	cfg := aws.NewConfig().WithRegion(region).WithMaxRetries(maxRetries)
	return rds.New(session.New(), cfg)
}

func init() {
	gucumber.Before("@rds", func() {
		gucumber.World["client"] = rds.New(smoke.Session)
	})
}