diff --git a/changelog/unreleased/postprocessing-events.md b/changelog/unreleased/postprocessing-events.md new file mode 100644 index 0000000000..7b32efcc31 --- /dev/null +++ b/changelog/unreleased/postprocessing-events.md @@ -0,0 +1,5 @@ +Bugfix: increase event processing workers + +We increased the number of go routines that pull events from the queue to three and made the number off workers configurable. Furthermore, the postprocessing delay no longer introduces a sleep that slows down pulling of events, but asynchronously triggers the next step. + +https://github.com/owncloud/ocis/pull/10368 diff --git a/services/postprocessing/pkg/config/config.go b/services/postprocessing/pkg/config/config.go index 70f0723702..8919df2f7a 100644 --- a/services/postprocessing/pkg/config/config.go +++ b/services/postprocessing/pkg/config/config.go @@ -25,7 +25,9 @@ type Config struct { // Postprocessing defines the config options for the postprocessing service. type Postprocessing struct { - Events Events `yaml:"events"` + Events Events `yaml:"events"` + Workers int `yaml:"workers" env:"POSTPROCESSING_WORKERS" desc:"The number of concurrent go routines that fetch events from the event queue." introductionVersion:"6.7"` + Steps []string `yaml:"steps" env:"POSTPROCESSING_STEPS" desc:"A list of postprocessing steps processed in order of their appearance. Currently supported values by the system are: 'virusscan', 'policies' and 'delay'. Custom steps are allowed. See the documentation for instructions. See the Environment Variable Types description for more details." introductionVersion:"pre5.0"` Delayprocessing time.Duration `yaml:"delayprocessing" env:"POSTPROCESSING_DELAY" desc:"After uploading a file but before making it available for download, a delay step can be added. Intended for developing purposes only. If a duration is set but the keyword 'delay' is not explicitely added to 'POSTPROCESSING_STEPS', the delay step will be processed as last step. In such a case, a log entry will be written on service startup to remind the admin about that situation. See the Environment Variable Types description for more details." introductionVersion:"pre5.0"` diff --git a/services/postprocessing/pkg/config/defaults/defaultconfig.go b/services/postprocessing/pkg/config/defaults/defaultconfig.go index 51a9bb6112..4f00a1ff9e 100644 --- a/services/postprocessing/pkg/config/defaults/defaultconfig.go +++ b/services/postprocessing/pkg/config/defaults/defaultconfig.go @@ -31,6 +31,7 @@ func DefaultConfig() *config.Config { Endpoint: "127.0.0.1:9233", Cluster: "ocis-cluster", }, + Workers: 3, RetryBackoffDuration: 5 * time.Second, MaxRetries: 14, }, diff --git a/services/postprocessing/pkg/postprocessing/postprocessing.go b/services/postprocessing/pkg/postprocessing/postprocessing.go index 1ce3962357..f5274e9875 100644 --- a/services/postprocessing/pkg/postprocessing/postprocessing.go +++ b/services/postprocessing/pkg/postprocessing/postprocessing.go @@ -75,9 +75,12 @@ func (pp *Postprocessing) CurrentStep() interface{} { } // Delay will sleep the configured time then continue -func (pp *Postprocessing) Delay() interface{} { - time.Sleep(pp.config.Delayprocessing) - return pp.next(events.PPStepDelay) +func (pp *Postprocessing) Delay(f func(next interface{})) { + next := pp.next(events.PPStepDelay) + go func() { + time.Sleep(pp.config.Delayprocessing) + f(next) + }() } // BackoffDuration calculates the duration for exponential backoff based on the number of failures. diff --git a/services/postprocessing/pkg/service/service.go b/services/postprocessing/pkg/service/service.go index 13441567d9..70918e8108 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 < pps.c.Workers; 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 } @@ -149,7 +154,11 @@ func (pps *PostprocessingService) processEvent(e events.Event) error { pps.log.Error().Str("uploadID", ev.UploadID).Err(err).Msg("cannot get upload") return fmt.Errorf("%w: cannot get upload", ErrEvent) } - next = pp.Delay() + pp.Delay(func(next interface{}) { + if err := events.Publish(ctx, pps.pub, next); err != nil { + pps.log.Error().Err(err).Msg("cannot publish event") + } + }) case events.UploadReady: if ev.Failed { // the upload failed - let's keep it around for a while - but mark it as finished