ecs.go

package ecs

import (
	"context"
	"fmt"
	"strings"
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/credentials"
	"github.com/aws/aws-sdk-go/aws/defaults"
	"github.com/aws/aws-sdk-go/aws/ec2metadata"
	"github.com/aws/aws-sdk-go/aws/request"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/ec2"
	"github.com/aws/aws-sdk-go/service/ecs"
	"github.com/cenk/backoff"
	"github.com/containous/traefik/job"
	"github.com/containous/traefik/log"
	"github.com/containous/traefik/provider"
	"github.com/containous/traefik/safe"
	"github.com/containous/traefik/types"
)

var _ provider.Provider = (*Provider)(nil)

// Provider holds configurations of the provider.
type Provider struct {
	provider.BaseProvider `mapstructure:",squash" export:"true"`

	Domain           string `description:"Default domain used"`
	ExposedByDefault bool   `description:"Expose containers by default" export:"true"`
	RefreshSeconds   int    `description:"Polling interval (in seconds)" export:"true"`

	// Provider lookup parameters
	Clusters             Clusters `description:"ECS Clusters name"`
	Cluster              string   `description:"deprecated - ECS Cluster name"` // deprecated
	AutoDiscoverClusters bool     `description:"Auto discover cluster" export:"true"`
	Region               string   `description:"The AWS region to use for requests" export:"true"`
	AccessKeyID          string   `description:"The AWS credentials access key to use for making requests"`
	SecretAccessKey      string   `description:"The AWS credentials access key to use for making requests"`
}

type ecsInstance struct {
	Name                string
	ID                  string
	task                *ecs.Task
	taskDefinition      *ecs.TaskDefinition
	container           *ecs.Container
	containerDefinition *ecs.ContainerDefinition
	machine             *ec2.Instance
	TraefikLabels       map[string]string
}

type awsClient struct {
	ecs *ecs.ECS
	ec2 *ec2.EC2
}

func (p *Provider) createClient() (*awsClient, error) {
	sess, err := session.NewSession()
	if err != nil {
		return nil, err
	}
	ec2meta := ec2metadata.New(sess)
	if p.Region == "" {
		log.Infoln("No EC2 region provided, querying instance metadata endpoint...")
		identity, err := ec2meta.GetInstanceIdentityDocument()
		if err != nil {
			return nil, err
		}
		p.Region = identity.Region
	}

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

	if p.Trace {
		cfg.WithLogger(aws.LoggerFunc(func(args ...interface{}) {
			log.Debug(args...)
		}))
	}

	return &awsClient{
		ecs.New(sess, cfg),
		ec2.New(sess, cfg),
	}, nil
}

// Provide allows the ecs provider to provide configurations to traefik
// using the given configuration channel.
func (p *Provider) Provide(configurationChan chan<- types.ConfigMessage, pool *safe.Pool, constraints types.Constraints) error {
	p.Constraints = append(p.Constraints, constraints...)

	handleCanceled := func(ctx context.Context, err error) error {
		if ctx.Err() == context.Canceled || err == context.Canceled {
			return nil
		}
		return err
	}

	pool.Go(func(stop chan bool) {
		ctx, cancel := context.WithCancel(context.Background())
		safe.Go(func() {
			<-stop
			cancel()
		})

		operation := func() error {
			awsClient, err := p.createClient()
			if err != nil {
				return err
			}

			configuration, err := p.loadECSConfig(ctx, awsClient)
			if err != nil {
				return handleCanceled(ctx, err)
			}

			configurationChan <- types.ConfigMessage{
				ProviderName:  "ecs",
				Configuration: configuration,
			}

			if p.Watch {
				reload := time.NewTicker(time.Second * time.Duration(p.RefreshSeconds))
				defer reload.Stop()
				for {
					select {
					case <-reload.C:
						configuration, err := p.loadECSConfig(ctx, awsClient)
						if err != nil {
							return handleCanceled(ctx, err)
						}

						configurationChan <- types.ConfigMessage{
							ProviderName:  "ecs",
							Configuration: configuration,
						}
					case <-ctx.Done():
						return handleCanceled(ctx, ctx.Err())
					}
				}
			}

			return nil
		}

		notify := func(err error, time time.Duration) {
			log.Errorf("Provider connection error %+v, retrying in %s", err, time)
		}
		err := backoff.RetryNotify(safe.OperationWithRecover(operation), job.NewBackOff(backoff.NewExponentialBackOff()), notify)
		if err != nil {
			log.Errorf("Cannot connect to Provider api %+v", err)
		}
	})

	return nil
}

