Propogate tag as a functional argument into the notification system to attach
tags to manifest push and pull event notifications. Signed-off-by: Richard Scothern <richard.scothern@gmail.com>
This commit is contained in:
parent
ec6ac0c05e
commit
afe2bdd1c5
9 changed files with 100 additions and 37 deletions
|
@ -53,12 +53,34 @@ func NewRequestRecord(id string, r *http.Request) RequestRecord {
|
|||
}
|
||||
}
|
||||
|
||||
func (b *bridge) ManifestPushed(repo reference.Named, sm distribution.Manifest) error {
|
||||
return b.createManifestEventAndWrite(EventActionPush, repo, sm)
|
||||
func (b *bridge) ManifestPushed(repo reference.Named, sm distribution.Manifest, options ...distribution.ManifestServiceOption) error {
|
||||
manifestEvent, err := b.createManifestEvent(EventActionPush, repo, sm)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, option := range options {
|
||||
if opt, ok := option.(distribution.WithTagOption); ok {
|
||||
manifestEvent.Target.Tag = opt.Tag
|
||||
break
|
||||
}
|
||||
}
|
||||
return b.sink.Write(*manifestEvent)
|
||||
}
|
||||
|
||||
func (b *bridge) ManifestPulled(repo reference.Named, sm distribution.Manifest) error {
|
||||
return b.createManifestEventAndWrite(EventActionPull, repo, sm)
|
||||
func (b *bridge) ManifestPulled(repo reference.Named, sm distribution.Manifest, options ...distribution.ManifestServiceOption) error {
|
||||
manifestEvent, err := b.createManifestEvent(EventActionPull, repo, sm)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, option := range options {
|
||||
if opt, ok := option.(distribution.WithTagOption); ok {
|
||||
manifestEvent.Target.Tag = opt.Tag
|
||||
break
|
||||
}
|
||||
}
|
||||
return b.sink.Write(*manifestEvent)
|
||||
}
|
||||
|
||||
func (b *bridge) ManifestDeleted(repo reference.Named, dgst digest.Digest) error {
|
||||
|
|
|
@ -3,6 +3,7 @@ package notifications
|
|||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/docker/distribution"
|
||||
"github.com/docker/distribution/digest"
|
||||
"github.com/docker/distribution/manifest/schema1"
|
||||
"github.com/docker/distribution/reference"
|
||||
|
@ -61,6 +62,38 @@ func TestEventBridgeManifestPushed(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestEventBridgeManifestPushedWithTag(t *testing.T) {
|
||||
l := createTestEnv(t, testSinkFn(func(events ...Event) error {
|
||||
checkCommonManifest(t, EventActionPush, events...)
|
||||
if events[0].Target.Tag != "latest" {
|
||||
t.Fatalf("missing or unexpected tag: %#v", events[0].Target)
|
||||
}
|
||||
|
||||
return nil
|
||||
}))
|
||||
|
||||
repoRef, _ := reference.ParseNamed(repo)
|
||||
if err := l.ManifestPushed(repoRef, sm, distribution.WithTag(m.Tag)); err != nil {
|
||||
t.Fatalf("unexpected error notifying manifest pull: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventBridgeManifestPulledWithTag(t *testing.T) {
|
||||
l := createTestEnv(t, testSinkFn(func(events ...Event) error {
|
||||
checkCommonManifest(t, EventActionPull, events...)
|
||||
if events[0].Target.Tag != "latest" {
|
||||
t.Fatalf("missing or unexpected tag: %#v", events[0].Target)
|
||||
}
|
||||
|
||||
return nil
|
||||
}))
|
||||
|
||||
repoRef, _ := reference.ParseNamed(repo)
|
||||
if err := l.ManifestPulled(repoRef, sm, distribution.WithTag(m.Tag)); err != nil {
|
||||
t.Fatalf("unexpected error notifying manifest pull: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEventBridgeManifestDeleted(t *testing.T) {
|
||||
l := createTestEnv(t, testSinkFn(func(events ...Event) error {
|
||||
checkDeleted(t, EventActionDelete, events...)
|
||||
|
|
|
@ -68,6 +68,9 @@ type Event struct {
|
|||
|
||||
// URL provides a direct link to the content.
|
||||
URL string `json:"url,omitempty"`
|
||||
|
||||
// Tag provides the tag
|
||||
Tag string `json:"tag,omitempty"`
|
||||
} `json:"target,omitempty"`
|
||||
|
||||
// Request covers the request that generated the event.
|
||||
|
|
|
@ -12,8 +12,8 @@ import (
|
|||
|
||||
// ManifestListener describes a set of methods for listening to events related to manifests.
|
||||
type ManifestListener interface {
|
||||
ManifestPushed(repo reference.Named, sm distribution.Manifest) error
|
||||
ManifestPulled(repo reference.Named, sm distribution.Manifest) error
|
||||
ManifestPushed(repo reference.Named, sm distribution.Manifest, options ...distribution.ManifestServiceOption) error
|
||||
ManifestPulled(repo reference.Named, sm distribution.Manifest, options ...distribution.ManifestServiceOption) error
|
||||
ManifestDeleted(repo reference.Named, dgst digest.Digest) error
|
||||
}
|
||||
|
||||
|
@ -81,7 +81,7 @@ func (msl *manifestServiceListener) Delete(ctx context.Context, dgst digest.Dige
|
|||
func (msl *manifestServiceListener) Get(ctx context.Context, dgst digest.Digest, options ...distribution.ManifestServiceOption) (distribution.Manifest, error) {
|
||||
sm, err := msl.ManifestService.Get(ctx, dgst)
|
||||
if err == nil {
|
||||
if err := msl.parent.listener.ManifestPulled(msl.parent.Repository.Named(), sm); err != nil {
|
||||
if err := msl.parent.listener.ManifestPulled(msl.parent.Repository.Named(), sm, options...); err != nil {
|
||||
logrus.Errorf("error dispatching manifest pull to listener: %v", err)
|
||||
}
|
||||
}
|
||||
|
@ -93,7 +93,7 @@ func (msl *manifestServiceListener) Put(ctx context.Context, sm distribution.Man
|
|||
dgst, err := msl.ManifestService.Put(ctx, sm, options...)
|
||||
|
||||
if err == nil {
|
||||
if err := msl.parent.listener.ManifestPushed(msl.parent.Repository.Named(), sm); err != nil {
|
||||
if err := msl.parent.listener.ManifestPushed(msl.parent.Repository.Named(), sm, options...); err != nil {
|
||||
logrus.Errorf("error dispatching manifest push to listener: %v", err)
|
||||
}
|
||||
}
|
||||
|
|
|
@ -44,7 +44,7 @@ func TestListener(t *testing.T) {
|
|||
"manifest:delete": 1,
|
||||
"layer:push": 2,
|
||||
"layer:pull": 2,
|
||||
"layer:delete": 2, // deletes not supported for now
|
||||
"layer:delete": 2,
|
||||
}
|
||||
|
||||
if !reflect.DeepEqual(tl.ops, expectedOps) {
|
||||
|
@ -57,13 +57,13 @@ type testListener struct {
|
|||
ops map[string]int
|
||||
}
|
||||
|
||||
func (tl *testListener) ManifestPushed(repo reference.Named, m distribution.Manifest) error {
|
||||
func (tl *testListener) ManifestPushed(repo reference.Named, m distribution.Manifest, options ...distribution.ManifestServiceOption) error {
|
||||
tl.ops["manifest:push"]++
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tl *testListener) ManifestPulled(repo reference.Named, m distribution.Manifest) error {
|
||||
func (tl *testListener) ManifestPulled(repo reference.Named, m distribution.Manifest, options ...distribution.ManifestServiceOption) error {
|
||||
tl.ops["manifest:pull"]++
|
||||
return nil
|
||||
}
|
||||
|
|
12
registry.go
12
registry.go
|
@ -58,6 +58,18 @@ type ManifestServiceOption interface {
|
|||
Apply(ManifestService) error
|
||||
}
|
||||
|
||||
// WithTag allows a tag to be passed into Put
|
||||
func WithTag(tag string) ManifestServiceOption {
|
||||
return WithTagOption{tag}
|
||||
}
|
||||
|
||||
type WithTagOption struct{ Tag string }
|
||||
|
||||
func (o WithTagOption) Apply(m ManifestService) error {
|
||||
// no implementation
|
||||
return nil
|
||||
}
|
||||
|
||||
// Repository is a named collection of manifests and layers.
|
||||
type Repository interface {
|
||||
// Named returns the name of the repository.
|
||||
|
|
|
@ -402,9 +402,9 @@ func (ms *manifests) Get(ctx context.Context, dgst digest.Digest, options ...dis
|
|||
)
|
||||
|
||||
for _, option := range options {
|
||||
if opt, ok := option.(withTagOption); ok {
|
||||
digestOrTag = opt.tag
|
||||
ref, err = reference.WithTag(ms.name, opt.tag)
|
||||
if opt, ok := option.(distribution.WithTagOption); ok {
|
||||
digestOrTag = opt.Tag
|
||||
ref, err = reference.WithTag(ms.name, opt.Tag)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
@ -465,21 +465,6 @@ func (ms *manifests) Get(ctx context.Context, dgst digest.Digest, options ...dis
|
|||
return nil, HandleErrorResponse(resp)
|
||||
}
|
||||
|
||||
// WithTag allows a tag to be passed into Put which enables the client
|
||||
// to build a correct URL.
|
||||
func WithTag(tag string) distribution.ManifestServiceOption {
|
||||
return withTagOption{tag}
|
||||
}
|
||||
|
||||
type withTagOption struct{ tag string }
|
||||
|
||||
func (o withTagOption) Apply(m distribution.ManifestService) error {
|
||||
if _, ok := m.(*manifests); ok {
|
||||
return nil
|
||||
}
|
||||
return fmt.Errorf("withTagOption is a client-only option")
|
||||
}
|
||||
|
||||
// Put puts a manifest. A tag can be specified using an options parameter which uses some shared state to hold the
|
||||
// tag name in order to build the correct upload URL.
|
||||
func (ms *manifests) Put(ctx context.Context, m distribution.Manifest, options ...distribution.ManifestServiceOption) (digest.Digest, error) {
|
||||
|
@ -487,9 +472,9 @@ func (ms *manifests) Put(ctx context.Context, m distribution.Manifest, options .
|
|||
var tagged bool
|
||||
|
||||
for _, option := range options {
|
||||
if opt, ok := option.(withTagOption); ok {
|
||||
if opt, ok := option.(distribution.WithTagOption); ok {
|
||||
var err error
|
||||
ref, err = reference.WithTag(ref, opt.tag)
|
||||
ref, err = reference.WithTag(ref, opt.Tag)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
|
|
@ -710,7 +710,7 @@ func TestV1ManifestFetch(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
manifest, err = ms.Get(ctx, dgst, WithTag("latest"))
|
||||
manifest, err = ms.Get(ctx, dgst, distribution.WithTag("latest"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
@ -723,7 +723,7 @@ func TestV1ManifestFetch(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
manifest, err = ms.Get(ctx, dgst, WithTag("badcontenttype"))
|
||||
manifest, err = ms.Get(ctx, dgst, distribution.WithTag("badcontenttype"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
@ -761,7 +761,7 @@ func TestManifestFetchWithEtag(t *testing.T) {
|
|||
if !ok {
|
||||
panic("wrong type for client manifest service")
|
||||
}
|
||||
_, err = clientManifestService.Get(ctx, d1, WithTag("latest"), AddEtagToTag("latest", d1.String()))
|
||||
_, err = clientManifestService.Get(ctx, d1, distribution.WithTag("latest"), AddEtagToTag("latest", d1.String()))
|
||||
if err != distribution.ErrManifestNotModified {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
@ -861,7 +861,7 @@ func TestManifestPut(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
if _, err := ms.Put(ctx, m1, WithTag(m1.Tag)); err != nil {
|
||||
if _, err := ms.Put(ctx, m1, distribution.WithTag(m1.Tag)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
|
|
|
@ -86,7 +86,11 @@ func (imh *imageManifestHandler) GetImageManifest(w http.ResponseWriter, r *http
|
|||
return
|
||||
}
|
||||
|
||||
manifest, err = manifests.Get(imh, imh.Digest)
|
||||
var options []distribution.ManifestServiceOption
|
||||
if imh.Tag != "" {
|
||||
options = append(options, distribution.WithTag(imh.Tag))
|
||||
}
|
||||
manifest, err = manifests.Get(imh, imh.Digest, options...)
|
||||
if err != nil {
|
||||
imh.Errors = append(imh.Errors, v2.ErrorCodeManifestUnknown.WithDetail(err))
|
||||
return
|
||||
|
@ -245,7 +249,11 @@ func (imh *imageManifestHandler) PutImageManifest(w http.ResponseWriter, r *http
|
|||
return
|
||||
}
|
||||
|
||||
_, err = manifests.Put(imh, manifest)
|
||||
var options []distribution.ManifestServiceOption
|
||||
if imh.Tag != "" {
|
||||
options = append(options, distribution.WithTag(imh.Tag))
|
||||
}
|
||||
_, err = manifests.Put(imh, manifest, options...)
|
||||
if err != nil {
|
||||
// TODO(stevvooe): These error handling switches really need to be
|
||||
// handled by an app global mapper.
|
||||
|
|
Loading…
Reference in a new issue