[consumer/consumertest] Add ProfileCount() (#14251)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

In #14239 `(Profiles)ProfileCount() int` was introduced. Provide this
function also in consumertests for other tests.


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

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

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

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

---------

Signed-off-by: Florian Lehner <florian.lehner@elastic.co>

Author: florianl

.chloggen/config.yaml

@@ -13,7 +13,6 @@ components:
     - connector/forward
     - connector/sample
     - consumer/consumererror/xconsumererror
-    - consumer/consumertest
     - consumer/xconsumer
     - docs/rfcs
     - exporter/debug
@@ -38,6 +37,7 @@ components:
     - pkg/config/configtelemetry
     - pkg/config/configtls
     - pkg/confmap
+    - pkg/consumer/consumertest
     - pkg/exporterhelper
     - pkg/otelcol
     - pkg/pdata

.chloggen/profiles-consumertest.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. receiver/otlp)
+component: pkg/consumer/consumertest
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add ProfileCount()
+
+# One or more tracking issues or pull requests related to the change
+issues: [14251]
+
+# (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:
+
+# 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]

consumer/consumertest/sink.go

@@ -201,10 +201,11 @@ func (sle *LogsSink) Contexts() []context.Context {
 // stores all profiles and allows querying them for testing.
 type ProfilesSink struct {
 	nonMutatingConsumer
-	mu          sync.Mutex
-	profiles    []pprofile.Profiles
-	contexts    []context.Context
-	sampleCount int
+	mu           sync.Mutex
+	profiles     []pprofile.Profiles
+	contexts     []context.Context
+	sampleCount  int
+	profileCount int
 }
 
 var _ xconsumer.Profiles = (*ProfilesSink)(nil)
@@ -217,6 +218,7 @@ func (ste *ProfilesSink) ConsumeProfiles(ctx context.Context, td pprofile.Profil
 	ste.profiles = append(ste.profiles, td)
 	ste.contexts = append(ste.contexts, ctx)
 	ste.sampleCount += td.SampleCount()
+	ste.profileCount += td.ProfileCount()
 
 	return nil
 }
@@ -231,13 +233,20 @@ func (ste *ProfilesSink) AllProfiles() []pprofile.Profiles {
 	return copyProfiles
 }
 
-// SampleCount returns the number of profiles stored by this sink since last Reset.
+// SampleCount returns the number of samples stored by this sink since last Reset.
 func (ste *ProfilesSink) SampleCount() int {
 	ste.mu.Lock()
 	defer ste.mu.Unlock()
 	return ste.sampleCount
 }
 
+// ProfileCount returns the number of profiles stored by this sink since last Reset.
+func (ste *ProfilesSink) ProfileCount() int {
+	ste.mu.Lock()
+	defer ste.mu.Unlock()
+	return ste.profileCount
+}
+
 // Reset deletes any stored data.
 func (ste *ProfilesSink) Reset() {
 	ste.mu.Lock()
@@ -246,6 +255,7 @@ func (ste *ProfilesSink) Reset() {
 	ste.profiles = nil
 	ste.contexts = nil
 	ste.sampleCount = 0
+	ste.profileCount = 0
 }
 
 // Contexts returns the contexts stored by this sink since last Reset.

consumer/consumertest/sink_test.go

@@ -78,10 +78,14 @@ func TestProfilesSink(t *testing.T) {
 		want = append(want, td)
 	}
 	assert.Equal(t, want, sink.AllProfiles())
+	// Each Profile in Profiles holds a single sample.
+	// So SampleCount() equals ProfileCount() in this case.
 	assert.Equal(t, len(want), sink.SampleCount())
+	assert.Equal(t, len(want), sink.ProfileCount())
 	sink.Reset()
 	assert.Empty(t, sink.AllProfiles())
-	assert.Empty(t, sink.SampleCount())
+	assert.Equal(t, 0, sink.SampleCount())
+	assert.Equal(t, 0, sink.ProfileCount())
 }
 
 func TestTracesSinkWithContext(t *testing.T) {