13090024238

Committed 01 Feb 2025 03:37PM UTC coverage: 41.618% (-1.3%) from 42.886%

Build # 13090024238

Build Type

Pull #1964

github

Committed by

kemingy

Commit Message

feat: use moby context by default

Signed-off-by: Keming <kemingyang@tensorchord.ai>

Pull Request Pull Request #1964: feat: use moby context by default

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

156 existing lines in 10 files now uncovered.

5022 of 12067 relevant lines covered (41.62%)

158.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

9.87

/pkg/driver/docker/docker.go

// Copyright 2023 The envd Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//      http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License 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 docker

import (
        "context"
        "encoding/base64"
        "encoding/json"
        "fmt"
        "io"
        "os"
        "path/filepath"
        "regexp"
        "strconv"
        "strings"
        "time"

        "github.com/cockroachdb/errors"
        "github.com/containers/image/v5/docker/reference"
        "github.com/containers/image/v5/pkg/docker/config"
        "github.com/docker/docker/api/types/container"
        "github.com/docker/docker/api/types/filters"
        dockerimage "github.com/docker/docker/api/types/image"
        "github.com/docker/docker/api/types/mount"
        "github.com/docker/docker/client"
        "github.com/docker/docker/pkg/jsonmessage"
        "github.com/moby/term"
        imagespec "github.com/opencontainers/image-spec/specs-go/v1"
        "github.com/sirupsen/logrus"

        "github.com/tensorchord/envd/pkg/driver"
        "github.com/tensorchord/envd/pkg/envd"
        containerType "github.com/tensorchord/envd/pkg/types"
        "github.com/tensorchord/envd/pkg/util/buildkitutil"
        "github.com/tensorchord/envd/pkg/util/fileutil"
)

const buildkitdConfigPath = "/etc/registry"

var (
        anchoredIdentifierRegexp = regexp.MustCompile(`^([a-f0-9]{64})$`)
        waitingInterval          = 1 * time.Second
)

type dockerClient struct {
        *client.Client
}

func NewClient(ctx context.Context) (driver.Client, error) {
        cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
        if err != nil {
                return nil, err
        }
        _, err = cli.Ping(ctx)
        if err != nil {
                // Special note needed to give users
                if strings.Contains(err.Error(), "permission denied") {
                        err = errors.New(`It seems that current user have no access to docker daemon,
please visit https://docs.docker.com/engine/install/linux-postinstall/ for more info.`)
                }
                return nil, err
        }
        return dockerClient{cli}, nil
}

// Normalize the name accord the spec of docker, It may support normalize image and container in the future.
func NormalizeName(s string) (string, error) {
        if ok := anchoredIdentifierRegexp.MatchString(s); ok {
                return "", errors.Newf("invalid repository name (%s), cannot specify 64-byte hexadecimal strings, please rename it", s)
        }
        var remoteName string
        var tagSep int
        if tagSep = strings.IndexRune(s, ':'); tagSep > -1 {
                remoteName = s[:tagSep]
        } else {
                remoteName = s
        }
        if strings.ToLower(remoteName) != remoteName {
                remoteName = strings.ToLower(remoteName)
                if tagSep > -1 {
                        s = remoteName + s[tagSep:]
                } else {
                        s = remoteName
                }
                logrus.Warnf("The working directory's name is not lowercased: %s, the image built will be lowercased to %s", remoteName, s)
        }
        // remove the spaces
        s = strings.ReplaceAll(s, " ", "")
        name, err := reference.Parse(s)
        if err != nil {
                return "", errors.Wrapf(err, "failed to parse the name '%s', please provide a valid image name", s)
        }
        return name.String(), nil
}

func (c dockerClient) ListImage(ctx context.Context) ([]dockerimage.Summary, error) {
        images, err := c.ImageList(ctx, dockerimage.ListOptions{
                Filters: dockerFilters(false),
        })
        return images, err
}

func (c dockerClient) RemoveImage(ctx context.Context, image string) error {
        _, err := c.ImageRemove(ctx, image, dockerimage.RemoveOptions{})
        if err != nil {
                logrus.WithError(err).Errorf("failed to remove image %s", image)
                return err
        }
        return nil
}

func (c dockerClient) PushImage(ctx context.Context, image string, platform string) error {
        ref, err := reference.ParseNormalizedNamed(image)
        if err != nil {
                return errors.Wrap(err, "failed to normalize the image name")
        }
        auth, err := config.GetCredentialsForRef(nil, ref)
        if err != nil {
                return errors.Wrap(err, "failed to get credentials for image")
        }
        buf, err := json.Marshal(auth)
        if err != nil {
                return errors.Wrap(err, "failed to marshal auth struct")
        }
        platformInfo := strings.Split(platform, "/")
        if len(platformInfo) != 2 {
                return errors.New("invalid platform format, should be <architecture>/<os>")
        }
        reader, err := c.ImagePush(ctx, image, dockerimage.PushOptions{
                RegistryAuth: base64.URLEncoding.EncodeToString(buf),
                Platform: &imagespec.Platform{
                        Architecture: platformInfo[0],
                        OS:           platformInfo[1],
                },
        })
        if err != nil {
                logrus.WithError(err).Errorf("failed to push image %s", image)
                return err
        }

        bar := envd.InitProgressBar(0)

        defer func() {
                reader.Close()
                bar.Finish()
        }()

        decoder := json.NewDecoder(reader)
        stats := new(jsonmessage.JSONMessage)
        for err := decoder.Decode(stats); !errors.Is(err, io.EOF); err = decoder.Decode(stats) {
                if err != nil {
                        return err
                }
                if stats.Error != nil {
                        return stats.Error
                }

                if stats.Status != "" {
                        if stats.ID == "" {
                                bar.UpdateTitle(stats.Status)
                        } else {
                                bar.UpdateTitle(fmt.Sprintf("Pushing image => [%s] %s %s", stats.ID, stats.Status, stats.Progress))
                        }
                }

                stats = new(jsonmessage.JSONMessage)
        }
        return nil
}