func wrapAws(ctx context.Context, req *request.Request) error {
	req.HTTPRequest = req.HTTPRequest.WithContext(ctx)
	return req.Send()
}

// Find all running Provider tasks in a cluster, also collect the task definitions (for docker labels)
// and the EC2 instance data
func (p *Provider) listInstances(ctx context.Context, client *awsClient) ([]ecsInstance, error) {
	var clustersArn []*string
	var clusters Clusters

	if p.AutoDiscoverClusters {
		input := &ecs.ListClustersInput{}
		for {
			result, err := client.ecs.ListClusters(input)
			if err != nil {
				return nil, err
			}
			if result != nil {
				clustersArn = append(clustersArn, result.ClusterArns...)
				input.NextToken = result.NextToken
				if result.NextToken == nil {
					break
				}
			} else {
				break
			}
		}
		for _, cArn := range clustersArn {
			clusters = append(clusters, *cArn)
		}
	} else if p.Cluster != "" {
		// TODO: Deprecated configuration - Need to be removed in the future
		clusters = Clusters{p.Cluster}
		log.Warn("Deprecated configuration found: ecs.cluster " +
			"Please use ecs.clusters instead.")
	} else {
		clusters = p.Clusters
	}

	var instances []ecsInstance

	log.Debugf("ECS Clusters: %s", clusters)

	for _, c := range clusters {

		req, _ := client.ecs.ListTasksRequest(&ecs.ListTasksInput{
			Cluster:       &c,
			DesiredStatus: aws.String(ecs.DesiredStatusRunning),
		})

		var taskArns []*string

		for ; req != nil; req = req.NextPage() {
			if err := wrapAws(ctx, req); err != nil {
				return nil, err
			}

			taskArns = append(taskArns, req.Data.(*ecs.ListTasksOutput).TaskArns...)
		}

		// Skip to the next cluster if there are no tasks found on
		// this cluster.
		if len(taskArns) == 0 {
			continue
		}

		chunkedTaskArns := chunkedTaskArns(taskArns)
		var tasks []*ecs.Task

		for _, arns := range chunkedTaskArns {
			req, taskResp := client.ecs.DescribeTasksRequest(&ecs.DescribeTasksInput{
				Tasks:   arns,
				Cluster: &c,
			})

			if err := wrapAws(ctx, req); err != nil {
				return nil, err
			}
			tasks = append(tasks, taskResp.Tasks...)

		}

		containerInstanceArns := make([]*string, 0)
		byContainerInstance := make(map[string]int)

		taskDefinitionArns := make([]*string, 0)
		byTaskDefinition := make(map[string]int)

		for _, task := range tasks {
			if _, found := byContainerInstance[aws.StringValue(task.ContainerInstanceArn)]; !found {
				byContainerInstance[aws.StringValue(task.ContainerInstanceArn)] = len(containerInstanceArns)
				containerInstanceArns = append(containerInstanceArns, task.ContainerInstanceArn)
			}
			if _, found := byTaskDefinition[aws.StringValue(task.TaskDefinitionArn)]; !found {
				byTaskDefinition[aws.StringValue(task.TaskDefinitionArn)] = len(taskDefinitionArns)
				taskDefinitionArns = append(taskDefinitionArns, task.TaskDefinitionArn)
			}
		}

		machines, err := p.lookupEc2Instances(ctx, client, &c, containerInstanceArns)
		if err != nil {
			return nil, err
		}

		taskDefinitions, err := p.lookupTaskDefinitions(ctx, client, taskDefinitionArns)
		if err != nil {
			return nil, err
		}

		for _, task := range tasks {

			machineIdx := byContainerInstance[aws.StringValue(task.ContainerInstanceArn)]
			taskDefIdx := byTaskDefinition[aws.StringValue(task.TaskDefinitionArn)]

			for _, container := range task.Containers {

				taskDefinition := taskDefinitions[taskDefIdx]
				var containerDefinition *ecs.ContainerDefinition
				for _, def := range taskDefinition.ContainerDefinitions {
					if aws.StringValue(container.Name) == aws.StringValue(def.Name) {
						containerDefinition = def
						break
					}
				}

				instances = append(instances, ecsInstance{
					Name:                fmt.Sprintf("%s-%s", strings.Replace(aws.StringValue(task.Group), ":", "-", 1), *container.Name),
					ID:                  (aws.StringValue(task.TaskArn))[len(aws.StringValue(task.TaskArn))-12:],
					task:                task,
					taskDefinition:      taskDefinition,
					container:           container,
					containerDefinition: containerDefinition,
					machine:             machines[machineIdx],
					TraefikLabels:       aws.StringValueMap(containerDefinition.DockerLabels),
				})
			}
		}
	}

	return instances, nil
}

