Compare commits
3 Commits
v3.6.3
..
8fc1635d11
| Author | SHA1 | Date | |
|---|---|---|---|
| 8fc1635d11 | |||
| 8d9bc1c422 | |||
| 30b7cef026 |
@@ -109,16 +109,22 @@ func (db *DB) InsertAccessLog(ctx context.Context, remoteName, path string, cach
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
func (db *DB) FindOrphanedBlobs(ctx context.Context) ([]models.Blob, error) {
|
// FindOrphanedBlobs returns blobs no longer referenced by any artifact or
|
||||||
|
// local file, restricted to those created before now()-minAge. The age cutoff
|
||||||
|
// is a grace period that avoids a TOCTOU race with in-flight dedup uploads,
|
||||||
|
// which insert the blob row before the referencing artifact/local_files row.
|
||||||
|
func (db *DB) FindOrphanedBlobs(ctx context.Context, minAge time.Duration) ([]models.Blob, error) {
|
||||||
|
cutoff := time.Now().Add(-minAge)
|
||||||
rows, err := db.Pool.Query(ctx, `
|
rows, err := db.Pool.Query(ctx, `
|
||||||
SELECT b.content_hash, b.s3_key, b.size_bytes, b.content_type, b.created_at
|
SELECT b.content_hash, b.s3_key, b.size_bytes, b.content_type, b.created_at
|
||||||
FROM blobs b
|
FROM blobs b
|
||||||
WHERE b.content_hash NOT IN (
|
WHERE b.created_at < $1
|
||||||
|
AND b.content_hash NOT IN (
|
||||||
SELECT content_hash FROM artifacts
|
SELECT content_hash FROM artifacts
|
||||||
UNION
|
UNION
|
||||||
SELECT content_hash FROM local_files
|
SELECT content_hash FROM local_files
|
||||||
)
|
)
|
||||||
`)
|
`, cutoff)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -30,6 +30,15 @@ func (db *DB) GetOverviewStats(ctx context.Context) (*models.OverviewStats, erro
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
err = db.Pool.QueryRow(ctx, `
|
||||||
|
SELECT COALESCE(SUM(size_bytes), 0)
|
||||||
|
FROM access_log
|
||||||
|
WHERE cache_hit = TRUE AND created_at > NOW() - INTERVAL '30 days'
|
||||||
|
`).Scan(&stats.BandwidthSaved30d)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
return &stats, nil
|
return &stats, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+6
-1
@@ -9,6 +9,11 @@ import (
|
|||||||
"git.unkin.net/unkin/artifactapi/internal/storage"
|
"git.unkin.net/unkin/artifactapi/internal/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// blobGracePeriod is how old an orphaned blob must be before GC will delete
|
||||||
|
// it. This avoids racing in-flight dedup uploads that insert the blob row
|
||||||
|
// before the referencing artifact/local_files row exists.
|
||||||
|
const blobGracePeriod = 1 * time.Hour
|
||||||
|
|
||||||
type Collector struct {
|
type Collector struct {
|
||||||
db *database.DB
|
db *database.DB
|
||||||
store *storage.S3
|
store *storage.S3
|
||||||
@@ -38,7 +43,7 @@ func (c *Collector) Run(ctx context.Context) {
|
|||||||
func (c *Collector) sweep(ctx context.Context) {
|
func (c *Collector) sweep(ctx context.Context) {
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
|
|
||||||
orphaned, err := c.db.FindOrphanedBlobs(ctx)
|
orphaned, err := c.db.FindOrphanedBlobs(ctx, blobGracePeriod)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
slog.Error("gc: find orphaned blobs", "error", err)
|
slog.Error("gc: find orphaned blobs", "error", err)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -65,11 +65,12 @@ func (m *HelmMerger) MergeIndexes(members []MemberIndex, proxyBaseURL string) ([
|
|||||||
if baseHost != "" && extractHost(u) != baseHost {
|
if baseHost != "" && extractHost(u) != baseHost {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
relPath := extractPathRelativeToBase(u, member.BaseURL)
|
||||||
ver.URLs[i] = fmt.Sprintf("%s/api/v1/%s/%s/%s",
|
ver.URLs[i] = fmt.Sprintf("%s/api/v1/%s/%s/%s",
|
||||||
strings.TrimRight(proxyBaseURL, "/"),
|
strings.TrimRight(proxyBaseURL, "/"),
|
||||||
routePrefix,
|
routePrefix,
|
||||||
member.RemoteName,
|
member.RemoteName,
|
||||||
extractPath(u))
|
relPath)
|
||||||
} else {
|
} else {
|
||||||
ver.URLs[i] = fmt.Sprintf("%s/api/v1/%s/%s/%s",
|
ver.URLs[i] = fmt.Sprintf("%s/api/v1/%s/%s/%s",
|
||||||
strings.TrimRight(proxyBaseURL, "/"),
|
strings.TrimRight(proxyBaseURL, "/"),
|
||||||
@@ -102,6 +103,18 @@ func extractHost(rawURL string) string {
|
|||||||
return rest[:slashIdx]
|
return rest[:slashIdx]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func extractPathRelativeToBase(rawURL, baseURL string) string {
|
||||||
|
fullPath := extractPath(rawURL)
|
||||||
|
basePath := extractPath(baseURL)
|
||||||
|
if basePath != "" {
|
||||||
|
basePath = strings.TrimRight(basePath, "/") + "/"
|
||||||
|
if strings.HasPrefix(fullPath, basePath) {
|
||||||
|
return fullPath[len(basePath):]
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return fullPath
|
||||||
|
}
|
||||||
|
|
||||||
func extractPath(rawURL string) string {
|
func extractPath(rawURL string) string {
|
||||||
idx := strings.Index(rawURL, "://")
|
idx := strings.Index(rawURL, "://")
|
||||||
if idx == -1 {
|
if idx == -1 {
|
||||||
|
|||||||
@@ -50,6 +50,11 @@ export function Dashboard() {
|
|||||||
value={formatNumber(stats.total_blobs_deduped)}
|
value={formatNumber(stats.total_blobs_deduped)}
|
||||||
sub="shared blobs"
|
sub="shared blobs"
|
||||||
/>
|
/>
|
||||||
|
<StatsCard
|
||||||
|
label="Bandwidth Saved"
|
||||||
|
value={formatBytes(stats.bandwidth_saved_30d)}
|
||||||
|
sub="last 30 days"
|
||||||
|
/>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
{health && (
|
{health && (
|
||||||
|
|||||||
Reference in New Issue
Block a user