mirror of
https://github.com/goharbor/harbor.git
synced 2024-12-18 22:57:38 +01:00
update blob controller & manager (#12101)
* update blob controller & manager 1, add two more attributes, version, update_time and status 2, add delete and fresh update time method in blob mgr & ctr. Signed-off-by: wang yan <wangyan@vmware.com>
This commit is contained in:
commit
53044da28f
@ -29,3 +29,9 @@ CREATE TABLE IF NOT EXISTS task (
|
||||
end_time timestamp,
|
||||
FOREIGN KEY (execution_id) REFERENCES execution(id)
|
||||
);
|
||||
|
||||
ALTER TABLE blob ADD COLUMN IF NOT EXISTS update_time timestamp default CURRENT_TIMESTAMP;
|
||||
ALTER TABLE blob ADD COLUMN IF NOT EXISTS status varchar(255);
|
||||
ALTER TABLE blob ADD COLUMN IF NOT EXISTS version BIGINT default 0;
|
||||
CREATE INDEX IF NOT EXISTS idx_status ON blob (status);
|
||||
CREATE INDEX IF NOT EXISTS idx_version ON blob (version);
|
@ -34,7 +34,6 @@ func init() {
|
||||
new(OIDCUser),
|
||||
new(NotificationPolicy),
|
||||
new(NotificationJob),
|
||||
new(Blob),
|
||||
new(ProjectBlob),
|
||||
new(ArtifactAndBlob),
|
||||
new(CVEWhitelist),
|
||||
|
@ -1,34 +0,0 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/docker/distribution/manifest/schema2"
|
||||
)
|
||||
|
||||
// Blob holds the details of a blob.
|
||||
type Blob struct {
|
||||
ID int64 `orm:"pk;auto;column(id)" json:"id"`
|
||||
Digest string `orm:"column(digest)" json:"digest"`
|
||||
ContentType string `orm:"column(content_type)" json:"content_type"`
|
||||
Size int64 `orm:"column(size)" json:"size"`
|
||||
CreationTime time.Time `orm:"column(creation_time);auto_now_add" json:"creation_time"`
|
||||
}
|
||||
|
||||
// TableName ...
|
||||
func (b *Blob) TableName() string {
|
||||
return "blob"
|
||||
}
|
||||
|
||||
// IsForeignLayer returns true if the blob is foreign layer
|
||||
func (b *Blob) IsForeignLayer() bool {
|
||||
return b.ContentType == schema2.MediaTypeForeignLayer
|
||||
}
|
||||
|
||||
// BlobQuery ...
|
||||
type BlobQuery struct {
|
||||
Digest string
|
||||
ContentType string
|
||||
Digests []string
|
||||
Pagination
|
||||
}
|
@ -17,15 +17,14 @@ package blob
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/garyburd/redigo/redis"
|
||||
"github.com/goharbor/harbor/src/common/models"
|
||||
util "github.com/goharbor/harbor/src/common/utils/redis"
|
||||
"github.com/goharbor/harbor/src/lib/errors"
|
||||
"github.com/goharbor/harbor/src/lib/log"
|
||||
"github.com/goharbor/harbor/src/lib/orm"
|
||||
"github.com/goharbor/harbor/src/pkg/blob"
|
||||
blob_models "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
)
|
||||
|
||||
var (
|
||||
@ -75,6 +74,15 @@ type Controller interface {
|
||||
|
||||
// GetAcceptedBlobSize returns the accepted size of stream upload blob.
|
||||
GetAcceptedBlobSize(sessionID string) (int64, error)
|
||||
|
||||
// Touch updates the blob status to StatusNone and increase version every time.
|
||||
Touch(ctx context.Context, blob *blob.Blob) error
|
||||
|
||||
// Update updates the blob, it cannot handle blob status transitions.
|
||||
Update(ctx context.Context, blob *blob.Blob) error
|
||||
|
||||
// Delete deletes the blob by its id
|
||||
Delete(ctx context.Context, id int64) error
|
||||
}
|
||||
|
||||
// NewController creates an instance of the default repository controller
|
||||
@ -184,7 +192,7 @@ func (c *controller) FindMissingAssociationsForProject(ctx context.Context, proj
|
||||
associated[blob.Digest] = true
|
||||
}
|
||||
|
||||
var results []*models.Blob
|
||||
var results []*blob.Blob
|
||||
for _, blob := range blobs {
|
||||
if !associated[blob.Digest] {
|
||||
results = append(results, blob)
|
||||
@ -260,7 +268,7 @@ func (c *controller) Sync(ctx context.Context, references []distribution.Descrip
|
||||
if len(updating) > 0 {
|
||||
orm.WithTransaction(func(ctx context.Context) error {
|
||||
for _, blob := range updating {
|
||||
if err := c.blobMgr.Update(ctx, blob); err != nil {
|
||||
if err := c.Update(ctx, blob); err != nil {
|
||||
log.G(ctx).Warningf("Failed to update blob %s, error: %v", blob.Digest, err)
|
||||
return err
|
||||
}
|
||||
@ -314,3 +322,23 @@ func (c *controller) GetAcceptedBlobSize(sessionID string) (int64, error) {
|
||||
|
||||
return size, nil
|
||||
}
|
||||
|
||||
func (c *controller) Touch(ctx context.Context, blob *blob.Blob) error {
|
||||
blob.Status = blob_models.StatusNone
|
||||
count, err := c.blobMgr.UpdateBlobStatus(ctx, blob)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if count == 0 {
|
||||
return errors.New(nil).WithMessage(fmt.Sprintf("no blob item is updated to StatusNone, id:%d, digest:%s", blob.ID, blob.Digest)).WithCode(errors.NotFoundCode)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *controller) Update(ctx context.Context, blob *blob.Blob) error {
|
||||
return c.blobMgr.Update(ctx, blob)
|
||||
}
|
||||
|
||||
func (c *controller) Delete(ctx context.Context, id int64) error {
|
||||
return c.blobMgr.Delete(ctx, id)
|
||||
}
|
||||
|
@ -17,6 +17,8 @@ package blob
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/goharbor/harbor/src/lib/errors"
|
||||
"github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
"testing"
|
||||
|
||||
"github.com/docker/distribution/manifest/schema2"
|
||||
@ -27,6 +29,8 @@ import (
|
||||
blobtesting "github.com/goharbor/harbor/src/testing/pkg/blob"
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/suite"
|
||||
|
||||
pkg_blob "github.com/goharbor/harbor/src/pkg/blob"
|
||||
)
|
||||
|
||||
type ControllerTestSuite struct {
|
||||
@ -267,6 +271,46 @@ func (suite *ControllerTestSuite) TestGetSetAcceptedBlobSize() {
|
||||
suite.Equal(int64(100), size)
|
||||
}
|
||||
|
||||
func (suite *ControllerTestSuite) TestTouch() {
|
||||
ctx := suite.Context()
|
||||
|
||||
err := Ctl.Touch(ctx, &blob.Blob{
|
||||
Status: models.StatusNone,
|
||||
})
|
||||
suite.NotNil(err)
|
||||
suite.True(errors.IsNotFoundErr(err))
|
||||
|
||||
digest := suite.prepareBlob()
|
||||
blob, err := Ctl.Get(ctx, digest)
|
||||
suite.Nil(err)
|
||||
blob.Status = models.StatusDelete
|
||||
_, err = pkg_blob.Mgr.UpdateBlobStatus(suite.Context(), blob)
|
||||
suite.Nil(err)
|
||||
|
||||
err = Ctl.Touch(ctx, blob)
|
||||
suite.Nil(err)
|
||||
suite.Equal(blob.Status, models.StatusNone)
|
||||
}
|
||||
|
||||
func (suite *ControllerTestSuite) TestDelete() {
|
||||
ctx := suite.Context()
|
||||
|
||||
digest := suite.DigestString()
|
||||
_, err := Ctl.Ensure(ctx, digest, "application/octet-stream", 100)
|
||||
suite.Nil(err)
|
||||
|
||||
blob, err := Ctl.Get(ctx, digest)
|
||||
suite.Nil(err)
|
||||
suite.Equal(digest, blob.Digest)
|
||||
|
||||
err = Ctl.Delete(ctx, blob.ID)
|
||||
suite.Nil(err)
|
||||
|
||||
exist, err := Ctl.Exist(ctx, digest)
|
||||
suite.Nil(err)
|
||||
suite.False(exist)
|
||||
}
|
||||
|
||||
func TestControllerTestSuite(t *testing.T) {
|
||||
suite.Run(t, &ControllerTestSuite{})
|
||||
}
|
||||
|
@ -7,6 +7,7 @@ import (
|
||||
"github.com/goharbor/harbor/src/jobservice/job"
|
||||
"github.com/goharbor/harbor/src/pkg/artifact"
|
||||
"github.com/goharbor/harbor/src/pkg/artifactrash/model"
|
||||
pkg_blob "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
artifacttesting "github.com/goharbor/harbor/src/testing/controller/artifact"
|
||||
projecttesting "github.com/goharbor/harbor/src/testing/controller/project"
|
||||
mockjobservice "github.com/goharbor/harbor/src/testing/jobservice"
|
||||
@ -93,7 +94,7 @@ func (suite *gcTestSuite) TestRemoveUntaggedBlobs() {
|
||||
},
|
||||
}, nil)
|
||||
|
||||
mock.OnAnything(suite.blobMgr, "List").Return([]*models.Blob{
|
||||
mock.OnAnything(suite.blobMgr, "List").Return([]*pkg_blob.Blob{
|
||||
{
|
||||
ID: 1234,
|
||||
Digest: "sha256:1234",
|
||||
@ -203,7 +204,7 @@ func (suite *gcTestSuite) TestRun() {
|
||||
},
|
||||
}, nil)
|
||||
|
||||
mock.OnAnything(suite.blobMgr, "List").Return([]*models.Blob{
|
||||
mock.OnAnything(suite.blobMgr, "List").Return([]*pkg_blob.Blob{
|
||||
{
|
||||
ID: 12345,
|
||||
Digest: "sha256:12345",
|
||||
|
@ -17,9 +17,12 @@ package dao
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/goharbor/harbor/src/lib/errors"
|
||||
"github.com/goharbor/harbor/src/lib/log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
beego_orm "github.com/astaxie/beego/orm"
|
||||
"github.com/docker/distribution/manifest/schema2"
|
||||
"github.com/goharbor/harbor/src/lib/orm"
|
||||
"github.com/goharbor/harbor/src/lib/q"
|
||||
@ -49,6 +52,9 @@ type DAO interface {
|
||||
// UpdateBlob update blob
|
||||
UpdateBlob(ctx context.Context, blob *models.Blob) error
|
||||
|
||||
// UpdateBlob update blob status
|
||||
UpdateBlobStatus(ctx context.Context, blob *models.Blob) (int64, error)
|
||||
|
||||
// ListBlobs list blobs by query
|
||||
ListBlobs(ctx context.Context, params models.ListParams) ([]*models.Blob, error)
|
||||
|
||||
@ -66,6 +72,9 @@ type DAO interface {
|
||||
|
||||
// ExistProjectBlob returns true when ProjectBlob exist
|
||||
ExistProjectBlob(ctx context.Context, projectID int64, blobDigest string) (bool, error)
|
||||
|
||||
// DeleteBlob delete blob
|
||||
DeleteBlob(ctx context.Context, id int64) (err error)
|
||||
}
|
||||
|
||||
// New returns an instance of the default DAO
|
||||
@ -162,13 +171,58 @@ func (d *dao) GetBlobByDigest(ctx context.Context, digest string) (*models.Blob,
|
||||
return blob, nil
|
||||
}
|
||||
|
||||
func (d *dao) UpdateBlobStatus(ctx context.Context, blob *models.Blob) (int64, error) {
|
||||
o, err := orm.FromContext(ctx)
|
||||
if err != nil {
|
||||
return -1, err
|
||||
}
|
||||
|
||||
// each update will auto increase version and update time
|
||||
data := make(beego_orm.Params)
|
||||
data["version"] = beego_orm.ColValue(beego_orm.ColAdd, 1)
|
||||
data["update_time"] = time.Now()
|
||||
data["status"] = blob.Status
|
||||
|
||||
qt := o.QueryTable(&models.Blob{})
|
||||
cond := beego_orm.NewCondition()
|
||||
var c *beego_orm.Condition
|
||||
|
||||
// In the multiple blob head scenario, if one request success mark the blob from StatusDelete to StatusNone, then version should increase one.
|
||||
// in the meantime, the other requests tries to do the same thing, use 'where version >= blob.version' can handle it.
|
||||
if blob.Status == models.StatusNone {
|
||||
c = cond.And("version__gte", blob.Version)
|
||||
} else {
|
||||
c = cond.And("version", blob.Version)
|
||||
}
|
||||
|
||||
/*
|
||||
generated simple sql string.
|
||||
UPDATE "blob" SET "version" = "version" + $1, "update_time" = $2, "status" = $3
|
||||
WHERE "id" IN ( SELECT T0."id" FROM "blob" T0 WHERE T0."version" >= $4 AND T0."id" = $5 AND T0."status" IN ('delete', 'deleting') )
|
||||
*/
|
||||
|
||||
count, err := qt.SetCond(c).Filter("id", blob.ID).
|
||||
Filter("status__in", models.StatusMap[blob.Status]).
|
||||
Update(data)
|
||||
if err != nil {
|
||||
return count, err
|
||||
}
|
||||
if count == 0 {
|
||||
log.Warningf("no blob is updated according to query condition, id: %d, status_in, %v", blob.ID, models.StatusMap[blob.Status])
|
||||
return 0, nil
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
// UpdateBlob cannot handle the status change and version increase, for handling blob status change, please call
|
||||
// for the UpdateBlobStatus.
|
||||
func (d *dao) UpdateBlob(ctx context.Context, blob *models.Blob) error {
|
||||
o, err := orm.FromContext(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
_, err = o.Update(blob)
|
||||
blob.UpdateTime = time.Now()
|
||||
_, err = o.Update(blob, "size", "content_type", "update_time")
|
||||
return err
|
||||
}
|
||||
|
||||
@ -318,3 +372,20 @@ func (d *dao) DeleteProjectBlob(ctx context.Context, projectID int64, blobIDs ..
|
||||
_, err = qs.Delete()
|
||||
return err
|
||||
}
|
||||
|
||||
func (d *dao) DeleteBlob(ctx context.Context, id int64) error {
|
||||
ormer, err := orm.FromContext(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
n, err := ormer.Delete(&models.Blob{
|
||||
ID: id,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n == 0 {
|
||||
return errors.NotFoundError(nil).WithMessage("blob %d not found", id)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
@ -15,11 +15,11 @@
|
||||
package dao
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/goharbor/harbor/src/lib/errors"
|
||||
"github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
htesting "github.com/goharbor/harbor/src/testing"
|
||||
"github.com/stretchr/testify/suite"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type DaoTestSuite struct {
|
||||
@ -129,20 +129,60 @@ func (suite *DaoTestSuite) TestUpdateBlob() {
|
||||
digest := suite.DigestString()
|
||||
|
||||
suite.dao.CreateBlob(ctx, &models.Blob{Digest: digest})
|
||||
|
||||
blob, err := suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(0), blob.Size)
|
||||
}
|
||||
|
||||
blob.Size = 100
|
||||
|
||||
if suite.Nil(suite.dao.UpdateBlob(ctx, blob)) {
|
||||
blob, err := suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(100), blob.Size)
|
||||
suite.Equal(int64(0), blob.Version)
|
||||
}
|
||||
}
|
||||
|
||||
blob.Status = "deleting"
|
||||
suite.Nil(suite.dao.UpdateBlob(ctx, blob), "cannot be updated.")
|
||||
blob, err = suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(0), blob.Version)
|
||||
suite.Equal(models.StatusNone, blob.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func (suite *DaoTestSuite) TestUpdateBlobStatus() {
|
||||
ctx := suite.Context()
|
||||
|
||||
digest := suite.DigestString()
|
||||
|
||||
suite.dao.CreateBlob(ctx, &models.Blob{Digest: digest})
|
||||
blob, err := suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(0), blob.Size)
|
||||
}
|
||||
|
||||
// StatusNone cannot be updated to StatusDeleting directly
|
||||
blob.Status = models.StatusDeleting
|
||||
count, err := suite.dao.UpdateBlobStatus(ctx, blob)
|
||||
suite.Nil(err)
|
||||
suite.Equal(int64(0), count)
|
||||
blob, err = suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(0), blob.Version)
|
||||
suite.Equal(models.StatusNone, blob.Status)
|
||||
}
|
||||
|
||||
blob.Status = models.StatusDelete
|
||||
count, err = suite.dao.UpdateBlobStatus(ctx, blob)
|
||||
suite.Nil(err)
|
||||
suite.Equal(int64(1), count)
|
||||
blob, err = suite.dao.GetBlobByDigest(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
suite.Equal(int64(1), blob.Version)
|
||||
suite.Equal(models.StatusDelete, blob.Status)
|
||||
}
|
||||
}
|
||||
|
||||
func (suite *DaoTestSuite) TestListBlobs() {
|
||||
@ -322,6 +362,20 @@ func (suite *DaoTestSuite) TestDeleteProjectBlob() {
|
||||
}
|
||||
}
|
||||
|
||||
func (suite *DaoTestSuite) TestDelete() {
|
||||
ctx := suite.Context()
|
||||
|
||||
err := suite.dao.DeleteBlob(ctx, 100021)
|
||||
suite.Require().NotNil(err)
|
||||
suite.True(errors.IsErr(err, errors.NotFoundCode))
|
||||
|
||||
digest := suite.DigestString()
|
||||
id, err := suite.dao.CreateBlob(ctx, &models.Blob{Digest: digest})
|
||||
suite.Nil(err)
|
||||
err = suite.dao.DeleteBlob(ctx, id)
|
||||
suite.Require().Nil(err)
|
||||
}
|
||||
|
||||
func TestDaoTestSuite(t *testing.T) {
|
||||
suite.Run(t, &DaoTestSuite{})
|
||||
}
|
||||
|
@ -16,7 +16,6 @@ package blob
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/goharbor/harbor/src/pkg/blob/dao"
|
||||
"github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
)
|
||||
@ -58,8 +57,14 @@ type Manager interface {
|
||||
// Update the blob
|
||||
Update(ctx context.Context, blob *Blob) error
|
||||
|
||||
// Update the blob status
|
||||
UpdateBlobStatus(ctx context.Context, blob *models.Blob) (int64, error)
|
||||
|
||||
// List returns blobs by params
|
||||
List(ctx context.Context, params ListParams) ([]*Blob, error)
|
||||
|
||||
// DeleteBlob delete blob
|
||||
Delete(ctx context.Context, id int64) (err error)
|
||||
}
|
||||
|
||||
type manager struct {
|
||||
@ -112,10 +117,18 @@ func (m *manager) Update(ctx context.Context, blob *Blob) error {
|
||||
return m.dao.UpdateBlob(ctx, blob)
|
||||
}
|
||||
|
||||
func (m *manager) UpdateBlobStatus(ctx context.Context, blob *models.Blob) (int64, error) {
|
||||
return m.dao.UpdateBlobStatus(ctx, blob)
|
||||
}
|
||||
|
||||
func (m *manager) List(ctx context.Context, params ListParams) ([]*Blob, error) {
|
||||
return m.dao.ListBlobs(ctx, params)
|
||||
}
|
||||
|
||||
func (m *manager) Delete(ctx context.Context, id int64) error {
|
||||
return m.dao.DeleteBlob(ctx, id)
|
||||
}
|
||||
|
||||
// NewManager returns blob manager
|
||||
func NewManager() Manager {
|
||||
return &manager{dao: dao.New()}
|
||||
|
@ -16,10 +16,11 @@ package blob
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
htesting "github.com/goharbor/harbor/src/testing"
|
||||
"github.com/stretchr/testify/suite"
|
||||
"testing"
|
||||
|
||||
"github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
)
|
||||
|
||||
type ManagerTestSuite struct {
|
||||
@ -200,6 +201,7 @@ func (suite *ManagerTestSuite) TestUpdate() {
|
||||
suite.Equal(digest, blob.Digest)
|
||||
suite.Equal("media type", blob.ContentType)
|
||||
suite.Equal(int64(1000), blob.Size)
|
||||
suite.Equal(models.StatusNone, blob.Status)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -256,6 +258,38 @@ func (suite *ManagerTestSuite) TestListByArtifact() {
|
||||
suite.Len(blobs, 3)
|
||||
}
|
||||
|
||||
func (suite *ManagerTestSuite) TestDelete() {
|
||||
ctx := suite.Context()
|
||||
digest := suite.DigestString()
|
||||
blobID, err := Mgr.Create(ctx, digest, "media type", 100)
|
||||
suite.Nil(err)
|
||||
|
||||
err = Mgr.Delete(ctx, blobID)
|
||||
suite.Nil(err)
|
||||
}
|
||||
|
||||
func (suite *ManagerTestSuite) TestUpdateStatus() {
|
||||
ctx := suite.Context()
|
||||
|
||||
digest := suite.DigestString()
|
||||
_, err := Mgr.Create(ctx, digest, "media type", 100)
|
||||
suite.Nil(err)
|
||||
|
||||
blob, err := Mgr.Get(ctx, digest)
|
||||
if suite.Nil(err) {
|
||||
blob.Status = models.StatusDelete
|
||||
_, err := Mgr.UpdateBlobStatus(ctx, blob)
|
||||
suite.Nil(err)
|
||||
|
||||
{
|
||||
blob, err := Mgr.Get(ctx, digest)
|
||||
suite.Nil(err)
|
||||
suite.Equal(digest, blob.Digest)
|
||||
suite.Equal(models.StatusDelete, blob.Status)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerTestSuite(t *testing.T) {
|
||||
suite.Run(t, &ManagerTestSuite{})
|
||||
}
|
||||
|
@ -15,16 +15,73 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"github.com/astaxie/beego/orm"
|
||||
"github.com/docker/distribution/manifest/schema2"
|
||||
"github.com/goharbor/harbor/src/common/models"
|
||||
"time"
|
||||
)
|
||||
|
||||
// TODO: move ArtifactAndBlob, Blob and ProjectBlob to here
|
||||
func init() {
|
||||
orm.RegisterModel(&Blob{})
|
||||
}
|
||||
|
||||
// TODO: move ArtifactAndBlob, ProjectBlob to here
|
||||
|
||||
// ArtifactAndBlob alias ArtifactAndBlob model
|
||||
type ArtifactAndBlob = models.ArtifactAndBlob
|
||||
|
||||
// Blob alias Blob model
|
||||
type Blob = models.Blob
|
||||
/*
|
||||
the status are used for Garbage Collection
|
||||
StatusNone, the blob is using in Harbor as normal.
|
||||
StatusDelete, the blob is marked as GC candidate.
|
||||
StatusDeleting, the blob undergo a GC blob deletion.
|
||||
StatusDeleteFailed, the blob is failed to delete from the backend storage.
|
||||
|
||||
The status transitions
|
||||
StatusNone -> StatusDelete : Mark the blob as candidate.
|
||||
StatusDelete -> StatusDeleting : Select the blob and call the API to delete asset in the backend storage.
|
||||
StatusDeleting -> Trash : Delete success from the backend storage.
|
||||
StatusDelete -> StatusNone : Client asks the existence of blob, remove it from the candidate.
|
||||
StatusDelete -> StatusDeleteFailed : The storage driver returns fail when to delete the real data from the configurated file system.
|
||||
StatusDeleteFailed -> StatusNone : The delete failed blobs can be pushed again, and back to normal.
|
||||
StatusDeleteFailed -> StatusDelete : The delete failed blobs should be in the candidate.
|
||||
*/
|
||||
const (
|
||||
StatusNone = ""
|
||||
StatusDelete = "delete"
|
||||
StatusDeleting = "deleting"
|
||||
StatusDeleteFailed = "deletefailed"
|
||||
)
|
||||
|
||||
// StatusMap key is the target status, values are the accept source status. For example, only StatusNone and StatusDeleteFailed can be convert to StatusDelete.
|
||||
var StatusMap = map[string][]string{
|
||||
StatusNone: {StatusNone, StatusDelete, StatusDeleteFailed},
|
||||
StatusDelete: {StatusNone, StatusDeleteFailed},
|
||||
StatusDeleting: {StatusDelete},
|
||||
StatusDeleteFailed: {StatusDeleting},
|
||||
}
|
||||
|
||||
// Blob holds the details of a blob.
|
||||
type Blob struct {
|
||||
ID int64 `orm:"pk;auto;column(id)" json:"id"`
|
||||
Digest string `orm:"column(digest)" json:"digest"`
|
||||
ContentType string `orm:"column(content_type)" json:"content_type"`
|
||||
Size int64 `orm:"column(size)" json:"size"`
|
||||
Status string `orm:"column(status)" json:"status"`
|
||||
UpdateTime time.Time `orm:"column(update_time);auto_now_add" json:"update_time"`
|
||||
Version int64 `orm:"column(version)" json:"version"`
|
||||
CreationTime time.Time `orm:"column(creation_time);auto_now_add" json:"creation_time"`
|
||||
}
|
||||
|
||||
// TableName ...
|
||||
func (b *Blob) TableName() string {
|
||||
return "blob"
|
||||
}
|
||||
|
||||
// IsForeignLayer returns true if the blob is foreign layer
|
||||
func (b *Blob) IsForeignLayer() bool {
|
||||
return b.ContentType == schema2.MediaTypeForeignLayer
|
||||
}
|
||||
|
||||
// ProjectBlob alias ProjectBlob model
|
||||
type ProjectBlob = models.ProjectBlob
|
||||
|
54
src/server/middleware/blob/head_blob.go
Normal file
54
src/server/middleware/blob/head_blob.go
Normal file
@ -0,0 +1,54 @@
|
||||
package blob
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"github.com/goharbor/harbor/src/controller/blob"
|
||||
"github.com/goharbor/harbor/src/lib"
|
||||
"github.com/goharbor/harbor/src/lib/errors"
|
||||
lib_http "github.com/goharbor/harbor/src/lib/http"
|
||||
"github.com/goharbor/harbor/src/lib/log"
|
||||
blob_models "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
"github.com/goharbor/harbor/src/server/middleware"
|
||||
"github.com/goharbor/harbor/src/server/middleware/requestid"
|
||||
"net/http"
|
||||
)
|
||||
|
||||
// HeadBlobMiddleware intercept the head blob request
|
||||
func HeadBlobMiddleware() func(http.Handler) http.Handler {
|
||||
return middleware.New(func(rw http.ResponseWriter, req *http.Request, next http.Handler) {
|
||||
if err := handleHead(req); err != nil {
|
||||
lib_http.SendError(rw, err)
|
||||
return
|
||||
}
|
||||
next.ServeHTTP(rw, req)
|
||||
})
|
||||
}
|
||||
|
||||
// handleHead ...
|
||||
func handleHead(req *http.Request) error {
|
||||
none := lib.ArtifactInfo{}
|
||||
// for head blob, the GetArtifactInfo is actually get the information of blob.
|
||||
blobInfo := lib.GetArtifactInfo(req.Context())
|
||||
if blobInfo == none {
|
||||
return errors.New("cannot get the blob information from request context").WithCode(errors.NotFoundCode)
|
||||
}
|
||||
|
||||
bb, err := blob.Ctl.Get(req.Context(), blobInfo.Digest)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch bb.Status {
|
||||
case blob_models.StatusNone, blob_models.StatusDelete:
|
||||
err := blob.Ctl.Touch(req.Context(), bb)
|
||||
if err != nil {
|
||||
log.Errorf("failed to update blob: %s status to StatusNone, error:%v", blobInfo.Digest, err)
|
||||
return errors.Wrapf(err, fmt.Sprintf("the request id is: %s", req.Header.Get(requestid.HeaderXRequestID)))
|
||||
}
|
||||
case blob_models.StatusDeleting, blob_models.StatusDeleteFailed:
|
||||
return errors.New(nil).WithMessage(fmt.Sprintf("the asking blob is in GC, mark it as non existing, request id: %s", req.Header.Get(requestid.HeaderXRequestID))).WithCode(errors.NotFoundCode)
|
||||
default:
|
||||
return errors.New(nil).WithMessage(fmt.Sprintf("wrong blob status, %s", bb.Status))
|
||||
}
|
||||
return nil
|
||||
}
|
89
src/server/middleware/blob/head_blob_test.go
Normal file
89
src/server/middleware/blob/head_blob_test.go
Normal file
@ -0,0 +1,89 @@
|
||||
package blob
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
beego_orm "github.com/astaxie/beego/orm"
|
||||
"github.com/goharbor/harbor/src/controller/blob"
|
||||
"github.com/goharbor/harbor/src/lib"
|
||||
"github.com/goharbor/harbor/src/lib/orm"
|
||||
pkg_blob "github.com/goharbor/harbor/src/pkg/blob"
|
||||
blob_models "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
htesting "github.com/goharbor/harbor/src/testing"
|
||||
"github.com/stretchr/testify/suite"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
|
||||
type HeadBlobUploadMiddlewareTestSuite struct {
|
||||
htesting.Suite
|
||||
}
|
||||
|
||||
func (suite *HeadBlobUploadMiddlewareTestSuite) SetupSuite() {
|
||||
suite.Suite.SetupSuite()
|
||||
suite.Suite.ClearTables = []string{"blob"}
|
||||
}
|
||||
|
||||
func (suite *HeadBlobUploadMiddlewareTestSuite) makeRequest(projectName, digest string) *http.Request {
|
||||
req := httptest.NewRequest("HEAD", fmt.Sprintf("/v2/%s/blobs/%s", projectName, digest), nil)
|
||||
info := lib.ArtifactInfo{
|
||||
Repository: fmt.Sprintf("%s/photon", projectName),
|
||||
Reference: "2.0",
|
||||
Tag: "2.0",
|
||||
Digest: digest,
|
||||
}
|
||||
*req = *(req.WithContext(orm.NewContext(req.Context(), beego_orm.NewOrm())))
|
||||
*req = *(req.WithContext(lib.WithArtifactInfo(req.Context(), info)))
|
||||
return req
|
||||
}
|
||||
|
||||
func (suite *HeadBlobUploadMiddlewareTestSuite) TestHeadBlobStatusNone() {
|
||||
suite.WithProject(func(projectID int64, projectName string) {
|
||||
digest := suite.DigestString()
|
||||
|
||||
_, err := blob.Ctl.Ensure(suite.Context(), digest, "application/octet-stream", 512)
|
||||
suite.Nil(err)
|
||||
|
||||
req := suite.makeRequest(projectName, digest)
|
||||
res := httptest.NewRecorder()
|
||||
next := suite.NextHandler(http.StatusOK, map[string]string{"Docker-Content-Digest": digest})
|
||||
HeadBlobMiddleware()(next).ServeHTTP(res, req)
|
||||
suite.Equal(http.StatusOK, res.Code)
|
||||
|
||||
blob, err := blob.Ctl.Get(suite.Context(), digest)
|
||||
suite.Nil(err)
|
||||
suite.Equal(digest, blob.Digest)
|
||||
suite.Equal(blob_models.StatusNone, blob.Status)
|
||||
})
|
||||
}
|
||||
|
||||
func (suite *HeadBlobUploadMiddlewareTestSuite) TestHeadBlobStatusDeleting() {
|
||||
suite.WithProject(func(projectID int64, projectName string) {
|
||||
digest := suite.DigestString()
|
||||
|
||||
id, err := blob.Ctl.Ensure(suite.Context(), digest, "application/octet-stream", 512)
|
||||
suite.Nil(err)
|
||||
|
||||
// status-none -> status-delete -> status-deleting
|
||||
_, err = pkg_blob.Mgr.UpdateBlobStatus(suite.Context(), &blob_models.Blob{ID: id, Status: blob_models.StatusDelete})
|
||||
suite.Nil(err)
|
||||
_, err = pkg_blob.Mgr.UpdateBlobStatus(suite.Context(), &blob_models.Blob{ID: id, Status: blob_models.StatusDeleting, Version: 1})
|
||||
suite.Nil(err)
|
||||
|
||||
req := suite.NewRequest(http.MethodHead, fmt.Sprintf("/v2/%s/blobs/%s", projectName, digest), nil)
|
||||
res := httptest.NewRecorder()
|
||||
|
||||
next := suite.NextHandler(http.StatusOK, map[string]string{"Docker-Content-Digest": digest})
|
||||
HeadBlobMiddleware()(next).ServeHTTP(res, req)
|
||||
suite.Equal(http.StatusNotFound, res.Code)
|
||||
|
||||
blob, err := blob.Ctl.Get(suite.Context(), digest)
|
||||
suite.Nil(err)
|
||||
suite.Equal(digest, blob.Digest)
|
||||
suite.Equal(blob_models.StatusDeleting, blob.Status)
|
||||
})
|
||||
}
|
||||
|
||||
func TestHeadBlobUploadMiddlewareTestSuite(t *testing.T) {
|
||||
suite.Run(t, &HeadBlobUploadMiddlewareTestSuite{})
|
||||
}
|
@ -22,7 +22,7 @@ var (
|
||||
V2ManifestURLRe = regexp.MustCompile(fmt.Sprintf(`^/v2/(?P<%s>%s)/manifests/(?P<%s>%s|%s)$`, RepositorySubexp, reference.NameRegexp.String(), ReferenceSubexp, reference.TagRegexp.String(), digest.DigestRegexp.String()))
|
||||
// V2TagListURLRe is the regular expression for matching request to v2 handler to list tags
|
||||
V2TagListURLRe = regexp.MustCompile(fmt.Sprintf(`^/v2/(?P<%s>%s)/tags/list`, RepositorySubexp, reference.NameRegexp.String()))
|
||||
// V2BlobURLRe is the regular expression for matching request to v2 handler to retrieve delete a blob
|
||||
// V2BlobURLRe is the regular expression for matching request to v2 handler to retrieve head/delete a blob
|
||||
V2BlobURLRe = regexp.MustCompile(fmt.Sprintf(`^/v2/(?P<%s>%s)/blobs/(?P<%s>%s)$`, RepositorySubexp, reference.NameRegexp.String(), DigestSubexp, digest.DigestRegexp.String()))
|
||||
// V2BlobUploadURLRe is the regular expression for matching the request to v2 handler to upload a blob, the upload uuid currently is not put into a group
|
||||
V2BlobUploadURLRe = regexp.MustCompile(fmt.Sprintf(`^/v2/(?P<%s>%s)/blobs/uploads[/a-zA-Z0-9\-_\.=]*$`, RepositorySubexp, reference.NameRegexp.String()))
|
||||
|
@ -85,6 +85,11 @@ func RegisterRoutes() {
|
||||
Middleware(quota.PutBlobUploadMiddleware()).
|
||||
Middleware(blob.PutBlobUploadMiddleware()).
|
||||
Handler(proxy)
|
||||
root.NewRoute().
|
||||
Method(http.MethodHead).
|
||||
Path("/*/blobs/:digest").
|
||||
Middleware(blob.HeadBlobMiddleware()).
|
||||
Handler(proxy)
|
||||
// others
|
||||
root.NewRoute().Path("/*").Handler(proxy)
|
||||
}
|
||||
|
@ -3,16 +3,15 @@
|
||||
package blob
|
||||
|
||||
import (
|
||||
blob "github.com/goharbor/harbor/src/controller/blob"
|
||||
blobmodels "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
|
||||
context "context"
|
||||
|
||||
blob "github.com/goharbor/harbor/src/controller/blob"
|
||||
|
||||
distribution "github.com/docker/distribution"
|
||||
|
||||
mock "github.com/stretchr/testify/mock"
|
||||
|
||||
models "github.com/goharbor/harbor/src/common/models"
|
||||
models "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
)
|
||||
|
||||
// Controller is an autogenerated mock type for the Controller type
|
||||
@ -83,6 +82,20 @@ func (_m *Controller) CalculateTotalSizeByProject(ctx context.Context, projectID
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
// Delete provides a mock function with given fields: ctx, id
|
||||
func (_m *Controller) Delete(ctx context.Context, id int64) error {
|
||||
ret := _m.Called(ctx, id)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
|
||||
r0 = rf(ctx, id)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// Ensure provides a mock function with given fields: ctx, digest, contentType, size
|
||||
func (_m *Controller) Ensure(ctx context.Context, digest string, contentType string, size int64) (int64, error) {
|
||||
ret := _m.Called(ctx, digest, contentType, size)
|
||||
@ -207,11 +220,11 @@ func (_m *Controller) GetAcceptedBlobSize(sessionID string) (int64, error) {
|
||||
}
|
||||
|
||||
// List provides a mock function with given fields: ctx, params
|
||||
func (_m *Controller) List(ctx context.Context, params blobmodels.ListParams) ([]*models.Blob, error) {
|
||||
func (_m *Controller) List(ctx context.Context, params models.ListParams) ([]*models.Blob, error) {
|
||||
ret := _m.Called(ctx, params)
|
||||
|
||||
var r0 []*models.Blob
|
||||
if rf, ok := ret.Get(0).(func(context.Context, blobmodels.ListParams) []*models.Blob); ok {
|
||||
if rf, ok := ret.Get(0).(func(context.Context, models.ListParams) []*models.Blob); ok {
|
||||
r0 = rf(ctx, params)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
@ -220,7 +233,7 @@ func (_m *Controller) List(ctx context.Context, params blobmodels.ListParams) ([
|
||||
}
|
||||
|
||||
var r1 error
|
||||
if rf, ok := ret.Get(1).(func(context.Context, blobmodels.ListParams) error); ok {
|
||||
if rf, ok := ret.Get(1).(func(context.Context, models.ListParams) error); ok {
|
||||
r1 = rf(ctx, params)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
@ -256,3 +269,31 @@ func (_m *Controller) Sync(ctx context.Context, references []distribution.Descri
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// Touch provides a mock function with given fields: ctx, _a1
|
||||
func (_m *Controller) Touch(ctx context.Context, _a1 *models.Blob) error {
|
||||
ret := _m.Called(ctx, _a1)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, *models.Blob) error); ok {
|
||||
r0 = rf(ctx, _a1)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// Update provides a mock function with given fields: ctx, _a1
|
||||
func (_m *Controller) Update(ctx context.Context, _a1 *models.Blob) error {
|
||||
ret := _m.Called(ctx, _a1)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, *models.Blob) error); ok {
|
||||
r0 = rf(ctx, _a1)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
@ -5,11 +5,8 @@ package blob
|
||||
import (
|
||||
context "context"
|
||||
|
||||
blobmodels "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
|
||||
models "github.com/goharbor/harbor/src/pkg/blob/models"
|
||||
mock "github.com/stretchr/testify/mock"
|
||||
|
||||
models "github.com/goharbor/harbor/src/common/models"
|
||||
)
|
||||
|
||||
// Manager is an autogenerated mock type for the Manager type
|
||||
@ -129,6 +126,20 @@ func (_m *Manager) Create(ctx context.Context, digest string, contentType string
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
// Delete provides a mock function with given fields: ctx, id
|
||||
func (_m *Manager) Delete(ctx context.Context, id int64) error {
|
||||
ret := _m.Called(ctx, id)
|
||||
|
||||
var r0 error
|
||||
if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
|
||||
r0 = rf(ctx, id)
|
||||
} else {
|
||||
r0 = ret.Error(0)
|
||||
}
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// Get provides a mock function with given fields: ctx, digest
|
||||
func (_m *Manager) Get(ctx context.Context, digest string) (*models.Blob, error) {
|
||||
ret := _m.Called(ctx, digest)
|
||||
@ -153,11 +164,11 @@ func (_m *Manager) Get(ctx context.Context, digest string) (*models.Blob, error)
|
||||
}
|
||||
|
||||
// List provides a mock function with given fields: ctx, params
|
||||
func (_m *Manager) List(ctx context.Context, params blobmodels.ListParams) ([]*models.Blob, error) {
|
||||
func (_m *Manager) List(ctx context.Context, params models.ListParams) ([]*models.Blob, error) {
|
||||
ret := _m.Called(ctx, params)
|
||||
|
||||
var r0 []*models.Blob
|
||||
if rf, ok := ret.Get(0).(func(context.Context, blobmodels.ListParams) []*models.Blob); ok {
|
||||
if rf, ok := ret.Get(0).(func(context.Context, models.ListParams) []*models.Blob); ok {
|
||||
r0 = rf(ctx, params)
|
||||
} else {
|
||||
if ret.Get(0) != nil {
|
||||
@ -166,7 +177,7 @@ func (_m *Manager) List(ctx context.Context, params blobmodels.ListParams) ([]*m
|
||||
}
|
||||
|
||||
var r1 error
|
||||
if rf, ok := ret.Get(1).(func(context.Context, blobmodels.ListParams) error); ok {
|
||||
if rf, ok := ret.Get(1).(func(context.Context, models.ListParams) error); ok {
|
||||
r1 = rf(ctx, params)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
@ -188,3 +199,24 @@ func (_m *Manager) Update(ctx context.Context, _a1 *models.Blob) error {
|
||||
|
||||
return r0
|
||||
}
|
||||
|
||||
// UpdateBlobStatus provides a mock function with given fields: ctx, _a1
|
||||
func (_m *Manager) UpdateBlobStatus(ctx context.Context, _a1 *models.Blob) (int64, error) {
|
||||
ret := _m.Called(ctx, _a1)
|
||||
|
||||
var r0 int64
|
||||
if rf, ok := ret.Get(0).(func(context.Context, *models.Blob) int64); ok {
|
||||
r0 = rf(ctx, _a1)
|
||||
} else {
|
||||
r0 = ret.Get(0).(int64)
|
||||
}
|
||||
|
||||
var r1 error
|
||||
if rf, ok := ret.Get(1).(func(context.Context, *models.Blob) error); ok {
|
||||
r1 = rf(ctx, _a1)
|
||||
} else {
|
||||
r1 = ret.Error(1)
|
||||
}
|
||||
|
||||
return r0, r1
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user