Display docker pull progress

runner/go.mod

@@ -17,13 +17,16 @@ require (
 	github.com/bluekeyes/go-gitdiff v0.6.0
 	github.com/docker/docker v20.10.6+incompatible
 	github.com/docker/go-connections v0.4.0
+	github.com/dustin/go-humanize v1.0.1
 	github.com/go-git/go-git/v5 v5.4.2
 	github.com/libp2p/go-reuseport v0.3.0
+	github.com/opencontainers/go-digest v1.0.0
 	github.com/sirupsen/logrus v1.8.1
 	github.com/stretchr/testify v1.8.1
 	github.com/urfave/cli/v2 v2.3.0
 	go.uber.org/atomic v1.4.0
 	golang.org/x/crypto v0.6.0
+	golang.org/x/sync v0.1.0
 	gopkg.in/yaml.v2 v2.4.0
 )
 
@@ -34,6 +37,7 @@ require (
 	cloud.google.com/go/longrunning v0.3.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/internal v1.2.0 // indirect
 	github.com/Azure/azure-sdk-for-go/sdk/keyvault/internal v0.7.0 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v0.9.0 // indirect
 	github.com/Microsoft/go-winio v0.4.17 // indirect
 	github.com/ProtonMail/go-crypto v0.0.0-20210428141323-04723f9f07d7 // indirect
@@ -50,7 +54,6 @@ require (
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/docker/distribution v2.7.1+incompatible // indirect
 	github.com/docker/go-units v0.4.0 // indirect
-	github.com/dustin/go-humanize v1.0.1 // indirect
 	github.com/emirpasic/gods v1.12.0 // indirect
 	github.com/go-git/gcfg v1.5.0 // indirect
 	github.com/go-git/go-billy/v5 v5.3.1 // indirect
@@ -74,7 +77,6 @@ require (
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/mattn/go-isatty v0.0.16 // indirect
 	github.com/mitchellh/go-homedir v1.1.0 // indirect
-	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/pkg/browser v0.0.0-20210911075715-681adbf594b8 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
@@ -89,7 +91,6 @@ require (
 	golang.org/x/mod v0.8.0 // indirect
 	golang.org/x/net v0.7.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect
-	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.5.0 // indirect
 	golang.org/x/text v0.7.0 // indirect
 	golang.org/x/tools v0.6.0 // indirect
@@ -135,5 +136,3 @@ require (
 	gopkg.in/yaml.v3 v3.0.1
 	modernc.org/sqlite v1.20.3
 )
-
-replace github.com/kahing/goofys => github.com/dstackai/goofys v0.24.1-0.20211210032445-aae1cc43d188

runner/go.sum

@@ -305,6 +305,7 @@ github.com/cpuguy83/go-md2man/v2 v2.0.0 h1:EoUDS0afbrsXAZ9YQ9jdu/mZ2sXgT1/2yyNng
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7DoTY=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
+github.com/creack/pty v1.1.11 h1:07n33Z8lZxZ2qwegKbObQohDhXDQxiMMz1NOUGYlesw=
 github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
 github.com/cyphar/filepath-securejoin v0.2.2/go.mod h1:FpkQEhXnPnOthhzymB7CGsFk2G9VLXONKD9G7QGMM+4=
 github.com/d2g/dhcp4 v0.0.0-20170904100407-a1d1b6c41b1c/go.mod h1:Ct2BUK8SB0YC1SMSibvLzxjeJLnrYEVLULFNiHY9YfQ=
@@ -338,7 +339,6 @@ github.com/docker/libtrust v0.0.0-20150114040149-fa567046d9b1/go.mod h1:cyGadeNE
 github.com/docker/spdystream v0.0.0-20160310174837-449fdfce4d96/go.mod h1:Qh8CwZgvJUkLughtfhJv5dyTYa91l1fOUCrgjqmcifM=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
 github.com/dustin/go-humanize v0.0.0-20171111073723-bb3d318650d4/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
-github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
 github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=

runner/internal/backend/aws/backend.go

@@ -6,7 +6,7 @@ import (
 	"fmt"
 	"github.com/dstackai/dstack/runner/internal/container"
 	"io"
-	"io/ioutil"
+	"os"
 	"path"
 	"strings"
 	"time"
@@ -45,7 +45,7 @@ func init() {
 	backend.RegisterBackend("aws", func(ctx context.Context, pathConfig string) (backend.Backend, error) {
 		file := File{}
 		log.Trace(ctx, "Read config file", "path", pathConfig)
-		theConfig, err := ioutil.ReadFile(pathConfig)
+		theConfig, err := os.ReadFile(pathConfig)
 		if err != nil {
 			return nil, gerrors.Wrap(err)
 		}

runner/internal/backend/aws/ec2.go

@@ -2,13 +2,12 @@ package aws
 
 import (
 	"context"
-	"io/ioutil"
-
 	"github.com/aws/aws-sdk-go-v2/config"
 	"github.com/aws/aws-sdk-go-v2/feature/ec2/imds"
 	"github.com/aws/aws-sdk-go-v2/service/ec2"
 	"github.com/dstackai/dstack/runner/internal/gerrors"
 	"github.com/dstackai/dstack/runner/internal/log"
+	"io"
 )
 
 type ClientEC2 struct {
@@ -98,7 +97,7 @@ func (ec *ClientEC2) getInstanceID(ctx context.Context) (string, error) {
 	if err != nil {
 		return "", gerrors.Wrap(err)
 	}
-	id, err := ioutil.ReadAll(meta.Content)
+	id, err := io.ReadAll(meta.Content)
 	if err != nil {
 		return "", gerrors.Wrap(err)
 	}

runner/internal/backend/azure/backend.go

@@ -172,7 +172,7 @@ func (azbackend *AzureBackend) CreateLogger(ctx context.Context, logGroup, logNa
 		azbackend.config.StorageAccount,
 	)
 	logger := NewAzureLogger(loggingClient, azbackend.state.Job.JobID, logGroup, logName)
-	logger.Launch(ctx)
+	_ = logger.Launch(ctx)
 	return logger
 }
 

runner/internal/backend/azure/logging.go

@@ -185,7 +185,7 @@ func (azlogger *AzureLogger) doFlush() {
 	if len(azlogger.logBuff) == 0 {
 		return
 	}
-	azlogger.writeLogs(azlogger.logBuff)
+	_ = azlogger.writeLogs(azlogger.logBuff)
 	azlogger.logBuff = azlogger.logBuff[:0]
 }
 

runner/internal/backend/azure/secret.go

@@ -40,7 +40,9 @@ func (azsecret AzureSecretManager) FetchCredentials(ctx context.Context, repoId
 	if err != nil {
 		return nil, gerrors.Wrap(err)
 	}
-	json.Unmarshal([]byte(data), &creds)
+	if err = json.Unmarshal([]byte(data), &creds); err != nil {
+		return nil, gerrors.Wrap(err)
+	}
 	return &creds, nil
 }
 

runner/internal/backend/backend.go

@@ -6,7 +6,6 @@ import (
 	"github.com/dstackai/dstack/runner/internal/backend/base"
 	"github.com/dstackai/dstack/runner/internal/container"
 	"io"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sync"
@@ -76,7 +75,7 @@ func New(ctx context.Context, path string) (Backend, error) {
 	if _, err := os.Stat(filepath.Join(path)); err != nil {
 		return DefaultBackend, nil
 	}
-	theConfig, err := ioutil.ReadFile(path)
+	theConfig, err := os.ReadFile(path)
 	if err != nil {
 		return nil, gerrors.Wrap(err)
 	}

runner/internal/backend/gcp/logging.go

@@ -67,10 +67,10 @@ func (glogger *GCPLogger) flushLogs(ctx context.Context, ticker *time.Ticker) {
 			// Backend will Close() the client on Shutdown(), flushing the logs.
 			// We don't want to Close() from this goroutine since google libraries may
 			// send audit logs after ctx.Done().
-			glogger.logger.Flush()
+			_ = glogger.logger.Flush()
 			return
 		case <-ticker.C:
-			glogger.logger.Flush()
+			_ = glogger.logger.Flush()
 		}
 	}
 }

runner/internal/backend/gcp/secret.go

@@ -46,7 +46,9 @@ func (sm *GCPSecretManager) FetchCredentials(ctx context.Context, repoId string)
 	if err != nil {
 		return nil, gerrors.Wrap(err)
 	}
-	json.Unmarshal([]byte(data), creds)
+	if err = json.Unmarshal([]byte(data), creds); err != nil {
+		return nil, gerrors.Wrap(err)
+	}
 	return creds, nil
 }
 

runner/internal/backend/lambda/backend.go

@@ -5,7 +5,6 @@ import (
 	"github.com/dstackai/dstack/runner/internal/backend/base"
 	"github.com/dstackai/dstack/runner/internal/container"
 	"io"
-	"io/ioutil"
 	"os"
 
 	"github.com/docker/docker/api/types/mount"
@@ -43,7 +42,7 @@ func init() {
 	backend.RegisterBackend("lambda", func(ctx context.Context, pathConfig string) (backend.Backend, error) {
 		config := LambdaConfig{}
 		log.Trace(ctx, "Read config file", "path", pathConfig)
-		fileContent, err := ioutil.ReadFile(pathConfig)
+		fileContent, err := os.ReadFile(pathConfig)
 		if err != nil {
 			return nil, gerrors.Wrap(err)
 		}
@@ -57,8 +56,8 @@ func init() {
 }
 
 func New(config LambdaConfig) *LambdaBackend {
-	os.Setenv("AWS_ACCESS_KEY_ID", config.StorageConfig.Credentials.AccessKey)
-	os.Setenv("AWS_SECRET_ACCESS_KEY", config.StorageConfig.Credentials.SecretKey)
+	_ = os.Setenv("AWS_ACCESS_KEY_ID", config.StorageConfig.Credentials.AccessKey)
+	_ = os.Setenv("AWS_SECRET_ACCESS_KEY", config.StorageConfig.Credentials.SecretKey)
 	return &LambdaBackend{
 		storageBackend: aws.New(config.StorageConfig.Region, config.StorageConfig.Bucket),
 		apiClient:      NewLambdaAPIClient(config.ApiKey),

runner/internal/backend/local/backend.go

@@ -6,7 +6,6 @@ import (
 	"github.com/dstackai/dstack/runner/internal/backend/base"
 	"github.com/dstackai/dstack/runner/internal/container"
 	"io"
-	"io/ioutil"
 	"os"
 	"path"
 	"path/filepath"
@@ -43,7 +42,7 @@ const LOCAL_BACKEND_DIR = "local_backend"
 func init() {
 	backend.RegisterBackend("local", func(ctx context.Context, pathConfig string) (backend.Backend, error) {
 		config := LocalConfigFile{}
-		fileContent, err := ioutil.ReadFile(pathConfig)
+		fileContent, err := os.ReadFile(pathConfig)
 		if err != nil {
 			fmt.Println("[ERROR]", err.Error())
 			return nil, err
@@ -58,12 +57,12 @@ func init() {
 }
 
 func New(namespace string) *Local {
-	path := filepath.Join(common.HomeDir(), consts.DSTACK_DIR_PATH, LOCAL_BACKEND_DIR, namespace)
+	storagePath := filepath.Join(common.HomeDir(), consts.DSTACK_DIR_PATH, LOCAL_BACKEND_DIR, namespace)
 	return &Local{
 		namespace: namespace,
-		path:      path,
-		storage:   NewLocalStorage(path),
-		cliSecret: NewClientSecret(path),
+		path:      storagePath,
+		storage:   NewLocalStorage(storagePath),
+		cliSecret: NewClientSecret(storagePath),
 	}
 }
 

runner/internal/backend/local/secret.go

@@ -27,7 +27,7 @@ func (sm *ClientSecret) fetchSecret(_ context.Context, path string, secrets map[
 	if err != nil {
 		return nil, gerrors.Wrap(err)
 	}
-	defer db.Close()
+	defer func() { _ = db.Close() }()
 	stmt, err := db.Prepare("SELECT secret_string FROM KV WHERE secret_name=?")
 	if err != nil {
 		return nil, gerrors.Wrap(err)
@@ -56,7 +56,7 @@ func (sm *ClientSecret) fetchCredentials(ctx context.Context, repoId string) *mo
 		log.Error(ctx, "Connecting database. Credentials Local", "RepoId", repoId, "err", err)
 		return nil
 	}
-	defer db.Close()
+	defer func() { _ = db.Close() }()
 	rows, err := db.Query("SELECT secret_string FROM KV WHERE secret_name=?", fmt.Sprintf("/dstack/credentials/%s", repoId))
 	if err != nil {
 		log.Error(ctx, "Fetching value credentials Local", "RepoId", repoId, "err", err)

runner/internal/container/engine.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/docker/docker/api/types/mount"
+	"github.com/docker/docker/pkg/jsonmessage"
 	"github.com/dstackai/dstack/runner/internal/environment"
 	"github.com/dstackai/dstack/runner/internal/models"
 	"io"
@@ -105,7 +106,7 @@ type DockerRuntime struct {
 
 func (r *Engine) Create(ctx context.Context, spec *Spec, logs io.Writer) (*DockerRuntime, error) {
 	log.Trace(ctx, "Start pull image")
-	err := r.pullImageIfAbsent(ctx, spec.Image, spec.RegistryAuthBase64)
+	err := r.pullImageIfAbsent(ctx, spec.Image, spec.RegistryAuthBase64, logs)
 	if err != nil {
 		log.Error(ctx, fmt.Sprintf("failed to download docker image: %s", err))
 		return nil, gerrors.Newf("failed to download docker image: %s", err)
@@ -265,7 +266,7 @@ func (r *DockerRuntime) Stop(ctx context.Context) error {
 	return nil
 }
 
-func (r *Engine) pullImageIfAbsent(ctx context.Context, image string, registryAuthBase64 string) error {
+func (r *Engine) pullImageIfAbsent(ctx context.Context, image string, registryAuthBase64 string, logs io.Writer) error {
 	if image == "" {
 		return gerrors.New("given image value is empty")
 	}
@@ -287,16 +288,23 @@ func (r *Engine) pullImageIfAbsent(ctx context.Context, image string, registryAu
 		return gerrors.Wrap(err)
 	}
 	defer func() { _ = reader.Close() }()
-	buf, err := io.ReadAll(reader)
-	if err != nil {
-		return gerrors.Wrap(err)
+
+	if logs != nil {
+		if err = jsonmessage.DisplayJSONMessagesStream(reader, logs, 0, false, nil); err != nil {
+			return gerrors.Wrap(err)
+		}
+	} else {
+		buf, err := io.ReadAll(reader)
+		if err != nil {
+			return gerrors.Wrap(err)
+		}
+		log.Trace(ctx, "Image pull stdout", "stdout", string(buf))
 	}
-	log.Trace(ctx, "Image pull stdout", "stdout", string(buf))
 	return nil
 }
 
-func (r *Engine) NewBuildSpec(ctx context.Context, job *models.Job, spec *Spec, secrets map[string]string, repoPath string) (*BuildSpec, error) {
-	err := r.pullImageIfAbsent(ctx, spec.Image, spec.RegistryAuthBase64)
+func (r *Engine) NewBuildSpec(ctx context.Context, job *models.Job, spec *Spec, secrets map[string]string, repoPath string, logs io.Writer) (*BuildSpec, error) {
+	err := r.pullImageIfAbsent(ctx, spec.Image, spec.RegistryAuthBase64, logs)
 	if err != nil {
 		return nil, gerrors.Wrap(err)
 	}

runner/internal/executor/config.go

@@ -2,7 +2,6 @@ package executor
 
 import (
 	"context"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -29,7 +28,7 @@ func (ex *Executor) loadConfig(configDir string) error {
 		log.Error(context.Background(), "Failed to load config", "err", gerrors.Wrap(err))
 		return err
 	}
-	theConfigFile, err := ioutil.ReadFile(thePathConfig)
+	theConfigFile, err := os.ReadFile(thePathConfig)
 	if err != nil {
 		log.Error(context.Background(), "Unexpected error, please try to rerun", "err", gerrors.Wrap(err))
 		return err

runner/internal/executor/executor.go

@@ -308,15 +308,6 @@ func (ex *Executor) runJob(ctx context.Context, erCh chan error, stoppedCh chan
 	defer func() { _ = fileLog.Close() }()
 	allLogs := io.MultiWriter(logger, ex.streamLogs, fileLog)
 
-	_, isLocalBackend := ex.backend.(*localbackend.Local)
-	if isLocalBackend {
-		err := ex.warnOnLongImagePull(ctx, job.Image)
-		if err != nil {
-			erCh <- gerrors.Wrap(err)
-			return
-		}
-	}
-
 	log.Trace(ctx, "Building container", "mode", job.BuildPolicy)
 	job.Status = states.Building
 	if err = ex.backend.UpdateState(jctx); err != nil {
@@ -716,7 +707,7 @@ func (ex *Executor) build(ctx context.Context, spec *container.Spec, stoppedCh c
 	if err != nil {
 		return gerrors.Wrap(err)
 	}
-	buildSpec, err := ex.engine.NewBuildSpec(ctx, job, spec, secrets, path.Join(ex.backend.GetTMPDir(ctx), consts.RUNS_DIR, job.RunName, job.JobID))
+	buildSpec, err := ex.engine.NewBuildSpec(ctx, job, spec, secrets, path.Join(ex.backend.GetTMPDir(ctx), consts.RUNS_DIR, job.RunName, job.JobID), logs)
 	if err != nil {
 		return gerrors.Wrap(err)
 	}

runner/internal/repo/archive.go

@@ -13,7 +13,7 @@ func ExtractArchive(ctx context.Context, src, dst string) error {
 	if err != nil {
 		return gerrors.Wrap(err)
 	}
-	defer file.Close()
+	defer func() { _ = file.Close() }()
 	log.Trace(ctx, "Extracting archive", "src", src, "dst", dst)
 	if err := extract.Archive(ctx, file, dst, nil); err != nil {
 		return gerrors.Wrap(err)

runner/internal/repo/diff_test.go

@@ -2,7 +2,6 @@ package repo
 
 import (
 	"context"
-	"io/ioutil"
 	"os"
 	"path"
 	"testing"
@@ -307,7 +306,7 @@ Last line.`
 	for _, c := range cases {
 		cc := c
 		t.Run(c.name, func(t *testing.T) {
-			dir, err := ioutil.TempDir("", "dstack-unit-")
+			dir, err := os.MkdirTemp("", "dstack-unit-")
 			assert.NoError(t, err, "create tmp directory for test")
 			if err != nil {
 				defer func() {