Initial commit.

registry/client.go

@@ -0,0 +1,245 @@
+package registry
+
+import (
+	"crypto/tls"
+	"fmt"
+	"regexp"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/hhkbp2/go-logging"
+	"github.com/parnurzeal/gorequest"
+	"github.com/tidwall/gjson"
+)
+
+// Client main class.
+type Client struct {
+	url       string
+	verifyTLS bool
+	username  string
+	password  string
+	request   *gorequest.SuperAgent
+	logger    logging.Logger
+	mux       sync.Mutex
+	tokens    map[string]string
+	repos     map[string][]string
+	tagCounts map[string]int
+	authURL   string
+}
+
+// NewClient initialize Client.
+func NewClient(url string, verifyTLS bool, username, password string) *Client {
+	c := &Client{
+		url:       strings.TrimRight(url, "/"),
+		verifyTLS: verifyTLS,
+		username:  username,
+		password:  password,
+
+		request:   gorequest.New().TLSClientConfig(&tls.Config{InsecureSkipVerify: !verifyTLS}),
+		logger:    setupLogging("registry.client"),
+		tokens:    map[string]string{},
+		repos:     map[string][]string{},
+		tagCounts: map[string]int{},
+	}
+	resp, _, errs := c.request.Get(c.url+"/v2/").Set("User-Agent", "docker-registry-ui").End()
+	if len(errs) > 0 {
+		c.logger.Error(errs[0])
+		return nil
+	}
+
+	authHeader := ""
+	if resp.StatusCode == 200 {
+		return c
+	} else if resp.StatusCode == 401 {
+		authHeader = resp.Header.Get("WWW-Authenticate")
+	} else {
+		c.logger.Error(resp.Status)
+		return nil
+	}
+
+	if strings.HasPrefix(authHeader, "Bearer") {
+		r, _ := regexp.Compile(`^Bearer realm="(http.+)",service="(.+)"`)
+		if m := r.FindStringSubmatch(authHeader); len(m) > 0 {
+			c.authURL = fmt.Sprintf("%s?service=%s", m[1], m[2])
+			c.logger.Info("Token auth service discovered at ", c.authURL)
+		}
+		if c.authURL == "" {
+			c.logger.Warn("No token auth service discovered from ", c.url)
+			return nil
+		}
+	} else if strings.HasPrefix(authHeader, "Basic") {
+		c.request = c.request.SetBasicAuth(c.username, c.password)
+		c.logger.Info("It was discovered the registry is configured with HTTP basic auth.")
+	}
+
+	return c
+}
+
+// getToken get existing or new auth token.
+func (c *Client) getToken(scope string) string {
+	// Check if we have already a token and it's not expired.
+	if token, ok := c.tokens[scope]; ok {
+		resp, _, _ := c.request.Get(c.url+"/v2/").Set("Authorization", fmt.Sprintf("Bearer %s", token)).Set("User-Agent", "docker-registry-ui").End()
+		if resp != nil && resp.StatusCode == 200 {
+			return token
+		}
+	}
+
+	request := gorequest.New().TLSClientConfig(&tls.Config{InsecureSkipVerify: !c.verifyTLS})
+	resp, data, errs := request.Get(fmt.Sprintf("%s&scope=%s", c.authURL, scope)).SetBasicAuth(c.username, c.password).Set("User-Agent", "docker-registry-ui").End()
+	if len(errs) > 0 {
+		c.logger.Error(errs[0])
+		return ""
+	}
+	if resp.StatusCode != 200 {
+		c.logger.Error("Failed to get token for scope ", scope, " from ", c.authURL)
+		return ""
+	}
+
+	c.tokens[scope] = gjson.Get(data, "token").String()
+	c.logger.Info("Received new token for scope ", scope)
+
+	return c.tokens[scope]
+}
+
+// callRegistry make an HTTP request to Docker registry.
+func (c *Client) callRegistry(uri, scope string, manifest uint, delete bool) (rdata, rdigest string) {
+	acceptHeader := fmt.Sprintf("application/vnd.docker.distribution.manifest.v%d+json", manifest)
+	authHeader := ""
+	if c.authURL != "" {
+		authHeader = fmt.Sprintf("Bearer %s", c.getToken(scope))
+	}
+
+	resp, data, errs := c.request.Get(c.url+uri).Set("Accept", acceptHeader).Set("Authorization", authHeader).Set("User-Agent", "docker-registry-ui").End()
+	if len(errs) > 0 {
+		c.logger.Error(errs[0])
+		return "", ""
+	}
+
+	c.logger.Info("GET ", uri, " ", resp.Status)
+	// Returns 404 when no tags in the repo.
+	if resp.StatusCode != 200 {
+		return "", ""
+	}
+	digest := resp.Header.Get("Docker-Content-Digest")
+
+	if delete {
+		// Delete by manifest digest reference.
+		parts := strings.Split(uri, "/manifests/")
+		uri = parts[0] + "/manifests/" + digest
+		resp, _, errs := c.request.Delete(c.url+uri).Set("Accept", acceptHeader).Set("Authorization", authHeader).Set("User-Agent", "docker-registry-ui").End()
+		if len(errs) > 0 {
+			c.logger.Error(errs[0])
+		} else {
+			// Returns 202 on success.
+			c.logger.Info("DELETE ", uri, " (", parts[1], ") ", resp.Status)
+		}
+		return "", ""
+	}
+
+	return data, digest
+}
+
+// Namespaces list repo namespaces.
+func (c *Client) Namespaces() []string {
+	namespaces := make([]string, 0, len(c.repos))
+	for k := range c.repos {
+		namespaces = append(namespaces, k)
+	}
+	sort.Strings(namespaces)
+	return namespaces
+}
+
+// Repositories list repos by namespaces where 'library' is the default one.
+func (c *Client) Repositories(useCache bool) map[string][]string {
+	// Return from cache if available.
+	if len(c.repos) > 0 && useCache {
+		return c.repos
+	}
+
+	c.mux.Lock()
+	defer c.mux.Unlock()
+	scope := "registry:catalog:*"
+	data, _ := c.callRegistry("/v2/_catalog", scope, 2, false)
+	if data == "" {
+		return c.repos
+	}
+
+	c.repos = map[string][]string{}
+	for _, r := range gjson.Get(data, "repositories").Array() {
+		namespace := "library"
+		repo := r.String()
+		if strings.Contains(repo, "/") {
+			f := strings.SplitN(repo, "/", 2)
+			namespace = f[0]
+			repo = f[1]
+		}
+		c.repos[namespace] = append(c.repos[namespace], repo)
+	}
+
+	return c.repos
+}
+
+// Tags get tags for the repo.
+func (c *Client) Tags(repo string) []string {
+	scope := fmt.Sprintf("repository:%s:*", repo)
+	data, _ := c.callRegistry(fmt.Sprintf("/v2/%s/tags/list", repo), scope, 2, false)
+	var tags []string
+	for _, t := range gjson.Get(data, "tags").Array() {
+		tags = append(tags, t.String())
+	}
+	return tags
+}
+
+// TagInfo get image info for the repo tag.
+func (c *Client) TagInfo(repo, tag string, v1only bool) (rsha256, rinfoV1, rinfoV2 string) {
+	scope := fmt.Sprintf("repository:%s:*", repo)
+	infoV1, _ := c.callRegistry(fmt.Sprintf("/v2/%s/manifests/%s", repo, tag), scope, 1, false)
+	if infoV1 == "" {
+		return "", "", ""
+	}
+
+	if v1only {
+		return "", infoV1, ""
+	}
+
+	infoV2, digest := c.callRegistry(fmt.Sprintf("/v2/%s/manifests/%s", repo, tag), scope, 2, false)
+	if infoV2 == "" || digest == "" {
+		return "", "", ""
+	}
+
+	sha256 := digest[7:]
+	return sha256, infoV1, infoV2
+}
+
+// TagCounts return map with tag counts.
+func (c *Client) TagCounts() map[string]int {
+	return c.tagCounts
+}
+
+// CountTags count repository tags in background regularly.
+func (c *Client) CountTags(interval uint8) {
+	for {
+		c.logger.Info("Calculating tags in background...")
+		catalog := c.Repositories(false)
+		for n, repos := range catalog {
+			for _, r := range repos {
+				repoPath := r
+				if n != "library" {
+					repoPath = fmt.Sprintf("%s/%s", n, r)
+				}
+				c.tagCounts[fmt.Sprintf("%s/%s", n, r)] = len(c.Tags(repoPath))
+			}
+		}
+		c.logger.Info("Tags calculation complete.")
+		time.Sleep(time.Duration(interval) * time.Minute)
+	}
+}
+
+// DeleteTag delete image tag.
+func (c *Client) DeleteTag(repo, tag string) {
+	scope := fmt.Sprintf("repository:%s:*", repo)
+	c.callRegistry(fmt.Sprintf("/v2/%s/manifests/%s", repo, tag), scope, 2, true)
+}

