<commit_before>// 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 engine

import (
	"archive/tar"
	"bufio"
	"io"
	"os"
	"strings"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/aws/amazon-ecs-agent/agent/api"
	"github.com/aws/amazon-ecs-agent/agent/engine/dockerauth"
	"github.com/aws/amazon-ecs-agent/agent/engine/dockerclient"
	"github.com/aws/amazon-ecs-agent/agent/engine/emptyvolume"
	"github.com/aws/amazon-ecs-agent/agent/utils"
	"github.com/aws/amazon-ecs-agent/agent/utils/ttime"

	docker "github.com/fsouza/go-dockerclient"
)

const dockerStopTimeoutSeconds = 30

// Timelimits for docker operations enforced above docker
const (
	pullImageTimeout        = 2 * time.Hour
	createContainerTimeout  = 1 * time.Minute
	startContainerTimeout   = 1 * time.Minute
	stopContainerTimeout    = 1 * time.Minute
	removeContainerTimeout  = 5 * time.Minute
	inspectContainerTimeout = 10 * time.Second
)

// Interface to make testing it easier
type DockerClient interface {
	ContainerEvents(ctx context.Context) (<-chan DockerContainerChangeEvent, error)

	PullImage(image string) DockerContainerMetadata
	CreateContainer(*docker.Config, string) DockerContainerMetadata
	StartContainer(string, *docker.HostConfig) DockerContainerMetadata
	StopContainer(string) DockerContainerMetadata
	DescribeContainer(string) (api.ContainerStatus, DockerContainerMetadata)

	RemoveContainer(string) error

	GetContainerName(string) (string, error)
	InspectContainer(string) (*docker.Container, error)

	Version() (string, error)
}

// Implements DockerClient
type DockerGoClient struct {
	dockerClient dockerclient.Client
}

func (dg *DockerGoClient) SetGoDockerClient(to dockerclient.Client) {
	dg.dockerClient = to
}

// pullLock is a temporary workaround for a devicemapper issue. See: https://github.com/docker/docker/issues/9718
var pullLock sync.Mutex

// scratchCreateLock guards against multiple 'scratch' image creations at once
var scratchCreateLock sync.Mutex

type DockerImageResponse struct {
	Images []docker.APIImages
}

func NewDockerGoClient() (*DockerGoClient, error) {
	endpoint := utils.DefaultIfBlank(os.Getenv(DOCKER_ENDPOINT_ENV_VARIABLE), DOCKER_DEFAULT_ENDPOINT)

	client, err := docker.NewVersionedClient(endpoint, "1.17")
	if err != nil {
		log.Error("Unable to connect to docker daemon . Ensure docker is running", "endpoint", endpoint, "err", err)
		return nil, err
	}

	// Even if we have a dockerclient, the daemon might not be running. Ping it
	// to ensure it's up.
	err = client.Ping()
	if err != nil {
		log.Error("Unable to ping docker daemon. Ensure docker is running", "endpoint", endpoint, "err", err)
		return nil, err
	}

	return &DockerGoClient{
		dockerClient: client,
	}, nil
}

func (dg *DockerGoClient) PullImage(image string) DockerContainerMetadata {
	timeout := ttime.After(pullImageTimeout)

	response := make(chan DockerContainerMetadata, 1)
	go func() { response <- dg.pullImage(image) }()
	select {
	case resp := <-response:
		return resp
	case <-timeout:
		return DockerContainerMetadata{Error: &DockerTimeoutError{pullImageTimeout, "pulled"}}
	}
}

