// New creates a new instance of the handlers merging in the provided configs
// on top of the SDK's default configurations. Once the Session is created it
// can be mutated to modify the Config or Handlers. The Session is safe to be
// read concurrently, but it should not be written to concurrently.
//
// If the AWS_SDK_LOAD_CONFIG environment is set to a truthy value, the New
// method could now encounter an error when loading the configuration. When
// The environment variable is set, and an error occurs, New will return a
// session that will fail all requests reporting the error that occured while
// loading the session. Use NewSession to get the error when creating the
// session.
//
// If the AWS_SDK_LOAD_CONFIG environment variable is set to a truthy value
// the shared config file (~/.aws/config) will also be loaded, in addition to
// the shared credentials file (~/.aws/config). Values set in both the
// shared config, and shared credentials will be taken from the shared
// credentials file.
//
// Deprecated: Use NewSession functiions to create sessions instead. NewSession
// has the same functionality as New except an error can be returned when the
// func is called instead of waiting to receive an error until a request is made.
func New(cfgs ...*aws.Config) *Session {
	// load initial config from environment
	envCfg := loadEnvConfig()

	if envCfg.EnableSharedConfig {
		s, err := newSession(envCfg, cfgs...)
		if err != nil {
			// Old session.New expected all errors to be discovered when
			// a request is made, and would report the errors then. This
			// needs to be replicated if an error occurs while creating
			// the session.
			msg := "failed to create session with AWS_SDK_LOAD_CONFIG enabled. " +
				"Use session.NewSession to handle errors occuring during session creation."

			// Session creation failed, need to report the error and prevent
			// any requests from succeeding.
			s = &Session{Config: defaults.Config()}
			s.Config.MergeIn(cfgs...)
			s.Config.Logger.Log("ERROR:", msg, "Error:", err)
			s.Handlers.Validate.PushBack(func(r *request.Request) {
				r.Error = err
			})
		}
		return s
	}

	return oldNewSession(cfgs...)
}

func oldNewSession(cfgs ...*aws.Config) *Session {
	cfg := defaults.Config()
	handlers := defaults.Handlers()

	// Apply the passed in configs so the configuration can be applied to the
	// default credential chain
	cfg.MergeIn(cfgs...)
	if cfg.EndpointResolver == nil {
		// An endpoint resolver is required for a session to be able to provide
		// endpoints for service client configurations.
		cfg.EndpointResolver = endpoints.DefaultResolver()
	}
	cfg.Credentials = defaults.CredChain(cfg, handlers)

	// Reapply any passed in configs to override credentials if set
	cfg.MergeIn(cfgs...)

	s := &Session{
		Config:   cfg,
		Handlers: handlers,
	}

	initHandlers(s)

	return s
}

func newSession(envCfg envConfig, cfgs ...*aws.Config) (*Session, error) {
	cfg := defaults.Config()
	handlers := defaults.Handlers()

	// Get a merged version of the user provided config to determine if
	// credentials were.
	userCfg := &aws.Config{}
	userCfg.MergeIn(cfgs...)

	// Order config files will be loaded in with later files overwriting
	// previous config file values.
	cfgFiles := []string{envCfg.SharedConfigFile, envCfg.SharedCredentialsFile}
	if !envCfg.EnableSharedConfig {
		// The shared config file (~/.aws/config) is only loaded if instructed
		// to load via the envConfig.EnableSharedConfig (AWS_SDK_LOAD_CONFIG).
		cfgFiles = cfgFiles[1:]
	}

	// Load additional config from file(s)
	sharedCfg, err := loadSharedConfig(envCfg.Profile, cfgFiles)
	if err != nil {
		return nil, err
	}

	mergeConfigSrcs(cfg, userCfg, envCfg, sharedCfg, handlers)

	s := &Session{
		Config:   cfg,
		Handlers: handlers,
	}

	initHandlers(s)

	return s, nil
}

