Merge pull request #31 from emilien-puget/fix-panic-writer

fix a writer panic

io/writer.go

@@ -7,6 +7,7 @@ import (
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/metric"
 	"go.opentelemetry.io/otel/trace"
+	nooptrace "go.opentelemetry.io/otel/trace/noop"
 )
 
 var _ io.WriteCloser = (*instrumentedWriter)(nil)
@@ -41,6 +42,7 @@ func NewInstrumentedWriterFactory(prefix string, attrT []attribute.KeyValue, att
 			writer: w,
 			track: ioTracking{
 				instr: instr,
+				span:  nooptrace.Span{},
 				ctx:   ctx,
 			},
 		}

io/writer_test.go

@@ -143,3 +143,114 @@ func TestWriterHappyPath(t *testing.T) {
 		return
 	}
 }
+
+func TestWriter_nowrite(t *testing.T) {
+	var err error
+
+	spanRecorder := sdktracetest.NewSpanRecorder()
+	tracerProvider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spanRecorder))
+	tracer := tracerProvider.Tracer("io-writer-test-tracer")
+	ctx, _ := tracer.Start(context.Background(), "io-writer-test-body-tracker")
+
+	metricReader := sdkmetric.NewManualReader()
+	metricProvider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(metricReader))
+	meter := metricProvider.Meter("io-writer-test-meter")
+
+	// this is the buffer where we want to write something:
+	var bw bytes.Buffer
+	bw.Grow(64 * 1024)
+
+	initialStartedSpansLen := len(spanRecorder.Started())
+
+	attrsT := []attribute.KeyValue{
+		attribute.Int("http.status_code", 201),
+		attribute.String("url.path.pattern", "/this/{matched}/path"),
+		attribute.String("url.path", "/this/some_value/path"),
+	}
+	attrsM := []attribute.KeyValue{
+		attribute.Int("http.status_code", 201),
+		attribute.String("url.path", "/this/{matched}/path"),
+	}
+	writer := NewInstrumentedWriter("", &bw, ctx, attrsT, attrsM, tracer, meter)
+
+	startedSpans := spanRecorder.Started()
+	if len(startedSpans) > initialStartedSpansLen {
+		t.Errorf("span cannot start until first write: %#v", startedSpans)
+		return
+	}
+
+	endedSpans := spanRecorder.Ended()
+	if len(endedSpans) > 0 {
+		t.Errorf("the writing does not end a span until its closed")
+		return
+	}
+
+	writer.Close()
+	endedSpans = spanRecorder.Ended()
+	if len(endedSpans) != 0 {
+		t.Errorf("num ended spans, want: 0, got: %d", len(endedSpans))
+		for idx, s := range endedSpans {
+			t.Errorf("%d -> %#v", idx, s)
+		}
+		return
+	}
+
+	// check the reported metrics
+	rm := metricdata.ResourceMetrics{}
+	err = metricReader.Collect(context.Background(), &rm)
+	if err != nil {
+		t.Errorf("collecting metrics err: %s", err.Error())
+		return
+	}
+
+	if len(rm.ScopeMetrics) != 1 {
+		t.Errorf("wrong amount of metrics, want: 1, got: %d", len(rm.ScopeMetrics))
+		for idx, sm := range rm.ScopeMetrics {
+			t.Errorf("%d -> %#v", idx, sm)
+		}
+		return
+	}
+
+	// --> check that we have all the metrics we want to report
+	sm := rm.ScopeMetrics[0]
+	wantedMetrics := map[string]bool{
+		"written.size":      false,
+		"written.size-hist": false,
+		"written.time":      false,
+		"written.time-hist": false,
+		// "written-errors":    false,
+	}
+	numWantedMetrics := len(wantedMetrics)
+	gotMetrics := make(map[string]metricdata.Metrics, numWantedMetrics)
+	for _, m := range sm.Metrics {
+		gotMetrics[m.Name] = m
+	}
+	for k := range wantedMetrics {
+		if _, ok := gotMetrics[k]; !ok {
+			t.Errorf("missing metric %s", k)
+			return
+		}
+	}
+
+	// --> check that the metrics have the expected attributes set
+	writtenSize := gotMetrics["written.size"]
+	writtenSizeSum, ok := writtenSize.Data.(metricdata.Sum[int64])
+	if !ok {
+		t.Errorf("cannot access written size aggregation: %#v", writtenSize.Data)
+		return
+	}
+	if len(writtenSizeSum.DataPoints) != 1 {
+		t.Errorf("written sum data points, want: 1, got: %d", len(writtenSizeSum.DataPoints))
+		return
+	}
+	dp := writtenSizeSum.DataPoints[0]
+
+	if dp.Attributes.Len() != 2 {
+		t.Errorf("missing attributes")
+		return
+	}
+	if dp.Value != 0 {
+		t.Errorf("metric size, want: %d, got: %d", 0, dp.Value)
+		return
+	}
+}