registry/common.go

@@ -0,0 +1,44 @@
+package registry
+
+import (
+	"fmt"
+	"reflect"
+	"sort"
+
+	"github.com/hhkbp2/go-logging"
+)
+
+// setupLogging configure logging.
+func setupLogging(name string) logging.Logger {
+	logger := logging.GetLogger(name)
+	handler := logging.NewStdoutHandler()
+	format := "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+	dateFormat := "%Y-%m-%d %H:%M:%S"
+	formatter := logging.NewStandardFormatter(format, dateFormat)
+	handler.SetFormatter(formatter)
+	logger.SetLevel(logging.LevelInfo)
+	logger.AddHandler(handler)
+	return logger
+}
+
+// SortedMapKeys sort keys of the map where values can be of any type.
+func SortedMapKeys(m interface{}) []string {
+	v := reflect.ValueOf(m)
+	keys := make([]string, 0, len(v.MapKeys()))
+	for _, key := range v.MapKeys() {
+		keys = append(keys, key.String())
+	}
+	sort.Strings(keys)
+	return keys
+}
+
+// PrettySize format bytes in more readable units.
+func PrettySize(size float64) string {
+	units := []string{"B", "KB", "MB", "GB"}
+	i := 0
+	for size > 1024 && i < len(units) {
+		size = size / 1024
+		i = i + 1
+	}
+	return fmt.Sprintf("%.*f %s", 0, size, units[i])
+}

