fix: crash when audit webhook queue_dir is not writable (#19854)

This is regression introduced in #19275 refactor
This commit is contained in:
Harshavardhana 2024-06-01 20:03:39 -07:00 committed by GitHub
parent 2a75225569
commit ba54b39c02
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 15 additions and 23 deletions

View File

@ -193,7 +193,20 @@ func (h *Target) initDiskStore(ctx context.Context) (err error) {
h.storeCtxCancel = cancel
h.lastStarted = time.Now()
go h.startQueueProcessor(ctx, true)
queueStore := store.NewQueueStore[interface{}](
filepath.Join(h.config.QueueDir, h.Name()),
uint64(h.config.QueueSize),
httpLoggerExtension,
)
if err := queueStore.Open(); err != nil {
return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
}
h.store = queueStore
store.StreamItems(h.store, h, ctx.Done(), h.config.LogOnceIf)
return nil
}
@ -314,10 +327,7 @@ func (h *Target) startQueueProcessor(ctx context.Context, mainWorker bool) {
enc := jsoniter.ConfigCompatibleWithStandardLibrary.NewEncoder(buf)
defer bytebufferpool.Put(buf)
isDirQueue := false
if h.config.QueueDir != "" {
isDirQueue = true
}
isDirQueue := h.config.QueueDir != ""
// globalBuffer is always created or adjusted
// before this method is launched.
@ -504,23 +514,6 @@ func New(config Config) (*Target, error) {
}
h.client = &http.Client{Transport: h.config.Transport}
if h.config.QueueDir != "" {
queueStore := store.NewQueueStore[interface{}](
filepath.Join(h.config.QueueDir, h.Name()),
uint64(h.config.QueueSize),
httpLoggerExtension,
)
if err := queueStore.Open(); err != nil {
return h, fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
}
h.store = queueStore
}
return h, nil
}

View File

@ -169,9 +169,8 @@ func (h *Target) Init(ctx context.Context) error {
}
func (h *Target) initQueueStore(ctx context.Context) (err error) {
var queueStore store.Store[interface{}]
queueDir := filepath.Join(h.kconfig.QueueDir, h.Name())
queueStore = store.NewQueueStore[interface{}](queueDir, uint64(h.kconfig.QueueSize), kafkaLoggerExtension)
queueStore := store.NewQueueStore[interface{}](queueDir, uint64(h.kconfig.QueueSize), kafkaLoggerExtension)
if err = queueStore.Open(); err != nil {
return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
}