Repository: skupperproject/skupper
Pull request #1327 (2023-12-15 20:40:01)

Fix races in the Flow Collector
Lots of reorganization for hopefully zero functional change! :sweat: 

Probably easiest to review commit by commit. Glad to split it up if y'all prefer - they may just seem random without the end state in view.

* Stole the mock messaging.ConnectionFactory from service_sync and put it somewhere multiple packages could use. Getting any useful coverage in this package without some way to send and receive messages was going to be a train wreck. Curious if there's a way to quickly stand up a skupper router isolated per-test-case instead. Bonus points - pkg/service_sync went from like 20 (mostly test-utility related) data races to 3.

* Fixed some obvious race cases in flow/messaging.go shutdown. Also removed a timer in the sender that was doing nothing constructive. Also allowed me to write tests that can wait for the collector to finish.

* Added test coverage and adapted current collector tests to remove racy/flaky behavior.

* Fixed one of the collector data races around shutdown. (more important than it seems - config-sync can loose a leader election and shut down it's collector-lite.)

* ~~Reorganize most of the collector beaconUpdates loop so that it doesn't touch any of the collector's state. Introduces a few channels to communicate with the main loop. This I'm most on-the-fence about. It could instead be a couple line change if we nixed the concurrent beaconUpdates loop and kept everything in the recordUpdates loop. That said, we have no clue how sensitive the send buffers on the other side are - if anything we should be more aggressively reading flow records.~~ Change of heart. Have dropped the concurrent processing of beacons for now in order to solve the race conditions with a minimal changelist. Hope to come back around to reorganize the way our cache is accessed/updated anyhow, esp if we find that a single event handling loop isn't cutting it.




Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index c6a73b04..40b03b91 100644
--- a/.circleci/config.yml
+++ b/.circleci/config.yml
@@ -322,6 +322,10 @@ commands:
             sudo apt-get update -qq
             sudo apt-get -qq -y install podman
             podman version
+            
+            # temporary fix for https://github.com/containers/podman/issues/21024
+            wget https://download.opensuse.org/repositories/devel:/kubic:/libcontainers:/stable/xUbuntu_22.04/amd64/conmon_2.1.2~0_amd64.deb -O /tmp/conmon_2.1.2.deb
+            sudo apt install /tmp/conmon_2.1.2.deb
 
             # Bypassing CircleCI issue with user session (see: https://github.com/containers/podman/issues/16529)
             mkdir -p ~/.config/containers/containers.conf.d
diff --git a/cmd/manifest/manifest.go b/cmd/manifest/manifest.go
index fb332039..1dafb566 100644
--- a/cmd/manifest/manifest.go
+++ b/cmd/manifest/manifest.go
@@ -1,89 +1,13 @@
 package main
 
 import (
-	"encoding/json"
 	"fmt"
+	"github.com/skupperproject/skupper/pkg/utils/configs"
 	"os"
 
-	"github.com/skupperproject/skupper/pkg/images"
 	"github.com/spf13/cobra"
 )
 
-type SkupperImage struct {
-	Name       string `yaml:"name"`
-	SHA        string `yaml:"sha"`
-	Repository string `yaml:"repository,omitempty"`
-}
-
-func generateManifestFile() error {
-	// Define a struct for the manifest file.
-	manifest := struct {
-		Images []SkupperImage `json:"images"`
-	}{
-		Images: []SkupperImage{
-			{
-				Name:       images.GetRouterImageName(),
-				SHA:        images.GetSha(images.GetRouterImageName()),
-				Repository: "https://github.com/skupperproject/skupper-router",
-			},
-			{
-				Name:       images.GetServiceControllerImageName(),
-				SHA:        images.GetSha(images.GetServiceControllerImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetControllerPodmanImageName(),
-				SHA:        images.GetSha(images.GetControllerPodmanImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetConfigSyncImageName(),
-				SHA:        images.GetSha(images.GetConfigSyncImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetFlowCollectorImageName(),
-				SHA:        images.GetSha(images.GetFlowCollectorImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name:       images.GetSiteControllerImageName(),
-				SHA:        images.GetSha(images.GetSiteControllerImageName()),
-				Repository: "https://github.com/skupperproject/skupper",
-			},
-			{
-				Name: images.GetPrometheusServerImageName(),
-				SHA:  images.GetSha(images.GetPrometheusServerImageName()),
-			},
-			{
-				Name: images.GetOauthProxyImageName(),
-				SHA:  images.GetSha(images.GetOauthProxyImageName()),
-			},
-		},
-	}
-
-	// Encode the manifest image list as JSON.
-	manifestListJSON, err := json.MarshalIndent(manifest, "", "   ")
-	if err != nil {
-		return fmt.Errorf("Error encoding manifest image list: %v\n", err)
-
-	}
-
-	// Create a new file.
-	file, err := os.Create("manifest.json")
-	if err != nil {
-		return fmt.Errorf("Error creating file: %v\n", err)
-	}
-
-	// Write the JSON data to the file.
-	_, err = file.Write(manifestListJSON)
-	if err != nil {
-		return fmt.Errorf("Error writing to file: %v\n", err)
-	}
-
-	return nil
-}
-
 func main() {
 
 	var command = &cobra.Command{
@@ -93,7 +17,9 @@ func main() {
 		Short: "generates a manifest.json file",
 		Args:  cobra.NoArgs,
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return generateManifestFile()
+			manifestManager := configs.ManifestManager{EnableSHA: true}
+			return manifestManager.CreateFile(manifestManager.GetConfiguredManifest())
+
 		},
 	}
 
diff --git a/cmd/skupper/skupper.go b/cmd/skupper/skupper.go
index a12ff51c..ee834f17 100644
--- a/cmd/skupper/skupper.go
+++ b/cmd/skupper/skupper.go
@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"github.com/skupperproject/skupper/pkg/utils/configs"
 	"os"
 	"reflect"
 	"strconv"
@@ -866,6 +867,21 @@ func NewCmdVersion(skupperClient SkupperSiteClient) *cobra.Command {
 	return cmd
 }
 
+func NewCmdVersionManifest() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "manifest",
+		Short: "Report the version of the Skupper images by default and the value of the environment variables",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, args []string) error {
+			silenceCobra(cmd)
+
+			manifestManager := configs.ManifestManager{}
+			return manifestManager.CreateFile(manifestManager.GetDefaultManifestWithEnv())
+		},
+	}
+	return cmd
+}
+
 func NewCmdDebug() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "debug dump <file>, debug events or debug service <service-name>",
@@ -1040,7 +1056,10 @@ func init() {
 	cmdStatusService := NewCmdServiceStatus(skupperCli.Service())
 	cmdLabelsService := NewCmdServiceLabel(skupperCli.Service())
 
+	cmdVersionManifest := NewCmdVersionManifest()
 	cmdVersion := NewCmdVersion(skupperCli.Site())
+	cmdVersion.AddCommand(cmdVersionManifest)
+
 	cmdDebugDump := NewCmdDebugDump(skupperCli.Debug())
 	cmdDebugEvents := NewCmdDebugEvents(skupperCli.Debug())
 	cmdDebugService := NewCmdDebugService(skupperCli.Debug())
diff --git a/pkg/flow/collector.go b/pkg/flow/collector.go
index ea85d201..436c196c 100644
--- a/pkg/flow/collector.go
+++ b/pkg/flow/collector.go
@@ -279,69 +279,55 @@ func getRealSizeOf(v interface{}) (int, error) {
 	return b.Len(), nil
 }
 
-func (c *FlowCollector) beaconUpdates(stopCh <-chan struct{}) {
-	for {
-		select {
-		case beaconUpdates := <-c.beaconsIncoming:
-			for _, beaconUpdate := range beaconUpdates {
-				beacon, ok := beaconUpdate.(BeaconRecord)
-				if !ok {
-					log.Println("COLLECTOR: Unable to convert interface to beacon")
-				} else {
-					if source, ok := c.eventSources[beacon.Identity]; !ok {
-						var receivers []*receiver
-						log.Printf("COLLECTOR: Detected event source %s of type %s \n", beacon.Identity, beacon.SourceType)
-						receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming))
-						if beacon.SourceType == recordNames[Router] {
-							switch c.mode {
-							case RecordMetrics:
-								receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".flows", c.recordsIncoming))
-							case RecordStatus:
-								receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".logs", c.recordsIncoming))
-							}
-						} else if beacon.SourceType == recordNames[Controller] {
-							receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".heartbeats", c.heartbeatsIncoming))
-						}
-						outgoing := make(chan interface{})
-						s := newSender(c.connectionFactory, beacon.Direct, false, outgoing)
-						if c.connectionFactory != nil {
-							s.start()
-						}
-						now := uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-						c.eventSources[beacon.Identity] = &eventSource{
-							EventSourceRecord: EventSourceRecord{
-								Base: Base{
-									RecType:   recordNames[EventSource],
-									Identity:  beacon.Identity,
-									StartTime: now,
-									EndTime:   0,
-								},
-								Beacon:    &beacon,
-								LastHeard: now,
-								Beacons:   1,
-							},
-							receivers: receivers,
-							send: &senderDirect{
-								sender:    s,
-								outgoing:  outgoing,
-								heartbeat: false,
-							},
-						}
-						if c.connectionFactory != nil {
-							for _, receiver := range receivers {
-								receiver.start()
-							}
-						}
-						c.pendingFlush[beacon.Direct] = c.eventSources[beacon.Identity].send
-					} else {
-						source.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
-						source.Beacons++
-					}
-				}
+func (c *FlowCollector) beaconUpdate(beacon BeaconRecord) {
+	if source, ok := c.eventSources[beacon.Identity]; !ok {
+		var receivers []*receiver
+		log.Printf("COLLECTOR: Detected event source %s of type %s \n", beacon.Identity, beacon.SourceType)
+		receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address, c.recordsIncoming))
+		if beacon.SourceType == recordNames[Router] {
+			switch c.mode {
+			case RecordMetrics:
+				receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".flows", c.recordsIncoming))
+			case RecordStatus:
+				receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".logs", c.recordsIncoming))
 			}
