qq

Author: koponk04

builder-config.yaml

@@ -6,3 +6,5 @@ dist:
 
 processors:
   - gomod: github.com/koponk04/otel-custom-components v0.1.0
+    import: github.com/koponk04/otel-custom-components/processor/httpprocessor
+    path: ./

processor/httpprocessor/config.go

@@ -1,7 +1,7 @@
 package httpprocessor
 
 import (
-	"go.opentelemetry.io/collector/component"
+	"fmt"
 )
 
 // Config defines the configuration for the HTTP processor
@@ -19,7 +19,7 @@ type Config struct {
 // Validate checks if the processor configuration is valid
 func (cfg *Config) Validate() error {
 	if cfg.Endpoint == "" {
-		return component.ErrNilNextConsumer
+		return fmt.Errorf("endpoint cannot be empty")
 	}
 	return nil
 }

processor/httpprocessor/factory.go

@@ -6,7 +6,6 @@ import (
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
-	"go.opentelemetry.io/collector/processor/processorhelper"
 )
 
 const (
@@ -42,16 +41,10 @@ func createTracesProcessor(
 	proc := &httpProcessor{
 		config: processorCfg,
 		logger: set.Logger,
+		next:   nextConsumer,
 	}
 
-	return processorhelper.NewTracesProcessor(
-		ctx,
-		set,
-		cfg,
-		nextConsumer,
-		proc.processTraces,
-		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}),
-	)
+	return proc, nil
 }
 
 func createMetricsProcessor(
@@ -61,19 +54,13 @@ func createMetricsProcessor(
 	nextConsumer consumer.Metrics,
 ) (processor.Metrics, error) {
 	processorCfg := cfg.(*Config)
-	proc := &httpProcessor{
+	proc := &httpMetricsProcessor{
 		config: processorCfg,
 		logger: set.Logger,
+		next:   nextConsumer,
 	}
 
-	return processorhelper.NewMetricsProcessor(
-		ctx,
-		set,
-		cfg,
-		nextConsumer,
-		proc.processMetrics,
-		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}),
-	)
+	return proc, nil
 }
 
 func createLogsProcessor(
@@ -83,17 +70,11 @@ func createLogsProcessor(
 	nextConsumer consumer.Logs,
 ) (processor.Logs, error) {
 	processorCfg := cfg.(*Config)
-	proc := &httpProcessor{
+	proc := &httpLogsProcessor{
 		config: processorCfg,
 		logger: set.Logger,
+		next:   nextConsumer,
 	}
 
-	return processorhelper.NewLogsProcessor(
-		ctx,
-		set,
-		cfg,
-		nextConsumer,
-		proc.processLogs,
-		processorhelper.WithCapabilities(consumer.Capabilities{MutatesData: true}),
-	)
+	return proc, nil
 }

processor/httpprocessor/processor.go

@@ -3,18 +3,34 @@ package httpprocessor
 import (
 	"context"
 
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
 	"go.uber.org/zap"
 )
 
+// httpProcessor handles traces
 type httpProcessor struct {
 	config *Config
 	logger *zap.Logger
+	next   consumer.Traces
 }
 
-func (p *httpProcessor) processTraces(ctx context.Context, td ptrace.Traces) (ptrace.Traces, error) {
+func (p *httpProcessor) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: true}
+}
+
+func (p *httpProcessor) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func (p *httpProcessor) Shutdown(ctx context.Context) error {
+	return nil
+}
+
+func (p *httpProcessor) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
 	// Implement your trace processing logic here
 	p.logger.Info("Processing traces", zap.Int("span_count", td.SpanCount()))
 
@@ -33,17 +49,55 @@ func (p *httpProcessor) processTraces(ctx context.Context, td ptrace.Traces) (pt
 		}
 	}
 
-	return td, nil
+	return p.next.ConsumeTraces(ctx, td)
+}
+
+// httpMetricsProcessor handles metrics
+type httpMetricsProcessor struct {
+	config *Config
+	logger *zap.Logger
+	next   consumer.Metrics
+}
+
+func (p *httpMetricsProcessor) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: true}
+}
+
+func (p *httpMetricsProcessor) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func (p *httpMetricsProcessor) Shutdown(ctx context.Context) error {
+	return nil
 }
 
-func (p *httpProcessor) processMetrics(ctx context.Context, md pmetric.Metrics) (pmetric.Metrics, error) {
+func (p *httpMetricsProcessor) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
 	// Implement your metric processing logic here
 	p.logger.Info("Processing metrics", zap.Int("metric_count", md.MetricCount()))
 
-	return md, nil
+	return p.next.ConsumeMetrics(ctx, md)
+}
+
+// httpLogsProcessor handles logs
+type httpLogsProcessor struct {
+	config *Config
+	logger *zap.Logger
+	next   consumer.Logs
+}
+
+func (p *httpLogsProcessor) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: true}
+}
+
+func (p *httpLogsProcessor) Start(ctx context.Context, host component.Host) error {
+	return nil
+}
+
+func (p *httpLogsProcessor) Shutdown(ctx context.Context) error {
+	return nil
 }
 
-func (p *httpProcessor) processLogs(ctx context.Context, ld plog.Logs) (plog.Logs, error) {
+func (p *httpLogsProcessor) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 	// Implement your log processing logic here
 	p.logger.Info("Processing logs", zap.Int("log_count", ld.LogRecordCount()))
 
@@ -62,5 +116,5 @@ func (p *httpProcessor) processLogs(ctx context.Context, ld plog.Logs) (plog.Log
 		}
 	}
 
-	return ld, nil
+	return p.next.ConsumeLogs(ctx, ld)
 }