Add MySQL along with sqlite3 support as a registry events storage.

This commit is contained in:
Roman Vynar 2018-05-28 15:36:27 +03:00
parent 20daa971e5
commit c5a5fe2e8d
14 changed files with 270 additions and 183 deletions

11
CHANGELOG.md Normal file
View File

@ -0,0 +1,11 @@
## Changelog
### 0.6
* Add MySQL along with sqlite3 support as a registry events storage.
New config settings `event_database_driver`, `event_database_location`.
* Bump Go version and dependencies.
### 0.5
* Initial public version.

View File

@ -10,6 +10,7 @@ ADD glide.* /opt/src/github.com/quiq/docker-registry-ui/
RUN cd /opt/src/github.com/quiq/docker-registry-ui && \
/opt/bin/glide install
ADD events /opt/src/github.com/quiq/docker-registry-ui/events
ADD registry /opt/src/github.com/quiq/docker-registry-ui/registry
ADD main.go /opt/src/github.com/quiq/docker-registry-ui/
RUN cd /opt/src/github.com/quiq/docker-registry-ui && \

View File

@ -11,6 +11,7 @@
* Automatically discover an authentication method (basic auth, token service etc.)
* Caching the list of repositories, tag counts and refreshing in background
* Event listener of notification events coming from Registry
* Store events in sqlite or MySQL database
* CLI option to maintain the tags retention: purge tags older than X days keeping at least Y tags
No TLS or authentication implemented on the UI web server itself.
@ -56,6 +57,23 @@ To receive events you need to configure Registry as follow:
Adjust url and token as appropriate.
## Using MySQL instead of sqlite3 for event listener
To use MySQL as a storage you need to change `event_database_driver` and `event_database_location`
settings in the config file. It is expected you create a database mentioned in the location DSN.
Minimal privileges are `SELECT`, `INSERT`, `DELETE`.
You can create a table manually if you don't want to grant `CREATE` permission:
CREATE TABLE events (
id INTEGER PRIMARY KEY AUTO_INCREMENT,
action CHAR(4) NULL,
repository VARCHAR(100) NULL,
tag VARCHAR(100) NULL,
ip VARCHAR(15) NULL,
user VARCHAR(50) NULL,
created DATETIME NULL
);
### Schedule a cron task for purging tags
The following example shows how to run a cron task to purge tags older than X days but also keep

View File

@ -19,6 +19,12 @@ event_listener_token: token
# Retention of records to keep.
event_retention_days: 7
# Event listener storage.
event_database_driver: sqlite3
event_database_location: data/registry_events.db
# event_database_driver: mysql
# event_database_location: user:password@tcp(localhost:3306)/docker_events
# Cache refresh interval in minutes.
# How long to cache repository list and tag counts.
cache_refresh_interval: 10

186
events/event_listener.go Normal file
View File