func TestOneDayRetrier(t *testing.T) {
	stateChangeClient := newSubmitStateChangeClient(defaults.Config())

	request, _ := stateChangeClient.SubmitContainerStateChangeRequest(&ecs.SubmitContainerStateChangeInput{})

	retrier := stateChangeClient.Retryer

	var totalDelay time.Duration
	for retries := 0; retries < retrier.MaxRetries(); retries++ {
		request.Error = errors.New("")
		request.Retryable = aws.Bool(true)
		request.HTTPResponse = &http.Response{StatusCode: 500}
		if request.WillRetry() && retrier.ShouldRetry(request) {
			totalDelay += retrier.RetryRules(request)
			request.RetryCount++
		}
	}

	request.Error = errors.New("")
	request.Retryable = aws.Bool(true)
	request.HTTPResponse = &http.Response{StatusCode: 500}
	if request.WillRetry() {
		t.Errorf("Expected request to not be retried after %v retries", retrier.MaxRetries())
	}

	if totalDelay > 25*time.Hour || totalDelay < 23*time.Hour {
		t.Errorf("Expected accumulated retry delay to be roughly 24 hours; was %v", totalDelay)
	}
}

func (dynamoDBSource *DynamoDBSource) Get() (map[string]interface{}, error) {

	config := defaults.Config()

	if dynamoDBSource.AccessKey != "" {
		config = config.WithCredentials(credentials.NewCredentials(&credentials.StaticProvider{
			Value: credentials.Value{
				AccessKeyID:     dynamoDBSource.AccessKey,
				SecretAccessKey: dynamoDBSource.SecretKey,
			},
		}))
	}

	if dynamoDBSource.Endpoint != "" {
		config = config.WithEndpoint(dynamoDBSource.Endpoint)
	}

	if dynamoDBSource.Region != "" {
		config = config.WithRegion(dynamoDBSource.Region)
	} else {
		config = config.WithRegion("us-west-1")
	}

	client := dynamodb.New(session.New(config))

	tableName := aws.String(dynamoDBSource.Table)

	describeTableInput := &dynamodb.DescribeTableInput{TableName: tableName}

	if _, err := client.DescribeTable(describeTableInput); err != nil {
		return nil, err
	}

	if err := client.WaitUntilTableExists(describeTableInput); err != nil {
		return nil, err
	}

	key := dynamoDBSource.Key

	response, err := client.GetItem(&dynamodb.GetItemInput{
		Key: map[string]*dynamodb.AttributeValue{
			"key": {S: aws.String(key)},
		},
		TableName:      tableName,
		ConsistentRead: aws.Bool(true),
	})

	if err != nil {
		return nil, err
	}

	result := make(map[string]interface{})
	err = dynamodbattribute.ConvertFromMap(response.Item, &result)

	delete(result, key)

	return result, err
}

// discoverEc2Hosts searches an AWS region, returning a list of instance ips
// where EC2TagKey = EC2TagValue
func (c *Config) discoverEc2Hosts(logger *log.Logger) ([]string, error) {
	config := c.RetryJoinEC2

	ec2meta := ec2metadata.New(session.New())
	if config.Region == "" {
		logger.Printf("[INFO] agent: No EC2 region provided, querying instance metadata endpoint...")
		identity, err := ec2meta.GetInstanceIdentityDocument()
		if err != nil {
			return nil, err
		}
		config.Region = identity.Region
	}

	awsConfig := &aws.Config{
		Region: &config.Region,
		Credentials: credentials.NewChainCredentials(
			[]credentials.Provider{
				&credentials.StaticProvider{
					Value: credentials.Value{
						AccessKeyID:     config.AccessKeyID,
						SecretAccessKey: config.SecretAccessKey,
					},
				},
				&credentials.EnvProvider{},
				&credentials.SharedCredentialsProvider{},
				defaults.RemoteCredProvider(*(defaults.Config()), defaults.Handlers()),
			}),
	}

	svc := ec2.New(session.New(), awsConfig)

	resp, err := svc.DescribeInstances(&ec2.DescribeInstancesInput{
		Filters: []*ec2.Filter{
			{
				Name: aws.String("tag:" + config.TagKey),
				Values: []*string{
					aws.String(config.TagValue),
				},
			},
		},
	})

	if err != nil {
		return nil, err
	}

	var servers []string
	for i := range resp.Reservations {
		for _, instance := range resp.Reservations[i].Instances {
			// Terminated instances don't have the PrivateIpAddress field
			if instance.PrivateIpAddress != nil {
				servers = append(servers, *instance.PrivateIpAddress)
			}
		}
	}

	return servers, nil
}

