feat(ecr): implement image-management control plane (ListImages, DescribeImages, BatchGetImage, PutImage, BatchDeleteImage) (#389)

Serve the five image-management JSON 1.1 actions inline in ECR_Request,
reusing the gateway-side OCI Registry for manifest bytes:

- ListImages / DescribeImages enumerate images by tag + digest (KV meta).
- BatchGetImage returns verbatim manifest bytes with Q14 partial-failure
  semantics (cap 100, digest wins over tag, 200 + failures array).
- PutImage stores a manifest + tag via a shared StoreManifest core that the
  OCI manifest PUT now delegates to.
- BatchDeleteImage removes tag pointers / manifest meta (blob GC deferred).

Adds DeleteManifestMeta across the MetaStore relay chain and ECR image
error codes.
This commit is contained in:
Josh Siviour
2026-06-18 22:14:57 +10:00
committed by GitHub
parent 1608ffa465
commit 2655178cb5
16 changed files with 1157 additions and 53 deletions
+12
View File
@@ -458,6 +458,12 @@ var (
ErrorRepositoryPolicyNotFound = "RepositoryPolicyNotFoundException"
ErrorRepositoryAlreadyExists = "RepositoryAlreadyExistsException"
ErrorRepositoryNotEmpty = "RepositoryNotEmptyException"
ErrorImageNotFound = "ImageNotFoundException"
ErrorImageAlreadyExists = "ImageAlreadyExistsException"
ErrorImageDigestDoesNotMatch = "ImageDigestDoesNotMatchException"
ErrorLayersNotFound = "LayersNotFoundException"
ErrorReferencedImagesNotFound = "ReferencedImagesNotFoundException"
ErrorTooManyTags = "TooManyTagsException"
// ELBv2-specific error codes
ErrorELBv2LoadBalancerNotFound = "LoadBalancerNotFound"
@@ -968,6 +974,12 @@ var ErrorLookup = map[string]ErrorMessage{
ErrorRepositoryPolicyNotFound: {HTTPCode: 400, Message: "The repository and registry do not have an associated repository policy."},
ErrorRepositoryAlreadyExists: {HTTPCode: 400, Message: "The repository already exists in the registry."},
ErrorRepositoryNotEmpty: {HTTPCode: 400, Message: "The repository contains images. Either delete the images or use the force option to delete the repository."},
ErrorImageNotFound: {HTTPCode: 400, Message: "The image requested does not exist in the specified repository."},
ErrorImageAlreadyExists: {HTTPCode: 400, Message: "The specified image has already been pushed, and there were no changes to the manifest or image tag after the last push."},
ErrorImageDigestDoesNotMatch: {HTTPCode: 400, Message: "The specified image digest does not match the digest of the supplied image manifest."},
ErrorLayersNotFound: {HTTPCode: 400, Message: "The specified layers could not be found, or the specified layer is not valid for this repository."},
ErrorReferencedImagesNotFound: {HTTPCode: 400, Message: "The manifest list is referencing an image that does not exist."},
ErrorTooManyTags: {HTTPCode: 400, Message: "The list of tags on the repository is over the limit. The maximum number of tags that can be applied to a repository is 50."},
// ELBv2 error codes
ErrorELBv2LoadBalancerNotFound: {HTTPCode: 400, Message: "One or more load balancers not found."},
+1
View File
@@ -976,6 +976,7 @@ func (d *Daemon) subscribeAll() error {
natsSub{handlers_ecr.SubjectManifestPut, d.handleECRManifestPut, "spinifex-workers"},
natsSub{handlers_ecr.SubjectManifestDescribe, d.handleECRManifestDescribe, "spinifex-workers"},
natsSub{handlers_ecr.SubjectManifestList, d.handleECRManifestList, "spinifex-workers"},
natsSub{handlers_ecr.SubjectManifestDelete, d.handleECRManifestDelete, "spinifex-workers"},
natsSub{handlers_ecr.SubjectUploadCreate, d.handleECRUploadCreate, "spinifex-workers"},
natsSub{handlers_ecr.SubjectUploadGet, d.handleECRUploadGet, "spinifex-workers"},
natsSub{handlers_ecr.SubjectUploadUpdate, d.handleECRUploadUpdate, "spinifex-workers"},
+4
View File
@@ -58,6 +58,10 @@ func (d *Daemon) handleECRManifestList(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.ManifestList)
}
func (d *Daemon) handleECRManifestDelete(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.ManifestDelete)
}
func (d *Daemon) handleECRUploadCreate(msg *nats.Msg) {
handleNATSRequest(msg, d.ecrMetaService.UploadCreate)
}
+7 -50
View File
@@ -551,64 +551,21 @@ func (reg *Registry) getManifest(w http.ResponseWriter, r *http.Request, name, r
// to ensure every referenced blob exists in the pool; image indexes are checked
// to ensure every referenced child manifest exists with a matching mediaType.
func (reg *Registry) putManifest(w http.ResponseWriter, r *http.Request, name, reference string) {
if err := reg.ensureRepo(name); err != nil {
reg.internal(w, "ensure repo", err)
return
}
body, err := io.ReadAll(io.LimitReader(r.Body, maxManifestBytes+1))
if err != nil {
reg.internal(w, "read manifest", err)
return
}
if len(body) > maxManifestBytes {
WriteError(w, http.StatusRequestEntityTooLarge, "MANIFEST_INVALID", "manifest too large")
return
}
contentType := r.Header.Get("Content-Type")
if contentType == "" {
contentType = detectManifestType(body)
}
digest := "sha256:" + hex.EncodeToString(sha256Sum(body))
if ecr.ValidateDigest(reference) && reference != digest {
WriteError(w, http.StatusBadRequest, "DIGEST_INVALID", "reference digest does not match content")
return
}
children, code, vErr := reg.validateManifest(name, contentType, body)
if vErr != nil {
WriteError(w, http.StatusBadRequest, code, vErr.Error())
return
}
if _, err := reg.Store.PutObject(&s3.PutObjectInput{
Bucket: aws.String(reg.bucket()),
Key: aws.String(ecr.ManifestKey(name, digest)),
Body: aws.ReadSeekCloser(strings.NewReader(string(body))),
ContentType: aws.String(contentType),
}); err != nil {
reg.internal(w, "store manifest", err)
return
}
if err := reg.Meta.PutManifestMeta(reg.AccountID, name, ecr.ManifestMeta{
Digest: digest,
MediaType: contentType,
Size: int64(len(body)),
PushedAt: time.Now().UTC(),
ChildDigests: children,
}); err != nil {
reg.internal(w, "store manifest meta", err)
return
}
if !ecr.ValidateDigest(reference) {
if err := reg.Meta.PutTag(reg.AccountID, name, reference, digest); err != nil {
reg.internal(w, "store tag", err)
digest, err := reg.StoreManifest(reg.AccountID, name, reference, r.Header.Get("Content-Type"), body)
if err != nil {
var mErr *ManifestStoreError
if errors.As(err, &mErr) {
WriteError(w, mErr.Status, mErr.Code, mErr.Msg)
return
}
reg.internal(w, "store manifest", err)
return
}
w.Header().Set("Docker-Content-Digest", digest)
+246
View File
@@ -0,0 +1,246 @@
package gateway_ecr
import (
"encoding/hex"
"errors"
"io"
"net/http"
"slices"
"strings"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/s3"
"github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"github.com/mulgadc/spinifex/spinifex/objectstore"
)
// ErrImageNotFound is returned by the image read/delete helpers when a tag or
// digest does not resolve to a stored manifest.
var ErrImageNotFound = errors.New("image not found")
// ManifestStoreError carries an OCI/HTTP status and code so the OCI PUT path can
// render the v2 error envelope while JSON PutImage maps it to an AWS error.
type ManifestStoreError struct {
Status int
Code string
Msg string
}
func (e *ManifestStoreError) Error() string { return e.Msg }
// ImageID pairs a manifest digest with an optional tag.
type ImageID struct {
Digest string
Tag string
}
// ImageRecord is a stored image's metadata, resolved with its tag set.
type ImageRecord struct {
Digest string
Tags []string
Size int64
PushedAt time.Time
MediaType string
}
// StoreManifest validates, stores and (for a tag reference) tags a manifest for
// the given account. It is the shared core of the OCI manifest PUT and the JSON
// PutImage action. ref is a tag or a digest; contentType may be empty, in which
// case it is detected. On a validation/size failure it returns a
// *ManifestStoreError carrying the OCI code.
func (reg *Registry) StoreManifest(account, repo, ref, contentType string, body []byte) (string, error) {
scoped := reg.forAccount(account)
if err := scoped.ensureRepo(repo); err != nil {
return "", err
}
if len(body) > maxManifestBytes {
return "", &ManifestStoreError{http.StatusRequestEntityTooLarge, "MANIFEST_INVALID", "manifest too large"}
}
if contentType == "" {
contentType = detectManifestType(body)
}
digest := "sha256:" + hex.EncodeToString(sha256Sum(body))
if ecr.ValidateDigest(ref) && ref != digest {
return "", &ManifestStoreError{http.StatusBadRequest, "DIGEST_INVALID", "reference digest does not match content"}
}
children, code, vErr := scoped.validateManifest(repo, contentType, body)
if vErr != nil {
return "", &ManifestStoreError{http.StatusBadRequest, code, vErr.Error()}
}
if _, err := scoped.Store.PutObject(&s3.PutObjectInput{
Bucket: aws.String(scoped.bucket()),
Key: aws.String(ecr.ManifestKey(repo, digest)),
Body: aws.ReadSeekCloser(strings.NewReader(string(body))),
ContentType: aws.String(contentType),
}); err != nil {
return "", err
}
if err := scoped.Meta.PutManifestMeta(account, repo, ecr.ManifestMeta{
Digest: digest,
MediaType: contentType,
Size: int64(len(body)),
PushedAt: time.Now().UTC(),
ChildDigests: children,
}); err != nil {
return "", err
}
if ref != "" && !ecr.ValidateDigest(ref) {
if err := scoped.Meta.PutTag(account, repo, ref, digest); err != nil {
return "", err
}
}
return digest, nil
}
// GetManifest returns the verbatim manifest bytes, media type and digest for a
// tag or digest reference. When acceptedTypes is non-empty the stored media type
// must match one of them; a mismatch yields ErrImageNotFound (Q14). A missing
// image yields ErrImageNotFound.
func (reg *Registry) GetManifest(account, repo, ref string, acceptedTypes []string) ([]byte, string, string, error) {
scoped := reg.forAccount(account)
digest, ok := scoped.resolveManifestDigest(repo, ref)
if !ok {
return nil, "", "", ErrImageNotFound
}
meta, err := scoped.Meta.GetManifestMeta(account, repo, digest)
if err != nil {
if errors.Is(err, ecr.ErrNotFound) {
return nil, "", "", ErrImageNotFound
}
return nil, "", "", err
}
if len(acceptedTypes) > 0 && !mediaTypeAccepted(acceptedTypes, meta.MediaType) {
return nil, "", "", ErrImageNotFound
}
out, err := scoped.Store.GetObject(&s3.GetObjectInput{
Bucket: aws.String(scoped.bucket()),
Key: aws.String(ecr.ManifestKey(repo, digest)),
})
if err != nil {
if objectstore.IsNoSuchKeyError(err) {
return nil, "", "", ErrImageNotFound
}
return nil, "", "", err
}
defer func() { _ = out.Body.Close() }()
bytes, err := io.ReadAll(out.Body)
if err != nil {
return nil, "", "", err
}
return bytes, meta.MediaType, digest, nil
}
// ListImages resolves every stored manifest in a repo with its tag set. It is
// the shared source for the ListImages and DescribeImages actions. A missing
// repo yields ecr.ErrNotFound.
func (reg *Registry) ListImages(account, repo string) ([]ImageRecord, error) {
scoped := reg.forAccount(account)
if _, err := scoped.Meta.GetRepo(account, repo); err != nil {
return nil, err
}
tags, err := scoped.Meta.ListTags(account, repo)
if err != nil {
return nil, err
}
tagsByDigest := make(map[string][]string)
for _, tag := range tags {
digest, err := scoped.Meta.GetTag(account, repo, tag)
if err != nil {
if errors.Is(err, ecr.ErrNotFound) {
continue
}
return nil, err
}
tagsByDigest[digest] = append(tagsByDigest[digest], tag)
}
// ListManifests yields lookup keys (the KV store tokenizes ':' to '-'); the
// canonical digest is the one stored in the manifest meta, which is what the
// tag reverse-index is keyed by. Resolve every record on meta.Digest so the
// digest form and tag association are correct regardless of store.
keys, err := scoped.Meta.ListManifests(account, repo)
if err != nil {
return nil, err
}
records := make([]ImageRecord, 0, len(keys))
for _, key := range keys {
meta, err := scoped.Meta.GetManifestMeta(account, repo, key)
if err != nil {
if errors.Is(err, ecr.ErrNotFound) {
continue
}
return nil, err
}
records = append(records, ImageRecord{
Digest: meta.Digest,
Tags: tagsByDigest[meta.Digest],
Size: meta.Size,
PushedAt: meta.PushedAt,
MediaType: meta.MediaType,
})
}
return records, nil
}
// DeleteImage removes an image by tag and/or digest. A tag deletes only that tag
// pointer; a digest deletes the manifest meta and every tag pointing at it. Blob
// and manifest-object GC is deferred (siv-361). The resolved digest is returned.
// A missing image yields ErrImageNotFound.
func (reg *Registry) DeleteImage(account, repo, tag, digest string) (string, error) {
scoped := reg.forAccount(account)
if tag != "" && digest == "" {
resolved, err := scoped.Meta.GetTag(account, repo, tag)
if err != nil {
if errors.Is(err, ecr.ErrNotFound) {
return "", ErrImageNotFound
}
return "", err
}
if err := scoped.Meta.DeleteTag(account, repo, tag); err != nil {
return "", err
}
return resolved, nil
}
if _, err := scoped.Meta.GetManifestMeta(account, repo, digest); err != nil {
if errors.Is(err, ecr.ErrNotFound) {
return "", ErrImageNotFound
}
return "", err
}
tags, err := scoped.Meta.ListTags(account, repo)
if err != nil {
return "", err
}
for _, t := range tags {
d, err := scoped.Meta.GetTag(account, repo, t)
if err != nil {
if errors.Is(err, ecr.ErrNotFound) {
continue
}
return "", err
}
if d == digest {
if err := scoped.Meta.DeleteTag(account, repo, t); err != nil {
return "", err
}
}
}
if err := scoped.Meta.DeleteManifestMeta(account, repo, digest); err != nil {
return "", err
}
return digest, nil
}
// mediaTypeAccepted reports whether mediaType is in the accepted set.
func mediaTypeAccepted(accepted []string, mediaType string) bool {
return slices.Contains(accepted, mediaType)
}
+176
View File
@@ -0,0 +1,176 @@
package gateway_ecr
import (
"errors"
"fmt"
"strings"
"testing"
"github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/mulgadc/spinifex/spinifex/testutil"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// seedImage pushes a config + layer blob and stores an image manifest tagged
// `tag`, returning the manifest bytes and digest.
func seedImage(t *testing.T, reg *Registry, repo, tag string) ([]byte, string) {
t.Helper()
cfgDg := pushBlob(t, reg, repo, []byte("cfg-"+repo+"-"+tag))
layerDg := pushBlob(t, reg, repo, []byte("layer-"+repo+"-"+tag))
manifest := fmt.Appendf(nil,
`{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[{"digest":"%s"}]}`,
mediaTypeDockerManifest, cfgDg, layerDg)
digest, err := reg.StoreManifest(testAccount, repo, tag, mediaTypeDockerManifest, manifest)
require.NoError(t, err)
return manifest, digest
}
func TestStoreAndGetManifest_RoundTrip(t *testing.T) {
reg := newTestRegistry()
manifest, digest := seedImage(t, reg, "team/app", "v1")
// Fetch by tag.
body, mediaType, gotDigest, err := reg.GetManifest(testAccount, "team/app", "v1", nil)
require.NoError(t, err)
assert.Equal(t, manifest, body)
assert.Equal(t, mediaTypeDockerManifest, mediaType)
assert.Equal(t, digest, gotDigest)
// Fetch by digest.
body, _, _, err = reg.GetManifest(testAccount, "team/app", digest, nil)
require.NoError(t, err)
assert.Equal(t, manifest, body)
// acceptedMediaTypes match.
_, _, _, err = reg.GetManifest(testAccount, "team/app", "v1", []string{mediaTypeDockerManifest})
require.NoError(t, err)
// acceptedMediaTypes mismatch -> ErrImageNotFound (Q14).
_, _, _, err = reg.GetManifest(testAccount, "team/app", "v1", []string{mediaTypeOCIIndex})
assert.ErrorIs(t, err, ErrImageNotFound)
// Unknown tag.
_, _, _, err = reg.GetManifest(testAccount, "team/app", "ghost", nil)
assert.ErrorIs(t, err, ErrImageNotFound)
}
func TestListImages_TaggedAndUntagged(t *testing.T) {
reg := newTestRegistry()
repo := "team/app"
seedImage(t, reg, repo, "v1")
seedImage(t, reg, repo, "v2")
// An untagged manifest: store with a digest reference so no tag is written.
manifest := fmt.Appendf(nil,
`{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`,
mediaTypeDockerManifest, pushBlob(t, reg, repo, []byte("cfg-untagged")))
untaggedDigest := digestOf(manifest)
_, err := reg.StoreManifest(testAccount, repo, untaggedDigest, mediaTypeDockerManifest, manifest)
require.NoError(t, err)
records, err := reg.ListImages(testAccount, repo)
require.NoError(t, err)
tags := map[string]bool{}
untagged := 0
for _, rec := range records {
if len(rec.Tags) == 0 {
untagged++
assert.Equal(t, untaggedDigest, rec.Digest)
continue
}
for _, tag := range rec.Tags {
tags[tag] = true
}
assert.Positive(t, rec.Size)
}
assert.True(t, tags["v1"])
assert.True(t, tags["v2"])
assert.Equal(t, 1, untagged)
// Missing repo -> ErrNotFound.
_, err = reg.ListImages(testAccount, "team/ghost")
assert.ErrorIs(t, err, ecr.ErrNotFound)
}
// TestListImages_KVDigestForm guards the production JetStream-KV path: its
// ListManifests returns tokenized keys (':'->'-'), so ListImages must
// canonicalize on meta.Digest to emit a real digest and keep tag association.
// The memory store returns real digests and would mask this.
func TestListImages_KVDigestForm(t *testing.T) {
_, _, js := testutil.StartTestJetStream(t)
reg := NewRegistry(objectstore.NewMemoryObjectStore(), ecr.NewKVMetaStore(js), testAccount)
repo := "team/app"
manifest := []byte(`{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.v2+json","layers":[]}`)
digest, err := reg.StoreManifest(testAccount, repo, "v1", mediaTypeDockerManifest, manifest)
require.NoError(t, err)
require.True(t, strings.HasPrefix(digest, "sha256:"))
records, err := reg.ListImages(testAccount, repo)
require.NoError(t, err)
require.Len(t, records, 1)
assert.Equal(t, digest, records[0].Digest, "digest must be the colon form, not the KV token")
assert.Equal(t, []string{"v1"}, records[0].Tags, "tag association must survive KV tokenization")
}
func TestDeleteImage_ByTagAndByDigest(t *testing.T) {
reg := newTestRegistry()
repo := "team/app"
_, digest := seedImage(t, reg, repo, "v1")
seedImage(t, reg, repo, "v2") // shares nothing; distinct digest
// Delete by tag removes only the tag pointer; the manifest stays.
got, err := reg.DeleteImage(testAccount, repo, "v1", "")
require.NoError(t, err)
assert.Equal(t, digest, got)
_, _, _, err = reg.GetManifest(testAccount, repo, "v1", nil)
assert.ErrorIs(t, err, ErrImageNotFound)
_, _, _, err = reg.GetManifest(testAccount, repo, digest, nil)
require.NoError(t, err) // manifest still resolvable by digest
// Delete by digest removes the manifest meta.
got, err = reg.DeleteImage(testAccount, repo, "", digest)
require.NoError(t, err)
assert.Equal(t, digest, got)
_, _, _, err = reg.GetManifest(testAccount, repo, digest, nil)
assert.ErrorIs(t, err, ErrImageNotFound)
// Deleting an absent tag/digest -> ErrImageNotFound.
_, err = reg.DeleteImage(testAccount, repo, "ghost", "")
assert.ErrorIs(t, err, ErrImageNotFound)
_, err = reg.DeleteImage(testAccount, repo, "", digestOf([]byte("absent")))
assert.ErrorIs(t, err, ErrImageNotFound)
}
func TestDeleteImage_ByDigestClearsTags(t *testing.T) {
reg := newTestRegistry()
repo := "team/app"
_, digest := seedImage(t, reg, repo, "v1")
// Add a second tag pointing at the same digest.
require.NoError(t, reg.Meta.PutTag(testAccount, repo, "latest", digest))
got, err := reg.DeleteImage(testAccount, repo, "", digest)
require.NoError(t, err)
assert.Equal(t, digest, got)
tags, err := reg.Meta.ListTags(testAccount, repo)
require.NoError(t, err)
assert.Empty(t, tags)
}
func TestStoreManifest_BadDigestReference(t *testing.T) {
reg := newTestRegistry()
repo := "team/app"
cfgDg := pushBlob(t, reg, repo, []byte("cfg"))
manifest := fmt.Appendf(nil,
`{"schemaVersion":2,"mediaType":"%s","config":{"digest":"%s"},"layers":[]}`,
mediaTypeDockerManifest, cfgDg)
_, err := reg.StoreManifest(testAccount, repo, digestOf([]byte("wrong")), mediaTypeDockerManifest, manifest)
var mErr *ManifestStoreError
require.True(t, errors.As(err, &mErr))
assert.Equal(t, "DIGEST_INVALID", mErr.Code)
}
+410
View File
@@ -0,0 +1,410 @@
package gateway
import (
"encoding/json"
"errors"
"io"
"log/slog"
"net/http"
"slices"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/ecr"
"github.com/mulgadc/spinifex/spinifex/awserrors"
gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr"
gateway_ecrapi "github.com/mulgadc/spinifex/spinifex/gateway/ecrapi"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
)
// maxImageBatch is the per-call cap on imageIds for the batch image actions.
const maxImageBatch = 100
// imageIdentifier is the AWS JSON 1.1 {imageDigest, imageTag} pair.
type imageIdentifier struct {
ImageDigest string `json:"imageDigest"`
ImageTag string `json:"imageTag"`
}
type tagStatusFilter struct {
TagStatus string `json:"tagStatus"`
}
type listImagesRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
Filter *tagStatusFilter `json:"filter"`
}
type describeImagesRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
ImageIds []imageIdentifier `json:"imageIds"`
Filter *tagStatusFilter `json:"filter"`
}
type batchGetImageRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
ImageIds []imageIdentifier `json:"imageIds"`
AcceptedMediaTypes []string `json:"acceptedMediaTypes"`
}
type putImageRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
ImageManifest string `json:"imageManifest"`
ImageManifestMediaType string `json:"imageManifestMediaType"`
ImageTag string `json:"imageTag"`
ImageDigest string `json:"imageDigest"`
}
type batchDeleteImageRequest struct {
RepositoryName string `json:"repositoryName"`
RegistryID string `json:"registryId"`
ImageIds []imageIdentifier `json:"imageIds"`
}
// ecrImageAccount reads the auth-context account and requires the OCI registry
// to be wired. Image actions need the gateway-side predastore Store, so a nil
// registry is a server fault.
func (gw *GatewayConfig) ecrImageAccount(r *http.Request) (string, error) {
accountID, _ := r.Context().Value(ctxAccountID).(string)
if accountID == "" {
slog.Error("ECR image action: no account ID in auth context")
return "", errors.New(awserrors.ErrorServerInternal)
}
if gw.ECRRegistry == nil {
slog.Error("ECR image action: OCI registry not configured")
return "", errors.New(awserrors.ErrorServerInternal)
}
return accountID, nil
}
// validateRepoAndRegistry rejects a malformed repository name and a registryId
// targeting another account.
func validateRepoAndRegistry(name, registryID, accountID string) error {
if err := handlers_ecr.ValidateRepoName(name); err != nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
if registryID != "" && registryID != accountID {
return errors.New(awserrors.ErrorAccessDenied)
}
return nil
}
func decodeJSONBody(r *http.Request, dst any) error {
body, err := io.ReadAll(r.Body)
if err != nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
if err := json.Unmarshal(body, dst); err != nil {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
return nil
}
// handleListImages returns every imageId in the repo, honoring an optional
// TAGGED/UNTAGGED filter. Tagged manifests yield one entry per tag; untagged
// manifests yield a digest-only entry.
func (gw *GatewayConfig) handleListImages(w http.ResponseWriter, r *http.Request) error {
accountID, err := gw.ecrImageAccount(r)
if err != nil {
return err
}
var req listImagesRequest
if err := decodeJSONBody(r, &req); err != nil {
return err
}
if err := validateRepoAndRegistry(req.RepositoryName, req.RegistryID, accountID); err != nil {
return err
}
records, err := gw.ecrListImages(accountID, req.RepositoryName)
if err != nil {
return err
}
status := ""
if req.Filter != nil {
status = req.Filter.TagStatus
}
ids := make([]*ecr.ImageIdentifier, 0, len(records))
for _, rec := range records {
if len(rec.Tags) == 0 {
if status == ecr.TagStatusTagged {
continue
}
ids = append(ids, &ecr.ImageIdentifier{ImageDigest: aws.String(rec.Digest)})
continue
}
if status == ecr.TagStatusUntagged {
continue
}
for _, tag := range rec.Tags {
ids = append(ids, &ecr.ImageIdentifier{ImageDigest: aws.String(rec.Digest), ImageTag: aws.String(tag)})
}
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.ListImagesOutput{ImageIds: ids})
return nil
}
// handleDescribeImages returns detailed metadata for the repo's images,
// optionally narrowed to the requested imageIds.
func (gw *GatewayConfig) handleDescribeImages(w http.ResponseWriter, r *http.Request) error {
accountID, err := gw.ecrImageAccount(r)
if err != nil {
return err
}
var req describeImagesRequest
if err := decodeJSONBody(r, &req); err != nil {
return err
}
if err := validateRepoAndRegistry(req.RepositoryName, req.RegistryID, accountID); err != nil {
return err
}
records, err := gw.ecrListImages(accountID, req.RepositoryName)
if err != nil {
return err
}
wanted := func(rec gateway_ecr.ImageRecord) bool {
if len(req.ImageIds) == 0 {
return true
}
for _, id := range req.ImageIds {
if id.ImageDigest != "" && id.ImageDigest == rec.Digest {
return true
}
if id.ImageTag != "" && slices.Contains(rec.Tags, id.ImageTag) {
return true
}
}
return false
}
details := make([]*ecr.ImageDetail, 0, len(records))
for _, rec := range records {
if !wanted(rec) {
continue
}
detail := &ecr.ImageDetail{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
ImageDigest: aws.String(rec.Digest),
ImageSizeInBytes: aws.Int64(rec.Size),
ImageManifestMediaType: aws.String(rec.MediaType),
}
if !rec.PushedAt.IsZero() {
detail.ImagePushedAt = aws.Time(rec.PushedAt)
}
for _, tag := range rec.Tags {
detail.ImageTags = append(detail.ImageTags, aws.String(tag))
}
details = append(details, detail)
}
if len(req.ImageIds) > 0 && len(details) == 0 {
return errors.New(awserrors.ErrorImageNotFound)
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.DescribeImagesOutput{ImageDetails: details})
return nil
}
// handleBatchGetImage returns verbatim manifest bytes for up to 100 imageIds.
// Per Q14 it answers HTTP 200 with a structured failures array on partial
// misses; the digest wins over the tag when both are supplied.
func (gw *GatewayConfig) handleBatchGetImage(w http.ResponseWriter, r *http.Request) error {
accountID, err := gw.ecrImageAccount(r)
if err != nil {
return err
}
var req batchGetImageRequest
if err := decodeJSONBody(r, &req); err != nil {
return err
}
if err := validateRepoAndRegistry(req.RepositoryName, req.RegistryID, accountID); err != nil {
return err
}
if len(req.ImageIds) > maxImageBatch {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
var images []*ecr.Image
var failures []*ecr.ImageFailure
for _, id := range req.ImageIds {
ref := id.ImageDigest
if ref == "" {
ref = id.ImageTag
}
if ref == "" {
failures = append(failures, imageFailure(id, ecr.ImageFailureCodeMissingDigestAndTag, "no digest or tag specified"))
continue
}
body, mediaType, digest, gErr := gw.ECRRegistry.GetManifest(accountID, req.RepositoryName, ref, req.AcceptedMediaTypes)
if errors.Is(gErr, gateway_ecr.ErrImageNotFound) {
failures = append(failures, imageFailure(id, ecr.ImageFailureCodeImageNotFound, "image not found"))
continue
}
if gErr != nil {
slog.Error("BatchGetImage: get manifest failed", "repo", req.RepositoryName, "err", gErr)
return errors.New(awserrors.ErrorServerInternal)
}
out := &ecr.Image{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
ImageId: &ecr.ImageIdentifier{ImageDigest: aws.String(digest)},
ImageManifest: aws.String(string(body)),
ImageManifestMediaType: aws.String(mediaType),
}
if id.ImageTag != "" {
out.ImageId.ImageTag = aws.String(id.ImageTag)
}
images = append(images, out)
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.BatchGetImageOutput{Images: images, Failures: failures})
return nil
}
// handlePutImage stores a manifest (and tag) supplied as JSON, the control-plane
// twin of an OCI manifest PUT. Used by `aws ecr put-image` to re-tag or copy an
// existing manifest.
func (gw *GatewayConfig) handlePutImage(w http.ResponseWriter, r *http.Request) error {
accountID, err := gw.ecrImageAccount(r)
if err != nil {
return err
}
var req putImageRequest
if err := decodeJSONBody(r, &req); err != nil {
return err
}
if err := validateRepoAndRegistry(req.RepositoryName, req.RegistryID, accountID); err != nil {
return err
}
if req.ImageManifest == "" {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
ref := req.ImageTag
if ref == "" {
ref = req.ImageDigest
}
digest, sErr := gw.ECRRegistry.StoreManifest(accountID, req.RepositoryName, ref, req.ImageManifestMediaType, []byte(req.ImageManifest))
if sErr != nil {
return mapStoreManifestError(sErr, req.RepositoryName)
}
image := &ecr.Image{
RegistryId: aws.String(accountID),
RepositoryName: aws.String(req.RepositoryName),
ImageId: &ecr.ImageIdentifier{ImageDigest: aws.String(digest)},
ImageManifest: aws.String(req.ImageManifest),
ImageManifestMediaType: aws.String(req.ImageManifestMediaType),
}
if req.ImageTag != "" {
image.ImageId.ImageTag = aws.String(req.ImageTag)
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.PutImageOutput{Image: image})
return nil
}
// handleBatchDeleteImage deletes images by tag or digest. A tag removes only
// that tag pointer; a digest removes the manifest and every tag pointing at it.
// Partial failures are reported in the failures array with HTTP 200.
func (gw *GatewayConfig) handleBatchDeleteImage(w http.ResponseWriter, r *http.Request) error {
accountID, err := gw.ecrImageAccount(r)
if err != nil {
return err
}
var req batchDeleteImageRequest
if err := decodeJSONBody(r, &req); err != nil {
return err
}
if err := validateRepoAndRegistry(req.RepositoryName, req.RegistryID, accountID); err != nil {
return err
}
if len(req.ImageIds) > maxImageBatch {
return errors.New(awserrors.ErrorInvalidParameterValue)
}
var deleted []*ecr.ImageIdentifier
var failures []*ecr.ImageFailure
for _, id := range req.ImageIds {
if id.ImageDigest == "" && id.ImageTag == "" {
failures = append(failures, imageFailure(id, ecr.ImageFailureCodeMissingDigestAndTag, "no digest or tag specified"))
continue
}
digest, dErr := gw.ECRRegistry.DeleteImage(accountID, req.RepositoryName, id.ImageTag, id.ImageDigest)
if errors.Is(dErr, gateway_ecr.ErrImageNotFound) {
failures = append(failures, imageFailure(id, ecr.ImageFailureCodeImageNotFound, "image not found"))
continue
}
if dErr != nil {
slog.Error("BatchDeleteImage: delete failed", "repo", req.RepositoryName, "err", dErr)
return errors.New(awserrors.ErrorServerInternal)
}
out := &ecr.ImageIdentifier{ImageDigest: aws.String(digest)}
if id.ImageTag != "" {
out.ImageTag = aws.String(id.ImageTag)
}
deleted = append(deleted, out)
}
gateway_ecrapi.WriteJSONResponse(w, &ecr.BatchDeleteImageOutput{ImageIds: deleted, Failures: failures})
return nil
}
// ecrListImages resolves the repo's image records, mapping a missing repo to
// RepositoryNotFound and any other backend fault to ServerInternal.
func (gw *GatewayConfig) ecrListImages(accountID, repo string) ([]gateway_ecr.ImageRecord, error) {
records, err := gw.ECRRegistry.ListImages(accountID, repo)
if errors.Is(err, handlers_ecr.ErrNotFound) {
return nil, errors.New(awserrors.ErrorRepositoryNotFound)
}
if err != nil {
slog.Error("ECR image action: list images failed", "repo", repo, "err", err)
return nil, errors.New(awserrors.ErrorServerInternal)
}
return records, nil
}
// mapStoreManifestError translates the OCI manifest-store error codes into AWS
// PutImage error codes.
func mapStoreManifestError(err error, repo string) error {
var mErr *gateway_ecr.ManifestStoreError
if errors.As(err, &mErr) {
switch mErr.Code {
case "DIGEST_INVALID":
return errors.New(awserrors.ErrorImageDigestDoesNotMatch)
case "MANIFEST_BLOB_UNKNOWN":
return errors.New(awserrors.ErrorLayersNotFound)
default:
return errors.New(awserrors.ErrorInvalidParameterValue)
}
}
slog.Error("PutImage: store manifest failed", "repo", repo, "err", err)
return errors.New(awserrors.ErrorServerInternal)
}
func imageFailure(id imageIdentifier, code, reason string) *ecr.ImageFailure {
failure := &ecr.ImageFailure{
FailureCode: aws.String(code),
FailureReason: aws.String(reason),
ImageId: &ecr.ImageIdentifier{},
}
if id.ImageDigest != "" {
failure.ImageId.ImageDigest = aws.String(id.ImageDigest)
}
if id.ImageTag != "" {
failure.ImageId.ImageTag = aws.String(id.ImageTag)
}
return failure
}
+216
View File
@@ -0,0 +1,216 @@
package gateway
import (
"context"
"encoding/json"
"fmt"
"net/http"
"net/http/httptest"
"strings"
"testing"
"github.com/aws/aws-sdk-go/service/ecr"
gateway_ecr "github.com/mulgadc/spinifex/spinifex/gateway/ecr"
handlers_ecr "github.com/mulgadc/spinifex/spinifex/handlers/ecr"
"github.com/mulgadc/spinifex/spinifex/objectstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// newImageGateway wires a GatewayConfig with an in-memory OCI registry (memory
// object store + memory meta) so the inline image handlers exercise the real
// Registry helpers without NATS.
func newImageGateway(t *testing.T) *GatewayConfig {
t.Helper()
reg := gateway_ecr.NewRegistry(objectstore.NewMemoryObjectStore(), handlers_ecr.NewMemoryMetaStore(), ecrTestAccount)
return &GatewayConfig{ECRRegistry: reg, Region: ecrTestRegion, InternalSuffix: ecrTestSuffix, DisableLogging: true}
}
// seedTaggedImage stores a layerless manifest under repo:tag, returning its
// digest. A unique manifest per tag yields a unique digest.
func seedTaggedImage(t *testing.T, gw *GatewayConfig, repo, tag string) string {
t.Helper()
manifest := fmt.Appendf(nil,
`{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.v2+json","layers":[],"annotations":{"tag":"%s"}}`, tag)
digest, err := gw.ECRRegistry.StoreManifest(ecrTestAccount, repo, tag, "application/vnd.docker.distribution.manifest.v2+json", manifest)
require.NoError(t, err)
return digest
}
func imageReq(t *testing.T, body string) *http.Request {
t.Helper()
req := httptest.NewRequest(http.MethodPost, "/", strings.NewReader(body))
return req.WithContext(context.WithValue(req.Context(), ctxAccountID, ecrTestAccount))
}
func callImage(t *testing.T, gw *GatewayConfig, h func(*GatewayConfig, http.ResponseWriter, *http.Request) error, body string) (*httptest.ResponseRecorder, error) {
t.Helper()
w := httptest.NewRecorder()
return w, h(gw, w, imageReq(t, body))
}
func TestListImages_TaggedUntaggedFilter(t *testing.T) {
gw := newImageGateway(t)
seedTaggedImage(t, gw, "team/app", "v1")
seedTaggedImage(t, gw, "team/app", "v2")
w, err := callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/app"}`)
require.NoError(t, err)
require.Equal(t, http.StatusOK, w.Code)
var out ecr.ListImagesOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
assert.Len(t, out.ImageIds, 2)
// TAGGED filter keeps both; UNTAGGED filter drops both.
w, err = callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/app","filter":{"tagStatus":"UNTAGGED"}}`)
require.NoError(t, err)
var untagged ecr.ListImagesOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &untagged))
assert.Empty(t, untagged.ImageIds)
}
func TestListImages_MissingRepo(t *testing.T) {
gw := newImageGateway(t)
_, err := callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/ghost"}`)
require.Error(t, err)
assert.Equal(t, "RepositoryNotFoundException", err.Error())
}
func TestDescribeImages_HappyAndNotFound(t *testing.T) {
gw := newImageGateway(t)
digest := seedTaggedImage(t, gw, "team/app", "v1")
w, err := callImage(t, gw, (*GatewayConfig).handleDescribeImages, `{"repositoryName":"team/app"}`)
require.NoError(t, err)
// AWS jsonutil emits imagePushedAt as an epoch float, which encoding/json
// cannot decode into the SDK struct's *time.Time, so assert via a local view.
var out struct {
ImageDetails []struct {
ImageDigest string `json:"imageDigest"`
ImageTags []string `json:"imageTags"`
ImageSizeInBytes int64 `json:"imageSizeInBytes"`
ImagePushedAt float64 `json:"imagePushedAt"`
} `json:"imageDetails"`
}
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
require.Len(t, out.ImageDetails, 1)
assert.Equal(t, digest, out.ImageDetails[0].ImageDigest)
assert.Equal(t, []string{"v1"}, out.ImageDetails[0].ImageTags)
assert.Positive(t, out.ImageDetails[0].ImageSizeInBytes)
assert.Positive(t, out.ImageDetails[0].ImagePushedAt)
// Asking for an absent imageId -> ImageNotFound.
_, err = callImage(t, gw, (*GatewayConfig).handleDescribeImages, `{"repositoryName":"team/app","imageIds":[{"imageTag":"ghost"}]}`)
require.Error(t, err)
assert.Equal(t, "ImageNotFoundException", err.Error())
}
func TestBatchGetImage_PartialAndDigestWins(t *testing.T) {
gw := newImageGateway(t)
digest := seedTaggedImage(t, gw, "team/app", "v1")
body := fmt.Sprintf(`{"repositoryName":"team/app","imageIds":[{"imageDigest":"%s","imageTag":"v1"},{"imageTag":"ghost"},{}]}`, digest)
w, err := callImage(t, gw, (*GatewayConfig).handleBatchGetImage, body)
require.NoError(t, err)
require.Equal(t, http.StatusOK, w.Code)
var out ecr.BatchGetImageOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
require.Len(t, out.Images, 1)
assert.Equal(t, digest, *out.Images[0].ImageId.ImageDigest)
assert.NotEmpty(t, *out.Images[0].ImageManifest)
require.Len(t, out.Failures, 2)
codes := map[string]bool{}
for _, f := range out.Failures {
codes[*f.FailureCode] = true
}
assert.True(t, codes["ImageNotFound"])
assert.True(t, codes["MissingDigestAndTag"])
}
func TestBatchGetImage_CapExceeded(t *testing.T) {
gw := newImageGateway(t)
ids := make([]string, 101)
for i := range ids {
ids[i] = `{"imageTag":"t"}`
}
body := `{"repositoryName":"team/app","imageIds":[` + strings.Join(ids, ",") + `]}`
_, err := callImage(t, gw, (*GatewayConfig).handleBatchGetImage, body)
require.Error(t, err)
assert.Equal(t, "InvalidParameterValue", err.Error())
}
func TestPutImage_HappyAndMissingManifest(t *testing.T) {
gw := newImageGateway(t)
manifest := `{"schemaVersion":2,"mediaType":"application/vnd.docker.distribution.manifest.v2+json","layers":[]}`
body, _ := json.Marshal(map[string]string{
"repositoryName": "team/app",
"imageManifest": manifest,
"imageManifestMediaType": "application/vnd.docker.distribution.manifest.v2+json",
"imageTag": "v1",
})
w, err := callImage(t, gw, (*GatewayConfig).handlePutImage, string(body))
require.NoError(t, err)
require.Equal(t, http.StatusOK, w.Code)
var out ecr.PutImageOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
assert.Equal(t, "v1", *out.Image.ImageId.ImageTag)
assert.True(t, strings.HasPrefix(*out.Image.ImageId.ImageDigest, "sha256:"))
// It is now listable.
w, err = callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/app"}`)
require.NoError(t, err)
var list ecr.ListImagesOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &list))
assert.Len(t, list.ImageIds, 1)
// Missing manifest -> InvalidParameterValue.
_, err = callImage(t, gw, (*GatewayConfig).handlePutImage, `{"repositoryName":"team/app","imageTag":"v2"}`)
require.Error(t, err)
assert.Equal(t, "InvalidParameterValue", err.Error())
}
func TestBatchDeleteImage_Partial(t *testing.T) {
gw := newImageGateway(t)
digest := seedTaggedImage(t, gw, "team/app", "v1")
body := fmt.Sprintf(`{"repositoryName":"team/app","imageIds":[{"imageDigest":"%s"},{"imageTag":"ghost"},{}]}`, digest)
w, err := callImage(t, gw, (*GatewayConfig).handleBatchDeleteImage, body)
require.NoError(t, err)
require.Equal(t, http.StatusOK, w.Code)
var out ecr.BatchDeleteImageOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &out))
require.Len(t, out.ImageIds, 1)
assert.Equal(t, digest, *out.ImageIds[0].ImageDigest)
require.Len(t, out.Failures, 2)
// The image is gone.
w, err = callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/app"}`)
require.NoError(t, err)
var list ecr.ListImagesOutput
require.NoError(t, json.Unmarshal(w.Body.Bytes(), &list))
assert.Empty(t, list.ImageIds)
}
func TestImageHandlers_GuardRails(t *testing.T) {
gw := newImageGateway(t)
// Cross-account registryId -> AccessDenied.
_, err := callImage(t, gw, (*GatewayConfig).handleListImages, `{"repositoryName":"team/app","registryId":"999999999999"}`)
require.Error(t, err)
assert.Equal(t, "AccessDenied", err.Error())
// No auth account -> ServerInternal.
w := httptest.NewRecorder()
err = gw.handleListImages(w, httptest.NewRequest(http.MethodPost, "/", strings.NewReader(`{"repositoryName":"team/app"}`)))
require.Error(t, err)
assert.Equal(t, "ServerInternal", err.Error())
// Nil registry -> ServerInternal.
bare := &GatewayConfig{DisableLogging: true}
err = bare.handleListImages(httptest.NewRecorder(), imageReq(t, `{"repositoryName":"team/app"}`))
require.Error(t, err)
assert.Equal(t, "ServerInternal", err.Error())
}
+3 -3
View File
@@ -60,12 +60,12 @@ func TestECRRequest_UnknownAction(t *testing.T) {
}
// A registered-but-unimplemented action resolves to the 501 stub until its
// handler lands. CreateRepository/DeleteRepository/DescribeRepositories are now
// served inline, so ListImages stands in as a still-stubbed action.
// handler lands. The repo/image actions are now served inline, so
// ListRepositories stands in as a still-stubbed action.
func TestECRRequest_KnownActionNotImplemented(t *testing.T) {
gw := &GatewayConfig{DisableLogging: true}
err := gw.ECR_Request(httptest.NewRecorder(),
setupECRRequest("AmazonEC2ContainerRegistry_V20150921.ListImages", "{}"))
setupECRRequest("AmazonEC2ContainerRegistry_V20150921.ListRepositories", "{}"))
require.Error(t, err)
assert.Equal(t, awserrors.ErrorNotImplemented, err.Error())
}
+16
View File
@@ -61,6 +61,22 @@ func (gw *GatewayConfig) ECR_Request(w http.ResponseWriter, r *http.Request) err
return gw.handleDeleteRepository(w, r)
}
// Image-management actions need the gateway-side predastore Store (manifest
// bytes) carried by the OCI Registry, which the relayed Handler signature
// does not reach, so they are served inline.
switch action {
case "ListImages":
return gw.handleListImages(w, r)
case "DescribeImages":
return gw.handleDescribeImages(w, r)
case "BatchGetImage":
return gw.handleBatchGetImage(w, r)
case "PutImage":
return gw.handlePutImage(w, r)
case "BatchDeleteImage":
return gw.handleBatchDeleteImage(w, r)
}
accountID, _ := r.Context().Value(ctxAccountID).(string)
if accountID == "" {
slog.Error("ECR_Request: no account ID in auth context")
+23
View File
@@ -74,6 +74,7 @@ type MetaStore interface {
PutManifestMeta(accountID, repo string, meta ManifestMeta) error
GetManifestMeta(accountID, repo, digest string) (ManifestMeta, error)
DeleteManifestMeta(accountID, repo, digest string) error
PutUpload(accountID, uploadID string, state UploadState) (uint64, error)
GetUpload(accountID, uploadID string) (UploadState, uint64, error)
@@ -348,6 +349,17 @@ func (s *KVMetaStore) GetManifestMeta(accountID, repo, digest string) (ManifestM
return m, nil
}
func (s *KVMetaStore) DeleteManifestMeta(accountID, repo, digest string) error {
kv, err := s.bucket(accountID)
if err != nil {
return err
}
if _, err := kv.Get(KVManifestKey(repo, digest)); err != nil {
return mapKVErr(err)
}
return kv.Delete(KVManifestKey(repo, digest))
}
func (s *KVMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) {
kv, err := s.bucket(accountID)
if err != nil {
@@ -600,6 +612,17 @@ func (m *MemoryMetaStore) GetManifestMeta(accountID, repo, digest string) (Manif
return meta, nil
}
func (m *MemoryMetaStore) DeleteManifestMeta(accountID, repo, digest string) error {
m.mu.Lock()
defer m.mu.Unlock()
key := repo + "|" + digest
if _, ok := m.manifests[accountID][key]; !ok {
return ErrNotFound
}
delete(m.manifests[accountID], key)
return nil
}
func (m *MemoryMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) {
m.mu.Lock()
defer m.mu.Unlock()
+5
View File
@@ -53,6 +53,11 @@ func TestKVMetaStore_FullCycle(t *testing.T) {
_, err = store.GetManifestMeta(acct, "team/app", "sha256:zzz")
assert.ErrorIs(t, err, ErrNotFound)
require.NoError(t, store.DeleteManifestMeta(acct, "team/app", "sha256:ccc"))
_, err = store.GetManifestMeta(acct, "team/app", "sha256:ccc")
assert.ErrorIs(t, err, ErrNotFound)
assert.ErrorIs(t, store.DeleteManifestMeta(acct, "team/app", "sha256:zzz"), ErrNotFound)
// Upload CAS lifecycle.
rev, err := store.PutUpload(acct, "u1", UploadState{RepoName: "team/app"})
require.NoError(t, err)
+11
View File
@@ -145,6 +145,17 @@ func (s *NATSMetaStore) GetManifestMeta(accountID, repo, digest string) (Manifes
return resp.Meta, nil
}
func (s *NATSMetaStore) DeleteManifestMeta(accountID, repo, digest string) error {
resp, err := utils.NATSRequest[ManifestDeleteResponse](s.conn, SubjectManifestDelete, &ManifestDeleteRequest{Repo: repo, Digest: digest}, metaRequestTimeout, accountID)
if err != nil {
return err
}
if !resp.Found {
return ErrNotFound
}
return nil
}
func (s *NATSMetaStore) PutUpload(accountID, uploadID string, state UploadState) (uint64, error) {
resp, err := utils.NATSRequest[UploadCreateResponse](s.conn, SubjectUploadCreate, &UploadCreateRequest{UploadID: uploadID, State: state}, metaRequestTimeout, accountID)
if err != nil {
+5
View File
@@ -82,6 +82,11 @@ func TestMemoryMetaStore_TagsAndManifests(t *testing.T) {
assert.Equal(t, int64(7), mm.Size)
_, err = m.GetManifestMeta(acct, "r", "sha256:zzz")
assert.ErrorIs(t, err, ErrNotFound)
require.NoError(t, m.DeleteManifestMeta(acct, "r", "sha256:ccc"))
_, err = m.GetManifestMeta(acct, "r", "sha256:ccc")
assert.ErrorIs(t, err, ErrNotFound)
assert.ErrorIs(t, m.DeleteManifestMeta(acct, "r", "sha256:ccc"), ErrNotFound)
}
func TestMemoryMetaStore_UploadCAS(t *testing.T) {
+11
View File
@@ -21,6 +21,7 @@ const (
SubjectManifestPut = "ecr.manifest.put"
SubjectManifestDescribe = "ecr.manifest.describe"
SubjectManifestList = "ecr.manifest.list"
SubjectManifestDelete = "ecr.manifest.delete"
SubjectUploadCreate = "ecr.upload.create"
SubjectUploadGet = "ecr.upload.get"
@@ -146,6 +147,15 @@ type ManifestListResponse struct {
Digests []string `json:"digests"`
}
type ManifestDeleteRequest struct {
Repo string `json:"repo"`
Digest string `json:"digest"`
}
type ManifestDeleteResponse struct {
Found bool `json:"found"`
}
type UploadCreateRequest struct {
UploadID string `json:"uploadID"`
State UploadState `json:"state"`
@@ -207,6 +217,7 @@ type MetaService interface {
ManifestPut(req *ManifestPutRequest, accountID string) (*ManifestPutResponse, error)
ManifestDescribe(req *ManifestDescribeRequest, accountID string) (*ManifestDescribeResponse, error)
ManifestList(req *ManifestListRequest, accountID string) (*ManifestListResponse, error)
ManifestDelete(req *ManifestDeleteRequest, accountID string) (*ManifestDeleteResponse, error)
UploadCreate(req *UploadCreateRequest, accountID string) (*UploadCreateResponse, error)
UploadGet(req *UploadGetRequest, accountID string) (*UploadGetResponse, error)
+11
View File
@@ -144,6 +144,17 @@ func (s *MetaServiceImpl) ManifestList(req *ManifestListRequest, accountID strin
return &ManifestListResponse{Digests: digests}, nil
}
func (s *MetaServiceImpl) ManifestDelete(req *ManifestDeleteRequest, accountID string) (*ManifestDeleteResponse, error) {
err := s.store.DeleteManifestMeta(accountID, req.Repo, req.Digest)
if errors.Is(err, ErrNotFound) {
return &ManifestDeleteResponse{Found: false}, nil
}
if err != nil {
return nil, err
}
return &ManifestDeleteResponse{Found: true}, nil
}
func (s *MetaServiceImpl) ManifestDescribe(req *ManifestDescribeRequest, accountID string) (*ManifestDescribeResponse, error) {
meta, err := s.store.GetManifestMeta(accountID, req.Repo, req.Digest)
if errors.Is(err, ErrNotFound) {