func (dg *DockerGoClient) pullImage(image string) DockerContainerMetadata {
	log.Debug("Pulling image", "image", image)
	client := dg.dockerClient

	// Special case; this image is not one that should be pulled, but rather
	// should be created locally if necessary
	if image == emptyvolume.Image+":"+emptyvolume.Tag {
		err := dg.createScratchImageIfNotExists()
		if err != nil {
			return DockerContainerMetadata{Error: &api.DefaultNamedError{Name: "CreateEmptyVolumeError", Err: "Could not create empty volume " + err.Error()}}
		}
		return DockerContainerMetadata{}
	}

	authConfig := dockerauth.GetAuthconfig(image)
	// Workaround for devicemapper bug. See:
	// https://github.com/docker/docker/issues/9718
	pullLock.Lock()
	defer pullLock.Unlock()

	pullDebugOut, pullWriter := io.Pipe()
	defer pullWriter.Close()
	opts := docker.PullImageOptions{
		Repository:   image,
		OutputStream: pullWriter,
	}
	go func() {
		reader := bufio.NewReader(pullDebugOut)
		var line string
		var err error
		for err == nil {
			line, err = reader.ReadString('\n')
			if err != nil {
				break
			}
			log.Debug("Pulling image", "image", image, "status", line)
			if strings.Contains(line, "already being pulled by another client. Waiting.") {
				// This can mean the deamon is 'hung' in pulling status for this image, but we can't be sure.
				log.Error("Image 'pull' status marked as already being pulled", "image", image, "status", line)
			}
		}
		if err != nil && err != io.EOF {
			log.Warn("Error reading pull image status", "image", image, "err", err)
		}
		log.Debug("Pulling image complete", "image", image)
	}()
	err := client.PullImage(opts, authConfig)
	if err != nil {
		return DockerContainerMetadata{Error: CannotXContainerError{"Pulled", err.Error()}}
	}
	return DockerContainerMetadata{}
}<commit_msg>Engine: Pull - jitter instead of lock<commit_after>// 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 engine

import (
	"archive/tar"
	"bufio"
	"io"
	"os"
	"strings"
	"sync"
	"time"

	"golang.org/x/net/context"

	"github.com/aws/amazon-ecs-agent/agent/api"
	"github.com/aws/amazon-ecs-agent/agent/engine/dockerauth"
	"github.com/aws/amazon-ecs-agent/agent/engine/dockerclient"
	"github.com/aws/amazon-ecs-agent/agent/engine/emptyvolume"
	"github.com/aws/amazon-ecs-agent/agent/utils"
	"github.com/aws/amazon-ecs-agent/agent/utils/ttime"

	docker "github.com/fsouza/go-dockerclient"
)

const dockerStopTimeoutSeconds = 30

// Timelimits for docker operations enforced above docker
const (
	pullImageTimeout        = 2 * time.Hour
	createContainerTimeout  = 1 * time.Minute
	startContainerTimeout   = 1 * time.Minute
	stopContainerTimeout    = 1 * time.Minute
	removeContainerTimeout  = 5 * time.Minute
	inspectContainerTimeout = 10 * time.Second

	// dockerPullBeginTimeout is the timeout from when a 'pull' is called to when
	// we expect to see output on the pull progress stream. This is to work
	// around a docker bug which sometimes results in pulls not progressing.
	dockerPullBeginTimeout = 45 * time.Second
	dockerPullJitter       = 10 * time.Second
)

// Interface to make testing it easier
type DockerClient interface {
	ContainerEvents(ctx context.Context) (<-chan DockerContainerChangeEvent, error)

	PullImage(image string) DockerContainerMetadata
	CreateContainer(*docker.Config, string) DockerContainerMetadata
	StartContainer(string, *docker.HostConfig) DockerContainerMetadata
	StopContainer(string) DockerContainerMetadata
	DescribeContainer(string) (api.ContainerStatus, DockerContainerMetadata)

	RemoveContainer(string) error

	GetContainerName(string) (string, error)
	InspectContainer(string) (*docker.Container, error)

	Version() (string, error)
}

// Implements DockerClient
type DockerGoClient struct {
	dockerClient dockerclient.Client
}

func (dg *DockerGoClient) SetGoDockerClient(to dockerclient.Client) {
	dg.dockerClient = to
}

// pullLock is a temporary workaround for a devicemapper issue. See: https://github.com/docker/docker/issues/9718
var pullLock sync.Mutex

// scratchCreateLock guards against multiple 'scratch' image creations at once
var scratchCreateLock sync.Mutex

type DockerImageResponse struct {
	Images []docker.APIImages
}

