Merge pull request #11046 from ywk253100/200312_replication

Replicate tag deletion between Harbor instances

src/api/event/handler/init.go

@@ -29,6 +29,7 @@ func init() {
 	notifier.Subscribe(event.TopicPushArtifact, &replication.Handler{})
 	notifier.Subscribe(event.TopicDeleteArtifact, &replication.Handler{})
 	notifier.Subscribe(event.TopicCreateTag, &replication.Handler{})
+	notifier.Subscribe(event.TopicDeleteTag, &replication.Handler{})
 
 	// audit logs
 	notifier.Subscribe(event.TopicPushArtifact, &auditlog.Handler{})

src/api/event/handler/replication/replication.go

@@ -42,6 +42,10 @@ func (r *Handler) Handle(value interface{}) error {
 	if ok {
 		return r.handleCreateTag(createTagEvent)
 	}
+	deleteTagEvent, ok := value.(*event.DeleteTagEvent)
+	if ok {
+		return r.handleDeleteTag(deleteTagEvent)
+	}
 	return nil
 }
 
@@ -50,8 +54,6 @@ func (r *Handler) IsStateful() bool {
 	return false
 }
 
-// TODO handle create tag
-
 func (r *Handler) handlePushArtifact(event *event.PushArtifactEvent) error {
 	art := event.Artifact
 	public := false
@@ -140,3 +142,27 @@ func (r *Handler) handleCreateTag(event *event.CreateTagEvent) error {
 	}
 	return replication.EventHandler.Handle(e)
 }
+
+func (r *Handler) handleDeleteTag(event *event.DeleteTagEvent) error {
+	art := event.AttachedArtifact
+	e := &repevent.Event{
+		Type: repevent.EventTypeTagDelete,
+		Resource: &model.Resource{
+			Type: model.ResourceTypeArtifact,
+			Metadata: &model.ResourceMetadata{
+				Repository: &model.Repository{
+					Name: event.Repository,
+				},
+				Artifacts: []*model.Artifact{
+					{
+						Type:   art.Type,
+						Digest: art.Digest,
+						Tags:   []string{event.Tag},
+					}},
+			},
+			Deleted:     true,
+			IsDeleteTag: true,
+		},
+	}
+	return replication.EventHandler.Handle(e)
+}

src/replication/adapter/adapter.go

@@ -51,17 +51,17 @@ type Adapter interface {
 	HealthCheck() (model.HealthStatus, error)
 }
 
-// ImageRegistry defines the capabilities that an image registry should have
-type ImageRegistry interface {
-	FetchImages(filters []*model.Filter) ([]*model.Resource, error)
+// ArtifactRegistry defines the capabilities that an artifact registry should have
+type ArtifactRegistry interface {
+	FetchArtifacts(filters []*model.Filter) ([]*model.Resource, error)
 	ManifestExist(repository, reference string) (exist bool, digest string, err error)
 	PullManifest(repository, reference string, accepttedMediaTypes ...string) (manifest distribution.Manifest, digest string, err error)
 	PushManifest(repository, reference, mediaType string, payload []byte) (string, error)
-	// the "reference" can be "tag" or "digest", the function needs to handle both
-	DeleteManifest(repository, reference string) error
+	DeleteManifest(repository, reference string) error // the "reference" can be "tag" or "digest", the function needs to handle both
 	BlobExist(repository, digest string) (exist bool, err error)
 	PullBlob(repository, digest string) (size int64, blob io.ReadCloser, err error)
 	PushBlob(repository, digest string, size int64, blob io.Reader) error
+	DeleteTag(repository, tag string) error
 }
 
 // ChartRegistry defines the capabilities that a chart registry should have

src/replication/adapter/aliacr/adapter.go

@@ -99,6 +99,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return getAdapterInfo()
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 // adapter for to aliyun docker registry
 type adapter struct {
 	*native.Adapter
@@ -183,7 +188,7 @@ func (a *adapter) listNamespaces(c *cr.Client) (namespaces []string, err error)
 		namespaces = append(namespaces, ns.Namespace)
 	}
 
-	log.Debugf("FetchImages.listNamespaces: %#v\n", namespaces)
+	log.Debugf("FetchArtifacts.listNamespaces: %#v\n", namespaces)
 
 	return
 }
@@ -202,9 +207,9 @@ func (a *adapter) listCandidateNamespaces(c *cr.Client, namespacePattern string)
 	return a.listNamespaces(c)
 }
 
-// FetchImages AliACR not support /v2/_catalog of Registry, we'll list all resources via Aliyun's API
-func (a *adapter) FetchImages(filters []*model.Filter) (resources []*model.Resource, err error) {
-	log.Debugf("FetchImages.filters: %#v\n", filters)
+// FetchArtifacts AliACR not support /v2/_catalog of Registry, we'll list all resources via Aliyun's API
+func (a *adapter) FetchArtifacts(filters []*model.Filter) (resources []*model.Resource, err error) {
+	log.Debugf("FetchArtifacts.filters: %#v\n", filters)
 
 	var client *cr.Client
 	client, err = cr.NewClientWithAccessKey(a.region, a.registry.Credential.AccessKey, a.registry.Credential.AccessSecret)
@@ -265,7 +270,7 @@ func (a *adapter) FetchImages(filters []*model.Filter) (resources []*model.Resou
 			}
 		}
 	}
