diff --git a/config.example.yaml b/config.example.yaml index a99dbebf..9ce998c0 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -117,7 +117,9 @@ storage: # How often to look for expired artifacts; at least "1m". sweep_interval: "10m" # Keep this many of the newest cached versions of each package from - # eviction by age (cargo, golang, npm and pypi). 0 keeps none. + # eviction by age (cargo, golang, npm and pypi). For oci it keeps the + # blobs of this many images per repository, those a tag most recently + # pointed at. 0 keeps none. # keep_latest: 2 # Store fetched artifacts. Set to false to stream every download from diff --git a/docs/architecture.md b/docs/architecture.md index 1d174266..c498cc7f 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -205,10 +205,21 @@ metadata_dates ( last_modified INTEGER NOT NULL, -- Unix seconds; BIGINT on Postgres PRIMARY KEY (ecosystem, proxy_url, name) ) + +image_manifests ( + package_purl TEXT NOT NULL, -- the repository, as its blobs are cached + digest TEXT NOT NULL, + refs TEXT NOT NULL, -- digests it references, one per line + tagged_at INTEGER NOT NULL, -- Unix ms a tag last pointed at it, 0 if none did; BIGINT on Postgres + recorded_at INTEGER NOT NULL, -- Unix ms; BIGINT on Postgres + PRIMARY KEY (package_purl, digest) +) ``` `metadata_dates` holds the `Last-Modified` served with rewritten metadata, such as npm packuments, Composer documents, PyPI simple pages and Helm indexes. A package's date is the second it started serving its current representation, and a new representation is only dated in a later second. Keeping it in the database means replicas sharing one serve the same date, and never give one representation's date to another, so `If-Modified-Since` cannot return 304 for a body the client does not have. +`image_manifests` is filled only with `storage.retention.keep_latest` set and a retention rule for `oci`. It holds which manifests and blobs each served container manifest references, so the retention sweep can keep the blobs of the newest images of a repository. The sweep also drops the rows of images that are no longer among the newest. + On PostgreSQL, `INTEGER PRIMARY KEY` becomes `SERIAL`, `DATETIME` becomes `TIMESTAMP`, `INTEGER DEFAULT 0` booleans become `BOOLEAN DEFAULT FALSE`, and size/count columns use `BIGINT`. The `MigrateSchema()` function handles backward compatibility with older git-pkgs databases by running named migrations that add missing columns and tables. See [migrations.md](migrations.md) for how to add new schema changes. diff --git a/docs/configuration.md b/docs/configuration.md index 1ac89a6b..6850d11a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -71,7 +71,7 @@ storage: | `storage.cache_artifacts` | `PROXY_STORAGE_CACHE_ARTIFACTS` | - | Store fetched artifacts (default: true); `false` streams them from upstream | | `storage.retention.default` | `PROXY_STORAGE_RETENTION_DEFAULT` | - | Evict artifacts not served for this long (e.g. "30d"); see [Retention](#retention) | | `storage.retention.sweep_interval` | `PROXY_STORAGE_RETENTION_SWEEP_INTERVAL` | - | How often expired artifacts are looked for (default: "10m", at least "1m") | -| `storage.retention.keep_latest` | - | - | Keep this many of the newest cached versions of each package from eviction by age (default: 0); see [Retention](#retention) | +| `storage.retention.keep_latest` | - | - | Keep this many of the newest cached versions of each package, or images of each repository, from eviction by age (default: 0); see [Retention](#retention) | `storage.max_size` counts cached artifacts and cached metadata documents (`cache_metadata`). When the total goes over the limit, the least recently served entries of either kind are evicted first; an evicted metadata document is fetched from upstream again on the next request for it. Pushed artifacts are not counted or evicted. An artifact replaced by a refetch stays in storage for at least an hour, or `storage.direct_serve_ttl` if longer, so requests already reading it can finish, and storage use can exceed the limit by what was replaced in that time. @@ -127,7 +127,16 @@ storage: keep_latest: 2 ``` -With this, the files of the two highest cached versions of a package are never evicted by age, and older versions expire by their rule as before. Versions are ordered by the ecosystem's version rules, not by when they were fetched, and a prerelease counts like any other version. Only versions in the cache count, not those upstream, so the kept ones need not be the latest releases. `keep_latest` applies to `cargo`, `golang`, `npm` and `pypi`; the other ecosystems ignore it, since many of them cache files under a digest or a build rather than a version. On startup the proxy logs the ecosystems it reaches (`keep_latest_covered`). `storage.max_size` can still evict a kept version. +With this, the files of the two highest cached versions of a package are never evicted by age, and older versions expire by their rule as before. Versions are ordered by the ecosystem's version rules, not by when they were fetched, and a prerelease counts like any other version. Only versions in the cache count, not those upstream, so the kept ones need not be the latest releases. `keep_latest` applies to `cargo`, `golang`, `npm`, `pypi` and, by image, `oci`; the other ecosystems ignore it, since many of them cache files under a digest or a build rather than a version. On startup the proxy logs the ecosystems it reaches (`keep_latest_covered`). `storage.max_size` can still evict a kept version. + +For `oci` there are no versions to order, so `keep_latest` counts images instead: it keeps the blobs of the images of a repository that a tag most recently pointed at. With `keep_latest: 2` and one tag such as `beta` in use, that is the image `beta` points at now and the one it pointed at before. The count is per repository, not per tag, so with several tags in use set it at least that high. A blob that a kept image shares with an older one stays. + +To know which blobs an image needs, the proxy records what each manifest it serves references, in the `image_manifests` table, while `keep_latest` is set and a retention rule applies to `oci`. This has a few consequences: + +- An image only counts once its tag was pulled or mirrored with `keep_latest` set, and its blobs are known once its platform manifest was pulled too. Until then they expire by their rule as before. +- An image pulled only by digest is not counted and not kept. +- A multi-platform tag counts as one image, also when a client that accepts no image index was served one platform's manifest for it. +- Homebrew bottles and Helm charts pulled over OCI are images here: the newest versions of a bottle or chart are those whose tags were last asked for. Generic retention is opt-in because release assets may disappear upstream and cannot then be downloaded again after eviction. An existing `default` never starts evicting generic downloads. Set `ecosystems.generic: "30d"` to cover all generic artifacts, or configure only selected GitHub repositories: diff --git a/internal/database/database_test.go b/internal/database/database_test.go index 500ad0d8..8a2e0574 100644 --- a/internal/database/database_test.go +++ b/internal/database/database_test.go @@ -613,7 +613,7 @@ func createTestPostgresDB(t *testing.T) *DB { // Drop and recreate every table CreateSchema creates for clean test state; // leftover migration records make the next CreateSchema fail on the // migrations primary key. - tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "job_runs", "fetch_leases", "upstream_holds", "metadata_snapshots", "projects", "project_components", "migrations", "schema_info"} + tables := []string{"artifacts", "pending_deletes", "versions", "packages", "vulnerabilities", "metadata_cache", "metadata_dates", "image_manifests", "job_runs", "fetch_leases", "upstream_holds", "metadata_snapshots", "projects", "project_components", "migrations", "schema_info"} for _, table := range tables { _, _ = db.Exec("DROP TABLE IF EXISTS " + table + " CASCADE") } diff --git a/internal/database/image_manifests.go b/internal/database/image_manifests.go new file mode 100644 index 00000000..4a26758d --- /dev/null +++ b/internal/database/image_manifests.go @@ -0,0 +1,224 @@ +package database + +import ( + "cmp" + "database/sql" + "errors" + "fmt" + "maps" + "slices" + "strings" + "time" + + "github.com/jmoiron/sqlx" +) + +const ( + // imageWalkDepth is how many levels of references NewestImageDigests + // follows below an image: nested indexes, a platform manifest, its blobs. + imageWalkDepth = 4 + // imageWalkBatch is how many digests one query looks up. + imageWalkBatch = 500 +) + +// ImageManifest is what one container image manifest of a repository +// references. +type ImageManifest struct { + // PackagePURL is the repository as its blobs are cached. + PackagePURL string + Digest string + // Refs are the digests of the manifests or blobs it references. + Refs []string + // TaggedAt is when a tag was last found to point at the manifest, or + // zero for one only asked for by digest. + TaggedAt time.Time +} + +type imageManifestRow struct { + Digest string `db:"digest"` + Refs string `db:"refs"` + TaggedAt int64 `db:"tagged_at"` +} + +func (r imageManifestRow) refs() []string { + return strings.Fields(r.Refs) +} + +// RecordImageManifest stores a manifest's references. For one recorded +// before it only moves the tag time forward. +func (db *DB) RecordImageManifest(m ImageManifest) error { + tagged := int64(0) + if !m.TaggedAt.IsZero() { + tagged = m.TaggedAt.UnixMilli() + } + _, err := db.Exec(db.Rebind(` + INSERT INTO image_manifests (package_purl, digest, refs, tagged_at, recorded_at) VALUES (?, ?, ?, ?, ?) + ON CONFLICT (package_purl, digest) DO UPDATE SET tagged_at = excluded.tagged_at + WHERE image_manifests.tagged_at < excluded.tagged_at + `), m.PackagePURL, m.Digest, strings.Join(m.Refs, "\n"), tagged, time.Now().UnixMilli()) + return err +} + +// ImageManifestTaggedAt returns when a tag was last found to point at a +// manifest, zero if none was, and whether the manifest is recorded at all. +func (db *DB) ImageManifestTaggedAt(packagePURL, digest string) (time.Time, bool, error) { + var tagged int64 + err := db.Get(&tagged, db.Rebind(` + SELECT tagged_at FROM image_manifests WHERE package_purl = ? AND digest = ? + `), packagePURL, digest) + if errors.Is(err, sql.ErrNoRows) { + return time.Time{}, false, nil + } + if err != nil || tagged == 0 { + return time.Time{}, err == nil, err + } + return time.UnixMilli(tagged), true, nil +} + +// NewestImageDigests returns the digests of the n images of a repository +// that a tag was most recently found to point at, and of every manifest and +// blob they reference. +func (db *DB) NewestImageDigests(packagePURL string, n int) (map[string]struct{}, error) { + digests, _, err := db.walkNewestImages(packagePURL, n) + return digests, err +} + +// ImageManifestRepositories returns the repositories that have recorded +// manifests. +func (db *DB) ImageManifestRepositories() ([]string, error) { + var repositories []string + err := db.Select(&repositories, `SELECT DISTINCT package_purl FROM image_manifests`) + return repositories, err +} + +// PruneImageManifests removes the manifests of a repository that its n +// newest images do not reference and that were recorded before the given +// time. The time leaves a manifest recorded shortly before the one that +// references it in place. +func (db *DB) PruneImageManifests(packagePURL string, n int, before time.Time) error { + _, manifests, err := db.walkNewestImages(packagePURL, n) + if err != nil { + return err + } + query := `DELETE FROM image_manifests WHERE package_purl = ? AND recorded_at < ?` + args := []any{packagePURL, before.UnixMilli()} + if len(manifests) > 0 { + query, args, err = sqlx.In(query+` AND digest NOT IN (?)`, packagePURL, before.UnixMilli(), manifests) + if err != nil { + return fmt.Errorf("building prune query: %w", err) + } + } + _, err = db.Exec(db.Rebind(query), args...) + return err +} + +// walkNewestImages returns every digest the n newest images of a repository +// reference, themselves included, and those among them that are recorded +// manifests. +func (db *DB) walkNewestImages(packagePURL string, n int) (map[string]struct{}, []string, error) { + digests := map[string]struct{}{} + if n <= 0 { + return digests, nil, nil + } + rows, err := db.newestImages(packagePURL, n) + if err != nil { + return nil, nil, err + } + + var manifests []string + for _, row := range rows { + digests[row.Digest] = struct{}{} + } + for depth := 0; len(rows) > 0; depth++ { + var next []string + for _, row := range rows { + manifests = append(manifests, row.Digest) + for _, ref := range row.refs() { + if _, seen := digests[ref]; seen { + continue + } + digests[ref] = struct{}{} + next = append(next, ref) + } + } + if depth == imageWalkDepth { + break + } + if rows, err = db.imageManifestRows(packagePURL, next); err != nil { + return nil, nil, err + } + } + return digests, manifests, nil +} + +// newestImages returns the n tagged manifests of a repository whose tag was +// seen last. A registry serves a tag as an index to one client and as one +// platform's manifest to a client that accepts no index, so a tagged +// manifest that another tagged one references is that image seen another +// way: it is no image of its own and lends the other its tag time. +func (db *DB) newestImages(packagePURL string, n int) ([]imageManifestRow, error) { + var rows []imageManifestRow + err := db.Select(&rows, db.Rebind(` + SELECT digest, refs, tagged_at FROM image_manifests + WHERE package_purl = ? AND tagged_at > 0 + `), packagePURL) + if err != nil { + return nil, err + } + + tagged := make(map[string]int64, len(rows)) + for _, row := range rows { + tagged[row.Digest] = row.TaggedAt + } + seen := maps.Clone(tagged) + referenced := map[string]struct{}{} + for _, row := range rows { + for _, ref := range row.refs() { + at, ok := tagged[ref] + if !ok || ref == row.Digest { + continue + } + referenced[ref] = struct{}{} + seen[row.Digest] = max(seen[row.Digest], at) + } + } + + images := rows[:0:0] + for _, row := range rows { + if _, ok := referenced[row.Digest]; !ok { + images = append(images, row) + } + } + if len(images) == 0 { + // Manifests that only reference each other; no registry serves these. + images = rows + } + slices.SortFunc(images, func(a, b imageManifestRow) int { + return cmp.Or( + cmp.Compare(seen[b.Digest], seen[a.Digest]), + cmp.Compare(b.TaggedAt, a.TaggedAt), + cmp.Compare(a.Digest, b.Digest), + ) + }) + return images[:min(n, len(images))], nil +} + +// imageManifestRows returns the recorded manifests among digests. +func (db *DB) imageManifestRows(packagePURL string, digests []string) ([]imageManifestRow, error) { + var rows []imageManifestRow + for start := 0; start < len(digests); start += imageWalkBatch { + batch := digests[start:min(start+imageWalkBatch, len(digests))] + query, args, err := sqlx.In(` + SELECT digest, refs FROM image_manifests WHERE package_purl = ? AND digest IN (?) + `, packagePURL, batch) + if err != nil { + return nil, fmt.Errorf("building image manifest query: %w", err) + } + var found []imageManifestRow + if err := db.Select(&found, db.Rebind(query), args...); err != nil { + return nil, err + } + rows = append(rows, found...) + } + return rows, nil +} diff --git a/internal/database/image_manifests_test.go b/internal/database/image_manifests_test.go new file mode 100644 index 00000000..b6074cb7 --- /dev/null +++ b/internal/database/image_manifests_test.go @@ -0,0 +1,228 @@ +package database + +import ( + "slices" + "testing" + "time" +) + +const testImageRepo = "pkg:oci/library%2Fnginx" + +func recordImage(t *testing.T, db *DB, repo, digest string, taggedAt time.Time, refs ...string) { + t.Helper() + err := db.RecordImageManifest(ImageManifest{PackagePURL: repo, Digest: digest, Refs: refs, TaggedAt: taggedAt}) + if err != nil { + t.Fatalf("RecordImageManifest(%s): %v", digest, err) + } +} + +func newestImageDigests(t *testing.T, db *DB, repo string, n int) []string { + t.Helper() + set, err := db.NewestImageDigests(repo, n) + if err != nil { + t.Fatalf("NewestImageDigests: %v", err) + } + digests := make([]string, 0, len(set)) + for d := range set { + digests = append(digests, d) + } + slices.Sort(digests) + return digests +} + +// The newest images are the manifests a tag was last found to point at. What +// they reference is followed through the manifests recorded for the same +// repository, so an index leads to its platform manifests and their blobs. +func TestNewestImageDigestsFollowsReferences(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-old", now.Add(-2*time.Hour), "sha256:arm-old", "sha256:amd-old") + recordImage(t, db, testImageRepo, "sha256:arm-old", time.Time{}, "sha256:config-old", "sha256:layer-shared", "sha256:layer-old") + recordImage(t, db, testImageRepo, "sha256:index-new", now.Add(-time.Hour), "sha256:arm-new", "sha256:amd-new") + recordImage(t, db, testImageRepo, "sha256:arm-new", time.Time{}, "sha256:config-new", "sha256:layer-shared", "sha256:layer-new") + // Another repository's image is newer still and must not count here. + recordImage(t, db, "pkg:oci/library%2Fredis", "sha256:redis", now, "sha256:layer-redis") + + want := []string{"sha256:amd-new", "sha256:arm-new", "sha256:config-new", "sha256:index-new", "sha256:layer-new", "sha256:layer-shared"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 11 { + t.Errorf("newest 2 = %v, want all 11 digests of both images", got) + } + if got := newestImageDigests(t, db, testImageRepo, 0); len(got) != 0 { + t.Errorf("newest 0 = %v, want none", got) + } + }) +} + +// A registry may serve a tag as an index to one client and as one platform's +// manifest to a client that accepts no index. Both are then tagged, and +// count as one image, with the time the tag was last seen through either. +func TestNewestImageDigestsCountsATagsRepresentationsOnce(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-other", now.Add(-2*time.Hour), "sha256:arm-other") + recordImage(t, db, testImageRepo, "sha256:index", now.Add(-3*time.Hour), "sha256:arm", "sha256:amd") + recordImage(t, db, testImageRepo, "sha256:arm", time.Time{}, "sha256:layer-arm") + recordImage(t, db, testImageRepo, "sha256:amd", now.Add(-time.Hour), "sha256:layer-amd") + + want := []string{"sha256:amd", "sha256:arm", "sha256:index", "sha256:layer-amd", "sha256:layer-arm"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want the whole index %v", got, want) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 7 { + t.Errorf("newest 2 = %v, want all 7 digests of both images", got) + } + }) +} + +// A platform manifest two indexes share lends both its tag time, and the +// index tagged later is then the newer image. +func TestNewestImageDigestsOrdersImagesSharingAManifest(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + // The digests are named so that a tie would pick the old index. + recordImage(t, db, testImageRepo, "sha256:a-index-old", now.Add(-3*time.Hour), "sha256:arm", "sha256:amd-old") + recordImage(t, db, testImageRepo, "sha256:b-index-new", now.Add(-2*time.Hour), "sha256:arm", "sha256:amd-new") + recordImage(t, db, testImageRepo, "sha256:arm", now.Add(-time.Hour), "sha256:layer-arm") + + want := []string{"sha256:amd-new", "sha256:arm", "sha256:b-index-new", "sha256:layer-arm"} + if got := newestImageDigests(t, db, testImageRepo, 1); !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + }) +} + +// A manifest only asked for by digest is not an image of its own. +func TestNewestImageDigestsIgnoresUntaggedManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + recordImage(t, db, testImageRepo, "sha256:pinned", time.Time{}, "sha256:layer-pinned") + if got := newestImageDigests(t, db, testImageRepo, 5); len(got) != 0 { + t.Errorf("newest = %v, want none", got) + } + }) +} + +// Recording a manifest again moves its tag time forward only, and a request +// by digest does not make it lose the time a tag gave it. +func TestRecordImageManifestKeepsLatestTagTime(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now().Truncate(time.Millisecond) + recordImage(t, db, testImageRepo, "sha256:a", now.Add(-time.Hour), "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:b", now.Add(-30*time.Minute), "sha256:layer-b") + + if at, ok, err := db.ImageManifestTaggedAt(testImageRepo, "sha256:missing"); err != nil || ok || !at.IsZero() { + t.Fatalf("ImageManifestTaggedAt(missing) = %v, %v, %v; want zero, false, nil", at, ok, err) + } + + recordImage(t, db, testImageRepo, "sha256:a", time.Time{}, "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:a", now.Add(-2*time.Hour), "sha256:layer-a") + at, ok, err := db.ImageManifestTaggedAt(testImageRepo, "sha256:a") + if err != nil || !ok || !at.Equal(now.Add(-time.Hour)) { + t.Fatalf("ImageManifestTaggedAt(a) = %v, %v, %v; want %v, true, nil", at, ok, err, now.Add(-time.Hour)) + } + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:b", "sha256:layer-b"}; !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + + // A tag found on a again makes it the newest image. + recordImage(t, db, testImageRepo, "sha256:a", now, "sha256:layer-a") + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:a", "sha256:layer-a"}; !slices.Equal(got, want) { + t.Errorf("newest 1 after retag = %v, want %v", got, want) + } + }) +} + +// A manifest that references itself, or a chain longer than any real index, +// ends the walk instead of looping. +func TestNewestImageDigestsStopsOnCycles(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:a", now, "sha256:b") + recordImage(t, db, testImageRepo, "sha256:b", time.Time{}, "sha256:a", "sha256:layer") + if got, want := newestImageDigests(t, db, testImageRepo, 1), []string{"sha256:a", "sha256:b", "sha256:layer"}; !slices.Equal(got, want) { + t.Errorf("newest 1 = %v, want %v", got, want) + } + }) +} + +// Pruning drops what the newest images no longer reference, once it is older +// than the grace time, so a platform manifest recorded just after its index +// is not lost. +func TestPruneImageManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + now := time.Now() + recordImage(t, db, testImageRepo, "sha256:index-old", now.Add(-2*time.Hour), "sha256:arm-old") + recordImage(t, db, testImageRepo, "sha256:arm-old", time.Time{}, "sha256:layer-old") + recordImage(t, db, testImageRepo, "sha256:index-new", now.Add(-time.Hour), "sha256:arm-new") + recordImage(t, db, testImageRepo, "sha256:arm-new", time.Time{}, "sha256:layer-new") + recordImage(t, db, testImageRepo, "sha256:pinned", time.Time{}, "sha256:layer-pinned") + recordImage(t, db, "pkg:oci/library%2Fredis", "sha256:redis-old", now.Add(-3*time.Hour), "sha256:layer-redis") + + // Everything was recorded just now, so nothing is old enough yet. + if err := db.PruneImageManifests(testImageRepo, 1, now.Add(-time.Minute)); err != nil { + t.Fatalf("PruneImageManifests: %v", err) + } + if got := newestImageDigests(t, db, testImageRepo, 2); len(got) != 6 { + t.Fatalf("after early prune, newest 2 = %v, want both images", got) + } + + if err := db.PruneImageManifests(testImageRepo, 1, now.Add(time.Minute)); err != nil { + t.Fatalf("PruneImageManifests: %v", err) + } + want := []string{"sha256:arm-new", "sha256:index-new", "sha256:layer-new"} + if got := newestImageDigests(t, db, testImageRepo, 2); !slices.Equal(got, want) { + t.Errorf("after prune, newest 2 = %v, want %v", got, want) + } + for _, digest := range []string{"sha256:index-old", "sha256:arm-old", "sha256:pinned"} { + if _, ok, err := db.ImageManifestTaggedAt(testImageRepo, digest); err != nil || ok { + t.Errorf("%s still recorded after prune (ok=%v, err=%v)", digest, ok, err) + } + } + if got := newestImageDigests(t, db, "pkg:oci/library%2Fredis", 1); len(got) != 2 { + t.Errorf("other repository after prune = %v, want its image kept", got) + } + }) +} + +// A database from before the table existed gets it from the migration. +func TestMigrateAddsImageManifests(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + if _, err := db.Exec("DROP TABLE image_manifests"); err != nil { + t.Fatal(err) + } + if _, err := db.Exec("DELETE FROM migrations WHERE name = '024_add_image_manifests'"); err != nil { + t.Fatal(err) + } + if err := db.MigrateSchema(); err != nil { + t.Fatalf("MigrateSchema: %v", err) + } + recordImage(t, db, testImageRepo, "sha256:a", time.Now(), "sha256:layer-a") + if got := newestImageDigests(t, db, testImageRepo, 1); len(got) != 2 { + t.Errorf("newest 1 after migration = %v, want the image", got) + } + }) +} + +// The repositories with records are listed once each, for pruning. +func TestImageManifestRepositories(t *testing.T) { + runWithBothDatabases(t, func(t *testing.T, db *DB) { + if got, err := db.ImageManifestRepositories(); err != nil || len(got) != 0 { + t.Fatalf("ImageManifestRepositories() = %v, %v; want none", got, err) + } + const redis = "pkg:oci/library%2Fredis" + recordImage(t, db, testImageRepo, "sha256:a", time.Now(), "sha256:layer-a") + recordImage(t, db, testImageRepo, "sha256:b", time.Time{}, "sha256:layer-b") + recordImage(t, db, redis, "sha256:c", time.Time{}, "sha256:layer-c") + + got, err := db.ImageManifestRepositories() + if err != nil { + t.Fatalf("ImageManifestRepositories: %v", err) + } + slices.Sort(got) + if want := []string{testImageRepo, redis}; !slices.Equal(got, want) { + t.Errorf("ImageManifestRepositories() = %v, want %v", got, want) + } + }) +} diff --git a/internal/database/schema.go b/internal/database/schema.go index bca37bd4..1c298fab 100644 --- a/internal/database/schema.go +++ b/internal/database/schema.go @@ -151,6 +151,15 @@ CREATE TABLE IF NOT EXISTS metadata_dates ( PRIMARY KEY (ecosystem, proxy_url, name) ); +CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at INTEGER NOT NULL DEFAULT 0, + recorded_at INTEGER NOT NULL, + PRIMARY KEY (package_purl, digest) +); + CREATE TABLE IF NOT EXISTS upstream_holds ( host TEXT NOT NULL PRIMARY KEY, until_ms INTEGER NOT NULL @@ -330,6 +339,15 @@ CREATE TABLE IF NOT EXISTS metadata_dates ( PRIMARY KEY (ecosystem, proxy_url, name) ); +CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at BIGINT NOT NULL DEFAULT 0, + recorded_at BIGINT NOT NULL, + PRIMARY KEY (package_purl, digest) +); + CREATE TABLE IF NOT EXISTS upstream_holds ( host TEXT NOT NULL PRIMARY KEY, until_ms BIGINT NOT NULL @@ -546,6 +564,7 @@ var migrations = []migration{ {"021_add_projects", migrateAddProjects}, {"022_add_artifact_mirrored_at", migrateAddArtifactMirroredAt}, {"023_add_metadata_upstream_url", migrateAddMetadataUpstreamURL}, + {"024_add_image_manifests", migrateAddImageManifests}, } // isTableNotFound returns true if the error indicates a missing table. @@ -1030,6 +1049,27 @@ func migrateAddMetadataDates(db *DB) error { return nil } +// migrateAddImageManifests creates the table of what container image +// manifests reference, which retention's keep_latest reads. +func migrateAddImageManifests(db *DB) error { + millis := "INTEGER" + if db.dialect == DialectPostgres { + millis = "BIGINT" + } + query := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS image_manifests ( + package_purl TEXT NOT NULL, + digest TEXT NOT NULL, + refs TEXT NOT NULL, + tagged_at %s NOT NULL DEFAULT 0, + recorded_at %s NOT NULL, + PRIMARY KEY (package_purl, digest) + )`, millis, millis) + if _, err := db.Exec(query); err != nil { + return fmt.Errorf("creating image_manifests table: %w", err) + } + return nil +} + // EnsureMetadataCacheTable creates the metadata_cache table if it doesn't exist. func (db *DB) EnsureMetadataCacheTable() error { has, err := db.HasTable("metadata_cache") diff --git a/internal/handler/container.go b/internal/handler/container.go index 40f5d0f2..c0453595 100644 --- a/internal/handler/container.go +++ b/internal/handler/container.go @@ -70,7 +70,7 @@ func isDockerHubKey(key string) bool { // Package rules write the repository name as a path, pkg:oci/library/nginx, // and are mapped to that form. func init() { - retention.Default.Register(retention.Spec{Key: "oci", CanonicalPackage: ociRetentionKey}) + retention.Default.Register(retention.Spec{Key: "oci", CanonicalPackage: ociRetentionKey, KeepLatestImages: true}) } // ociRetentionKey maps a package rule to the stored package PURL. It refuses diff --git a/internal/handler/container_image_refs.go b/internal/handler/container_image_refs.go new file mode 100644 index 00000000..fbe523cd --- /dev/null +++ b/internal/handler/container_image_refs.go @@ -0,0 +1,97 @@ +package handler + +import ( + "encoding/json" + "regexp" + "slices" + "time" + + "github.com/git-pkgs/proxy/internal/database" +) + +// maxImageManifestRefs is how many references of one manifest are recorded. +// A manifest with more is not recorded. +const maxImageManifestRefs = 4096 + +// imageSubjectTagPattern matches the tags that hold what refers to an image +// rather than an image of their own: the referrers tag schema, and the +// sha256-.sig, .att and .sbom tags of cosign. +var imageSubjectTagPattern = regexp.MustCompile(`^sha256-[a-f0-9]{64}(\.[a-z0-9]+)?$`) + +// recordImageManifest notes which manifests and blobs a served manifest +// references, and for a tag that it pointed at the manifest, so retention +// can keep the blobs of the newest images of a repository. It does nothing +// unless storage.retention.keep_latest covers container images. The +// retention sweep drops the records of images that are no longer among the +// newest. +func (h *ContainerHandler) recordImageManifest(cacheName, reference string, manifest *cachedContainerManifest, stale bool) { + if h.proxy.KeepLatestImages <= 0 || h.proxy.DB == nil { + return + } + // A manifest asked for by digest is recorded under that digest, the one + // the manifest referencing it names. + digest, tagged := reference, time.Time{} + if !manifestDigestReferencePattern.MatchString(reference) { + digest = manifest.contentDigest + if !imageSubjectTagPattern.MatchString(reference) { + // The time upstream last confirmed the tag, which a stale copy + // served during an outage does not move. + if tagged = manifest.fetchedAt; tagged.IsZero() && !stale { + tagged = time.Now() + } + } + } + if !mirrorDigestPattern.MatchString(digest) { + return + } + + packagePURL := canonicalPackagePURL("oci", cacheName) + known, recorded, err := h.proxy.DB.ImageManifestTaggedAt(packagePURL, digest) + if err != nil { + h.proxy.Logger.Warn("failed to read image manifest record", "package_purl", packagePURL, "digest", digest, "error", err) + return + } + // Tag times are stored in milliseconds. + if recorded && tagged.UnixMilli() <= known.UnixMilli() { + return + } + var refs []string + if !recorded { + // The answer to a HEAD for a manifest that is not cached has no + // body to read references from. + if refs = imageManifestRefs(manifest.body); len(refs) == 0 { + return + } + } + record := database.ImageManifest{PackagePURL: packagePURL, Digest: digest, Refs: refs, TaggedAt: tagged} + if err := h.proxy.DB.RecordImageManifest(record); err != nil { + h.proxy.Logger.Warn("failed to record image manifest", "package_purl", packagePURL, "digest", digest, "error", err) + } +} + +// imageManifestRefs returns the digests of the manifests an index lists, or +// of the config and layers of an image manifest. It returns nil for a +// document that names none, or more than maxImageManifestRefs. +func imageManifestRefs(body []byte) []string { + var manifest ociManifest + if len(body) == 0 || json.Unmarshal(body, &manifest) != nil { + return nil + } + descriptors := slices.Concat(manifest.Manifests, manifest.Layers) + if manifest.Config != nil { + descriptors = append(descriptors, *manifest.Config) + } + seen := make(map[string]struct{}, len(descriptors)) + var refs []string + for _, descriptor := range descriptors { + if _, dup := seen[descriptor.Digest]; dup || !mirrorDigestPattern.MatchString(descriptor.Digest) { + continue + } + seen[descriptor.Digest] = struct{}{} + refs = append(refs, descriptor.Digest) + } + if len(refs) > maxImageManifestRefs { + return nil + } + return refs +} diff --git a/internal/handler/container_image_refs_test.go b/internal/handler/container_image_refs_test.go new file mode 100644 index 00000000..9b2dac6b --- /dev/null +++ b/internal/handler/container_image_refs_test.go @@ -0,0 +1,192 @@ +package handler + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +const imageRefsRepo = "team/app" + +func imageRefsHandler(t *testing.T, keepLatest int) (*ContainerHandler, *fakeImageRegistry, fakeImage) { + t.Helper() + registry, image := newFakeImageRegistry(t) + proxy := newMirrorTestProxy(t, registry) + proxy.KeepLatestImages = keepLatest + return NewContainerHandlerWithRegistry(proxy, "http://proxy.example", registry.server.URL), registry, image +} + +func pullImageManifest(t *testing.T, h *ContainerHandler, reference string) { + t.Helper() + if response := pull(t, h, "/"+imageRefsRepo+"/manifests/"+reference, containerdManifestAccept); response.Code != http.StatusOK { + t.Fatalf("manifest %s: status %d: %s", reference, response.Code, response.Body.String()) + } +} + +// newestImageDigests returns what the newest image of the repository needs. +func newestImageDigests(t *testing.T, h *ContainerHandler) map[string]struct{} { + t.Helper() + digests, err := h.proxy.DB.NewestImageDigests(canonicalPackagePURL("oci", imageRefsRepo), 1) + if err != nil { + t.Fatalf("NewestImageDigests: %v", err) + } + return digests +} + +func imageTaggedAt(t *testing.T, h *ContainerHandler, digest string) (time.Time, bool) { + t.Helper() + at, ok, err := h.proxy.DB.ImageManifestTaggedAt(canonicalPackagePURL("oci", imageRefsRepo), digest) + if err != nil { + t.Fatalf("ImageManifestTaggedAt: %v", err) + } + return at, ok +} + +// moveTag points a tag of the fake registry at another manifest. +func (r *fakeImageRegistry) moveTag(tag, digest string) { + r.mu.Lock() + defer r.mu.Unlock() + r.manifests["/v2/"+imageRefsRepo+"/manifests/"+tag] = r.manifests["/v2/"+imageRefsRepo+"/manifests/"+digest] +} + +// A pull by tag makes the manifest an image of the repository, and the +// manifests pulled by digest after it say which blobs it needs. +func TestContainerPullRecordsWhatAnImageReferences(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + + pullImageManifest(t, h, "multi") + if at, ok := imageTaggedAt(t, h, image.indexDigest); !ok || at.IsZero() { + t.Fatalf("index after a pull by tag: recorded=%v tagged=%v, want both", ok, at) + } + pullImageManifest(t, h, image.arm64Digest) + if at, ok := imageTaggedAt(t, h, image.arm64Digest); !ok || !at.IsZero() { + t.Fatalf("platform manifest after a pull by digest: recorded=%v tagged=%v, want recorded and untagged", ok, at) + } + + digests := newestImageDigests(t, h) + for name, digest := range map[string]string{ + "index": image.indexDigest, "arm64 manifest": image.arm64Digest, + "amd64 manifest": image.amd64Digest, "arm64 layer": image.arm64Layer, + } { + if _, ok := digests[digest]; !ok { + t.Errorf("%s %s is not among the newest image's digests", name, digest) + } + } + // The amd64 manifest was never pulled, so its blobs are unknown. + if _, ok := digests[image.amd64Layer]; ok { + t.Errorf("amd64 layer is among the digests though its manifest was not pulled") + } +} + +// Without keep_latest nothing is recorded. +func TestContainerPullRecordsNothingWithoutKeepLatest(t *testing.T) { + h, _, image := imageRefsHandler(t, 0) + pullImageManifest(t, h, "multi") + pullImageManifest(t, h, image.arm64Digest) + if _, ok := imageTaggedAt(t, h, image.indexDigest); ok { + t.Errorf("index recorded though keep_latest is off") + } + if _, ok := imageTaggedAt(t, h, image.arm64Digest); ok { + t.Errorf("platform manifest recorded though keep_latest is off") + } +} + +// Referrers and cosign tags hold what refers to an image. They are recorded +// like a pull by digest, so they don't count as images of their own. +func TestContainerPullOfReferrersTagIsNotAnImage(t *testing.T) { + h, registry, image := imageRefsHandler(t, 1) + subject := "sha256-" + image.indexDigest[len("sha256:"):] + for _, tag := range []string{subject, subject + ".sig", subject + ".att"} { + registry.moveTag(tag, image.singleDigest) + pullImageManifest(t, h, tag) + } + if at, ok := imageTaggedAt(t, h, image.singleDigest); !ok || !at.IsZero() { + t.Errorf("manifest behind referrers tags: recorded=%v tagged=%v, want recorded and untagged", ok, at) + } +} + +// A tag that is asked for again moves the image's tag time forward, also +// when the client only sends a HEAD, as containerd does for an image it has. +func TestContainerPullAgainMovesTagTimeForward(t *testing.T) { + for _, method := range []string{http.MethodGet, http.MethodHead} { + t.Run(method, func(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + pullImageManifest(t, h, "single") + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ?"), 1000); err != nil { + t.Fatal(err) + } + + request := httptest.NewRequest(method, "/"+imageRefsRepo+"/manifests/single", nil) + request.Header.Set("Accept", containerdManifestAccept) + response := httptest.NewRecorder() + h.Routes().ServeHTTP(response, request) + if response.Code != http.StatusOK { + t.Fatalf("status %d", response.Code) + } + if at, _ := imageTaggedAt(t, h, image.singleDigest); time.Since(at) > time.Minute { + t.Errorf("tag time after the second request = %v, want now", at) + } + }) + } +} + +// When a tag moves, the image it points at now is the newest, and the one +// it pointed at before no longer is. +func TestContainerMovedTagMakesTheNewImageNewest(t *testing.T) { + h, registry, image := imageRefsHandler(t, 1) + registry.moveTag("stable", image.singleDigest) + pullImageManifest(t, h, "stable") + if _, ok := newestImageDigests(t, h)[image.singleLayer]; !ok { + t.Fatalf("layer of the first image is not kept") + } + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ?"), 1000); err != nil { + t.Fatal(err) + } + + registry.moveTag("stable", image.indexDigest) + pullImageManifest(t, h, "stable") + pullImageManifest(t, h, image.arm64Digest) + + digests := newestImageDigests(t, h) + if _, ok := digests[image.arm64Layer]; !ok { + t.Errorf("layer of the image the tag moved to is not kept") + } + if _, ok := digests[image.singleLayer]; ok { + t.Errorf("layer of the image the tag moved away from is still kept") + } +} + +// A scheduled mirror run resolves tags itself, and records them like a pull. +func TestMirrorImageRecordsWhatAnImageReferences(t *testing.T) { + h, _, image := imageRefsHandler(t, 1) + platform, err := ParsePlatform("linux/arm64") + if err != nil { + t.Fatal(err) + } + ref := ImageRef{Name: imageRefsRepo, References: []string{"multi"}} + if _, err := h.MirrorImage(context.Background(), ref, []Platform{platform}); err != nil { + t.Fatalf("MirrorImage: %v", err) + } + if at, ok := imageTaggedAt(t, h, image.indexDigest); !ok || at.IsZero() { + t.Fatalf("index after a mirror run: recorded=%v tagged=%v, want both", ok, at) + } + if _, ok := newestImageDigests(t, h)[image.arm64Layer]; !ok { + t.Errorf("arm64 layer is not among the mirrored image's digests") + } + + // A later run that finds the tag unchanged still moves its time forward. + db := h.proxy.DB + if _, err := db.Exec(db.Rebind("UPDATE image_manifests SET tagged_at = ? WHERE tagged_at > 0"), 1000); err != nil { + t.Fatal(err) + } + if _, err := h.MirrorImage(context.Background(), ref, []Platform{platform}); err != nil { + t.Fatalf("second MirrorImage: %v", err) + } + if at, _ := imageTaggedAt(t, h, image.indexDigest); time.Since(at) > time.Minute { + t.Errorf("tag time after the second run = %v, want now", at) + } +} diff --git a/internal/handler/container_manifest.go b/internal/handler/container_manifest.go index 6d129d0a..fa857f37 100644 --- a/internal/handler/container_manifest.go +++ b/internal/handler/container_manifest.go @@ -63,14 +63,14 @@ func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, } cached, _ := h.loadContainerManifestForAccept(r.Context(), registryURL, name, reference, req.accept, req.cacheAccept) if cached != nil && (req.immutable() || h.containerManifestFresh(cached)) { - h.writeManifest(w, r, cacheName, cached, false) + h.writeManifest(w, r, cacheName, reference, cached, false) return } outcome := h.coalescedManifest(r.Context(), req) switch { case outcome.manifest != nil: - h.writeManifest(w, r, cacheName, outcome.manifest, false) + h.writeManifest(w, r, cacheName, reference, outcome.manifest, false) case outcome.relay != nil: outcome.relay.write(h.proxy, w, r) case outcome.digestInvalid: @@ -83,7 +83,7 @@ func (h *ContainerHandler) serveManifest(w http.ResponseWriter, r *http.Request, func (h *ContainerHandler) serveStaleManifestOrError(w http.ResponseWriter, r *http.Request, cacheName, reference string, cached *cachedContainerManifest, err error) { if cached != nil { h.proxy.Logger.Warn("upstream manifest fetch failed, serving stale cache", "error", err) - h.writeManifest(w, r, cacheName, cached, true) + h.writeManifest(w, r, cacheName, reference, cached, true) return } if referrersTagPattern.MatchString(reference) { @@ -109,11 +109,13 @@ func (h *ContainerHandler) manifestDenied(w http.ResponseWriter, cacheName, refe return true } -// writeManifest writes manifest unless the denylist names its digest. -func (h *ContainerHandler) writeManifest(w http.ResponseWriter, r *http.Request, cacheName string, manifest *cachedContainerManifest, stale bool) { +// writeManifest writes the manifest that reference resolved to, unless the +// denylist names its digest. +func (h *ContainerHandler) writeManifest(w http.ResponseWriter, r *http.Request, cacheName, reference string, manifest *cachedContainerManifest, stale bool) { if h.manifestDenied(w, cacheName, manifest.contentDigest) { return } + h.recordImageManifest(cacheName, reference, manifest, stale) writeContainerManifest(w, r, manifest, stale) } diff --git a/internal/handler/container_mirror.go b/internal/handler/container_mirror.go index f19ac736..19a52fc7 100644 --- a/internal/handler/container_mirror.go +++ b/internal/handler/container_mirror.go @@ -255,6 +255,8 @@ func (m *imageMirror) mirrorTag(ctx context.Context, tag string) (string, error) return digest, nil } m.h.storeContainerManifestForAccept(ctx, m.registryURL, m.upstreamName, tag, accept, cacheAccept, manifest) + // The walk recorded the manifest as one asked for by digest. + m.h.recordImageManifest(m.cacheName, tag, manifest, false) return digest, nil } diff --git a/internal/handler/handler.go b/internal/handler/handler.go index 9786eeac..f245aaff 100644 --- a/internal/handler/handler.go +++ b/internal/handler/handler.go @@ -244,6 +244,12 @@ type Proxy struct { DirectServeTTL time.Duration HTTPClient *http.Client + // KeepLatestImages is storage.retention.keep_latest where a retention + // rule applies to container images. Above zero, the container handler + // records what each served manifest references, so retention can keep + // the blobs of that many images per repository. + KeepLatestImages int + // RateLimits is the gate under the upstream clients. Artifact fetches // check it before calling Fetcher, so held hosts never reach its breaker. RateLimits UpstreamHolds diff --git a/internal/handler/profile.go b/internal/handler/profile.go index 8e6ad611..e9cde1af 100644 --- a/internal/handler/profile.go +++ b/internal/handler/profile.go @@ -14,7 +14,7 @@ func (p *Proxy) WithCooldown(policy CooldownPolicy, shortest time.Duration) *Pro GradleReadOnly: p.GradleReadOnly, HostedEnabled: p.HostedEnabled, GradleMaxUploadSize: p.GradleMaxUploadSize, NPMFullMetadata: p.NPMFullMetadata, DirectServe: p.DirectServe, DirectServeTTL: p.DirectServeTTL, - HTTPClient: p.HTTPClient, + HTTPClient: p.HTTPClient, KeepLatestImages: p.KeepLatestImages, RateLimits: p.RateLimits, HostedNPMScopes: p.HostedNPMScopes, HostedComposerVendors: p.HostedComposerVendors, StreamArtifacts: p.StreamArtifacts, Scanners: p.Scanners, ScanSigningKey: p.ScanSigningKey, ScanFetchBaseURL: p.ScanFetchBaseURL, diff --git a/internal/handler/profile_test.go b/internal/handler/profile_test.go index a85d6ca3..c2a768f5 100644 --- a/internal/handler/profile_test.go +++ b/internal/handler/profile_test.go @@ -34,6 +34,7 @@ func TestWithCooldownCopiesProxyFields(t *testing.T) { p.NPMFullMetadata = true p.DirectServe = true p.DirectServeTTL = time.Minute + p.KeepLatestImages = 2 p.RateLimits = registryclient.NewRateLimitGate(http.DefaultTransport) p.HostedNPMScopes = []string{"@example"} p.HostedComposerVendors = []string{"example"} diff --git a/internal/retention/keep_latest_test.go b/internal/retention/keep_latest_test.go index 5d26eabe..c5fcfa8d 100644 --- a/internal/retention/keep_latest_test.go +++ b/internal/retention/keep_latest_test.go @@ -56,8 +56,9 @@ func TestRegistryKeepLatestKeys(t *testing.T) { reg.Register(Spec{Key: "npm", CompareVersions: VersionOrder("npm")}) reg.Register(Spec{Key: "gem"}) reg.Register(Spec{Key: "cargo", CompareVersions: VersionOrder("cargo")}) + reg.Register(Spec{Key: "oci", KeepLatestImages: true}) - if got := reg.KeepLatestKeys(); !slices.Equal(got, []string{"cargo", "npm"}) { - t.Errorf("KeepLatestKeys() = %v, want [cargo npm]", got) + if got := reg.KeepLatestKeys(); !slices.Equal(got, []string{"cargo", "npm", "oci"}) { + t.Errorf("KeepLatestKeys() = %v, want [cargo npm oci]", got) } } diff --git a/internal/retention/retention.go b/internal/retention/retention.go index 8379dc5d..4a1acd8e 100644 --- a/internal/retention/retention.go +++ b/internal/retention/retention.go @@ -63,6 +63,12 @@ type Spec struct { // applies to the ecosystem. Nil means the stored versions cannot be // ordered, as with digests and hashes. CompareVersions func(a, b string) int + + // KeepLatestImages marks an ecosystem that stores the blobs of container + // images under their digests and records which of them each image + // references. Rules.KeepLatest then keeps the blobs of the newest + // images of a repository. + KeepLatestImages bool } // VersionOrder returns a CompareVersions for an ecosystem whose handler @@ -173,7 +179,7 @@ func (r *Registry) Keys() []string { func (r *Registry) KeepLatestKeys() []string { var keys []string for _, key := range r.Keys() { - if s, _ := r.Lookup(key); s.CompareVersions != nil { + if s, _ := r.Lookup(key); s.CompareVersions != nil || s.KeepLatestImages { keys = append(keys, key) } } @@ -258,7 +264,8 @@ type Rules struct { Sources map[string]map[string]time.Duration // KeepLatest is how many of the highest cached versions of each package // are never evicted by age, in ecosystems whose Spec sets - // CompareVersions. Zero keeps none. + // CompareVersions. Where it sets KeepLatestImages, it is how many of the + // newest images of each repository keep their blobs. Zero keeps none. KeepLatest int } diff --git a/internal/server/retention.go b/internal/server/retention.go index d7798497..5003dfd7 100644 --- a/internal/server/retention.go +++ b/internal/server/retention.go @@ -4,8 +4,10 @@ import ( "context" "log/slog" "slices" + "strings" "time" + "github.com/git-pkgs/proxy/internal/config" "github.com/git-pkgs/proxy/internal/database" "github.com/git-pkgs/proxy/internal/metrics" "github.com/git-pkgs/proxy/internal/retention" @@ -15,6 +17,11 @@ import ( // retentionWindow is how many artifact ids one query scans. const retentionWindow = 5000 +// imageRecordGrace is how long a manifest record is safe from pruning, so +// one written while a sweep runs, or shortly before the tag that leads to +// it is recorded, is not lost. +const imageRecordGrace = time.Hour + // retentionMaxClears caps how many artifacts one sweep clears, so a first // sweep over a large, old cache spreads its deletes and the refetches that // follow over several intervals. A variable so tests can lower it. @@ -96,6 +103,8 @@ func sweepRetention(ctx context.Context, db *database.DB, logger *slog.Logger, return } + pruneImageRecords(db, logger, rules.KeepLatest, now) + // Without the buffered hits written, an artifact being served right now // can look expired, so skip this sweep rather than clear it. if err := db.FlushHits(); err != nil { @@ -152,6 +161,41 @@ func sweepRetention(ctx context.Context, db *database.DB, logger *slog.Logger, } } +// keepLatestImages returns storage.retention.keep_latest if a retention rule +// can evict container image blobs, and zero otherwise: without such a rule +// nothing reads what the container handler would record. +func keepLatestImages(cfg *config.Config) int { + rules, err := cfg.RetentionRules(retention.Default) + if err != nil || rules.KeepLatest <= 0 { + return 0 + } + if rules.For("oci", "") > 0 { + return rules.KeepLatest + } + for packagePURL, d := range rules.Packages { + if d > 0 && strings.HasPrefix(packagePURL, "pkg:oci/") { + return rules.KeepLatest + } + } + return 0 +} + +// pruneImageRecords drops the container manifest records that the keep +// newest images of their repository do not reference, which is all of them +// once keep_latest is off. +func pruneImageRecords(db *database.DB, logger *slog.Logger, keep int, now time.Time) { + repositories, err := db.ImageManifestRepositories() + if err != nil { + logger.Warn("retention: failed to list image manifest records", "error", err) + return + } + for _, packagePURL := range repositories { + if err := db.PruneImageManifests(packagePURL, keep, now.Add(-imageRecordGrace)); err != nil { + logger.Warn("retention: failed to prune image manifest records", "package_purl", packagePURL, "error", err) + } + } +} + // clearIfExpired clears one candidate when its own rule says it has expired // and keep does not hold it back, and queues its object for deletion, // reporting whether it did. @@ -180,8 +224,8 @@ func clearIfExpired(db *database.DB, logger *slog.Logger, reg *retention.Registr } // latestKeeper answers, for one sweep, whether an artifact belongs to one of -// the highest cached versions of its package, which keep_latest holds back -// from eviction by age. +// the highest cached versions of its package, or to one of the newest images +// of its repository, which keep_latest holds back from eviction by age. type latestKeeper struct { db *database.DB logger *slog.Logger @@ -197,13 +241,14 @@ func newLatestKeeper(db *database.DB, logger *slog.Logger, reg *retention.Regist } // keeps reports whether c, an artifact of the ecosystem key, is a file of -// one of the highest cached versions of its package. +// one of the highest cached versions of its package, or a blob of one of the +// newest images of its repository. func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { if k.n <= 0 { return false } spec, _ := k.reg.Lookup(key) - if spec.CompareVersions == nil { + if spec.CompareVersions == nil && !spec.KeepLatestImages { return false } kept, ok := k.kept[c.PackagePURL] @@ -212,7 +257,7 @@ func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { if kept, err = k.newest(spec, c.PackagePURL); err != nil { // Not knowing which versions are the highest, keep the artifact // for this sweep rather than evict one of them. - k.logger.Warn("retention: failed to list cached versions, keeping artifact", + k.logger.Warn("retention: failed to list the versions to keep, keeping artifact", "package_purl", c.PackagePURL, "error", err) return true } @@ -223,8 +268,16 @@ func (k *latestKeeper) keeps(key string, c database.RetentionCandidate) bool { } // newest returns the version PURLs of the highest cached versions of a -// package. +// package. For a repository of container images, whose versions are blob +// digests, these are the blobs its newest images reference. func (k *latestKeeper) newest(spec retention.Spec, packagePURL string) (map[string]struct{}, error) { + var imageDigests map[string]struct{} + if spec.KeepLatestImages { + var err error + if imageDigests, err = k.db.NewestImageDigests(packagePURL, k.n); err != nil { + return nil, err + } + } purls, err := k.db.CachedVersionPURLs(packagePURL) if err != nil { return nil, err @@ -242,7 +295,15 @@ func (k *latestKeeper) newest(spec retention.Spec, packagePURL string) (map[stri byVersion[p.Version] = append(byVersion[p.Version], versionPURL) } kept := map[string]struct{}{} - for _, v := range retention.NewestVersions(spec.CompareVersions, versions, k.n) { + if spec.KeepLatestImages { + versions = slices.DeleteFunc(versions, func(digest string) bool { + _, referenced := imageDigests[digest] + return !referenced + }) + } else { + versions = retention.NewestVersions(spec.CompareVersions, versions, k.n) + } + for _, v := range versions { for _, versionPURL := range byVersion[v] { kept[versionPURL] = struct{}{} } diff --git a/internal/server/retention_keep_latest_oci_test.go b/internal/server/retention_keep_latest_oci_test.go new file mode 100644 index 00000000..50739b24 --- /dev/null +++ b/internal/server/retention_keep_latest_oci_test.go @@ -0,0 +1,314 @@ +package server + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/git-pkgs/proxy/internal/config" + "github.com/git-pkgs/proxy/internal/database" + "github.com/git-pkgs/proxy/internal/handler" + "github.com/git-pkgs/proxy/internal/retention" + "github.com/git-pkgs/purl" + "github.com/git-pkgs/registries/fetch" +) + +const ( + keepImagesRepo = "team/app" + keepImagesIndexType = "application/vnd.oci.image.index.v1+json" + keepImagesImageType = "application/vnd.oci.image.manifest.v1+json" +) + +// keepImagesRegistry is a registry with one repository whose tags can be +// moved between images. +type keepImagesRegistry struct { + srv *httptest.Server + mu sync.Mutex + // manifests holds the manifest documents by digest, tags the digest + // each tag points at. + manifests map[string]string + types map[string]string + tags map[string]string + blobs map[string]string +} + +// keepImage is one image of the repository: an index with one platform +// manifest, which has a config and a layer. +type keepImage struct { + index, manifest string + config, layer ociBlob +} + +func newKeepImagesRegistry(t *testing.T) *keepImagesRegistry { + t.Helper() + r := &keepImagesRegistry{manifests: map[string]string{}, types: map[string]string{}, tags: map[string]string{}, blobs: map[string]string{}} + r.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) { + r.mu.Lock() + defer r.mu.Unlock() + const manifestsPrefix, blobsPrefix = "/v2/" + keepImagesRepo + "/manifests/", "/v2/" + keepImagesRepo + "/blobs/" + if len(req.URL.Path) > len(blobsPrefix) && req.URL.Path[:len(blobsPrefix)] == blobsPrefix { + if content, ok := r.blobs[req.URL.Path[len(blobsPrefix):]]; ok { + _, _ = io.WriteString(w, content) + return + } + } + if len(req.URL.Path) > len(manifestsPrefix) && req.URL.Path[:len(manifestsPrefix)] == manifestsPrefix { + digest := req.URL.Path[len(manifestsPrefix):] + if tagged, ok := r.tags[digest]; ok { + digest = tagged + } + if body, ok := r.manifests[digest]; ok { + w.Header().Set("Content-Type", r.types[digest]) + w.Header().Set("Docker-Content-Digest", digest) + if req.Method != http.MethodHead { + _, _ = io.WriteString(w, body) + } + return + } + } + http.NotFound(w, req) + })) + t.Cleanup(r.srv.Close) + return r +} + +func (r *keepImagesRegistry) addManifest(t *testing.T, mediaType string, doc map[string]any) string { + t.Helper() + doc["schemaVersion"], doc["mediaType"] = 2, mediaType + body, err := json.Marshal(doc) + if err != nil { + t.Fatal(err) + } + digest := ociBlob{content: string(body)}.digest() + r.manifests[digest], r.types[digest] = string(body), mediaType + return digest +} + +// addImage adds an image whose blobs are named after it. +func (r *keepImagesRegistry) addImage(t *testing.T, name string) keepImage { + t.Helper() + r.mu.Lock() + defer r.mu.Unlock() + image := keepImage{ + config: ociBlob{keepImagesRepo, name + " config"}, + layer: ociBlob{keepImagesRepo, name + " layer"}, + } + for _, b := range []ociBlob{image.config, image.layer} { + r.blobs[b.digest()] = b.content + } + image.manifest = r.addManifest(t, keepImagesImageType, map[string]any{ + "config": map[string]any{"mediaType": "application/vnd.oci.image.config.v1+json", "digest": image.config.digest()}, + "layers": []map[string]any{{"mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "digest": image.layer.digest()}}, + }) + image.index = r.addManifest(t, keepImagesIndexType, map[string]any{ + "manifests": []map[string]any{{ + "mediaType": keepImagesImageType, "digest": image.manifest, + "platform": map[string]string{"os": "linux", "architecture": "arm64"}, + }}, + }) + return image +} + +func (r *keepImagesRegistry) tag(tag string, image keepImage) { + r.mu.Lock() + defer r.mu.Unlock() + r.tags[tag] = image.index +} + +// newKeepImagesProxy serves the registry through a proxy that keeps the +// blobs of keepLatest images per repository. +func newKeepImagesProxy(t *testing.T, registry *keepImagesRegistry, keepLatest int) (*httptest.Server, *database.DB) { + t.Helper() + db, store := setupEvictionTest(t) + fetcher := fetch.NewFetcher(fetch.WithHTTPClient(registry.srv.Client()), fetch.WithMaxRetries(0)) + t.Cleanup(func() { _ = fetcher.Close() }) + proxy := handler.NewProxy(db, store, fetcher, fetch.NewResolver(), discardLogger()) + proxy.HTTPClient = registry.srv.Client() + proxy.KeepLatestImages = keepLatest + srv := httptest.NewServer(handler.NewContainerHandlerWithRegistry(proxy, "http://proxy.test", registry.srv.URL).Routes()) + t.Cleanup(srv.Close) + return srv, db +} + +// pullImage pulls a tag the way a client does: the tag, the platform +// manifest by digest, then its blobs. +func pullImage(t *testing.T, proxyURL, tag string, image keepImage) { + t.Helper() + base := proxyURL + "/" + keepImagesRepo + ociGet(t, base+"/manifests/"+tag) + ociGet(t, base+"/manifests/"+image.manifest) + ociGet(t, base+"/blobs/"+image.config.digest()) + ociGet(t, base+"/blobs/"+image.layer.digest()) +} + +// ageImages makes every cached blob, every record and every tag sighting +// look old, with the images given first seen longest ago. +func ageImages(t *testing.T, db *database.DB, now time.Time, images ...keepImage) { + t.Helper() + recorded := now.Add(-30 * retentionDay).UnixMilli() + if _, err := db.Exec(db.Rebind(`UPDATE image_manifests SET recorded_at = ?`), recorded); err != nil { + t.Fatalf("backdating records: %v", err) + } + for i, image := range images { + old := now.Add(-time.Duration(30-i) * retentionDay) + ociBackdate(t, db, image.config, old, old) + ociBackdate(t, db, image.layer, old, old) + query := db.Rebind(`UPDATE image_manifests SET tagged_at = ? WHERE digest = ? AND tagged_at > 0`) + if _, err := db.Exec(query, old.UnixMilli(), image.index); err != nil { + t.Fatalf("backdating image %d: %v", i, err) + } + } +} + +func imageCached(t *testing.T, db *database.DB, image keepImage) bool { + t.Helper() + config := ociArtifact(t, db, keepImagesRepo, image.config).StoragePath.Valid + layer := ociArtifact(t, db, keepImagesRepo, image.layer).StoragePath.Valid + if config != layer { + t.Errorf("config cached = %v, layer cached = %v, want the same", config, layer) + } + return layer +} + +// With keep_latest, the blobs of the images a tag most recently pointed at +// outlive their rule. Blobs of images the tag left longer ago are evicted. +func TestSweepRetentionKeepsBlobsOfNewestImages(t *testing.T) { + for _, tt := range []struct { + keepLatest int + wantCached [3]bool + }{ + {keepLatest: 0, wantCached: [3]bool{false, false, false}}, + {keepLatest: 1, wantCached: [3]bool{false, false, true}}, + {keepLatest: 2, wantCached: [3]bool{false, true, true}}, + } { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, tt.keepLatest) + images := []keepImage{registry.addImage(t, "first"), registry.addImage(t, "second"), registry.addImage(t, "third")} + for _, image := range images { + registry.tag("beta", image) + pullImage(t, srv.URL, "beta", image) + } + now := time.Now() + ageImages(t, db, now, images...) + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: tt.keepLatest} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + + for i, image := range images { + if got := imageCached(t, db, image); got != tt.wantCached[i] { + t.Errorf("keep_latest %d: image %d cached = %v, want %v", tt.keepLatest, i+1, got, tt.wantCached[i]) + } + } + } +} + +// A sweep drops the records of images that are no longer among the newest, +// so the table holds a few images per repository, and all records once +// keep_latest is turned off. +func TestSweepRetentionPrunesImageRecords(t *testing.T) { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, 1) + images := []keepImage{registry.addImage(t, "first"), registry.addImage(t, "second")} + for _, image := range images { + registry.tag("beta", image) + pullImage(t, srv.URL, "beta", image) + } + now := time.Now() + ageImages(t, db, now, images...) + recorded := func(digest string) bool { + t.Helper() + _, ok, err := db.ImageManifestTaggedAt(purl.MakePURLString("oci", keepImagesRepo, ""), digest) + if err != nil { + t.Fatalf("ImageManifestTaggedAt: %v", err) + } + return ok + } + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: 1} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + for _, digest := range []string{images[0].index, images[0].manifest} { + if recorded(digest) { + t.Errorf("manifest %s of the older image is still recorded", digest) + } + } + for _, digest := range []string{images[1].index, images[1].manifest} { + if !recorded(digest) { + t.Errorf("manifest %s of the newest image is no longer recorded", digest) + } + } + + rules.KeepLatest = 0 + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + if recorded(images[1].index) { + t.Error("records are left after keep_latest was turned off") + } +} + +// Manifests are recorded only when a retention rule can evict container +// blobs, since nothing reads the records otherwise. +func TestKeepLatestImagesNeedsAnOCIRule(t *testing.T) { + for _, tt := range []struct { + name string + def string + ecosystems map[string]string + packages map[string]string + keepLatest int + want int + }{ + {name: "no keep_latest", ecosystems: map[string]string{"oci": "1d"}, want: 0}, + {name: "no rule", keepLatest: 2, want: 0}, + {name: "rule for another ecosystem", ecosystems: map[string]string{"npm": "7d"}, keepLatest: 2, want: 0}, + {name: "oci rule", ecosystems: map[string]string{"oci": "1d"}, keepLatest: 2, want: 2}, + {name: "default", def: "30d", keepLatest: 2, want: 2}, + {name: "default with oci exempt", def: "30d", ecosystems: map[string]string{"oci": "0"}, keepLatest: 2, want: 0}, + {name: "rule for one repository", packages: map[string]string{"pkg:oci/library/nginx": "3d"}, keepLatest: 2, want: 2}, + {name: "rule for another ecosystem's package", packages: map[string]string{"pkg:npm/lodash": "3d"}, keepLatest: 2, want: 0}, + } { + t.Run(tt.name, func(t *testing.T) { + cfg := config.Default() + cfg.Storage.Retention.Default = tt.def + cfg.Storage.Retention.Ecosystems = tt.ecosystems + cfg.Storage.Retention.Packages = tt.packages + cfg.Storage.Retention.KeepLatest = tt.keepLatest + if got := keepLatestImages(cfg); got != tt.want { + t.Errorf("keepLatestImages() = %d, want %d", got, tt.want) + } + }) + } +} + +// proxy mirror runs in a process of its own and has to record the images it +// mirrors as serve does, or the sweep would not know their blobs. +func TestMirrorProxyRecordsImagesForKeepLatest(t *testing.T) { + cfg := config.Default() + cfg.Storage.Retention.Ecosystems = map[string]string{"oci": "1d"} + cfg.Storage.Retention.KeepLatest = 2 + proxy, _ := NewMirrorProxy(cfg, nil, nil, slog.New(slog.NewTextHandler(io.Discard, nil))) + if proxy.KeepLatestImages != 2 { + t.Errorf("KeepLatestImages = %d, want 2", proxy.KeepLatestImages) + } +} + +// An image pulled only by digest is not one of the newest images, so its +// blobs are evicted by age as without keep_latest. +func TestSweepRetentionKeepLatestIgnoresImagesPulledByDigest(t *testing.T) { + registry := newKeepImagesRegistry(t) + srv, db := newKeepImagesProxy(t, registry, 1) + image := registry.addImage(t, "pinned") + pullImage(t, srv.URL, image.index, image) + now := time.Now() + ageImages(t, db, now, image) + + rules := retention.Rules{Ecosystems: map[string]time.Duration{"oci": retentionDay}, KeepLatest: 1} + sweepRetention(context.Background(), db, discardLogger(), retention.Default, rules, now) + + if imageCached(t, db, image) { + t.Error("blobs of an image pulled only by digest were kept") + } +} diff --git a/internal/server/retention_keep_latest_test.go b/internal/server/retention_keep_latest_test.go index 9ee138ba..dd809854 100644 --- a/internal/server/retention_keep_latest_test.go +++ b/internal/server/retention_keep_latest_test.go @@ -206,9 +206,9 @@ func TestSweepRetentionWithoutKeepLatestEvictsNewestToo(t *testing.T) { } // The ecosystems whose handlers store plain release versions support -// keep_latest. +// keep_latest, and so does oci, which keeps by image. func TestKeepLatestEcosystems(t *testing.T) { - want := []string{"cargo", "golang", "npm", "pypi"} + want := []string{"cargo", "golang", "npm", "oci", "pypi"} if got := retention.Default.KeepLatestKeys(); !slices.Equal(got, want) { t.Errorf("KeepLatestKeys() = %v, want %v", got, want) } diff --git a/internal/server/server.go b/internal/server/server.go index 59717d44..39d43c9d 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -308,6 +308,7 @@ func (s *Server) serve(listener net.Listener) error { return fmt.Errorf("configuring scanners: %w", err) } configureMetadata(proxy, s.cfg) + proxy.KeepLatestImages = keepLatestImages(s.cfg) proxy.GradleReadOnly = s.cfg.Gradle.BuildCache.ReadOnly proxy.HostedEnabled = s.cfg.Hosted.Enabled proxy.HostedNPMScopes = s.cfg.Hosted.NPMScopes @@ -632,6 +633,7 @@ func NewMirrorProxy(cfg *config.Config, db *database.DB, store storage.Storage, configureMirrorResolvers(proxy, cfg.Upstream) proxy.HTTPClient = metadataClient proxy.RateLimits = s.rateLimits + proxy.KeepLatestImages = keepLatestImages(cfg) return proxy, s.newContainerHandler(proxy) }