TBS: perf: Pool APMEvent in sampling decision handler (#13539)

Pool APMEvent struct in ReadTraceEvents for ~20% perf gain.

x-pack/apm-server/sampling/eventstorage/storage.go

@@ -300,16 +300,16 @@ func (rw *ReadWriter) ReadTraceEvents(traceID string, out *modelpb.Batch) error
 		}
 		switch item.UserMeta() {
 		case entryMetaTraceEvent:
-			var event modelpb.APMEvent
+			event := modelpb.APMEventFromVTPool()
 			if err := item.Value(func(data []byte) error {
-				if err := rw.s.codec.DecodeEvent(data, &event); err != nil {
+				if err := rw.s.codec.DecodeEvent(data, event); err != nil {
 					return fmt.Errorf("codec failed to decode event: %w", err)
 				}
 				return nil
 			}); err != nil {
 				return err
 			}
-			*out = append(*out, &event)
+			*out = append(*out, event)
 		default:
 			// Unknown entry meta: ignore.
 			continue

x-pack/apm-server/sampling/processor.go

@@ -473,6 +473,7 @@ func (p *Processor) Run() error {
 		}
 	})
 	g.Go(func() error {
+		var events modelpb.Batch
 		// TODO(axw) pace the publishing over the flush interval?
 		// Alternatively we can rely on backpressure from the reporter,
 		// removing the artificial one second timeout from publisher code
@@ -510,7 +511,8 @@ func (p *Processor) Run() error {
 					"received error writing sampled trace: %s", err,
 				)
 			}
-			var events modelpb.Batch
+
+			events = events[:0]
 			if err := p.eventStore.ReadTraceEvents(traceID, &events); err != nil {
 				p.rateLimitedLogger.Warnf(
 					"received error reading trace events: %s", err,
@@ -543,6 +545,11 @@ func (p *Processor) Run() error {
 				if err := p.config.BatchProcessor.ProcessBatch(gracefulContext, &events); err != nil {
 					p.logger.With(logp.Error(err)).Warn("failed to report events")
 				}
+
+				for i := range events {
+					events[i].ReturnToVTPool()
+					events[i] = nil // not required but ensure that there is no ref to the freed event
+				}
 			}
 		}
 	})

x-pack/apm-server/sampling/processor_test.go

@@ -432,7 +432,7 @@ func TestProcessRemoteTailSampling(t *testing.T) {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case reported <- *batch:
+		case reported <- batch.Clone():
 			return nil
 		}
 	})