Non-durable SSE streams (owncloud#7986)

* make sse streams non-durable Signed-off-by: jkoberg <[email protected]> * bump reva Signed-off-by: jkoberg <[email protected]> --------- Signed-off-by: jkoberg <[email protected]>
2403905 · Jan 24, 2024 · 12168e2 · 12168e2
1 parent 35fe83d
commit 12168e2
Show file tree

Hide file tree

Showing 31 changed files with 173 additions and 60 deletions.
diff --git a/changelog/unreleased/bump-reva.md b/changelog/unreleased/bump-reva.md
@@ -6,3 +6,4 @@ https://github.com/owncloud/ocis/pull/7793
 https://github.com/owncloud/ocis/pull/7978
 https://github.com/owncloud/ocis/pull/7979
 https://github.com/owncloud/ocis/pull/7963
+https://github.com/owncloud/ocis/pull/7986
diff --git a/changelog/unreleased/sse-non-durable-streams.md b/changelog/unreleased/sse-non-durable-streams.md
@@ -0,0 +1,5 @@
+Bugfix: Non durable streams for sse service
+
+Configure sse streams to be non-durable. This functionality is not needed for the sse service
+
+https://github.com/owncloud/ocis/pull/7986
diff --git a/go.mod b/go.mod
@@ -13,7 +13,7 @@ require (
 	github.com/coreos/go-oidc v2.2.1+incompatible
 	github.com/coreos/go-oidc/v3 v3.9.0
 	github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781
-	github.com/cs3org/reva/v2 v2.17.1-0.20231215155002-7810e3d8be38
+	github.com/cs3org/reva/v2 v2.17.1-0.20231218091701-6a3a91e35514
 	github.com/dhowden/tag v0.0.0-20230630033851-978a0926ee25
 	github.com/disintegration/imaging v1.6.2
 	github.com/dutchcoders/go-clamd v0.0.0-20170520113014-b970184f4d9e
@@ -35,7 +35,7 @@ require (
 	github.com/go-micro/plugins/v4/registry/nats v1.2.2-0.20230723205323-1ada01245674
 	github.com/go-micro/plugins/v4/server/grpc v1.2.0
 	github.com/go-micro/plugins/v4/server/http v1.2.2
-	github.com/go-micro/plugins/v4/store/nats-js v1.2.1-0.20231129143103-d72facc652f0
+	github.com/go-micro/plugins/v4/store/nats-js-kv v0.0.0-00010101000000-000000000000
 	github.com/go-micro/plugins/v4/wrapper/breaker/gobreaker v1.2.0
 	github.com/go-micro/plugins/v4/wrapper/monitoring/prometheus v1.2.0
 	github.com/go-micro/plugins/v4/wrapper/trace/opentelemetry v1.2.0
@@ -193,8 +193,8 @@ require (
 	github.com/go-logfmt/logfmt v0.5.1 // indirect
 	github.com/go-logr/logr v1.3.0 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
-	github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7 // indirect
-	github.com/go-micro/plugins/v4/store/nats-js-kv v0.0.0-00010101000000-000000000000 // indirect
+	github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20231215124540-f7f8d3274bf9 // indirect
+	github.com/go-micro/plugins/v4/store/nats-js v1.2.1-0.20231129143103-d72facc652f0 // indirect
 	github.com/go-micro/plugins/v4/store/redis v1.2.1-0.20230510195111-07cd57e1bc9d // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect

diff --git a/go.sum b/go.sum
@@ -1021,8 +1021,8 @@ github.com/crewjam/saml v0.4.14 h1:g9FBNx62osKusnFzs3QTN5L9CVA/Egfgm+stJShzw/c=
 github.com/crewjam/saml v0.4.14/go.mod h1:UVSZCf18jJkk6GpWNVqcyQJMD5HsRugBPf4I1nl2mME=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781 h1:BUdwkIlf8IS2FasrrPg8gGPHQPOrQ18MS1Oew2tmGtY=
 github.com/cs3org/go-cs3apis v0.0.0-20231023073225-7748710e0781/go.mod h1:UXha4TguuB52H14EMoSsCqDj7k8a/t7g4gVP+bgY5LY=
-github.com/cs3org/reva/v2 v2.17.1-0.20231215155002-7810e3d8be38 h1:vkA/Ty82yETTDrpLV/b5/9VXUJ/9o7vTRs7ampZC5LU=
-github.com/cs3org/reva/v2 v2.17.1-0.20231215155002-7810e3d8be38/go.mod h1:5Yxh1DneWZQvMBOiBVv3LJaSsSlRCoqCBa4Wws7PWHw=
+github.com/cs3org/reva/v2 v2.17.1-0.20231218091701-6a3a91e35514 h1:ThhcxiI3Iq8sweI6ZGaEkTzFhum+ev0EGGnDiAK8eW8=
+github.com/cs3org/reva/v2 v2.17.1-0.20231218091701-6a3a91e35514/go.mod h1:QW31Q1IQ9ZCJMFv3u8/SdHSyLfCcSVNcRbqIJj+Y+7o=
 github.com/cyberdelia/templates v0.0.0-20141128023046-ca7fffd4298c/go.mod h1:GyV+0YP4qX0UQ7r2MoYZ+AvYDp12OF5yg4q8rGnyNh4=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
@@ -1195,8 +1195,8 @@ github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
 github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
 github.com/go-micro/plugins/v4/client/grpc v1.2.1 h1:7xAwZRCO6mdUtBHsYIQs1/eCTdhCrnjF70GB+AVd6L0=
 github.com/go-micro/plugins/v4/client/grpc v1.2.1/go.mod h1:3fDuzyfYLwEImn8+lkhKl3W4Ay1jFevkTeC32PBlgQs=
-github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7 h1:/RpJVLKmKT2OcEnKCPaS6n+zygNzYDzwoYgPQEgcEiQ=
-github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20230807070816-bc05fb076ce7/go.mod h1:lYuiEYKQTpbE2LA8HEcC8D6kQ29M7ILfEak3dzeucEg=
+github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20231215124540-f7f8d3274bf9 h1:YOIavj+ZgO9HzukpdXZCvQv+AahjW/fTVFVF4QFRabw=
+github.com/go-micro/plugins/v4/events/natsjs v1.2.2-0.20231215124540-f7f8d3274bf9/go.mod h1:cL0O63th39fZ+M/aRJvajz7Qnmv+UTXugOq1k3qrYiQ=
 github.com/go-micro/plugins/v4/logger/zerolog v1.2.0 h1:JZ516VQ9zekRoi868XG7x0EWxZ2AMq/euHIBChITsTI=
 github.com/go-micro/plugins/v4/logger/zerolog v1.2.0/go.mod h1:AieYOIeOxobYa5B8WGEqxXM3Ndi26tDIu9fZ4RYkCvQ=
 github.com/go-micro/plugins/v4/registry/consul v1.2.1 h1:3wctYMtstwQLCjoJ1HA6mKGGFF1hcdKDv5MzHakB1jE=

diff --git a/services/antivirus/pkg/service/service.go b/services/antivirus/pkg/service/service.go
@@ -96,7 +96,7 @@ func (av Antivirus) Run() error {
 		evtsCfg.TLSInsecure = false
 	}
 
-	natsStream, err := stream.NatsFromConfig(av.c.Service.Name, stream.NatsConfig(av.c.Events))
+	natsStream, err := stream.NatsFromConfig(av.c.Service.Name, false, stream.NatsConfig(av.c.Events))
 	if err != nil {
 		return err
 	}

diff --git a/services/audit/pkg/command/server.go b/services/audit/pkg/command/server.go
@@ -43,7 +43,7 @@ func Server(cfg *config.Config) *cli.Command {
 			)
 			defer cancel()
 
-			client, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+			client, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 			if err != nil {
 				return err
 			}

diff --git a/services/clientlog/pkg/command/server.go b/services/clientlog/pkg/command/server.go
@@ -57,7 +57,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 			defer cancel()
 
-			stream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+			stream, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 			if err != nil {
 				return err
 			}

diff --git a/services/eventhistory/pkg/command/server.go b/services/eventhistory/pkg/command/server.go
@@ -60,7 +60,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 			metrics.BuildInfo.WithLabelValues(version.GetString()).Set(1)
 
-			consumer, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+			consumer, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 			if err != nil {
 				return err
 			}

diff --git a/services/frontend/pkg/command/events.go b/services/frontend/pkg/command/events.go
@@ -39,7 +39,7 @@ var _registeredEvents = []events.Unmarshaller{
 
 // ListenForEvents listens for events and acts accordingly
 func ListenForEvents(ctx context.Context, cfg *config.Config, l log.Logger) error {
-	bus, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+	bus, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 	if err != nil {
 		l.Error().Err(err).Msg("cannot connect to nats")
 		return err

diff --git a/services/graph/pkg/server/http/server.go b/services/graph/pkg/server/http/server.go
@@ -53,7 +53,7 @@ func Server(opts ...Option) (http.Service, error) {
 
 	if options.Config.Events.Endpoint != "" {
 		var err error
-		publisher, err = stream.NatsFromConfig(options.Config.Service.Name, stream.NatsConfig(options.Config.Events))
+		publisher, err = stream.NatsFromConfig(options.Config.Service.Name, false, stream.NatsConfig(options.Config.Events))
 		if err != nil {
 			options.Logger.Error().
 				Err(err).

diff --git a/services/notifications/pkg/command/server.go b/services/notifications/pkg/command/server.go
@@ -89,7 +89,7 @@ func Server(cfg *config.Config) *cli.Command {
 				events.SpaceUnshared{},
 				events.SpaceMembershipExpired{},
 			}
-			client, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Notifications.Events))
+			client, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Notifications.Events))
 			if err != nil {
 				return err
 			}

diff --git a/services/policies/pkg/command/server.go b/services/policies/pkg/command/server.go
@@ -110,7 +110,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 			{
 
-				bus, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+				bus, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 				if err != nil {
 					return err
 				}

diff --git a/services/postprocessing/pkg/command/postprocessing.go b/services/postprocessing/pkg/command/postprocessing.go
@@ -30,7 +30,7 @@ func RestartPostprocessing(cfg *config.Config) *cli.Command {
 			return configlog.ReturnFatal(parser.ParseConfig(cfg))
 		},
 		Action: func(c *cli.Context) error {
-			stream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Postprocessing.Events))
+			stream, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Postprocessing.Events))
 			if err != nil {
 				return err
 			}

diff --git a/services/postprocessing/pkg/command/server.go b/services/postprocessing/pkg/command/server.go
@@ -54,7 +54,7 @@ func Server(cfg *config.Config) *cli.Command {
 			}
 
 			{
-				bus, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Postprocessing.Events))
+				bus, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Postprocessing.Events))
 				if err != nil {
 					return err
 				}

diff --git a/services/search/pkg/service/grpc/v0/service.go b/services/search/pkg/service/grpc/v0/service.go
@@ -75,7 +75,7 @@ func NewHandler(opts ...Option) (searchsvc.SearchProviderHandler, func(), error)
 		return nil, teardown, fmt.Errorf("unknown search extractor: %s", cfg.Extractor.Type)
 	}
 
