added logging with timestamp and a client based lock

Signed-off-by: Gergely Brautigam <[email protected]>

api/oci/extensions/repositories/ocireg/repository.go

@@ -11,6 +11,7 @@ import (
 	"github.com/containerd/errdefs"
 	"github.com/mandelsoft/goutils/errors"
 	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	"oras.land/oras-go/v2/registry/remote/auth"
 	"oras.land/oras-go/v2/registry/remote/retry"
 
@@ -179,6 +180,7 @@ func (r *RepositoryImpl) getResolver(comp string) (oras.Resolver, error) {
 		Client:    authClient,
 		PlainHTTP: r.info.Scheme == "http",
 		Logger:    logger,
+		Lock:      locker.New(),
 	}), nil
 }
 

api/tech/oras/client.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/containerd/containerd/errdefs"
 	"github.com/mandelsoft/logging"
+	"github.com/moby/locker"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	oraserr "oras.land/oras-go/v2/errdef"
 	"oras.land/oras-go/v2/registry/remote"
@@ -17,26 +18,28 @@ type ClientOptions struct {
 	Client    *auth.Client
 	PlainHTTP bool
 	Logger    logging.Logger
+	Lock      *locker.Locker
 }
 
 type Client struct {
 	client    *auth.Client
 	plainHTTP bool
 	logger    logging.Logger
+	lock      *locker.Locker
 }
 
 var _ Resolver = &Client{}
 
 func New(opts ClientOptions) *Client {
-	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP, logger: opts.Logger}
+	return &Client{client: opts.Client, plainHTTP: opts.PlainHTTP, logger: opts.Logger, lock: opts.Lock}
 }
 
 func (c *Client) Fetcher(ctx context.Context, ref string) (Fetcher, error) {
 	return &OrasFetcher{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
 }
 
 func (c *Client) Pusher(ctx context.Context, ref string) (Pusher, error) {
-	return &OrasPusher{client: c.client, ref: ref, plainHTTP: c.plainHTTP}, nil
+	return &OrasPusher{client: c.client, ref: ref, plainHTTP: c.plainHTTP, lock: c.lock}, nil
 }
 
 func (c *Client) Lister(ctx context.Context, ref string) (Lister, error) {

api/tech/oras/fetcher.go

@@ -5,7 +5,11 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"log"
+	"sync"
+	"time"
 
+	"github.com/google/uuid"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras-go/v2/registry/remote/auth"
 )
@@ -14,9 +18,19 @@ type OrasFetcher struct {
 	client    *auth.Client
 	ref       string
 	plainHTTP bool
+	mu        sync.Mutex
 }
 
 func (c *OrasFetcher) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.ReadCloser, error) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	start := time.Now()
