Merge pull request #12755 from heww/expired-blob-size

fix(blob): make blob size in redis expired in 24 hours

src/controller/blob/controller.go

@@ -17,6 +17,7 @@ package blob
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/docker/distribution"
 	"github.com/goharbor/harbor/src/lib/errors"
@@ -92,12 +93,14 @@ type Controller interface {
 // NewController creates an instance of the default repository controller
 func NewController() Controller {
 	return &controller{
-		blobMgr: blob.Mgr,
+		blobMgr:            blob.Mgr,
+		blobSizeExpiration: time.Hour * 24, // keep the size of blob in redis with 24 hours
 	}
 }
 
 type controller struct {
-	blobMgr blob.Manager
+	blobMgr            blob.Manager
+	blobSizeExpiration time.Duration
 }
 
 func (c *controller) AssociateWithArtifact(ctx context.Context, blobDigests []string, artifactDigest string) error {
@@ -297,14 +300,17 @@ func (c *controller) SetAcceptedBlobSize(sessionID string, size int64) error {
 	conn := redislib.DefaultPool().Get()
 	defer conn.Close()
 
-	key := fmt.Sprintf("upload:%s:size", sessionID)
+	key := blobSizeKey(sessionID)
-	reply, err := redis.String(conn.Do("SET", key, size))
+	reply, err := redis.String(conn.Do("SET", key, size, "EX", int64(c.blobSizeExpiration/time.Second)))
 	if err != nil {
+		log.Errorf("failed to set accepted blob size for session %s in redis, error: %v", sessionID, err)
 		return err
 	}
 
 	if reply != "OK" {
-		return fmt.Errorf("bad reply value")
+		err := fmt.Errorf("failed to set accepted blob size for session %s in redis, error: expected reply is 'OK' but actual it is '%s'", sessionID, reply)
+		log.Errorf("%s", err.Error())
+		return err
 	}
 
 	return nil
@@ -314,7 +320,7 @@ func (c *controller) GetAcceptedBlobSize(sessionID string) (int64, error) {
 	conn := redislib.DefaultPool().Get()
 	defer conn.Close()
 
-	key := fmt.Sprintf("upload:%s:size", sessionID)
+	key := blobSizeKey(sessionID)
 	size, err := redis.Int64(conn.Do("GET", key))
 	if err != nil {
 		if err == redis.ErrNil {
@@ -358,3 +364,7 @@ func (c *controller) Update(ctx context.Context, blob *blob.Blob) error {
 func (c *controller) Delete(ctx context.Context, id int64) error {
 	return c.blobMgr.Delete(ctx, id)
 }
+
+func blobSizeKey(sessionID string) string {
+	return fmt.Sprintf("upload:%s:size", sessionID)
+}

src/controller/blob/controller_test.go

@@ -17,9 +17,11 @@ package blob
 import (
 	"context"
 	"fmt"
+	"testing"
+	"time"
+
 	"github.com/goharbor/harbor/src/lib/errors"
 	"github.com/goharbor/harbor/src/pkg/blob/models"
-	"testing"
 
 	"github.com/docker/distribution/manifest/schema2"
 	"github.com/goharbor/harbor/src/pkg/blob"
@@ -258,17 +260,42 @@ func (suite *ControllerTestSuite) TestSync() {
 }
 
 func (suite *ControllerTestSuite) TestGetSetAcceptedBlobSize() {
-	sessionID := uuid.New().String()
+	{
+		sessionID := uuid.New().String()
 
-	size, err := Ctl.GetAcceptedBlobSize(sessionID)
+		size, err := Ctl.GetAcceptedBlobSize(sessionID)
-	suite.Nil(err)
+		suite.Nil(err)
-	suite.Equal(int64(0), size)
+		suite.Equal(int64(0), size)
 
-	suite.Nil(Ctl.SetAcceptedBlobSize(sessionID, 100))
+		suite.Nil(Ctl.SetAcceptedBlobSize(sessionID, 100))
 
-	size, err = Ctl.GetAcceptedBlobSize(sessionID)
+		size, err = Ctl.GetAcceptedBlobSize(sessionID)
-	suite.Nil(err)
+		suite.Nil(err)
-	suite.Equal(int64(100), size)
+		suite.Equal(int64(100), size)
+	}
+
+	{
+		// test blob size expired
+		ctl := &controller{blobSizeExpiration: time.Second * 5}
+
+		sessionID := uuid.New().String()
+
+		size, err := ctl.GetAcceptedBlobSize(sessionID)
+		suite.Nil(err)
+		suite.Equal(int64(0), size)
+
+		suite.Nil(ctl.SetAcceptedBlobSize(sessionID, 100))
+
+		size, err = ctl.GetAcceptedBlobSize(sessionID)
+		suite.Nil(err)
+		suite.Equal(int64(100), size)
+
+		time.Sleep(time.Second * 10)
+
+		size, err = ctl.GetAcceptedBlobSize(sessionID)
+		suite.Nil(err)
+		suite.Equal(int64(0), size)
+	}
 }
 
 func (suite *ControllerTestSuite) TestTouch() {

src/jobservice/job/impl/gc/garbage_collection.go

@@ -45,7 +45,6 @@ const (
 	dialWriteTimeout      = 10 * time.Second
 	blobPrefix            = "blobs::*"
 	repoPrefix            = "repository::*"
-	uploadSizePattern     = "upload:*:size"
 )
 
 // GarbageCollector is the struct to run registry's garbage collection
@@ -348,8 +347,7 @@ func (gc *GarbageCollector) cleanCache() error {
 	// sample of keys in registry redis:
 	// 1) "blobs::sha256:1a6fd470b9ce10849be79e99529a88371dff60c60aab424c077007f6979b4812"
 	// 2) "repository::library/hello-world::blobs::sha256:4ab4c602aa5eed5528a6620ff18a1dc4faef0e1ab3a5eddeddb410714478c67f"
-	// 3) "upload:fbd2e0a3-262d-40bb-abe4-2f43aa6f9cda:size"
+	patterns := []string{blobPrefix, repoPrefix}
-	patterns := []string{blobPrefix, repoPrefix, uploadSizePattern}
 	for _, pattern := range patterns {
 		if err := delKeys(con, pattern); err != nil {
 			gc.logger.Errorf("failed to clean registry cache %v, pattern %s", err, pattern)