api_client.go

// Copyright 2014-2015 Amazon.com, Inc. or its affiliates. All Rights Reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"). You may
// not use this file except in compliance with the License. A copy of the
// License is located at
//
//	http://aws.amazon.com/apache2.0/
//
// or in the "license" file accompanying this file. This file is distributed
// on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
// express or implied. See the License for the specific language governing
// permissions and limitations under the License.

package api

import (
	"crypto/tls"
	"errors"
	"runtime"
	"strconv"
	"strings"

	"github.com/aws/amazon-ecs-agent/agent/ecs_client/awsjson/codec"
	"github.com/aws/amazon-ecs-agent/agent/ecs_client/client/dialer"
	svc "github.com/aws/amazon-ecs-agent/agent/ecs_client/ecs_autogenerated_client"

	"github.com/docker/docker/pkg/system"

	"github.com/aws/amazon-ecs-agent/agent/ecs_client/authv4"
	"github.com/aws/amazon-ecs-agent/agent/ecs_client/authv4/credentials"

	"github.com/aws/amazon-ecs-agent/agent/config"
	"github.com/aws/amazon-ecs-agent/agent/ec2"
	"github.com/aws/amazon-ecs-agent/agent/logger"
	"github.com/aws/amazon-ecs-agent/agent/utils"
)

var log = logger.ForModule("api client")

type ECSClient interface {
	CredentialProvider() credentials.AWSCredentialProvider
	RegisterContainerInstance() (string, error)
	SubmitTaskStateChange(change ContainerStateChange) utils.RetriableError
	SubmitContainerStateChange(change ContainerStateChange) utils.RetriableError
	DiscoverPollEndpoint(containerInstanceArn string) (string, error)
}

type ApiECSClient struct {
	credentialProvider credentials.AWSCredentialProvider
	config             *config.Config
	insecureSkipVerify bool

	// Swappable impl for testing
	serviceClientFn func() (svc.AmazonEC2ContainerServiceV20141113, error)
}

const (
	ECS_SERVICE = "ecs"
)

const EcsMaxReasonLength = 255

func NewECSClient(credentialProvider credentials.AWSCredentialProvider, config *config.Config, insecureSkipVerify bool) ECSClient {
	client := &ApiECSClient{credentialProvider: credentialProvider,
		config:             config,
		insecureSkipVerify: insecureSkipVerify,
	}
	client.serviceClientFn = func() (svc.AmazonEC2ContainerServiceV20141113, error) {
		return client.serviceClientImpl()
	}
	return client
}

// serviceClient provides a client for interacting with the ECS service apis
func (client *ApiECSClient) serviceClient() (svc.AmazonEC2ContainerServiceV20141113, error) {
	return client.serviceClientFn()
}

// serviceClientImpl is the default serviceClient provider.
func (client *ApiECSClient) serviceClientImpl() (svc.AmazonEC2ContainerServiceV20141113, error) {
	config := client.config

	signer := authv4.NewHttpSigner(config.AWSRegion, ECS_SERVICE, client.CredentialProvider(), nil)

	c := codec.AwsJson{Host: config.APIEndpoint, SignerV4: signer}
	endpoint := config.APIEndpoint
	port := uint16(443)
	if parts := strings.Split(endpoint, ":"); len(parts) == 2 {
		endpoint = parts[0]
		tmpPort, err := strconv.Atoi(parts[1])
		port = uint16(tmpPort)
		if err != nil {
			return nil, errors.New("Malformed endpoint url given")
		}
	}

	d, err := dialer.TLS(endpoint, port, &tls.Config{InsecureSkipVerify: client.insecureSkipVerify})

	if err != nil {
		log.Error("Cannot resolve url", "url", endpoint, "port", port, "err", err)
		return nil, err
	}

	ecs := svc.NewAmazonEC2ContainerServiceV20141113Client(d, c)
	return ecs, nil
}

func (client *ApiECSClient) CredentialProvider() credentials.AWSCredentialProvider {
	return client.credentialProvider
}

func getCpuAndMemory() (int32, int32) {
	memInfo, err := system.ReadMemInfo()
	mem := int32(memInfo.MemTotal / 1024 / 1024) // MB
	if err != nil {
		log.Error("Unable to get memory info", "err", err)
		mem = 0
	}
	cpu := runtime.NumCPU() * 1024

	return int32(cpu), mem
}

