Migrate from bidi streaming to unary RPC

Add SendListenerMessage unary RPC to ListenerService, enabling
NodeListener to use simple request-reply instead of bidirectional
streaming. This improves AWS ALB compatibility (no HTTP/2 PING
needed), simplifies the code (2 goroutines instead of 3, no unacked
message queue), and enables better load balancing.

Changes:
- Proto: add SendListenerMessage RPC to ListenerService
- Server: add SendListenerMessage handler with Redis push and ACK
- BaseListener: add RunUnary (2-goroutine lifecycle), SendUnaryMessage,
  and gRPC built-in retry config for transient failures
- NodeListener: switch from Run/SendMessage to RunUnary/SendUnaryMessage
- Tests: add coverage for unary handler, base listener RunUnary,
  and node listener sendMessages paths

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: xutongNV

src/operator/event_listener.go

@@ -33,7 +33,7 @@ import (
 	pb "go.corp.nvidia.com/osmo/proto/operator"
 )
 
-// EventListener manages the bidirectional gRPC stream connection for Kubernetes events
+// EventListener manages unary RPC calls for Kubernetes events
 type EventListener struct {
 	*utils.BaseListener
 	args utils.ListenerArgs
@@ -51,12 +51,12 @@ func NewEventListener(args utils.ListenerArgs, inst *utils.Instruments) *EventLi
 	return el
 }
 
