Activitylog event handler split

services/activitylog/pkg/apierrors/errors.go

@@ -0,0 +1,12 @@
+// apierrors package defines common business API errors that can be used across the service. It is intended to be used by both the service and the API layer to ensure consistent error handling and messaging.
+package apierrors
+
+import "errors"
+
+var (
+	ErrNotFound     = errors.New("query target not found")
+	ErrBadRequest   = errors.New("bad request")
+	ErrUnauthorized = errors.New("unauthorized")
+	ErrForbidden    = errors.New("forbidden")
+	ErrMissingEmail = errors.New("missing email address")
+)

services/activitylog/pkg/command/server.go

@@ -2,18 +2,18 @@ package command
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 
-	"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"
 	"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 +24,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/raw"
+	"github.com/opencloud-eu/reva/v2/pkg/rgrpc/todo/pool"
 )
 
 var _registeredEvents = []events.Unmarshaller{
@@ -62,19 +68,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 +97,102 @@ 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(cfg, kv,
+				activitylog.Logger(logger),
+				activitylog.MaxActivities(cfg.MaxActivities),
+			)
+			if err != nil {
+				logger.Error().Err(err).Msg("Failed to initialize activity log")
+				return err
+			}
+
+			if !cfg.HTTP.Disabled {
 
-			{
-				svc, err := http.Server(
+				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 := raw.FromConfig(ctx, connName, raw.Config{
+					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,
+					MaxAckPending:        cfg.Events.MaxAckPending,
+					AckWait:              cfg.Events.AckWait,
+				})
+				if err != nil {
+					logger.Error().Err(err).Msg("Failed to initialize event stream")
+					return err
+				}
 
+				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", "http").Msg("Failed to initialize server")
+					logger.Error().Err(err).Str("transport", "event").Msg("Failed to initialize server")
 					return err
 				}
 
-				gr.Add(runner.NewGoMicroHttpServerRunner(cfg.Service.Name+".http", svc))
+				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,47 @@ func Server(cfg *config.Config) *cobra.Command {
 		},
 	}
 }
+
+func ConnectNatsKV(cfg config.Store) (nats.KeyValue, error) {
+	// Connect to NATS servers
+	natsOptions := nats.Options{
+		Servers: cfg.Nodes,
+	}
+	if cfg.EnableTLS {
+		if cfg.TLSRootCACertificate != "" {
+			// when root ca is configured use it. an insecure flag is ignored.
+			nats.RootCAs(cfg.TLSRootCACertificate)(&natsOptions)
+		} else {
+			// enable tls and use insecure flag
+			nats.Secure(&tls.Config{MinVersion: tls.VersionTLS12, InsecureSkipVerify: cfg.TLSInsecure})(&natsOptions)
+		}
+	}
+	if cfg.AuthUsername != "" && cfg.AuthPassword != "" {
+		nats.UserInfo(cfg.AuthUsername, cfg.AuthPassword)(&natsOptions)
+	}
+	conn, err := natsOptions.Connect()
+	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,17 +35,21 @@ 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 {
-	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"`
-	TLSRootCACertificate string `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided NOTIFICATIONS_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"1.0.0"`
-	EnableTLS            bool   `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
-	AuthUsername         string `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
-	AuthPassword         string `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	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"`
+	TLSRootCACertificate string        `yaml:"tls_root_ca_certificate" env:"OC_EVENTS_TLS_ROOT_CA_CERTIFICATE" desc:"The root CA certificate used to validate the server's TLS certificate. If provided NOTIFICATIONS_EVENTS_TLS_INSECURE will be seen as false." introductionVersion:"1.0.0"`
+	EnableTLS            bool          `yaml:"enable_tls" env:"OC_EVENTS_ENABLE_TLS" desc:"Enable TLS for the connection to the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	AuthUsername         string        `yaml:"username" env:"OC_EVENTS_AUTH_USERNAME" desc:"The username to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	AuthPassword         string        `yaml:"password" env:"OC_EVENTS_AUTH_PASSWORD" desc:"The password to authenticate with the events broker. The events broker is the OpenCloud service which receives and delivers events between the services." introductionVersion:"1.0.0"`
+	MaxAckPending        int           `yaml:"max_ack_pending" env:"ACTIVITYLOG_EVENTS_MAX_ACK_PENDING" desc:"The maximum number of unacknowledged messages. This is used to limit the number of messages that can be in flight at the same time." introductionVersion:"%NEXT%"`
+	AckWait              time.Duration `yaml:"ack_wait" env:"ACTIVITYLOG_EVENTS_ACK_WAIT" desc:"The time to wait for an ack before the message is redelivered. This is used to ensure that messages are not lost if the consumer crashes." introductionVersion:"%NEXT%"`
 }
 
 // Store configures the store to use
@@ -77,6 +81,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

@@ -29,9 +29,11 @@ func DefaultConfig() *config.Config {
 			Name: "activitylog",
 		},
 		Events: config.Events{
-			Endpoint:  "127.0.0.1:9233",
-			Cluster:   "opencloud-cluster",
-			EnableTLS: false,
+			Endpoint:      "127.0.0.1:9233",
+			Cluster:       "opencloud-cluster",
+			EnableTLS:     false,
+			MaxAckPending: 1000,
+			AckWait:       1 * time.Minute,
 		},
 		Store: config.Store{
 			Store:    "nats-js-kv",
@@ -52,7 +54,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/data/rawactivity.go

@@ -0,0 +1,10 @@
+package data
+
+import "time"
+
+// RawActivity represents an activity as it is stored in the activitylog store
+type RawActivity struct {
+	EventID   string    `json:"event_id"`
+	Depth     int       `json:"depth"`
+	Timestamp time.Time `json:"timestamp"`
+}