Add repo into DB, update code per comments

Deploy/db/registry.sql

@@ -105,6 +105,22 @@ create table access_log (
  FOREIGN KEY (project_id) REFERENCES project (project_id)
 );
 
+create table repository (
+ repository_id int NOT NULL AUTO_INCREMENT,
+ name varchar(255) NOT NULL,
+ project_id int NOT NULL,
+ owner_id int NOT NULL,
+ description text,
+ pull_count int DEFAULT 0 NOT NULL,
+ star_count int DEFAULT 0 NOT NULL,
+ creation_time timestamp default CURRENT_TIMESTAMP,
+ update_time timestamp default CURRENT_TIMESTAMP on update CURRENT_TIMESTAMP,
+ primary key (repository_id),
+ FOREIGN KEY (owner_id) REFERENCES user(user_id),
+ FOREIGN KEY (project_id) REFERENCES project(project_id),
+ UNIQUE (name)
+);
+
 create table replication_policy (
  id int NOT NULL AUTO_INCREMENT,
  name varchar(256),

api/internal.go

@@ -0,0 +1,51 @@
+/*
+   Copyright (c) 2016 VMware, Inc. All Rights Reserved.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package api
+
+import (
+	"net/http"
+
+	"github.com/vmware/harbor/dao"
+	"github.com/vmware/harbor/utils/log"
+)
+
+// InternalAPI handles request of harbor admin...
+type InternalAPI struct {
+	BaseAPI
+}
+
+// Prepare validates the URL and parms
+func (ia *InternalAPI) Prepare() {
+	var currentUserID int
+	currentUserID = ia.ValidateUser()
+	isAdmin, err := dao.IsAdminRole(currentUserID)
+	if err != nil {
+		log.Errorf("Error occurred in IsAdminRole:%v", err)
+		ia.CustomAbort(http.StatusInternalServerError, "Internal error.")
+	}
+	if !isAdmin {
+		log.Error("Guests doesn't have the permisson to request harbor internal API.")
+		ia.CustomAbort(http.StatusForbidden, "Guests doesn't have the permisson to request harbor internal API.")
+	}
+}
+
+// SyncRegistry ...
+func (ia *InternalAPI) SyncRegistry() {
+	err := SyncRegistry()
+	if err != nil {
+		ia.CustomAbort(http.StatusInternalServerError, "internal error")
+	}
+}

api/utils.go

@@ -20,15 +20,19 @@ import (
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"net"
 	"net/http"
 	"os"
+	"sort"
 	"strings"
+	"time"
 
 	"github.com/vmware/harbor/dao"
 	"github.com/vmware/harbor/models"
 	"github.com/vmware/harbor/service/cache"
 	"github.com/vmware/harbor/utils"
 	"github.com/vmware/harbor/utils/log"
+	"github.com/vmware/harbor/utils/registry"
 )
 
 func checkProjectPermission(userID int, projectID int64) bool {
@@ -235,6 +239,146 @@ func addAuthentication(req *http.Request) {
 	}
 }
 
+// SyncRegistry syncs the repositories of registry with database.
+func SyncRegistry() error {
+
+	log.Debugf("Start syncing repositories from registry to DB... ")
+	rc, err := initRegistryClient()
+	if err != nil {
+		log.Errorf("error occurred while initializing registry client for %v", err)
+		return err
+	}
+
+	reposInRegistry, err := rc.Catalog()
+	if err != nil {
+		log.Error(err)
+		return err
+	}
+
+	var repoRecordsInDB []models.RepoRecord
+	repoRecordsInDB, err = dao.GetAllRepositories()
+	if err != nil {
+		log.Errorf("error occurred while getting all registories. %v", err)
+		return err
+	}
+	var reposInDB []string
+	for _, repoRecordInDB := range repoRecordsInDB {
+		reposInDB = append(reposInDB, repoRecordInDB.Name)
+	}
+
+	var reposToAdd []string
+	var reposToDel []string
+	reposToAdd, reposToDel = diffRepos(reposInRegistry, reposInDB)
+
+	if len(reposToAdd) > 0 {
+		log.Debugf("Start adding repositories into DB... ")
+		for _, repoToAdd := range reposToAdd {
+			project, _ := utils.ParseRepository(repoToAdd)
+			user, err := dao.GetAccessLogCreator(repoToAdd)
+			if err != nil {
+				log.Errorf("Error happens when getting the repository owner from access log: %v", err)
+			}
+			if len(user) == 0 {
+				user = "anonymous"
+			}
+			pullCount, err := dao.CountPull(repoToAdd)
+			if err != nil {
+				log.Errorf("Error happens when counting pull count from access log: %v", err)
+			}
+			repoRecord := models.RepoRecord{Name: repoToAdd, OwnerName: user, ProjectName: project, PullCount: pullCount}
+			if err := dao.AddRepository(repoRecord); err != nil {
+				log.Errorf("Error happens when adding the missing repository: %v", err)
+			}
+			log.Debugf("Add repository: %s success.", repoToAdd)
+		}
+	}
+
+	if len(reposToDel) > 0 {
+		log.Debugf("Start deleting repositories from DB... ")
+		for _, repoToDel := range reposToDel {
+			if err := dao.DeleteRepository(repoToDel); err != nil {
+				log.Errorf("Error happens when deleting the repository: %v", err)
+			}
+			log.Debugf("Delete repository: %s success.", repoToDel)
+		}
+	}
+
+	log.Debugf("Sync repositories from registry to DB is done.")
+	return nil
+}
+
+func diffRepos(reposInRegistry []string, reposInDB []string) ([]string, []string) {
+	var needsAdd []string
+	var needsDel []string
+
+	sort.Strings(reposInRegistry)
+	sort.Strings(reposInDB)
+
+	i, j := 0, 0
+	for i < len(reposInRegistry) && j < len(reposInDB) {
+		d := strings.Compare(reposInRegistry[i], reposInDB[j])
+		if d < 0 {
+			needsAdd = append(needsAdd, reposInRegistry[i])
+			i++
+		} else if d > 0 {
+			needsDel = append(needsDel, reposInDB[j])
+			j++
+		} else {
+			i++
+			j++
+		}
+	}
+
+	for i < len(reposInRegistry) {
+		needsAdd = append(needsAdd, reposInRegistry[i])
+		i++
+	}
+
+	for j < len(reposInDB) {
+		needsDel = append(needsDel, reposInDB[j])
+		j++
+	}
+
+	return needsAdd, needsDel
+}
+
+func initRegistryClient() (r *registry.Registry, err error) {
+	endpoint := os.Getenv("REGISTRY_URL")
+
+	addr := ""
+	if strings.Contains(endpoint, "/") {
+		addr = endpoint[strings.LastIndex(endpoint, "/")+1:]
+	}
+
+	ch := make(chan int, 1)
+	go func() {
+		var err error
+		var c net.Conn
+		for {
+			c, err = net.DialTimeout("tcp", addr, 20*time.Second)
+			if err == nil {
+				c.Close()
+				ch <- 1
+			} else {
+				log.Errorf("failed to connect to registry client, retry after 2 seconds :%v", err)
+				time.Sleep(2 * time.Second)
+			}
+		}
+	}()
+	select {
+	case <-ch:
+	case <-time.After(60 * time.Second):
+		panic("Failed to connect to registry client after 60 seconds")
+	}
+
+	registryClient, err := cache.NewRegistryClient(endpoint, true, "admin",
+		"registry", "catalog", "*")
+	if err != nil {
+		return nil, err
+	}
+	return registryClient, nil
+}
+
 func buildReplicationURL() string {
 	url := getJobServiceURL()
 	return fmt.Sprintf("%s/api/jobs/replication", url)

dao/accesslog.go

@@ -202,3 +202,32 @@ func GetTopRepos(countNum int) ([]models.TopRepo, error) {
 	}
 	return list, nil
 }
+
+// GetAccessLogCreator ...
+func GetAccessLogCreator(repoName string) (string, error) {
+	o := GetOrmer()
+	sql := "select * from user where user_id = (select user_id from access_log where operation = 'push' and repo_name = ? order by op_time desc limit 1)"
+
+	var u []models.User
+	n, err := o.Raw(sql, repoName).QueryRows(&u)
+
+	if err != nil {
+		return "", err
+	}
+	if n == 0 {
+		return "", nil
+	}
+
+	return u[0].Username, nil
+}
+
+// CountPull ...
+func CountPull(repoName string) (int64, error) {
+	o := GetOrmer()
+	num, err := o.QueryTable("access_log").Filter("repo_name", repoName).Filter("operation", "pull").Count()
+	if err != nil {
+		log.Errorf("error in CountPull: %v ", err)
+		return 0, err
+	}
+	return num, nil
+}

dao/repository.go

@@ -0,0 +1,86 @@
+/*
+   Copyright (c) 2016 VMware, Inc. All Rights Reserved.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package dao
+
+import (
+	"fmt"
+
+	"github.com/astaxie/beego/orm"
+	"github.com/vmware/harbor/models"
+)
+
+// AddRepository adds a repo to the database.
+func AddRepository(repo models.RepoRecord) error {
+	o := GetOrmer()
+	sql := "insert into repository (owner_id, project_id, name, description, pull_count, star_count, creation_time, update_time) " +
+		"select (select user_id as owner_id from user where username=?), " +
+		"(select project_id as project_id from project where name=?), ?, ?, ?, ?, NOW(), NULL "
+
+	_, err := o.Raw(sql, repo.OwnerName, repo.ProjectName, repo.Name, repo.Description, repo.PullCount, repo.StarCount).Exec()
+	return err
+}
+
+// GetRepositoryByName ...
+func GetRepositoryByName(name string) (*models.RepoRecord, error) {
+	o := GetOrmer()
+	r := models.RepoRecord{Name: name}
+	err := o.Read(&r)
+	if err == orm.ErrNoRows {
+		return nil, nil
+	}
+	return &r, err
+}
+
+// GetAllRepositories ...
+func GetAllRepositories() ([]models.RepoRecord, error) {
+	o := GetOrmer()
+	var repos []models.RepoRecord
+	_, err := o.QueryTable("repository").All(&repos)
+	return repos, err
+}
+
+// DeleteRepository ...
+func DeleteRepository(name string) error {
+	o := GetOrmer()
+	_, err := o.QueryTable("repository").Filter("name", name).Delete()
+	return err
+}
+
+// UpdateRepository ...
+func UpdateRepository(repo models.RepoRecord) error {
+	o := GetOrmer()
+	_, err := o.Update(&repo)
+	return err
+}
+
+// IncreasePullCount ...
+func IncreasePullCount(name string) (err error) {
+	o := GetOrmer()
+	num, err := o.QueryTable("repository").Filter("name", name).Update(
+		orm.Params{
+			"pull_count": orm.ColValue(orm.ColAdd, 1),
+		})
+	if num == 0 {
+		err = fmt.Errorf("Failed to increase repository pull count with name: %s %s", name, err.Error())
+	}
+	return err
+}
+
+//RepositoryExists returns whether the repository exists according to its name.
+func RepositoryExists(name string) bool {
+	o := GetOrmer()
+	return o.QueryTable("repository").Filter("name", name).Exist()
+}

models/base.go

@@ -26,5 +26,6 @@ func init() {
 		new(User),
 		new(Project),
 		new(Role),
-		new(AccessLog))
+		new(AccessLog),
+		new(RepoRecord))
 }

models/repo.go

@@ -0,0 +1,38 @@
+/*
+   Copyright (c) 2016 VMware, Inc. All Rights Reserved.
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package models
+
+import (
+	"time"
+)
+
+// RepoRecord holds the record of an repository in DB, all the infors are from the registry notification event.
+type RepoRecord struct {
+	RepositoryID string    `orm:"column(repository_id);pk" json:"repository_id"`
+	Name         string    `orm:"column(name)" json:"name"`
+	OwnerName    string    `orm:"-"`
+	OwnerID      int64     `orm:"column(owner_id)"  json:"owner_id"`
+	ProjectName  string    `orm:"-"`
+	ProjectID    int64     `orm:"column(project_id)"  json:"project_id"`
+	Description  string    `orm:"column(description)" json:"description"`
+	PullCount    int64     `orm:"column(pull_count)" json:"pull_count"`
+	StarCount    int64     `orm:"column(star_count)" json:"star_count"`
+	CreationTime time.Time `orm:"column(creation_time);auto_now_add" json:"creation_time"`
+	UpdateTime   time.Time `orm:"column(update_time);auto_now" json:"update_time"`
+}
+
+//TableName is required by by beego orm to map RepoRecord to table repository
+func (rp *RepoRecord) TableName() string {
+	return "repository"
+}

service/notification.go

@@ -24,6 +24,7 @@ import (
 	"github.com/vmware/harbor/dao"
 	"github.com/vmware/harbor/models"
 	"github.com/vmware/harbor/service/cache"
+	"github.com/vmware/harbor/utils"
 	"github.com/vmware/harbor/utils/log"
 
 	"github.com/astaxie/beego"
@@ -55,11 +56,7 @@ func (n *NotificationHandler) Post() {
 	for _, event := range events {
 		repository := event.Target.Repository
 
-		project := ""
-		if strings.Contains(repository, "/") {
-			project = repository[0:strings.LastIndex(repository, "/")]
-		}
-
+		project, _ := utils.ParseRepository(repository)
 		tag := event.Target.Tag
 		action := event.Action
 
@@ -80,6 +77,18 @@ func (n *NotificationHandler) Post() {
 				}
 			}()
 
+			go func() {
+				exist := dao.RepositoryExists(repository)
+				if exist {
+					return
+				}
+				log.Debugf("Add repository %s into DB.", repository)
+				repoRecord := models.RepoRecord{Name: repository, OwnerName: user, ProjectName: project}
+				if err := dao.AddRepository(repoRecord); err != nil {
+					log.Errorf("Error happens when adding repository: %v", err)
+				}
+			}()
+
 			operation := ""
 			if action == "push" {
 				operation = models.RepOpTransfer
@@ -87,6 +96,14 @@ func (n *NotificationHandler) Post() {
 
 			go api.TriggerReplicationByRepository(repository, []string{tag}, operation)
 		}
+		if action == "pull" {
+			go func() {
+				log.Debugf("Increase the repository %s pull count.", repository)
+				if err := dao.IncreasePullCount(repository); err != nil {
+					log.Errorf("Error happens when increasing pull count: %v", repository)
+				}
+			}()
+		}
 	}
 }
 

ui/main.go

@@ -17,11 +17,11 @@ package main
 
 import (
 	"fmt"
+	"os"
 
 	log "github.com/vmware/harbor/utils/log"
 
-	"os"
-
+	"github.com/vmware/harbor/api"
 	_ "github.com/vmware/harbor/auth/db"
 	_ "github.com/vmware/harbor/auth/ldap"
 	"github.com/vmware/harbor/dao"
@@ -80,5 +80,8 @@ func main() {
 		log.Error(err)
 	}
 	initRouters()
+	if err := api.SyncRegistry(); err != nil {
+		log.Error(err)
+	}
 	beego.Run()
 }

ui/router.go

@@ -65,6 +65,7 @@ func initRouters() {
 	beego.Router("/api/projects/:id([0-9]+)/logs/filter", &api.ProjectAPI{}, "post:FilterAccessLog")
 	beego.Router("/api/users/?:id", &api.UserAPI{})
 	beego.Router("/api/users/:id([0-9]+)/password", &api.UserAPI{}, "put:ChangePassword")
+	beego.Router("/api/internal/syncregistry", &api.InternalAPI{}, "post:SyncRegistry")
 	beego.Router("/api/repositories", &api.RepositoryAPI{})
 	beego.Router("/api/repositories/tags", &api.RepositoryAPI{}, "get:GetTags")
 	beego.Router("/api/repositories/manifests", &api.RepositoryAPI{}, "get:GetManifests")