feat: improve GCS-next driver debuggability (fd44b208) · Commits · GitLab.org / container-registry

docs/configuration.md

+2 −0

Original line number	Diff line number	Diff line
		@@ -108,6 +108,7 @@ storage:
		bucket: bucketname
		keyfile: /path/to/keyfile
		useragent: container-registry
		debug_log: false
		credentials:
		type: service_account
		project_id: project_id_string
		@@ -504,6 +505,7 @@ storage:
		bucket: bucketname
		keyfile: /path/to/keyfile
		useragent: container-registry
		debug_log: false
		credentials:
		type: service_account
		project_id: project_id_string

go.mod

+4 −0

Original line number	Diff line number	Diff line
		@@ -35,12 +35,14 @@ require (
		github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438
		github.com/jackc/pgx/v5 v5.7.5
		github.com/lib/pq v1.10.9
		github.com/neilotoole/slogt v1.1.0
		github.com/olekukonko/tablewriter v1.0.7
		github.com/opencontainers/go-digest v1.0.0
		github.com/opencontainers/image-spec v1.1.1
		github.com/prometheus/client_golang v1.22.0
		github.com/redis/go-redis/v9 v9.10.0
		github.com/rubenv/sql-migrate v1.8.0
		github.com/samber/slog-logrus/v2 v2.5.2
		github.com/schollz/progressbar/v3 v3.18.0
		github.com/sirupsen/logrus v1.9.3
		github.com/spf13/cobra v1.9.1
		@@ -177,6 +179,8 @@ require (
		github.com/prometheus/procfs v0.15.1 // indirect
		github.com/rivo/uniseg v0.4.7 // indirect
		github.com/sagikazarmark/locafero v0.7.0 // indirect
		github.com/samber/lo v1.47.0 // indirect
		github.com/samber/slog-common v0.18.1 // indirect
		github.com/sebest/xff v0.0.0-20210106013422-671bd2870b3a // indirect
		github.com/shirou/gopsutil/v4 v4.25.3 // indirect
		github.com/sourcegraph/conc v0.3.0 // indirect

go.sum

+8 −0

Original line number	Diff line number	Diff line
		@@ -602,6 +602,8 @@ github.com/nats-io/nats.go v1.9.1/go.mod h1:ZjDU1L/7fJ09jvUSRVBR2e7+RnLiiIQyqyzE
		github.com/nats-io/nkeys v0.1.0/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
		github.com/nats-io/nkeys v0.1.3/go.mod h1:xpnFELMwJABBLVhffcfd1MZx6VsNRFpEugbxziKVo7w=
		github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
		github.com/neilotoole/slogt v1.1.0 h1:c7qE92sq+V0yvCuaxph+RQ2jOKL61c4hqS1Bv9W7FZE=
		github.com/neilotoole/slogt v1.1.0/go.mod h1:RCrGXkPc/hYybNulqQrMHRtvlQ7F6NktNVLuLwk6V+w=
		github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
		github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
		github.com/oklog/oklog v0.3.2/go.mod h1:FCV+B7mhrz4o+ueLpx+KqkyXRGMWOYEvfiXtdGtbWGs=
		@@ -718,6 +720,12 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
		github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
		github.com/sagikazarmark/locafero v0.7.0 h1:5MqpDsTGNDhY8sGp0Aowyf0qKsPrhewaLSsFaodPcyo=
		github.com/sagikazarmark/locafero v0.7.0/go.mod h1:2za3Cg5rMaTMoG/2Ulr9AwtFaIppKXTRYnozin4aB5k=
		github.com/samber/lo v1.47.0 h1:z7RynLwP5nbyRscyvcD043DWYoOcYRv3mV8lBeqOCLc=
		github.com/samber/lo v1.47.0/go.mod h1:RmDH9Ct32Qy3gduHQuKJ3gW1fMHAnE/fAzQuf6He5cU=
		github.com/samber/slog-common v0.18.1 h1:c0EipD/nVY9HG5shgm/XAs67mgpWDMF+MmtptdJNCkQ=
		github.com/samber/slog-common v0.18.1/go.mod h1:QNZiNGKakvrfbJ2YglQXLCZauzkI9xZBjOhWFKS3IKk=
		github.com/samber/slog-logrus/v2 v2.5.2 h1:pReWs5r4u/NbZokrJkFoFhURA5CnlUAgcMcPV1U1LuY=
		github.com/samber/slog-logrus/v2 v2.5.2/go.mod h1:wHDewbid6WQqF8E1HfMGeQtd+nAs9aqZgWCHoiGy4sQ=
		github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
		github.com/schollz/progressbar/v3 v3.18.0 h1:uXdoHABRFmNIjUfte/Ex7WtuyVslrw2wVPQmCN62HpA=
		github.com/schollz/progressbar/v3 v3.18.0/go.mod h1:IsO3lpbaGuzh8zIMzgY3+J8l4C8GjO0Y9S69eFvNsec=

registry/storage/driver/gcs/common.go

+54 −1

Original line number	Diff line number	Diff line
		@@ -5,6 +5,7 @@ import (
		"encoding/json"
		"errors"
		"fmt"
		"log/slog"
		"math/rand/v2"
		"net/http"
		"os"
		@@ -18,6 +19,7 @@ import (

		"cloud.google.com/go/storage"
		"github.com/benbjohnson/clock"
		dcontext "github.com/docker/distribution/context"
		"github.com/docker/distribution/registry/internal"
		dstorage "github.com/docker/distribution/registry/storage"
		storagedriver "github.com/docker/distribution/registry/storage/driver"
		@@ -26,6 +28,7 @@ import (
		"github.com/docker/distribution/registry/storage/driver/internal/parse"
		"github.com/docker/distribution/registry/storage/internal/metrics"
		"github.com/docker/distribution/version"
		sloglogrus "github.com/samber/slog-logrus/v2"
		"github.com/sirupsen/logrus"
		"golang.org/x/oauth2"
		"golang.org/x/oauth2/google"
		@@ -76,6 +79,7 @@ func init() {

		// driverParameters is a struct that encapsulates all the driver parameters after all values have been set
		type driverParameters struct {
		logger dcontext.Logger
		bucket string
		email string
		privateKey []byte
		@@ -218,7 +222,7 @@ func parseParameters(parameters map[string]any, useNext bool) (*driverParameters
		case int, uint, int32, uint32, uint64, int64:
		chunkSize = reflect.ValueOf(v).Convert(reflect.TypeOf(chunkSize)).Int()
		default:
		return nil, fmt.Errorf("invalid valud for chunksize: %#v", chunkSizeParam)
		return nil, fmt.Errorf("invalid value for chunksize: %#v", chunkSizeParam)
		}

		if chunkSize < minChunkSize {
		@@ -280,8 +284,56 @@ func parseParameters(parameters map[string]any, useNext bool) (*driverParameters
		return nil, fmt.Errorf("maxconcurrency config error: %s", err)
		}

		logger := parameters[storagedriver.ParamLogger].(dcontext.Logger)

		opts := []option.ClientOption{option.WithTokenSource(ts)}
		if useNext {
		debugLogging := false

		if _, ok = parameters["debug_log"]; ok {
		debugLogging, err = parse.Bool(parameters, "debug_log", false)
		if err != nil {
		return nil, fmt.Errorf("parsing parameter %s: %w", "debug_log", err)
		}
		}
		if debugLogging {
		// NOTE(prozlach): Casting directly to logrus.Entry is a shortcut here,
		// as we require Logrus logger for the adapter. In theory we should be
		// using the context.Logger interface instead of peeking into the
		// implementation, but this requies a deeper refactoring.
		//
		// Hopefully we will switch to slog/zap at some point and this will
		// become non-issue.
		logrusEntry, ok := logger.(*logrus.Entry)
		if !ok {
		return nil, fmt.Errorf("debug logging requires logrus logger, got %T", logger)
		}
		slogger := slog.New(
		sloglogrus.Option{
		Level: slog.LevelDebug,
		Logger: logrusEntry.Logger,
		}.NewLogrusHandler(),
		)
		// NOTE(prozlach): This does not work really, see https://github.com/googleapis/google-cloud-go/issues/12475
		// As a workaround we set the env variable as well, but this only
		// makes GCS print logs to stdout using JSON format. Not ideal.
		//
		// Only set the environment variable if it's not already set
		// nolint: revive // max-control-nesting
		if existingLevel := os.Getenv("GOOGLE_SDK_GO_LOGGING_LEVEL"); existingLevel != "" {
		logger.WithFields(logrus.Fields{
		"existing_value": existingLevel,
		"requested_value": "debug",
		}).Warn("GOOGLE_SDK_GO_LOGGING_LEVEL environment variable is already set, not overriding")
		} else {
		err = os.Setenv("GOOGLE_SDK_GO_LOGGING_LEVEL", "debug")
		if err != nil {
		return nil, fmt.Errorf("setting `GOOGLE_SDK_GO_LOGGING_LEVEL` env var: %w", err)
		}
		}
		opts = append(opts, option.WithLogger(slogger))
		}

		// NOTE(prozlach): By default, reads are made using the Cloud Storage XML
		// API. GCS SDK recommends using the JSON API instead, which is done
		// here by setting WithJSONReads. This ensures consistency with other
		@@ -310,6 +362,7 @@ func parseParameters(parameters map[string]any, useNext bool) (*driverParameters
		}

		return &driverParameters{
		logger: logger,
		bucket: fmt.Sprint(bucket),
		rootDirectory: fmt.Sprint(rootDirectory),
		email: jwtConf.Email,

registry/storage/driver/gcs/gcs_next.go

+104 −18

Original line number	Diff line number	Diff line
		@@ -31,6 +31,7 @@ import (
		"time"

		"cloud.google.com/go/storage"
		dcontext "github.com/docker/distribution/context"
		storagedriver "github.com/docker/distribution/registry/storage/driver"
		"github.com/docker/distribution/registry/storage/driver/base"
		"github.com/hashicorp/go-multierror"
		@@ -50,6 +51,7 @@ func NewNext(params *driverParameters) (storagedriver.StorageDriver, error) {
		return nil, fmt.Errorf("invalid chunksize: %d is not a positive multiple of %d", params.chunkSize, minChunkSize)
		}
		d := &driverNext{
		logger: params.logger,
		bucket: params.storageClient.Bucket(params.bucket),
		rootDirectory: rootDirectory,
		email: params.email,
		@@ -71,6 +73,7 @@ func NewNext(params *driverParameters) (storagedriver.StorageDriver, error) {
		// driverNext is a storagedriver.StorageDriver implementation backed by GCS
		// Objects are stored at absolute keys in the provided bucket.
		type driverNext struct {
		logger dcontext.Logger
		client *http.Client
		bucket *storage.BucketHandle
		email string
		@@ -130,7 +133,7 @@ func (d *driverNext) PutContent(ctx context.Context, path string, contents []byt
		wc.MD5 = h.Sum(nil)

		if len(contents) == 0 {
		logrus.WithFields(logrus.Fields{
		d.logger.WithFields(logrus.Fields{
		"path": path,
		"length": len(contents),
		"stack": string(debug.Stack()),
		@@ -182,6 +185,13 @@ func (d *driverNext) Reader(ctx context.Context, path string, offset int64) (io.
		_ = r.Close()
		return nil, storagedriver.PathNotFoundError{Path: path, DriverName: driverName}
		}
		if offset == r.Attrs.Size {
		d.logger.WithFields(logrus.Fields{
		"path": path,
		"fileSize": r.Attrs.Size,
		"requestedOffset": offset,
		}).Info("Range request at EOF, returning empty reader")
		}
		return r, nil
		}

		@@ -243,16 +253,23 @@ func putContentsCloseNext(wc *storage.Writer, contents []byte) error {
		}

		// putChunkNext either completes upload or submits another chunk.
		func putChunkNext(ctx context.Context, client *http.Client, sessionURI string, chunk []byte, from, totalSize int64) (int64, error) {
		func (w *writerNext) putChunkNext(chunk []byte, totalSize int64) (int64, error) {
		w.logger.WithFields(logrus.Fields{
		"chunkStart": w.offset,
		"chunkEnd": w.offset + int64(len(chunk)) - 1,
		"chunkSize": len(chunk),
		"totalSize": totalSize,
		}).Debug("Uploading chunk")

		// Documentation: https://cloud.google.com/storage/docs/performing-resumable-uploads#chunked-upload
		bytesPut := int64(0)
		err := retry(func() error {
		req, err := http.NewRequestWithContext(ctx, http.MethodPut, sessionURI, bytes.NewReader(chunk))
		req, err := http.NewRequestWithContext(w.ctx, http.MethodPut, w.sessionURI, bytes.NewReader(chunk))
		if err != nil {
		return fmt.Errorf("creating new http request: %w", err)
		}
		length := int64(len(chunk))
		to := from + length - 1
		to := w.offset + length - 1
		req.Header.Set("Content-Type", "application/octet-stream")
		// NOTE(prozlach) fake-gcs-server documents this behavior well:
		// https://github.com/fsouza/fake-gcs-server/blob/1e954726309326217b4c533bb9646e30f683fa66/fakestorage/upload.go#L467-L501
		@@ -294,14 +311,14 @@ func putChunkNext(ctx context.Context, client *http.Client, sessionURI string, c
		if totalSize >= 0 {
		size = strconv.FormatInt(totalSize, 10)
		}
		if from == to+1 {
		if w.offset == to+1 {
		req.Header.Set("Content-Range", fmt.Sprintf("bytes */%v", size))
		} else {
		req.Header.Set("Content-Range", fmt.Sprintf("bytes %v-%v/%v", from, to, size))
		req.Header.Set("Content-Range", fmt.Sprintf("bytes %v-%v/%v", w.offset, to, size))
		}
		req.Header.Set("Content-Length", strconv.FormatInt(length, 10))

		resp, err := client.Do(req)
		resp, err := w.client.Do(req)
		if err != nil {
		return fmt.Errorf("executing http request: %w", err)
		}
		@@ -324,14 +341,20 @@ func putChunkNext(ctx context.Context, client *http.Client, sessionURI string, c
		if err != nil {
		return fmt.Errorf("parsing Range header: %w", err)
		}
		bytesPut = end - from + 1
		bytesPut = end - w.offset + 1

		w.logger.WithFields(logrus.Fields{
		"bytesUploaded": bytesPut,
		"totalExpected": len(chunk),
		"rangeHeader": resp.Header.Get("Range"),
		}).Debug("Partial chunk upload, continuing")
		return nil
		}
		err = googleapi.CheckMediaResponse(resp)
		if err != nil {
		return fmt.Errorf("committing resumable upload: %w", err)
		}
		bytesPut = to - from + 1
		bytesPut = to - w.offset + 1
		return nil
		})
		return bytesPut, err
		@@ -341,6 +364,7 @@ func putChunkNext(ctx context.Context, client *http.Client, sessionURI string, c
		// at the location designated by "path" after the call to Commit.
		func (d *driverNext) Writer(ctx context.Context, path string, doAppend bool) (storagedriver.FileWriter, error) {
		w := &writerNext{
		logger: d.logger,
		ctx: ctx,
		client: d.client,
		object: d.bucket.Object(d.pathToKey(path)),
		@@ -449,6 +473,19 @@ func (d *driverNext) List(ctx context.Context, path string) ([]string, error) {
		return nil, fmt.Errorf("fetching next page from iterator: %w", err)
		}

		if !attrs.Deleted.IsZero() {
		d.logger.WithFields(logrus.Fields{
		"path": d.keyToPath(attrs.Name),
		"deletedTime": attrs.Deleted,
		}).Debug("Filtered out deleted object from listing due to eventual consistency")
		}

		if attrs.ContentType == uploadSessionContentType {
		d.logger.WithFields(logrus.Fields{
		"path": d.keyToPath(attrs.Name),
		}).Debug("Filtered out temporary upload session object from listing")
		}

		// GCS does not guarantee strong consistency between
		// DELETE and LIST operations. Check that the object is not deleted,
		// and filter out any objects with a non-zero time-deleted
		@@ -489,7 +526,12 @@ func (d *driverNext) Move(ctx context.Context, sourcePath, destPath string) erro
		// if deleting the file fails, log the error, but do not fail; the file was successfully copied,
		// and the original should eventually be cleaned when purging the uploads folder.
		if err != nil {
		logrus.Infof("error deleting file: %v due to %v", sourcePath, err)
		d.logger.WithFields(logrus.Fields{
		"sourcePath": sourcePath,
		"destPath": destPath,
		"copySucceeded": true,
		"deleteError": err.Error(),
		}).Info("Move operation: copy succeeded but delete failed, file will need cleanup")
		}
		return nil
		}
		@@ -572,6 +614,11 @@ func (d *driverNext) DeleteFiles(ctx context.Context, paths []string) (int, erro
		g, gctx := errgroup.WithContext(ctx)
		g.SetLimit(maxDeleteConcurrency)

		d.logger.WithFields(logrus.Fields{
		"totalFiles": len(paths),
		"maxConcurrency": maxDeleteConcurrency,
		}).Debug("Starting concurrent file deletion")

		for _, path := range paths {
		g.Go(func() error {
		// Check if any context was canceled, if so - skip calling Delete
		@@ -607,6 +654,12 @@ func (d *driverNext) DeleteFiles(ctx context.Context, paths []string) (int, erro
		// multierror var
		_ = g.Wait()

		d.logger.WithFields(logrus.Fields{
		"totalFiles": len(paths),
		"successfulDeletes": int(count.Load()),
		"errors": errs.Len(),
		}).Debug("Completed concurrent file deletion")

		return int(count.Load()), errs.ErrorOrNil()
		}

		@@ -644,6 +697,20 @@ func (d *driverNext) URLFor(_ context.Context, path string, options map[string]a
		// If we have a private key and email from service account JSON, use them directly
		opts.GoogleAccessID = d.email
		opts.PrivateKey = d.privateKey

		d.logger.WithFields(logrus.Fields{
		"path": path,
		"method": methodString,
		"signingMethod": "service_account_key",
		"expires": expiresTime,
		}).Debug("Generating signed URL with service account credentials")
		} else {
		d.logger.WithFields(logrus.Fields{
		"path": path,
		"method": methodString,
		"signingMethod": "instance_profile",
		"expires": expiresTime,
		}).Debug("Generating signed URL with instance profile credentials")
		}

		// Signing a URL requires credentials authorized to sign a URL. They can be
		@@ -678,22 +745,22 @@ func (d *driverNext) WalkParallel(ctx context.Context, path string, f storagedri

		// startSessionNext starts a new resumable upload session. Documentation can be
		// found here: https://cloud.google.com/storage/docs/performing-resumable-uploads#initiate-session
		func startSessionNext(ctx context.Context, client *http.Client, bucket, name string) (uri string, err error) {
		func (w *writerNext) startSessionNext() (uri string, err error) {
		u := &url.URL{
		Scheme: "https",
		// https://cloud.google.com/storage/docs/request-endpoints#typical
		Host: "storage.googleapis.com",
		Path: fmt.Sprintf("/upload/storage/v1/b/%v/o", bucket),
		RawQuery: fmt.Sprintf("uploadType=resumable&name=%v", name),
		Path: fmt.Sprintf("/upload/storage/v1/b/%v/o", w.object.BucketName()),
		RawQuery: fmt.Sprintf("uploadType=resumable&name=%v", w.object.ObjectName()),
		}
		err = retry(func() error {
		req, err := http.NewRequestWithContext(ctx, http.MethodPost, u.String(), nil)
		req, err := http.NewRequestWithContext(w.ctx, http.MethodPost, u.String(), nil)
		if err != nil {
		return err
		}
		req.Header.Set("X-Upload-Content-Type", "application/octet-stream")
		req.Header.Set("Content-Length", "0")
		resp, err := client.Do(req)
		resp, err := w.client.Do(req)
		if err != nil {
		return fmt.Errorf("executing http request: %w", err)
		}
		@@ -705,6 +772,10 @@ func startSessionNext(ctx context.Context, client *http.Client, bucket, name str
		uri = resp.Header.Get("Location")
		return nil
		})
		w.logger.WithFields(logrus.Fields{
		"bucket": w.object.BucketName(),
		"object": w.object.ObjectName(),
		}).Debug("Created new resumable upload session")
		return uri, err
		}

		@@ -721,6 +792,7 @@ func (d *driverNext) keyToPath(key string) string {
		}

		type writerNext struct {
		logger dcontext.Logger
		ctx context.Context
		client *http.Client
		object *storage.ObjectHandle
		@@ -853,7 +925,7 @@ func (w *writerNext) Commit() error {
		// is committed even when the buffer is empty. The empty/first loop with
		// zero-length will actually commit the file.
		for {
		n, err := putChunkNext(w.ctx, w.client, w.sessionURI, w.buffer[nn:w.buffSize], w.offset, w.size)
		n, err := w.putChunkNext(w.buffer[nn:w.buffSize], w.size)
		nn += n
		w.offset += n
		if err != nil {
		@@ -878,12 +950,12 @@ func (w *writerNext) writeChunk() error {
		}
		// if their is no sessionURI yet, obtain one by starting the session
		if w.sessionURI == "" {
		w.sessionURI, err = startSessionNext(w.ctx, w.client, w.object.BucketName(), w.object.ObjectName())
		w.sessionURI, err = w.startSessionNext()
		}
		if err != nil {
		return err
		}
		nn, err := putChunkNext(w.ctx, w.client, w.sessionURI, w.buffer[0:chunkSize], w.offset, -1)
		nn, err := w.putChunkNext(w.buffer[0:chunkSize], -1)
		w.offset += nn
		// shift the remaining bytes to the start of the buffer
		w.buffSize = int64(copy(w.buffer, w.buffer[nn:w.buffSize])) // nolint: gosec // copy() is always going to be non-negative
		@@ -916,6 +988,13 @@ func (w *writerNext) Write(p []byte) (int, error) {
		// * this follows the semantics of Azure and S3 drivers
		w.size += int64(n)
		if w.buffSize == int64(cap(w.buffer)) {
		w.logger.WithFields(logrus.Fields{
		"path": w.object.ObjectName(),
		"bufferSize": w.buffSize,
		"totalWritten": w.size,
		"chunkNumber": w.offset/int64(cap(w.buffer)) + 1,
		}).Debug("Buffer full, initiating chunk upload")

		err = w.writeChunk()
		if err != nil {
		break
		@@ -983,5 +1062,12 @@ func (w *writerNext) init() error {
		w.sessionURI = attrs.Metadata["Session-URI"]
		w.offset = offset
		w.size = offset + w.buffSize

		w.logger.WithFields(logrus.Fields{
		"path": w.object.ObjectName(),
		"resumedOffset": offset,
		"bufferedBytes": w.buffSize,
		}).Debug("Resumed upload session")

		return nil
		}