Add obs for logs in scraper/scraperhelper (#12036)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description
This PR adds obs for logs in scraper/scraperhelper, also introduced new
metrics for scraping logs.

<!-- Issue number if applicable -->
#### Link to tracking issue
Relates to #11238 

<!--Describe what testing was performed and which tests were added.-->
#### Testing
Added

<!--Describe the documentation added.-->
#### Documentation
Added

<!--Please delete paragraphs that you did not use before submitting.-->

Author: sincejune

.chloggen/add-obs-for-logs.yaml

@@ -0,0 +1,25 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. otlpreceiver)
+component: scraper/scraperhelper
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add obs_logs for scraper/scraperhelper
+
+# One or more tracking issues or pull requests related to the change
+issues: [12036]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: This change adds obs for logs in scraper/scraperhelper, also introduced new metrics for scraping logs.
+
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

scraper/scraperhelper/documentation.md

@@ -6,6 +6,14 @@
 
 The following telemetry is emitted by this component.
 
+### otelcol_scraper_errored_log_records
+
+Number of log records that were unable to be scraped. [alpha]
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {datapoints} | Sum | Int | true |
+
 ### otelcol_scraper_errored_metric_points
 
 Number of metric points that were unable to be scraped. [alpha]
@@ -14,6 +22,14 @@ Number of metric points that were unable to be scraped. [alpha]
 | ---- | ----------- | ---------- | --------- |
 | {datapoints} | Sum | Int | true |
 
+### otelcol_scraper_scraped_log_records
+
+Number of log records successfully scraped. [alpha]
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {datapoints} | Sum | Int | true |
+
 ### otelcol_scraper_scraped_metric_points
 
 Number of metric points successfully scraped. [alpha]

scraper/scraperhelper/internal/metadata/generated_telemetry.go

@@ -26,4 +26,24 @@ telemetry:
       unit: "{datapoints}"
       sum:
         value_type: int
-        monotonic: true
+        monotonic: true
+
+    scraper_scraped_log_records:
+      enabled: true
+      stability:
+        level: alpha
+      description: Number of log records successfully scraped.
+      unit: "{datapoints}"
+      sum:
+        value_type: int
+        monotonic: true
+
+    scraper_errored_log_records:
+      enabled: true
+      stability:
+        level: alpha
+      description: Number of log records that were unable to be scraped.
+      unit: "{datapoints}"
+      sum:
+        value_type: int
+        monotonic: true

scraper/scraperhelper/internal/metadatatest/generated_telemetrytest_test.go

@@ -0,0 +1,81 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package scraperhelper // import "go.opentelemetry.io/collector/scraper/scraperhelper"
+
+import (
+	"context"
+	"errors"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/metric"
+	"go.uber.org/zap"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pipeline"
+	"go.opentelemetry.io/collector/scraper"
+	"go.opentelemetry.io/collector/scraper/scrapererror"
+	"go.opentelemetry.io/collector/scraper/scraperhelper/internal/metadata"
+)
+
+const (
+	// scrapedLogRecordsKey used to identify log records scraped by the
+	// Collector.
+	scrapedLogRecordsKey = "scraped_log_records"
+	// erroredLogRecordsKey used to identify log records errored (i.e.
+	// unable to be scraped) by the Collector.
+	erroredLogRecordsKey = "errored_log_records"
+)
+
+func newObsLogs(delegate scraper.ScrapeLogsFunc, receiverID component.ID, scraperID component.ID, telSettings component.TelemetrySettings) (scraper.ScrapeLogsFunc, error) {
+	telemetryBuilder, errBuilder := metadata.NewTelemetryBuilder(telSettings)
+	if errBuilder != nil {
+		return nil, errBuilder
+	}
+
+	tracer := metadata.Tracer(telSettings)
+	spanName := scraperKey + spanNameSep + scraperID.String() + spanNameSep + "ScrapeLogs"
+	otelAttrs := metric.WithAttributeSet(attribute.NewSet(
+		attribute.String(receiverKey, receiverID.String()),
+		attribute.String(scraperKey, scraperID.String()),
+	))
+
+	return func(ctx context.Context) (plog.Logs, error) {
+		ctx, span := tracer.Start(ctx, spanName)
+		defer span.End()
+
+		md, err := delegate(ctx)
+		numScrapedLogs := 0
+		numErroredLogs := 0
+		if err != nil {
+			telSettings.Logger.Error("Error scraping logs", zap.Error(err))
+			var partialErr scrapererror.PartialScrapeError
+			if errors.As(err, &partialErr) {
+				numErroredLogs = partialErr.Failed
+				numScrapedLogs = md.LogRecordCount()
+			}
+		} else {
+			numScrapedLogs = md.LogRecordCount()
+		}
+
+		telemetryBuilder.ScraperScrapedLogRecords.Add(ctx, int64(numScrapedLogs), otelAttrs)
+		telemetryBuilder.ScraperErroredLogRecords.Add(ctx, int64(numErroredLogs), otelAttrs)
+
+		// end span according to errors
+		if span.IsRecording() {
+			span.SetAttributes(
+				attribute.String(formatKey, pipeline.SignalMetrics.String()),
+				attribute.Int64(scrapedLogRecordsKey, int64(numScrapedLogs)),
+				attribute.Int64(erroredLogRecordsKey, int64(numErroredLogs)),
+			)
+
+			if err != nil {
+				span.SetStatus(codes.Error, err.Error())
+			}
+		}
+
+		return md, err
+	}, nil
+}

