fix: manager doesn't release read locks

also some general cleanup and tests for the mgr

Author: ecordell

manager/controller_test.go

@@ -43,7 +43,9 @@ func ExampleNewOwnedResourceController() {
 	mgr := NewManager(ctrlmanageropts.RecommendedDebuggingOptions().DebuggingConfiguration, ":", broadcaster, eventSink)
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Millisecond)
 	defer cancel()
-	_ = mgr.Start(ctx, controller)
+	readyc := make(chan struct{})
+	_ = mgr.Start(ctx, readyc, controller)
+	<-readyc
 	// Output:
 }
 
@@ -65,9 +67,12 @@ func TestControllerQueueDone(t *testing.T) {
 	mgr := NewManager(ctrlmanageropts.RecommendedDebuggingOptions().DebuggingConfiguration, ":", broadcaster, eventSink)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+
+	readyc := make(chan struct{})
 	go func() {
-		_ = mgr.Start(ctx, controller)
+		_ = mgr.Start(ctx, readyc, controller)
 	}()
+	<-readyc
 
 	// add many keys
 	for i := 0; i < 10; i++ {
@@ -105,9 +110,11 @@ func TestControllerEventsBroadcast(t *testing.T) {
 	mgr := NewManager(ctrlmanageropts.RecommendedDebuggingOptions().DebuggingConfiguration, ":8888", broadcaster, eventSink)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
+	readyc := make(chan struct{})
 	go func() {
-		_ = mgr.Start(ctx, controller)
+		_ = mgr.Start(ctx, readyc, controller)
 	}()
+	<-readyc
 	require.Eventually(t, healthCheckPassing(), 1*time.Second, 50*time.Millisecond)
 
 	recorder.Event(&v1.ObjectReference{Namespace: "test", Name: "a"}, v1.EventTypeNormal, "test", "test")

manager/manager.go

@@ -2,6 +2,7 @@ package manager
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"runtime"
@@ -45,6 +46,9 @@ type Manager struct {
 // information for its managed set of controllers.
 func NewManager(debugConfig *componentconfig.DebuggingConfiguration, address string, broadcaster record.EventBroadcaster, sink record.EventSink) *Manager {
 	handler := healthz.NewMutableHealthzHandler()
+	if broadcaster == nil {
+		broadcaster = record.NewBroadcaster()
+	}
 	return &Manager{
 		healthzHandler: handler,
 		srv: &http.Server{
@@ -61,13 +65,15 @@ func NewManager(debugConfig *componentconfig.DebuggingConfiguration, address str
 // Start starts a set of controllers in an errgroup and serves
 // health / debug endpoints for them. It stops when the context is cancelled.
 // It will only have an effect the first time it is called.
-func (m *Manager) Start(ctx context.Context, controllers ...Controller) error {
+func (m *Manager) Start(ctx context.Context, readyc chan<- struct{}, controllers ...Controller) error {
 	m.RLock()
 	if m.errG != nil {
+		m.RUnlock()
 		return fmt.Errorf("manager already started")
 	}
 	m.RUnlock()
 
+	var startErr error
 	m.once.Do(func() {
 		m.Lock()
 		m.errG, ctx = errgroup.WithContext(ctx)
@@ -76,40 +82,65 @@ func (m *Manager) Start(ctx context.Context, controllers ...Controller) error {
 
 		// start controllers
 		if err := m.Go(controllers...); err != nil {
+			startErr = err
 			return
 		}
 
 		// start health / debug server
 		m.errG.Go(func() error {
-			return m.srv.ListenAndServe()
+			if err := m.srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
+				return err
+			}
+			return nil
 		})
 
 		// start broadcaster
 		m.errG.Go(func() error {
 			m.broadcaster.StartStructuredLogging(2)
-			m.broadcaster.StartRecordingToSink(m.sink)
+			if m.sink != nil {
+				m.broadcaster.StartRecordingToSink(m.sink)
+			}
 			return nil
 		})
 
-		// stop health / debug server when context is cancelled
+		// stop health / debug server and all controllers when context is
+		// cancelled
 		m.errG.Go(func() error {
 			<-ctx.Done()
 			m.broadcaster.Shutdown()
-			return m.srv.Shutdown(ctx)
+
+			m.Lock()
+			for ctrl, cancel := range m.cancelFuncs {
+				cancel()
+				delete(m.cancelFuncs, ctrl)
+			}
+			m.Unlock()
+
+			// no context passed to shutdown; the errg will block
+			// until the server is closed
+			return m.srv.Shutdown(context.Background())
 		})
 	})
+
+	close(readyc)
+
+	if startErr != nil {
+		return startErr
+	}
+
 	if err := m.errG.Wait(); err != nil {
 		return err
 	}
 
-	return ctx.Err()
+	return nil
 }
 
 // Go adds controllers into the existing manager's errgroup
 func (m *Manager) Go(controllers ...Controller) error {
 	m.RLock()
 	errG := m.errG
 	if errG == nil {
+		m.RUnlock()
 		return fmt.Errorf("cannot add controllers to an unstarted manager")
 	}
 	ctx := m.errGCtx

manager/manager_test.go

@@ -0,0 +1,93 @@
+package manager
+
+import (
+	"context"
+	"net"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/component-base/config"
+	"k8s.io/klog/v2/klogr"
+
+	"github.com/authzed/controller-idioms/queue"
+	"github.com/authzed/controller-idioms/typed"
+)
+
+func TestManager(t *testing.T) {
+	t.Parallel()
+	ctx, cancel := context.WithCancel(context.Background())
+
+	m := NewManager(&config.DebuggingConfiguration{
+		EnableProfiling:           false,
+		EnableContentionProfiling: false,
+	}, ":"+getFreePort(t), nil, nil)
+
+	ready := make(chan struct{})
+	go func() {
+		require.NoError(t, m.Start(ctx, ready, testController(t, "a")))
+	}()
+	<-ready
+
+	requireCancelFnCount(t, m, 1)
+
+	// Ensure that the manager can't be started twice.
+	require.Error(t, m.Start(ctx, ready), "manager already started")
+
+	// Add some controllers after start
+	require.NoError(t, m.Go(testController(t, "b"), testController(t, "c")))
+	requireCancelFnCount(t, m, 3)
+
+	specificCtrl := testController(t, "d")
+	require.NoError(t, m.Go(specificCtrl))
+	requireCancelFnCount(t, m, 4)
+
+	// stop a specific controller
+	m.Cancel(specificCtrl)
+	requireCancelFnCount(t, m, 3)
+
+	// cancel the manager's context, which will clean up all controllers
+	cancel()
+
+	requireCancelFnCount(t, m, 0)
+}
+
+func getFreePort(t testing.TB) string {
+	t.Helper()
+
+	a, err := net.ResolveTCPAddr("tcp", "localhost:0")
+	require.NoError(t, err)
+	l, err := net.ListenTCP("tcp", a)
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, l.Close())
+	}()
+	return strconv.Itoa(l.Addr().(*net.TCPAddr).Port)
+}
+
+func testController(t *testing.T, name string) Controller {
+	gvr := schema.GroupVersionResource{
+		Group:    "example.com",
+		Version:  "v1",
+		Resource: "mytypes",
+	}
+	CtxQueue := queue.NewQueueOperationsCtx()
+	registry := typed.NewRegistry()
+	broadcaster := record.NewBroadcaster()
+
+	return NewOwnedResourceController(klogr.New(), name, gvr, CtxQueue, registry, broadcaster, func(_ context.Context, gvr schema.GroupVersionResource, namespace, name string) {
+		t.Log("processing", gvr, namespace, name)
+	})
+}
+
+func requireCancelFnCount(t *testing.T, m *Manager, count int) {
+	t.Helper()
+	require.Eventually(t, func() bool {
+		m.RLock()
+		defer m.RUnlock()
+		return len(m.cancelFuncs) == count
+	}, 100*time.Second, 10*time.Millisecond)
+}