dotnet receiver: convert stream data to metrics (#2608)

This is the fifth PR in a series for the dotnet core diagnostics
receiver.

This PR converts the raw metric data extracted from the dotnet stream
into Otel metrics and sends them down the pipeline.

For the full implementation please see #2200

receiver/dotnetdiagnosticsreceiver/config.go

@@ -36,4 +36,16 @@ type Config struct {
 	// be displayed by the `dotnet-counters` tool:
 	// https://docs.microsoft.com/en-us/dotnet/core/diagnostics/dotnet-counters
 	Counters []string `mapstructure:"counters"`
+
+	// LocalDebugDir takes an optional directory name where stream data can be written for
+	// offline analysis and troubleshooting. If LocalDebugDir is empty, no stream data is
+	// written. If it has a value, MaxLocalDebugFiles also needs to be set, and stream
+	// data will be written to disk at the specified location using the naming
+	// convention `msg.%d.bin` as each message is received, where %d is the current
+	// message number.
+	LocalDebugDir string `mapstructure:"local_debug_dir"`
+	// MaxLocalDebugFiles indicates the maximum number of files kept in LocalDebugDir. When a
+	// file is written, the oldest one will be deleted if necessary to keep the
+	// number of files in LocalDebugDir at the specified maximum.
+	MaxLocalDebugFiles int `mapstructure:"max_local_debug_files"`
 }

receiver/dotnetdiagnosticsreceiver/dotnet/event_header_test.go

@@ -24,7 +24,9 @@ import (
 
 func TestEventHeader(t *testing.T) {
 	h := eventHeader{}
-	err := parseEventHeader(network.NewMultiReader(network.NewDefaultFakeRW("", "", "")), &h)
+	rw := network.NewDefaultFakeRW("", "", "")
+	mr := network.NewMultiReader(rw, &network.NopBlobWriter{})
+	err := parseEventHeader(mr, &h)
 	require.NoError(t, err)
 }
 
@@ -49,6 +51,7 @@ func testEventHeaderError(t *testing.T, flags headerFlags, errPos int) {
 			0: {byte(flags)},
 		},
 	}
-	err := parseEventHeader(network.NewMultiReader(pr), &eventHeader{})
+	mr := network.NewMultiReader(pr, &network.NopBlobWriter{})
+	err := parseEventHeader(mr, &eventHeader{})
 	require.Error(t, err)
 }

receiver/dotnetdiagnosticsreceiver/dotnet/event_parser.go

@@ -71,7 +71,9 @@ func parseEventBlock(r network.MultiReader, fm fieldMetadataMap) (metrics []Metr
 		if err != nil {
 			return
 		}
-		metrics = append(metrics, m)
+		if len(m) > 0 {
+			metrics = append(metrics, m)
+		}
 	}
 
 	return

receiver/dotnetdiagnosticsreceiver/dotnet/event_parser_test.go

@@ -28,7 +28,7 @@ func TestEventParser(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 4)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(1131)
 	require.NoError(t, err)
 	metrics, err := parseEventBlock(reader, fms())
@@ -73,9 +73,9 @@ func TestEventParserErrors(t *testing.T) {
 
 func testEventParserError(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(1131)
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 	require.NoError(t, err)
 	_, err = parseEventBlock(reader, fms())
 	require.Error(t, err)

receiver/dotnetdiagnosticsreceiver/dotnet/ipc_parser_test.go

@@ -31,7 +31,7 @@ func TestIPCParser_NoErrors(t *testing.T) {
 			0: []byte(magicTerminated),
 		},
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseIPC(r)
 	require.NoError(t, err)
 }
@@ -43,7 +43,7 @@ func TestIPCParser_BadMagic(t *testing.T) {
 			0: []byte("DOTNET_IPC_V2"),
 		},
 	}
-	err := parseIPC(network.NewMultiReader(rw))
+	err := parseIPC(network.NewMultiReader(rw, &network.NopBlobWriter{}))
 	require.EqualError(t, err, `ipc header: expected magic "DOTNET_IPC_V1" got "DOTNET_IPC_V2"`)
 }
 
@@ -55,7 +55,7 @@ func TestIPCParser_BadResponseID(t *testing.T) {
 			3: {responseError},
 		},
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseIPC(r)
 	assert.EqualError(t, err, "ipc header: got error response")
 }
