Merge pull request #12 from rafabene/HYPERFLEET-641

HYPERFLEET-641: Add Health() method, optimize integration tests, fix goroutine leak

Author: openshift-merge-bot[bot]

Makefile

@@ -14,9 +14,9 @@ fmt:
 test:
 	go test -v ./broker/... ./pkg/... -timeout 10m
 
-# Run tests in the test folder
+# Run tests in the test folder (sequential packages: CI has 1 CPU so parallel execution causes timeouts)
 test-integration:
-	go test -v ./test/integration/... -timeout 10m
+	go test -v -p 1 ./test/integration/... -timeout 10m
 
 # Run all tests
 test-all: test test-integration

README.md

@@ -13,6 +13,7 @@ The current implementation uses [Watermill](https://github.com/ThreeDotsLabs/wat
 - **Flexible Configuration**: YAML configuration files with environment variable overrides via Viper
 - **Worker Pools**: Configurable parallel message processing for subscribers
 - **Subscription Management**: Flexible subscription IDs for load balancing (shared subscriptions) or fanout (separate subscriptions)
+- **Health Checks**: Built-in `Health()` method on `Publisher` for readiness probes (per [HyperFleet Health Endpoints standard](https://github.com/openshift-hyperfleet/architecture/blob/main/hyperfleet/standards/health-endpoints.md))
 - **Simple API**: Clean, easy-to-use interface that hides Watermill complexity
 
 ## Installation
@@ -52,7 +53,7 @@ appLogger := logger.NewTestLogger()
 publisher, err := broker.NewPublisher(appLogger)
 
 // Use JSON format for the default logger
-appLogger := logger.NewTestLogger(logger.FormatJSON)
+appLogger := logger.NewTestLogger(logger.WithFormat(logger.FormatJSON))
 publisher, err := broker.NewPublisher(appLogger)
 
 // Use your own logger implementation with config
@@ -113,9 +114,23 @@ func main() {
 
 </details>
 
-Note for Google PubSub: The Google Pub/Sub publisher implementation (via Watermill/Google Cloud SDK) starts background goroutines (for batching, connection management, etc.). 
-The app should call Close() to not leak 
+Note for Google Pub/Sub: The Google Pub/Sub publisher implementation (via Watermill/Google Cloud SDK) starts background goroutines (for batching, connection management, etc.).
+The app should call `Close()` to avoid leaking these background goroutines and resources.
 
+### Health Check
+
+The `Publisher` interface exposes a `Health(ctx context.Context) error` method that consumers can use to verify broker connectivity, typically in `/readyz` endpoints:
+
+```go
+if err := publisher.Health(ctx); err != nil {
+    // Broker is unhealthy — return 503
+    log.Printf("Broker health check failed: %v", err)
+}
+```
+
+The health check implementation varies by broker:
+- **RabbitMQ**: Checks if the AMQP connection is open and not closed (in-memory, no network call)
+- **Google Pub/Sub**: Performs a lightweight `GetTopic` API call (3s timeout) on a non-existent probe topic to verify connectivity. A `NotFound` response confirms the round-trip succeeded. Requires the `pubsub.topics.get` permission, which is **not** included in `roles/pubsub.publisher`. You must grant an additional role such as `roles/pubsub.viewer`, `roles/pubsub.editor`, or a custom role containing `pubsub.topics.get`.
 
 <details>
 <summary><strong>Subscriber Example</strong></summary>
@@ -456,9 +471,9 @@ The subscription ID concept enables two messaging patterns:
 - **Load Balancing**: Multiple subscribers with the same subscription ID share messages
 - **Fanout**: Subscribers with different subscription IDs each receive all messages
 
-This is implemented consistently across brokers:
+This is implemented differently per broker:
 - **RabbitMQ**: Queue names are `{topic}-{subscriptionId}`
-- **Google Pub/Sub**: Subscription names are `{topic}-{subscriptionId}`
+- **Google Pub/Sub**: Subscription names are `{subscriptionId}`
 
 ### 4. Worker Pool Architecture
 

broker/broker.go

@@ -4,9 +4,11 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"strings"
 	"time"
 
+	"cloud.google.com/go/pubsub/v2"
 	"github.com/ThreeDotsLabs/watermill"
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/cloudevents/sdk-go/v2/event"
@@ -58,25 +60,42 @@ func NewPublisher(log logger.Logger, configMap ...map[string]string) (Publisher,
 	}
 
 	var pub message.Publisher
+	var hc healthCheckFunc
+	var healthCloser io.Closer
 
 	switch cfg.Broker.Type {
 	case "rabbitmq":
 		pub, err = newRabbitMQPublisher(cfg, watermillLogger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create RabbitMQ publisher: %w", err)
 		}
+		hc = newRabbitMQHealthCheck(pub)
 	case "googlepubsub":
 		pub, err = newGooglePubSubPublisher(cfg, watermillLogger)
 		if err != nil {
 			return nil, fmt.Errorf("failed to create Google Pub/Sub publisher: %w", err)
 		}
+
+		// Create a persistent Pub/Sub client for health checks, reused across calls.
+		var healthClient *pubsub.Client
+		healthClient, err = pubsub.NewClient(context.Background(), cfg.Broker.GooglePubSub.ProjectID)
+		if err != nil {
+			if closeErr := pub.Close(); closeErr != nil {
+				return nil, fmt.Errorf("failed to create health check client: %w (also failed to close publisher: %v)", err, closeErr)
+			}
+			return nil, fmt.Errorf("failed to create health check client: %w", err)
+		}
+		hc = newGooglePubSubHealthCheck(healthClient, cfg.Broker.GooglePubSub.ProjectID)
+		healthCloser = healthClient
 	default:
 		return nil, fmt.Errorf("unsupported broker type: %s", cfg.Broker.Type)
 	}
 
 	return &publisher{
-		pub:    pub,
-		logger: log,
+		pub:          pub,
+		logger:       log,
+		healthCheck:  hc,
+		healthCloser: healthCloser,
 	}, nil
 }
 

broker/googlepubsub.go

@@ -77,6 +77,36 @@ func parseGoogleCloudDuration(s string) (time.Duration, error) {
 	return 0, fmt.Errorf("invalid duration format: %s", s)
 }
 
+// healthCheckTopicName is a non-existent topic used as a connectivity probe.
+// GetTopic returning NotFound proves the round-trip to Pub/Sub succeeded.
+const healthCheckTopicName = "hyperfleet-health-probe"
+
+// newGooglePubSubHealthCheck creates a health check function for a Google Pub/Sub publisher.
+// It reuses the provided pubsub.Client to perform a lightweight GetTopic API call
+// with a 3-second timeout to verify connectivity.
+// Requires pubsub.topics.get (not included in roles/pubsub.publisher; grant roles/pubsub.viewer or a custom role).
+// The caller is responsible for closing the client (via publisher.healthCloser).
+func newGooglePubSubHealthCheck(client *pubsub.Client, projectID string) healthCheckFunc {
+	fullTopicName := fmt.Sprintf("projects/%s/topics/%s", projectID, healthCheckTopicName)
+
+	return func(ctx context.Context) error {
+		ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
+		defer cancel()
+
+		_, err := client.TopicAdminClient.GetTopic(ctx, &pubsubpb.GetTopicRequest{
+			Topic: fullTopicName,
+		})
+		if err != nil {
+			// NotFound means the API is reachable — the topic just doesn't exist.
+			if st, ok := status.FromError(err); ok && st.Code() == codes.NotFound {
+				return nil
+			}
+			return fmt.Errorf("google pub/sub health check failed: %w", err)
+		}
+		return nil
+	}
+}
+
 // newGooglePubSubPublisher creates a Google Pub/Sub publisher
 func newGooglePubSubPublisher(cfg *config, logger watermill.LoggerAdapter) (message.Publisher, error) {
 	gps := cfg.Broker.GooglePubSub
@@ -161,8 +191,6 @@ func newGooglePubSubSubscriber(cfg *config, logger watermill.LoggerAdapter, subs
 	}
 
 	// Configure subscription name generator to use subscription ID
-	// The topic passed to Subscribe will be the original topic (no colon)
-	// We append subscription ID to create unique subscription names
 	pubsubConfig := googlepubsub.SubscriberConfig{
 		ProjectID: gps.ProjectID,
 		GenerateSubscriptionName: func(topic string) string {

broker/health_test.go

@@ -0,0 +1,60 @@
+package broker
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/ThreeDotsLabs/watermill/message"
+	"github.com/openshift-hyperfleet/hyperfleet-broker/pkg/logger"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPublisherHealthWithCustomHealthCheck(t *testing.T) {
+	mockLogger := logger.NewMockLogger()
+
+	t.Run("healthy publisher returns nil", func(t *testing.T) {
+		p := &publisher{
+			logger: mockLogger,
+			healthCheck: func(_ context.Context) error {
+				return nil
+			},
+		}
+		err := p.Health(context.Background())
+		assert.NoError(t, err)
+	})
+
+	t.Run("unhealthy publisher returns error", func(t *testing.T) {
+		p := &publisher{
+			logger: mockLogger,
+			healthCheck: func(_ context.Context) error {
+				return fmt.Errorf("connection lost")
+			},
+		}
+		err := p.Health(context.Background())
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "connection lost")
+	})
+}
+
+func TestNewRabbitMQHealthCheck(t *testing.T) {
+	t.Run("returns error for non-AMQP publisher", func(t *testing.T) {
+		// Pass a fakeWatermillPublisher (non-AMQP) to newRabbitMQHealthCheck
+		// to trigger the AMQP type assertion failure
+		hc := newRabbitMQHealthCheck(&fakeWatermillPublisher{})
+		err := hc(context.Background())
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "unexpected publisher type")
+	})
+}
+
+// fakeWatermillPublisher implements message.Publisher for testing type assertion failures
+type fakeWatermillPublisher struct{}
+
+func (f *fakeWatermillPublisher) Publish(_ string, _ ...*message.Message) error {
+	return nil
+}
+
+func (f *fakeWatermillPublisher) Close() error {
+	return nil
+}

broker/publisher.go

@@ -2,6 +2,8 @@ package broker
 
 import (
 	"context"
+	"fmt"
+	"io"
 
 	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/cloudevents/sdk-go/v2/event"
@@ -12,14 +14,25 @@ import (
 type Publisher interface {
 	// Publish publishes a CloudEvent to the specified topic with context
 	Publish(ctx context.Context, topic string, event *event.Event) error
+	// Health checks if the underlying broker connection is healthy.
+	// Returns nil if healthy, or an error describing the failure.
+	// The provided context controls the deadline/cancellation of the check.
+	Health(ctx context.Context) error
 	// Close closes the underlying publisher
 	Close() error
 }
 
+// healthCheckFunc is a function that checks broker connectivity.
+// Returns nil if healthy, or an error describing the failure.
+// The provided context controls the deadline/cancellation of the check.
+type healthCheckFunc func(ctx context.Context) error
+
 // publisher wraps a Watermill publisher and provides a simplified interface
 type publisher struct {
-	pub    message.Publisher
-	logger logger.Logger // Caller's logger (always present - default logger if not provided)
+	pub          message.Publisher
+	logger       logger.Logger // Caller's logger (always present - default logger if not provided)
+	healthCheck  healthCheckFunc
+	healthCloser io.Closer // optional resource to close with publisher (e.g. Pub/Sub health check client)
 }
 
 // Publish publishes a CloudEvent to the specified topic with context
@@ -45,7 +58,17 @@ func (p *publisher) Publish(ctx context.Context, topic string, event *event.Even
 	return nil
 }
 
-// Close closes the underlying publisher
+// Health checks if the underlying broker connection is healthy.
+// Returns nil if healthy, or an error describing the failure.
+// The provided context controls the deadline/cancellation of the check.
+func (p *publisher) Health(ctx context.Context) error {
+	if p == nil || p.healthCheck == nil {
+		return fmt.Errorf("health check not configured")
+	}
+	return p.healthCheck(ctx)
+}
+
+// Close closes the underlying publisher and any health check resources.
 func (p *publisher) Close() error {
 	p.logger.Info(context.Background(), "Closing publisher")
 
@@ -54,5 +77,14 @@ func (p *publisher) Close() error {
 		p.logger.Errorf(context.Background(), "Failed to close publisher: %v", err)
 	}
 
+	if p.healthCloser != nil {
+		if closeErr := p.healthCloser.Close(); closeErr != nil {
+			p.logger.Errorf(context.Background(), "Failed to close health check client: %v", closeErr)
+			if err == nil {
+				err = closeErr
+			}
+		}
+	}
+
 	return err
 }

broker/rabbitmq.go

@@ -1,6 +1,7 @@
 package broker
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
@@ -22,6 +23,28 @@ type rabbitMQConfig struct {
 	PublisherConfirm bool   `mapstructure:"publisher_confirm"`
 }
 
+// newRabbitMQHealthCheck creates a health check function for a RabbitMQ publisher.
+// It uses the AMQP ConnectionWrapper's IsConnected() and Closed() methods
+// to determine if the broker connection is alive.
+func newRabbitMQHealthCheck(pub message.Publisher) healthCheckFunc {
+	return func(_ context.Context) error {
+		amqpPub, ok := pub.(*amqp.Publisher)
+		if !ok {
+			return fmt.Errorf("unexpected publisher type for RabbitMQ health check")
+		}
+		if amqpPub == nil {
+			return fmt.Errorf("RabbitMQ publisher is nil")
+		}
+		if amqpPub.Closed() {
+			return fmt.Errorf("RabbitMQ connection is closed")
+		}
+		if !amqpPub.IsConnected() {
+			return fmt.Errorf("RabbitMQ connection is not established")
+		}
+		return nil
+	}
+}
+
 // newRabbitMQPublisher creates a RabbitMQ publisher
 func newRabbitMQPublisher(cfg *config, logger watermill.LoggerAdapter) (message.Publisher, error) {
 	amqpConfig := amqp.NewDurablePubSubConfig(