README
¶
Docker Events Package
The Docker events package implements a composable event distribution package
for Go.
Originally created to implement the notifications in Docker Registry 2, we've found the pattern to be useful in other applications. This package is most of the same code with slightly updated interfaces. Much of the internals have been made available.
Usage
The events package centers around a Sink type. Events are written with
calls to Sink.Write(event Event). Sinks can be wired up in various
configurations to achieve interesting behavior.
The canonical example is that employed by the
docker/distribution/notifications
package. Let's say we have a type httpSink where we'd like to queue
notifications. As a rule, it should send a single http request and return an
error if it fails:
func (h *httpSink) Write(event Event) error {
p, err := json.Marshal(event)
if err != nil {
return err
}
body := bytes.NewReader(p)
resp, err := h.client.Post(h.url, "application/json", body)
if err != nil {
return err
}
defer resp.Body.Close()
if resp.Status != 200 {
return errors.New("unexpected status")
}
return nil
}
// implement (*httpSink).Close()
With just that, we can start using components from this package. One can call
(*httpSink).Write to send events as the body of a post request to a
configured URL.
Retries
HTTP can be unreliable. The first feature we'd like is to have some retry:
hs := newHTTPSink(/*...*/)
retry := NewRetryingSink(hs, NewBreaker(5, time.Second))
We now have a sink that will retry events against the httpSink until they
succeed. The retry will backoff for one second after 5 consecutive failures
using the breaker strategy.
Queues
This isn't quite enough. We we want a sink that doesn't block while we are
waiting for events to be sent. Let's add a Queue:
queue := NewQueue(retry)
Now, we have an unbounded queue that will work through all events sent with
(*Queue).Write. Events can be added asynchronously to the queue without
blocking the current execution path. This is ideal for use in an http request.
Broadcast
It usually turns out that you want to send to more than one listener. We can
use Broadcaster to support this:
var broadcast = NewBroadcaster() // make it available somewhere in your application.
broadcast.Add(queue) // add your queue!
broadcast.Add(queue2) // and another!
With the above, we can now call broadcast.Write in our http handlers and have
all the events distributed to each queue. Because the events are queued, not
listener blocks another.
Extending
For the most part, the above is sufficient for a lot of applications. However,
extending the above functionality can be done implementing your own Sink. The
behavior and semantics of the sink can be completely dependent on the
application requirements. The interface is provided below for reference:
type Sink {
Write(Event) error
Close() error
}
Application behavior can be controlled by how Write behaves. The examples
above are designed to queue the message and return as quickly as possible.
Other implementations may block until the event is committed to durable
storage.
Copyright and license
Copyright © 2016 Docker, Inc. go-events is licensed under the Apache License, Version 2.0. See LICENSE for the full license text.
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ( // DefaultExponentialBackoffConfig provides a default configuration for // exponential backoff. DefaultExponentialBackoffConfig = ExponentialBackoffConfig{ Base: time.Second, Factor: time.Second, Max: 20 * time.Second, } )
var ( // ErrSinkClosed is returned if a write is issued to a sink that has been // closed. If encountered, the error should be considered terminal and // retries will not be successful. ErrSinkClosed = fmt.Errorf("events: sink closed") )
Functions ¶
This section is empty.
Types ¶
type Breaker ¶
type Breaker struct {
// contains filtered or unexported fields
}
Breaker implements a circuit breaker retry strategy.
The current implementation never drops events.
func NewBreaker ¶
NewBreaker returns a breaker that will backoff after the threshold has been tripped. A Breaker is thread safe and may be shared by many goroutines.
type Broadcaster ¶
type Broadcaster struct {
// contains filtered or unexported fields
}
Broadcaster sends events to multiple, reliable Sinks. The goal of this component is to dispatch events to configured endpoints. Reliability can be provided by wrapping incoming sinks.
func NewBroadcaster ¶
func NewBroadcaster(sinks ...Sink) *Broadcaster
NewBroadcaster appends one or more sinks to the list of sinks. The broadcaster behavior will be affected by the properties of the sink. Generally, the sink should accept all messages and deal with reliability on its own. Use of EventQueue and RetryingSink should be used here.
func (*Broadcaster) Add ¶
func (b *Broadcaster) Add(sink Sink) error
Add the sink to the broadcaster.
The provided sink must be comparable with equality. Typically, this just works with a regular pointer type.
func (*Broadcaster) Close ¶
func (b *Broadcaster) Close() error
Close the broadcaster, ensuring that all messages are flushed to the underlying sink before returning.
func (*Broadcaster) Remove ¶
func (b *Broadcaster) Remove(sink Sink) error
Remove the provided sink.
func (*Broadcaster) String ¶
func (b *Broadcaster) String() string
func (*Broadcaster) Write ¶
func (b *Broadcaster) Write(event Event) error
Write accepts an event to be dispatched to all sinks. This method will never fail and should never block (hopefully!). The caller cedes the memory to the broadcaster and should not modify it after calling write.
type Channel ¶
type Channel struct {
C chan Event
// contains filtered or unexported fields
}
Channel provides a sink that can be listened on. The writer and channel listener must operate in separate goroutines.
Consumers should listen on Channel.C until Closed is closed.
func NewChannel ¶
NewChannel returns a channel. If buffer is zero, the channel is unbuffered.
type ExponentialBackoff ¶
type ExponentialBackoff struct {
// contains filtered or unexported fields
}
ExponentialBackoff implements random backoff with exponentially increasing bounds as the number consecutive failures increase.
func NewExponentialBackoff ¶
func NewExponentialBackoff(config ExponentialBackoffConfig) *ExponentialBackoff
NewExponentialBackoff returns an exponential backoff strategy with the desired config. If config is nil, the default is returned.
func (*ExponentialBackoff) Failure ¶
func (b *ExponentialBackoff) Failure(event Event, err error) bool
Failure increments the failure counter.
func (*ExponentialBackoff) Proceed ¶
func (b *ExponentialBackoff) Proceed(event Event) time.Duration
Proceed returns the next randomly bound exponential backoff time.
func (*ExponentialBackoff) Success ¶
func (b *ExponentialBackoff) Success(event Event)
Success resets the failures counter.
type ExponentialBackoffConfig ¶
type ExponentialBackoffConfig struct {
// Base is the minimum bound for backing off after failure.
Base time.Duration
// Factor sets the amount of time by which the backoff grows with each
// failure.
Factor time.Duration
// Max is the absolute maxiumum bound for a single backoff.
Max time.Duration
}
ExponentialBackoffConfig configures backoff parameters.
Note that these parameters operate on the upper bound for choosing a random value. For example, at Base=1s, a random value in [0,1s) will be chosen for the backoff value.
type Filter ¶
type Filter struct {
// contains filtered or unexported fields
}
Filter provides an event sink that sends only events that are accepted by a Matcher. No methods on filter are goroutine safe.
type MatcherFunc ¶
MatcherFunc implements matcher with just a function.
func (MatcherFunc) Match ¶
func (fn MatcherFunc) Match(event Event) bool
Match calls the wrapped function.
type Queue ¶
type Queue struct {
// contains filtered or unexported fields
}
Queue accepts all messages into a queue for asynchronous consumption by a sink. It is unbounded and thread safe but the sink must be reliable or events will be dropped.
type RetryStrategy ¶
type RetryStrategy interface {
// Proceed is called before every event send. If proceed returns a
// positive, non-zero integer, the retryer will back off by the provided
// duration.
//
// An event is provided, by may be ignored.
Proceed(event Event) time.Duration
// Failure reports a failure to the strategy. If this method returns true,
// the event should be dropped.
Failure(event Event, err error) bool
// Success should be called when an event is sent successfully.
Success(event Event)
}
RetryStrategy defines a strategy for retrying event sink writes.
All methods should be goroutine safe.
type RetryingSink ¶
type RetryingSink struct {
// contains filtered or unexported fields
}
RetryingSink retries the write until success or an ErrSinkClosed is returned. Underlying sink must have p > 0 of succeeding or the sink will block. Retry is configured with a RetryStrategy. Concurrent calls to a retrying sink are serialized through the sink, meaning that if one is in-flight, another will not proceed.
func NewRetryingSink ¶
func NewRetryingSink(sink Sink, strategy RetryStrategy) *RetryingSink
NewRetryingSink returns a sink that will retry writes to a sink, backing off on failure. Parameters threshold and backoff adjust the behavior of the circuit breaker.
func (*RetryingSink) Close ¶
func (rs *RetryingSink) Close() error
Close closes the sink and the underlying sink.
func (*RetryingSink) String ¶
func (rs *RetryingSink) String() string
func (*RetryingSink) Write ¶
func (rs *RetryingSink) Write(event Event) error
Write attempts to flush the events to the downstream sink until it succeeds or the sink is closed.
type Sink ¶
type Sink interface {
// Write an event to the Sink. If no error is returned, the caller will
// assume that all events have been committed to the sink. If an error is
// received, the caller may retry sending the event.
Write(event Event) error
// Close the sink, possibly waiting for pending events to flush.
Close() error
}
Sink accepts and sends events.
Source Files