-// Run manages the bidirectional streaming lifecycle
+// Run manages the unary RPC lifecycle
 func (el *EventListener) Run(ctx context.Context) error {
 	ch := make(chan *pb.ListenerMessage, el.args.EventChanSize)
 	return el.BaseListener.Run(
 		ctx,
-		"Connected to the service, event listener stream established",
+		"Connected to operator service, unary event listener established",
 		ch,
 		el.watchEvents,
 		el.sendMessages,
@@ -70,6 +70,15 @@ func (el *EventListener) sendMessages(
 ) error {
 	el.Logf("Starting message sender for event channel")
 	defer el.Logf("Stopping event message sender")
+	defer el.DrainMessageChannel(ch)
+
+	// Resend any unacked messages from a previous connection before processing new ones
+	err := el.GetUnackedMessages().ResendAll(
+		ctx, el.SendMessage)
+	if err != nil {
+		el.Logf("Failed to resend unacked messages: %v", err)
+		return fmt.Errorf("failed to resend unacked messages: %w", err)
+	}
 
 	progressTicker := time.NewTicker(time.Duration(el.args.ProgressFrequencySec) * time.Second)
 	defer progressTicker.Stop()
@@ -90,7 +99,8 @@ func (el *EventListener) sendMessages(
 				el.inst.MessageChannelClosedUnexpectedly.Add(ctx, 1, el.MetricAttrs)
 				return fmt.Errorf("event watcher stopped")
 			}
-			if err := el.BaseListener.SendMessage(ctx, msg); err != nil {
+			if err := el.SendMessage(ctx, msg); err != nil {
+				el.GetUnackedMessages().AddMessageDropOldest(msg)
 				return fmt.Errorf("failed to send message: %w", err)
 			}
 		}

src/operator/event_listener_test.go

@@ -17,12 +17,15 @@
 package main
 
 import (
+	"context"
+	"fmt"
 	"testing"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
+	"go.corp.nvidia.com/osmo/operator/utils"
 	pb "go.corp.nvidia.com/osmo/proto/operator"
 )
 
@@ -227,6 +230,107 @@ func TestEventKeyComposite(t *testing.T) {
 	}
 }
 
+// TestEventListenerDrainPopulatesUnackedQueue verifies that drainMessageChannel
+// moves pending channel messages into the unacked queue using drop-oldest semantics.
+func TestEventListenerDrainPopulatesUnackedQueue(t *testing.T) {
+	args := utils.ListenerArgs{
+		EventChanSize:        10,
+		ProgressFrequencySec: 60,
+	}
+	inst := utils.NewNoopInstruments()
+	el := NewEventListener(args, inst)
+
+	ch := make(chan *pb.ListenerMessage, 5)
+	// Pre-fill channel with messages
+	for i := 0; i < 3; i++ {
+		ch <- &pb.ListenerMessage{Uuid: fmt.Sprintf("drain-%d", i)}
+	}
+
+	el.DrainMessageChannel(ch)
+
+	unacked := el.GetUnackedMessages()
+	if unacked.Qsize() != 3 {
+		t.Errorf("Qsize() = %d, expected 3 after drain", unacked.Qsize())
+	}
+
+	messages := unacked.ListMessages()
+	for i, msg := range messages {
+		expected := fmt.Sprintf("drain-%d", i)
+		if msg.Uuid != expected {
+			t.Errorf("messages[%d].Uuid = %s, expected %s", i, msg.Uuid, expected)
+		}
+	}
+}
+
+// TestEventListenerDrainRespectsCapacity verifies that drainMessageChannel
+// uses drop-oldest semantics when the queue is at capacity.
+func TestEventListenerDrainRespectsCapacity(t *testing.T) {
+	args := utils.ListenerArgs{
+		EventChanSize:        10,
+		MaxUnackedMessages:   3,
+		ProgressFrequencySec: 60,
+	}
+	inst := utils.NewNoopInstruments()
+	el := NewEventListener(args, inst)
+
+	ch := make(chan *pb.ListenerMessage, 10)
+	// Pre-fill channel with 5 messages (exceeds capacity of 3)
+	for i := 0; i < 5; i++ {
+		ch <- &pb.ListenerMessage{Uuid: fmt.Sprintf("drain-%d", i)}
+	}
+
+	el.DrainMessageChannel(ch)
+
+	unacked := el.GetUnackedMessages()
+	if unacked.Qsize() != 3 {
+		t.Errorf("Qsize() = %d, expected 3 (capped at capacity)", unacked.Qsize())
+	}
+
+	// Oldest messages should have been evicted, newest 3 remain
+	messages := unacked.ListMessages()
+	for i, msg := range messages {
+		expected := fmt.Sprintf("drain-%d", i+2)
+		if msg.Uuid != expected {
+			t.Errorf("messages[%d].Uuid = %s, expected %s", i, msg.Uuid, expected)
+		}
+	}
+}
+
+// TestEventListenerResendBeforeNewMessages verifies that ResendAll is invoked
+// on the unacked queue at the start of sendMessages, before reading from the channel.
+func TestEventListenerResendBeforeNewMessages(t *testing.T) {
+	// This test verifies the ResendAll integration at the UnackMessages level,
+	// since sendMessages requires a live gRPC connection for SendMessage.
+	unacked := utils.NewUnackMessages(10)
+	unacked.AddMessageDropOldest(&pb.ListenerMessage{Uuid: "old-1"})
+	unacked.AddMessageDropOldest(&pb.ListenerMessage{Uuid: "old-2"})
+
+	var sentOrder []string
+	sendFunc := func(_ context.Context, msg *pb.ListenerMessage) error {
+		sentOrder = append(sentOrder, msg.Uuid)
+		return nil
+	}
+
+	ctx := context.Background()
+	err := unacked.ResendAll(ctx, sendFunc)
+	if err != nil {
+		t.Fatalf("ResendAll failed: %v", err)
+	}
+
+	// Verify old messages were sent in order
+	if len(sentOrder) != 2 {
+		t.Fatalf("Expected 2 messages sent, got %d", len(sentOrder))
+	}
+	if sentOrder[0] != "old-1" || sentOrder[1] != "old-2" {
+		t.Errorf("Expected send order [old-1, old-2], got %v", sentOrder)
+	}
+
+	// Queue should be empty after successful resend
+	if unacked.Qsize() != 0 {
+		t.Errorf("Qsize() = %d, expected 0 after successful resend", unacked.Qsize())
+	}
+}
+
 // TestListenerMessageTypeAssertion verifies the message is correctly typed
 func TestListenerMessageTypeAssertion(t *testing.T) {
 	event := &corev1.Event{

src/operator/node_listener.go

@@ -59,12 +59,12 @@ func NewNodeListener(
 	return nl
 }
 
-// Run manages the bidirectional streaming lifecycle
+// Run manages the unary RPC lifecycle for node events
 func (nl *NodeListener) Run(ctx context.Context) error {
 	ch := make(chan *pb.ListenerMessage, nl.args.NodeUpdateChanSize)
 	return nl.BaseListener.Run(
 		ctx,
-		"Connected to operator service, node stream established",
+		"Connected to operator service, unary node listener established",
 		ch,
 		nl.watchNodes,
 		nl.sendMessages,
@@ -96,7 +96,7 @@ func (nl *NodeListener) sendMessages(
 				nl.inst.MessageChannelClosedUnexpectedly.Add(ctx, 1, nl.MetricAttrs)
 				return fmt.Errorf("node watcher stopped")
 			}
-			if err := nl.BaseListener.SendMessage(ctx, msg); err != nil {
+			if err := nl.SendMessage(ctx, msg); err != nil {
 				return fmt.Errorf("failed to send node message: %w", err)
 			}
 		}

src/operator/node_listener_test.go

@@ -17,10 +17,12 @@
 package main
 
 import (
+	"context"
 	"testing"
 	"time"
 
 	"go.corp.nvidia.com/osmo/operator/utils"
+	pb "go.corp.nvidia.com/osmo/proto/operator"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -93,3 +95,154 @@ func TestNewNodeListener(t *testing.T) {
 		t.Error("Expected unackedMessages to be initialized")
 	}
 }
+
+func TestNodeListener_SendMessages_ChannelClosed(t *testing.T) {
+	args := utils.ListenerArgs{
+		ServiceURL:           "http://localhost:8000",
+		Backend:              "test-backend",
+		Namespace:            "osmo",
+		NodeUpdateChanSize:   100,
+		MaxUnackedMessages:   100,
+		NodeConditionPrefix:  "osmo.nvidia.com/",
+		ProgressDir:          "/tmp/osmo/operator/",
+		ProgressFrequencySec: 15,
+	}
+
+	nodeConditionRules := utils.NewNodeConditionRules()
+	listener := NewNodeListener(args, nodeConditionRules, utils.NewNoopInstruments())
+
+	ch := make(chan *pb.ListenerMessage, 10)
+	close(ch)
+
+	ctx := context.Background()
+	err := listener.sendMessages(ctx, ch)
+	if err == nil {
+		t.Fatal("expected error when channel is closed, got nil")
+	}
+	expectedMsg := "node watcher stopped"
+	if err.Error() != expectedMsg {
+		t.Errorf("expected error %q, got %q", expectedMsg, err.Error())
+	}
+}
+
+func TestNodeListener_SendMessages_ContextCancelled(t *testing.T) {
+	args := utils.ListenerArgs{
+		ServiceURL:           "http://localhost:8000",
+		Backend:              "test-backend",
+		Namespace:            "osmo",
+		NodeUpdateChanSize:   100,
+		MaxUnackedMessages:   100,
+		NodeConditionPrefix:  "osmo.nvidia.com/",
+		ProgressDir:          "/tmp/osmo/operator/",
+		ProgressFrequencySec: 15,
+	}
+
+	nodeConditionRules := utils.NewNodeConditionRules()
+	listener := NewNodeListener(args, nodeConditionRules, utils.NewNoopInstruments())
+
+	ch := make(chan *pb.ListenerMessage, 10)
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	err := listener.sendMessages(ctx, ch)
+	if err != nil {
+		t.Fatalf("expected nil error on context cancellation, got: %v", err)
+	}
+}
+
+func TestNodeListener_SendMessages_ProgressReport(t *testing.T) {
+	args := utils.ListenerArgs{
+		ServiceURL:           "http://localhost:8000",
+		Backend:              "test-backend",
+		Namespace:            "osmo",
+		NodeUpdateChanSize:   100,
+		MaxUnackedMessages:   100,
+		NodeConditionPrefix:  "osmo.nvidia.com/",
+		ProgressDir:          "/tmp/osmo/operator/",
+		ProgressFrequencySec: 1,
+	}
+
+	nodeConditionRules := utils.NewNodeConditionRules()
+	listener := NewNodeListener(args, nodeConditionRules, utils.NewNoopInstruments())
+
+	ch := make(chan *pb.ListenerMessage, 10)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	errChan := make(chan error, 1)
+	go func() {
+		errChan <- listener.sendMessages(ctx, ch)
+	}()
+
+	select {
+	case err := <-errChan:
+		if err != nil {
+			t.Fatalf("expected nil error, got: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("test timed out")
+	}
+}
+
+func TestNodeListener_BuildResourceMessage(t *testing.T) {
+	args := utils.ListenerArgs{
+		ServiceURL:           "http://localhost:8000",
+		Backend:              "test-backend",
+		Namespace:            "osmo",
+		NodeUpdateChanSize:   100,
+		MaxUnackedMessages:   100,
+		NodeConditionPrefix:  "osmo.nvidia.com/",
+		ProgressDir:          "/tmp/osmo/operator/",
+		ProgressFrequencySec: 15,
+	}
+
+	nodeConditionRules := utils.NewNodeConditionRules()
+	listener := NewNodeListener(args, nodeConditionRules, utils.NewNoopInstruments())
+
+	tracker := utils.NewNodeStateTracker(1 * time.Minute)
+	node := &corev1.Node{
+		ObjectMeta: metav1.ObjectMeta{
+			Labels: map[string]string{
+				"kubernetes.io/hostname": "test-node",
+			},
+		},
+		Status: corev1.NodeStatus{
+			Conditions: []corev1.NodeCondition{
+				{
+					Type:   corev1.NodeReady,
+					Status: corev1.ConditionTrue,
+				},
+			},
+		},
+	}
+
+	msg := listener.buildResourceMessage(node, tracker, false, nil)
+	if msg == nil {
+		t.Fatal("expected non-nil message for new node")
+	}
+
+	updateNode := msg.GetUpdateNode()
+	if updateNode == nil {
+		t.Fatal("expected UpdateNode body")
+	}
+	if updateNode.Hostname != "test-node" {
+		t.Errorf("expected hostname test-node, got %s", updateNode.Hostname)
+	}
+
+	// Second call should return nil (no change)
+	msg2 := listener.buildResourceMessage(node, tracker, false, nil)
+	if msg2 != nil {
+		t.Error("expected nil message for unchanged node")
+	}
+
+	// Delete should always return a message
+	msg3 := listener.buildResourceMessage(node, tracker, true, nil)
+	if msg3 == nil {
+		t.Fatal("expected non-nil message for delete event")
+	}
+	if !msg3.GetUpdateNode().Delete {
+		t.Error("expected Delete to be true")
+	}
+}
+

src/operator/node_usage_listener.go

@@ -32,7 +32,7 @@ import (
 	pb "go.corp.nvidia.com/osmo/proto/operator"
 )
 
-// NodeUsageListener manages the bidirectional gRPC stream for pod resource usage
+// NodeUsageListener manages unary RPC calls for pod resource usage
 type NodeUsageListener struct {
 	*utils.BaseListener
 	args       utils.ListenerArgs
@@ -52,12 +52,12 @@ func NewNodeUsageListener(args utils.ListenerArgs, inst *utils.Instruments) *Nod
 	return nul
 }
 
-// Run manages the bidirectional streaming lifecycle
+// Run manages the unary RPC lifecycle
 func (nul *NodeUsageListener) Run(ctx context.Context) error {
 	ch := make(chan *pb.ListenerMessage, nul.args.UsageChanSize)
 	return nul.BaseListener.Run(
 		ctx,
-		"Connected to the service, node usage listener stream established",
+		"Connected to operator service, unary node usage listener established",
 		ch,
 		nul.watchPods,
 		nul.sendMessages,
@@ -88,7 +88,7 @@ func (nul *NodeUsageListener) sendMessages(
 				nul.inst.MessageChannelClosedUnexpectedly.Add(ctx, 1, nul.MetricAttrs)
 				return fmt.Errorf("usage watcher stopped")
 			}
-			if err := nul.BaseListener.SendMessage(ctx, msg); err != nil {
+			if err := nul.SendMessage(ctx, msg); err != nil {
 				return fmt.Errorf("failed to send UpdateNodeUsageBody message: %w", err)
 			}
 		}