registry/common_test.go

@@ -0,0 +1,47 @@
+package registry
+
+import (
+	"testing"
+	"time"
+
+	"github.com/smartystreets/goconvey/convey"
+)
+
+func TestSortedMapKeys(t *testing.T) {
+	a := map[string]string{
+		"foo": "bar",
+		"abc": "bar",
+		"zoo": "bar",
+	}
+	b := map[string]timeSlice{
+		"zoo": []tagData{tagData{name: "1", created: time.Now()}},
+		"abc": []tagData{tagData{name: "1", created: time.Now()}},
+		"foo": []tagData{tagData{name: "1", created: time.Now()}},
+	}
+	c := map[string][]string{
+		"zoo": []string{"1", "2"},
+		"foo": []string{"1", "2"},
+		"abc": []string{"1", "2"},
+	}
+	expect := []string{"abc", "foo", "zoo"}
+	convey.Convey("Sort map keys", t, func() {
+		convey.So(SortedMapKeys(a), convey.ShouldResemble, expect)
+		convey.So(SortedMapKeys(b), convey.ShouldResemble, expect)
+		convey.So(SortedMapKeys(c), convey.ShouldResemble, expect)
+	})
+}
+
+func TestPrettySize(t *testing.T) {
+	convey.Convey("Format bytes", t, func() {
+		input := map[float64]string{
+			123:        "123 B",
+			23123:      "23 KB",
+			23923:      "23 KB",
+			723425120:  "690 MB",
+			8534241213: "8 GB",
+		}
+		for key, val := range input {
+			convey.So(PrettySize(key), convey.ShouldEqual, val)
+		}
+	})
+}

registry/event_listener.go