func TestSessionCopy(t *testing.T) {
	oldEnv := initSessionTestEnv()
	defer popEnv(oldEnv)

	os.Setenv("AWS_REGION", "orig_region")

	s := Session{
		Config:   defaults.Config(),
		Handlers: defaults.Handlers(),
	}

	newSess := s.Copy(&aws.Config{Region: aws.String("new_region")})

	assert.Equal(t, "orig_region", *s.Config.Region)
	assert.Equal(t, "new_region", *newSess.Config.Region)
}

func oldNewSession(cfgs ...*aws.Config) *Session {
	cfg := defaults.Config()
	handlers := defaults.Handlers()

	// Apply the passed in configs so the configuration can be applied to the
	// default credential chain
	cfg.MergeIn(cfgs...)
	cfg.Credentials = defaults.CredChain(cfg, handlers)

	// Reapply any passed in configs to override credentials if set
	cfg.MergeIn(cfgs...)

	s := &Session{
		Config:   cfg,
		Handlers: handlers,
	}

	initHandlers(s)

	return s
}

func main() {
	flag.Set("logtostderr", "1")

	flag.Parse()

	envToFlag("AWS_REGION", "region")
	envToFlag("ES", "es")
	envToFlag("LISTEN", "listen")
	envToFlag("GLOG_v", "v")

	if esEndpoint == "" {
		glog.Fatal("elasticsearch endpoint flag (es) is required")
	}
	if awsRegion == "" {
		glog.Fatal("AWS region flag (region) is required")
	}

	target, err := url.Parse(esEndpoint)
	if err != nil {
		glog.Fatalf("cannot parse es argument (%q) as URL", esEndpoint)
	}

	proxy := httputil.NewSingleHostReverseProxy(target)

	credentials := defaults.CredChain(defaults.Config(), defaults.Handlers())

	signingRoundTripper := NewSigningRoundTripper(proxy.Transport, awsRegion, credentials)
	proxy.Transport = signingRoundTripper

	s := &http.Server{
		Addr:           listenHostAddr,
		Handler:        proxy,
		ReadTimeout:    120 * time.Second,
		WriteTimeout:   120 * time.Second,
		MaxHeaderBytes: 1 << 20,
	}
	glog.Infof("Listening on %s", listenHostAddr)
	err = s.ListenAndServe()
	glog.Fatalf("error listening on %q for http requests: %v", listenHostAddr, err)
}