+	id := uuid.New()
+
+	log.Printf("START fetch %s; %s: %s\n", id.String(), c.ref, start)
+	defer log.Printf("END fetch %s; %s: %s\n", id.String(), c.ref, time.Since(start))
+
 	src, err := createRepository(c.ref, c.client, c.plainHTTP)
 	if err != nil {
 		return nil, fmt.Errorf("failed to resolve ref %q: %w", c.ref, err)
@@ -30,8 +44,10 @@ func (c *OrasFetcher) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.Read
 	// that the mediatype is not set at this point so we don't want ORAS to try to
 	// select the wrong layer to fetch from.
 	if desc.Size < 1 || desc.Digest == "" {
+		log.Printf("Trying to resolve manifest %s: %s", id.String(), c.ref)
 		rdesc, err := src.Manifests().Resolve(ctx, desc.Digest.String())
 		if err != nil {
+			log.Printf("Failed to resolve manifest, trying to resolve blob %s: %s", id.String(), c.ref)
 			var berr error
 			rdesc, berr = src.Blobs().Resolve(ctx, desc.Digest.String())
 			if berr != nil {
@@ -41,10 +57,12 @@ func (c *OrasFetcher) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.Read
 				return nil, fmt.Errorf("failed to resolve fetch blob %q: %w", desc.Digest.String(), err)
 			}
 
+			log.Printf("Blob resolved, fetching reader %s: %s", id.String(), c.ref)
 			reader, err := src.Blobs().Fetch(ctx, rdesc)
 			if err != nil {
 				return nil, fmt.Errorf("failed to fetch blob: %w", err)
 			}
+			log.Printf("Reader fetched, returning %s: %s", id.String(), c.ref)
 
 			return reader, nil
 		}
@@ -55,10 +73,12 @@ func (c *OrasFetcher) Fetch(ctx context.Context, desc ociv1.Descriptor) (io.Read
 
 	// manifest resolve succeeded return the reader directly
 	// mediatype of the descriptor should now be set to the correct type.
+	log.Printf("Fetching manifest %s: %s", id.String(), c.ref)
 	fetch, err := src.Fetch(ctx, desc)
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch manifest: %w", err)
 	}
 
+	log.Printf("Manifest fetched; returining %s: %s", id.String(), c.ref)
 	return fetch, nil
 }

api/tech/oras/pusher.go

@@ -3,9 +3,12 @@ package oras
 import (
 	"context"
 	"fmt"
-	"sync"
+	"log"
+	"time"
 
 	"github.com/containerd/errdefs"
+	"github.com/google/uuid"
+	"github.com/moby/locker"
 	ociv1 "github.com/opencontainers/image-spec/specs-go/v1"
 	"oras.land/oras-go/v2/registry"
 	"oras.land/oras-go/v2/registry/remote/auth"
@@ -17,12 +20,17 @@ type OrasPusher struct {
 	client    *auth.Client
 	ref       string
 	plainHTTP bool
-	mu        sync.Mutex
+	lock      *locker.Locker
 }
 
 func (c *OrasPusher) Push(ctx context.Context, d ociv1.Descriptor, src Source) (retErr error) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
+	c.lock.Lock(c.ref)
+	defer c.lock.Unlock(c.ref)
+	start := time.Now()
+	id := uuid.New()
+
+	log.Printf("START push %s; %s: %s\n", id.String(), c.ref, start)
+	defer log.Printf("END push %s; %s: %s\n", id.String(), c.ref, time.Since(start))
 
 	reader, err := src.Reader()
 	if err != nil {
@@ -52,10 +60,12 @@ func (c *OrasPusher) Push(ctx context.Context, d ociv1.Descriptor, src Source) (
 		// layer with the reference included. PushReference then will tag
 		// that layer resulting in the created tag pointing to the right
 		// blob data.
+		log.Printf("Pushing Reference %s: %s", ref, id.String())
 		if err := repository.PushReference(ctx, d, reader, c.ref); err != nil {
 			return fmt.Errorf("failed to push tag: %w", err)
 		}
 
+		log.Printf("Pushing Reference done %s: %s", ref, id.String())
 		return nil
 	}
 
@@ -70,9 +80,12 @@ func (c *OrasPusher) Push(ctx context.Context, d ociv1.Descriptor, src Source) (
 
 	// We have a digest, so we use plain push for the digest.
 	// Push here decides if it's a Manifest or a Blob.
+	log.Printf("Pushing blob %s: %s", ref, id.String())
 	if err := repository.Push(ctx, d, reader); err != nil {
 		return fmt.Errorf("failed to push: %w, %s", err, c.ref)
 	}
 
+	log.Printf("Pushing blob done %s: %s", ref, id.String())
+
 	return nil
 }

go.mod

@@ -37,6 +37,7 @@ require (
 	github.com/gobwas/glob v0.2.3
 	github.com/golang/mock v1.6.0
 	github.com/google/go-github/v45 v45.2.0
+	github.com/google/uuid v1.6.0
 	github.com/hashicorp/vault-client-go v0.4.3
 	github.com/klauspost/compress v1.17.11
 	github.com/klauspost/pgzip v1.2.6
@@ -49,6 +50,7 @@ require (
 	github.com/mikefarah/yq/v4 v4.44.6
 	github.com/mitchellh/copystructure v1.2.0
 	github.com/mittwald/go-helm-client v0.12.15
+	github.com/moby/locker v1.0.1
 	github.com/modern-go/reflect2 v1.0.2
 	github.com/onsi/ginkgo/v2 v2.22.0
 	github.com/onsi/gomega v1.36.1
@@ -217,7 +219,6 @@ require (
 	github.com/google/pprof v0.0.0-20241029153458-d1b30febd7db // indirect
 	github.com/google/s2a-go v0.1.8 // indirect
 	github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.4 // indirect
 	github.com/googleapis/gax-go/v2 v2.14.0 // indirect
 	github.com/gorilla/mux v1.8.1 // indirect
@@ -258,7 +259,6 @@ require (
 	github.com/mitchellh/mapstructure v1.5.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
-	github.com/moby/locker v1.0.1 // indirect
 	github.com/moby/spdystream v0.5.0 // indirect
 	github.com/moby/sys/capability v0.3.0 // indirect
 	github.com/moby/sys/mountinfo v0.7.2 // indirect