Expand Result to Include Signal Types

input/elasticapm/processor.go

@@ -296,19 +296,25 @@ func (p *Processor) handleStream(
 	if n == 0 {
 		return readErr
 	}
-
-	if err := p.processBatch(ctx, processor, batch); err != nil {
+	if err := processor.ProcessBatch(ctx, batch); err != nil {
 		return fmt.Errorf("cannot process batch: %w", err)
 	}
+	for _, v := range *batch {
+		switch v.Type() {
+		case modelpb.SpanEventType:
+			result.SpanAccepted++
+		case modelpb.TransactionEventType:
+			result.TransactionAccepted++
+		case modelpb.MetricEventType:
+			result.MetricAccepted++
+		case modelpb.LogEventType:
+			result.LogAccepted++
+		}
+	}
 	result.Accepted += n
 	return readErr
 }
 
-// processBatch processes the batch and returns the events to the pool after it's been processed.
-func (p *Processor) processBatch(ctx context.Context, processor modelpb.BatchProcessor, batch *modelpb.Batch) error {
-	return processor.ProcessBatch(ctx, batch)
-}
-
 // getStreamReader returns a streamReader that reads ND-JSON lines from r.
 func (p *Processor) getStreamReader(r io.Reader) *streamReader {
 	return &streamReader{

input/elasticapm/processor_test.go

@@ -35,6 +35,7 @@ import (
 
 func TestHandleStreamReaderError(t *testing.T) {
 	readErr := errors.New("read failed")
+	cnt := 5
 	var calls int
 	var reader readerFunc = func(p []byte) (int, error) {
 		calls++
@@ -43,7 +44,7 @@ func TestHandleStreamReaderError(t *testing.T) {
 		}
 		buf := bytes.NewBuffer(nil)
 		buf.WriteString(validMetadata + "\n")
-		for i := 0; i < 5; i++ {
+		for i := 0; i < cnt; i++ {
 			buf.WriteString(validTransaction + "\n")
 		}
 		return copy(p, buf.Bytes()), nil
@@ -60,7 +61,10 @@ func TestHandleStreamReaderError(t *testing.T) {
 		reader, 10, nopBatchProcessor{}, &actualResult,
 	)
 	assert.ErrorIs(t, err, readErr)
-	assert.Equal(t, Result{Accepted: 5}, actualResult)
+	assert.Equal(t, Result{
+		Accepted:            cnt,
+		TransactionAccepted: cnt,
+	}, actualResult)
 }
 
 type readerFunc func([]byte) (int, error)

input/elasticapm/result.go

@@ -23,18 +23,34 @@ import (
 
 type Result struct {
 	errorsSpace [5]error
+
 	// Errors holds a limited number of errors that occurred while
 	// processing the event stream. If the limit is reached, the
 	// counters above are still incremented.
 	Errors []error
+
 	// Accepted holds the number of valid events accepted.
-	Accepted int
+	Accepted            int
+	SpanAccepted        int
+	TransactionAccepted int
+	MetricAccepted      int
+	LogAccepted         int
+
 	// TooLarge holds the number of events that were rejected due
 	// to exceeding the event size limit.
-	TooLarge int
+	TooLarge            int
+	SpanTooLarge        int
+	TransactionTooLarge int
+	MetricTooLarge      int
+	LogTooLarge         int
+
 	// Invalid holds the number of events that were rejected due
 	// to being invalid, excluding those that are counted by TooLarge.
-	Invalid int
+	Invalid            int
+	SpanInvalid        int
+	TransactionInvalid int
+	MetricInvalid      int
+	LogInvalid         int
 }
 
 func (r *Result) addError(err error) {