From d6958f3a3eda3d27b8828a2d6cee6c51936b7c1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?J=C3=B6rn=20Friedrich=20Dreyer?= Date: Tue, 22 Oct 2024 10:13:16 +0200 Subject: [PATCH] introduce event processing workers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Jörn Friedrich Dreyer --- services/postprocessing/pkg/config/config.go | 1 + .../pkg/config/defaults/defaultconfig.go | 1 + .../postprocessing/pkg/service/service.go | 27 +++++++++++-------- 3 files changed, 18 insertions(+), 11 deletions(-) diff --git a/services/postprocessing/pkg/config/config.go b/services/postprocessing/pkg/config/config.go index 70f072370..3fb9559d8 100644 --- a/services/postprocessing/pkg/config/config.go +++ b/services/postprocessing/pkg/config/config.go @@ -35,6 +35,7 @@ type Postprocessing struct { // Events combines the configuration options for the event bus. type Events struct { + Workers int `yaml:"workers" env:"POSTPROCESSING_EVENTS_WORKERS" desc:"The number of concurrent go routines that fetch events from the event queue." introductionVersion:"%%NEXT%%"` Endpoint string `yaml:"endpoint" env:"OCIS_EVENTS_ENDPOINT;POSTPROCESSING_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"pre5.0"` Cluster string `yaml:"cluster" env:"OCIS_EVENTS_CLUSTER;POSTPROCESSING_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"pre5.0"` diff --git a/services/postprocessing/pkg/config/defaults/defaultconfig.go b/services/postprocessing/pkg/config/defaults/defaultconfig.go index 51a9bb611..a649440b9 100644 --- a/services/postprocessing/pkg/config/defaults/defaultconfig.go +++ b/services/postprocessing/pkg/config/defaults/defaultconfig.go @@ -28,6 +28,7 @@ func DefaultConfig() *config.Config { }, Postprocessing: config.Postprocessing{ Events: config.Events{ + Workers: 3, Endpoint: "127.0.0.1:9233", Cluster: "ocis-cluster", }, diff --git a/services/postprocessing/pkg/service/service.go b/services/postprocessing/pkg/service/service.go index ec29966ac..1bc9065b0 100644 --- a/services/postprocessing/pkg/service/service.go +++ b/services/postprocessing/pkg/service/service.go @@ -65,18 +65,23 @@ func NewPostprocessingService(ctx context.Context, stream events.Stream, logger // Run to fulfil Runner interface func (pps *PostprocessingService) Run() error { - for e := range pps.events { - err := pps.processEvent(e) - if err != nil { - switch { - case errors.Is(err, ErrFatal): - return err - case errors.Is(err, ErrEvent): - continue - default: - pps.log.Fatal().Err(err).Msg("unknown error - exiting") + // Spawn workers that'll concurrently work the queue + for i := 0; i < 3; i++ { + go (func() { + for e := range pps.events { + err := pps.processEvent(e) + if err != nil { + switch { + case errors.Is(err, ErrFatal): + pps.log.Fatal().Err(err).Msg("fatal error - exiting") + case errors.Is(err, ErrEvent): + pps.log.Error().Err(err).Msg("continuing") + default: + pps.log.Fatal().Err(err).Msg("unknown error - exiting") + } + } } - } + })() } return nil }