func _main() int {
	defer log.Flush()
	flagset := flag.NewFlagSet("Amazon ECS Agent", flag.ContinueOnError)
	versionFlag := flagset.Bool("version", false, "Print the agent version information and exit")
	logLevel := flagset.String("loglevel", "", "Loglevel: [<crit>|<error>|<warn>|<info>|<debug>]")
	acceptInsecureCert := flagset.Bool("k", false, "Disable SSL certificate verification. We do not recommend setting this option.")
	licenseFlag := flagset.Bool("license", false, "Print the LICENSE and NOTICE files and exit")
	blackholeEc2Metadata := flagset.Bool("blackhole-ec2-metadata", false, "Blackhole the EC2 Metadata requests. Setting this option can cause the ECS Agent to fail to work properly.  We do not recommend setting this option")
	err := flagset.Parse(os.Args[1:])
	if err != nil {
		return exitcodes.ExitTerminal
	}

	if *licenseFlag {
		license := utils.NewLicenseProvider()
		text, err := license.GetText()
		if err != nil {
			fmt.Fprintln(os.Stderr, err)
			return exitcodes.ExitError
		}
		fmt.Println(text)
		return exitcodes.ExitSuccess
	}

	logger.SetLevel(*logLevel)
	ec2MetadataClient := ec2.DefaultClient
	if *blackholeEc2Metadata {
		ec2MetadataClient = ec2.NewBlackholeEC2MetadataClient()
	}

	log.Infof("Starting Agent: %s", version.String())
	if *acceptInsecureCert {
		log.Warn("SSL certificate verification disabled. This is not recommended.")
	}
	log.Info("Loading configuration")
	cfg, cfgErr := config.NewConfig(ec2MetadataClient)
	// Load cfg and create Docker client before doing 'versionFlag' so that it has the DOCKER_HOST variable loaded if needed
	clientFactory := dockerclient.NewFactory(cfg.DockerEndpoint)
	dockerClient, err := engine.NewDockerGoClient(clientFactory, *acceptInsecureCert, cfg)
	if err != nil {
		log.Criticalf("Error creating Docker client: %v", err)
		return exitcodes.ExitError
	}

	ctx := context.Background()
	// Create the DockerContainerChange event stream for tcs
	containerChangeEventStream := eventstream.NewEventStream(ContainerChangeEventStream, ctx)
	containerChangeEventStream.StartListening()

	// Create credentials manager. This will be used by the task engine and
	// the credentials handler
	credentialsManager := credentials.NewManager()
	// Create image manager. This will be used by the task engine for saving image states
	state := dockerstate.NewDockerTaskEngineState()
	imageManager := engine.NewImageManager(cfg, dockerClient, state)
	if *versionFlag {
		versionableEngine := engine.NewTaskEngine(cfg, dockerClient, credentialsManager, containerChangeEventStream, imageManager, state)
		version.PrintVersion(versionableEngine)
		return exitcodes.ExitSuccess
	}

	sighandlers.StartDebugHandler()

	if cfgErr != nil {
		log.Criticalf("Error loading config: %v", err)
		// All required config values can be inferred from EC2 Metadata, so this error could be transient.
		return exitcodes.ExitError
	}
	log.Debug("Loaded config: " + cfg.String())

	var currentEc2InstanceID, containerInstanceArn string
	var taskEngine engine.TaskEngine

	if cfg.Checkpoint {
		log.Info("Checkpointing is enabled. Attempting to load state")
		var previousCluster, previousEc2InstanceID, previousContainerInstanceArn string
		previousTaskEngine := engine.NewTaskEngine(cfg, dockerClient, credentialsManager, containerChangeEventStream, imageManager, state)
		// previousState is used to verify that our current runtime configuration is
		// compatible with our past configuration as reflected by our state-file
		previousState, err := initializeStateManager(cfg, previousTaskEngine, &previousCluster, &previousContainerInstanceArn, &previousEc2InstanceID)
		if err != nil {
			log.Criticalf("Error creating state manager: %v", err)
			return exitcodes.ExitTerminal
		}

		err = previousState.Load()
		if err != nil {
			log.Criticalf("Error loading previously saved state: %v", err)
			return exitcodes.ExitTerminal
		}

		if previousCluster != "" {
			// TODO Handle default cluster in a sane and unified way across the codebase
			configuredCluster := cfg.Cluster
			if configuredCluster == "" {
				log.Debug("Setting cluster to default; none configured")
				configuredCluster = config.DefaultClusterName
			}
			if previousCluster != configuredCluster {
				log.Criticalf("Data mismatch; saved cluster '%v' does not match configured cluster '%v'. Perhaps you want to delete the configured checkpoint file?", previousCluster, configuredCluster)
//.........这里部分代码省略.........