Merge pull request #3192 from owncloud/nats-check-for-error

[full-ci] use NATS JetStream

audit/pkg/command/server.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/asim/go-micro/plugins/events/nats/v4"
+	"github.com/asim/go-micro/plugins/events/natsjs/v4"
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/events/server"
 	"github.com/owncloud/ocis/audit/pkg/config"
@@ -35,7 +35,10 @@ func Server(cfg *config.Config) *cli.Command {
 			defer cancel()
 
 			evtsCfg := cfg.Events
-			client, err := server.NewNatsStream(nats.Address(evtsCfg.Endpoint), nats.ClusterID(evtsCfg.Cluster))
+			client, err := server.NewNatsStream(
+				natsjs.Address(evtsCfg.Endpoint),
+				natsjs.ClusterID(evtsCfg.Cluster),
+			)
 			if err != nil {
 				return err
 			}

audit/pkg/config/defaults/defaultconfig.go

@@ -20,7 +20,7 @@ func DefaultConfig() *config.Config {
 		},
 		Events: config.Events{
 			Endpoint:      "127.0.0.1:9233",
-			Cluster:       "test-cluster",
+			Cluster:       "ocis-cluster",
 			ConsumerGroup: "audit",
 		},
 		Auditlog: config.Auditlog{

changelog/unreleased/change-nats-backend.md

@@ -0,0 +1,6 @@
+Change: Switch NATS backend
+
+We've switched the NATS backend from Streaming to JetStream, since NATS Streaming is depreciated.
+
+https://github.com/owncloud/ocis/pull/3192
+https://github.com/cs3org/reva/pull/2574

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/GeertJohan/yubigo v0.0.0-20190917122436-175bc097e60e
 	github.com/ReneKroon/ttlcache/v2 v2.11.0
 	github.com/asim/go-micro/plugins/client/grpc/v4 v4.0.0-20220118152736-9e0be6c85d75
-	github.com/asim/go-micro/plugins/events/nats/v4 v4.0.0-20220118152736-9e0be6c85d75
+	github.com/asim/go-micro/plugins/events/natsjs/v4 v4.0.0-20220311080335-e5a35d38f931
 	github.com/asim/go-micro/plugins/logger/zerolog/v4 v4.0.0-20220118152736-9e0be6c85d75
 	github.com/asim/go-micro/plugins/registry/consul/v4 v4.0.0-20220118152736-9e0be6c85d75
 	github.com/asim/go-micro/plugins/registry/etcd/v4 v4.0.0-20220118152736-9e0be6c85d75
@@ -22,7 +22,7 @@ require (
 	github.com/blevesearch/bleve/v2 v2.3.1
 	github.com/coreos/go-oidc/v3 v3.1.0
 	github.com/cs3org/go-cs3apis v0.0.0-20220126114148-64c025ccdd19
-	github.com/cs3org/reva/v2 v2.0.0-20220324071330-5bbdbf17c339
+	github.com/cs3org/reva/v2 v2.0.0-20220324071614-82800d7ef768
 	github.com/disintegration/imaging v1.6.2
 	github.com/glauth/glauth/v2 v2.0.0-20211021011345-ef3151c28733
 	github.com/go-chi/chi/v5 v5.0.7
@@ -47,7 +47,6 @@ require (
 	github.com/mitchellh/mapstructure v1.4.3
 	github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826
 	github.com/nats-io/nats-server/v2 v2.7.4
-	github.com/nats-io/nats-streaming-server v0.24.3
 	github.com/nmcclain/asn1-ber v0.0.0-20170104154839-2661553a0484
 	github.com/nmcclain/ldap v0.0.0-20210720162743-7f8d1e44eeba
 	github.com/oklog/run v1.1.0
@@ -171,7 +170,6 @@ require (
 	github.com/hashicorp/go-plugin v1.4.3 // indirect
 	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
 	github.com/hashicorp/golang-lru v0.5.4 // indirect
-	github.com/hashicorp/raft v1.3.6 // indirect
 	github.com/hashicorp/serf v0.9.6 // indirect
 	github.com/hashicorp/yamux v0.0.0-20211028200310-0bc27b27de87 // indirect
 	github.com/huandu/xstrings v1.3.2 // indirect
@@ -194,7 +192,7 @@ require (
 	github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect
 	github.com/maxymania/go-system v0.0.0-20170110133659-647cc364bf0b // indirect
 	github.com/mendsley/gojwk v0.0.0-20141217222730-4d5ec6e58103 // indirect
-	github.com/miekg/dns v1.1.44 // indirect
+	github.com/miekg/dns v1.1.46 // indirect
 	github.com/mileusna/useragent v1.0.2 // indirect
 	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/minio/md5-simd v1.1.2 // indirect
@@ -212,7 +210,6 @@ require (
 	github.com/nats-io/nats.go v1.13.1-0.20220308171302-2f2f6968e98d // indirect
 	github.com/nats-io/nkeys v0.3.0 // indirect
 	github.com/nats-io/nuid v1.0.1 // indirect
-	github.com/nats-io/stan.go v0.10.2 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect
 	github.com/orcaman/concurrent-map v1.0.0 // indirect
 	github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect
@@ -260,7 +257,7 @@ require (
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5 // indirect
 	golang.org/x/text v0.3.7 // indirect
-	golang.org/x/time v0.0.0-20211116232009-f0f3c7e86c11 // indirect
+	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	golang.org/x/tools v0.1.8 // indirect
 	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

nats/pkg/command/server.go

@@ -37,9 +37,11 @@ func Server(cfg *config.Config) *cli.Command {
 
 			natsServer, err := nats.NewNATSServer(
 				ctx,
+				logging.NewLogWrapper(logger),
 				nats.Host(cfg.Nats.Host),
 				nats.Port(cfg.Nats.Port),
-				nats.Logger(logging.NewLogWrapper(logger)),
+				nats.ClusterID(cfg.Nats.ClusterID),
+				nats.StoreDir(cfg.Nats.StoreDir),
 			)
 			if err != nil {
 				return err

nats/pkg/config/config.go

@@ -22,6 +22,8 @@ type Config struct {
 
 // Nats is the nats config
 type Nats struct {
-	Host string `ocisConfig:"host" env:"NATS_NATS_HOST"`
-	Port int    `ocisConfig:"port" env:"NATS_NATS_PORT"`
+	Host      string `ocisConfig:"host" env:"NATS_NATS_HOST"`
+	Port      int    `ocisConfig:"port" env:"NATS_NATS_PORT"`
+	ClusterID string `ocisConfig:"clusterid" env:"NATS_NATS_CLUSTER_ID"`
+	StoreDir  string `ocisConfig:"store_dir" env:"NATS_NATS_STORE_DIR"`
 }

nats/pkg/config/defaults/defaultconfig.go

@@ -1,6 +1,11 @@
 package defaults
 
-import "github.com/owncloud/ocis/nats/pkg/config"
+import (
+	"path"
+
+	"github.com/owncloud/ocis/nats/pkg/config"
+	"github.com/owncloud/ocis/ocis-pkg/config/defaults"
+)
 
 // NOTE: Most of this configuration is not needed to keep it as simple as possible
 // TODO: Clean up unneeded configuration
@@ -20,8 +25,10 @@ func DefaultConfig() *config.Config {
 			Name: "nats",
 		},
 		Nats: config.Nats{
-			Host: "127.0.0.1",
-			Port: 9233,
+			Host:      "127.0.0.1",
+			Port:      9233,
+			ClusterID: "ocis-cluster",
+			StoreDir:  path.Join(defaults.BaseDataPath(), "nats"),
 		},
 	}
 }

nats/pkg/server/nats/nats.go

@@ -4,64 +4,44 @@ import (
 	"context"
 	"time"
 
-	natsServer "github.com/nats-io/nats-server/v2/server"
-	stanServer "github.com/nats-io/nats-streaming-server/server"
+	nserver "github.com/nats-io/nats-server/v2/server"
 )
 
 var NATSListenAndServeLoopTimer = 1 * time.Second
 
 type NATSServer struct {
-	ctx context.Context
-
-	natsOpts *natsServer.Options
-	stanOpts *stanServer.Options
-
-	server *stanServer.StanServer
+	ctx    context.Context
+	server *nserver.Server
 }
 
-// NewNATSServer returns a new NATSServer
-func NewNATSServer(ctx context.Context, opts ...Option) (*NATSServer, error) {
-
-	server := &NATSServer{
-		ctx:      ctx,
-		natsOpts: &stanServer.DefaultNatsServerOptions,
-		stanOpts: stanServer.GetDefaultOptions(),
-	}
+func NewNATSServer(ctx context.Context, logger nserver.Logger, opts ...NatsOption) (*NATSServer, error) {
+	natsOpts := &nserver.Options{}
 
 	for _, o := range opts {
-		o(server.natsOpts, server.stanOpts)
+		o(natsOpts)
 	}
 
-	return server, nil
-}
+	// enable JetStream
+	natsOpts.JetStream = true
 
-// ListenAndServe runs the NATSServer in a blocking way until the server is shutdown or an error occurs
-func (n *NATSServer) ListenAndServe() (err error) {
-	n.server, err = stanServer.RunServerWithOpts(
-		n.stanOpts,
-		n.natsOpts,
-	)
+	server, err := nserver.NewServer(natsOpts)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
-	defer n.Shutdown()
+	server.SetLoggerV2(logger, true, true, false)
 
-	for {
-		// check if NATs server has an encountered an error
-		if err := n.server.LastError(); err != nil {
-			return err
-		}
-		// check if the NATs server is still running
-		if n.server.State() == stanServer.Shutdown {
-			return nil
-		}
-		// check if context was cancelled
-		if n.ctx.Err() != nil {
-			return nil
-		}
-		time.Sleep(NATSListenAndServeLoopTimer)
-	}
+	return &NATSServer{
+		ctx:    ctx,
+		server: server,
+	}, nil
+}
+
+// ListenAndServe runs the NATSServer in a blocking way until the server is shutdown or an error occurs
+func (n *NATSServer) ListenAndServe() (err error) {
+	go n.server.Start()
+	<-n.ctx.Done()
+	return nil
 }
 
 func (n *NATSServer) Shutdown() {

nats/pkg/server/nats/options.go

@@ -1,31 +1,36 @@
 package nats
 
 import (
-	natsServer "github.com/nats-io/nats-server/v2/server"
-	"github.com/nats-io/nats-streaming-server/logger"
-	stanServer "github.com/nats-io/nats-streaming-server/server"
+	nserver "github.com/nats-io/nats-server/v2/server"
 )
 
-// Option configures the nats server
-type Option func(*natsServer.Options, *stanServer.Options)
+// NatsOption configures the nats server
+type NatsOption func(*nserver.Options)
 
 // Host sets the host URL for the nats server
-func Host(url string) Option {
-	return func(no *natsServer.Options, _ *stanServer.Options) {
-		no.Host = url
+func Host(url string) NatsOption {
+	return func(o *nserver.Options) {
+		o.Host = url
 	}
 }
 
 // Port sets the host URL for the nats server
-func Port(port int) Option {
-	return func(no *natsServer.Options, _ *stanServer.Options) {
-		no.Port = port
+func Port(port int) NatsOption {
+	return func(o *nserver.Options) {
+		o.Port = port
 	}
 }
 
-// Port sets the host URL for the nats server
-func Logger(logger logger.Logger) Option {
-	return func(no *natsServer.Options, so *stanServer.Options) {
-		so.CustomLogger = logger
+// ClusterID sets the name for the nats cluster
+func ClusterID(clusterID string) NatsOption {
+	return func(o *nserver.Options) {
+		o.Cluster.Name = clusterID
+	}
+}
+
+// StoreDir sets the folder for persistence
+func StoreDir(StoreDir string) NatsOption {
+	return func(o *nserver.Options) {
+		o.StoreDir = StoreDir
 	}
 }

notifications/pkg/command/server.go

@@ -3,7 +3,7 @@ package command
 import (
 	"fmt"
 
-	"github.com/asim/go-micro/plugins/events/nats/v4"
+	"github.com/asim/go-micro/plugins/events/natsjs/v4"
 	"github.com/cs3org/reva/v2/pkg/events"
 	"github.com/cs3org/reva/v2/pkg/events/server"
 	"github.com/owncloud/ocis/notifications/pkg/channels"
@@ -31,7 +31,10 @@ func Server(cfg *config.Config) *cli.Command {
 			}
 
 			evtsCfg := cfg.Notifications.Events
-			client, err := server.NewNatsStream(nats.Address(evtsCfg.Endpoint), nats.ClusterID(evtsCfg.Cluster))
+			client, err := server.NewNatsStream(
+				natsjs.Address(evtsCfg.Endpoint),
+				natsjs.ClusterID(evtsCfg.Cluster),
+			)
 			if err != nil {
 				return err
 			}

notifications/pkg/config/defaults/defaultconfig.go

@@ -28,7 +28,7 @@ func DefaultConfig() *config.Config {
 			},
 			Events: config.Events{
 				Endpoint:      "127.0.0.1:9233",
-				Cluster:       "test-cluster",
+				Cluster:       "ocis-cluster",
 				ConsumerGroup: "notifications",
 			},
 			RevaGateway:       "127.0.0.1:9142",

storage/pkg/config/defaults/defaultconfig.go

@@ -364,7 +364,7 @@ func DefaultConfig() *config.Config {
 				UserStorageMountID:               "",
 				Events: config.Events{
 					Address:   "127.0.0.1:9233",
-					ClusterID: "test-cluster",
+					ClusterID: "ocis-cluster",
 				},
 			},
 			StorageShares: config.StoragePort{