func NewDockerGoClient() (*DockerGoClient, error) {
	endpoint := utils.DefaultIfBlank(os.Getenv(DOCKER_ENDPOINT_ENV_VARIABLE), DOCKER_DEFAULT_ENDPOINT)

	client, err := docker.NewVersionedClient(endpoint, "1.17")
	if err != nil {
		log.Error("Unable to connect to docker daemon . Ensure docker is running", "endpoint", endpoint, "err", err)
		return nil, err
	}

	// Even if we have a dockerclient, the daemon might not be running. Ping it
	// to ensure it's up.
	err = client.Ping()
	if err != nil {
		log.Error("Unable to ping docker daemon. Ensure docker is running", "endpoint", endpoint, "err", err)
		return nil, err
	}

	return &DockerGoClient{
		dockerClient: client,
	}, nil
}

func (dg *DockerGoClient) PullImage(image string) DockerContainerMetadata {
	timeout := ttime.After(pullImageTimeout)

	response := make(chan DockerContainerMetadata, 1)
	go func() { response <- dg.pullImage(image) }()
	select {
	case resp := <-response:
		return resp
	case <-timeout:
		return DockerContainerMetadata{Error: &DockerTimeoutError{pullImageTimeout, "pulled"}}
	}
}

func (dg *DockerGoClient) pullImage(image string) DockerContainerMetadata {
	log.Debug("Pulling image", "image", image)
	client := dg.dockerClient

	// Special case; this image is not one that should be pulled, but rather
	// should be created locally if necessary
	if image == emptyvolume.Image+":"+emptyvolume.Tag {
		err := dg.createScratchImageIfNotExists()
		if err != nil {
			return DockerContainerMetadata{Error: &api.DefaultNamedError{Name: "CreateEmptyVolumeError", Err: "Could not create empty volume " + err.Error()}}
		}
		return DockerContainerMetadata{}
	}

	authConfig := dockerauth.GetAuthconfig(image)
	// Workaround for devicemapper bug. See:
	// https://github.com/docker/docker/issues/9718
	pullLock.Lock()
	ttime.Sleep(utils.AddJitter(0, dockerPullJitter))
	pullLock.Unlock()

	pullDebugOut, pullWriter := io.Pipe()
	defer pullWriter.Close()
	opts := docker.PullImageOptions{
		Repository:   image,
		OutputStream: pullWriter,
	}
	timeout := ttime.After(dockerPullBeginTimeout)
	pullBegan := make(chan bool, 1)
	pullBeganOnce := sync.Once{}

	go func() {
		reader := bufio.NewReader(pullDebugOut)
		var line string
		var err error
		for err == nil {
			line, err = reader.ReadString('\n')
			if err != nil {
				break
			}
			pullBeganOnce.Do(func() {
				pullBegan <- true
			})
			log.Debug("Pulling image", "image", image, "status", line)
			if strings.Contains(line, "already being pulled by another client. Waiting.") {
				// This can mean the deamon is 'hung' in pulling status for this image, but we can't be sure.
				log.Error("Image 'pull' status marked as already being pulled", "image", image, "status", line)
			}
		}
		if err != nil && err != io.EOF {
			log.Warn("Error reading pull image status", "image", image, "err", err)
		}
	}()
	pullFinished := make(chan error, 1)
	go func() {
		pullFinished <- client.PullImage(opts, authConfig)
		log.Debug("Pulling image complete", "image", image)
	}()

	select {
	case <-pullBegan:
		break
	case err := <-pullFinished:
		if err != nil {
			return DockerContainerMetadata{Error: CannotXContainerError{"Pulled", err.Error()}}
		}
		return DockerContainerMetadata{}
	case <-timeout:
		return DockerContainerMetadata{Error: &DockerTimeoutError{dockerPullBeginTimeout, "pullBegin"}}
	}
	log.Debug("Pull began for image", "image", image)

	select {
	case err := <-pullFinished:
		if err != nil {
			return DockerContainerMetadata{Error: CannotXContainerError{"Pulled", err.Error()}}
		}
		return DockerContainerMetadata{}
	case <-timeout:
		return DockerContainerMetadata{Error: &DockerTimeoutError{dockerPullBeginTimeout, "pullBegin"}}
	}
}
{'fraction_non_alphanumeric': 0.11368543431323701, 'fraction_numerical': 0.00606211592758678, 'mean_word_length': 4.479071883530482, 'pattern_counts': {'":': 2, '<': 18, '<?xml version=': 0, '>': 3, 'https://': 4, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '236200', 'n_tokens_mistral': 4033, 'n_tokens_neox': 3665, 'n_words': 1080}