9f0c8d6616
This changeset implements webhook notification endpoints for dispatching registry events. Repository instances can be decorated by a listener that converts calls into context-aware events, using a bridge. Events generated in the bridge are written to a sink. Implementations of sink include a broadcast and endpoint sink which can be used to configure event dispatch. Endpoints represent a webhook notification target, with queueing and retries built in. They can be added to a Broadcaster, which is a simple sink that writes a block of events to several sinks, to provide a complete dispatch mechanism. The main caveat to the current approach is that all unsent notifications are inmemory. Best effort is made to ensure that notifications are not dropped, to the point where queues may back up on faulty endpoints. If the endpoint is fixed, the events will be retried and all messages will go through. Internally, this functionality is all made up of Sink objects. The queuing functionality is implemented with an eventQueue sink and retries are implemented with retryingSink. Replacing the inmemory queuing with something persistent should be as simple as replacing broadcaster with a remote queue and that sets up the sinks to be local workers listening to that remote queue. Metrics are kept for each endpoint and exported via expvar. This may not be a permanent appraoch but should provide enough information for troubleshooting notification problems. Signed-off-by: Stephen J Day <stephen.day@docker.com>
132 lines
3.9 KiB
Go
132 lines
3.9 KiB
Go
package notifications
|
|
|
|
import (
|
|
"encoding/json"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
// TestEventJSONFormat provides silly test to detect if the event format or
|
|
// envelope has changed. If this code fails, the revision of the protocol may
|
|
// need to be incremented.
|
|
func TestEventEnvelopeJSONFormat(t *testing.T) {
|
|
var expected = strings.TrimSpace(`
|
|
{
|
|
"events": [
|
|
{
|
|
"id": "asdf-asdf-asdf-asdf-0",
|
|
"timestamp": "2006-01-02T15:04:05Z",
|
|
"action": "push",
|
|
"target": {
|
|
"type": "manifest",
|
|
"name": "library/test",
|
|
"digest": "sha256:0123456789abcdef0",
|
|
"tag": "latest",
|
|
"url": "http://example.com/v2/library/test/manifests/latest"
|
|
},
|
|
"actor": {
|
|
"name": "test-actor",
|
|
"addr": "client.local",
|
|
"host": "registrycluster.local",
|
|
"requestID": "asdfasdf"
|
|
},
|
|
"source": {
|
|
"addr": "hostname.local:port"
|
|
}
|
|
},
|
|
{
|
|
"id": "asdf-asdf-asdf-asdf-1",
|
|
"timestamp": "2006-01-02T15:04:05Z",
|
|
"action": "push",
|
|
"target": {
|
|
"type": "blob",
|
|
"name": "library/test",
|
|
"digest": "tarsum.v2+sha256:0123456789abcdef1",
|
|
"url": "http://example.com/v2/library/test/manifests/latest"
|
|
},
|
|
"actor": {
|
|
"name": "test-actor",
|
|
"addr": "client.local",
|
|
"host": "registrycluster.local",
|
|
"requestID": "asdfasdf"
|
|
},
|
|
"source": {
|
|
"addr": "hostname.local:port"
|
|
}
|
|
},
|
|
{
|
|
"id": "asdf-asdf-asdf-asdf-2",
|
|
"timestamp": "2006-01-02T15:04:05Z",
|
|
"action": "push",
|
|
"target": {
|
|
"type": "blob",
|
|
"name": "library/test",
|
|
"digest": "tarsum.v2+sha256:0123456789abcdef2",
|
|
"url": "http://example.com/v2/library/test/manifests/latest"
|
|
},
|
|
"actor": {
|
|
"name": "test-actor",
|
|
"addr": "client.local",
|
|
"host": "registrycluster.local",
|
|
"requestID": "asdfasdf"
|
|
},
|
|
"source": {
|
|
"addr": "hostname.local:port"
|
|
}
|
|
}
|
|
]
|
|
}
|
|
`)
|
|
|
|
tm, err := time.Parse(time.RFC3339, time.RFC3339[:len(time.RFC3339)-5])
|
|
if err != nil {
|
|
t.Fatalf("error creating time: %v", err)
|
|
}
|
|
|
|
var prototype Event
|
|
prototype.Action = "push"
|
|
prototype.Timestamp = tm
|
|
prototype.Actor.Addr = "client.local"
|
|
prototype.Actor.Name = "test-actor"
|
|
prototype.Actor.RequestID = "asdfasdf"
|
|
prototype.Actor.Host = "registrycluster.local"
|
|
prototype.Source.Addr = "hostname.local:port"
|
|
|
|
var manifestPush Event
|
|
manifestPush = prototype
|
|
manifestPush.ID = "asdf-asdf-asdf-asdf-0"
|
|
manifestPush.Target.Digest = "sha256:0123456789abcdef0"
|
|
manifestPush.Target.Type = "manifest"
|
|
manifestPush.Target.Name = "library/test"
|
|
manifestPush.Target.Tag = "latest"
|
|
manifestPush.Target.URL = "http://example.com/v2/library/test/manifests/latest"
|
|
|
|
var layerPush0 Event
|
|
layerPush0 = prototype
|
|
layerPush0.ID = "asdf-asdf-asdf-asdf-1"
|
|
layerPush0.Target.Digest = "tarsum.v2+sha256:0123456789abcdef1"
|
|
layerPush0.Target.Type = "blob"
|
|
layerPush0.Target.Name = "library/test"
|
|
layerPush0.Target.URL = "http://example.com/v2/library/test/manifests/latest"
|
|
|
|
var layerPush1 Event
|
|
layerPush1 = prototype
|
|
layerPush1.ID = "asdf-asdf-asdf-asdf-2"
|
|
layerPush1.Target.Digest = "tarsum.v2+sha256:0123456789abcdef2"
|
|
layerPush1.Target.Type = "blob"
|
|
layerPush1.Target.Name = "library/test"
|
|
layerPush1.Target.URL = "http://example.com/v2/library/test/manifests/latest"
|
|
|
|
var envelope Envelope
|
|
envelope.Events = append(envelope.Events, manifestPush, layerPush0, layerPush1)
|
|
|
|
p, err := json.MarshalIndent(envelope, "", " ")
|
|
if err != nil {
|
|
t.Fatalf("unexpected error marshaling envelope: %v", err)
|
|
}
|
|
if string(p) != expected {
|
|
t.Fatalf("format has changed\n%s\n != \n%s", string(p), expected)
|
|
}
|
|
}
|