Support pull based replication from Dockerhub

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

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

@@ -3,6 +3,7 @@ package dockerhub
 import (
 	"bytes"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io/ioutil"
 	"net/http"
@@ -53,6 +54,12 @@ func (a *adapter) Info() (*model.RegistryInfo, error) {
 		SupportedResourceTypes: []model.ResourceType{
 			model.ResourceTypeRepository,
 		},
+		SupportedResourceFilters: []*model.FilterStyle{
+			{
+				Type:  model.FilterTypeName,
+				Style: model.FilterStyleTypeText,
+			},
+		},
 		SupportedTriggers: []model.TriggerType{
 			model.TriggerTypeManual,
 		},
@@ -177,3 +184,140 @@ func (a *adapter) getNamespace(namespace string) (*model.Namespace, error) {
 		Name: namespace,
 	}, nil
 }
+
+// FetchImages fetches images
+func (a *adapter) FetchImages(namespaces []string, filters []*model.Filter) ([]*model.Resource, error) {
+	var repos []Repo
+	nameFilter := a.getFilter(model.FilterTypeName, filters)
+	for _, ns := range namespaces {
+		name := ""
+		if nameFilter != nil {
+			v, ok := nameFilter.Value.(string)
+			if !ok {
+				msg := fmt.Sprintf("expect name filter value to be string, but got: %v", nameFilter.Value)
+				log.Error(msg)
+				return nil, errors.New(msg)
+			}
+			name = v
+		}
+
+		page := 1
+		pageSize := 100
+		for {
+			pageRepos, err := a.getRepos(ns, name, page, pageSize)
+			if err != nil {
+				return nil, fmt.Errorf("get repos for namespace '%s' from DockerHub error: %v", ns, err)
+			}
+			repos = append(repos, pageRepos.Repos...)
+
+			if len(pageRepos.Next) == 0 {
+				break
+			}
+			page++
+		}
+	}
+
+	log.Infof("%d repos found for namespaces: %v", len(repos), namespaces)
+	var resources []*model.Resource
+	// TODO(ChenDe): Get tags for repos in parallel
+	for _, repo := range repos {
+		var tags []string
+		page := 1
+		pageSize := 100
+		for {
+			pageTags, err := a.getTags(repo.Namespace, repo.Name, page, pageSize)
+			if err != nil {
+				return nil, fmt.Errorf("get tags for repo '%s/%s' from DockerHub error: %v", repo.Namespace, repo.Name, err)
+			}
+			for _, t := range pageTags.Tags {
+				tags = append(tags, t.Name)
+			}
+
+			if len(pageTags.Next) == 0 {
+				break
+			}
+			page++
+		}
+
+		// If the repo has no tags, skip it
+		if len(tags) == 0 {
+			continue
+		}
+
+		resources = append(resources, &model.Resource{
+			Type:     model.ResourceTypeRepository,
+			Registry: a.registry,
+			Metadata: &model.ResourceMetadata{
+				Namespace: repo.Namespace,
+				Name:      fmt.Sprintf("%s/%s", repo.Namespace, repo.Name),
+				Vtags:     tags,
+			},
+		})
+	}
+
+	return resources, nil
+}
+
+// getRepos gets a page of repos from DockerHub
+func (a *adapter) getRepos(namespace, name string, page, pageSize int) (*ReposResp, error) {
+	resp, err := a.client.Do(http.MethodGet, listReposPath(namespace, name, page, pageSize), nil)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode/100 != 2 {
+		log.Errorf("list repos error: %d -- %s", resp.StatusCode, string(body))
+		return nil, fmt.Errorf("%d -- %s", resp.StatusCode, body)
+	}
+
+	repos := &ReposResp{}
+	err = json.Unmarshal(body, repos)
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal repos list %s error: %v", string(body), err)
+	}
+
+	return repos, nil
+}
+
+// getTags gets a page of tags for a repo from DockerHub
+func (a *adapter) getTags(namespace, repo string, page, pageSize int) (*TagsResp, error) {
+	resp, err := a.client.Do(http.MethodGet, listTagsPath(namespace, repo, page, pageSize), nil)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	body, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	if resp.StatusCode/100 != 2 {
+		log.Errorf("list tags error: %d -- %s", resp.StatusCode, string(body))
+		return nil, fmt.Errorf("%d -- %s", resp.StatusCode, body)
+	}
+
+	tags := &TagsResp{}
+	err = json.Unmarshal(body, tags)
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal tags list %s error: %v", string(body), err)
+	}
+
+	return tags, nil
+}
+
+// getFilter gets specific type filter from filters list.
+func (a *adapter) getFilter(filterType model.FilterType, filters []*model.Filter) *model.Filter {
+	for _, f := range filters {
+		if f.Type == filterType {
+			return f
+		}
+	}
+	return nil
+}

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