@@ -70,7 +70,7 @@ func testIPCError(t *testing.T, idx int) {
 	rw := &network.FakeRW{
 		ReadErrIdx: idx,
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseIPC(r)
 	require.EqualError(t, err, fmt.Sprintf("deliberate error on read %d", idx))
 }

receiver/dotnetdiagnosticsreceiver/dotnet/metadata_parser_test.go

@@ -28,7 +28,7 @@ func TestParseMetadata(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 4)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(159)
 	require.NoError(t, err)
 	msgs := fieldMetadataMap{}
@@ -62,11 +62,11 @@ func TestParseMetadataErrors(t *testing.T) {
 
 func testParseMetadataError(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(159)
 	require.NoError(t, err)
 
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 
 	msgs := fieldMetadataMap{}
 	err = parseMetadataBlock(reader, msgs)

receiver/dotnetdiagnosticsreceiver/dotnet/nettrace_parser_test.go

@@ -31,7 +31,7 @@ func TestParseNettrace(t *testing.T) {
 			2: []byte(nettraceSerialization),
 		},
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseNettrace(r)
 	require.NoError(t, err)
 }
@@ -43,7 +43,7 @@ func TestParseNettrace_BadHeaderName(t *testing.T) {
 			0: []byte("nettrace"),
 		},
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseNettrace(r)
 	require.EqualError(t, err, `header name: expected "Nettrace" got "nettrace"`)
 }
@@ -58,7 +58,7 @@ func TestParseNettrace_BadSerializationName(t *testing.T) {
 			2: []byte(serType),
 		},
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := parseNettrace(r)
 	require.EqualError(t, err, `serialization type: expected "!FastSerialization.1" got "foo"`)
 }
@@ -74,6 +74,6 @@ func testParseNettraceReadErr(i int) error {
 	rw := &network.FakeRW{
 		ReadErrIdx: i,
 	}
-	r := network.NewMultiReader(rw)
+	r := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	return parseNettrace(r)
 }

receiver/dotnetdiagnosticsreceiver/dotnet/parser.go