@@ -0,0 +1,128 @@
+package registry
+
+import (
+	"database/sql"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"github.com/grafana/grafana/pkg/cmd/grafana-cli/logger"
+	// 🐒 patching of "database/sql".
+	_ "github.com/mattn/go-sqlite3"
+	"github.com/tidwall/gjson"
+)
+
+const (
+	dbFile = "data/registry_events.db"
+	schema = `
+	CREATE TABLE IF NOT EXISTS events (
+		id INTEGER PRIMARY KEY AUTOINCREMENT,
+		action CHAR(4) NULL,
+		repository VARCHAR(100) NULL,
+		tag VARCHAR(100) NULL,
+		ip VARCHAR(15) NULL,
+		user VARCHAR(50) NULL,
+		created DATETIME NULL
+	);
+`
+)
+
+type eventData struct {
+	Events []interface{} `json:"events"`
+}
+
+// EventRow event row from sqlite
+type EventRow struct {
+	ID         int
+	Action     string
+	Repository string
+	Tag        string
+	IP         string
+	User       string
+	Created    time.Time
+}
+
+// ProcessEvents parse and store registry events
+func ProcessEvents(request *http.Request, retention int) {
+	logger := setupLogging("registry.event_listener")
+	decoder := json.NewDecoder(request.Body)
+	var t eventData
+	if err := decoder.Decode(&t); err != nil {
+		logger.Errorf("Problem decoding event from request: %+v", request)
+		return
+	}
+	logger.Debugf("Received event: %+v", t)
+	j, _ := json.Marshal(t)
+
+	db, err := sql.Open("sqlite3", dbFile)
+	if err != nil {
+		logger.Error("Error opening sqlite db: ", err)
+		return
+	}
+	defer db.Close()
+
+	_, err = db.Exec(schema)
+	if err != nil {
+		logger.Error("Error creating a table: ", err)
+		return
+	}
+
+	stmt, _ := db.Prepare("INSERT INTO events(action, repository, tag, ip, user, created) values(?,?,?,?,?,DateTime('now'))")
+	for _, e := range gjson.GetBytes(j, "events").Array() {
+		// Ignore calls by docker-registry-ui itself.
+		if e.Get("request.useragent").String() == "docker-registry-ui" {
+			continue
+		}
+		action := e.Get("action").String()
+		repository := e.Get("target.repository").String()
+		tag := e.Get("target.tag").String()
+		// Tag is empty in case of signed pull.
+		if tag == "" {
+			tag = e.Get("target.digest").String()
+		}
+		ip := strings.Split(e.Get("request.addr").String(), ":")[0]
+		user := e.Get("actor.name").String()
+		logger.Debugf("Parsed event data: %s %s:%s %s %s ", action, repository, tag, ip, user)
+
+		res, err := stmt.Exec(action, repository, tag, ip, user)
+		if err != nil {
+			logger.Error("Error inserting a row: ", err)
+			return
+		}
+		id, _ := res.LastInsertId()
+		logger.Debug("New event added with id ", id)
+	}
+
+	// Purge old records.
+	stmt, _ = db.Prepare("DELETE FROM events WHERE created < DateTime('now',?)")
+	res, _ := stmt.Exec(fmt.Sprintf("-%d day", retention))
+	count, _ := res.RowsAffected()
+	logger.Debug("Rows deleted: ", count)
+}
+
+// GetEvents retrieve events from sqlite db
+func GetEvents() []EventRow {
+	var events []EventRow
+	db, err := sql.Open("sqlite3", dbFile)
+	if err != nil {
+		logger.Error("Error opening sqlite db: ", err)
+		return events
+	}
+	defer db.Close()
+
+	rows, err := db.Query("SELECT * FROM events ORDER BY id DESC LIMIT 1000")
+	if err != nil {
+		logger.Error("Error selecting from table: ", err)
+		return events
+	}
+
+	for rows.Next() {
+		var row EventRow
+		rows.Scan(&row.ID, &row.Action, &row.Repository, &row.Tag, &row.IP, &row.User, &row.Created)
+		events = append(events, row)
+	}
+	rows.Close()
+	return events
+}

registry/tasks.go