func (c dockerClient) GetImage(ctx context.Context, image string) (dockerimage.Summary, error) {
        images, err := c.ImageList(ctx, dockerimage.ListOptions{
                Filters: dockerFiltersWithName(image),
        })
        if err != nil {
                return dockerimage.Summary{}, err
        }
        if len(images) == 0 {
                return dockerimage.Summary{}, errors.Errorf("image %s not found", image)
        }
        return images[0], nil
}

func (c dockerClient) GetImageWithCacheHashLabel(ctx context.Context, image string, hash string) (dockerimage.Summary, error) {
        images, err := c.ImageList(ctx, dockerimage.ListOptions{
                Filters: dockerFiltersWithCacheLabel(image, hash),
        })
        if err != nil {
                return dockerimage.Summary{}, err
        }
        if len(images) == 0 {
                return dockerimage.Summary{}, errors.Errorf("image with hash %s not found", hash)
        }
        return images[0], nil
}

func (c dockerClient) PauseContainer(ctx context.Context, name string) (string, error) {
        logger := logrus.WithField("container", name)
        err := c.ContainerPause(ctx, name)
        if err != nil {
                errCause := errors.UnwrapAll(err).Error()
                switch {
                case strings.Contains(errCause, "is already paused"):
                        logger.Debug("container is already paused, there is no need to pause it again")
                        return "", nil
                case strings.Contains(errCause, "No such container"):
                        logger.Debug("container is not found, there is no need to pause it")
                        return "", errors.New("container not found")
                default:
                        return "", errors.Wrap(err, "failed to pause container")
                }
        }
        return name, nil
}

func (c dockerClient) ResumeContainer(ctx context.Context, name string) (string, error) {
        logger := logrus.WithField("container", name)
        err := c.ContainerUnpause(ctx, name)
        if err != nil {
                errCause := errors.UnwrapAll(err).Error()
                switch {
                case strings.Contains(errCause, "is not paused"):
                        logger.Debug("container is not paused, there is no need to resume")
                        return "", nil
                case strings.Contains(errCause, "No such container"):
                        logger.Debug("container is not found, there is no need to resume it")
                        return "", errors.New("container not found")
                default:
                        return "", errors.Wrap(err, "failed to resume container")
                }
        }
        return name, nil
}

func (c dockerClient) RemoveContainer(ctx context.Context, name string) (string, error) {
        logger := logrus.WithField("container", name)
        err := c.ContainerRemove(ctx, name, container.RemoveOptions{})
        if err != nil {
                errCause := errors.UnwrapAll(err).Error()
                switch {
                case strings.Contains(errCause, "No such container"):
                        logger.Debug("container is not found, there is no need to remove it")
                        return "", errors.New("container not found")
                default:
                        return "", errors.Wrap(err, "failed to remove container")
                }
        }
        return name, nil
}

func (c dockerClient) StartBuildkitd(ctx context.Context, tag, name string, bc *buildkitutil.BuildkitConfig, timeout time.Duration) (string, error) {
        logger := logrus.WithFields(logrus.Fields{
                "tag":             tag,
                "container":       name,
                "buildkit-config": bc,
        })
        logger.Debug("starting buildkitd")
        if _, _, err := c.ImageInspectWithRaw(ctx, tag); err != nil {
                if !client.IsErrNotFound(err) {
                        return "", errors.Wrap(err, "failed to inspect image")
                }

                // Pull the image.
                logger.Debug("pulling image")
                body, err := c.ImagePull(ctx, tag, dockerimage.PullOptions{})
                if err != nil {
                        return "", errors.Wrap(err, "failed to pull image")
                }
                defer body.Close()
                termFd, isTerm := term.GetFdInfo(os.Stdout)
                err = jsonmessage.DisplayJSONMessagesStream(body, os.Stdout, termFd, isTerm, nil)
                if err != nil {
                        logger.WithError(err).Warningln("failed to display image pull output")
                }
        }
        config := &container.Config{
                Image: tag,
        }
        hostConfig := &container.HostConfig{
                Privileged: true,
                AutoRemove: true,
                Mounts: []mount.Mount{
                        {
                                Type:   mount.TypeBind,
                                Source: fileutil.DefaultConfigDir,
                                Target: buildkitdConfigPath,
                        },
                },
        }

        err := bc.Save()
        if err != nil {
                return "", errors.Wrap(err, "failed to generate buildkit config")
        }
        config.Entrypoint = []string{
                "buildkitd", "--config", filepath.Join(buildkitdConfigPath, "buildkitd.toml"),
        }
        created, _ := c.Exists(ctx, name)
        if created {
                status, err := c.GetStatus(ctx, name)
                if err != nil {
                        return name, errors.Wrap(err, "failed to get container status")
                }

                err = c.handleContainerCreated(ctx, name, status, timeout)
                if err != nil {
                        return name, errors.Wrap(err, "failed to handle container created condition")
                }

                // When status is StatusDead/StatusRemoving, we need to create and start the container later(not to return directly).
                if status != containerType.StatusDead && status != containerType.StatusRemoving {
                        return name, nil
                }
        }
        resp, err := c.ContainerCreate(ctx, config, hostConfig, nil, nil, name)
        if err != nil {
                return "", errors.Wrap(err, "failed to create container")
        }

        for _, w := range resp.Warnings {
                logger.Warnf("run with warnings: %s", w)
        }

        if err := c.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {
                return "", errors.Wrap(err, "failed to start container")
        }

        container, err := c.ContainerInspect(ctx, resp.ID)
        if err != nil {
                return "", errors.Wrap(err, "failed to inspect container")
        }

        err = c.waitUntilRunning(ctx, container.Name, timeout)
        if err != nil {
                return "", err
        }

        return container.Name, nil
}