scraper/scraperhelper/metadata.yaml

@@ -0,0 +1,136 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package scraperhelper
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/testdata"
+	"go.opentelemetry.io/collector/scraper/scraperhelper/internal/metadatatest"
+)
+
+func TestScrapeLogsDataOp(t *testing.T) {
+	tt := metadatatest.SetupTelemetry()
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	tel := tt.NewTelemetrySettings()
+	// TODO: Add capability for tracing testing in metadatatest.
+	spanRecorder := new(tracetest.SpanRecorder)
+	tel.TracerProvider = sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(spanRecorder))
+
+	parentCtx, parentSpan := tel.TracerProvider.Tracer("test").Start(context.Background(), t.Name())
+	defer parentSpan.End()
+
+	params := []testParams{
+		{items: 23, err: partialErrFake},
+		{items: 29, err: errFake},
+		{items: 15, err: nil},
+	}
+	for i := range params {
+		sf, err := newObsLogs(func(context.Context) (plog.Logs, error) {
+			return testdata.GenerateLogs(params[i].items), params[i].err
+		}, receiverID, scraperID, tel)
+		require.NoError(t, err)
+		_, err = sf.ScrapeLogs(parentCtx)
+		require.ErrorIs(t, err, params[i].err)
+	}
+
+	spans := spanRecorder.Ended()
+	require.Equal(t, len(params), len(spans))
+
+	var scrapedLogRecords, erroredLogRecords int
+	for i, span := range spans {
+		assert.Equal(t, "scraper/"+scraperID.String()+"/ScrapeLogs", span.Name())
+		switch {
+		case params[i].err == nil:
+			scrapedLogRecords += params[i].items
+			require.Contains(t, span.Attributes(), attribute.Int64(scrapedLogRecordsKey, int64(params[i].items)))
+			require.Contains(t, span.Attributes(), attribute.Int64(erroredLogRecordsKey, 0))
+			assert.Equal(t, codes.Unset, span.Status().Code)
+		case errors.Is(params[i].err, errFake):
+			// Since we get an error, we cannot record any metrics because we don't know if the returned plog.Logs is valid instance.
+			require.Contains(t, span.Attributes(), attribute.Int64(scrapedLogRecordsKey, 0))
+			require.Contains(t, span.Attributes(), attribute.Int64(erroredLogRecordsKey, 0))
+			assert.Equal(t, codes.Error, span.Status().Code)
+			assert.Equal(t, params[i].err.Error(), span.Status().Description)
+		case errors.Is(params[i].err, partialErrFake):
+			scrapedLogRecords += params[i].items
+			erroredLogRecords += 2
+			require.Contains(t, span.Attributes(), attribute.Int64(scrapedLogRecordsKey, int64(params[i].items)))
+			require.Contains(t, span.Attributes(), attribute.Int64(erroredLogRecordsKey, 2))
+			assert.Equal(t, codes.Error, span.Status().Code)
+			assert.Equal(t, params[i].err.Error(), span.Status().Description)
+		default:
+			t.Fatalf("unexpected err param: %v", params[i].err)
+		}
+	}
+
+	checkScraperLogs(t, tt, receiverID, scraperID, int64(scrapedLogRecords), int64(erroredLogRecords))
+}
+
+func TestCheckScraperLogs(t *testing.T) {
+	tt := metadatatest.SetupTelemetry()
+	t.Cleanup(func() { require.NoError(t, tt.Shutdown(context.Background())) })
+
+	sf, err := newObsLogs(func(context.Context) (plog.Logs, error) {
+		return testdata.GenerateLogs(7), nil
+	}, receiverID, scraperID, tt.NewTelemetrySettings())
+	require.NoError(t, err)
+	_, err = sf.ScrapeLogs(context.Background())
+	require.NoError(t, err)
+
+	checkScraperLogs(t, tt, receiverID, scraperID, 7, 0)
+}
+
+func checkScraperLogs(t *testing.T, tt metadatatest.Telemetry, receiver component.ID, scraper component.ID, scrapedLogRecords int64, erroredLogRecords int64) {
+	tt.AssertMetrics(t, []metricdata.Metrics{
+		{
+			Name:        "otelcol_scraper_scraped_log_records",
+			Description: "Number of log records successfully scraped. [alpha]",
+			Unit:        "{datapoints}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String(receiverKey, receiver.String()),
+							attribute.String(scraperKey, scraper.String())),
+						Value: scrapedLogRecords,
+					},
+				},
+			},
+		},
+		{
+			Name:        "otelcol_scraper_errored_log_records",
+			Description: "Number of log records that were unable to be scraped. [alpha]",
+			Unit:        "{datapoints}",
+			Data: metricdata.Sum[int64]{
+				Temporality: metricdata.CumulativeTemporality,
+				IsMonotonic: true,
+				DataPoints: []metricdata.DataPoint[int64]{
+					{
+						Attributes: attribute.NewSet(
+							attribute.String(receiverKey, receiver.String()),
+							attribute.String(scraperKey, scraper.String())),
+						Value: erroredLogRecords,
+					},
+				},
+			},
+		},
+	}, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreExemplars())
+}