fix(logger): keep reloaded webhook and audit targets alive after a peer reload

A runtime change to audit_webhook, logger_webhook or audit_kafka is
applied on every peer by serviceReloadDynamic, under a 30-second context
that is cancelled as soon as the handler returns. applyDynamicConfigForSubSys
passed that context to the logger target updates, and the targets tie
their queue worker and, with queue_dir, their disk replay to the Init
context. On every server except the one that served the admin request
the new targets therefore stopped right after they were created: records
filled the in-memory queue and were then dropped with "log buffer full"
until the server restarted.

- Build logger webhook, audit webhook and audit Kafka targets with
  GlobalContext, as startup and the serving node already do. The
  caller's context is still used for logging.
- Open the HTTP disk queue store before starting the queue worker, and
  cancel the derived context when it cannot be opened. A worker started
  before a failed Open dereferenced the nil store on the first record and
  crashed the server; with the process context that would have happened
  on every server.
- Let Cancel hand records still waiting in the channel to the shared
  buffer of the replacement target. A target whose queue store never
  opened has no worker to do this, so repairing queue_dir and applying
  the configuration again used to discard the records it had accepted.
- TestDynamicWebhookOutlivesConfigContext applies each case through
  applyDynamicConfigForSubSys under a short context, cancels it as the
  peer handler does, and requires delivery: audit and logger webhooks
  with a memory queue, a disk queue and an unopenable disk queue
  (recovered by applying the configuration again), and audit Kafka with
  and without queue_dir against a sarama mock broker. Each case runs in
  a child process without logger and audit environment variables.

This completes earlier uncommitted maintainer work on this issue, which
covered the two HTTP call sites, the store-before-worker order and the
first version of the test.

Refs #259

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Feng Ruohang <rh@vonng.com>

internal/logger/target/http/http.go

@@ -196,8 +196,6 @@ func (h *Target) Init(ctx context.Context) (err error) {
 func (h *Target) initDiskStore(ctx context.Context) (err error) {
 	ctx, cancel := context.WithCancel(ctx)
 	h.storeCtxCancel = cancel
-	h.lastStarted = time.Now()
-	go h.startQueueProcessor(ctx, true)
 
 	queueStore := store.NewQueueStore[any](
 		filepath.Join(h.config.QueueDir, h.Name()),
@@ -206,10 +204,14 @@ func (h *Target) initDiskStore(ctx context.Context) (err error) {
 	)
 
 	if err := queueStore.Open(); err != nil {
+		cancel()
 		return fmt.Errorf("unable to initialize the queue store of %s webhook: %w", h.Name(), err)
 	}
 
 	h.store = queueStore
+	// The writer may immediately consume entries migrated from an old target.
+	h.lastStarted = time.Now()
+	go h.startQueueProcessor(ctx, true)
 	store.StreamItems(h.store, h, ctx.Done(), h.config.LogOnceIf)
 
 	return nil
@@ -627,6 +629,25 @@ func (h *Target) Cancel() {
 	// and finish the existing ones.
 	// All future ones will be discarded.
 	h.logChMu.Lock()
+	// Hand entries that no worker drained to the replacement, for example
+	// when the queue store never opened and no worker was started.
+	logChLock.Lock()
+	if migrateBuffer, ok := logChBuffers[h.Name()]; ok {
+	drain:
+		for {
+			select {
+			case v := <-h.logCh:
+				select {
+				case migrateBuffer <- v:
+				default:
+					break drain
+				}
+			default:
+				break drain
+			}
+		}
+	}
+	logChLock.Unlock()
 	xioutil.SafeClose(h.logCh)
 	h.logCh = nil
 	h.logChMu.Unlock()