// CreateCluster creates a cluster from a given name and returns its arn
func (client *ApiECSClient) CreateCluster(clusterName string) (string, error) {
	svcRequest := svc.NewCreateClusterRequest()
	svcRequest.SetClusterName(&clusterName)

	svcClient, err := client.serviceClient()
	if err != nil {
		log.Error("Unable to get service client for frontend", "err", err)
		return "", err
	}

	resp, err := svcClient.CreateCluster(svcRequest)
	if err != nil {
		log.Crit("Could not create cluster", "err", err)
		return "", NewAPIError(err)
	}
	log.Info("Created a cluster!", "clusterName", clusterName)
	return *resp.Cluster().ClusterName(), nil

}

func (client *ApiECSClient) describeCluster(clusterName string) (clusterRef string, clusterStatus string, err error) {
	svcRequest := svc.NewDescribeClustersRequest()
	clusterNames := []*string{&clusterName}
	svcRequest.SetClusters(clusterNames)

	svcClient, err := client.serviceClient()
	if err != nil {
		log.Error("Unable to get service client for frontend", "err", err)
		return
	}

	resp, err := svcClient.DescribeClusters(svcRequest)
	if err != nil {
		log.Error("Unable to describe cluster", "cluster", clusterName, "err", err)
		return
	}
	for _, cluster := range resp.Clusters() {
		if *cluster.ClusterName() == clusterName {
			clusterRef = *cluster.ClusterName()
			clusterStatus = *cluster.Status()
			return
		}
	}
	return
}

func (client *ApiECSClient) RegisterContainerInstance() (string, error) {
	clusterRef := client.config.Cluster
	// If our clusterRef is empty, we should try to create the default
	if clusterRef == "" {
		clusterRef = config.DEFAULT_CLUSTER_NAME
		defer func() {
			// Update the config value to reflect the cluster we end up in
			client.config.Cluster = clusterRef
		}()
		// Attempt to register without checking existence of the cluster so we don't require
		// excess permissions in the case where the cluster already exists and is active
		containerInstanceArn, err := client.registerContainerInstance(clusterRef)
		if err == nil {
			return containerInstanceArn, nil
		}
		// If trying to register fails, try to create the cluster before calling
		// register again
		clusterRef, err = client.CreateCluster(clusterRef)
		if err != nil {
			return "", NewAPIError(err)
		}
	}
	return client.registerContainerInstance(clusterRef)
}

func (client *ApiECSClient) registerContainerInstance(clusterRef string) (string, error) {
	svcRequest := svc.NewRegisterContainerInstanceRequest()
	svcRequest.SetCluster(&clusterRef)

	instanceIdentityDoc, err := ec2.ReadResource(ec2.INSTANCE_IDENTITY_DOCUMENT_RESOURCE)
	iidRetrieved := true
	if err != nil {
		log.Error("Unable to get instance identity document", "err", err)
		iidRetrieved = false
		instanceIdentityDoc = []byte{}
	}
	strIid := string(instanceIdentityDoc)
	svcRequest.SetInstanceIdentityDocument(&strIid)

	instanceIdentitySignature := []byte{}
	if iidRetrieved {
		instanceIdentitySignature, err = ec2.ReadResource(ec2.INSTANCE_IDENTITY_DOCUMENT_SIGNATURE_RESOURCE)
		if err != nil {
			log.Error("Unable to get instance identity signature", "err", err)
		}
	}

	strIidSig := string(instanceIdentitySignature)
	svcRequest.SetInstanceIdentityDocumentSignature(&strIidSig)

	integerStr := "INTEGER"

	cpu, mem := getCpuAndMemory()

	cpuResource := svc.NewResource()
	cpuResource.SetName(utils.Strptr("CPU"))
	cpuResource.SetType(&integerStr)
	cpuResource.SetIntegerValue(&cpu)

	memResource := svc.NewResource()
	memResource.SetName(utils.Strptr("MEMORY"))
	memResource.SetType(&integerStr)
	memResource.SetIntegerValue(&mem)

	portResource := svc.NewResource()
	portResource.SetName(utils.Strptr("PORTS"))
	portResource.SetType(utils.Strptr("STRINGSET"))
	portResource.SetStringSetValue(utils.Uint16SliceToStringSlice(client.config.ReservedPorts))

	resources := []svc.Resource{cpuResource, memResource, portResource}
	svcRequest.SetTotalResources(resources)

	ecs, err := client.serviceClient()
	if err != nil {
		log.Error("Unable to get service client for frontend", "err", err)
		return "", err
	}

	resp, err := ecs.RegisterContainerInstance(svcRequest)
	if err != nil {
		log.Error("Could not register", "err", err)
		return "", NewAPIError(err)
	}
	log.Info("Registered!")
	return *resp.ContainerInstance().ContainerInstanceArn(), nil
}

