Merge branch 'main' into rmconsumerdata

Author: bogdandrutu

CHANGELOG.md

@@ -6,6 +6,7 @@
 
 - Remove `consumerdata.TraceData` (#2551)
 - Move `consumerdata.MetricsData` to `internaldata.MetricsData` (#2512)
+- Remove custom OpenCensus sematic conventions that have equivalent in otel (#2552)
 
 ## v0.21.0 Beta
 

component/componenttest/shutdown_verifier.go

@@ -0,0 +1,69 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package componenttest
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/config/configerror"
+	"go.opentelemetry.io/collector/config/configmodels"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/internal/testdata"
+)
+
+func verifyTraceProcessorDoesntProduceAfterShutdown(t *testing.T, factory component.ProcessorFactory, cfg configmodels.Processor) {
+	// Create a processor and output its produce to a sink.
+	nextSink := new(consumertest.TracesSink)
+	processor, err := factory.CreateTracesProcessor(
+		context.Background(),
+		component.ProcessorCreateParams{Logger: zap.NewNop()},
+		cfg,
+		nextSink,
+	)
+	if err != nil {
+		if err == configerror.ErrDataTypeIsNotSupported {
+			return
+		}
+		require.NoError(t, err)
+	}
+	err = processor.Start(context.Background(), NewNopHost())
+	assert.NoError(t, err)
+
+	// Send some traces to the processor.
+	const generatedCount = 10
+	for i := 0; i < generatedCount; i++ {
+		processor.ConsumeTraces(context.Background(), testdata.GenerateTraceDataOneSpan())
+	}
+
+	// Now shutdown the processor.
+	err = processor.Shutdown(context.Background())
+	assert.NoError(t, err)
+
+	// The Shutdown() is done. It means the processor must have sent everything we
+	// gave it to the next sink.
+	assert.EqualValues(t, generatedCount, nextSink.SpansCount())
+}
+
+func VerifyProcessorShutdown(t *testing.T, factory component.ProcessorFactory, cfg configmodels.Processor) {
+	verifyTraceProcessorDoesntProduceAfterShutdown(t, factory, cfg)
+	// TODO: add metrics and logs verification.
+	// TODO: add other shutdown verifications.
+}

internal/goldendataset/span_generator.go

@@ -476,33 +476,36 @@ func calculateListSize(listCnt PICTInputSpanChild) int {
 
 func generateSpanEvent(index int) *otlptrace.Span_Event {
 	t := time.Now().Add(-75 * time.Microsecond)
+	name, attributes := generateEventNameAndAttributes(index)
 	return &otlptrace.Span_Event{
 		TimeUnixNano:           uint64(t.UnixNano()),
-		Name:                   "message",
-		Attributes:             generateEventAttributes(index),
+		Name:                   name,
+		Attributes:             attributes,
 		DroppedAttributesCount: 0,
 	}
 }
 
-func generateEventAttributes(index int) []otlpcommon.KeyValue {
-	if index%4 == 2 {
-		return nil
-	}
-	attrMap := make(map[string]interface{})
-	if index%2 == 0 {
-		attrMap[conventions.AttributeMessageType] = "SENT"
-	} else {
-		attrMap[conventions.AttributeMessageType] = "RECEIVED"
-	}
-	attrMap[conventions.AttributeMessageID] = int64(index)
-	attrMap[conventions.AttributeMessageCompressedSize] = int64(17 * index)
-	attrMap[conventions.AttributeMessageUncompressedSize] = int64(24 * index)
-	if index%4 == 1 {
-		attrMap["app.inretry"] = true
-		attrMap["app.progress"] = 0.6
-		attrMap["app.statemap"] = "14|5|202"
+func generateEventNameAndAttributes(index int) (string, []otlpcommon.KeyValue) {
+	switch index % 4 {
+	case 0, 3:
+		attrMap := make(map[string]interface{})
+		if index%2 == 0 {
+			attrMap[conventions.AttributeMessageType] = "SENT"
+		} else {
+			attrMap[conventions.AttributeMessageType] = "RECEIVED"
+		}
+		attrMap[conventions.AttributeMessageID] = int64(index / 4)
+		attrMap[conventions.AttributeMessageCompressedSize] = int64(17 * index)
+		attrMap[conventions.AttributeMessageUncompressedSize] = int64(24 * index)
+		return "message", convertMapToAttributeKeyValues(attrMap)
+	case 1:
+		return "custom", convertMapToAttributeKeyValues(map[string]interface{}{
+			"app.inretry":  true,
+			"app.progress": 0.6,
+			"app.statemap": "14|5|202"})
+	default:
+		return "annotation", nil
 	}
-	return convertMapToAttributeKeyValues(attrMap)
 }
 
 func generateSpanLink(random io.Reader, index int) *otlptrace.Span_Link {

internal/goldendataset/traces_generator.go

@@ -17,24 +17,25 @@ package goldendataset
 import (
 	"fmt"
 	"io"
+	"math/rand"
 
+	"go.opentelemetry.io/collector/consumer/pdata"
 	otlpcommon "go.opentelemetry.io/collector/internal/data/protogen/common/v1"
 	otlptrace "go.opentelemetry.io/collector/internal/data/protogen/trace/v1"
 )
 
-// GenerateResourceSpans generates a slice of OTLP ResourceSpans objects based on the PICT-generated pairwise
+// GenerateTraces generates a slice of OTLP ResourceSpans objects based on the PICT-generated pairwise
 // parameters defined in the parameters file specified by the tracePairsFile parameter. The pairs to generate
-// spans for for defined in the file specified by the spanPairsFile parameter. The random parameter injects the
-// random number generator to use in generating IDs and other random values.
+// spans for for defined in the file specified by the spanPairsFile parameter.
 // The slice of ResourceSpans are returned. If an err is returned, the slice elements will be nil.
-func GenerateResourceSpans(tracePairsFile string, spanPairsFile string,
-	random io.Reader) ([]*otlptrace.ResourceSpans, error) {
+func GenerateTraces(tracePairsFile string, spanPairsFile string) ([]pdata.Traces, error) {
+	random := io.Reader(rand.New(rand.NewSource(42)))
 	pairsData, err := loadPictOutputFile(tracePairsFile)
 	if err != nil {
 		return nil, err
 	}
 	pairsTotal := len(pairsData) - 1
-	spans := make([]*otlptrace.ResourceSpans, pairsTotal)
+	traces := make([]pdata.Traces, pairsTotal)
 	for index, values := range pairsData {
 		if index == 0 {
 			continue
@@ -44,22 +45,22 @@ func GenerateResourceSpans(tracePairsFile string, spanPairsFile string,
 			InstrumentationLibrary: PICTInputInstrumentationLibrary(values[TracesColumnInstrumentationLibrary]),
 			Spans:                  PICTInputSpans(values[TracesColumnSpans]),
 		}
-		rscSpan, spanErr := GenerateResourceSpan(tracingInputs, spanPairsFile, random)
+		rscSpan, spanErr := generateResourceSpan(tracingInputs, spanPairsFile, random)
 		if spanErr != nil {
 			err = spanErr
 		}
-		spans[index-1] = rscSpan
+		traces[index-1] = pdata.TracesFromOtlp([]*otlptrace.ResourceSpans{rscSpan})
 	}
-	return spans, err
+	return traces, err
 }
 
-// GenerateResourceSpan generates a single OTLP ResourceSpans populated based on the provided inputs. They are:
+// generateResourceSpan generates a single OTLP ResourceSpans populated based on the provided inputs. They are:
 //   tracingInputs - the pairwise combination of field value variations for this ResourceSpans
 //   spanPairsFile - the file with the PICT-generated parameter combinations to generate spans for
 //   random - the random number generator to use in generating ID values
 //
 // The generated resource spans. If err is not nil, some or all of the resource spans fields will be nil.
-func GenerateResourceSpan(tracingInputs *PICTTracingInputs, spanPairsFile string,
+func generateResourceSpan(tracingInputs *PICTTracingInputs, spanPairsFile string,
 	random io.Reader) (*otlptrace.ResourceSpans, error) {
 	libSpans, err := generateLibrarySpansArray(tracingInputs, spanPairsFile, random)
 	return &otlptrace.ResourceSpans{

internal/goldendataset/traces_generator_test.go

@@ -15,17 +15,14 @@
 package goldendataset
 
 import (
-	"io"
-	"math/rand"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
 func TestGenerateTraces(t *testing.T) {
-	random := io.Reader(rand.New(rand.NewSource(42)))
-	rscSpans, err := GenerateResourceSpans("testdata/generated_pict_pairs_traces.txt",
-		"testdata/generated_pict_pairs_spans.txt", random)
+	rscSpans, err := GenerateTraces("testdata/generated_pict_pairs_traces.txt",
+		"testdata/generated_pict_pairs_spans.txt")
 	assert.Nil(t, err)
 	assert.Equal(t, 32, len(rscSpans))
 }

processor/batchprocessor/batch_processor.go

@@ -108,6 +108,8 @@ func (bp *batchProcessor) Start(context.Context, component.Host) error {
 // Shutdown is invoked during service shutdown.
 func (bp *batchProcessor) Shutdown(context.Context) error {
 	bp.cancel()
+
+	// Wait until current batch is drained.
 	<-bp.done
 	return nil
 }
@@ -132,6 +134,7 @@ func (bp *batchProcessor) startProcessingCycle() {
 				// make it cancellable using the context that Shutdown gets as a parameter
 				bp.sendItems(statTimeoutTriggerSend)
 			}
+			// Indicate that we finished draining.
 			close(bp.done)
 			return
 		case item := <-bp.newItem:

processor/batchprocessor/batch_processor_test.go

@@ -689,3 +689,8 @@ func logsReceivedByName(lds []pdata.Logs) map[string]pdata.LogRecord {
 	}
 	return logsReceivedByName
 }
+
+func TestShutdown(t *testing.T) {
+	factory := NewFactory()
+	componenttest.VerifyProcessorShutdown(t, factory, factory.CreateDefaultConfig())
+}

receiver/prometheusreceiver/README.md

@@ -25,6 +25,11 @@ services. It supports the full set of Prometheus configuration, including
 service discovery. Just like you would write in a YAML configuration file
 before starting Prometheus, such as with:
 
+**Note**: Since the collector configuration supports env variable substitution
+`$` charaters in your prometheus configuration are interpreted as environment
+variables.  If you want to use $ charaters in your prometheus configuration,
+you must escape them using `$$`.
+
 ```shell
 prometheus --config.file=prom.yaml
 ```

testbed/correctness/traces/correctness_test.go

@@ -61,8 +61,7 @@ func testWithTracingGoldenDataset(
 	dataProvider := testbed.NewGoldenDataProvider(
 		"../../../internal/goldendataset/testdata/generated_pict_pairs_traces.txt",
 		"../../../internal/goldendataset/testdata/generated_pict_pairs_spans.txt",
-		"",
-		161803)
+		"")
 	factories, err := defaultcomponents.Components()
 	require.NoError(t, err, "default components resulted in: %v", err)
 	runner := testbed.NewInProcessCollector(factories)

testbed/testbed/data_providers.go

@@ -17,9 +17,7 @@ package testbed
 import (
 	"encoding/binary"
 	"fmt"
-	"io"
 	"log"
-	"math/rand"
 	"os"
 	"strconv"
 	"time"
@@ -216,12 +214,12 @@ func (dp *PerfTestDataProvider) GenerateLogs() (pdata.Logs, bool) {
 type GoldenDataProvider struct {
 	tracePairsFile     string
 	spanPairsFile      string
-	random             io.Reader
 	batchesGenerated   *atomic.Uint64
 	dataItemsGenerated *atomic.Uint64
-	resourceSpans      []*otlptrace.ResourceSpans
-	spansIndex         int
-	spansMap           map[string]*otlptrace.Span
+
+	tracesGenerated []pdata.Traces
+	tracesIndex     int
+	spansMap        map[string]*otlptrace.Span
 
 	metricPairsFile  string
 	metricsGenerated []pdata.Metrics
@@ -230,13 +228,11 @@ type GoldenDataProvider struct {
 
 // NewGoldenDataProvider creates a new instance of GoldenDataProvider which generates test data based
 // on the pairwise combinations specified in the tracePairsFile and spanPairsFile input variables.
-// The supplied randomSeed is used to initialize the random number generator used in generating tracing IDs.
-func NewGoldenDataProvider(tracePairsFile string, spanPairsFile string, metricPairsFile string, randomSeed int64) *GoldenDataProvider {
+func NewGoldenDataProvider(tracePairsFile string, spanPairsFile string, metricPairsFile string) *GoldenDataProvider {
 	return &GoldenDataProvider{
 		tracePairsFile:  tracePairsFile,
 		spanPairsFile:   spanPairsFile,
 		metricPairsFile: metricPairsFile,
-		random:          io.Reader(rand.New(rand.NewSource(randomSeed))),
 	}
 }
 
@@ -246,27 +242,22 @@ func (dp *GoldenDataProvider) SetLoadGeneratorCounters(batchesGenerated *atomic.
 }
 
 func (dp *GoldenDataProvider) GenerateTraces() (pdata.Traces, bool) {
-	if dp.resourceSpans == nil {
+	if dp.tracesGenerated == nil {
 		var err error
-		dp.resourceSpans, err = goldendataset.GenerateResourceSpans(dp.tracePairsFile, dp.spanPairsFile, dp.random)
+		dp.tracesGenerated, err = goldendataset.GenerateTraces(dp.tracePairsFile, dp.spanPairsFile)
 		if err != nil {
 			log.Printf("cannot generate traces: %s", err)
-			dp.resourceSpans = make([]*otlptrace.ResourceSpans, 0)
+			dp.tracesGenerated = nil
 		}
 	}
 	dp.batchesGenerated.Inc()
-	if dp.spansIndex >= len(dp.resourceSpans) {
-		return pdata.TracesFromOtlp(make([]*otlptrace.ResourceSpans, 0)), true
-	}
-	resourceSpans := make([]*otlptrace.ResourceSpans, 1)
-	resourceSpans[0] = dp.resourceSpans[dp.spansIndex]
-	dp.spansIndex++
-	spanCount := uint64(0)
-	for _, libSpans := range resourceSpans[0].InstrumentationLibrarySpans {
-		spanCount += uint64(len(libSpans.Spans))
+	if dp.tracesIndex >= len(dp.tracesGenerated) {
+		return pdata.NewTraces(), true
 	}
-	dp.dataItemsGenerated.Add(spanCount)
-	return pdata.TracesFromOtlp(resourceSpans), false
+	td := dp.tracesGenerated[dp.tracesIndex]
+	dp.tracesIndex++
+	dp.dataItemsGenerated.Add(uint64(td.SpanCount()))
+	return td, false
 }
 
 func (dp *GoldenDataProvider) GenerateMetrics() (pdata.Metrics, bool) {
@@ -277,8 +268,8 @@ func (dp *GoldenDataProvider) GenerateMetrics() (pdata.Metrics, bool) {
 			log.Printf("cannot generate metrics: %s", err)
 		}
 	}
-	numMetricsGenerated := len(dp.metricsGenerated)
-	if dp.metricsIndex == numMetricsGenerated {
+	dp.batchesGenerated.Inc()
+	if dp.metricsIndex == len(dp.metricsGenerated) {
 		return pdata.Metrics{}, true
 	}
 	pdm := dp.metricsGenerated[dp.metricsIndex]
@@ -294,7 +285,11 @@ func (dp *GoldenDataProvider) GenerateLogs() (pdata.Logs, bool) {
 
 func (dp *GoldenDataProvider) GetGeneratedSpan(traceID pdata.TraceID, spanID pdata.SpanID) *otlptrace.Span {
 	if dp.spansMap == nil {
-		dp.spansMap = populateSpansMap(dp.resourceSpans)
+		var resourceSpansList []*otlptrace.ResourceSpans
+		for _, td := range dp.tracesGenerated {
+			resourceSpansList = append(resourceSpansList, pdata.TracesToOtlp(td)...)
+		}
+		dp.spansMap = populateSpansMap(resourceSpansList)
 	}
 	key := traceIDAndSpanIDToString(traceID, spanID)
 	return dp.spansMap[key]