-		case <-stopCh:
-			return
+		} else if beacon.SourceType == recordNames[Controller] {
+			receivers = append(receivers, newReceiver(c.connectionFactory, beacon.Address+".heartbeats", c.heartbeatsIncoming))
+		}
+		outgoing := make(chan interface{})
+		s := newSender(c.connectionFactory, beacon.Direct, false, outgoing)
+		if c.connectionFactory != nil {
+			s.start()
 		}
+		now := uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		c.eventSources[beacon.Identity] = &eventSource{
+			EventSourceRecord: EventSourceRecord{
+				Base: Base{
+					RecType:   recordNames[EventSource],
+					Identity:  beacon.Identity,
+					StartTime: now,
+					EndTime:   0,
+				},
+				Beacon:    &beacon,
+				LastHeard: now,
+				Beacons:   1,
+			},
+			receivers: receivers,
+			send: &senderDirect{
+				sender:    s,
+				outgoing:  outgoing,
+				heartbeat: false,
+			},
+		}
+		if c.connectionFactory != nil {
+			for _, receiver := range receivers {
+				receiver.start()
+			}
+		}
+		c.pendingFlush[beacon.Direct] = c.eventSources[beacon.Identity].send
+	} else {
+		source.LastHeard = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+		source.Beacons++
 	}
 }
 
