tail sampling: Change monitoring metrics

Introduce sampled, head_unsampled and unprocessed.
Adapt logic for dropped to only contain trace events.

implements elastic#6269

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

@@ -56,9 +56,12 @@ type Processor struct {
 }
 
 type eventMetrics struct {
-	processed int64
-	dropped   int64
-	stored    int64
+	processed      int64
+	dropped        int64
+	stored         int64
+	sampled        int64
+	head_unsampled int64
+	unprocessed    int64
 }
 
 // NewProcessor returns a new Processor, for tail-sampling trace events.
@@ -117,6 +120,9 @@ func (p *Processor) CollectMonitoring(_ monitoring.Mode, V monitoring.Visitor) {
 		monitoring.ReportInt(V, "processed", atomic.LoadInt64(&p.eventMetrics.processed))
 		monitoring.ReportInt(V, "dropped", atomic.LoadInt64(&p.eventMetrics.dropped))
 		monitoring.ReportInt(V, "stored", atomic.LoadInt64(&p.eventMetrics.stored))
+		monitoring.ReportInt(V, "sampled", atomic.LoadInt64(&p.eventMetrics.sampled))
+		monitoring.ReportInt(V, "head_unsampled", atomic.LoadInt64(&p.eventMetrics.head_unsampled))
+		monitoring.ReportInt(V, "unprocessed", atomic.LoadInt64(&p.eventMetrics.unprocessed))
 	})
 }
 
@@ -149,13 +155,17 @@ func (p *Processor) ProcessBatch(ctx context.Context, batch *model.Batch) error
 			if err != nil {
 				return err
 			}
+			p.updateProcessorMetrics(report, stored)
 		case model.SpanProcessor:
 			var err error
 			atomic.AddInt64(&p.eventMetrics.processed, 1)
 			report, stored, err = p.processSpan(event)
 			if err != nil {
 				return err
 			}
+			p.updateProcessorMetrics(report, stored)
+		default:
+			atomic.AddInt64(&p.eventMetrics.unprocessed, 1)
 		}
 		if !report {
 			// We shouldn't report this event, so remove it from the slice.
@@ -164,7 +174,6 @@ func (p *Processor) ProcessBatch(ctx context.Context, batch *model.Batch) error
 			events = events[:n-1]
 			i--
 		}
-		p.updateProcessorMetrics(report, stored)
 	}
 	*batch = events
 	return nil
@@ -191,6 +200,7 @@ func (p *Processor) processTransaction(event *model.APMEvent) (report, stored bo
 	if !event.Transaction.Sampled {
 		// (Head-based) unsampled transactions are passed through
 		// by the tail sampler.
+		atomic.AddInt64(&p.eventMetrics.head_unsampled, 1)
 		return true, false, nil
 	}
 
@@ -200,6 +210,9 @@ func (p *Processor) processTransaction(event *model.APMEvent) (report, stored bo
 		// Tail-sampling decision has been made: report the transaction
 		// if it was sampled.
 		report := traceSampled
+		if report {
+			atomic.AddInt64(&p.eventMetrics.sampled, 1)
+		}
 		return report, false, nil
 	case eventstorage.ErrNotFound:
 		// Tail-sampling decision has not yet been made.
@@ -263,6 +276,7 @@ func (p *Processor) processSpan(event *model.APMEvent) (report, stored bool, _ e
 	if !traceSampled {
 		return false, false, nil
 	}
+	atomic.AddInt64(&p.eventMetrics.sampled, 1)
 	return true, false, nil
 }
 
@@ -460,6 +474,7 @@ func (p *Processor) Run() error {
 						}
 					}
 				}
+				atomic.AddInt64(&p.eventMetrics.processed, int64(len(events)))
 				if err := p.config.BatchProcessor.ProcessBatch(ctx, &events); err != nil {
 					p.logger.With(logp.Error(err)).Warn("failed to report events")
 				}