feat: #73 support registry data prune

Author: bohdan-shulha

go.mod

@@ -6,10 +6,10 @@ require (
 	github.com/aws/aws-sdk-go-v2 v1.32.2
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.41
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.65.3
+	github.com/distribution/reference v0.6.0
 	github.com/docker/docker v27.0.0+incompatible
 	github.com/mackerelio/go-osstat v0.2.5
 	github.com/ncruces/go-sqlite3 v0.18.4
-	github.com/opencontainers/image-spec v1.1.0
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.14.0
 	github.com/prometheus/common v0.42.0
@@ -31,24 +31,29 @@ require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/distribution v2.8.3+incompatible // indirect
 	github.com/docker/go-connections v0.5.0 // indirect
+	github.com/docker/go-metrics v0.0.1 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
+	github.com/docker/libtrust v0.0.0-20160708172513-aabc10ec26b7 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/protobuf v1.5.2 // indirect
+	github.com/gorilla/mux v1.8.1 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/moby/docker-image-spec v1.3.1 // indirect
 	github.com/moby/term v0.5.0 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/ncruces/julianday v1.0.0 // indirect
 	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/client_model v0.3.0 // indirect
 	github.com/prometheus/procfs v0.8.0 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
 	github.com/tetratelabs/wazero v1.8.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.52.0 // indirect
 	go.opentelemetry.io/otel v1.27.0 // indirect

go.sum

@@ -6,12 +6,15 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"slices"
 	"strings"
 
+	"github.com/distribution/reference"
 	"github.com/docker/docker/api/types/mount"
 	"github.com/pkg/errors"
 	"github.com/ptah-sh/ptah-agent/internal/app/ptah-agent/busybox"
 	"github.com/ptah-sh/ptah-agent/internal/app/ptah-agent/docker/config"
+	"github.com/ptah-sh/ptah-agent/internal/app/ptah-agent/registry"
 	t "github.com/ptah-sh/ptah-agent/internal/pkg/ptah-client"
 )
 
@@ -145,3 +148,78 @@ func (e *taskExecutor) pullImage(ctx context.Context, req *t.PullImageReq) (*t.P
 
 	return &t.PullImageRes{Output: output}, nil
 }
+
+func (e *taskExecutor) pruneDockerRegistry(ctx context.Context, req *t.PruneDockerRegistryReq) (*t.PruneDockerRegistryRes, error) {
+	log := Logger(ctx)
+
+	var result t.PruneDockerRegistryRes
+
+	tagsToKeep := make(map[string][]string)
+	for _, imageRef := range req.KeepImages {
+		ref, err := reference.ParseNamed(imageRef)
+		if err != nil {
+			return nil, fmt.Errorf("prune docker registry: %w", err)
+		}
+
+		repo := reference.Path(ref)
+
+		repoTags, ok := tagsToKeep[repo]
+		if !ok {
+			repoTags = make([]string, 0)
+		}
+
+		taggedRef, ok := reference.TagNameOnly(ref).(reference.Tagged)
+		if !ok {
+			return nil, fmt.Errorf("prune docker registry: can not get tag from ref: %s", imageRef)
+		}
+
+		tagsToKeep[repo] = append(repoTags, taggedRef.Tag())
+
+		log.Debug("parsed image ref", "ref", ref, "repo", repo, "tag", taggedRef.Tag())
+	}
+
+	registry := registry.New("http://registry.ptah.local:5050")
+
+	catalog, err := registry.Catalog(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("prune docker registry: %w", err)
+	}
+
+	for _, repo := range catalog.Repositories {
+		log.Debug("processing repo", "repo", repo)
+
+		tags, err := registry.TagsList(ctx, repo)
+		if err != nil {
+			return nil, fmt.Errorf("prune docker registry: %w", err)
+		}
+
+		for _, tag := range tags.Tags {
+			log.Debug("processing tag", "tag", tag)
+
+			if _, ok := tagsToKeep[repo]; !ok || !slices.Contains(tagsToKeep[repo], tag) {
+				manifest, err := registry.ManifestHead(ctx, repo, tag)
+				if err != nil {
+					return nil, fmt.Errorf("prune docker registry: %w", err)
+				}
+
+				log.Debug("deleting image", "repo", repo, "tag", tag, "digest", manifest.Digest)
+
+				if err := registry.ManifestDelete(ctx, repo, manifest.Digest); err != nil {
+					return nil, fmt.Errorf("prune docker registry: %w", err)
+				}
+			}
+		}
+	}
+
+	// ref, err := reference.ParseNamed("ptah/test")
+	// if err != nil {
+	// 	return nil, fmt.Errorf("prune docker registry: %w", err)
+	// }
+
+	// info, err := dockerRegistry.ParseRepositoryInfo(ref)
+	// if err != nil {
+	// 	return nil, fmt.Errorf("prune docker registry: %w", err)
+	// }
+
+	return &result, nil
+}

