Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
Root cause of DIGEST_INVALID errors was registry disk exhaustion. Project teardown wasn't cleaning up container images, causing the registry PVC to fill up over time. Changes: - Add RegistryProvider port interface for registry operations - Extend zot.Client with DeleteProjectRepositories method - Wire registry provider into ProjectInfraService - Delete images during DeleteProject cleanup (step 4) The zot client uses the OCI distribution API: - Lists all repos, filters by project prefix - Gets manifest digests via HEAD request - Deletes manifests by digest to trigger GC Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
268 lines
7.5 KiB
Go
268 lines
7.5 KiB
Go
// Package zot provides a client for interacting with the zot container registry.
|
|
package zot
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/orchard9/rdev/internal/domain"
|
|
"github.com/orchard9/rdev/internal/port"
|
|
)
|
|
|
|
// Client interacts with the zot container registry.
|
|
type Client struct {
|
|
url string
|
|
httpClient *http.Client
|
|
logger *slog.Logger
|
|
}
|
|
|
|
// Ensure Client implements port.RegistryProvider at compile time.
|
|
var _ port.RegistryProvider = (*Client)(nil)
|
|
|
|
// NewClient creates a new zot client.
|
|
// The URL should be the registry base URL (e.g., "https://registry.threesix.ai").
|
|
func NewClient(url string) *Client {
|
|
return &Client{
|
|
url: url,
|
|
httpClient: &http.Client{
|
|
Timeout: 30 * time.Second,
|
|
},
|
|
logger: slog.Default(),
|
|
}
|
|
}
|
|
|
|
// WithLogger sets the logger for the client.
|
|
func (c *Client) WithLogger(logger *slog.Logger) *Client {
|
|
c.logger = logger
|
|
return c
|
|
}
|
|
|
|
// Check returns the health status of the registry.
|
|
// A 200 or 401 response indicates the registry is healthy (401 means auth required but registry is up).
|
|
func (c *Client) Check(ctx context.Context) domain.RegistryStatus {
|
|
start := time.Now()
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url+"/v2/", nil)
|
|
if err != nil {
|
|
return domain.RegistryStatus{
|
|
Healthy: false,
|
|
URL: c.url,
|
|
Error: fmt.Sprintf("failed to create request: %v", err),
|
|
LastChecked: time.Now().UTC(),
|
|
}
|
|
}
|
|
|
|
resp, err := c.httpClient.Do(req)
|
|
latency := time.Since(start)
|
|
|
|
if err != nil {
|
|
return domain.RegistryStatus{
|
|
Healthy: false,
|
|
URL: c.url,
|
|
Latency: latency.String(),
|
|
Error: fmt.Sprintf("connection error: %v", err),
|
|
LastChecked: time.Now().UTC(),
|
|
}
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
// 200 = healthy, 401 = healthy but requires auth
|
|
healthy := resp.StatusCode == http.StatusOK || resp.StatusCode == http.StatusUnauthorized
|
|
|
|
status := domain.RegistryStatus{
|
|
Healthy: healthy,
|
|
URL: c.url,
|
|
Latency: latency.String(),
|
|
LastChecked: time.Now().UTC(),
|
|
}
|
|
|
|
if !healthy {
|
|
status.Error = fmt.Sprintf("unexpected status code: %d", resp.StatusCode)
|
|
}
|
|
|
|
return status
|
|
}
|
|
|
|
// catalogResponse is the OCI catalog API response.
|
|
type catalogResponse struct {
|
|
Repositories []string `json:"repositories"`
|
|
}
|
|
|
|
// tagsResponse is the OCI tags list API response.
|
|
type tagsResponse struct {
|
|
Name string `json:"name"`
|
|
Tags []string `json:"tags"`
|
|
}
|
|
|
|
// ListRepositories returns all repositories in the registry.
|
|
func (c *Client) ListRepositories(ctx context.Context) ([]string, error) {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url+"/v2/_catalog", nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create request: %w", err)
|
|
}
|
|
|
|
resp, err := c.httpClient.Do(req)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("request failed: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, fmt.Errorf("unexpected status: %d", resp.StatusCode)
|
|
}
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read body: %w", err)
|
|
}
|
|
|
|
var catalog catalogResponse
|
|
if err := json.Unmarshal(body, &catalog); err != nil {
|
|
return nil, fmt.Errorf("parse catalog: %w", err)
|
|
}
|
|
|
|
return catalog.Repositories, nil
|
|
}
|
|
|
|
// ListProjectRepositories returns all repositories for a specific project.
|
|
// This includes both the main repo and sub-repos like cache.
|
|
func (c *Client) ListProjectRepositories(ctx context.Context, projectID string) ([]string, error) {
|
|
allRepos, err := c.ListRepositories(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var projectRepos []string
|
|
prefix := projectID + "/"
|
|
for _, repo := range allRepos {
|
|
if repo == projectID || strings.HasPrefix(repo, prefix) {
|
|
projectRepos = append(projectRepos, repo)
|
|
}
|
|
}
|
|
return projectRepos, nil
|
|
}
|
|
|
|
// DeleteRepository deletes all tags and manifests for a repository.
|
|
// This triggers Zot's garbage collection to reclaim storage.
|
|
func (c *Client) DeleteRepository(ctx context.Context, repo string) error {
|
|
// List all tags for the repository
|
|
tags, err := c.listTags(ctx, repo)
|
|
if err != nil {
|
|
return fmt.Errorf("list tags for %s: %w", repo, err)
|
|
}
|
|
|
|
// Delete each tag's manifest
|
|
for _, tag := range tags {
|
|
if err := c.deleteManifest(ctx, repo, tag); err != nil {
|
|
c.logger.Warn("failed to delete manifest", "repo", repo, "tag", tag, "error", err)
|
|
// Continue with other tags
|
|
}
|
|
}
|
|
|
|
c.logger.Info("deleted repository", "repo", repo, "tags_deleted", len(tags))
|
|
return nil
|
|
}
|
|
|
|
// DeleteProjectRepositories deletes all repositories for a project.
|
|
func (c *Client) DeleteProjectRepositories(ctx context.Context, projectID string) error {
|
|
repos, err := c.ListProjectRepositories(ctx, projectID)
|
|
if err != nil {
|
|
return fmt.Errorf("list project repos: %w", err)
|
|
}
|
|
|
|
for _, repo := range repos {
|
|
if err := c.DeleteRepository(ctx, repo); err != nil {
|
|
c.logger.Warn("failed to delete repo", "repo", repo, "error", err)
|
|
// Continue with other repos
|
|
}
|
|
}
|
|
|
|
c.logger.Info("deleted project repositories", "project", projectID, "repos_deleted", len(repos))
|
|
return nil
|
|
}
|
|
|
|
// listTags returns all tags for a repository.
|
|
func (c *Client) listTags(ctx context.Context, repo string) ([]string, error) {
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.url+"/v2/"+repo+"/tags/list", nil)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("create request: %w", err)
|
|
}
|
|
|
|
resp, err := c.httpClient.Do(req)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("request failed: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode == http.StatusNotFound {
|
|
return nil, nil // Repo doesn't exist, nothing to delete
|
|
}
|
|
if resp.StatusCode != http.StatusOK {
|
|
return nil, fmt.Errorf("unexpected status: %d", resp.StatusCode)
|
|
}
|
|
|
|
body, err := io.ReadAll(resp.Body)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("read body: %w", err)
|
|
}
|
|
|
|
var tags tagsResponse
|
|
if err := json.Unmarshal(body, &tags); err != nil {
|
|
return nil, fmt.Errorf("parse tags: %w", err)
|
|
}
|
|
|
|
return tags.Tags, nil
|
|
}
|
|
|
|
// deleteManifest deletes a specific manifest by tag.
|
|
func (c *Client) deleteManifest(ctx context.Context, repo, tag string) error {
|
|
// First, get the manifest digest
|
|
headReq, err := http.NewRequestWithContext(ctx, http.MethodHead, c.url+"/v2/"+repo+"/manifests/"+tag, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("create head request: %w", err)
|
|
}
|
|
headReq.Header.Set("Accept", "application/vnd.oci.image.manifest.v1+json, application/vnd.docker.distribution.manifest.v2+json")
|
|
|
|
headResp, err := c.httpClient.Do(headReq)
|
|
if err != nil {
|
|
return fmt.Errorf("head request failed: %w", err)
|
|
}
|
|
defer func() { _ = headResp.Body.Close() }()
|
|
|
|
if headResp.StatusCode == http.StatusNotFound {
|
|
return nil // Already deleted
|
|
}
|
|
if headResp.StatusCode != http.StatusOK {
|
|
return fmt.Errorf("head unexpected status: %d", headResp.StatusCode)
|
|
}
|
|
|
|
digest := headResp.Header.Get("Docker-Content-Digest")
|
|
if digest == "" {
|
|
return fmt.Errorf("no digest in response")
|
|
}
|
|
|
|
// Delete by digest
|
|
delReq, err := http.NewRequestWithContext(ctx, http.MethodDelete, c.url+"/v2/"+repo+"/manifests/"+digest, nil)
|
|
if err != nil {
|
|
return fmt.Errorf("create delete request: %w", err)
|
|
}
|
|
|
|
delResp, err := c.httpClient.Do(delReq)
|
|
if err != nil {
|
|
return fmt.Errorf("delete request failed: %w", err)
|
|
}
|
|
defer func() { _ = delResp.Body.Close() }()
|
|
|
|
if delResp.StatusCode != http.StatusAccepted && delResp.StatusCode != http.StatusOK && delResp.StatusCode != http.StatusNotFound {
|
|
return fmt.Errorf("delete unexpected status: %d", delResp.StatusCode)
|
|
}
|
|
|
|
return nil
|
|
}
|