use processorhelperprofiles

processor/k8sattributesprocessor/factory.go

@@ -9,9 +9,9 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/consumer/consumerprofiles"
-	"go.opentelemetry.io/collector/pdata/pprofile"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+	"go.opentelemetry.io/collector/processor/processorhelper/processorhelperprofiles"
 	"go.opentelemetry.io/collector/processor/processorprofiles"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/k8sconfig"
@@ -141,34 +141,25 @@ func createLogsProcessorWithOptions(
 		processorhelper.WithShutdown(kp.Shutdown))
 }
 
-type profiles struct {
-	component.StartFunc
-	component.ShutdownFunc
-	consumerprofiles.Profiles
-}
-
 func createProfilesProcessorWithOptions(
-	_ context.Context,
+	ctx context.Context,
 	set processor.Settings,
 	cfg component.Config,
 	nextProfilesConsumer consumerprofiles.Profiles,
 	options ...option,
 ) (processorprofiles.Profiles, error) {
 	kp := createKubernetesProcessor(set, cfg, options...)
 
-	profilesConsumer, err := consumerprofiles.NewProfiles(func(ctx context.Context, pd pprofile.Profiles) (err error) {
-		pd = kp.processProfiles(ctx, pd)
-		return nextProfilesConsumer.ConsumeProfiles(ctx, pd)
-	}, consumer.WithCapabilities(consumerCapabilities))
-	if err != nil {
-		return nil, err
-	}
-
-	return &profiles{
-		StartFunc:    kp.Start,
-		ShutdownFunc: kp.Shutdown,
-		Profiles:     profilesConsumer,
-	}, nil
+	return processorhelperprofiles.NewProfiles(
+		ctx,
+		set,
+		cfg,
+		nextProfilesConsumer,
+		kp.processProfiles,
+		processorhelperprofiles.WithCapabilities(consumerCapabilities),
+		processorhelperprofiles.WithStart(kp.Start),
+		processorhelperprofiles.WithShutdown(kp.Shutdown),
+	)
 }
 
 func createKubernetesProcessor(

processor/k8sattributesprocessor/go.mod

@@ -100,6 +100,7 @@ require (
 	go.opentelemetry.io/collector/extension v0.112.0 // indirect
 	go.opentelemetry.io/collector/extension/auth v0.112.0 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.112.0 // indirect
+	go.opentelemetry.io/collector/processor/processorhelper/processorhelperprofiles v0.0.0-20241029112935-002a74860455 // indirect
 	go.opentelemetry.io/collector/receiver/receiverprofiles v0.112.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.56.0 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.56.0 // indirect

processor/k8sattributesprocessor/processor.go

@@ -119,13 +119,13 @@ func (kp *kubernetesprocessor) processLogs(ctx context.Context, ld plog.Logs) (p
 }
 
 // processProfiles process profiles and add k8s metadata using resource IP, hostname or incoming IP as pod origin.
-func (kp *kubernetesprocessor) processProfiles(ctx context.Context, pd pprofile.Profiles) pprofile.Profiles {
+func (kp *kubernetesprocessor) processProfiles(ctx context.Context, pd pprofile.Profiles) (pprofile.Profiles, error) {
 	rp := pd.ResourceProfiles()
 	for i := 0; i < rp.Len(); i++ {
 		kp.processResource(ctx, rp.At(i).Resource())
 	}
 
-	return pd
+	return pd, nil
 }
 
 // processResource adds Pod metadata tags to resource based on pod association configuration