Activitylog event handler split

services/activitylog/pkg/command/server.go

@@ -3,17 +3,18 @@ package command
 import (
 	"context"
 	"fmt"
+	"strings"
 
-	"github.com/opencloud-eu/opencloud/pkg/log"
-	"github.com/opencloud-eu/opencloud/pkg/runner"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
+	"github.com/nats-io/nats.go"
+	"github.com/olekukonko/errors"
 	"github.com/spf13/cobra"
 
 	"github.com/opencloud-eu/opencloud/pkg/config/configlog"
 	"github.com/opencloud-eu/opencloud/pkg/generators"
+	"github.com/opencloud-eu/opencloud/pkg/log"
+	natspkg "github.com/opencloud-eu/opencloud/pkg/nats"
 	"github.com/opencloud-eu/opencloud/pkg/registry"
+	"github.com/opencloud-eu/opencloud/pkg/runner"
 	ogrpc "github.com/opencloud-eu/opencloud/pkg/service/grpc"
 	"github.com/opencloud-eu/opencloud/pkg/tracing"
 	"github.com/opencloud-eu/opencloud/pkg/version"
@@ -24,6 +25,12 @@ import (
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/metrics"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/server/debug"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/server/http"
+	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/activitylog"
+	svcEvents "github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/events"
+	svcHttp "github.com/opencloud-eu/opencloud/services/activitylog/pkg/service/http"
+	"github.com/opencloud-eu/reva/v2/pkg/events"
+	"github.com/opencloud-eu/reva/v2/pkg/events/stream"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 )
 
 var _registeredEvents = []events.Unmarshaller{
@@ -62,19 +69,11 @@ func Server(cfg *config.Config) *cobra.Command {
 
 			gr := runner.NewGroup()
 			ctx, cancel := context.WithCancel(cmd.Context())
+			defer cancel()
 
 			mtrcs := metrics.New()
 			mtrcs.BuildInfo.WithLabelValues(version.GetString()).Set(1)
 
-			defer cancel()
-
-			connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
-			evStream, err := stream.NatsFromConfig(connName, false, stream.NatsConfig(cfg.Events))
-			if err != nil {
-				logger.Error().Err(err).Msg("Failed to initialize event stream")
-				return err
-			}
-
 			tm, err := pool.StringToTLSMode(cfg.GRPCClientTLS.Mode)
 			if err != nil {
 				logger.Error().Err(err).Msg("Failed to parse tls mode")
@@ -99,28 +98,101 @@ func Server(cfg *config.Config) *cobra.Command {
 				return err
 			}
 
-			hClient := ehsvc.NewEventHistoryService("eu.opencloud.api.eventhistory", grpcClient)
-			vClient := settingssvc.NewValueService("eu.opencloud.api.settings", grpcClient)
+			kv, err := ConnectNatsKV(cfg.Store)
+			if err != nil {
+				return err
+			}
+			activityLog, err := activitylog.New(kv,
+				activitylog.Logger(logger),
+				activitylog.MaxActivities(cfg.MaxActivities),
+				activitylog.WriteBufferDuration(cfg.WriteBufferDuration),
+			)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to initialize activity log")
+				return err
+			}
 
-			{
-				svc, err := http.Server(
+			if !cfg.HTTP.Disabled {
+
+				hClient := ehsvc.NewEventHistoryService("eu.opencloud.api.eventhistory", grpcClient)
+
+				svc, err := svcHttp.New(
+					activityLog,
+					svcHttp.Logger(logger),
+					svcHttp.GatewaySelector(gatewaySelector),
+					svcHttp.RegisteredEvents(_registeredEvents),
+					//svcHttp.TraceProvider(tracerProvider),
+					svcHttp.HistoryClient(hClient),
+				)
+				if err != nil {
+					logger.Error().Err(err).Msg("handler init")
+					return err
+				}
+				// TODO svc = service.NewInstrument(svc, metrics)
+				// TODO svc = service.NewLogging(svc, logger) // this logs service specific data
+				// TODO svc = service.NewTracing(svc, traceProvider)
+				vClient := settingssvc.NewValueService("eu.opencloud.api.settings", grpcClient)
+
+				server, err := http.Server(
+					http.ValueClient(vClient),
 					http.Logger(logger),
+					http.Context(ctx),
 					http.Config(cfg),
-					http.Context(ctx), // NOTE: not passing this "option" leads to a panic in go-micro
-					http.TraceProvider(tracerProvider),
-					http.Stream(evStream),
-					http.GatewaySelector(gatewaySelector),
-					http.HistoryClient(hClient),
-					http.ValueClient(vClient),
-					http.RegisteredEvents(_registeredEvents),
+					http.Service(svc),
 				)
+				if err != nil {
+					logger.Info().
+						Err(err).
+						Str("transport", "http").
+						Msg("Failed to initialize server")
+
+					return err
+				}
+
+				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", server))
+			} else {
+				logger.Info().Msg("HTTP server disabled, not starting HTTP service")
+			}
 
+			if !cfg.Events.Disabled {
+
+				connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus)
+				evStream, err := stream.NatsFromConfig(connName, false, stream.NatsConfig{
+					Endpoint:             cfg.Events.Endpoint,
+					Cluster:              cfg.Events.Cluster,
+					EnableTLS:            cfg.Events.EnableTLS,
+					TLSInsecure:          cfg.Events.TLSInsecure,
+					TLSRootCACertificate: cfg.Events.TLSRootCACertificate,
+					AuthUsername:         cfg.Events.AuthUsername,
+					AuthPassword:         cfg.Events.AuthPassword,
+				})
 				if err != nil {
-					logger.Error().Err(err).Str("transport", "http").Msg("Failed to initialize server")
+					logger.Error().Err(err).Msg("Failed to initialize event stream")
 					return err
 				}
 
-				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", svc))
+				eventSvc, err := svcEvents.New(
+					activityLog,
+					evStream,
+					svcEvents.Context(ctx),
+					svcEvents.Logger(logger),
+					svcEvents.ServiceAccount(cfg.ServiceAccount),
+					svcEvents.GatewaySelector(gatewaySelector),
+					svcEvents.RegisteredEvents(_registeredEvents),
+					svcEvents.NumConsumers(cfg.NumConsumers),
+				)
+				if err != nil {
+					logger.Error().Err(err).Str("transport", "event").Msg("Failed to initialize server")
+					return err
+				}
+
+				gr.Add(runner.New(cfg.Service.Name+".svc", func() error {
+					return eventSvc.Run()
+				}, func() {
+					eventSvc.Close()
+				}))
+			} else {
+				logger.Info().Msg("event listening disabled, not starting event service")
 			}
 
 			{
@@ -149,3 +221,33 @@ func Server(cfg *config.Config) *cobra.Command {
 		},
 	}
 }
+
+func ConnectNatsKV(cfg config.Store) (nats.KeyValue, error) {
+	// Connect to NATS servers
+	secureOption := natspkg.Secure(cfg.EnableTLS, cfg.TLSInsecure, cfg.TLSRootCACertificate)
+	conn, err := nats.Connect(strings.Join(cfg.Nodes, ","), secureOption, nats.UserInfo(cfg.AuthUsername, cfg.AuthPassword))
+	if err != nil {
+		return nil, err
+	}
+
+	js, err := conn.JetStream()
+	if err != nil {
+		return nil, err
+	}
+
+	kv, err := js.KeyValue(cfg.Database)
+	if err != nil {
+		if !errors.Is(err, nats.ErrBucketNotFound) {
+			return nil, errors.Wrapf(err, "Failed to get bucket (%s)", cfg.Database)
+		}
+
+		kv, err = js.CreateKeyValue(&nats.KeyValueConfig{
+			Bucket: cfg.Database,
+		})
+		if err != nil {
+			return nil, errors.Wrapf(err, "Failed to create bucket (%s)", cfg.Database)
+		}
+	}
+
+	return kv, nil
+}

services/activitylog/pkg/config/config.go

@@ -35,10 +35,12 @@ type Config struct {
 
 	WriteBufferDuration time.Duration `yaml:"write_buffer_duration" env:"ACTIVITYLOG_WRITE_BUFFER_DURATION" desc:"The duration to wait before flushing the write buffer. This is used to reduce the number of writes to the store." introductionVersion:"4.0.0"`
 	MaxActivities       int           `yaml:"max_activities" env:"ACTIVITYLOG_MAX_ACTIVITIES" desc:"The maximum number of activities to keep in the store per resource. If the number of activities exceeds this value, the oldest activities will be removed." introductionVersion:"4.0.0"`
+	NumConsumers        int           `yaml:"num_consumers" env:"ACTIVITYLOG_NUM_CONSUMERS" desc:"The amount of concurrent event consumers to start. Event consumers are used for updating the list of activities. Multiple consumers increase parallelisation, but will also increase CPU and memory demands." introductionVersion:"%NEXT%"`
 }
 
 // Events combines the configuration options for the event bus.
 type Events struct {
+	Disabled             bool   `yaml:"disabled" env:"ACTIVITYLOG_EVENTS_DISABLED" desc:"Disables listening for events. Set this to true if the service should only handle HTTP requests." introductionVersion:"%NEXT%"`
 	Endpoint             string `yaml:"endpoint" env:"OC_EVENTS_ENDPOINT" desc:"The address of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture." introductionVersion:"1.0.0"`
 	Cluster              string `yaml:"cluster" env:"OC_EVENTS_CLUSTER" desc:"The clusterID of the event system. The event system is the message queuing service. It is used as message broker for the microservice architecture. Mandatory when using NATS as event system." introductionVersion:"1.0.0"`
 	TLSInsecure          bool   `yaml:"tls_insecure" env:"OC_INSECURE;OC_EVENTS_TLS_INSECURE" desc:"Whether to verify the server TLS certificates." introductionVersion:"1.0.0"`
@@ -77,6 +79,7 @@ type CORS struct {
 
 // HTTP defines the available http configuration.
 type HTTP struct {
+	Disabled  bool                  `yaml:"disabled" env:"ACTIVITYLOG_HTTP_DISABLED" desc:"Disables the HTTP service. Set this to true if the service should only handle events." introductionVersion:"1.0.0"`
 	Addr      string                `yaml:"addr" env:"ACTIVITYLOG_HTTP_ADDR" desc:"The bind address of the HTTP service." introductionVersion:"1.0.0"`
 	Namespace string                `yaml:"-"`
 	Root      string                `yaml:"root" env:"ACTIVITYLOG_HTTP_ROOT" desc:"Subdirectory that serves as the root for this HTTP service." introductionVersion:"1.0.0"`

services/activitylog/pkg/config/defaults/defaultconfig.go

@@ -52,7 +52,9 @@ func DefaultConfig() *config.Config {
 			},
 		},
 		WriteBufferDuration: 10 * time.Second,
-		MaxActivities:       6000,
+		// Nats runs into max payload exceeded errors at around 7k activities. Let's keep a buffer.
+		MaxActivities: 6000,
+		NumConsumers:  1,
 	}
 }
 

services/activitylog/pkg/server/http/option.go

@@ -3,38 +3,28 @@ package http
 import (
 	"context"
 
-	gateway "github.com/cs3org/go-cs3apis/cs3/gateway/v1beta1"
 	"github.com/opencloud-eu/opencloud/pkg/log"
-	ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0"
 	settingssvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/settings/v0"
 	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/config"
-	"github.com/opencloud-eu/opencloud/services/activitylog/pkg/metrics"
-	"github.com/opencloud-eu/reva/v2/pkg/events"
-	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
-
 	"github.com/spf13/pflag"
-	"go-micro.dev/v4/store"
 	"go.opentelemetry.io/otel/trace"
+	"go.opentelemetry.io/otel/trace/noop"
 )
 
 // Option defines a single option function.
 type Option func(o *Options)
 
 // Options defines the available options for this package.
 type Options struct {
-	Logger           log.Logger
-	Context          context.Context
-	Config           *config.Config
-	Metrics          *metrics.Metrics
-	Flags            []pflag.Flag
-	Namespace        string
-	Store            store.Store
-	Stream           events.Stream
-	GatewaySelector  pool.Selectable[gateway.GatewayAPIClient]
-	TraceProvider    trace.TracerProvider
-	HistoryClient    ehsvc.EventHistoryService
-	ValueClient      settingssvc.ValueService
-	RegisteredEvents []events.Unmarshaller
+	Name          string
+	Namespace     string
+	Logger        log.Logger
+	Context       context.Context
+	Config        *config.Config
+	Flags         []pflag.Flag
+	Service       ActivityLogService
+	TraceProvider trace.TracerProvider
+	ValueClient   settingssvc.ValueService
 }
 
 // newOptions initializes the available default options.
@@ -69,10 +59,10 @@ func Config(val *config.Config) Option {
 	}
 }
 
-// Metrics provides a function to set the metrics option.
-func Metrics(val *metrics.Metrics) Option {
+// Service provides a function to set the service option.
+func Service(val ActivityLogService) Option {
 	return func(o *Options) {
-		o.Metrics = val
+		o.Service = val
 	}
 }
 
@@ -83,58 +73,20 @@ func Flags(flags ...pflag.Flag) Option {
 	}
 }
 
-// Namespace provides a function to set the Namespace option.
-func Namespace(val string) Option {
-	return func(o *Options) {
-		o.Namespace = val
-	}
-}
-
-// Store provides a function to configure the store
-func Store(store store.Store) Option {
-	return func(o *Options) {
-		o.Store = store
-	}
-}
-
-// Stream provides a function to configure the stream
-func Stream(stream events.Stream) Option {
-	return func(o *Options) {
-		o.Stream = stream
-	}
-}
-
-// GatewaySelector provides a function to configure the gateway client selector
-func GatewaySelector(gatewaySelector pool.Selectable[gateway.GatewayAPIClient]) Option {
-	return func(o *Options) {
-		o.GatewaySelector = gatewaySelector
-	}
-}
-
-// HistoryClient provides a function to configure the event history client
-func HistoryClient(h ehsvc.EventHistoryService) Option {
-	return func(o *Options) {
-		o.HistoryClient = h
-	}
-}
-
-// RegisteredEvents provides a function to register events
-func RegisteredEvents(evs []events.Unmarshaller) Option {
-	return func(o *Options) {
-		o.RegisteredEvents = evs
-	}
-}
-
-// TraceProvider provides a function to set the TracerProvider option
-func TraceProvider(val trace.TracerProvider) Option {
+// TraceProvider provides a function to configure the trace provider
+func TraceProvider(traceProvider trace.TracerProvider) Option {
 	return func(o *Options) {
-		o.TraceProvider = val
+		if traceProvider != nil {
+			o.TraceProvider = traceProvider
+		} else {
+			o.TraceProvider = noop.NewTracerProvider()
+		}
 	}
 }
 
-// ValueClient provides a function to set the ValueClient options
-func ValueClient(val settingssvc.ValueService) Option {
+// ValueClient adds a grpc client for the value service
+func ValueClient(vs settingssvc.ValueService) Option {
 	return func(o *Options) {
-		o.ValueClient = val
+		o.ValueClient = vs
 	}
 }