Add valid method to adapter to valid resource

Signed-off-by: cd1989 <chende@caicloud.io>

src/replication/ng/adapter/adapter.go

@@ -39,6 +39,9 @@ type Adapter interface {
 	// PrepareForPush does the prepare work that needed for pushing/uploading the resource
 	// eg: create the namespace or repository
 	PrepareForPush(*model.Resource) error
+	// ValidResource checks whether a resource is valid for the registry. For example, DockerHub don't support
+	// multiple parts repo name like 'a/b/c'.
+	ValidResource(*model.Resource) bool
 	// Get the namespace specified by the name, the returning value should
 	// contain the metadata about the namespace if it has
 	// TODO remove this method?

src/replication/ng/adapter/dockerhub/adapter.go

@@ -51,7 +51,8 @@ var _ adp.Adapter = (*adapter)(nil)
 // Info returns information of the registry
 func (a *adapter) Info() (*model.RegistryInfo, error) {
 	return &model.RegistryInfo{
-		Type: model.RegistryTypeDockerHub,
+		Type:             model.RegistryTypeDockerHub,
+		SupportNamespace: true,
 		SupportedResourceTypes: []model.ResourceType{
 			model.ResourceTypeRepository,
 		},
@@ -67,13 +68,54 @@ func (a *adapter) Info() (*model.RegistryInfo, error) {
 		},
 		SupportedTriggers: []model.TriggerType{
 			model.TriggerTypeManual,
+			model.TriggerTypeScheduled,
 		},
 	}, nil
 }
 
-// HealthCheck checks health status of the registry
-func (a *adapter) HealthCheck() (model.HealthStatus, error) {
-	return model.Healthy, nil
+// ValidResource checks whether a resource is valid, in DockerHub, multi-parts repo name like 'a/b/c' is not supported.
+func (a *adapter) ValidResource(resource *model.Resource) bool {
+	if resource == nil || resource.Metadata == nil || resource.Metadata.Repository == nil {
+		return false
+	}
+
+	if len(strings.Split(resource.Metadata.Repository.Name, "/")) != 1 {
+		return false
+	}
+
+	return true
+}
+
+// ConvertResourceMetadata converts the namespace and repository part of the resource metadata
+// to the one that the adapter can handle
+func (a *adapter) ConvertResourceMetadata(meta *model.ResourceMetadata, namespace *model.Namespace) (*model.ResourceMetadata, error) {
+	return meta, nil
+}
+
+// PrepareForPush does the prepare work that needed for pushing/uploading the resource
+// eg: create the namespace or repository
+func (a *adapter) PrepareForPush(resource *model.Resource) error {
+	if resource == nil {
+		return errors.New("the resource cannot be null")
+	}
+	if resource.Metadata == nil {
+		return errors.New("the metadata of resource cannot be null")
+	}
+	if resource.Metadata.Namespace == nil {
+		return errors.New("the namespace of resource cannot be null")
+	}
+	if len(resource.Metadata.Namespace.Name) == 0 {
+		return errors.New("the name of the namespace cannot be null")
+	}
+
+	err := a.CreateNamespace(&model.Namespace{
+		Name: resource.Metadata.Namespace.Name,
+	})
+	if err != nil {
+		return fmt.Errorf("create namespace '%s' in DockerHub error: %v", resource.Metadata.Namespace.Name, err)
+	}
+
+	return nil
 }
 
 // ListNamespaces lists namespaces from DockerHub with the provided query conditions.
@@ -273,9 +315,13 @@ func (a *adapter) FetchImages(namespaces []string, filters []*model.Filter) ([]*
 			Type:     model.ResourceTypeRepository,
 			Registry: a.registry,
 			Metadata: &model.ResourceMetadata{
-				Namespace: repo.Namespace,
-				Name:      fmt.Sprintf("%s/%s", repo.Namespace, repo.Name),
-				Vtags:     tags,
+				Namespace: &model.Namespace{
+					Name: repo.Namespace,
+				},
+				Repository: &model.Repository{
+					Name: repo.Name,
+				},
+				Vtags: tags,
 			},
 		})
 	}