@@ -17,12 +17,12 @@ const (
 
 func getAdapter(t *testing.T) adp.Adapter {
 	assert := assert.New(t)
-	factory, err := adp.GetFactory(registryTypeDockerHub)
+	factory, err := adp.GetFactory(model.RegistryTypeDockerHub)
 	assert.Nil(err)
 	assert.NotNil(factory)
 
 	adapter, err := factory(&model.Registry{
-		Type: registryTypeDockerHub,
+		Type: model.RegistryTypeDockerHub,
 		Credential: &model.Credential{
 			AccessKey:    testUser,
 			AccessSecret: testPassword,

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

@@ -16,3 +16,15 @@ const (
 func getNamespacePath(namespace string) string {
 	return fmt.Sprintf("/v2/orgs/%s/", namespace)
 }
+
+func listReposPath(namespace, name string, page, pageSize int) string {
+	if len(name) == 0 {
+		return fmt.Sprintf("/v2/repositories/%s/?page=%d&page_size=%d", namespace, page, pageSize)
+	}
+
+	return fmt.Sprintf("/v2/repositories/%s/?name=%s&page=%d&page_size=%d", namespace, name, page, pageSize)
+}
+
+func listTagsPath(namespace, repo string, page, pageSize int) string {
+	return fmt.Sprintf("/v2/repositories/%s/%s/tags/?page=%d&page_size=%d", namespace, repo, page, pageSize)
+}

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

@@ -32,3 +32,61 @@ type NewOrgReq struct {
 	// GravatarEmail ...
 	GravatarEmail string `json:"gravatar_email"`
 }
+
+// Repo describes a repo in DockerHub
+type Repo struct {
+	// User ...
+	User string `json:"user"`
+	// Name of the repo
+	Name string `json:"name"`
+	// Namespace of the repo
+	Namespace string `json:"namespace"`
+	// RepoType is type of the repo, e.g. 'image'
+	RepoType string `json:"repository_type"`
+	// Status ...
+	Status int `json:"status"`
+	// Description ...
+	Description string `json:"description"`
+	// IsPrivate indicates whether the repo is private
+	IsPrivate bool `json:"is_private"`
+	// IsAutomated ...
+	IsAutomated bool `json:"is_automated"`
+	// CanEdit ...
+	CanEdit bool `json:"can_edit"`
+	// StarCount ..
+	StarCount int `json:"star_count"`
+	// PullCount ...
+	PullCount int `json:"pull_count"`
+}
+
+// ReposResp is response of repo list request
+type ReposResp struct {
+	// Count is total number of repos
+	Count int `json:"count"`
+	// Next is the URL of the next page
+	Next string `json:"next"`
+	// Previous is the URL of the previous page
+	Previous string `json:"previous"`
+	// Repos is repo list
+	Repos []Repo `json:"results"`
+}
+
+// Tag describes a tag in DockerHub
+type Tag struct {
+	// Name of the tag
+	Name string `json:"name"`
+	// FullSize is size of the image
+	FullSize int64 `json:"full_size"`
+}
+
+// TagsResp is response of tag list request
+type TagsResp struct {
+	// Count is total number of repos
+	Count int `json:"count"`
+	// Next is the URL of the next page
+	Next string `json:"next"`
+	// Previous is the URL of the previous page
+	Previous string `json:"previous"`
+	// Repos is tags list
+	Tags []Tag `json:"results"`
+}

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

@@ -317,5 +317,10 @@ func getResourceName(res *model.Resource) string {
 	if len(meta.Vtags) == 0 {
 		return meta.GetResourceName()
 	}
-	return meta.GetResourceName() + ":[" + strings.Join(meta.Vtags, ",") + "]"
+
+	if len(meta.Vtags) <= 5 {
+		return meta.GetResourceName() + ":[" + strings.Join(meta.Vtags, ",") + "]"
+	}
+
+	return fmt.Sprintf("%s:[%s ... %d in total]", meta.GetResourceName(), strings.Join(meta.Vtags[:5], ","), len(meta.Vtags))
 }