@ -0,0 +1,186 @@
package events
import (
"database/sql"
"encoding/json"
"fmt"
"net/http"
"os"
"strings"
"github.com/hhkbp2/go-logging"
"github.com/quiq/docker-registry-ui/registry"
// 🐒 patching of "database/sql".
_ "github.com/go-sql-driver/mysql"
_ "github.com/mattn/go-sqlite3"
"github.com/tidwall/gjson"
)
const (
schemaSQLite = `
CREATE TABLE 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
);
`
)
// EventListener event listener
type EventListener struct {
databaseDriver string
databaseLocation string
retention int
logger logging.Logger
}
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 string
}
// NewEventListener initialize EventListener.
func NewEventListener(databaseDriver, databaseLocation string, retention int) *EventListener {
return &EventListener{
databaseDriver: databaseDriver,
databaseLocation: databaseLocation,
retention: retention,
logger: registry.SetupLogging("events.event_listener"),
}
}
// ProcessEvents parse and store registry events
func (e *EventListener) ProcessEvents(request *http.Request) {
decoder := json.NewDecoder(request.Body)
var t eventData
if err := decoder.Decode(&t); err != nil {
e.logger.Errorf("Problem decoding event from request: %+v", request)
return
}
e.logger.Debugf("Received event: %+v", t)
j, _ := json.Marshal(t)
db, err := e.getDababaseHandler()
if err != nil {
e.logger.Error(err)
return
}
defer db.Close()
now := "DateTime('now')"
if e.databaseDriver == "mysql" {
now = "NOW()"
}
stmt, _ := db.Prepare("INSERT INTO events(action, repository, tag, ip, user, created) values(?,?,?,?,?," + now + ")")
for _, i := range gjson.GetBytes(j, "events").Array() {
// Ignore calls by docker-registry-ui itself.
if i.Get("request.useragent").String() == "docker-registry-ui" {
continue
}
action := i.Get("action").String()
repository := i.Get("target.repository").String()
tag := i.Get("target.tag").String()
// Tag is empty in case of signed pull.
if tag == "" {
tag = i.Get("target.digest").String()
}
ip := strings.Split(i.Get("request.addr").String(), ":")[0]
user := i.Get("actor.name").String()
e.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 {
e.logger.Error("Error inserting a row: ", err)
return
}
id, _ := res.LastInsertId()
e.logger.Debug("New event added with id ", id)
}
// Purge old records.
var res sql.Result
if e.databaseDriver == "mysql" {
stmt, _ := db.Prepare("DELETE FROM events WHERE created < DATE_SUB(NOW(), INTERVAL ? DAY)")
res, _ = stmt.Exec(e.retention)
} else {
stmt, _ := db.Prepare("DELETE FROM events WHERE created < DateTime('now',?)")
res, _ = stmt.Exec(fmt.Sprintf("-%d day", e.retention))
}
count, _ := res.RowsAffected()
e.logger.Debug("Rows deleted: ", count)
}
// GetEvents retrieve events from sqlite db
func (e *EventListener) GetEvents(repository string) []EventRow {
var events []EventRow
db, err := e.getDababaseHandler()
if err != nil {
e.logger.Error(err)
return events
}
defer db.Close()
query := "SELECT * FROM events ORDER BY id DESC LIMIT 1000"
if repository != "" {
query = fmt.Sprintf("SELECT * FROM events WHERE repository='%s' ORDER BY id DESC LIMIT 5", repository)
}
rows, err := db.Query(query)
if err != nil {
e.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
}
func (e *EventListener) getDababaseHandler() (*sql.DB, error) {
firstRun := false
schema := schemaSQLite
if e.databaseDriver == "sqlite3" {
if _, err := os.Stat(e.databaseLocation); os.IsNotExist(err) {
firstRun = true
}
}
// Open db connection.
db, err := sql.Open(e.databaseDriver, e.databaseLocation)
if err != nil {
return nil, fmt.Errorf("Error opening %s db: %s", e.databaseDriver, err)
}
if e.databaseDriver == "mysql" {
schema = strings.Replace(schema, "AUTOINCREMENT", "AUTO_INCREMENT", 1)
if _, err := db.Query("SELECT * FROM events LIMIT 1"); err != nil {
firstRun = true
}
}
// Create table on first run.
if firstRun {
if _, err = db.Exec(schema); err != nil {
return nil, fmt.Errorf("Error creating a table: %s", err)
}
}
return db, nil
}

14
glide.lock generated
View File

@ -1,5 +1,5 @@
hash: 35e7a8422393703991e14940d1fa372418c561f5dd5200a2a5a4d7e09813eeee
updated: 2018-05-28T11:05:19.474208+03:00
hash: d156899e94e2d0d92ed200d5729bec5d0d205b5b98bbf2bd89f7f91c9ed7a518
updated: 2018-05-28T13:28:11.313447+03:00
imports:
- name: github.com/CloudyKit/fastprinter
version: 74b38d55f37af5d6c05ca11147d616b613a3420e
@ -7,10 +7,8 @@ imports:
version: 2b064536b25ab0e9c54245f9e2cc5bd4766033fe
- name: github.com/dgrijalva/jwt-go
version: 06ea1031745cb8b3dab3f6a236daf2b0aa468b7e
- name: github.com/grafana/grafana
version: 238139fad67d0d6e7f5e54d515568f2ab789a26e
subpackages:
- pkg/cmd/grafana-cli/logger
- name: github.com/go-sql-driver/mysql
version: 64db0f7ebe171b596aa9b26f39a79f7413a3b617
- name: github.com/hhkbp2/go-logging
version: 377ba05d98977baa2a0d9e13f13aac2f3a47ac4d
- name: github.com/hhkbp2/go-strftime
@ -59,6 +57,10 @@ imports:
version: 7c87d13f8e835d2fb3a70a2912c811ed0c1d241b
subpackages:
- unix
- name: google.golang.org/appengine
version: b1f26356af11148e710935ed1ac8a7f5702c7612
subpackages:
- cloudsql
- name: gopkg.in/yaml.v2
version: 5420a8b6744d3b0345ab293f6fcba19c978f1183
testImports:

View File

@ -13,6 +13,7 @@ import:
version: v1.1.0
- package: github.com/mattn/go-sqlite3
version: 1.7.0
- package: github.com/go-sql-driver/mysql
testImport:
- package: github.com/smartystreets/goconvey
version: 1.6.2

55
main.go
View File

@ -13,25 +13,28 @@ import (
"github.com/CloudyKit/jet"
"github.com/labstack/echo"
"github.com/labstack/echo/middleware"
"github.com/quiq/docker-registry-ui/events"
"github.com/quiq/docker-registry-ui/registry"
"github.com/tidwall/gjson"
"gopkg.in/yaml.v2"
)
type configData struct {
ListenAddr string `yaml:"listen_addr"`
RegistryURL string `yaml:"registry_url"`
VerifyTLS bool `yaml:"verify_tls"`
Username string `yaml:"registry_username"`
Password string `yaml:"registry_password"`
EventListenerToken string `yaml:"event_listener_token"`
EventRetentionDays int `yaml:"event_retention_days"`
CacheRefreshInterval uint8 `yaml:"cache_refresh_interval"`
AnyoneCanDelete bool `yaml:"anyone_can_delete"`
Admins []string `yaml:"admins"`
Debug bool `yaml:"debug"`
PurgeTagsKeepDays int `yaml:"purge_tags_keep_days"`
PurgeTagsKeepCount int `yaml:"purge_tags_keep_count"`
ListenAddr string `yaml:"listen_addr"`
RegistryURL string `yaml:"registry_url"`
VerifyTLS bool `yaml:"verify_tls"`
Username string `yaml:"registry_username"`
Password string `yaml:"registry_password"`
EventListenerToken string `yaml:"event_listener_token"`
EventRetentionDays int `yaml:"event_retention_days"`
EventDatabaseDriver string `yaml:"event_database_driver"`
EventDatabaseLocation string `yaml:"event_database_location"`
CacheRefreshInterval uint8 `yaml:"cache_refresh_interval"`
AnyoneCanDelete bool `yaml:"anyone_can_delete"`
Admins []string `yaml:"admins"`
Debug bool `yaml:"debug"`
PurgeTagsKeepDays int `yaml:"purge_tags_keep_days"`
PurgeTagsKeepCount int `yaml:"purge_tags_keep_count"`
}
type template struct {
@ -39,8 +42,9 @@ type template struct {
}
type apiClient struct {
client *registry.Client
config configData
client *registry.Client
eventListener *events.EventListener
config configData
}
func main() {
@ -87,6 +91,11 @@ func main() {
// Count tags in background.
go a.client.CountTags(a.config.CacheRefreshInterval)
if a.config.EventDatabaseDriver != "sqlite3" && a.config.EventDatabaseDriver != "mysql" {
panic(fmt.Errorf("event_database_driver should be either sqlite3 or mysql"))
}
a.eventListener = events.NewEventListener(a.config.EventDatabaseDriver, a.config.EventDatabaseLocation, a.config.EventRetentionDays)
// Template engine init.
view := jet.NewHTMLSet("templates")
view.SetDevelopmentMode(a.config.Debug)
@ -102,7 +111,9 @@ func main() {
return registry.PrettySize(value)
})
view.AddGlobal("pretty_time", func(datetime interface{}) string {
return strings.Split(strings.Replace(datetime.(string), "T", " ", 1), ".")[0]
d := strings.Replace(datetime.(string), "T", " ", 1)
d = strings.Replace(d, "Z", "", 1)
return strings.Split(d, ".")[0]
})
view.AddGlobal("parse_map", func(m interface{}) string {
var res string
@ -130,7 +141,7 @@ func main() {
return token == a.config.EventListenerToken, nil
}),
}))
p.POST("/events", a.eventListener)
p.POST("/events", a.receiveEvents)
e.Logger.Fatal(e.Start(a.config.ListenAddr))
}
@ -185,7 +196,7 @@ func (a *apiClient) viewTags(c echo.Context) error {
data.Set("repo", repo)
data.Set("tags", tags)
data.Set("deleteAllowed", deleteAllowed)
data.Set("events", registry.GetEvents(repo))
data.Set("events", a.eventListener.GetEvents(repo))
return c.Render(http.StatusOK, "tags.html", data)
}
@ -281,13 +292,13 @@ func (a *apiClient) checkDeletePermission(user string) bool {
// viewLog view events from sqlite.
func (a *apiClient) viewLog(c echo.Context) error {
data := jet.VarMap{}
data.Set("events", registry.GetEvents(""))
data.Set("events", a.eventListener.GetEvents(""))
return c.Render(http.StatusOK, "event_log.html", data)
}
// eventListener listen events from registry.
func (a *apiClient) eventListener(c echo.Context) error {
registry.ProcessEvents(c.Request(), a.config.EventRetentionDays)
// receiveEvents receive events.
func (a *apiClient) receiveEvents(c echo.Context) error {
a.eventListener.ProcessEvents(c.Request())
return c.String(http.StatusOK, "OK")
}

View File

@ -38,7 +38,7 @@ func NewClient(url string, verifyTLS bool, username, password string) *Client {
password: password,
request: gorequest.New().TLSClientConfig(&tls.Config{InsecureSkipVerify: !verifyTLS}),
logger: setupLogging("registry.client"),
logger: SetupLogging("registry.client"),
tokens: map[string]string{},
repos: map[string][]string{},
tagCounts: map[string]int{},

View File

@ -8,8 +8,8 @@ import (
"github.com/hhkbp2/go-logging"
)
// setupLogging configure logging.
func setupLogging(name string) logging.Logger {
// 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"

View File

@ -1,149 +0,0 @@
package registry
import (
"database/sql"
"encoding/json"
"fmt"
"net/http"
"os"
"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 := getDababaseHandler()
defer db.Close()
if err != nil {
logger.Error(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(repository string) []EventRow {
var events []EventRow
db, err := getDababaseHandler()
defer db.Close()
if err != nil {
logger.Error(err)
return events
}
query := "SELECT * FROM events ORDER BY id DESC LIMIT 1000"
if repository != "" {
query = fmt.Sprintf("SELECT * FROM events WHERE repository='%s' ORDER BY id DESC LIMIT 5", repository)
}
rows, err := db.Query(query)
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
}
func getDababaseHandler() (*sql.DB, error) {
firstRun := false
if _, err := os.Stat(dbFile); os.IsNotExist(err) {
firstRun = true
}
// Open db file and create if needed.
db, err := sql.Open("sqlite3", dbFile)
if err != nil {
return nil, fmt.Errorf("Error opening sqlite db: %s", err)
}
// Create table on first run.
if firstRun {
if _, err = db.Exec(schema); err != nil {
return nil, fmt.Errorf("Error creating a table: %s", err)
}
}
return db, nil
}

View File

@ -34,7 +34,7 @@ func (p timeSlice) Swap(i, j int) {
// PurgeOldTags purge old tags.
func PurgeOldTags(client *Client, purgeDryRun bool, purgeTagsKeepDays, purgeTagsKeepCount int) {
logger := setupLogging("registry.tasks.PurgeOldTags")
logger := SetupLogging("registry.tasks.PurgeOldTags")
// Reduce client logging.
client.logger.SetLevel(logging.LevelError)

View File

@ -39,7 +39,7 @@
{{end}}
<td>{{ e.IP }}</td>
<td>{{ e.User }}</td>
<td>{{ e.Created }}</td>
<td>{{ e.Created|pretty_time }}</td>
</tr>
{{end}}
</tbody>

View File

@ -78,7 +78,7 @@
{{end}}
<td>{{ e.IP }}</td>
<td>{{ e.User }}</td>
<td>{{ e.Created }}</td>
<td>{{ e.Created|pretty_time }}</td>
</tr>
{{end}}
</tbody>