-	bus, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig{
+	bus, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig{
 		Endpoint:             cfg.Events.Endpoint,
 		Cluster:              cfg.Events.Cluster,
 		EnableTLS:            cfg.Events.EnableTLS,

diff --git a/services/sse/pkg/command/server.go b/services/sse/pkg/command/server.go
@@ -59,7 +59,7 @@ func Server(cfg *config.Config) *cli.Command {
 			}
 
 			{
-				natsStream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+				natsStream, err := stream.NatsFromConfig(cfg.Service.Name, true, stream.NatsConfig(cfg.Events))
 				if err != nil {
 					return err
 				}

diff --git a/services/storage-users/pkg/event/event.go b/services/storage-users/pkg/event/event.go
@@ -8,7 +8,7 @@ import (
 
 // NewStream prepares the requested nats stream and returns it.
 func NewStream(cfg *config.Config) (events.Stream, error) {
-	return stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig{
+	return stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig{
 		Endpoint:             cfg.Events.Addr,
 		Cluster:              cfg.Events.ClusterID,
 		EnableTLS:            cfg.Events.EnableTLS,

diff --git a/services/userlog/pkg/command/server.go b/services/userlog/pkg/command/server.go
@@ -82,7 +82,7 @@ func Server(cfg *config.Config) *cli.Command {
 
 			defer cancel()
 
-			stream, err := stream.NatsFromConfig(cfg.Service.Name, stream.NatsConfig(cfg.Events))
+			stream, err := stream.NatsFromConfig(cfg.Service.Name, false, stream.NatsConfig(cfg.Events))
 			if err != nil {
 				return err
 			}

diff --git a/vendor/github.com/cs3org/reva/v2/internal/grpc/interceptors/eventsmiddleware/events.go b/vendor/github.com/cs3org/reva/v2/internal/grpc/interceptors/eventsmiddleware/events.go
diff --git a/vendor/github.com/cs3org/reva/v2/internal/grpc/services/storageprovider/storageprovider.go b/vendor/github.com/cs3org/reva/v2/internal/grpc/services/storageprovider/storageprovider.go
diff --git a/vendor/github.com/cs3org/reva/v2/internal/http/services/dataprovider/dataprovider.go b/vendor/github.com/cs3org/reva/v2/internal/http/services/dataprovider/dataprovider.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/events/stream/nats.go b/vendor/github.com/cs3org/reva/v2/pkg/events/stream/nats.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/metrics/metrics.go b/vendor/github.com/cs3org/reva/v2/pkg/rhttp/datatx/metrics/metrics.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go b/vendor/github.com/cs3org/reva/v2/pkg/share/manager/jsoncs3/jsoncs3.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/decomposedfs.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/node/node.go
diff --git a/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go b/vendor/github.com/cs3org/reva/v2/pkg/storage/utils/decomposedfs/upload.go