func (p *Provider) lookupEc2Instances(ctx context.Context, client *awsClient, clusterName *string, containerArns []*string) ([]*ec2.Instance, error) {

	order := make(map[string]int)
	instanceIds := make([]*string, len(containerArns))
	instances := make([]*ec2.Instance, len(containerArns))
	for i, arn := range containerArns {
		order[aws.StringValue(arn)] = i
	}

	req, _ := client.ecs.DescribeContainerInstancesRequest(&ecs.DescribeContainerInstancesInput{
		ContainerInstances: containerArns,
		Cluster:            clusterName,
	})

	for ; req != nil; req = req.NextPage() {
		if err := wrapAws(ctx, req); err != nil {
			return nil, err
		}

		containerResp := req.Data.(*ecs.DescribeContainerInstancesOutput)
		for i, container := range containerResp.ContainerInstances {
			order[aws.StringValue(container.Ec2InstanceId)] = order[aws.StringValue(container.ContainerInstanceArn)]
			instanceIds[i] = container.Ec2InstanceId
		}
	}

	req, _ = client.ec2.DescribeInstancesRequest(&ec2.DescribeInstancesInput{
		InstanceIds: instanceIds,
	})

	for ; req != nil; req = req.NextPage() {
		if err := wrapAws(ctx, req); err != nil {
			return nil, err
		}

		instancesResp := req.Data.(*ec2.DescribeInstancesOutput)
		for _, r := range instancesResp.Reservations {
			for _, i := range r.Instances {
				if i.InstanceId != nil {
					instances[order[aws.StringValue(i.InstanceId)]] = i
				}
			}
		}
	}
	return instances, nil
}

func (p *Provider) lookupTaskDefinitions(ctx context.Context, client *awsClient, taskDefArns []*string) ([]*ecs.TaskDefinition, error) {
	taskDefinitions := make([]*ecs.TaskDefinition, len(taskDefArns))
	for i, arn := range taskDefArns {

		req, resp := client.ecs.DescribeTaskDefinitionRequest(&ecs.DescribeTaskDefinitionInput{
			TaskDefinition: arn,
		})

		if err := wrapAws(ctx, req); err != nil {
			return nil, err
		}

		taskDefinitions[i] = resp.TaskDefinition
	}
	return taskDefinitions, nil
}

func (p *Provider) loadECSConfig(ctx context.Context, client *awsClient) (*types.Configuration, error) {
	instances, err := p.listInstances(ctx, client)
	if err != nil {
		return nil, err
	}

	return p.buildConfiguration(instances)
}

// Provider expects no more than 100 parameters be passed to a DescribeTask call; thus, pack
// each string into an array capped at 100 elements
func chunkedTaskArns(tasks []*string) [][]*string {
	var chunkedTasks [][]*string
	for i := 0; i < len(tasks); i += 100 {
		var sliceEnd int
		if i+100 < len(tasks) {
			sliceEnd = i + 100
		} else {
			sliceEnd = len(tasks)
		}
		chunkedTasks = append(chunkedTasks, tasks[i:sliceEnd])
	}
	return chunkedTasks
}