func (c dockerClient) Exists(ctx context.Context, cname string) (bool, error) {
        _, err := c.ContainerInspect(ctx, cname)
        if err != nil {
                if client.IsErrNotFound(err) {
                        return false, nil
                }
                return false, err
        }
        return true, nil
}

func (c dockerClient) IsRunning(ctx context.Context, cname string) (bool, error) {
        container, err := c.ContainerInspect(ctx, cname)
        if err != nil {
                if client.IsErrNotFound(err) {
                        return false, nil
                }
                return false, err
        }
        return container.State.Running, nil
}

func (c dockerClient) GetStatus(ctx context.Context, cname string) (containerType.ContainerStatus, error) {
        container, err := c.ContainerInspect(ctx, cname)
        if err != nil {
                if client.IsErrNotFound(err) {
                        return "", nil
                }
                return "", err
        }
        return containerType.ContainerStatus(container.State.Status), nil
}

// Load loads the docker image from the reader into the docker host.
// It's up to the caller to close the io.ReadCloser.
func (c dockerClient) Load(ctx context.Context, r io.ReadCloser, quiet bool) error {
        resp, err := c.ImageLoad(ctx, r, quiet)
        if err != nil {
                return err
        }

        defer resp.Body.Close()
        return nil
}

func (c dockerClient) Exec(ctx context.Context, cname string, cmd []string) error {
        execConfig := container.ExecOptions{
                Cmd:    cmd,
                Detach: true,
        }
        resp, err := c.ContainerExecCreate(ctx, cname, execConfig)
        if err != nil {
                return err
        }
        execID := resp.ID
        return c.ContainerExecStart(ctx, execID, container.ExecStartOptions{
                Detach: true,
        })
}

func (c dockerClient) PruneImage(ctx context.Context) (dockerimage.PruneReport, error) {
        pruneReport, err := c.ImagesPrune(ctx, filters.Args{})
        if err != nil {
                return dockerimage.PruneReport{}, errors.Wrap(err, "failed to prune images")
        }
        return pruneReport, nil
}

func (c dockerClient) Stats(ctx context.Context, cname string, statChan chan<- *driver.Stats, done <-chan bool) (retErr error) {
        errC := make(chan error, 1)
        containerStats, err := c.ContainerStats(ctx, cname, true)
        readCloser := containerStats.Body
        quit := make(chan struct{})
        defer func() {
                close(statChan)
                close(quit)

                if err := <-errC; err != nil && retErr == nil {
                        retErr = err
                }

                if err := readCloser.Close(); err != nil && retErr == nil {
                        retErr = err
                }
        }()

        go func() {
                // block here waiting for the signal to stop function
                select {
                case <-done:
                        readCloser.Close()
                case <-quit:
                        return
                }
        }()

        if err != nil {
                return err
        }
        decoder := json.NewDecoder(readCloser)
        stats := new(driver.Stats)
        for err := decoder.Decode(stats); !errors.Is(err, io.EOF); err = decoder.Decode(stats) {
                if err != nil {
                        return err
                }
                statChan <- stats
                stats = new(driver.Stats)
        }
        return nil
}

func (c dockerClient) waitUntilRunning(ctx context.Context,
        name string, timeout time.Duration) error {
        logger := logrus.WithField("container", name)
        logger.Debug("waiting to start")

        // First, wait for the container to be marked as started.
        ctxTimeout, cancel := context.WithTimeout(ctx, timeout)
        defer cancel()
        for {
                select {
                case <-time.After(waitingInterval):
                        isRunning, err := c.IsRunning(ctxTimeout, name)
                        if err != nil {
                                // Has not yet started. Keep waiting.
                                return errors.Wrap(err, "failed to check if container is running")
                        }
                        if isRunning {
                                logger.Debug("the container is running")
                                return nil
                        }

                case <-ctxTimeout.Done():
                        container, err := c.ContainerInspect(ctx, name)
                        if err != nil {
                                logger.Debugf("failed to inspect container %s", name)
                        }
                        state, err := json.Marshal(container.State)
                        if err != nil {
                                logger.Debug("failed to marshal container state")
                        }
                        logger.Debugf("container state: %s", state)
                        return errors.Errorf("timeout %s: container did not start", timeout)
                }
        }
}

func (c dockerClient) waitUntilRemoved(ctx context.Context,
        name string, timeout time.Duration) error {
        logger := logrus.WithField("container", name)
        logger.Debug("waiting to be removed")

        // Wait for the container to be removed
        ctxTimeout, cancel := context.WithTimeout(ctx, timeout)
        defer cancel()
        for {
                select {
                case <-time.After(waitingInterval):
                        exist, err := c.Exists(ctxTimeout, name)
                        if err != nil {
                                return errors.Wrap(err, "failed to check if container has been removed")
                        }
                        if !exist {
                                logger.Debug("the container has been removed")
                                return nil
                        }
                case <-ctxTimeout.Done():
                        container, err := c.ContainerInspect(ctx, name)
                        if err != nil {
                                logger.Debugf("failed to inspect container %s", name)
                        }
                        state, err := json.Marshal(container.State)
                        if err != nil {
                                logger.Debug("failed to marshal container state")
                        }
                        logger.Debugf("container state: %s", state)
                        return errors.Errorf("timeout %s: container can't be removed", timeout)
                }
        }
}