func (client *ApiECSClient) SubmitTaskStateChange(change ContainerStateChange) utils.RetriableError {
	if change.TaskStatus == TaskStatusNone {
		log.Warn("SubmitTaskStateChange called with an invalid change", "change", change)
		return NewAPIError(errors.New("SubmitTaskStateChange called with an invalid change"))
	}

	stat := change.TaskStatus.String()
	if stat == "DEAD" {
		stat = "STOPPED"
	}
	if stat != "STOPPED" && stat != "RUNNING" {
		log.Debug("Not submitting unsupported upstream task state", "state", stat)
		// Not really an error
		return nil
	}

	req := svc.NewSubmitTaskStateChangeRequest()
	req.SetTask(&change.TaskArn)
	req.SetStatus(&stat)
	req.SetCluster(&client.config.Cluster)

	c, err := client.serviceClient()
	if err != nil {
		return NewAPIError(err)
	}
	_, err = c.SubmitTaskStateChange(req)
	if err != nil {
		log.Warn("Could not submit a task state change", "err", err)
		return NewAPIError(err)
	}
	return nil
}

func (client *ApiECSClient) SubmitContainerStateChange(change ContainerStateChange) utils.RetriableError {
	req := svc.NewSubmitContainerStateChangeRequest()
	req.SetTask(&change.TaskArn)
	req.SetContainerName(&change.ContainerName)
	stat := change.Status.String()
	if stat == "DEAD" {
		stat = "STOPPED"
	}
	if stat != "STOPPED" && stat != "RUNNING" {
		log.Info("Not submitting not supported upstream container state", "state", stat)
		return nil
	}

	req.SetStatus(&stat)
	req.SetCluster(&client.config.Cluster)
	if change.ExitCode != nil {
		exitCode := int32(*change.ExitCode)
		req.SetExitCode(&exitCode)
	}
	if change.Reason != "" {
		reason := change.Reason
		if len(reason) > EcsMaxReasonLength {
			reason = reason[:EcsMaxReasonLength]
		}
		req.SetReason(&reason)
	}
	networkBindings := make([]svc.NetworkBinding, len(change.PortBindings))
	for i, binding := range change.PortBindings {
		aBinding := svc.NewNetworkBinding()
		aBinding.SetBindIP(&binding.BindIp)
		hostPort := int32(binding.HostPort)
		aBinding.SetHostPort(&hostPort)
		containerPort := int32(binding.ContainerPort)
		aBinding.SetContainerPort(&containerPort)
		networkBindings[i] = aBinding
	}
	req.SetNetworkBindings(networkBindings)

	c, err := client.serviceClient()
	if err != nil {
		return NewAPIError(err)
	}
	_, err = c.SubmitContainerStateChange(req)
	if err != nil {
		log.Warn("Could not submit a container state change", "change", change, "err", err)
		return NewAPIError(err)
	}
	return nil
}

func (client *ApiECSClient) DiscoverPollEndpoint(containerInstanceArn string) (string, error) {
	req := svc.NewDiscoverPollEndpointRequest()
	req.SetContainerInstance(&containerInstanceArn)
	req.SetCluster(&client.config.Cluster)

	c, err := client.serviceClient()
	if err != nil {
		return "", NewAPIError(err)
	}

	resp, err := c.DiscoverPollEndpoint(req)
	if err != nil {
		return "", NewAPIError(err)
	}

	return *resp.Endpoint(), nil
}