@@ -0,0 +1,132 @@
+package registry
+
+import (
+	"fmt"
+	"sort"
+	"time"
+
+	"github.com/hhkbp2/go-logging"
+	"github.com/tidwall/gjson"
+)
+
+type tagData struct {
+	name    string
+	created time.Time
+}
+
+func (t tagData) String() string {
+	return fmt.Sprintf(`"%s <%s>"`, t.name, t.created.Format("2006-01-02 15:04:05"))
+}
+
+type timeSlice []tagData
+
+func (p timeSlice) Len() int {
+	return len(p)
+}
+
+func (p timeSlice) Less(i, j int) bool {
+	return p[i].created.After(p[j].created)
+}
+
+func (p timeSlice) Swap(i, j int) {
+	p[i], p[j] = p[j], p[i]
+}
+
+// PurgeOldTags purge old tags.
+func PurgeOldTags(client *Client, purgeDryRun bool, purgeTagsKeepDays, purgeTagsKeepCount int) {
+	logger := setupLogging("registry.tasks.PurgeOldTags")
+	// Reduce client logging.
+	client.logger.SetLevel(logging.LevelError)
+
+	dryRunText := ""
+	if purgeDryRun {
+		logger.Warn("Dry-run mode enabled.")
+		dryRunText = "skipped"
+	}
+	logger.Info("Scanning registry for repositories, tags and their creation dates...")
+	catalog := client.Repositories(true)
+	// catalog := map[string][]string{"library": []string{""}}
+	now := time.Now().UTC()
+	repos := map[string]timeSlice{}
+	count := 0
+	for namespace := range catalog {
+		count = count + len(catalog[namespace])
+		for _, repo := range catalog[namespace] {
+			if namespace != "library" {
+				repo = fmt.Sprintf("%s/%s", namespace, repo)
+			}
+
+			tags := client.Tags(repo)
+			logger.Infof("[%s] scanning %d tags...", repo, len(tags))
+			if len(tags) == 0 {
+				continue
+			}
+			for _, tag := range tags {
+				_, infoV1, _ := client.TagInfo(repo, tag, true)
+				if infoV1 == "" {
+					logger.Errorf("[%s] manifest missed for tag %s", repo, tag)
+					continue
+				}
+				created := gjson.Get(gjson.Get(infoV1, "history.0.v1Compatibility").String(), "created").Time()
+				repos[repo] = append(repos[repo], tagData{name: tag, created: created})
+			}
+		}
+	}
+
+	logger.Infof("Scanned %d repositories.", count)
+	logger.Info("Filtering out tags for purging...")
+	purgeTags := map[string][]string{}
+	keepTags := map[string][]string{}
+	count = 0
+	for _, repo := range SortedMapKeys(repos) {
+		// Sort tags by "created" from newest to oldest.
+		sortedTags := make(timeSlice, 0, len(repos[repo]))
+		for _, d := range repos[repo] {
+			sortedTags = append(sortedTags, d)
+		}
+		sort.Sort(sortedTags)
+		repos[repo] = sortedTags
+
+		// Filter out tags by retention days.
+		for _, tag := range repos[repo] {
+			delta := int(now.Sub(tag.created).Hours() / 24)
+			if delta > purgeTagsKeepDays {
+				purgeTags[repo] = append(purgeTags[repo], tag.name)
+			} else {
+				keepTags[repo] = append(keepTags[repo], tag.name)
+			}
+		}
+
+		// Keep minimal count of tags no matter how old they are.
+		if len(repos[repo])-len(purgeTags[repo]) < purgeTagsKeepCount {
+			if len(purgeTags[repo]) > purgeTagsKeepCount {
+				keepTags[repo] = append(keepTags[repo], purgeTags[repo][:purgeTagsKeepCount]...)
+				purgeTags[repo] = purgeTags[repo][purgeTagsKeepCount:]
+			} else {
+				keepTags[repo] = append(keepTags[repo], purgeTags[repo]...)
+				delete(purgeTags, repo)
+			}
+		}
+
+		count = count + len(purgeTags[repo])
+		logger.Infof("[%s] All %d: %v", repo, len(repos[repo]), repos[repo])
+		logger.Infof("[%s] Keep %d: %v", repo, len(keepTags[repo]), keepTags[repo])
+		logger.Infof("[%s] Purge %d: %v", repo, len(purgeTags[repo]), purgeTags[repo])
+	}
+
+	logger.Infof("There are %d tags to purge.", count)
+	if count > 0 {
+		logger.Info("Purging old tags...")
+	}
+
+	for _, repo := range SortedMapKeys(purgeTags) {
+		logger.Infof("[%s] Purging %d tags... %s", repo, len(purgeTags[repo]), dryRunText)
+		if purgeDryRun {
+			continue
+		}
+		for _, tag := range purgeTags[repo] {
+			client.DeleteTag(repo, tag)
+		}
+	}
+	logger.Info("Done.")
+}