-	log.Debugf("FetchImages.repositories: %#v\n", repositories)
+	log.Debugf("FetchArtifacts.repositories: %#v\n", repositories)
 
 	var rawResources = make([]*model.Resource, len(repositories))
 	runner := utils.NewLimitedConcurrentRunner(adp.MaxConcurrency)
@@ -316,7 +321,7 @@ func (a *adapter) FetchImages(filters []*model.Filter) (resources []*model.Resou
 	runner.Wait()
 
 	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchImages error when collect tags for repos")
+		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
 	}
 
 	for _, r := range rawResources {

src/replication/adapter/aliacr/adapter_test.go

@@ -118,11 +118,11 @@ func BenchmarkGetRegion(b *testing.B) {
 	}
 }
 
-func Test_adapter_FetchImages(t *testing.T) {
+func Test_adapter_FetchArtifacts(t *testing.T) {
 	a, s := getMockAdapter(t, true, true)
 	defer s.Close()
 	var filters = []*model.Filter{}
-	var resources, err = a.FetchImages(filters)
+	var resources, err = a.FetchArtifacts(filters)
 	assert.NotNil(t, err)
 	assert.Nil(t, resources)
 }

src/replication/adapter/awsecr/adapter.go

@@ -81,6 +81,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return getAdapterInfo()
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 type adapter struct {
 	*native.Adapter
 	registry      *model.Registry

src/replication/adapter/awsecr/adapter_test.go

@@ -196,7 +196,7 @@ func TestAdapter_PrepareForPush(t *testing.T) {
 func TestAdapter_FetchImages(t *testing.T) {
 	a, s := getMockAdapter(t, true, true)
 	defer s.Close()
-	resources, err := a.FetchImages([]*model.Filter{
+	resources, err := a.FetchArtifacts([]*model.Filter{
 		{
 			Type:  model.FilterTypeName,
 			Value: "*",

src/replication/adapter/azurecr/adapter.go

@@ -38,8 +38,10 @@ type adapter struct {
 	*native.Adapter
 }
 
-// Ensure '*adapter' implements interface 'Adapter'.
-var _ adp.Adapter = (*adapter)(nil)
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
 
 // Info returns information of the registry
 func (a *adapter) Info() (*model.RegistryInfo, error) {

src/replication/adapter/dockerhub/adapter.go

@@ -55,6 +55,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return getAdapterInfo()
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 type adapter struct {
 	*native.Adapter
 	registry *model.Registry
@@ -232,8 +237,8 @@ func (a *adapter) getNamespace(namespace string) (*model.Namespace, error) {
 	}, nil
 }
 
-// FetchImages fetches images
-func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error) {
+// FetchArtifacts fetches images
+func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, error) {
 	var repos []Repo
 	nameFilter, err := a.getStringFilterValue(model.FilterTypeName, filters)
 	if err != nil {
@@ -339,7 +344,7 @@ func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error
 	runner.Wait()
 
 	if runner.IsCancelled() {
-		return nil, fmt.Errorf("FetchImages error when collect tags for repos")
+		return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
 	}
 
 	var resources []*model.Resource

src/replication/adapter/dockerhub/adapter_test.go

@@ -66,10 +66,10 @@ func TestListNamespaces(t *testing.T) {
 	}
 }
 
-func TestFetchImages(t *testing.T) {
+func TestFetchArtifacts(t *testing.T) {
 	ad := getAdapter(t)
 	adapter := ad.(*adapter)
-	_, err := adapter.FetchImages([]*model.Filter{
+	_, err := adapter.FetchArtifacts([]*model.Filter{
 		{
 			Type:  model.FilterTypeName,
 			Value: "goharbor/harbor-core",

src/replication/adapter/gitlab/adapter.go

@@ -30,6 +30,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return nil
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 type adapter struct {
 	*native.Adapter
 	registry        *model.Registry
@@ -71,8 +76,8 @@ func (a *adapter) Info() (info *model.RegistryInfo, err error) {
 	}, nil
 }
 
-// FetchImages fetches images
-func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error) {
+// FetchArtifacts fetches images
+func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, error) {
 	var resources []*model.Resource
 	var projects []*Project
 	var err error

src/replication/adapter/googlegcr/adapter.go

@@ -49,6 +49,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return getAdapterInfo()
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 type adapter struct {
 	*native.Adapter
 	registry *model.Registry

src/replication/adapter/googlegcr/adapter_test.go

@@ -142,7 +142,7 @@ func TestAdapter_PrepareForPush(t *testing.T) {
 func TestAdapter_FetchImages(t *testing.T) {
 	a, s := getMockAdapter(t, true, true)
 	defer s.Close()
-	resources, err := a.FetchImages([]*model.Filter{
+	resources, err := a.FetchArtifacts([]*model.Filter{
 		{
 			Type:  model.FilterTypeName,
 			Value: "*",

src/replication/adapter/harbor/adapter.go

@@ -54,6 +54,12 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return nil
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+	_ adp.ChartRegistry    = (*adapter)(nil)
+)
+
 type adapter struct {
 	*native.Adapter
 	registry *model.Registry

src/replication/adapter/harbor/image_registry.go

@@ -28,7 +28,7 @@ import (
 	"strings"
 )
 
-func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error) {
+func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, error) {
 	projects, err := a.listProjects(filters)
 	if err != nil {
 		return nil, err
@@ -78,7 +78,7 @@ func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error
 		runner.Wait()
 
 		if runner.IsCancelled() {
-			return nil, fmt.Errorf("FetchImages error when collect tags for repos")
+			return nil, fmt.Errorf("FetchArtifacts error when collect tags for repos")
 		}
 
 		for _, r := range rawResources {
@@ -168,3 +168,10 @@ func (a *adapter) listArtifacts(repository string, filters []*model.Filter) ([]*
 	}
 	return filter.DoFilterArtifacts(arts, filters)
 }
+
+func (a *adapter) DeleteTag(repository, tag string) error {
+	project, repository := utils.ParseRepository(repository)
+	url := fmt.Sprintf("%s/api/%s/projects/%s/repositories/%s/artifacts/%s/tags/%s",
+		a.getURL(), api.APIVersion, project, repository, tag, tag)
+	return a.client.Delete(url)
+}

src/replication/adapter/harbor/image_registry_test.go

@@ -24,7 +24,7 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
-func TestFetchImages(t *testing.T) {
+func TestFetchArtifacts(t *testing.T) {
 	server := test.NewServer([]*test.RequestHandlerMapping{
 		{
 			Method:  http.MethodGet,
@@ -80,7 +80,7 @@ func TestFetchImages(t *testing.T) {
 	adapter, err := newAdapter(registry)
 	require.Nil(t, err)
 	// nil filter
-	resources, err := adapter.FetchImages(nil)
+	resources, err := adapter.FetchArtifacts(nil)
 	require.Nil(t, err)
 	assert.Equal(t, 1, len(resources))
 	assert.Equal(t, model.ResourceTypeArtifact, resources[0].Type)
@@ -99,7 +99,7 @@ func TestFetchImages(t *testing.T) {
 			Value: "1.0",
 		},
 	}
-	resources, err = adapter.FetchImages(filters)
+	resources, err = adapter.FetchArtifacts(filters)
 	require.Nil(t, err)
 	assert.Equal(t, 1, len(resources))
 	assert.Equal(t, model.ResourceTypeArtifact, resources[0].Type)

src/replication/adapter/helmhub/adapter.go

@@ -53,6 +53,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	}
 }
 
+var (
+	_ adp.Adapter       = (*adapter)(nil)
+	_ adp.ChartRegistry = (*adapter)(nil)
+)
+
 type adapter struct {
 	registry *model.Registry
 	client   *Client

src/replication/adapter/huawei/huawei_adapter.go

@@ -40,6 +40,11 @@ func (f *factory) AdapterPattern() *model.AdapterPattern {
 	return nil
 }
 
+var (
+	_ adp.Adapter          = (*adapter)(nil)
+	_ adp.ArtifactRegistry = (*adapter)(nil)
+)
+
 // Adapter is for images replications between harbor and Huawei image repository(SWR)
 type adapter struct {
 	*native.Adapter

src/replication/adapter/huawei/image_registry.go

@@ -10,8 +10,8 @@ import (
 	"github.com/goharbor/harbor/src/replication/model"
 )
 
-// FetchImages gets resources from Huawei SWR
-func (a *adapter) FetchImages(filters []*model.Filter) ([]*model.Resource, error) {
+// FetchArtifacts gets resources from Huawei SWR
+func (a *adapter) FetchArtifacts(filters []*model.Filter) ([]*model.Resource, error) {
 
 	resources := []*model.Resource{}
 

src/replication/adapter/huawei/image_registry_test.go

@@ -28,8 +28,8 @@ func init() {
 	HWAdapter = *adp.(*adapter)
 }
 
-func TestAdapter_FetchImages(t *testing.T) {
-	resources, err := HWAdapter.FetchImages(nil)
+func TestAdapter_FetchArtifacts(t *testing.T) {
+	resources, err := HWAdapter.FetchArtifacts(nil)
 	if err != nil {
 		if strings.HasPrefix(err.Error(), "[401]") {
 			t.Log("huawei ak/sk is not available", err.Error())