configurate the vault srorage Postprocessing

Author: 2403905

pkg/events/postprocessing.go

@@ -103,6 +103,7 @@ type PostprocessingStepFinished struct {
 	UploadID      string
 	ExecutingUser *user.User
 	Filename      string
+	ResourceID    *provider.ResourceId
 
 	FinishedStep Postprocessingstep    // name of the step
 	Result       interface{}           // result information see VirusscanResult for example
@@ -145,6 +146,7 @@ type VirusscanResult struct {
 type PostprocessingFinished struct {
 	UploadID          string
 	Filename          string
+	ResourceID        *provider.ResourceId
 	SpaceOwner        *user.UserId
 	ExecutingUser     *user.User
 	Result            map[Postprocessingstep]interface{} // it is a map[step]Event

pkg/storage/utils/decomposedfs/decomposedfs.go

@@ -258,7 +258,7 @@ func New(o *options.Options, aspects aspects.Aspects, log *zerolog.Logger) (stor
 			return nil, errors.New("need nats for async file processing")
 		}
 
-		ch, err := events.Consume(fs.stream, "dcfs", _registeredEvents...)
+		ch, err := events.Consume(fs.stream, o.Events.ConsumerGroup, _registeredEvents...)
 		if err != nil {
 			return nil, err
 		}
@@ -285,6 +285,10 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 		switch ev := event.Event.(type) {
 		case events.PostprocessingFinished:
 			sublog := log.With().Str("event", "PostprocessingFinished").Str("uploadid", ev.UploadID).Logger()
+			if ev.ResourceID != nil && ev.ResourceID.GetStorageId() != "" && ev.ResourceID.GetStorageId() != fs.o.MountID {
+				sublog.Debug().Msg("ignoring event for different storage")
+				continue
+			}
 			session, err := fs.sessionStore.Get(ctx, ev.UploadID)
 			if err != nil {
 				sublog.Error().Err(err).Msg("Failed to get upload")
@@ -450,6 +454,10 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 			session.Cleanup(true, !ev.KeepUpload, !ev.KeepUpload, true)
 		case events.RevertRevision:
 			sublog := log.With().Str("event", "RevertRevision").Interface("nodeid", ev.ResourceID).Logger()
+			if ev.ResourceID != nil && ev.ResourceID.GetStorageId() != "" && ev.ResourceID.GetStorageId() != fs.o.MountID {
+				sublog.Debug().Msg("ignoring event for different storage")
+				continue
+			}
 			n, err := fs.lu.NodeFromID(ctx, ev.ResourceID)
 			if err != nil {
 				sublog.Error().Err(err).Msg("Failed to get node")
@@ -462,6 +470,10 @@ func (fs *Decomposedfs) Postprocessing(ch <-chan events.Event) {
 			}
 		case events.PostprocessingStepFinished:
 			sublog := log.With().Str("event", "PostprocessingStepFinished").Str("uploadid", ev.UploadID).Logger()
+			if ev.ResourceID != nil && ev.ResourceID.GetStorageId() != "" && ev.ResourceID.GetStorageId() != fs.o.MountID {
+				sublog.Debug().Msg("ignoring event for different storage")
+				continue
+			}
 			if ev.FinishedStep != events.PPStepAntivirus {
 				// atm we are only interested in antivirus results
 				continue

pkg/storage/utils/decomposedfs/options/options.go

@@ -23,10 +23,10 @@ import (
 	"strings"
 	"time"
 
+	"github.com/mitchellh/mapstructure"
 	"github.com/owncloud/reva/v2/pkg/rgrpc/todo/pool"
 	"github.com/owncloud/reva/v2/pkg/sharedconf"
 	"github.com/owncloud/reva/v2/pkg/storage/cache"
-	"github.com/mitchellh/mapstructure"
 	"github.com/pkg/errors"
 )
 
@@ -103,7 +103,8 @@ type AsyncPropagatorOptions struct {
 
 // EventOptions are the configurable options for events
 type EventOptions struct {
-	NumConsumers int `mapstructure:"numconsumers"`
+	NumConsumers  int    `mapstructure:"numconsumers"`
+	ConsumerGroup string `mapstructure:"consumer_group"`
 }
 
 // TokenOptions are the configurable option for tokens
@@ -172,5 +173,9 @@ func New(m map[string]interface{}) (*Options, error) {
 		o.UploadDirectory = filepath.Join(o.Root, "uploads")
 	}
 
+	if o.Events.ConsumerGroup == "" {
+		o.Events.ConsumerGroup = "dcfs"
+	}
+
 	return o, nil
 }