24920332435

Committed 25 Apr 2026 02:22AM UTC coverage: 43.086% (+0.03%) from 43.054%

Build # 24920332435

Build Type

push

github

Committed by

web-flow

Commit Message

chore(deps): bump go.opentelemetry.io/otel/sdk from 1.40.0 to 1.43.0 (#2096)

Bumps [go.opentelemetry.io/otel/sdk](https://github.com/open-telemetry/opentelemetry-go) from 1.40.0 to 1.43.0.
- [Release notes](https://github.com/open-telemetry/opentelemetry-go/releases)
- [Changelog](https://github.com/open-telemetry/opentelemetry-go/blob/main/CHANGELOG.md)
- [Commits](https://github.com/open-telemetry/opentelemetry-go/compare/v1.40.0...v1.43.0)

---
updated-dependencies:
- dependency-name: go.opentelemetry.io/otel/sdk
  dependency-version: 1.43.0
  dependency-type: indirect
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>

Coverage Stats

5412 of 12561 relevant lines covered (43.09%)

169.68 hits per line

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

9.93

/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/containerd/errdefs"
        "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 !errdefs.IsNotFound(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 errdefs.IsNotFound(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 errdefs.IsNotFound(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 errdefs.IsNotFound(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,
        })

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

tensorchord / envd / 24920332435

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