Adds buffered chan to inventory client events in all RMs (#61)

Author: Raphael V Rosa

attestationstatus/pkg/invclient/invclient.go

@@ -29,6 +29,8 @@ const (
 	//  and 600ms per request on average.
 	// TODO: fine tune this longer timeout based on target scale and inventory client batch size.
 	ListAllDefaultTimeout = time.Minute // Longer timeout for reconciling all resources
+	// eventsWatcherBufSize is the buffer size for the events channel.
+	eventsWatcherBufSize = 10
 )
 
 var (
@@ -44,7 +46,7 @@ func StartInventoryClient(wg *sync.WaitGroup, conf config.AttestationStatusMgrCo
 
 	zlog.InfraSec().Info().Msg("Starting Inventory gRPC Client")
 
-	events := make(chan *inv_client.WatchEvents)
+	events := make(chan *inv_client.WatchEvents, eventsWatcherBufSize)
 	cfg := inv_client.InventoryClientConfig{
 		Name:                      "attestmgr",
 		Address:                   conf.InventoryAddr,

host/pkg/hostmgr/hostmgr.go

@@ -44,6 +44,8 @@ const (
 	backoffRetries  = uint64(5)
 	// SetHostConnectionLost does 2 operation with Inventory, plus we leave some slack time.
 	nOperationInventoryHostConnLost = 2.05
+	// eventsWatcherBufSize is the buffer size for the events channel.
+	eventsWatcherBufSize = 10
 )
 
 func EnableAuth(enable bool) Option {
@@ -105,7 +107,7 @@ func StartInvGrpcCli(
 	}
 	zlog.InfraSec().Info().Msg("initial Inv Grpc Client start.")
 
-	events := make(chan *inv_client.WatchEvents)
+	events := make(chan *inv_client.WatchEvents, eventsWatcherBufSize)
 	cfg := inv_client.InventoryClientConfig{
 		Name:                      "hostmgr",
 		Address:                   conf.InventoryAddr,

maintenance/pkg/maintmgr/maintmgr.go

@@ -29,6 +29,9 @@ import (
 
 var zlog = logging.GetLogger("MaintenanceManager")
 
+// eventsWatcherBufSize is the buffer size for the events channel.
+var eventsWatcherBufSize = 10
+
 // TODO(max): remove global instances.
 var invMgrCli invclient.InvGrpcClient
 
@@ -104,7 +107,7 @@ func StartInvGrpcCli(
 		inv_v1.ResourceKind_RESOURCE_KIND_REPEATEDSCHEDULE,
 	}
 
-	events := make(chan *inv_client.WatchEvents)
+	events := make(chan *inv_client.WatchEvents, eventsWatcherBufSize)
 
 	cfg := inv_client.InventoryClientConfig{
 		Name:                      "maintmgr",

networking/internal/clients/invclient.go

@@ -30,6 +30,8 @@ const (
 
 	// TODO: fine tune this longer timeout based on target scale and inventory client batch size.
 	ListAllDefaultTimeout = time.Minute // Longer timeout for reconciling all resources
+	// eventsWatcherBufSize is the buffer size for the events channel.
+	eventsWatcherBufSize = 10
 )
 
 var (
@@ -94,7 +96,7 @@ func NewNetInventoryClientWithOptions(opts ...Option) (*NetInventoryClient, erro
 	for _, opt := range opts {
 		opt(&options)
 	}
-	eventsWatcher := make(chan *client.WatchEvents)
+	eventsWatcher := make(chan *client.WatchEvents, eventsWatcherBufSize)
 	wg := sync.WaitGroup{}
 	clientCfg := client.InventoryClientConfig{
 		Name:                      clientName,

os-resource/internal/invclient/invclient.go

@@ -35,6 +35,8 @@ var (
 const (
 	DefaultInventoryTimeout = 5 * time.Second
 	ListAllDefaultTimeout   = time.Minute // Longer timeout for reconciling all resources
+	// eventsWatcherBufSize is the buffer size for the events channel.
+	eventsWatcherBufSize = 10
 )
 
 type InventoryClient struct {
@@ -57,7 +59,7 @@ func NewInventoryClient(
 		inv_v1.ResourceKind_RESOURCE_KIND_TENANT,
 	}
 
-	events := make(chan *inv_client.WatchEvents)
+	events := make(chan *inv_client.WatchEvents, eventsWatcherBufSize)
 	cfg := inv_client.InventoryClientConfig{
 		Name:                      "osrm_clientsrv",
 		Address:                   invsvcaddr,

telemetry/internal/invclient/invclient.go

@@ -31,6 +31,8 @@ var (
 	zlog       = logging.GetLogger(clientName)
 
 	inventoryTimeout = flag.Duration("invTimeout", DefaultInventoryTimeout, "Inventory API calls timeout")
+	// eventsWatcherBufSize is the buffer size for the events channel.
+	eventsWatcherBufSize = 10
 )
 
 type TelemetryInventoryClient struct {
@@ -107,7 +109,7 @@ func NewTelemetryInventoryClientWithOptions(wg *sync.WaitGroup, opts ...Option)
 		opt(&options)
 	}
 
-	eventsWatcher := make(chan *client.WatchEvents)
+	eventsWatcher := make(chan *client.WatchEvents, eventsWatcherBufSize)
 	cfg := client.InventoryClientConfig{
 		Name:                      clientName,
 		Address:                   options.InventoryAddress,