13754521523

Committed 10 Mar 2025 01:00AM UTC coverage: 42.542% (+0.4%) from 42.15%

Build # 13754521523

Build Type

Pull #1992

github

Committed by

kemingy

Commit Message

fix lint

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

Pull Request Pull Request #1992: chore: bump dep version, check dep monthly

Run Details

26 of 32 new or added lines in 5 files covered. (81.25%)

2 existing lines in 1 file now uncovered.

5157 of 12122 relevant lines covered (42.54%)

158.91 hits per line

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

9.84

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

tensorchord / envd / 13754521523

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