Merge pull request #47 from ctrliq/mirror-configs

Update mirror plugin

charts/beskar-mirror/values.yaml

@@ -127,4 +127,7 @@ configData:
     gcs:
       bucket: beskar-mirror
     azure:
-      container: beskar-mirror
+      container: beskar-mirror
+
+  sync:
+    max_worker_count: 50

go.mod

@@ -8,9 +8,9 @@ require (
 	github.com/RussellLuo/kun v0.4.5
 	github.com/RussellLuo/validating/v3 v3.0.0-beta.1
 	github.com/adlio/schema v1.3.4
-	github.com/antoniomika/go-rsync v0.0.0-20220817021523-f831db35f9a3
 	github.com/aws/aws-sdk-go v1.48.10
 	github.com/cavaliergopher/rpm v1.2.0
+	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cenkalti/backoff/v4 v4.2.1
 	github.com/distribution/distribution/v3 v3.0.0-alpha.1
 	github.com/distribution/reference v0.5.0
@@ -34,6 +34,7 @@ require (
 	github.com/twmb/murmur3 v1.1.8
 	github.com/ulikunitz/xz v0.5.11
 	github.com/vishvananda/netlink v1.2.1-beta.2
+	go.ciq.dev/go-rsync v0.0.0-20240304021629-0a3bb196e6d1
 	go.opentelemetry.io/otel v1.21.0
 	go.opentelemetry.io/otel/trace v1.21.0
 	gocloud.dev v0.32.0

go.sum

@@ -57,8 +57,6 @@ github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuy
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
-github.com/antoniomika/go-rsync v0.0.0-20220817021523-f831db35f9a3 h1:5mL0NCuUVoX5omCZhvf7yPW1wzgRlo3cWGLDUnN6kkM=
-github.com/antoniomika/go-rsync v0.0.0-20220817021523-f831db35f9a3/go.mod h1:rKzRO3ppwfCUpHMf/IEnJuwuGsr6yi0rlG7/RE32oEY=
 github.com/apache/thrift v0.12.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/apache/thrift v0.13.0/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/arbovm/levenshtein v0.0.0-20160628152529-48b4e1c0c4d0 h1:jfIu9sQUG6Ig+0+Ap1h4unLjW6YQJpKZVmUzxsD4E/Q=
@@ -129,6 +127,7 @@ github.com/bytecodealliance/wasmtime-go/v3 v3.0.2/go.mod h1:RnUjnIXxEJcL6BgCvNyz
 github.com/casbin/casbin/v2 v2.1.2/go.mod h1:YcPU1XXisHhLzuxH9coDNf2FbKpjGlbCg3n9yuLkIJQ=
 github.com/cavaliergopher/rpm v1.2.0 h1:s0h+QeVK252QFTolkhGiMeQ1f+tMeIMhGl8B1HUmGUc=
 github.com/cavaliergopher/rpm v1.2.0/go.mod h1:R0q3vTqa7RUvPofAZYrnjJ63hh2vngjFfphuXiExVos=
+github.com/cenkalti/backoff v2.2.1+incompatible h1:tNowT99t7UNflLxfYYSlKYsBpXdEet03Pg2g16Swow4=
 github.com/cenkalti/backoff v2.2.1+incompatible/go.mod h1:90ReRw6GdpyfrHakVjL/QHaoyV4aDUVVkXQJJJ3NXXM=
 github.com/cenkalti/backoff/v4 v4.2.1 h1:y4OZtCnogmCPw98Zjyt5a6+QwPLGkiQsYW5oUqylYbM=
 github.com/cenkalti/backoff/v4 v4.2.1/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
@@ -690,6 +689,8 @@ github.com/yashtewari/glob-intersection v0.2.0/go.mod h1:LK7pIC3piUjovexikBbJ26Y
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+go.ciq.dev/go-rsync v0.0.0-20240304021629-0a3bb196e6d1 h1:lYxtzhvoRGnoET/RcKJDnRnmaHuGKBCUIj3D1ZubBNg=
+go.ciq.dev/go-rsync v0.0.0-20240304021629-0a3bb196e6d1/go.mod h1:xOHMiPHUTm8AQpxu4n14T8bRuT/izQISy8ycm/Q3LLY=
 go.etcd.io/bbolt v1.3.3/go.mod h1:IbVyRI1SCnLcuJnV2u8VeU0CEYM7e686BmAb1XKL+uU=
 go.etcd.io/etcd v0.0.0-20191023171146-3cf2f69b5738/go.mod h1:dnLIgRNXwCJa5e+c6mIZCrds/GIG4ncV9HhK5PX7jPg=
 go.opencensus.io v0.20.1/go.mod h1:6WKK9ahsWS3RSO+PY9ZHZUfv2irvY6gN279GOPZjmmk=

internal/plugins/mirror/api.go

@@ -53,13 +53,27 @@ func (p *Plugin) SyncRepository(ctx context.Context, repository string, wait boo
 	return p.repositoryManager.Get(ctx, repository).SyncRepository(ctx, wait)
 }
 
+func (p *Plugin) GenerateRepository(ctx context.Context, repository string) (err error) {
+	if err := checkRepository(repository); err != nil {
+		return err
+	}
+	return p.repositoryManager.Get(ctx, repository).GenerateRepository(ctx)
+}
+
 func (p *Plugin) GetRepositorySyncStatus(ctx context.Context, repository string) (syncStatus *apiv1.SyncStatus, err error) {
 	if err := checkRepository(repository); err != nil {
 		return nil, err
 	}
 	return p.repositoryManager.Get(ctx, repository).GetRepositorySyncStatus(ctx)
 }
 
+func (p *Plugin) GetRepositorySyncPlan(ctx context.Context, repository string) (syncPlan *apiv1.RepositorySyncPlan, err error) {
+	if err := checkRepository(repository); err != nil {
+		return nil, err
+	}
+	return p.repositoryManager.Get(ctx, repository).GetRepositorySyncPlan(ctx)
+}
+
 func (p *Plugin) ListRepositoryLogs(ctx context.Context, repository string, page *apiv1.Page) (logs []apiv1.RepositoryLog, err error) {
 	if err := checkRepository(repository); err != nil {
 		return nil, err
@@ -80,3 +94,17 @@ func (p *Plugin) GetRepositoryFile(ctx context.Context, repository, file string)
 	}
 	return p.repositoryManager.Get(ctx, repository).GetRepositoryFile(ctx, file)
 }
+
+func (p *Plugin) GetRepositoryFileCount(ctx context.Context, repository string) (count int, err error) {
+	if err := checkRepository(repository); err != nil {
+		return 0, err
+	}
+	return p.repositoryManager.Get(ctx, repository).GetRepositoryFileCount(ctx)
+}
+
+func (p *Plugin) DeleteRepositoryFile(ctx context.Context, repository, file string) (err error) {
+	if err := checkRepository(repository); err != nil {
+		return err
+	}
+	return p.repositoryManager.Get(ctx, repository).DeleteRepositoryFile(ctx, file)
+}

internal/plugins/mirror/pkg/config/beskar-mirror.go

@@ -31,14 +31,15 @@ const (
 var defaultBeskarMirrorConfig string
 
 type BeskarMirrorConfig struct {
-	Version         string         `yaml:"version"`
-	Log             log.Config     `yaml:"log"`
-	Addr            string         `yaml:"addr"`
-	Gossip          gossip.Config  `yaml:"gossip"`
-	Storage         storage.Config `yaml:"storage"`
-	Profiling       bool           `yaml:"profiling"`
-	DataDir         string         `yaml:"datadir"`
-	ConfigDirectory string         `yaml:"-"`
+	Version         string            `yaml:"version"`
+	Log             log.Config        `yaml:"log"`
+	Addr            string            `yaml:"addr"`
+	Gossip          gossip.Config     `yaml:"gossip"`
+	Storage         storage.Config    `yaml:"storage"`
+	Profiling       bool              `yaml:"profiling"`
+	DataDir         string            `yaml:"datadir"`
+	ConfigDirectory string            `yaml:"-"`
+	Sync            config.SyncConfig `yaml:"sync"`
 }
 
 func (bc BeskarMirrorConfig) ListenIP() (string, error) {

internal/plugins/mirror/pkg/config/beskar-mirror_test.go

@@ -47,4 +47,6 @@ func TestParseBeskarMirrorConfig(t *testing.T) {
 	require.Equal(t, "0.0.0.0:5501", bc.Gossip.Addr)
 	require.Equal(t, "XD1IOhcp0HWFgZJ/HAaARqMKJwfMWtz284Yj7wxmerA=", bc.Gossip.Key)
 	require.Equal(t, []string{"127.0.0.1:5102"}, bc.Gossip.Peers)
+
+	require.Equal(t, 50, bc.Sync.MaxWorkerCount)
 }

internal/plugins/mirror/pkg/config/default/beskar-mirror.yaml

@@ -34,4 +34,7 @@ storage:
   azure:
     container: beskar-mirror
     account-name: account_name
-    account-key: base64_encoded_account_key
+    account-key: base64_encoded_account_key
+
+sync:
+  max_worker_count: 50

internal/plugins/mirror/pkg/index/embedded/index.html.tpl

@@ -11,7 +11,9 @@ td, th {
 <hr>
 <table>
   <tbody>
+{{- if .Previous }}
     <td><a href="{{ .Previous }}">../</td>
+{{- end }}
 {{- range $dir := .Directories }}
     <tr>
       <td><a href="{{ $dir.Ref }}/">{{ $dir.Name }}/</a></td>

internal/plugins/mirror/pkg/mirrordb/repository.go

@@ -20,10 +20,13 @@ var repositorySchemas embed.FS
 type RepositoryFile struct {
 	Tag          string `db:"tag"`
 	Name         string `db:"name"`
+	Reference    string `db:"reference"`
+	Parent       string `db:"parent"`
 	Link         string `db:"link"`
 	ModifiedTime int64  `db:"modified_time"`
 	Mode         uint32 `db:"mode"`
 	Size         uint64 `db:"size"`
+	ConfigID     uint64 `db:"config_id"`
 }
 
 type RepositoryDB struct {
@@ -56,15 +59,15 @@ func (db *RepositoryDB) AddFile(ctx context.Context, file *RepositoryFile) error
 	}
 
 	//nolint:gosec
-	s := md5.Sum([]byte(file.Name))
+	s := md5.Sum([]byte(file.Reference))
 	file.Tag = hex.EncodeToString(s[:])
 
 	db.Lock()
 	result, err := db.NamedExecContext(
 		ctx,
 		// BE CAREFUL and respect the table's columns order !!
-		"INSERT INTO files VALUES(:tag, :name, :link, :modified_time, :mode, :size) "+
-			"ON CONFLICT (tag) DO UPDATE SET name = :name, link = :link, modified_time = :modified_time, mode = :mode, size = :size",
+		"INSERT INTO files VALUES(:tag, :name, :reference, :parent, :link, :modified_time, :mode, :size, :config_id) "+
+			"ON CONFLICT (tag) DO UPDATE SET name = :name, reference = :reference, parent = :parent, link = :link, modified_time = :modified_time, mode = :mode, size = :size, config_id = :config_id",
 		file,
 	)
 	db.Unlock()
@@ -159,6 +162,56 @@ func (db *RepositoryDB) GetFileByName(ctx context.Context, name string) (*Reposi
 	return file, nil
 }
 
+func (db *RepositoryDB) GetFileByReference(ctx context.Context, reference string) (*RepositoryFile, error) {
+	db.Reference.Add(1)
+	defer db.Reference.Add(-1)
+
+	if err := db.Open(ctx); err != nil {
+		return nil, err
+	}
+
+	rows, err := db.QueryxContext(ctx, "SELECT * FROM files WHERE reference = ? LIMIT 1", reference)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	file := new(RepositoryFile)
+
+	if !rows.Next() {
+		return nil, sqlite.ErrNoEntryFound
+	}
+	if err := rows.StructScan(file); err != nil {
+		return nil, err
+	}
+
+	return file, nil
+}
+
+func (db *RepositoryDB) DeleteFileByName(ctx context.Context, name string) error {
+	db.Reference.Add(1)
+	defer db.Reference.Add(-1)
+
+	if err := db.Open(ctx); err != nil {
+		return err
+	}
+
+	db.Lock()
+	result, err := db.ExecContext(ctx, "DELETE FROM files WHERE name = ?", name)
+	db.Unlock()
+
+	if err != nil {
+		return err
+	}
+
+	_, err = result.RowsAffected()
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
 type WalkFileFunc func(*RepositoryFile) error
 
 func (db *RepositoryDB) WalkFiles(ctx context.Context, walkFn WalkFileFunc) error {
@@ -223,6 +276,101 @@ func (db *RepositoryDB) WalkSymlinks(ctx context.Context, walkFn WalkFileFunc) e
 	return nil
 }
 
+func (db *RepositoryDB) WalkFilesByParent(ctx context.Context, parent string, walkFn WalkFileFunc) error {
+	if walkFn == nil {
+		return fmt.Errorf("no file walk function provided")
+	}
+
+	db.Reference.Add(1)
+	defer db.Reference.Add(-1)
+
+	if err := db.Open(ctx); err != nil {
+		return err
+	}
+
+	rows, err := db.QueryxContext(ctx, "SELECT * FROM files WHERE parent = ?", parent)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		file := new(RepositoryFile)
+		err := rows.StructScan(file)
+		if err != nil {
+			return err
+		} else if err := walkFn(file); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (db *RepositoryDB) WalkFilesByConfigID(ctx context.Context, configID uint64, walkFn WalkFileFunc) error {
+	if walkFn == nil {
+		return fmt.Errorf("no file walk function provided")
+	}
+
+	db.Reference.Add(1)
+	defer db.Reference.Add(-1)
+
+	if err := db.Open(ctx); err != nil {
+		return err
+	}
+
+	rows, err := db.QueryxContext(ctx, "SELECT * FROM files WHERE config_id = ?", configID)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		file := new(RepositoryFile)
+		err := rows.StructScan(file)
+		if err != nil {
+			return err
+		} else if err := walkFn(file); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+type WalkStringFunc func(*string) error
+
+func (db *RepositoryDB) WalkFilesByDistinctParent(ctx context.Context, walkFn WalkStringFunc) error {
+	if walkFn == nil {
+		return fmt.Errorf("no file walk function provided")
+	}
+
+	db.Reference.Add(1)
+	defer db.Reference.Add(-1)
+
+	if err := db.Open(ctx); err != nil {
+		return err
+	}
+
+	rows, err := db.QueryxContext(ctx, "SELECT DISTINCT parent FROM files")
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+
+	for rows.Next() {
+		parent := new(string)
+		err := rows.Scan(parent)
+		if err != nil {
+			return err
+		} else if err := walkFn(parent); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 func (db *RepositoryDB) CountFiles(ctx context.Context) (int, error) {
 	db.Reference.Add(1)
 	defer db.Reference.Add(-1)

internal/plugins/mirror/pkg/mirrordb/schema/repository/000000_create_files_table.sql

@@ -1,10 +1,16 @@
 CREATE TABLE IF NOT EXISTS files (
     tag TEXT PRIMARY KEY,
     name TEXT,
+    reference TEXT,
+    parent TEXT,
     link TEXT,
     modified_time INTEGER,
     mode INTEGER,
-    size INTEGER
+    size INTEGER,
+    config_id INTEGER
 );
 
-CREATE INDEX filename_idx ON files(name);
+CREATE INDEX files_name_idx ON files(name);
+CREATE INDEX files_reference_idx ON files(reference);
+CREATE INDEX files_parent_idx ON files(parent);
+CREATE INDEX files_config_id_idx ON files(config_id);

internal/plugins/mirror/pkg/mirrordb/schema/status/000000_create_properties_table.sql

@@ -2,7 +2,8 @@ CREATE TABLE IF NOT EXISTS properties (
     id INTEGER PRIMARY KEY,
     created BOOLEAN,
     mirror BOOLEAN,
-    mirror_urls BLOB
+    mirror_configs BLOB,
+    web_config BLOB
 );
 
-INSERT INTO properties VALUES(1, false, false, '');
+INSERT INTO properties VALUES(1, false, false, '', '');