func (c dockerClient) handleContainerCreated(ctx context.Context,
        cname string, status containerType.ContainerStatus, timeout time.Duration) error {
        logger := logrus.WithFields(logrus.Fields{
                "container": cname,
                "status":    status,
        })

        if status == containerType.StatusPaused {
                logger.Info("container was paused, unpause it now...")
                _, err := c.ResumeContainer(ctx, cname)
                if err != nil {
                        logger.WithError(err).Error("can not run buildkitd")
                        return errors.Wrap(err, "failed to unpause container")
                }
        } else if status == containerType.StatusExited {
                logger.Info("container exited, try to start it...")
                err := c.ContainerStart(ctx, cname, container.StartOptions{})
                if err != nil {
                        logger.WithError(err).Error("can not run buildkitd")
                        return errors.Wrap(err, "failed to start exited cotaniner")
                }
        } else if status == containerType.StatusDead {
                logger.Info("container is dead, try to remove it...")
                err := c.ContainerRemove(ctx, cname, container.RemoveOptions{})
                if err != nil {
                        logger.WithError(err).Error("can not run buildkitd")
                        return errors.Wrap(err, "failed to remove container")
                }
        } else if status == containerType.StatusCreated {
                logger.Info("container is being created")
                err := c.waitUntilRunning(ctx, cname, timeout)
                if err != nil {
                        logger.WithError(err).Error("can not run buildkitd")
                        return errors.Wrap(err, "failed to start container")
                }
        } else if status == containerType.StatusRemoving {
                logger.Info("container is being removed.")
                err := c.waitUntilRemoved(ctx, cname, timeout)
                if err != nil {
                        logger.WithError(err).Error("can not run buildkitd")
                        return errors.Wrap(err, "failed to remove container")
                }
        }
        // No process for StatusRunning

        return nil
}

func GetDockerVersion() (int, error) {

        ctx := context.Background()
        cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())
        if err != nil {
                return -1, err
        }
        defer cli.Close()

        info, err := cli.Info(ctx)
        if err != nil {
                return -1, err
        }
        version, err := strconv.Atoi(strings.Split(info.ServerVersion, ".")[0])
        if err != nil {
                return -1, err
        }
        return version, nil
}

