Streamline config, fix tests
This commit is contained in:
@@ -34,6 +34,7 @@ type Config struct {
|
|||||||
Context context.Context `yaml:"-"`
|
Context context.Context `yaml:"-"`
|
||||||
|
|
||||||
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:"%%NEXT%%"`
|
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:"%%NEXT%%"`
|
||||||
|
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:"%%NEXT%%"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// Events combines the configuration options for the event bus.
|
// Events combines the configuration options for the event bus.
|
||||||
|
|||||||
@@ -53,6 +53,7 @@ func DefaultConfig() *config.Config {
|
|||||||
},
|
},
|
||||||
},
|
},
|
||||||
WriteBufferDuration: 10 * time.Second,
|
WriteBufferDuration: 10 * time.Second,
|
||||||
|
MaxActivities: 6000,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -88,7 +88,6 @@ func Server(opts ...Option) (http.Service, error) {
|
|||||||
svc.HistoryClient(options.HistoryClient),
|
svc.HistoryClient(options.HistoryClient),
|
||||||
svc.ValueClient(options.ValueClient),
|
svc.ValueClient(options.ValueClient),
|
||||||
svc.RegisteredEvents(options.RegisteredEvents),
|
svc.RegisteredEvents(options.RegisteredEvents),
|
||||||
svc.WriteBufferDuration(options.Config.WriteBufferDuration),
|
|
||||||
)
|
)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return http.Service{}, err
|
return http.Service{}, err
|
||||||
|
|||||||
@@ -103,17 +103,3 @@ func ValueClient(vs settingssvc.ValueService) Option {
|
|||||||
o.ValueClient = vs
|
o.ValueClient = vs
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// WriteBufferDuration sets the write buffer duration
|
|
||||||
func WriteBufferDuration(d time.Duration) Option {
|
|
||||||
return func(o *Options) {
|
|
||||||
o.WriteBufferDuration = d
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// MaxActivities sets the maximum number of activities to store
|
|
||||||
func MaxActivities(max int) Option {
|
|
||||||
return func(o *Options) {
|
|
||||||
o.MaxActivities = max
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|||||||
@@ -206,10 +206,10 @@ func New(opts ...Option) (*ActivitylogService, error) {
|
|||||||
tp: o.TraceProvider,
|
tp: o.TraceProvider,
|
||||||
tracer: o.TraceProvider.Tracer("github.com/opencloud-eu/opencloud/services/activitylog/pkg/service"),
|
tracer: o.TraceProvider.Tracer("github.com/opencloud-eu/opencloud/services/activitylog/pkg/service"),
|
||||||
parentIdCache: cache,
|
parentIdCache: cache,
|
||||||
maxActivities: o.MaxActivities,
|
maxActivities: o.Config.MaxActivities,
|
||||||
natskv: kv,
|
natskv: kv,
|
||||||
}
|
}
|
||||||
s.debouncer = NewDebouncer(o.WriteBufferDuration, s.storeActivity)
|
s.debouncer = NewDebouncer(o.Config.WriteBufferDuration, s.storeActivity)
|
||||||
|
|
||||||
s.mux.Get("/graph/v1beta1/extensions/org.libregraph/activities", s.HandleGetItemActivities)
|
s.mux.Get("/graph/v1beta1/extensions/org.libregraph/activities", s.HandleGetItemActivities)
|
||||||
|
|
||||||
|
|||||||
@@ -66,7 +66,7 @@ var _ = SynchronizedAfterSuite(func() {
|
|||||||
var _ = Describe("ActivitylogService", func() {
|
var _ = Describe("ActivitylogService", func() {
|
||||||
var (
|
var (
|
||||||
alog *ActivitylogService
|
alog *ActivitylogService
|
||||||
getResource func(ref *provider.Reference) (*provider.ResourceInfo, error)
|
getResource func(_ context.Context, ref *provider.Reference) (*provider.ResourceInfo, error)
|
||||||
writebufferduration = 100 * time.Millisecond
|
writebufferduration = 100 * time.Millisecond
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -84,9 +84,9 @@ var _ = Describe("ActivitylogService", func() {
|
|||||||
Nodes: []string{server.Addr().String()},
|
Nodes: []string{server.Addr().String()},
|
||||||
Database: "activitylog-test-" + uuid.New().String(),
|
Database: "activitylog-test-" + uuid.New().String(),
|
||||||
},
|
},
|
||||||
|
MaxActivities: 4,
|
||||||
|
WriteBufferDuration: writebufferduration,
|
||||||
}),
|
}),
|
||||||
MaxActivities(4),
|
|
||||||
WriteBufferDuration(writebufferduration),
|
|
||||||
Stream(stream),
|
Stream(stream),
|
||||||
TraceProvider(noop.NewTracerProvider()),
|
TraceProvider(noop.NewTracerProvider()),
|
||||||
Mux(chi.NewMux()),
|
Mux(chi.NewMux()),
|
||||||
@@ -148,7 +148,7 @@ var _ = Describe("ActivitylogService", func() {
|
|||||||
tc := tc // capture range variable
|
tc := tc // capture range variable
|
||||||
Context(tc.Name, func() {
|
Context(tc.Name, func() {
|
||||||
JustBeforeEach(func() {
|
JustBeforeEach(func() {
|
||||||
getResource = func(ref *provider.Reference) (*provider.ResourceInfo, error) {
|
getResource = func(_ context.Context, ref *provider.Reference) (*provider.ResourceInfo, error) {
|
||||||
return tc.Tree[ref.GetResourceId().GetOpaqueId()], nil
|
return tc.Tree[ref.GetResourceId().GetOpaqueId()], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -179,9 +179,13 @@ var _ = Describe("ActivitylogService", func() {
|
|||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
|
BeforeEach(func() {
|
||||||
|
writebufferduration = 100 * time.Millisecond
|
||||||
|
})
|
||||||
|
|
||||||
Describe("addActivity", func() {
|
Describe("addActivity", func() {
|
||||||
var (
|
var (
|
||||||
getResource = func(ref *provider.Reference) (*provider.ResourceInfo, error) {
|
getResource = func(_ context.Context, ref *provider.Reference) (*provider.ResourceInfo, error) {
|
||||||
return tree[ref.GetResourceId().GetOpaqueId()], nil
|
return tree[ref.GetResourceId().GetOpaqueId()], nil
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
@@ -234,7 +238,7 @@ var _ = Describe("ActivitylogService", func() {
|
|||||||
|
|
||||||
Describe("Activities", func() {
|
Describe("Activities", func() {
|
||||||
It("combines multiple batches", func() {
|
It("combines multiple batches", func() {
|
||||||
getResource = func(ref *provider.Reference) (*provider.ResourceInfo, error) {
|
getResource = func(_ context.Context, ref *provider.Reference) (*provider.ResourceInfo, error) {
|
||||||
return tree[ref.GetResourceId().GetOpaqueId()], nil
|
return tree[ref.GetResourceId().GetOpaqueId()], nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user