src/replication/ng/adapter/dockerhub/adapter_test.go

@@ -48,21 +48,3 @@ func TestListNamespaces(t *testing.T) {
 		fmt.Println(ns.Name)
 	}
 }
-
-func TestCreateNamespace(t *testing.T) {
-	if testUser == "" {
-		return
-	}
-
-	assert := assert.New(t)
-	adapter := getAdapter(t)
-
-	err := adapter.CreateNamespace(&model.Namespace{
-		Name: "harborns",
-		Metadata: map[string]interface{}{
-			metadataKeyFullName: "harbor namespace",
-			metadataKeyCompany:  "harbor",
-		},
-	})
-	assert.Nil(err)
-}

src/replication/ng/adapter/huawei/huawei_adapter.go

@@ -118,6 +118,11 @@ func (adapter Adapter) ConvertResourceMetadata(resourceMetadata *model.ResourceM
 	return metadata, nil
 }
 
+// ValidResource ...
+func (adapter Adapter) ValidResource(*model.Resource) bool {
+	return true
+}
+
 // PrepareForPush prepare for push to Huawei SWR
 func (adapter Adapter) PrepareForPush(resource *model.Resource) error {
 

src/replication/ng/adapter/image_registry.go

@@ -276,6 +276,11 @@ func (d *DefaultImageRegistry) PushBlob(repository, digest string, size int64, b
 	return client.PushBlob(digest, size, blob)
 }
 
+// ValidResource ...
+func (d *DefaultImageRegistry) ValidResource(*model.Resource) bool {
+	return true
+}
+
 func isDigest(str string) bool {
 	return strings.Contains(str, ":")
 }

src/replication/ng/model/resource.go

@@ -69,4 +69,6 @@ type Resource struct {
 	Deleted bool `json:"deleted"`
 	// indicate whether the resource can be overridden
 	Override bool `json:"override"`
+	// Invalid indicates the resource is invalid for the registry
+	Invalid bool `json:"invalid"`
 }

src/replication/ng/operation/controller_test.go

@@ -149,6 +149,9 @@ func (f *fakedAdapter) PrepareForPush(*model.Resource) error {
 func (f *fakedAdapter) HealthCheck() (model.HealthStatus, error) {
 	return model.Healthy, nil
 }
+func (f *fakedAdapter) ValidResource(*model.Resource) bool {
+	return true
+}
 func (f *fakedAdapter) GetNamespace(ns string) (*model.Namespace, error) {
 	var namespace *model.Namespace
 	if ns == "library" {

src/replication/ng/operation/flow/copy.go

@@ -61,6 +61,13 @@ func (c *copyFlow) Run(interface{}) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+
+	for i := range dstResources {
+		if !dstAdapter.ValidResource(dstResources[i]) {
+			dstResources[i].Invalid = true
+		}
+	}
+
 	if err = prepareForPush(dstAdapter, dstResources); err != nil {
 		return 0, err
 	}
@@ -71,7 +78,14 @@ func (c *copyFlow) Run(interface{}) (int, error) {
 	if err = createTasks(c.executionMgr, c.executionID, items); err != nil {
 		return 0, err
 	}
-	return schedule(c.scheduler, c.executionMgr, items)
+
+	var filtered []*scheduler.ScheduleItem
+	for _, item := range items {
+		if !item.DstResource.Invalid {
+			filtered = append(filtered, item)
+		}
+	}
+	return schedule(c.scheduler, c.executionMgr, filtered)
 }
 
 // mark the execution as success in database

src/replication/ng/operation/flow/deletion.go

@@ -64,6 +64,13 @@ func (d *deletionFlow) Run(interface{}) (int, error) {
 	if err != nil {
 		return 0, err
 	}
+
+	for i := range dstResources {
+		if !dstAdapter.ValidResource(dstResources[i]) {
+			dstResources[i].Invalid = true
+		}
+	}
+
 	items, err := preprocess(d.scheduler, srcResources, dstResources)
 	if err != nil {
 		return 0, err
@@ -71,5 +78,13 @@ func (d *deletionFlow) Run(interface{}) (int, error) {
 	if err = createTasks(d.executionMgr, d.executionID, items); err != nil {
 		return 0, err
 	}
-	return schedule(d.scheduler, d.executionMgr, items)
+
+	var filtered []*scheduler.ScheduleItem
+	for _, item := range items {
+		if !item.DstResource.Invalid {
+			filtered = append(filtered, item)
+		}
+	}
+
+	return schedule(d.scheduler, d.executionMgr, filtered)
 }

src/replication/ng/operation/flow/stage.go

@@ -241,6 +241,7 @@ func createTasks(mgr execution.Manager, executionID int64, items []*scheduler.Sc
 		if item.DstResource.Deleted {
 			operation = "deletion"
 		}
+
 		task := &models.Task{
 			ExecutionID:  executionID,
 			Status:       models.TaskStatusInitialized,
@@ -249,6 +250,12 @@ func createTasks(mgr execution.Manager, executionID int64, items []*scheduler.Sc
 			DstResource:  getResourceName(item.DstResource),
 			Operation:    operation,
 		}
+
+		if item.DstResource.Invalid {
+			task.Status = models.TaskStatusFailed
+			task.EndTime = time.Now()
+		}
+
 		id, err := mgr.CreateTask(task)
 		if err != nil {
 			// if failed to create the task for one of the items,

src/replication/ng/operation/flow/stage_test.go

@@ -62,6 +62,9 @@ func (f *fakedAdapter) PrepareForPush(*model.Resource) error {
 func (f *fakedAdapter) HealthCheck() (model.HealthStatus, error) {
 	return model.Healthy, nil
 }
+func (f *fakedAdapter) ValidResource(*model.Resource) bool {
+	return true
+}
 func (f *fakedAdapter) GetNamespace(ns string) (*model.Namespace, error) {
 	var namespace *model.Namespace
 	if ns == "library" {

src/testing/job/mock_client.go

@@ -17,31 +17,31 @@ type MockJobClient struct {
 // GetJobLog ...
 func (mjc *MockJobClient) GetJobLog(uuid string) ([]byte, error) {
 	if uuid == "500" {
-		return nil, &http.Error{500, "Server side error"}
+		return nil, &http.Error{500, "server side error"}
 	}
 	if mjc.validUUID(uuid) {
 		return []byte("some log"), nil
 	}
-	return nil, &http.Error{404, "Not Found"}
+	return nil, &http.Error{404, "not Found"}
 }
 
 // SubmitJob ...
 func (mjc *MockJobClient) SubmitJob(data *models.JobData) (string, error) {
-	if data.Name == job.ImageScanAllJob || data.Name == job.ImageReplicate || data.Name == job.ImageGC || data.Name == job.ImageScanJob {
+	if data.Name == job.ImageScanAllJob || data.Name == job.Replication || data.Name == job.ImageGC || data.Name == job.ImageScanJob {
 		uuid := fmt.Sprintf("u-%d", rand.Int())
 		mjc.JobUUID = append(mjc.JobUUID, uuid)
 		return uuid, nil
 	}
-	return "", fmt.Errorf("Unsupported job %s", data.Name)
+	return "", fmt.Errorf("unsupported job %s", data.Name)
 }
 
 // PostAction ...
 func (mjc *MockJobClient) PostAction(uuid, action string) error {
 	if "500" == uuid {
-		return &http.Error{500, "Server side error"}
+		return &http.Error{500, "server side error"}
 	}
 	if !mjc.validUUID(uuid) {
-		return &http.Error{404, "Not Found"}
+		return &http.Error{404, "not Found"}
 	}
 	return nil
 }