@@ -355,6 +341,15 @@ func (c *FlowCollector) recordUpdates(stopCh <-chan struct{}) {
 
 	for {
 		select {
+		case beaconUpdates := <-c.beaconsIncoming:
+			for _, beaconUpdate := range beaconUpdates {
+				beacon, ok := beaconUpdate.(BeaconRecord)
+				if !ok {
+					log.Println("COLLECTOR: Unable to convert interface to beacon")
+				} else {
+					c.beaconUpdate(beacon)
+				}
+			}
 		case heartbeatUpdates := <-c.heartbeatsIncoming:
 			for _, heartbeatUpdate := range heartbeatUpdates {
 				heartbeat, ok := heartbeatUpdate.(HeartbeatRecord)
@@ -414,9 +409,13 @@ func (c *FlowCollector) run(stopCh <-chan struct{}) {
 	c.beaconReceiver = newReceiver(c.connectionFactory, BeaconAddress, c.beaconsIncoming)
 	c.beaconReceiver.start()
 
-	go c.beaconUpdates(stopCh)
-	go c.recordUpdates(stopCh)
-	<-stopCh
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		c.recordUpdates(stopCh)
+	}()
+	<-done
+	log.Println("COLLECTOR: Finished running. Shutting down")
 	for _, eventsource := range c.eventSources {
 		for _, receiver := range eventsource.receivers {
 			receiver.stop()
diff --git a/pkg/flow/messaging.go b/pkg/flow/messaging.go
index 32ed3a37..2bcdb3a9 100644
--- a/pkg/flow/messaging.go
+++ b/pkg/flow/messaging.go
@@ -2,7 +2,7 @@ package flow
 
 import (
 	"log"
-	"time"
+	"sync"
 
 	amqp "github.com/interconnectedcloud/go-amqp"
 	"github.com/skupperproject/skupper/pkg/event"
@@ -16,36 +16,31 @@ const (
 )
 
 type base struct {
+	stopOnce          sync.Once
+	done              chan struct{}
 	connectionFactory messaging.ConnectionFactory
-	closed            bool
-	client            messaging.Connection
 	incoming          chan []interface{}
 	outgoing          chan interface{}
 	address           string
 }
 
 func (c *base) stop() {
-	c.closed = true
-	if c.client != nil {
-		c.client.Close()
-	}
+	c.stopOnce.Do(func() { close(c.done) })
 }
 
 type sender struct {
 	base
 	sendSettled bool
-	ticker      *time.Ticker
-	request     *amqp.Message
 }
 
 func (c *sender) start() {
-	c.closed = false
 	go c.send()
 }
 
 func newSender(connectionFactory messaging.ConnectionFactory, address string, sendSettled bool, update chan interface{}) *sender {
 	return &sender{
 		base: base{
+			done:              make(chan struct{}),
 			connectionFactory: connectionFactory,
 			outgoing:          update,
 			address:           address,
@@ -55,15 +50,18 @@ func newSender(connectionFactory messaging.ConnectionFactory, address string, se
 }
 
 func (c *sender) send() {
-	c.ticker = time.NewTicker(5 * time.Second)
-	defer c.ticker.Stop()
-	for !c.closed {
-		err := c._send()
-		if err != nil {
-			log.Printf("COLLECTOR: Error sending out updates %s", err.Error())
+	for {
+		select {
+		case <-c.done:
+			log.Println("COLLECTOR: Flow process stopped sending")
+			return
+		default:
+			if err := c._send(); err != nil {
+				log.Printf("COLLECTOR: Error sending out updates %s", err.Error())
+			}
 		}
+
 	}
-	log.Println("COLLECTOR: Flow process stopped sending")
 }
 
 func (c *sender) _send() error {
@@ -71,7 +69,6 @@ func (c *sender) _send() error {
 	if err != nil {
 		return err
 	}
-	c.client = client
 	log.Printf("COLLECTOR: Connection for sender %s to %s established\n", c.address, c.connectionFactory.Url())
 	defer client.Close()
 
@@ -82,15 +79,18 @@ func (c *sender) _send() error {
 
 	defer sender.Close()
 
+	var request *amqp.Message
 	for {
 		select {
+		case <-c.done:
+			return nil
 		case update := <-c.outgoing:
 			if beacon, ok := update.(*BeaconRecord); ok {
 				msg, err := encodeBeacon(beacon)
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if heartbeat, ok := update.(*HeartbeatRecord); ok {
@@ -98,7 +98,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if fr, ok := update.(*FlushRecord); ok {
@@ -106,7 +106,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if site, ok := update.(*SiteRecord); ok {
@@ -114,7 +114,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if process, ok := update.(*ProcessRecord); ok {
@@ -122,7 +122,7 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
 			if host, ok := update.(*HostRecord); ok {
@@ -130,22 +130,19 @@ func (c *sender) _send() error {
 				if err != nil {
 					event.Recordf(FlowControllerEvent, "Failed to encode message for flow controller: %s", err.Error())
 				} else {
-					c.request = msg
+					request = msg
 				}
 			}
-		case <-c.ticker.C:
 		}
-		if c.request != nil {
-			c.request.SendSettled = c.sendSettled
-			err = sender.Send(c.request)
+		if request != nil {
+			request.SendSettled = c.sendSettled
+			err = sender.Send(request)
 			if err != nil {
 				return err
-			} else {
-				c.request = nil
 			}
+			request = nil
 		}
 	}
-	return nil
 }
 
 type receiver struct {
@@ -155,6 +152,7 @@ type receiver struct {
 func newReceiver(connectionFactory messaging.ConnectionFactory, address string, updates chan []interface{}) *receiver {
 	return &receiver{
 		base: base{
+			done:              make(chan struct{}),
 			connectionFactory: connectionFactory,
 			incoming:          updates,
 			address:           address,
@@ -163,15 +161,18 @@ func newReceiver(connectionFactory messaging.ConnectionFactory, address string,
 }
 
 func (r *receiver) start() {
-	r.closed = false
 	go r.receive()
 }
 
 func (r *receiver) receive() {
-	for !r.closed {
-		err := r._receive()
-		if err != nil {
-			log.Printf("COLLECTOR: Receiver %s %s\n", r.address, err.Error())
+	for {
+		select {
+		case <-r.done:
+			return
+		default:
+			if err := r._receive(); err != nil {
+				log.Printf("COLLECTOR: Receiver %s %s\n", r.address, err.Error())
+			}
 		}
 	}
 }
@@ -181,7 +182,6 @@ func (r *receiver) _receive() error {
 	if err != nil {
 		return err
 	}
-	r.client = client
 	log.Printf("COLLECTOR: Connection for receiver %s to %s established\n", r.address, r.connectionFactory.Url())
 	defer client.Close()
 
@@ -192,6 +192,11 @@ func (r *receiver) _receive() error {
 	defer receiver.Close()
 
 	for {
+		select {
+		case <-r.done:
+			return nil
+		default:
+		}
 		msg, err := receiver.Receive()
 		if err != nil {
 			return err
@@ -200,5 +205,4 @@ func (r *receiver) _receive() error {
 		results := decode(msg)
 		r.incoming <- results
 	}
-	return nil
 }
diff --git a/pkg/messaging/mock.go b/pkg/messaging/mock.go
new file mode 100644
index 00000000..e4f9fc7e
--- /dev/null
+++ b/pkg/messaging/mock.go
@@ -0,0 +1,189 @@
+package messaging
+
+import (
+	"fmt"
+	"sync"
+	"testing"
+	"time"
+
+	amqp "github.com/interconnectedcloud/go-amqp"
+)
+
+type MockConnectionFactory struct {
+	URL    string
+	Broker *Broker
+}
+
+// NewMockConnectionFactory creates an in-memory message brokering stub.
+// Messages will be sent multicast and are buffered for all unclosed receivers.
+func NewMockConnectionFactory(t *testing.T, url string) *MockConnectionFactory {
+	return &MockConnectionFactory{
+		URL:    url,
+		Broker: NewBroker(),
+	}
+}
+
+func (f *MockConnectionFactory) Connect() (Connection, error) {
+	c := &mockConnection{broker: f.Broker, done: make(chan struct{})}
+	return c, nil
+}
+
+func (f *MockConnectionFactory) Url() string {
+	return f.URL
+}
+
+type Broker struct {
+	mu     sync.Mutex
+	topics map[string]*multicast
+}
+
+func NewBroker() *Broker {
+	return &Broker{
+		topics: map[string]*multicast{},
+	}
+}
+func (b *Broker) AwaitReceivers(address string, n int) {
+	topic := b.get(address)
+	ct := func() int {
+		topic.mu.Lock()
+		defer topic.mu.Unlock()
+		return len(topic.receivers)
+	}
+	for n > ct() {
+		time.Sleep(time.Millisecond * 5)
+	}
+}
+
+func (b *Broker) subscribe(address string, r *mockReceiver) {
+	b.get(address).subscribe(r)
+}
+
+func (b *Broker) send(address string, msg *amqp.Message) {
+	b.get(address).send(msg)
+}
+
+func (b *Broker) get(address string) *multicast {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	t, ok := b.topics[address]
+	if !ok {
+		t = newMulticast()
+		b.topics[address] = t
+	}
+	return t
+}
+
+type multicast struct {
+	mu        sync.Mutex
+	receivers []*mockReceiver
+}
+
+func newMulticast() *multicast {
+	return &multicast{}
+}
+
+func (m *multicast) send(msg *amqp.Message) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	for _, r := range m.receivers {
+		r.send(msg)
+	}
+}
+
+func (m *multicast) subscribe(r *mockReceiver) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.receivers = append(m.receivers, r)
+}
+
+type mockConnection struct {
+	broker    *Broker
+	closeOnce sync.Once
+	done      chan struct{}
+}
+
+func (c *mockConnection) Sender(address string) (Sender, error) {
+	return &mockSender{address: address, connection: c, done: make(chan struct{})}, nil
+}
+
+func (c *mockConnection) Receiver(address string, credit uint32) (Receiver, error) {
+	r := &mockReceiver{connection: c, channel: make(chan *amqp.Message, credit), done: make(chan struct{})}
+	c.broker.subscribe(address, r)
+	return r, nil
+}
+
+func (c *mockConnection) Close() {
+	c.closeOnce.Do(func() { close(c.done) })
+}
+
+func (c *mockConnection) send(address string, msg *amqp.Message) error {
+	select {
+	case <-c.done:
+		return fmt.Errorf("channel closed")
+	default:
+		c.broker.send(address, msg)
+	}
+	return nil
+}
+
+func (c *mockConnection) receive(r *mockReceiver) (*amqp.Message, error) {
+	select {
+	case msg := <-r.channel:
+		if msg == nil {
+			return nil, fmt.Errorf("Failed to receive")
+		}
+		return msg, nil
+	case <-c.done:
+		return nil, fmt.Errorf("connection closed")
+	}
+}
+
+type mockSender struct {
+	connection *mockConnection
+	address    string
+	closeOnce  sync.Once
+	done       chan struct{}
+}
+
+func (s *mockSender) Send(msg *amqp.Message) error {
+	select {
+	case <-s.done:
+		return fmt.Errorf("sender closed")
+	default:
+		return s.connection.send(s.address, msg)
+	}
+}
+
+func (s *mockSender) Close() error {
+	s.closeOnce.Do(func() { close(s.done) })
+	return nil
+}
+
+type mockReceiver struct {
+	connection *mockConnection
+	channel    chan *amqp.Message
+	closeOnce  sync.Once
+	done       chan struct{}
+}
+
+func (r *mockReceiver) send(msg *amqp.Message) {
+	select {
+	case <-r.done:
+		return
+	default:
+		r.channel <- msg
+	}
+}
+
+func (r *mockReceiver) Receive() (*amqp.Message, error) {
+	return r.connection.receive(r)
+}
+
+func (r *mockReceiver) Accept(msg *amqp.Message) error {
+	return nil
+}
+
+func (r *mockReceiver) Close() error {
+	r.closeOnce.Do(func() { close(r.done) })
+	return nil
+}
diff --git a/pkg/utils/configs/manifest.go b/pkg/utils/configs/manifest.go
new file mode 100644
index 00000000..47b510bc
--- /dev/null
+++ b/pkg/utils/configs/manifest.go
@@ -0,0 +1,208 @@
+package configs
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/skupperproject/skupper/pkg/images"
+	"os"
+	"strings"
+)
+
+type SkupperImage struct {
+	Name       string `json:"name"`
+	SHA        string `json:"sha,omitempty"`
+	Repository string `json:"repository,omitempty"`
+}
+
+type Manifest struct {
+	Images    []SkupperImage     `json:"images"`
+	Variables *map[string]string `json:"variables,omitempty"`
+}
+
+type ManifestManager struct {
+	EnableSHA bool
+}
+
+type ManifestGenerator interface {
+	GetConfiguredManifest() Manifest
+	GetDefaultManifestWithEnv() Manifest
+	CreateFile(m Manifest) error
+}
+
+func (manager *ManifestManager) GetConfiguredManifest() Manifest {
+	return Manifest{
+		Images: getSkupperConfiguredImages(manager.EnableSHA),
+	}
+}
+
+func (manager *ManifestManager) GetDefaultManifestWithEnv() Manifest {
+	return Manifest{
+		Images:    getSkupperDefaultImages(),
+		Variables: getEnvironmentVariableMap(),
+	}
+}
+
+func (manager *ManifestManager) CreateFile(m Manifest) error {
+	// Encode the manifest image list as JSON.
+	manifestListJSON, err := json.MarshalIndent(m, "", "   ")
+	if err != nil {
+		return fmt.Errorf("Error encoding manifest image list: %v\n", err)
+
+	}
+
+	// Create a new file.
+	file, err := os.Create("manifest.json")
+	if err != nil {
+		return fmt.Errorf("Error creating file: %v\n", err)
+	}
+
+	// Write the JSON data to the file.
+	_, err = file.Write(manifestListJSON)
+	if err != nil {
+		return fmt.Errorf("Error writing to file: %v\n", err)
+	}
+
+	return nil
+}
+
+func getSkupperConfiguredImages(enableSHA bool) []SkupperImage {
+	return []SkupperImage{
+		{
+			Name:       images.GetRouterImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetRouterImageName()),
+			Repository: "https://github.com/skupperproject/skupper-router",
+		},
+		{
+			Name:       images.GetServiceControllerImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetServiceControllerImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetControllerPodmanImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetControllerPodmanImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetConfigSyncImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetConfigSyncImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetFlowCollectorImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetFlowCollectorImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       images.GetSiteControllerImageName(),
+			SHA:        getSHAIfEnabled(enableSHA, images.GetSiteControllerImageName()),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name: images.GetPrometheusServerImageName(),
+			SHA:  getSHAIfEnabled(enableSHA, images.GetPrometheusServerImageName()),
+		},
+		{
+			Name: images.GetOauthProxyImageName(),
+			SHA:  getSHAIfEnabled(enableSHA, images.GetOauthProxyImageName()),
+		},
+	}
+}
+
+func getSkupperDefaultImages() []SkupperImage {
+	return []SkupperImage{
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper-router",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ServiceControllerImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ControllerPodmanImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.FlowCollectorImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name:       strings.Join([]string{images.DefaultImageRegistry, images.SiteControllerImageName}, "/"),
+			Repository: "https://github.com/skupperproject/skupper",
+		},
+		{
+			Name: strings.Join([]string{images.PrometheusImageRegistry, images.PrometheusServerImageName}, "/"),
+		},
+		{
+			Name: strings.Join([]string{images.OauthProxyImageRegistry, images.OauthProxyImageName}, "/"),
+		},
+	}
+}
+
+func getEnvironmentVariableMap() *map[string]string {
+
+	envVariables := make(map[string]string)
+
+	skupperImageRegistry := os.Getenv(images.SkupperImageRegistryEnvKey)
+	if skupperImageRegistry != "" {
+		envVariables[images.SkupperImageRegistryEnvKey] = skupperImageRegistry
+	}
+
+	prometheusImageRegistry := os.Getenv(images.PrometheusImageRegistryEnvKey)
+	if prometheusImageRegistry != "" {
+		envVariables[images.PrometheusImageRegistry] = prometheusImageRegistry
+	}
+
+	oauthImageRegistry := os.Getenv(images.OauthProxyImageRegistry)
+	if oauthImageRegistry != "" {
+		envVariables[images.OauthProxyImageRegistry] = oauthImageRegistry
+	}
+
+	routerImage := os.Getenv(images.RouterImageEnvKey)
+	if routerImage != "" {
+		envVariables[images.RouterImageEnvKey] = routerImage
+	}
+
+	serviceControllerImage := os.Getenv(images.ServiceControllerImageEnvKey)
+	if serviceControllerImage != "" {
+		envVariables[images.ServiceControllerImageEnvKey] = serviceControllerImage
+	}
+
+	controllerPodmanImage := os.Getenv(images.ControllerPodmanImageEnvKey)
+	if controllerPodmanImage != "" {
+		envVariables[images.ControllerPodmanImageEnvKey] = controllerPodmanImage
+	}
+
+	configSyncImage := os.Getenv(images.ConfigSyncImageEnvKey)
+	if configSyncImage != "" {
+		envVariables[images.ConfigSyncImageEnvKey] = configSyncImage
+	}
+
+	flowCollectorImage := os.Getenv(images.FlowCollectorImageEnvKey)
+	if flowCollectorImage != "" {
+		envVariables[images.FlowCollectorImageEnvKey] = flowCollectorImage
+	}
+
+	prometheusImage := os.Getenv(images.PrometheusServerImageEnvKey)
+	if prometheusImage != "" {
+		envVariables[images.PrometheusServerImageEnvKey] = prometheusImage
+	}
+
+	oauthImage := os.Getenv(images.OauthProxyImageEnvKey)
+	if oauthImage != "" {
+		envVariables[images.OauthProxyImageEnvKey] = oauthImage
+	}
+
+	return &envVariables
+}
+
+func getSHAIfEnabled(enableSHA bool, imageName string) string {
+	if !enableSHA {
+		return ""
+	}
+	return images.GetSha(imageName)
+}


Test patch:
diff --git a/pkg/flow/collector_test.go b/pkg/flow/collector_test.go
index 3a7ce08a..8f024506 100644
--- a/pkg/flow/collector_test.go
+++ b/pkg/flow/collector_test.go
@@ -1,12 +1,15 @@
 package flow
 
 import (
+	"fmt"
 	"reflect"
 	"strconv"
+	"sync"
 	"testing"
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/skupperproject/skupper/pkg/messaging"
 	"gotest.tools/assert"
 )
 
@@ -59,54 +62,197 @@ func TestUpdates(t *testing.T) {
 }
 
 func TestRecordUpdates(t *testing.T) {
-	reg := prometheus.NewRegistry()
-	u, _ := time.ParseDuration("5m")
+	factory := messaging.NewMockConnectionFactory(t, "mockamqp://local")
 	fc := NewFlowCollector(FlowCollectorSpec{
 		Mode:              RecordMetrics,
 		Origin:            "origin",
-		PromReg:           reg,
-		ConnectionFactory: nil,
-		FlowRecordTtl:     u,
+		PromReg:           prometheus.NewRegistry(),
+		ConnectionFactory: factory,
+		FlowRecordTtl:     time.Minute * 5,
 	})
-	fc.metrics = fc.NewMetrics(fc.prometheusReg)
 
-	assert.Assert(t, fc != nil)
 	stopCh := make(chan struct{})
-	go fc.beaconUpdates(stopCh)
-	go fc.recordUpdates(stopCh)
+	fc.Start(stopCh)
+	factory.Broker.AwaitReceivers(BeaconAddress, 1)
+
+	conn, err := factory.Connect()
+	assert.Assert(t, err)
+	beaconSender, err := conn.Sender(BeaconAddress)
+	assert.Assert(t, err)
+	flushReceiver, err := conn.Receiver("sfe.1234", 32)
+	assert.Assert(t, err)
 
 	beacon := BeaconRecord{
 		Version:    1,
-		SourceType: "CONTROLLER",
+		SourceType: "ROUTER",
 		Address:    RecordPrefix + "1234",
 		Direct:     "sfe.1234",
 		Identity:   "1234",
 	}
-
-	var beacons []interface{}
-	beacons = append(beacons, beacon)
-	fc.beaconsIncoming <- beacons
-	time.Sleep(1 * time.Second)
-	eventSource, ok := fc.eventSources["1234"]
-	assert.Assert(t, ok)
-	assert.Equal(t, eventSource.Beacons, 1)
-	fc.beaconsIncoming <- beacons
-	time.Sleep(1 * time.Second)
-	assert.Equal(t, eventSource.Beacons, 2)
+	msgBeacon, err := encodeBeacon(&beacon)
+	assert.Assert(t, err)
+	assert.Assert(t, beaconSender.Send(msgBeacon))
+	assert.Assert(t, beaconSender.Send(msgBeacon))
 
 	heartbeat := HeartbeatRecord{
 		Version:  1,
 		Identity: "1234",
 		Source:   "sfe.1234",
 	}
+	msgHB, err := encodeHeartbeat(&heartbeat)
+	assert.Assert(t, err)
 
-	var heartbeats []interface{}
-	heartbeats = append(heartbeats, heartbeat)
-	fc.heartbeatsIncoming <- heartbeats
-	time.Sleep(1 * time.Second)
-	assert.Equal(t, eventSource.Heartbeats, 1)
+	factory.Broker.AwaitReceivers("mc/sfe.1234", 1)
+	heartBeatSender, err := conn.Sender("mc/sfe.1234")
+	heartBeatSender.Send(msgHB)
 
-	time.Sleep(1 * time.Second)
+	// we don't have a safe way to access the internal state of the collector
+	// without stopping it (or standing up a stub http mux). Instead, we'll
+	// wait for a flush record so we know that the internal state is settled
+	// before stopping it and cracking things open.
+	msg, err := flushReceiver.Receive()
+	assert.Assert(t, err)
+	assert.Equal(t, msg.Properties.Subject, "FLUSH")
 
 	close(stopCh)
+	<-fc.beaconReceiver.done // only way to tell the flow collector is done
+
+	eventSource, ok := fc.eventSources["1234"]
+	t.Log(fc.eventSources)
+	assert.Assert(t, ok)
+	assert.Equal(t, eventSource.Beacons, 2)
+	assert.Equal(t, eventSource.Heartbeats, 1)
+}
+
+func TestStartupShutdown(t *testing.T) {
+	scenarios := []struct {
+		Name        string
+		Spec        FlowCollectorSpec
+		Controllers int
+	}{
+		{
+			Name:        "RecordMetrics basic",
+			Controllers: 0,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordMetrics,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordMetrics four controllers",
+			Controllers: 4,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordMetrics,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordStatus basic",
+			Controllers: 0,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordStatus,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		}, {
+			Name:        "RecordStatus many controllers",
+			Controllers: 128,
+			Spec: FlowCollectorSpec{
+				Mode:              RecordStatus,
+				Origin:            "origin",
+				PromReg:           prometheus.NewRegistry(),
+				ConnectionFactory: messaging.NewMockConnectionFactory(t, "local://test"),
+				FlowRecordTtl:     5 * time.Minute,
+			},
+		},
+	}
+	for _, _testspec := range scenarios {
+		s := _testspec
+		t.Run(s.Name, func(t *testing.T) {
+			t.Parallel()
+			fc := NewFlowCollector(s.Spec)
+
+			done := make(chan struct{})
+
+			conn, err := s.Spec.ConnectionFactory.Connect()
+			assert.Assert(t, err)
+
+			flushReceivers := []messaging.Receiver{}
+			for site := 1; site < s.Controllers; site++ {
+
+				r, err := conn.Receiver(fmt.Sprintf("sfe.%d", site), 128)
+				assert.Assert(t, err)
+				flushReceivers = append(flushReceivers, r)
+
+				// simulated controller - beacons+heartbeats
+				go runMockCollector(t, strconv.Itoa(site), conn, done)
+			}
+			// start flow collector
+			fc.Start(done)
+			mockFactory := s.Spec.ConnectionFactory.(*messaging.MockConnectionFactory)
+			mockFactory.Broker.AwaitReceivers(BeaconAddress, 1)
+
+			// wait for flushes
+			var allFlushed sync.WaitGroup
+			for _, r := range flushReceivers {
+				allFlushed.Add(1)
+				go func(receiver messaging.Receiver) {
+					defer allFlushed.Done()
+					msg, err := receiver.Receive()
+					assert.Assert(t, err)
+					assert.Equal(t, msg.Properties.Subject, "FLUSH")
+				}(r)
+			}
+			allFlushed.Wait()
+			close(done)
+
+			// last thing FlowCollector does on stop is close the beaconReceiver
+			// make sure that is done
+			<-fc.beaconReceiver.done
+
+		})
+	}
+}
+func runMockCollector(t *testing.T, siteID string, conn messaging.Connection, done chan struct{}) {
+	ticker := time.NewTicker(time.Second / 15)
+	defer ticker.Stop()
+	mcAddr := fmt.Sprintf("mc/sfe.%s", siteID)
+	directAddr := fmt.Sprintf("sfe.%s", siteID)
+	heartSender, err := conn.Sender(mcAddr + ".heartbeats")
+	assert.Assert(t, err)
+	beaconSender, err := conn.Sender(BeaconAddress)
+	assert.Assert(t, err)
+	beacon := BeaconRecord{
+		Version:    1,
+		SourceType: "CONTROLLER",
+		Address:    mcAddr,
+		Direct:     directAddr,
+		Identity:   siteID,
+	}
+	msgBeacon, err := encodeBeacon(&beacon)
+	assert.Assert(t, err)
+	heartbeat := HeartbeatRecord{
+		Version:  1,
+		Identity: siteID,
+		Source:   directAddr,
+	}
+	assert.Assert(t, err)
+	for {
+		select {
+		case <-done:
+			return
+		case <-ticker.C:
+			assert.Assert(t, beaconSender.Send(msgBeacon))
+			heartbeat.Now = uint64(time.Now().UnixNano()) / uint64(time.Microsecond)
+			msgHeart, err := encodeHeartbeat(&heartbeat)
+			assert.Assert(t, err)
+			assert.Assert(t, heartSender.Send(msgHeart))
+		}
+	}
 }
diff --git a/pkg/flow/controller_test.go b/pkg/flow/controller_test.go
index 90e136f5..27adcbaf 100644
--- a/pkg/flow/controller_test.go
+++ b/pkg/flow/controller_test.go
@@ -145,10 +145,7 @@ func TestUpdateRecords(t *testing.T) {
 	var flushes []interface{}
 	flushes = append(flushes, flush)
 	fc.flushIncoming <- flushes
-
-	time.Sleep(5 * time.Second)
 	close(stopCh)
-
 }
 
 func TestSiteController(t *testing.T) {
diff --git a/pkg/service_sync/messaging_test.go b/pkg/service_sync/messaging_test.go
index 4ab925c8..bab9c736 100644
--- a/pkg/service_sync/messaging_test.go
+++ b/pkg/service_sync/messaging_test.go
@@ -1,11 +1,9 @@
 package service_sync
 
 import (
-	"fmt"
 	"reflect"
 	"testing"
 
-	amqp "github.com/interconnectedcloud/go-amqp"
 	"gotest.tools/assert"
 
 	"github.com/skupperproject/skupper/api/types"
@@ -13,173 +11,6 @@ import (
 	"github.com/skupperproject/skupper/pkg/messaging"
 )
 
-type multicast struct {
-	receivers []*MockReceiver
-	channel   chan int
-}
-
-func newMulticast() *multicast {
-	return &multicast{
-		channel: make(chan int, 5),
-	}
-}
-
-func (m *multicast) send(msg *amqp.Message) {
-	for _, r := range m.receivers {
-		r.send(msg)
-	}
-}
-
-func (m *multicast) subscribe(r *MockReceiver) {
-	m.receivers = append(m.receivers, r)
-	m.channel <- len(m.receivers)
-}
-
-func (m *multicast) waitForReceivers(count int) {
-	for len(m.receivers) < count {
-		<-m.channel
-	}
-}
-
-type broker struct {
-	topics map[string]*multicast
-}
-
-func newBroker() *broker {
-	return &broker{
-		topics: map[string]*multicast{},
-	}
-}
-
-func (b *broker) newTopic(address string) *multicast {
-	t := newMulticast()
-	b.topics[address] = t
-	return t
-}
-
-func (b *broker) subscribe(address string, r *MockReceiver) {
-	t, ok := b.topics[address]
-	if !ok {
-		t = newMulticast()
-		b.topics[address] = t
-	}
-	t.subscribe(r)
-}
-
-func (b *broker) send(address string, msg *amqp.Message) {
-	if t, ok := b.topics[address]; ok {
-		t.send(msg)
-	}
-}
-
-type MockConnectionFactory struct {
-	url    string
-	topics *broker
-}
-
-func NewMockConnectionFactory(url string) *MockConnectionFactory {
-	return &MockConnectionFactory{
-		url:    url,
-		topics: newBroker(),
-	}
-}
-
-func (f *MockConnectionFactory) Connect() (messaging.Connection, error) {
-	c := &MockConnection{topics: f.topics, channel: make(chan bool)}
-	return c, nil
-}
-
-func (f *MockConnectionFactory) Url() string {
-	return f.url
-}
-
-type MockConnection struct {
-	topics  *broker
-	channel chan bool
-	closed  bool
-}
-
-func (c *MockConnection) Sender(address string) (messaging.Sender, error) {
-	return &MockSender{address: address, connection: c}, nil
-}
-
-func (c *MockConnection) Receiver(address string, credit uint32) (messaging.Receiver, error) {
-	r := &MockReceiver{connection: c, channel: make(chan *amqp.Message, credit)}
-	c.topics.subscribe(address, r)
-	return r, nil
-}
-
-func (c *MockConnection) Close() {
-	if !c.closed {
-		c.closed = true
-		close(c.channel)
-	}
-}
-
-func (c *MockConnection) send(address string, msg *amqp.Message) error {
-	if c.closed {
-		return fmt.Errorf("Channel closed")
-	}
-	c.topics.send(address, msg)
-	return nil
-}
-
-func (c *MockConnection) receive(r *MockReceiver) (*amqp.Message, error) {
-	if c.closed {
-		return nil, fmt.Errorf("Channel closed")
-	}
-	select {
-	case msg := <-r.channel:
-		if msg == nil {
-			return nil, fmt.Errorf("Failed to receive")
-		}
-		return msg, nil
-	case <-c.channel:
-		return nil, fmt.Errorf("Channel closed")
-	}
-}
-
-type MockSender struct {
-	connection *MockConnection
-	address    string
-}
-
-func (s *MockSender) Send(msg *amqp.Message) error {
-	return s.connection.send(s.address, msg)
-}
-
-func (s *MockSender) Close() error {
-	return nil
-}
-
-type MockReceiver struct {
-	connection *MockConnection
-	channel    chan *amqp.Message
-	closed     bool
-}
-
-func (r *MockReceiver) send(msg *amqp.Message) {
-	if !r.closed {
-		r.channel <- msg
-	}
-}
-
-func (r *MockReceiver) Receive() (*amqp.Message, error) {
-	return r.connection.receive(r)
-}
-
-func (r *MockReceiver) Accept(msg *amqp.Message) error {
-	return nil
-}
-
-func (r *MockReceiver) Close() error {
-	if !r.closed {
-		close(r.channel)
-		r.closed = true
-	}
-	return nil
-}
-
 func TestSender(t *testing.T) {
 	stopper := make(chan struct{})
 	event.StartDefaultEventStore(stopper)
@@ -234,7 +65,7 @@ func TestSender(t *testing.T) {
 
 	for _, s := range scenarios {
 		t.Run(s.name, func(t *testing.T) {
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			outgoing := make(chan ServiceUpdate)
 			sender := newSender(factory, outgoing, event.NewDefaultEventLogger())
 			sender.start()
@@ -326,16 +157,21 @@ func TestReceiver(t *testing.T) {
 
 	for _, s := range scenarios {
 		t.Run(s.name, func(t *testing.T) {
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			incoming := make(chan ServiceUpdate)
 			receiver := newReceiver(factory, incoming, event.NewDefaultEventLogger())
 			receiver.start()
-			factory.topics.newTopic(ServiceSyncAddress).waitForReceivers(1)
+			factory.Broker.AwaitReceivers(ServiceSyncAddress, 1)
+
+			tConn, err := factory.Connect()
+			assert.Assert(t, err)
+			serviceSyncSender, err := tConn.Sender(ServiceSyncAddress)
+			assert.Assert(t, err)
 
 			for _, update := range s.updates {
 				msg, err := encode(&update)
 				assert.Assert(t, err)
-				factory.topics.send(ServiceSyncAddress, msg)
+				serviceSyncSender.Send(msg)
 				received := <-incoming
 
 				assert.Assert(t, err)
diff --git a/pkg/service_sync/service_sync_test.go b/pkg/service_sync/service_sync_test.go
index 79445019..893269fd 100644
--- a/pkg/service_sync/service_sync_test.go
+++ b/pkg/service_sync/service_sync_test.go
@@ -1,17 +1,19 @@
 package service_sync
 
 import (
-	vanClient "github.com/skupperproject/skupper/client"
-	"k8s.io/client-go/kubernetes/fake"
 	"reflect"
 	"strings"
 	"testing"
 	"time"
 
+	vanClient "github.com/skupperproject/skupper/client"
+	"k8s.io/client-go/kubernetes/fake"
+
 	"gotest.tools/assert"
 
 	"github.com/skupperproject/skupper/api/types"
 	"github.com/skupperproject/skupper/pkg/event"
+	"github.com/skupperproject/skupper/pkg/messaging"
 )
 
 type updateRecord struct {
@@ -126,14 +128,14 @@ func TestServiceSync(t *testing.T) {
 		event.StartDefaultEventStore(stopper)
 		t.Run(s.name, func(t *testing.T) {
 			//hook up two instances of service sync
-			factory := NewMockConnectionFactory("test-channel")
+			factory := messaging.NewMockConnectionFactory(t, "test-channel")
 			updates1 := newUpdateChannel()
 			updates2 := newUpdateChannel()
 			site1 := NewServiceSync("foo", 0, "v1", factory, updates1.handler, event.NewDefaultEventLogger())
 			site2 := NewServiceSync("bar", 0, "v1", factory, updates2.handler, event.NewDefaultEventLogger())
 			site1.Start(stopper)
 			site2.Start(stopper)
-			factory.topics.newTopic(ServiceSyncAddress).waitForReceivers(2)
+			factory.Broker.AwaitReceivers(ServiceSyncAddress, 2)
 			site1.LocalDefinitionsUpdated(s.site1.definitions)
 			site2.LocalDefinitionsUpdated(s.site2.definitions)
 			site1Done := false
@@ -184,7 +186,7 @@ func TestRemoveStaleDefinitions(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
@@ -231,7 +233,7 @@ func TestUpdateRemoteDefinitions(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
@@ -295,7 +297,7 @@ func TestLocalDefinitionsUpdated(t *testing.T) {
 	event.StartDefaultEventStore(stopper)
 
 	updates := newUpdateCollector()
-	factory := NewMockConnectionFactory("test-channel")
+	factory := messaging.NewMockConnectionFactory(t, "test-channel")
 	site := NewServiceSync("foo", 0, "v1", factory, updates.handler, event.NewDefaultEventLogger())
 
 	defs := map[string]types.ServiceInterface{
diff --git a/pkg/utils/configs/manifest_test.go b/pkg/utils/configs/manifest_test.go
new file mode 100644
index 00000000..c1bed966
--- /dev/null
+++ b/pkg/utils/configs/manifest_test.go
@@ -0,0 +1,137 @@
+package configs
+
+import (
+	"fmt"
+	"github.com/skupperproject/skupper/pkg/images"
+	"gotest.tools/assert"
+	"os"
+	"strings"
+	"testing"
+)
+
+func TestManifestManager(t *testing.T) {
+	testcases := []struct {
+		title                          string
+		envVariablesWithValue          []string
+		expectedConfiguredManifest     Manifest
+		expectedDefaultManifestWithEnv Manifest
+	}{
+		{
+			title: "configured manifest has different images that the default manifest",
+			envVariablesWithValue: []string{
+				images.SkupperImageRegistryEnvKey,
+				images.ConfigSyncImageEnvKey,
+				images.RouterImageEnvKey},
+			expectedConfiguredManifest: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: "SKUPPER_CONFIG_SYNC_IMAGE_TESTING",
+					},
+					{
+						Name: "QDROUTERD_IMAGE_TESTING"},
+				},
+			},
+			expectedDefaultManifestWithEnv: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+				Variables: &map[string]string{
+					images.SkupperImageRegistryEnvKey: "SKUPPER_IMAGE_REGISTRY_TESTING",
+					images.ConfigSyncImageEnvKey:      "SKUPPER_CONFIG_SYNC_IMAGE_TESTING",
+					images.RouterImageEnvKey:          "QDROUTERD_IMAGE_TESTING",
+				},
+			},
+		},
+		{
+			title:                 "configured manifest the same images that the default manifest",
+			envVariablesWithValue: []string{},
+			expectedConfiguredManifest: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+			},
+			expectedDefaultManifestWithEnv: Manifest{
+				Images: []SkupperImage{
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.ConfigSyncImageName}, "/"),
+					},
+					{
+						Name: strings.Join([]string{images.DefaultImageRegistry, images.RouterImageName}, "/"),
+					},
+				},
+				Variables: &map[string]string{},
+			},
+		},
+	}
+
+	for _, c := range testcases {
+
+		setUpEnvVariables(c.envVariablesWithValue)
+
+		manifestManager := ManifestManager{EnableSHA: false}
+		configuredManifest := manifestManager.GetConfiguredManifest()
+		defaultManifest := manifestManager.GetDefaultManifestWithEnv()
+
+		for _, expectedImage := range c.expectedConfiguredManifest.Images {
+
+			configuredImage := getSkupperImageFromManifest(&configuredManifest, expectedImage.Name)
+
+			assert.Check(t, configuredImage != nil)
+		}
+
+		assert.Check(t, configuredManifest.Variables == nil)
+
+		for _, expectedImage := range c.expectedDefaultManifestWithEnv.Images {
+
+			defaultImage := getSkupperImageFromManifest(&defaultManifest, expectedImage.Name)
+
+			assert.Check(t, defaultImage != nil)
+		}
+
+		assert.DeepEqual(t, c.expectedDefaultManifestWithEnv.Variables, defaultManifest.Variables)
+
+		clearEnvVariables(c.envVariablesWithValue)
+	}
+}
+
+func setUpEnvVariables(variables []string) {
+	for _, variable := range variables {
+		err := os.Setenv(variable, strings.Join([]string{variable, "TESTING"}, "_"))
+		if err != nil {
+			fmt.Printf("error while setting %s", variable)
+			return
+		}
+	}
+}
+
+func clearEnvVariables(variables []string) {
+	for _, variable := range variables {
+		err := os.Unsetenv(variable)
+		if err != nil {
+			fmt.Printf("error while unssetting %s", variable)
+			return
+		}
+	}
+}
+
+func getSkupperImageFromManifest(m *Manifest, name string) *SkupperImage {
+
+	for _, skImage := range m.Images {
+
+		if skImage.Name == name {
+			return &skImage
+		}
+	}
+
+	return nil
+}