rdev/internal/adapter/zot/client.go
jordan 4486042155
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed
fix(registry): delete container images on project teardown
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>
2026-02-08 02:56:18 -07:00

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
}