make antivirus workers configurable

Signed-off-by: Jörn Friedrich Dreyer <jfd@butonic.de>

services/antivirus/pkg/config/config.go

@@ -18,8 +18,10 @@ type Config struct {
 
 	InfectedFileHandling string `yaml:"infected-file-handling" env:"ANTIVIRUS_INFECTED_FILE_HANDLING" desc:"Defines the behaviour when a virus has been found. Supported options are: 'delete', 'continue' and 'abort '. Delete will delete the file. Continue will mark the file as infected but continues further processing. Abort will keep the file in the uploads folder for further admin inspection and will not move it to its final destination." introductionVersion:"pre5.0"`
 	Events               Events
-	Scanner              Scanner
-	MaxScanSize          string `yaml:"max-scan-size" env:"ANTIVIRUS_MAX_SCAN_SIZE" desc:"The maximum scan size the virus scanner can handle. Only this many bytes of a file will be scanned. 0 means unlimited and is the default. Usable common abbreviations: [KB, KiB, MB, MiB, GB, GiB, TB, TiB, PB, PiB, EB, EiB], example: 2GB." introductionVersion:"pre5.0"`
+	Workers              int `yaml:"workers" env:"ANTIVIRUS_WORKERS" desc:"The number of concurrent go routines that fetch events from the event queue." introductionVersion:"6.7"`
+
+	Scanner     Scanner
+	MaxScanSize string `yaml:"max-scan-size" env:"ANTIVIRUS_MAX_SCAN_SIZE" desc:"The maximum scan size the virus scanner can handle. Only this many bytes of a file will be scanned. 0 means unlimited and is the default. Usable common abbreviations: [KB, KiB, MB, MiB, GB, GiB, TB, TiB, PB, PiB, EB, EiB], example: 2GB." introductionVersion:"pre5.0"`
 
 	Context context.Context `yaml:"-" json:"-"`
 

services/antivirus/pkg/scanners/icap.go

@@ -86,6 +86,7 @@ func (s ICAP) Scan(in Input) (Result, error) {
 	if err != nil {
 		return result, err
 	}
+	result.ScanTime = time.Now()
 
 	// TODO: make header configurable. See oc10 documentation: https://doc.owncloud.com/server/10.12/admin_manual/configuration/server/virus-scanner-support.html
 	if data, infected := res.Header["X-Infection-Found"]; infected {

services/antivirus/pkg/service/service.go

@@ -116,20 +116,23 @@ func (av Antivirus) Run() error {
 		return err
 	}
 
-	for e := range ch {
-		err := av.processEvent(e, natsStream)
-		if err != nil {
-			switch {
-			case errors.Is(err, ErrFatal):
-				return err
-			case errors.Is(err, ErrEvent):
-				// Right now logging of these happens in the processEvent method, might be cleaner to do it here.
-				continue
-			default:
-				av.l.Fatal().Err(err).Msg("unknown error - exiting")
+	// Spawn workers that'll concurrently work the queue
+	for i := 0; i < av.c.Workers; i++ {
+		go (func() {
+			for e := range ch {
+				err := av.processEvent(e, natsStream)
+				if err != nil {
+					switch {
+					case errors.Is(err, ErrFatal):
+						av.l.Fatal().Err(err).Msg("fatal error - exiting")
+					case errors.Is(err, ErrEvent):
+						av.l.Error().Err(err).Msg("continuing")
+					default:
+						av.l.Fatal().Err(err).Msg("unknown error - exiting")
+					}
+				}
 			}
-		}
-
+		})()
 	}
 
 	return nil
@@ -168,10 +171,12 @@ func (av Antivirus) processEvent(e events.Event, s events.Publisher) error {
 
 	av.l.Debug().Str("uploadid", ev.UploadID).Str("filename", ev.Filename).Msg("Starting virus scan.")
 	var errmsg string
+	start := time.Now()
 	res, err := av.process(ev)
 	if err != nil {
 		errmsg = err.Error()
 	}
+	duration := time.Since(start)
 
 	var outcome events.PostprocessingOutcome
 	switch {
@@ -186,7 +191,7 @@ func (av Antivirus) processEvent(e events.Event, s events.Publisher) error {
 		outcome = events.PPOutcomeAbort
 	}
 
-	av.l.Info().Str("uploadid", ev.UploadID).Interface("resourceID", ev.ResourceID).Str("virus", res.Description).Str("outcome", string(outcome)).Str("filename", ev.Filename).Str("user", ev.ExecutingUser.GetId().GetOpaqueId()).Bool("infected", res.Infected).Msg("File scanned")
+	av.l.Info().Str("uploadid", ev.UploadID).Interface("resourceID", ev.ResourceID).Str("virus", res.Description).Str("outcome", string(outcome)).Str("filename", ev.Filename).Str("user", ev.ExecutingUser.GetId().GetOpaqueId()).Bool("infected", res.Infected).Dur("duration", duration).Msg("File scanned")
 	if err := events.Publish(ctx, s, events.PostprocessingStepFinished{
 		FinishedStep:  events.PPStepAntivirus,
 		Outcome:       outcome,