From 6e87f6665b10ff7071899b68338a6ed172dfb0d5 Mon Sep 17 00:00:00 2001 From: Alex Ababii Date: Fri, 14 Aug 2026 11:56:23 +0200 Subject: [PATCH] eventhistory service: fix consumer isolation and conditional initialization --- services/eventhistory/pkg/command/server.go | 51 +++--- .../pkg/server/consumer/option.go | 58 +++++++ .../pkg/server/consumer/server.go | 78 +++++++++ .../eventhistory/pkg/server/grpc/option.go | 18 -- .../eventhistory/pkg/server/grpc/server.go | 13 -- services/eventhistory/pkg/service/service.go | 45 +---- .../eventhistory/pkg/service/service_test.go | 163 ++++++++++-------- 7 files changed, 257 insertions(+), 169 deletions(-) create mode 100644 services/eventhistory/pkg/server/consumer/option.go create mode 100644 services/eventhistory/pkg/server/consumer/server.go diff --git a/services/eventhistory/pkg/command/server.go b/services/eventhistory/pkg/command/server.go index 48e489d685..66ec3d591f 100644 --- a/services/eventhistory/pkg/command/server.go +++ b/services/eventhistory/pkg/command/server.go @@ -6,19 +6,19 @@ import ( "os/signal" "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/runner" ogrpc "github.com/opencloud-eu/opencloud/pkg/service/grpc" "github.com/opencloud-eu/opencloud/pkg/tracing" "github.com/opencloud-eu/opencloud/pkg/version" + ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config/parser" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/metrics" + "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/consumer" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/debug" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/grpc" - "github.com/opencloud-eu/reva/v2/pkg/events" - "github.com/opencloud-eu/reva/v2/pkg/events/stream" + svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service" "github.com/opencloud-eu/reva/v2/pkg/store" "github.com/spf13/cobra" @@ -59,25 +59,6 @@ func Server(cfg *config.Config) *cobra.Command { gr := runner.NewGroup() - var consumer events.Stream - if !cfg.Events.Disabled { - connName := generators.GenerateConnectionName(cfg.Service.Name, generators.NTypeBus) - consumer, err = stream.NatsFromConfig(connName, false, stream.NatsConfig{ - Endpoint: cfg.Events.Endpoint, - Cluster: cfg.Events.Cluster, - TLSInsecure: cfg.Events.TLSInsecure, - TLSRootCACertificate: cfg.Events.TLSRootCACertificate, - EnableTLS: cfg.Events.EnableTLS, - AuthUsername: cfg.Events.AuthUsername, - AuthPassword: cfg.Events.AuthPassword, - }) - if err != nil { - return err - } - } else { - logger.Info().Msg("event listening disabled, not starting event consumer") - } - st := store.Create( store.Store(cfg.Store.Store), store.TTL(cfg.Store.TTL), @@ -90,7 +71,27 @@ func Server(cfg *config.Config) *cobra.Command { store.TLSRootCA(cfg.Store.TLSRootCACertificate), ) + if !cfg.Events.Disabled { + evConsumer, err := consumer.NewConsumer( + consumer.Logger(logger), + consumer.Config(cfg), + consumer.Persistence(st), + ) + if err != nil { + return err + } + + go evConsumer.StoreEvents() + } else { + logger.Info().Msg("event listening disabled, not starting event consumer") + } + if !cfg.GRPC.Disabled { + eh, err := svc.NewEventHistoryService(cfg, st, logger) + if err != nil { + return err + } + service := grpc.NewService( grpc.Logger(logger), grpc.Context(ctx), @@ -99,11 +100,13 @@ func Server(cfg *config.Config) *cobra.Command { grpc.Namespace(cfg.GRPC.Namespace), grpc.Address(cfg.GRPC.Addr), grpc.Metrics(m), - grpc.Consumer(consumer), - grpc.Persistence(st), grpc.TraceProvider(traceProvider), ) + if err := ehsvc.RegisterEventHistoryServiceHandler(service.Server(), eh); err != nil { + return err + } + gr.Add(runner.NewGoMicroGrpcServerRunner(cfg.Service.Name+".grpc", service)) } else { logger.Info().Msg("gRPC server disabled, not starting gRPC service") diff --git a/services/eventhistory/pkg/server/consumer/option.go b/services/eventhistory/pkg/server/consumer/option.go new file mode 100644 index 0000000000..a12a1973db --- /dev/null +++ b/services/eventhistory/pkg/server/consumer/option.go @@ -0,0 +1,58 @@ +package consumer + +import ( + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" + "github.com/opencloud-eu/reva/v2/pkg/events" + "go-micro.dev/v4/store" +) + +// 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 + Config *config.Config + Persistence store.Store + Consumer events.Consumer +} + +// newOptions initializes the available default options. +func newOptions(opts ...Option) Options { + opt := Options{} + + for _, o := range opts { + o(&opt) + } + + return opt +} + +// Logger provides a function to set the logger option. +func Logger(val log.Logger) Option { + return func(o *Options) { + o.Logger = val + } +} + +// Config provides a function to set the config option. +func Config(val *config.Config) Option { + return func(o *Options) { + o.Config = val + } +} + +// Persistence provides a function to configure the store +func Persistence(store store.Store) Option { + return func(o *Options) { + o.Persistence = store + } +} + +// Stream provides a function to set the event source to consume from. +func Stream(consumer events.Consumer) Option { + return func(o *Options) { + o.Consumer = consumer + } +} diff --git a/services/eventhistory/pkg/server/consumer/server.go b/services/eventhistory/pkg/server/consumer/server.go new file mode 100644 index 0000000000..25aa38277c --- /dev/null +++ b/services/eventhistory/pkg/server/consumer/server.go @@ -0,0 +1,78 @@ +package consumer + +import ( + "encoding/json" + + "github.com/opencloud-eu/opencloud/pkg/generators" + "github.com/opencloud-eu/opencloud/pkg/log" + "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" + svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service" + "github.com/opencloud-eu/reva/v2/pkg/events" + "github.com/opencloud-eu/reva/v2/pkg/events/stream" + "go-micro.dev/v4/store" +) + +// Consumer consumes all events and stores them in the store +type Consumer struct { + ch <-chan events.Event + store store.Store + cfg *config.Config + log log.Logger +} + +// NewConsumer initializes the event consumer +func NewConsumer(opts ...Option) (*Consumer, error) { + options := newOptions(opts...) + + cons := options.Consumer + if cons == nil { + connName := generators.GenerateConnectionName(options.Config.Service.Name, generators.NTypeBus) + var err error + cons, err = stream.NatsFromConfig(connName, false, stream.NatsConfig{ + Endpoint: options.Config.Events.Endpoint, + Cluster: options.Config.Events.Cluster, + TLSInsecure: options.Config.Events.TLSInsecure, + TLSRootCACertificate: options.Config.Events.TLSRootCACertificate, + EnableTLS: options.Config.Events.EnableTLS, + AuthUsername: options.Config.Events.AuthUsername, + AuthPassword: options.Config.Events.AuthPassword, + }) + if err != nil { + return nil, err + } + } + + ch, err := events.ConsumeAll(cons, "evhistory") + if err != nil { + return nil, err + } + + return &Consumer{ch: ch, store: options.Persistence, cfg: options.Config, log: options.Logger}, nil +} + +// StoreEvents consumes all events and stores them in the store. Will block +func (c *Consumer) StoreEvents() { + for event := range c.ch { + ev, err := json.Marshal(svc.StoreEvent{ + ID: event.ID, + Type: event.Type, + Event: event.Event.([]byte), + }) + if err != nil { + c.log.Error().Err(err).Str("eventid", event.ID).Msg("could not marshal event") + continue + } + if err := c.store.Write(&store.Record{ + Key: event.ID, + Value: ev, + Expiry: c.cfg.Store.TTL, + Metadata: map[string]any{ + "type": event.Type, + }, + }); err != nil { + // we can't store. That's it for us. + c.log.Error().Err(err).Str("eventid", event.ID).Msg("could not store event") + continue + } + } +} diff --git a/services/eventhistory/pkg/server/grpc/option.go b/services/eventhistory/pkg/server/grpc/option.go index fa0653a675..2ebb42450c 100644 --- a/services/eventhistory/pkg/server/grpc/option.go +++ b/services/eventhistory/pkg/server/grpc/option.go @@ -6,8 +6,6 @@ import ( "github.com/opencloud-eu/opencloud/pkg/log" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/metrics" - "github.com/opencloud-eu/reva/v2/pkg/events" - "go-micro.dev/v4/store" "go.opentelemetry.io/otel/trace" ) @@ -23,8 +21,6 @@ type Options struct { Config *config.Config Metrics *metrics.Metrics Namespace string - Persistence store.Store - Consumer events.Consumer TraceProvider trace.TracerProvider } @@ -88,20 +84,6 @@ func Namespace(val string) Option { } } -// Persistence provides a function to configure the store -func Persistence(store store.Store) Option { - return func(o *Options) { - o.Persistence = store - } -} - -// Consumer provides a function to configure the consumer -func Consumer(consumer events.Consumer) Option { - return func(o *Options) { - o.Consumer = consumer - } -} - // TraceProvider provides a function to configure the trace provider func TraceProvider(traceProvider trace.TracerProvider) Option { return func(o *Options) { diff --git a/services/eventhistory/pkg/server/grpc/server.go b/services/eventhistory/pkg/server/grpc/server.go index 311e331a90..b5d149ebb1 100644 --- a/services/eventhistory/pkg/server/grpc/server.go +++ b/services/eventhistory/pkg/server/grpc/server.go @@ -3,8 +3,6 @@ package grpc import ( "github.com/opencloud-eu/opencloud/pkg/service/grpc" "github.com/opencloud-eu/opencloud/pkg/version" - ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0" - svc "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service" ) // NewService initializes the grpc service and server. @@ -32,16 +30,5 @@ func NewService(opts ...Option) grpc.Service { return grpc.Service{} } - eh, err := svc.NewEventHistoryService(options.Config, options.Consumer, options.Persistence, options.Logger) - if err != nil { - options.Logger.Fatal().Err(err).Msg("Error creating event history service") - return grpc.Service{} - } - - _ = ehsvc.RegisterEventHistoryServiceHandler( - service.Server(), - eh, - ) - return service } diff --git a/services/eventhistory/pkg/service/service.go b/services/eventhistory/pkg/service/service.go index 7599a58e16..521c3dc860 100644 --- a/services/eventhistory/pkg/service/service.go +++ b/services/eventhistory/pkg/service/service.go @@ -11,7 +11,6 @@ import ( ehmsg "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/messages/eventhistory/v0" ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" - "github.com/opencloud-eu/reva/v2/pkg/events" "go-micro.dev/v4/store" ) @@ -24,54 +23,14 @@ type StoreEvent struct { // EventHistoryService is the service responsible for event history type EventHistoryService struct { - ch <-chan events.Event store store.Store cfg *config.Config log log.Logger } // NewEventHistoryService returns an EventHistory service -func NewEventHistoryService(cfg *config.Config, consumer events.Consumer, store store.Store, log log.Logger) (*EventHistoryService, error) { - eh := &EventHistoryService{store: store, cfg: cfg, log: log} - - if consumer != nil { - ch, err := events.ConsumeAll(consumer, "evhistory") - if err != nil { - return nil, err - } - - eh.ch = ch - go eh.StoreEvents() - } - - return eh, nil -} - -// StoreEvents consumes all events and stores them in the store. Will block -func (eh *EventHistoryService) StoreEvents() { - for event := range eh.ch { - ev, err := json.Marshal(StoreEvent{ - ID: event.ID, - Type: event.Type, - Event: event.Event.([]byte), - }) - if err != nil { - eh.log.Error().Err(err).Str("eventid", event.ID).Msg("could not marshal event") - continue - } - if err := eh.store.Write(&store.Record{ - Key: event.ID, - Value: ev, - Expiry: eh.cfg.Store.TTL, - Metadata: map[string]any{ - "type": event.Type, - }, - }); err != nil { - // we can't store. That's it for us. - eh.log.Error().Err(err).Str("eventid", event.ID).Msg("could not store event") - continue - } - } +func NewEventHistoryService(cfg *config.Config, store store.Store, log log.Logger) (*EventHistoryService, error) { + return &EventHistoryService{store: store, cfg: cfg, log: log}, nil } // GetEvents allows retrieving events from the eventstore by id diff --git a/services/eventhistory/pkg/service/service_test.go b/services/eventhistory/pkg/service/service_test.go index afafc207b5..ed712d9a6e 100644 --- a/services/eventhistory/pkg/service/service_test.go +++ b/services/eventhistory/pkg/service/service_test.go @@ -14,6 +14,7 @@ import ( "github.com/opencloud-eu/opencloud/pkg/log" ehsvc "github.com/opencloud-eu/opencloud/protogen/gen/opencloud/services/eventhistory/v0" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/config" + "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/server/consumer" "github.com/opencloud-eu/opencloud/services/eventhistory/pkg/service" "github.com/opencloud-eu/reva/v2/pkg/events" "github.com/opencloud-eu/reva/v2/pkg/store" @@ -34,103 +35,123 @@ var _ = Describe("EventHistoryService", func() { BeforeEach(func() { var err error sto = store.Create() - bus = testBus(make(chan events.Event)) - eh, err = service.NewEventHistoryService(cfg, bus, sto, log.Logger{}) + eh, err = service.NewEventHistoryService(cfg, sto, log.Logger{}) Expect(err).ToNot(HaveOccurred()) }) - AfterEach(func() { - close(bus) - }) + Context("with event consumer", func() { + BeforeEach(func() { + bus = testBus(make(chan events.Event)) + evConsumer, err := consumer.NewConsumer( + consumer.Logger(log.Logger{}), + consumer.Config(cfg), + consumer.Persistence(sto), + consumer.Stream(bus), + ) + Expect(err).ToNot(HaveOccurred()) + go evConsumer.StoreEvents() + }) - It("Records events, stores them and allows them to be retrieved", func() { - id := bus.Publish(events.UploadReady{}) + AfterEach(func() { + close(bus) + }) - // service will store eventually - time.Sleep(500 * time.Millisecond) + It("Records events, stores them and allows them to be retrieved", func() { + id := bus.Publish(events.UploadReady{}) - resp := &ehsvc.GetEventsResponse{} - err := eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp) - Expect(err).ToNot(HaveOccurred()) - Expect(resp).ToNot(BeNil()) + // service will store eventually + time.Sleep(500 * time.Millisecond) - Expect(len(resp.Events)).To(Equal(1)) - Expect(resp.Events[0].Id).To(Equal(id)) - }) + resp := &ehsvc.GetEventsResponse{} + err := eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp) + Expect(err).ToNot(HaveOccurred()) + Expect(resp).ToNot(BeNil()) - It("Gets all events", func() { - ids := make([]string, 3) - ids[0] = bus.Publish(events.UploadReady{ - ExecutingUser: &userv1beta1.User{ - Id: &userv1beta1.UserId{ - OpaqueId: "test-id", + Expect(len(resp.Events)).To(Equal(1)) + Expect(resp.Events[0].Id).To(Equal(id)) + }) + + It("Gets all events", func() { + ids := make([]string, 3) + ids[0] = bus.Publish(events.UploadReady{ + ExecutingUser: &userv1beta1.User{ + Id: &userv1beta1.UserId{ + OpaqueId: "test-id", + }, }, - }, - Failed: false, - Timestamp: utils.TimeToTS(time.Time{}), - }) - ids[1] = bus.Publish(events.UserCreated{ - UserID: "another-id", - }) - ids[2] = bus.Publish(events.UserDeleted{ - Executant: &userv1beta1.UserId{ - OpaqueId: "another-id", - }, - UserID: "test-id", + Failed: false, + Timestamp: utils.TimeToTS(time.Time{}), + }) + ids[1] = bus.Publish(events.UserCreated{ + UserID: "another-id", + }) + ids[2] = bus.Publish(events.UserDeleted{ + Executant: &userv1beta1.UserId{ + OpaqueId: "another-id", + }, + UserID: "test-id", + }) + + time.Sleep(500 * time.Millisecond) + + resp := &ehsvc.GetEventsResponse{} + err := eh.GetEventsForUser(context.Background(), &ehsvc.GetEventsForUserRequest{UserID: "test-id"}, resp) + Expect(err).ToNot(HaveOccurred()) + Expect(resp).ToNot(BeNil()) + + // Events don't always come back in the same order as they were sent, so we need to sort them and + // do the same for the expected IDs as well. + expectedIDs := []string{ids[0], ids[2]} + sort.Strings(expectedIDs) + var gotIDs []string + for _, ev := range resp.Events { + gotIDs = append(gotIDs, ev.Id) + } + sort.Strings(gotIDs) + + Expect(len(gotIDs)).To(Equal(len(expectedIDs))) + Expect(gotIDs[0]).To(Equal(expectedIDs[0])) + Expect(gotIDs[1]).To(Equal(expectedIDs[1])) }) - time.Sleep(500 * time.Millisecond) + It("Stores events and verifies them in the store directly (consumer-only mode)", func() { + id := bus.Publish(events.UploadReady{}) + time.Sleep(500 * time.Millisecond) - resp := &ehsvc.GetEventsResponse{} - err := eh.GetEventsForUser(context.Background(), &ehsvc.GetEventsForUserRequest{UserID: "test-id"}, resp) - Expect(err).ToNot(HaveOccurred()) - Expect(resp).ToNot(BeNil()) + records, err := sto.Read(id) + Expect(err).ToNot(HaveOccurred()) + Expect(len(records)).To(Equal(1)) - // Events don't always come back in the same order as they were sent, so we need to sort them and - // do the same for the expected IDs as well. - expectedIDs := []string{ids[0], ids[2]} - sort.Strings(expectedIDs) - var gotIDs []string - for _, ev := range resp.Events { - gotIDs = append(gotIDs, ev.Id) - } - sort.Strings(gotIDs) - - Expect(len(gotIDs)).To(Equal(len(expectedIDs))) - Expect(gotIDs[0]).To(Equal(expectedIDs[0])) - Expect(gotIDs[1]).To(Equal(expectedIDs[1])) + var stored service.StoreEvent + err = json.Unmarshal(records[0].Value, &stored) + Expect(err).ToNot(HaveOccurred()) + Expect(stored.ID).To(Equal(id)) + }) }) It("Starts without consumer and reads events from shared store", func() { - ehNilConsumer, err := service.NewEventHistoryService(cfg, nil, sto, log.Logger{}) + // A separate consumer instance stores events into the store shared with the + // consumer-less instance (gRPC-only mode), which can still find them. + bus = testBus(make(chan events.Event)) + defer close(bus) + evConsumer, err := consumer.NewConsumer( + consumer.Logger(log.Logger{}), + consumer.Config(cfg), + consumer.Persistence(sto), + consumer.Stream(bus), + ) Expect(err).ToNot(HaveOccurred()) - Expect(ehNilConsumer).ToNot(BeNil()) + go evConsumer.StoreEvents() - // We assume that the event was consumed and stored by the original instance, so the nil-consumer - // instance (gRPC-only mode) can still find it in the shared store. id := bus.Publish(events.UploadReady{}) time.Sleep(500 * time.Millisecond) resp := &ehsvc.GetEventsResponse{} - err = ehNilConsumer.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp) + err = eh.GetEvents(context.Background(), &ehsvc.GetEventsRequest{Ids: []string{id}}, resp) Expect(err).ToNot(HaveOccurred()) Expect(len(resp.Events)).To(Equal(1)) Expect(resp.Events[0].Id).To(Equal(id)) }) - - It("Stores events and verifies them in the store directly (consumer-only mode)", func() { - id := bus.Publish(events.UploadReady{}) - time.Sleep(500 * time.Millisecond) - - records, err := sto.Read(id) - Expect(err).ToNot(HaveOccurred()) - Expect(len(records)).To(Equal(1)) - - var stored service.StoreEvent - err = json.Unmarshal(records[0].Value, &stored) - Expect(err).ToNot(HaveOccurred()) - Expect(stored.ID).To(Equal(id)) - }) }) type testBus chan events.Event