@@ -26,15 +26,20 @@ import (
 
 // Parser encapsulates all of the functionality to parse an IPC stream.
 type Parser struct {
-	r      network.MultiReader
-	logger *zap.Logger
+	r       network.MultiReader
+	consume func([]Metric)
+	logger  *zap.Logger
 }
 
-// NewParser accepts an io.Reader and logger returns a Parser for processing an
-// IPC stream.
-func NewParser(rdr io.Reader, logger *zap.Logger) *Parser {
-	r := network.NewMultiReader(rdr)
-	return &Parser{r: r, logger: logger}
+// MetricsConsumer is a function that accepts a slice of Metrics. Parser has a
+// member consumer function, used to send Metrics as they are created.
+type MetricsConsumer func([]Metric)
+
+// NewParser accepts an io.Reader, a MetricsConsumer, and logger, and returns a
+// Parser for processing an IPC stream.
+func NewParser(rdr io.Reader, mc MetricsConsumer, bw network.BlobWriter, logger *zap.Logger) *Parser {
+	r := network.NewMultiReader(rdr, bw)
+	return &Parser{r: r, consume: mc, logger: logger}
 }
 
 // ParseIPC parses the IPC response from the initial request to a dotnet process.
@@ -58,6 +63,8 @@ func (p *Parser) ParseAll(ctx context.Context) error {
 			return nil
 		default:
 			err = p.parseBlock(fms)
+			// flush regardless of error
+			p.r.Flush()
 			if err != nil {
 				return err
 			}
@@ -97,10 +104,12 @@ func (p *Parser) parseBlock(fms fieldMetadataMap) error {
 			return err
 		}
 	case "EventBlock":
-		_, err = parseEventBlock(p.r, fms)
+		var metrics []Metric
+		metrics, err = parseEventBlock(p.r, fms)
 		if err != nil {
 			return err
 		}
+		p.consume(metrics)
 	case "SPBlock":
 		err = parseSPBlock(p.r)
 		if err != nil {

receiver/dotnetdiagnosticsreceiver/dotnet/parser_test.go

@@ -15,6 +15,7 @@
 package dotnet
 
 import (
+	"context"
 	"path"
 	"testing"
 
@@ -36,7 +37,12 @@ func TestParser(t *testing.T) {
 			8: []byte(fastSerialization),
 		},
 	}
-	p := NewParser(rw, zap.NewNop())
+	p := NewParser(
+		rw,
+		func(metrics []Metric) {},
+		&network.NopBlobWriter{},
+		zap.NewNop(),
+	)
 	err := p.ParseIPC()
 	require.NoError(t, err)
 	err = p.ParseNettrace()
@@ -47,7 +53,7 @@ func TestParser_TestData(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 16)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	parser := NewParser(rw, zap.NewNop())
+	parser := NewParser(rw, func([]Metric) {}, &network.NopBlobWriter{}, zap.NewNop())
 	err = parser.ParseIPC()
 	require.NoError(t, err)
 	err = parser.ParseNettrace()
@@ -85,10 +91,10 @@ func testParseBlockError(t *testing.T, data [][]byte, offset, errIdx int) {
 
 func testParseBlock(t *testing.T, data [][]byte, offset, errIdx int) error {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(offset)
 	require.NoError(t, err)
-	rw.SetReadError(errIdx)
+	rw.ErrOnRead(errIdx)
 	msgs := fieldMetadataMap{}
 	parser := &Parser{r: reader, logger: zap.NewNop()}
 	for i := 0; i < 16; i++ {
@@ -99,3 +105,46 @@ func testParseBlock(t *testing.T, data [][]byte, offset, errIdx int) error {
 	}
 	return nil
 }
+
+func TestParser_ParseAll_Error(t *testing.T) {
+	data, err := network.ReadBlobData(path.Join("..", "testdata"), 16)
+	require.NoError(t, err)
+	rw := network.NewBlobReader(data)
+	parser := NewParser(rw, func([]Metric) {}, &network.NopBlobWriter{}, zap.NewNop())
+	err = parser.ParseIPC()
+	require.NoError(t, err)
+	err = parser.ParseNettrace()
+	require.NoError(t, err)
+	rw.StopOnRead(0)
+	errCh := make(chan error)
+	go func() {
+		err = parser.ParseAll(context.Background())
+		errCh <- err
+	}()
+	<-rw.Gate()
+	rw.ErrOnRead(0)
+	rw.Gate() <- struct{}{}
+	require.Error(t, <-errCh)
+}
+
+func TestParser_ParseAll_Cancel(t *testing.T) {
+	data, err := network.ReadBlobData(path.Join("..", "testdata"), 16)
+	require.NoError(t, err)
+	rw := network.NewBlobReader(data)
+	parser := NewParser(rw, func([]Metric) {}, &network.NopBlobWriter{}, zap.NewNop())
+	err = parser.ParseIPC()
+	require.NoError(t, err)
+	err = parser.ParseNettrace()
+	require.NoError(t, err)
+	rw.StopOnRead(0)
+	errCh := make(chan error)
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		err = parser.ParseAll(ctx)
+		errCh <- err
+	}()
+	<-rw.Gate()
+	cancel()
+	rw.Gate() <- struct{}{}
+	require.NoError(t, <-errCh)
+}

receiver/dotnetdiagnosticsreceiver/dotnet/sequence_parser_test.go

@@ -27,7 +27,7 @@ func TestParseSPBlock(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 16)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(36870)
 	require.NoError(t, err)
 	err = parseSPBlock(reader)
@@ -44,10 +44,10 @@ func TestParseSPBlock_Errors(t *testing.T) {
 
 func testParseSPBlockError(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(36870)
 	require.NoError(t, err)
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 	err = parseSPBlock(reader)
 	require.Error(t, err)
 }

receiver/dotnetdiagnosticsreceiver/dotnet/serialization_type_test.go

@@ -28,7 +28,7 @@ func TestParseSerializationType(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 1)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(61)
 	require.NoError(t, err)
 	st, err := parseSerializationType(reader)
@@ -48,11 +48,11 @@ func TestParseSerializationType_Error(t *testing.T) {
 
 func testParseSerializationTypeErr(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	// metadata block is 61 bytes in
 	err := reader.Seek(61)
 	require.NoError(t, err)
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 	_, err = parseSerializationType(reader)
 	require.Error(t, err)
 }

receiver/dotnetdiagnosticsreceiver/dotnet/stack_parser_test.go

@@ -27,7 +27,7 @@ func TestParseStackBlock(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 4)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(972)
 	require.NoError(t, err)
 	err = parseStackBlock(reader)
@@ -44,10 +44,10 @@ func TestParseStackBlockErrors(t *testing.T) {
 
 func testParseStackBlockError(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(972)
 	require.NoError(t, err)
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 	err = parseStackBlock(reader)
 	require.Error(t, err)
 }

receiver/dotnetdiagnosticsreceiver/dotnet/trace_parser_test.go

@@ -27,7 +27,7 @@ func TestTraceParser(t *testing.T) {
 	data, err := network.ReadBlobData(path.Join("..", "testdata"), 2)
 	require.NoError(t, err)
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err = reader.Seek(81)
 	require.NoError(t, err)
 	err = parseTraceMessage(reader)
@@ -45,10 +45,10 @@ func TestTraceParser_Errors(t *testing.T) {
 
 func testTraceParserReadError(t *testing.T, data [][]byte, i int) {
 	rw := network.NewBlobReader(data)
-	reader := network.NewMultiReader(rw)
+	reader := network.NewMultiReader(rw, &network.NopBlobWriter{})
 	err := reader.Seek(81)
 	require.NoError(t, err)
-	rw.SetReadError(i)
+	rw.ErrOnRead(i)
 	err = parseTraceMessage(reader)
 	require.Error(t, err)
 }