hubble/peer: handle burst of change notifications

pkg/hubble/peer/buffer.go

@@ -0,0 +1,105 @@
+// Copyright 2020 Authors of Cilium
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package peer
+
+import (
+	"fmt"
+	"io"
+
+	peerpb "github.com/cilium/cilium/api/v1/peer"
+	"github.com/cilium/cilium/pkg/lock"
+)
+
+type buffer struct {
+	max    int
+	buf    []*peerpb.ChangeNotification
+	mu     lock.Mutex
+	notify chan struct{}
+	stop   chan struct{}
+}
+
+// newBuffer creates a buffer of ChangeNotification that is safe for concurrent
+// use. The buffer is created with an initial size of 0 and is allowed to grow
+// until max is reached.
+func newBuffer(max int) *buffer {
+	return &buffer{
+		max:    max,
+		notify: nil,
+		stop:   make(chan struct{}),
+	}
+}
+
+// Len returns the number of elements in the buffer.
+func (b *buffer) Len() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return len(b.buf)
+}
+
+// Cap returns the capacity of the buffer.
+func (b *buffer) Cap() int {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return cap(b.buf)
+}
+
+// Push appends cn to the end of the buffer. An error is returned if its
+// maximum capacity is reached.
+func (b *buffer) Push(cn *peerpb.ChangeNotification) error {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if len(b.buf) == b.max {
+		return fmt.Errorf("max buffer size=%d reached", b.max)
+	}
+	b.buf = append(b.buf, cn)
+	if b.notify != nil {
+		close(b.notify)
+		b.notify = nil
+	}
+	return nil
+}
+
+// Pop removes and returns the first element in the buffer. If the buffer is
+// empty, Pop blocks until an element is added or Close is called in which case
+// io.EOF is returned.
+func (b *buffer) Pop() (*peerpb.ChangeNotification, error) {
+	b.mu.Lock()
+	if len(b.buf) == 0 {
+		if b.notify == nil {
+			b.notify = make(chan struct{})
+		}
+		notify := b.notify
+		b.mu.Unlock()
+		select {
+		case <-notify:
+			b.mu.Lock()
+		case <-b.stop:
+			return nil, io.EOF
+		}
+	}
+	cn := b.buf[0]
+	b.buf[0] = nil
+	b.buf = b.buf[1:]
+	b.mu.Unlock()
+	return cn, nil
+}
+
+// Close closes the buffer and frees the underlying memory.
+func (b *buffer) Close() {
+	close(b.stop)
+	b.mu.Lock()
+	b.buf = nil
+	b.mu.Unlock()
+}

pkg/hubble/peer/buffer_test.go