internal/app/ptah-agent/images.go

@@ -64,6 +64,8 @@ func parseTask(taskType int, payload string) (interface{}, error) {
 		return unmarshalTask(payload, &ptahClient.BuildImageReq{})
 	case 26:
 		return unmarshalTask(payload, &ptahClient.BuildImageWithNixpacksReq{})
+	case 27:
+		return unmarshalTask(payload, &ptahClient.PruneDockerRegistryReq{})
 	default:
 		return nil, fmt.Errorf("parse task: unknown task type %d", taskType)
 	}

internal/app/ptah-agent/parse_task.go

@@ -0,0 +1,29 @@
+package registry
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+)
+
+type CatalogRes struct {
+	Repositories []string `json:"repositories"`
+}
+
+func (r *Registry) Catalog(ctx context.Context) (*CatalogRes, error) {
+	url := fmt.Sprintf("%s/v2/_catalog", r.baseUrl)
+
+	resp, err := r.client.Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("get catalog: %w", err)
+	}
+
+	defer resp.Body.Close()
+
+	var result CatalogRes
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("decode catalog: %w", err)
+	}
+
+	return &result, nil
+}

internal/app/ptah-agent/registry/catalog.go

@@ -0,0 +1,51 @@
+package registry
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+)
+
+type ManifestHeadRes struct {
+	Digest string `json:"digest"`
+}
+
+func (r *Registry) ManifestHead(ctx context.Context, repo, tag string) (*ManifestHeadRes, error) {
+	url := fmt.Sprintf("%s/v2/%s/manifests/%s", r.baseUrl, repo, tag)
+
+	req, err := http.NewRequestWithContext(ctx, "HEAD", url, nil)
+	if err != nil {
+		return nil, fmt.Errorf("create manifest head request: %w", err)
+	}
+
+	req.Header.Set("Accept", "application/vnd.docker.distribution.manifest.v2+json")
+
+	resp, err := r.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("do manifest head request: %w", err)
+	}
+
+	defer resp.Body.Close()
+
+	return &ManifestHeadRes{
+		Digest: resp.Header.Get("Docker-Content-Digest"),
+	}, nil
+}
+
+func (r *Registry) ManifestDelete(ctx context.Context, repo, digest string) error {
+	url := fmt.Sprintf("%s/v2/%s/manifests/%s", r.baseUrl, repo, digest)
+
+	req, err := http.NewRequestWithContext(ctx, "DELETE", url, nil)
+	if err != nil {
+		return fmt.Errorf("create manifest delete request: %w", err)
+	}
+
+	resp, err := r.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("do manifest delete request: %w", err)
+	}
+
+	defer resp.Body.Close()
+
+	return nil
+}

internal/app/ptah-agent/registry/manifests.go

@@ -0,0 +1,15 @@
+package registry
+
+import "net/http"
+
+type Registry struct {
+	baseUrl string
+	client  *http.Client
+}
+
+func New(baseUrl string) *Registry {
+	return &Registry{
+		baseUrl: baseUrl,
+		client:  &http.Client{},
+	}
+}

internal/app/ptah-agent/registry/registry.go

@@ -0,0 +1,29 @@
+package registry
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+)
+
+type TagsListRes struct {
+	Tags []string `json:"tags"`
+}
+
+func (r *Registry) TagsList(ctx context.Context, repo string) (*TagsListRes, error) {
+	url := fmt.Sprintf("%s/v2/%s/tags/list", r.baseUrl, repo)
+
+	resp, err := r.client.Get(url)
+	if err != nil {
+		return nil, fmt.Errorf("get tags list: %w", err)
+	}
+
+	defer resp.Body.Close()
+
+	var result TagsListRes
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("decode tags list: %w", err)
+	}
+
+	return &result, nil
+}

internal/app/ptah-agent/registry/tags.go

@@ -58,6 +58,7 @@ func (e *taskExecutor) monitorDaemonServiceLaunch(ctx context.Context, service *
 
 	successfullChecks := 0
 
+	// FIXME: check if the container is continiously restarting.
 	for {
 		select {
 		case <-ctx.Done():

internal/app/ptah-agent/service_monitor.go

@@ -74,6 +74,8 @@ func (e *taskExecutor) executeTask(ctx context.Context, anyTask interface{}) (in
 		return e.buildImage(ctx, task)
 	case *t.BuildImageWithNixpacksReq:
 		return e.buildImageWithNixpacks(ctx, task)
+	case *t.PruneDockerRegistryReq:
+		return e.pruneDockerRegistry(ctx, task)
 	default:
 		return nil, fmt.Errorf("execute task: unknown task type %T", task)
 	}