Use Publisher Everywhere (#3803)

* Use Publisher Eveywhere Use the pulsarutils.Publisher everywhere we publish eventsequences Motivations for this are: - Same code used to publish to pulsar used in all places - Better testing on that piece of code - Less work if we need to work on publisher in future (only 1 place to update) Signed-off-by: JamesMurkin <[email protected]> * Fix config validation Signed-off-by: JamesMurkin <[email protected]> * imports Signed-off-by: JamesMurkin <[email protected]> * gofumpt Signed-off-by: JamesMurkin <[email protected]> * Fix producer names Signed-off-by: JamesMurkin <[email protected]> --------- Signed-off-by: JamesMurkin <[email protected]>
armadaproject · Jul 22, 2024 · 7d4ed88 · 7d4ed88
1 parent 56c9d9f
commit 7d4ed88
Show file tree

Hide file tree

Showing 23 changed files with 432 additions and 362 deletions.
diff --git a/config/armada/config.yaml b/config/armada/config.yaml
@@ -78,6 +78,7 @@ pulsar:
   eventsPrinterSubscription: "EventsPrinter"
   maxAllowedEventsPerMessage: 1000
   maxAllowedMessageSize: 4194304 # 4MB
+  sendTimeout: 5s
   receiverQueueSize: 100
 postgres:
   connection:

diff --git a/config/eventingester/config.yaml b/config/eventingester/config.yaml
@@ -7,7 +7,6 @@ redis:
 pulsar:
   URL: pulsar://pulsar:6650
   jobsetEventsTopic: events
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "events-ingester"

diff --git a/config/lookoutingesterv2/config.yaml b/config/lookoutingesterv2/config.yaml
@@ -10,7 +10,6 @@ metricsPort: 9002
 pulsar:
   URL: "pulsar://pulsar:6650"
   jobsetEventsTopic: "events"
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "lookout-ingester-v2"

diff --git a/config/scheduler/config.yaml b/config/scheduler/config.yaml
@@ -39,6 +39,7 @@ pulsar:
   compressionLevel: faster
   maxAllowedEventsPerMessage: 1000
   maxAllowedMessageSize: 4194304 #4Mi
+  sendTimeout: 5s
 armadaApi:
   armadaUrl: "server:50051"
   forceNoTls: true

diff --git a/config/scheduleringester/config.yaml b/config/scheduleringester/config.yaml
@@ -11,7 +11,6 @@ metrics:
 pulsar:
   URL: "pulsar://localhost:6650"
   jobsetEventsTopic: "events"
-  receiveTimeout: 5s
   backoffTime: 1s
   receiverQueueSize: 100
 subscriptionName: "scheduler-ingester"

diff --git a/internal/armada/configuration/types.go b/internal/armada/configuration/types.go
@@ -70,8 +70,8 @@ type PulsarConfig struct {
 	MaxAllowedEventsPerMessage int `validate:"gte=0"`
 	// Maximum allowed message size in bytes
 	MaxAllowedMessageSize uint
-	// Timeout when polling pulsar for messages
-	ReceiveTimeout time.Duration
+	// Timeout when sending messages asynchronously
+	SendTimeout time.Duration `validate:"required"`
 	// Backoff from polling when Pulsar returns an error
 	BackoffTime time.Duration
 	// Number of pulsar messages that will be queued by the pulsar consumer.

diff --git a/internal/armada/mocks/generate.go b/internal/armada/mocks/generate.go
@@ -1,6 +1,5 @@
 package mocks
 
 // Mock implementations used by tests
-//go:generate mockgen -destination=./mock_repository.go -package=mocks "github.com/armadaproject/armada/internal/armada/queue" QueueRepository
-//go:generate mockgen -destination=./mock_deduplicator.go -package=mocks "github.com/armadaproject/armada/internal/armada/submit" Deduplicator,Publisher
+//go:generate mockgen -destination=./mock_deduplicator.go -package=mocks "github.com/armadaproject/armada/internal/armada/submit" Deduplicator
 //go:generate mockgen -destination=./mock_authorizer.go -package=mocks "github.com/armadaproject/armada/internal/armada/server" ActionAuthorizer
diff --git a/internal/armada/mocks/mock_deduplicator.go b/internal/armada/mocks/mock_deduplicator.go
diff --git a/internal/armada/server.go b/internal/armada/server.go
@@ -128,7 +128,7 @@ func Serve(ctx *armadacontext.Context, config *configuration.ArmadaConfig, healt
 		CompressionLevel: config.Pulsar.CompressionLevel,
 		BatchingMaxSize:  config.Pulsar.MaxAllowedMessageSize,
 		Topic:            config.Pulsar.JobsetEventsTopic,
-	}, config.Pulsar.MaxAllowedEventsPerMessage, config.Pulsar.MaxAllowedMessageSize)
+	}, config.Pulsar.MaxAllowedEventsPerMessage, config.Pulsar.MaxAllowedMessageSize, config.Pulsar.SendTimeout)
 	if err != nil {
 		return errors.Wrapf(err, "error creating pulsar producer")
 	}