@@ -0,0 +1,84 @@
+// Copyright 2020 Authors of Cilium
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build !privileged_tests
+
+package peer
+
+import (
+	"fmt"
+	"io"
+	"testing"
+
+	peerpb "github.com/cilium/cilium/api/v1/peer"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestBufferPush(t *testing.T) {
+	max := 8
+	buf := newBuffer(max)
+	assert.NotNil(t, buf)
+	assert.Equal(t, 0, buf.Len())
+	assert.Equal(t, 0, buf.Cap())
+	for i := 0; i < max; i++ {
+		assert.NoError(t, buf.Push(&peerpb.ChangeNotification{}))
+		assert.Equal(t, i+1, buf.Len())
+	}
+	err := buf.Push(&peerpb.ChangeNotification{})
+	assert.Equal(t, fmt.Errorf("max buffer size=%d reached", max), err)
+	assert.Equal(t, max, buf.Len())
+}
+
+func TestBufferPop(t *testing.T) {
+	max := 8
+	buf := newBuffer(max)
+	assert.NotNil(t, buf)
+	assert.Equal(t, 0, buf.Len())
+	assert.Equal(t, 0, buf.Cap())
+	for i := 0; i < max; i++ {
+		assert.NoError(t, buf.Push(&peerpb.ChangeNotification{Name: fmt.Sprintf("#%d", i)}))
+		assert.Equal(t, i+1, buf.Len())
+	}
+	for i := 0; i < max; i++ {
+		cn, err := buf.Pop()
+		assert.NoError(t, err)
+		assert.Equal(t, &peerpb.ChangeNotification{Name: fmt.Sprintf("#%d", i)}, cn)
+	}
+
+	// pop should block until a CN is pushed
+	done := make(chan struct{})
+	go func() {
+		assert.NoError(t, buf.Push(&peerpb.ChangeNotification{Name: "test"}))
+		done <- struct{}{}
+	}()
+	cn, err := buf.Pop()
+	<-done
+	assert.NoError(t, err)
+	assert.Equal(t, &peerpb.ChangeNotification{Name: "test"}, cn)
+
+	// pop should block until `Close` is called
+	go func() {
+		buf.Close()
+		done <- struct{}{}
+	}()
+	cn, err = buf.Pop()
+	<-done
+	assert.Nil(t, cn)
+	assert.Equal(t, io.EOF, err)
+
+	// the buffer's underlying memory should be freed
+	assert.Equal(t, 0, buf.Len())
+	assert.Equal(t, 0, buf.Cap())
+}

pkg/hubble/peer/service.go

@@ -17,6 +17,7 @@ package peer
 import (
 	"context"
 	"errors"
+	"io"
 
 	peerpb "github.com/cilium/cilium/api/v1/peer"
 	"github.com/cilium/cilium/pkg/hubble/peer/serviceoption"
@@ -76,20 +77,17 @@ func (s *Service) Notify(_ *peerpb.NotifyRequest, stream peerpb.Peer_NotifyServe
 	})
 
 	// read from the handler's channel and fill the buffer until it's full
-	buf := make(chan *peerpb.ChangeNotification, s.opts.SendBufferSize)
+	buf := newBuffer(s.opts.MaxSendBufferSize)
 	g.Go(func() error {
-		defer close(buf)
+		defer buf.Close()
 		for {
 			select {
 			case cn, ok := <-h.C:
 				if !ok {
 					// channel is closed, stop buffering
 					return nil
 				}
-				select {
-				case buf <- cn:
-				default:
-					// buffer is full, messages are taking to long to send
+				if err := buf.Push(cn); err != nil {
 					return ErrStreamSendBlocked
 				}
 			case <-ctx.Done():
@@ -101,17 +99,16 @@ func (s *Service) Notify(_ *peerpb.NotifyRequest, stream peerpb.Peer_NotifyServe
 	// read from the buffer end send to the client
 	g.Go(func() error {
 		for {
-			select {
-			case cn, ok := <-buf:
-				if !ok {
-					// channel is closed, we're done
-					return nil
-				}
+			cn, err := buf.Pop()
+			switch err {
+			case nil:
 				if err := stream.Send(cn); err != nil {
 					return err
 				}
-			case <-ctx.Done():
+			case io.EOF:
 				return nil
+			default:
+				return err
 			}
 		}
 	})

pkg/hubble/peer/service_test.go

@@ -346,7 +346,7 @@ func TestService_NotifyWithBlockedSend(t *testing.T) {
 		},
 	}
 	notif := newNotifier(cb, init)
-	svc := NewService(notif, serviceoption.WithSendBufferSize(2))
+	svc := NewService(notif, serviceoption.WithMaxSendBufferSize(2))
 	done := make(chan struct{})
 	go func() {
 		err := svc.Notify(&peerpb.NotifyRequest{}, fakeServer)

pkg/hubble/peer/serviceoption/defaults.go

@@ -16,5 +16,5 @@ package serviceoption
 
 // Default serves only as reference point for default values.
 var Default = Options{
-	SendBufferSize: 10,
+	MaxSendBufferSize: 65_536,
 }

pkg/hubble/peer/serviceoption/option.go

@@ -16,17 +16,20 @@ package serviceoption
 
 // Options stores all the configuration values for the peer service.
 type Options struct {
-	SendBufferSize int
+	MaxSendBufferSize int
 }
 
 // Option customizes then configuration of the peer service.
 type Option func(o *Options)
 
-// WithSendBufferSize sets the size of the send buffer. When the send buffer is
-// full, for example due to errors in the transport, the server disconnects the
-// corresponding client. The send buffer is per connected client.
-func WithSendBufferSize(size int) Option {
+// WithMaxSendBufferSize sets the maximum size of the send buffer. When the
+// send buffer is full, for example due to errors in the transport, the server
+// disconnects the corresponding client.
+// The maximum buffer size should be large enough to accomodate the burst of
+// peer change notifications than happens on an initial call where all nodes in
+// the cluster are notified as being added.
+func WithMaxSendBufferSize(size int) Option {
 	return func(o *Options) {
-		o.SendBufferSize = size
+		o.MaxSendBufferSize = size
 	}
 }