1	// Copyright 2023 The envd Authors
2	//
3	// Licensed under the Apache License, Version 2.0 (the "License");
4	// you may not use this file except in compliance with the License.
5	// You may obtain a copy of the License at
6	//
7	// http://www.apache.org/licenses/LICENSE-2.0
8	//
9	// Unless required by applicable law or agreed to in writing, software
10	// distributed under the License is distributed on an "AS IS" BASIS,
11	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	// See the License for the specific language governing permissions and
13	// limitations under the License.
14
15	package docker
16
17	import (
18	"context"
19	"encoding/base64"
20	"encoding/json"
21	"fmt"
22	"io"
23	"os"
24	"path/filepath"
25	"regexp"
26	"strconv"
27	"strings"
28	"time"
29
30	"github.com/cockroachdb/errors"
31	"github.com/containers/image/v5/docker/reference"
32	"github.com/containers/image/v5/pkg/docker/config"
33	"github.com/docker/docker/api/types/container"
34	"github.com/docker/docker/api/types/filters"
35	dockerimage "github.com/docker/docker/api/types/image"
36	"github.com/docker/docker/api/types/mount"
37	"github.com/docker/docker/client"
38	"github.com/docker/docker/pkg/jsonmessage"
39	"github.com/moby/term"
40	imagespec "github.com/opencontainers/image-spec/specs-go/v1"
41	"github.com/sirupsen/logrus"
42
43	"github.com/tensorchord/envd/pkg/driver"
44	"github.com/tensorchord/envd/pkg/envd"
45	containerType "github.com/tensorchord/envd/pkg/types"
46	"github.com/tensorchord/envd/pkg/util/buildkitutil"
47	"github.com/tensorchord/envd/pkg/util/fileutil"
48	)
49
50	const buildkitdConfigPath = "/etc/registry"
51
52	var (
53	anchoredIdentifierRegexp = regexp.MustCompile(`^([a-f0-9]{64})$`)
54	waitingInterval = 1 * time.Second
55	)
56
57	type dockerClient struct {
58	*client.Client
59	}
60
61	func NewClient(ctx context.Context) (driver.Client, error) {	13✔
62	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())	13✔
63	if err != nil {	13✔
64	return nil, err	×
65	}	×
66	_, err = cli.Ping(ctx)	13✔
67	if err != nil {	14✔
68	// Special note needed to give users	1✔
69	if strings.Contains(err.Error(), "permission denied") {	1✔
70	err = errors.New(`It seems that current user have no access to docker daemon,	×
71	please visit https://docs.docker.com/engine/install/linux-postinstall/ for more info.`)	×
72	}	×
73	return nil, err	1✔
74	}
75	return dockerClient{cli}, nil	12✔
76	}
77
78	// Normalize the name accord the spec of docker, It may support normalize image and container in the future.
79	func NormalizeName(s string) (string, error) {	25✔
80	if ok := anchoredIdentifierRegexp.MatchString(s); ok {	25✔
81	return "", errors.Newf("invalid repository name (%s), cannot specify 64-byte hexadecimal strings, please rename it", s)	×
82	}	×
83	var remoteName string	25✔
84	var tagSep int	25✔
85	if tagSep = strings.IndexRune(s, ':'); tagSep > -1 {	50✔
86	remoteName = s[:tagSep]	25✔
87	} else {	25✔
88	remoteName = s	×
89	}	×
90	if strings.ToLower(remoteName) != remoteName {	26✔
91	remoteName = strings.ToLower(remoteName)	1✔
92	if tagSep > -1 {	2✔
93	s = remoteName + s[tagSep:]	1✔
94	} else {	1✔
95	s = remoteName	×
96	}	×
97	logrus.Warnf("The working directory's name is not lowercased: %s, the image built will be lowercased to %s", remoteName, s)	1✔
98	}
99	// remove the spaces
100	s = strings.ReplaceAll(s, " ", "")	25✔
101	name, err := reference.Parse(s)	25✔
102	if err != nil {	25✔
103	return "", errors.Wrapf(err, "failed to parse the name '%s', please provide a valid image name", s)	×
104	}	×
105	return name.String(), nil	25✔
106	}
107
108	func (c dockerClient) ListImage(ctx context.Context) ([]dockerimage.Summary, error) {	×
109	images, err := c.ImageList(ctx, dockerimage.ListOptions{	×
110	Filters: dockerFilters(false),	×
111	})	×
112	return images, err	×
113	}	×
114
115	func (c dockerClient) RemoveImage(ctx context.Context, image string) error {	1✔
116	_, err := c.ImageRemove(ctx, image, dockerimage.RemoveOptions{})	1✔
117	if err != nil {	1✔
118	logrus.WithError(err).Errorf("failed to remove image %s", image)	×
119	return err	×
120	}	×
121	return nil	1✔
122	}
123
124	func (c dockerClient) PushImage(ctx context.Context, image string, platform string) error {	×
125	ref, err := reference.ParseNormalizedNamed(image)	×
126	if err != nil {	×
127	return errors.Wrap(err, "failed to normalize the image name")	×
128	}	×
129	auth, err := config.GetCredentialsForRef(nil, ref)	×
130	if err != nil {	×
131	return errors.Wrap(err, "failed to get credentials for image")	×
132	}	×
133	buf, err := json.Marshal(auth)	×
134	if err != nil {	×
135	return errors.Wrap(err, "failed to marshal auth struct")	×
136	}	×
137	platformInfo := strings.Split(platform, "/")	×
138	if len(platformInfo) != 2 {	×
139	return errors.New("invalid platform format, should be <architecture>/<os>")	×
140	}	×
141	reader, err := c.ImagePush(ctx, image, dockerimage.PushOptions{	×
142	RegistryAuth: base64.URLEncoding.EncodeToString(buf),	×
143	Platform: &imagespec.Platform{	×
144	Architecture: platformInfo[0],	×
145	OS: platformInfo[1],	×
146	},	×
147	})	×
148	if err != nil {	×
149	logrus.WithError(err).Errorf("failed to push image %s", image)	×
150	return err	×
151	}	×
152
153	bar := envd.InitProgressBar(0)	×
154		×
155	defer func() {	×
156	reader.Close()	×
157	bar.Finish()	×
158	}()	×
159
160	decoder := json.NewDecoder(reader)	×
161	stats := new(jsonmessage.JSONMessage)	×
162	for err := decoder.Decode(stats); !errors.Is(err, io.EOF); err = decoder.Decode(stats) {	×
163	if err != nil {	×
164	return err	×
165	}	×
166	if stats.Error != nil {	×
167	return stats.Error	×
168	}	×
169
170	if stats.Status != "" {	×
171	if stats.ID == "" {	×
172	bar.UpdateTitle(stats.Status)	×
173	} else {	×
174	bar.UpdateTitle(fmt.Sprintf("Pushing image => [%s] %s %s", stats.ID, stats.Status, stats.Progress))	×
175	}	×
176	}
177
178	stats = new(jsonmessage.JSONMessage)	×
179	}
180	return nil	×
181	}
182
183	func (c dockerClient) GetImage(ctx context.Context, image string) (dockerimage.Summary, error) {	×
184	images, err := c.ImageList(ctx, dockerimage.ListOptions{	×
185	Filters: dockerFiltersWithName(image),	×
186	})	×
187	if err != nil {	×
188	return dockerimage.Summary{}, err	×
189	}	×
190	if len(images) == 0 {	×
191	return dockerimage.Summary{}, errors.Errorf("image %s not found", image)	×
192	}	×
193	return images[0], nil	×
194	}
195
196	func (c dockerClient) GetImageWithCacheHashLabel(ctx context.Context, image string, hash string) (dockerimage.Summary, error) {	5✔
197	images, err := c.ImageList(ctx, dockerimage.ListOptions{	5✔
198	Filters: dockerFiltersWithCacheLabel(image, hash),	5✔
199	})	5✔
200	if err != nil {	5✔
201	return dockerimage.Summary{}, err	×
202	}	×
203	if len(images) == 0 {	8✔
204	return dockerimage.Summary{}, errors.Errorf("image with hash %s not found", hash)	3✔
205	}	3✔
206	return images[0], nil	2✔
207	}
208
209	func (c dockerClient) PauseContainer(ctx context.Context, name string) (string, error) {	×
210	logger := logrus.WithField("container", name)	×
211	err := c.ContainerPause(ctx, name)	×
212	if err != nil {	×
213	errCause := errors.UnwrapAll(err).Error()	×
214	switch {	×
215	case strings.Contains(errCause, "is already paused"):	×
216	logger.Debug("container is already paused, there is no need to pause it again")	×
217	return "", nil	×
218	case strings.Contains(errCause, "No such container"):	×
219	logger.Debug("container is not found, there is no need to pause it")	×
220	return "", errors.New("container not found")	×
221	default:	×
222	return "", errors.Wrap(err, "failed to pause container")	×
223	}
224	}
225	return name, nil	×
226	}
227
228	func (c dockerClient) ResumeContainer(ctx context.Context, name string) (string, error) {	×
229	logger := logrus.WithField("container", name)	×
230	err := c.ContainerUnpause(ctx, name)	×
231	if err != nil {	×
232	errCause := errors.UnwrapAll(err).Error()	×
233	switch {	×
234	case strings.Contains(errCause, "is not paused"):	×
235	logger.Debug("container is not paused, there is no need to resume")	×
236	return "", nil	×
237	case strings.Contains(errCause, "No such container"):	×
238	logger.Debug("container is not found, there is no need to resume it")	×
239	return "", errors.New("container not found")	×
240	default:	×
241	return "", errors.Wrap(err, "failed to resume container")	×
242	}
243	}
244	return name, nil	×
245	}
246
247	func (c dockerClient) RemoveContainer(ctx context.Context, name string) (string, error) {	×
248	logger := logrus.WithField("container", name)	×
249	err := c.ContainerRemove(ctx, name, container.RemoveOptions{})	×
250	if err != nil {	×
251	errCause := errors.UnwrapAll(err).Error()	×
252	switch {	×
253	case strings.Contains(errCause, "No such container"):	×
254	logger.Debug("container is not found, there is no need to remove it")	×
255	return "", errors.New("container not found")	×
256	default:	×
257	return "", errors.Wrap(err, "failed to remove container")	×
258	}
259	}
260	return name, nil	×
261	}
262
UNCOV 263	func (c dockerClient) StartBuildkitd(ctx context.Context, tag, name string, bc *buildkitutil.BuildkitConfig, timeout time.Duration) (string, error) {	×
UNCOV 264	logger := logrus.WithFields(logrus.Fields{	×
UNCOV 265	"tag": tag,	×
UNCOV 266	"container": name,	×
UNCOV 267	"buildkit-config": bc,	×
UNCOV 268	})	×
UNCOV 269	logger.Debug("starting buildkitd")	×
UNCOV 270	if _, _, err := c.ImageInspectWithRaw(ctx, tag); err != nil {	×
UNCOV 271	if !client.IsErrNotFound(err) {	×
272	return "", errors.Wrap(err, "failed to inspect image")	×
273	}	×
274
275	// Pull the image.
UNCOV 276	logger.Debug("pulling image")	×
UNCOV 277	body, err := c.ImagePull(ctx, tag, dockerimage.PullOptions{})	×
UNCOV 278	if err != nil {	×
279	return "", errors.Wrap(err, "failed to pull image")	×
280	}	×
UNCOV 281	defer body.Close()	×
UNCOV 282	termFd, isTerm := term.GetFdInfo(os.Stdout)	×
UNCOV 283	err = jsonmessage.DisplayJSONMessagesStream(body, os.Stdout, termFd, isTerm, nil)	×
UNCOV 284	if err != nil {	×
285	logger.WithError(err).Warningln("failed to display image pull output")	×
286	}	×
287	}
UNCOV 288	config := &container.Config{	×
UNCOV 289	Image: tag,	×
UNCOV 290	}	×
UNCOV 291	hostConfig := &container.HostConfig{	×
UNCOV 292	Privileged: true,	×
UNCOV 293	AutoRemove: true,	×
UNCOV 294	Mounts: []mount.Mount{	×
UNCOV 295	{	×
UNCOV 296	Type: mount.TypeBind,	×
UNCOV 297	Source: fileutil.DefaultConfigDir,	×
UNCOV 298	Target: buildkitdConfigPath,	×
UNCOV 299	},	×
UNCOV 300	},	×
UNCOV 301	}	×
UNCOV 302		×
UNCOV 303	err := bc.Save()	×
UNCOV 304	if err != nil {	×
305	return "", errors.Wrap(err, "failed to generate buildkit config")	×
306	}	×
UNCOV 307	config.Entrypoint = []string{	×
UNCOV 308	"buildkitd", "--config", filepath.Join(buildkitdConfigPath, "buildkitd.toml"),	×
UNCOV 309	}	×
UNCOV 310	created, _ := c.Exists(ctx, name)	×
UNCOV 311	if created {	×
UNCOV 312	status, err := c.GetStatus(ctx, name)	×
UNCOV 313	if err != nil {	×
314	return name, errors.Wrap(err, "failed to get container status")	×
315	}	×
316
UNCOV 317	err = c.handleContainerCreated(ctx, name, status, timeout)	×
UNCOV 318	if err != nil {	×
319	return name, errors.Wrap(err, "failed to handle container created condition")	×
320	}	×
321
322	// When status is StatusDead/StatusRemoving, we need to create and start the container later(not to return directly).
UNCOV 323	if status != containerType.StatusDead && status != containerType.StatusRemoving {	×
UNCOV 324	return name, nil	×
UNCOV 325	}	×
326	}
UNCOV 327	resp, err := c.ContainerCreate(ctx, config, hostConfig, nil, nil, name)	×
UNCOV 328	if err != nil {	×
329	return "", errors.Wrap(err, "failed to create container")	×
330	}	×
331
UNCOV 332	for _, w := range resp.Warnings {	×
333	logger.Warnf("run with warnings: %s", w)	×
334	}	×
335
UNCOV 336	if err := c.ContainerStart(ctx, resp.ID, container.StartOptions{}); err != nil {	×
337	return "", errors.Wrap(err, "failed to start container")	×
338	}	×
339
UNCOV 340	container, err := c.ContainerInspect(ctx, resp.ID)	×
UNCOV 341	if err != nil {	×
342	return "", errors.Wrap(err, "failed to inspect container")	×
343	}	×
344
UNCOV 345	err = c.waitUntilRunning(ctx, container.Name, timeout)	×
UNCOV 346	if err != nil {	×
347	return "", err	×
348	}	×
349
UNCOV 350	return container.Name, nil	×
351	}
352
UNCOV 353	func (c dockerClient) Exists(ctx context.Context, cname string) (bool, error) {	×
UNCOV 354	_, err := c.ContainerInspect(ctx, cname)	×
UNCOV 355	if err != nil {	×
UNCOV 356	if client.IsErrNotFound(err) {	×
UNCOV 357	return false, nil	×
UNCOV 358	}	×
359	return false, err	×
360	}
UNCOV 361	return true, nil	×
362	}
363
UNCOV 364	func (c dockerClient) IsRunning(ctx context.Context, cname string) (bool, error) {	×
UNCOV 365	container, err := c.ContainerInspect(ctx, cname)	×
UNCOV 366	if err != nil {	×
367	if client.IsErrNotFound(err) {	×
368	return false, nil	×
369	}	×
370	return false, err	×
371	}
UNCOV 372	return container.State.Running, nil	×
373	}
374
UNCOV 375	func (c dockerClient) GetStatus(ctx context.Context, cname string) (containerType.ContainerStatus, error) {	×
UNCOV 376	container, err := c.ContainerInspect(ctx, cname)	×
UNCOV 377	if err != nil {	×
378	if client.IsErrNotFound(err) {	×
379	return "", nil	×
380	}	×
381	return "", err	×
382	}
UNCOV 383	return containerType.ContainerStatus(container.State.Status), nil	×
384	}
385
386	// Load loads the docker image from the reader into the docker host.
387	// It's up to the caller to close the io.ReadCloser.
388	func (c dockerClient) Load(ctx context.Context, r io.ReadCloser, quiet bool) error {	2✔
389	resp, err := c.ImageLoad(ctx, r, quiet)	2✔
390	if err != nil {	2✔
391	return err	×
392	}	×
393
394	defer resp.Body.Close()	2✔
395	return nil	2✔
396	}
397
398	func (c dockerClient) Exec(ctx context.Context, cname string, cmd []string) error {	×
399	execConfig := container.ExecOptions{	×
400	Cmd: cmd,	×
401	Detach: true,	×
402	}	×
403	resp, err := c.ContainerExecCreate(ctx, cname, execConfig)	×
404	if err != nil {	×
405	return err	×
406	}	×
407	execID := resp.ID	×
408	return c.ContainerExecStart(ctx, execID, container.ExecStartOptions{	×
409	Detach: true,	×
410	})	×
411	}
412
413	func (c dockerClient) PruneImage(ctx context.Context) (dockerimage.PruneReport, error) {	×
414	pruneReport, err := c.ImagesPrune(ctx, filters.Args{})	×
415	if err != nil {	×
416	return dockerimage.PruneReport{}, errors.Wrap(err, "failed to prune images")	×
417	}	×
418	return pruneReport, nil	×
419	}
420
421	func (c dockerClient) Stats(ctx context.Context, cname string, statChan chan<- *driver.Stats, done <-chan bool) (retErr error) {	×
422	errC := make(chan error, 1)	×
423	containerStats, err := c.ContainerStats(ctx, cname, true)	×
424	readCloser := containerStats.Body	×
425	quit := make(chan struct{})	×
426	defer func() {	×
427	close(statChan)	×
428	close(quit)	×
429		×
430	if err := <-errC; err != nil && retErr == nil {	×
431	retErr = err	×
432	}	×
433
434	if err := readCloser.Close(); err != nil && retErr == nil {	×
435	retErr = err	×
436	}	×
437	}()
438
439	go func() {	×
440	// block here waiting for the signal to stop function	×
441	select {	×
442	case <-done:	×
443	readCloser.Close()	×
444	case <-quit:	×
445	return	×
446	}
447	}()
448
449	if err != nil {	×
450	return err	×
451	}	×
452	decoder := json.NewDecoder(readCloser)	×
453	stats := new(driver.Stats)	×
454	for err := decoder.Decode(stats); !errors.Is(err, io.EOF); err = decoder.Decode(stats) {	×
455	if err != nil {	×
456	return err	×
457	}	×
458	statChan <- stats	×
459	stats = new(driver.Stats)	×
460	}
461	return nil	×
462	}
463
464	func (c dockerClient) waitUntilRunning(ctx context.Context,
UNCOV 465	name string, timeout time.Duration) error {	×
UNCOV 466	logger := logrus.WithField("container", name)	×
UNCOV 467	logger.Debug("waiting to start")	×
UNCOV 468		×
UNCOV 469	// First, wait for the container to be marked as started.	×
UNCOV 470	ctxTimeout, cancel := context.WithTimeout(ctx, timeout)	×
UNCOV 471	defer cancel()	×
UNCOV 472	for {	×
UNCOV 473	select {	×
UNCOV 474	case <-time.After(waitingInterval):	×
UNCOV 475	isRunning, err := c.IsRunning(ctxTimeout, name)	×
UNCOV 476	if err != nil {	×
477	// Has not yet started. Keep waiting.	×
478	return errors.Wrap(err, "failed to check if container is running")	×
479	}	×
UNCOV 480	if isRunning {	×
UNCOV 481	logger.Debug("the container is running")	×
UNCOV 482	return nil	×
UNCOV 483	}	×
484
485	case <-ctxTimeout.Done():	×
486	container, err := c.ContainerInspect(ctx, name)	×
487	if err != nil {	×
488	logger.Debugf("failed to inspect container %s", name)	×
489	}	×
490	state, err := json.Marshal(container.State)	×
491	if err != nil {	×
492	logger.Debug("failed to marshal container state")	×
493	}	×
494	logger.Debugf("container state: %s", state)	×
495	return errors.Errorf("timeout %s: container did not start", timeout)	×
496	}
497	}
498	}
499
500	func (c dockerClient) waitUntilRemoved(ctx context.Context,
501	name string, timeout time.Duration) error {	×
502	logger := logrus.WithField("container", name)	×
503	logger.Debug("waiting to be removed")	×
504		×
505	// Wait for the container to be removed	×
506	ctxTimeout, cancel := context.WithTimeout(ctx, timeout)	×
507	defer cancel()	×
508	for {	×
509	select {	×
510	case <-time.After(waitingInterval):	×
511	exist, err := c.Exists(ctxTimeout, name)	×
512	if err != nil {	×
513	return errors.Wrap(err, "failed to check if container has been removed")	×
514	}	×
515	if !exist {	×
516	logger.Debug("the container has been removed")	×
517	return nil	×
518	}	×
519	case <-ctxTimeout.Done():	×
520	container, err := c.ContainerInspect(ctx, name)	×
521	if err != nil {	×
522	logger.Debugf("failed to inspect container %s", name)	×
523	}	×
524	state, err := json.Marshal(container.State)	×
525	if err != nil {	×
526	logger.Debug("failed to marshal container state")	×
527	}	×
528	logger.Debugf("container state: %s", state)	×
529	return errors.Errorf("timeout %s: container can't be removed", timeout)	×
530	}
531	}
532	}
533
534	func (c dockerClient) handleContainerCreated(ctx context.Context,
UNCOV 535	cname string, status containerType.ContainerStatus, timeout time.Duration) error {	×
UNCOV 536	logger := logrus.WithFields(logrus.Fields{	×
UNCOV 537	"container": cname,	×
UNCOV 538	"status": status,	×
UNCOV 539	})	×
UNCOV 540		×
UNCOV 541	if status == containerType.StatusPaused {	×
542	logger.Info("container was paused, unpause it now...")	×
543	_, err := c.ResumeContainer(ctx, cname)	×
544	if err != nil {	×
545	logger.WithError(err).Error("can not run buildkitd")	×
546	return errors.Wrap(err, "failed to unpause container")	×
547	}	×
UNCOV 548	} else if status == containerType.StatusExited {	×
549	logger.Info("container exited, try to start it...")	×
550	err := c.ContainerStart(ctx, cname, container.StartOptions{})	×
551	if err != nil {	×
552	logger.WithError(err).Error("can not run buildkitd")	×
553	return errors.Wrap(err, "failed to start exited cotaniner")	×
554	}	×
UNCOV 555	} else if status == containerType.StatusDead {	×
556	logger.Info("container is dead, try to remove it...")	×
557	err := c.ContainerRemove(ctx, cname, container.RemoveOptions{})	×
558	if err != nil {	×
559	logger.WithError(err).Error("can not run buildkitd")	×
560	return errors.Wrap(err, "failed to remove container")	×
561	}	×
UNCOV 562	} else if status == containerType.StatusCreated {	×
563	logger.Info("container is being created")	×
564	err := c.waitUntilRunning(ctx, cname, timeout)	×
565	if err != nil {	×
566	logger.WithError(err).Error("can not run buildkitd")	×
567	return errors.Wrap(err, "failed to start container")	×
568	}	×
UNCOV 569	} else if status == containerType.StatusRemoving {	×
570	logger.Info("container is being removed.")	×
571	err := c.waitUntilRemoved(ctx, cname, timeout)	×
572	if err != nil {	×
573	logger.WithError(err).Error("can not run buildkitd")	×
574	return errors.Wrap(err, "failed to remove container")	×
575	}	×
576	}
577	// No process for StatusRunning
578
UNCOV 579	return nil	×
580	}
581
582	func GetDockerVersion() (int, error) {	×
583		×
584	ctx := context.Background()	×
585	cli, err := client.NewClientWithOpts(client.FromEnv, client.WithAPIVersionNegotiation())	×
586	if err != nil {	×
587	return -1, err	×
588	}	×
589	defer cli.Close()	×
590		×
591	info, err := cli.Info(ctx)	×
592	if err != nil {	×
593	return -1, err	×
594	}	×
595	version, err := strconv.Atoi(strings.Split(info.ServerVersion, ".")[0])	×
596	if err != nil {	×
597	return -1, err	×
598	}	×
599	return version, nil	×
600	}

tensorchord / envd / 13090024238

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous