Requeue events in the shim publisher
@@ -26,21 +26,37 @@ import (
"github.com/containerd/containerd/events"
"github.com/containerd/containerd/namespaces"
"github.com/containerd/typeurl"
+ "github.com/sirupsen/logrus"
)
+const (
+ queueSize = 2048
+ maxRequeue = 5
+)
+
+type item struct {
+ ev *v1.Envelope
+ ctx context.Context
+ count int
+}
+
func newPublisher(address string) *remoteEventsPublisher {
- return &remoteEventsPublisher{
+ l := &remoteEventsPublisher{
dialer: newDialier(func() (net.Conn, error) {
return connect(address, dial)
}),
closed: make(chan struct{}),
+ requeue: make(chan *item, queueSize),
}
+ go l.processQueue()
+ return l
}
type remoteEventsPublisher struct {
dialer *dialer
closed chan struct{}
closer sync.Once
+ requeue chan *item
}
func (l *remoteEventsPublisher) Done() <-chan struct{} {
@@ -55,11 +71,42 @@ func (l *remoteEventsPublisher) Close() (err error) {
return err
}
-func (l *remoteEventsPublisher) Publish(ctx context.Context, topic string, event events.Event) error {
+func (l *remoteEventsPublisher) processQueue() {
+ for i := range l.requeue {
+ if i.count > maxRequeue {
+ logrus.Errorf("evicting %s from queue because of retry count", i.ev.Topic)
+ // drop the event
+ continue
+ }
+
client, err := l.dialer.Get()
if err != nil {
- return err
+ l.dialer.Put(err)
+
+ l.queue(i)
+ logrus.WithError(err).Error("get events client")
+ continue
+ }
+ if _, err := client.Forward(i.ctx, &v1.ForwardRequest{
+ Envelope: i.ev,
+ }); err != nil {
+ l.dialer.Put(err)
+ logrus.WithError(err).Error("forward event")
+ l.queue(i)
}
+ }
+}
+
+func (l *remoteEventsPublisher) queue(i *item) {
+ go func() {
+ i.count++
+ // re-queue after a short delay
+ time.Sleep(time.Duration(1*i.count) * time.Second)
+ l.requeue <- i
+ }()
+}
+
+func (l *remoteEventsPublisher) Publish(ctx context.Context, topic string, event events.Event) error {
ns, err := namespaces.NamespaceRequired(ctx)
if err != nil {
return err
@@ -68,15 +115,26 @@ func (l *remoteEventsPublisher) Publish(ctx context.Context, topic string, event
if err != nil {
return err
}
- if _, err := client.Forward(ctx, &v1.ForwardRequest{
- Envelope: &v1.Envelope{
+ i := &item{
+ ev: &v1.Envelope{
Timestamp: time.Now(),
Namespace: ns,
Topic: topic,
Event: any,
},
+ ctx: ctx,
+ }
+ client, err := l.dialer.Get()
+ if err != nil {
+ l.dialer.Put(err)
+ l.queue(i)
+ return err
+ }
+ if _, err := client.Forward(i.ctx, &v1.ForwardRequest{
+ Envelope: i.ev,
}); err != nil {
l.dialer.Put(err)
+ l.queue(i)
return err
}
return nil