diff --git a/internal/armada/submit/submit_test.go b/internal/armada/submit/submit_test.go
@@ -16,14 +16,15 @@ import (
 	"github.com/armadaproject/armada/internal/armada/submit/testfixtures"
 	"github.com/armadaproject/armada/internal/common/armadacontext"
 	"github.com/armadaproject/armada/internal/common/auth/permission"
+	commonMocks "github.com/armadaproject/armada/internal/common/mocks"
 	"github.com/armadaproject/armada/internal/common/util"
 	"github.com/armadaproject/armada/pkg/api"
 	"github.com/armadaproject/armada/pkg/armadaevents"
 	"github.com/armadaproject/armada/pkg/client/queue"
 )
 
 type mockObjects struct {
-	publisher    *mocks.MockPublisher
+	publisher    *commonMocks.MockPublisher
 	queueRepo    *mocks.MockQueueRepository
 	deduplicator *mocks.MockDeduplicator
 	authorizer   *mocks.MockActionAuthorizer
@@ -32,7 +33,7 @@ type mockObjects struct {
 func createMocks(t *testing.T) *mockObjects {
 	ctrl := gomock.NewController(t)
 	return &mockObjects{
-		publisher:    mocks.NewMockPublisher(ctrl),
+		publisher:    commonMocks.NewMockPublisher(ctrl),
 		queueRepo:    mocks.NewMockQueueRepository(ctrl),
 		deduplicator: mocks.NewMockDeduplicator(ctrl),
 		authorizer:   mocks.NewMockActionAuthorizer(ctrl),

diff --git a/internal/common/ingest/ingestion_pipeline_test.go b/internal/common/ingest/ingestion_pipeline_test.go
@@ -347,8 +347,7 @@ func TestRun_LimitsProcessingBatchSize(t *testing.T) {
 func testPipeline(consumer pulsar.Consumer, converter InstructionConverter[*simpleMessages], sink Sink[*simpleMessages]) *IngestionPipeline[*simpleMessages] {
 	return &IngestionPipeline[*simpleMessages]{
 		pulsarConfig: configuration.PulsarConfig{
-			ReceiveTimeout: 10 * time.Second,
-			BackoffTime:    time.Second,
+			BackoffTime: time.Second,
 		},
 		pulsarSubscriptionName: "subscription",
 		pulsarBatchDuration:    batchDuration,

diff --git a/internal/common/mocks/generate.go b/internal/common/mocks/generate.go
@@ -2,4 +2,5 @@ package mocks
 
 // Mock implementations used by tests
 //go:generate mockgen -destination=./mock_pulsar.go -package=mocks "github.com/apache/pulsar-client-go/pulsar" Client,Producer,Message
+//go:generate mockgen -destination=./mock_publisher.go -package=mocks "github.com/armadaproject/armada/internal/common/pulsarutils" Publisher
 //go:generate mockgen -destination=./mock_executorapi.go -package=mocks "github.com/armadaproject/armada/pkg/executorapi" ExecutorApiClient,ExecutorApi_LeaseJobRunsClient
diff --git a/internal/common/mocks/mock_publisher.go b/internal/common/mocks/mock_publisher.go
diff --git a/internal/common/pulsarutils/eventsequence.go b/internal/common/pulsarutils/eventsequence.go