Merge branch 'main' of https://github.com/odigos-io/odigos

DEVELOPMENT.md

@@ -0,0 +1,43 @@
+# Development Guide
+
+This guide provides advanced instructions for contributors and maintainers, covering topics such as debugging specific components, analyzing performance profiles, and working with internal tools. It complements the `CONTRIBUTING.md` by offering insights into advanced development workflows and optimization techniques.
+
+---
+
+## CPU and Memory Profiling for the Collectors
+
+### Step 1: Port Forward the Gateway or Data Collection Pod
+Forward the relevant pod to your local machine to enable profiling access:
+
+kubectl port-forward pod/<pod-name> -n odigos-system 1777:1777
+
+
+### Step 2: Collect Profiling Data
+
+- **CPU Profile**  
+   Captures data about the amount of time your application spends executing functions. Use this profile to identify performance bottlenecks, optimize CPU-intensive operations, and analyze which parts of the code consume the most CPU resources.  
+
+   ``` bash
+    curl -o cpu_profile.prof http://localhost:1777/debug/pprof/profile?seconds=30 
+   ```
+
+- **Heap Memory Profile**  
+   Captures a snapshot of memory currently in use by your application after the latest garbage collection. Use this profile to identify memory leaks, track high memory usage, and analyze memory consumption by specific parts of the code.  
+   ``` bash
+   curl -o heap.out http://localhost:1777/debug/pprof/heap
+   ```
+
+- **Historical Memory Allocation**  
+   Provides insights into all memory allocations made by the program since it started running, including memory that has already been freed by the garbage collector (GC). This is useful for understanding memory allocation patterns and optimizing allocation behavior.  
+    ``` bash
+   curl -o allocs.out http://localhost:1777/debug/pprof/allocs
+   ```
+
+### Step 3: Analyze the Profiles
+After collecting the profiling data, use the `go tool pprof` command to analyze the profiles visually in your web browser. Replace `<output file>` with the appropriate file (`cpu_profile.prof`, `heap.out`, or `allocs.out`):
+``` bash
+go tool pprof -http=:8080 <output file>
+```
+This opens an interactive interface in your browser where you can:
+- **Visualize Hotspots**: View flame graphs or directed graphs for easy identification of bottlenecks.
+- **Drill Down**: Explore specific functions or memory allocations for detailed insights.

Makefile

@@ -246,3 +246,7 @@ dev-debug-destination:
 .PHONY: dev-add-nop-destination
 dev-nop-destination:
 	kubectl apply -f ./tests/nop-exporter.yaml
+
+.PHONY: dev-add-backpressue-destination
+dev-backpressue-destination:
+	kubectl apply -f ./tests/backpressure-exporter.yaml

autoscaler/controllers/datacollection/configmap.go

@@ -239,6 +239,7 @@ func calculateConfigMapData(nodeCG *odigosv1.CollectorsGroup, apps *odigosv1.Ins
 			"health_check": config.GenericMap{
 				"endpoint": "0.0.0.0:13133",
 			},
+			"pprof": config.GenericMap{},
 		},
 		Service: config.Service{
 			Pipelines: map[string]config.Pipeline{
@@ -248,7 +249,7 @@ func calculateConfigMapData(nodeCG *odigosv1.CollectorsGroup, apps *odigosv1.Ins
 					Exporters:  []string{"otlp/odigos-own-telemetry-ui"},
 				},
 			},
-			Extensions: []string{"health_check"},
+			Extensions: []string{"health_check", "pprof"},
 			Telemetry: config.Telemetry{
 				Metrics: config.GenericMap{
 					"address": fmt.Sprintf("0.0.0.0:%d", ownMetricsPort),

collector/builder-config.yaml

@@ -19,6 +19,7 @@ exporters:
   - gomod: go.opentelemetry.io/collector/exporter/otlphttpexporter v0.106.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/azureblobstorageexporter v0.106.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/googlecloudstorageexporter v0.106.0
+  - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/mockdestinationexporter v0.106.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awss3exporter v0.106.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/awsxrayexporter v0.106.0
   - gomod: github.com/open-telemetry/opentelemetry-collector-contrib/exporter/azuredataexplorerexporter v0.106.0
@@ -116,5 +117,6 @@ replaces:
   - github.com/open-telemetry/opentelemetry-collector-contrib/odigos/processor/odigossqldboperationprocessor => ../processors/odigossqldboperationprocessor
   - github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/azureblobstorageexporter => ../exporters/azureblobstorageexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/googlecloudstorageexporter => ../exporters/googlecloudstorageexporter
+  - github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/mockdestinationexporter => ../exporters/mockdestinationexporter
   - github.com/open-telemetry/opentelemetry-collector-contrib/odigos/processor/odigostrafficmetrics => ../processors/odigostrafficmetrics
   - go.opentelemetry.io/collector/odigos/providers/odigosfileprovider => ../providers/odigosfileprovider

collector/exporters/mockdestinationexporter/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

collector/exporters/mockdestinationexporter/README.md

@@ -0,0 +1,19 @@
+# Mock Destination Exporter
+
+This exporter can be used for development and testing.
+It allows you to mock a specific behavior of a destination exporter.
+
+## Configuration
+
+The following configuration options are available:
+
+- `response_duration` can be used to set the duration of time until the export response is returned. can be used to simulate slow receivers (due to errors, network issues, etc).
+- `reject_fraction` number from 0 to 1 that determines the fraction of exports that mocks a rejection of the export request.
+
+Example:
+
+```yaml
+│   mockdestination:
+│     reject_fraction: 0.5
+│     response_duration: 500ms
+```

collector/exporters/mockdestinationexporter/config.go

@@ -0,0 +1,33 @@
+package mockdestinationexporter
+
+import (
+	"fmt"
+	"time"
+
+	"go.opentelemetry.io/collector/component"
+)
+
+// Config contains the main configuration options for the mockdestination exporter
+type Config struct {
+
+	// ResponseDuration is the amount of time the exporter will wait before returning a response.
+	// It can be used to simulate loaded and slow destinations.
+	ResponseDuration time.Duration `mapstructure:"response_duration"`
+
+	// RejectFraction is the fraction of exports that will randomly be rejected.
+	// Set to 0 to disable rejection, and to 1 to reject all exports.
+	// Can be used to simulate destinations that are back-pressuring the collector.
+	RejectFraction float64 `mapstructure:"reject_fraction"`
+}
+
+func (c *Config) Validate() error {
+	if c.ResponseDuration < 0 {
+		return fmt.Errorf("response_duration must be a non-negative duration")
+	}
+	if c.RejectFraction < 0 || c.RejectFraction > 1 {
+		return fmt.Errorf("reject_fraction must be a fraction between 0 and 1")
+	}
+	return nil
+}
+
+var _ component.Config = (*Config)(nil)

collector/exporters/mockdestinationexporter/config_test.go

@@ -0,0 +1,15 @@
+package mockdestinationexporter
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+func TestUnmarshalDefaultConfig(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	assert.NoError(t, confmap.New().Unmarshal(&cfg))
+	assert.Equal(t, factory.CreateDefaultConfig(), cfg)
+}

collector/exporters/mockdestinationexporter/exporter.go

@@ -0,0 +1,62 @@
+package mockdestinationexporter
+
+import (
+	"context"
+	"errors"
+	"math/rand/v2"
+	"time"
+
+	"go.opentelemetry.io/collector/exporter"
+
+	"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"
+)
+
+type MockDestinationExporter struct {
+	config *Config
+	logger *zap.Logger
+}
+
+func NewMockDestinationExporter(config *Config,
+	params exporter.Settings) (*MockDestinationExporter, error) {
+
+	if config == nil {
+		return nil, errors.New("mock destination exporter config is nil")
+	}
+
+	logger := params.Logger
+
+	mockDestinationExporter := &MockDestinationExporter{
+		config: config,
+		logger: logger,
+	}
+	return mockDestinationExporter, nil
+}
+
+func (e *MockDestinationExporter) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (e *MockDestinationExporter) ConsumeTraces(ctx context.Context, traces ptrace.Traces) error {
+	return e.mockExport(ctx)
+}
+
+func (e *MockDestinationExporter) ConsumeMetrics(ctx context.Context, metrics pmetric.Metrics) error {
+	return e.mockExport(ctx)
+}
+
+func (e *MockDestinationExporter) ConsumeLogs(ctx context.Context, logs plog.Logs) error {
+	return e.mockExport(ctx)
+}
+
+func (e *MockDestinationExporter) mockExport(context.Context) error {
+	// not taking care of ctx cancel and shutdown as this is a dummy exporter and not used in production
+	<-time.After(e.config.ResponseDuration)
+	if rand.Float64() < e.config.RejectFraction {
+		return errors.New("export rejected by mock destination")
+	}
+	return nil
+}

collector/exporters/mockdestinationexporter/factory.go

@@ -0,0 +1,85 @@
+package mockdestinationexporter
+
+import (
+	"context"
+	"time"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/odigos/exporter/mockdestinationexporter/internal/metadata"
+	"go.opentelemetry.io/collector/exporter"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/exporter/exporterhelper"
+)
+
+// NewFactory creates a factory for GCS exporter.
+func NewFactory() exporter.Factory {
+	return exporter.NewFactory(
+		metadata.Type,
+		createDefaultConfig,
+		exporter.WithLogs(createLogsExporter, component.StabilityLevelBeta),
+		exporter.WithTraces(createTracesExporter, component.StabilityLevelBeta),
+		exporter.WithMetrics(createMetricsExporter, component.StabilityLevelBeta))
+}
+
+func createDefaultConfig() component.Config {
+	return &Config{
+		ResponseDuration: time.Millisecond * 100,
+		RejectFraction:   0,
+	}
+}
+
+func createLogsExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	cfg component.Config) (exporter.Logs, error) {
+
+	pCfg := cfg.(*Config)
+	gcsExporter, err := NewMockDestinationExporter(pCfg, set)
+	if err != nil {
+		return nil, err
+	}
+
+	return exporterhelper.NewLogsExporter(
+		ctx,
+		set,
+		cfg,
+		gcsExporter.ConsumeLogs)
+}
+
+func createTracesExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	cfg component.Config) (exporter.Traces, error) {
+
+	pCfg := cfg.(*Config)
+	gcsExporter, err := NewMockDestinationExporter(pCfg, set)
+	if err != nil {
+		return nil, err
+	}
+
+	return exporterhelper.NewTracesExporter(
+		ctx,
+		set,
+		cfg,
+		gcsExporter.ConsumeTraces,
+	)
+}
+
+func createMetricsExporter(
+	ctx context.Context,
+	set exporter.Settings,
+	cfg component.Config) (exporter.Metrics, error) {
+
+	pCfg := cfg.(*Config)
+	gcsExporter, err := NewMockDestinationExporter(pCfg, set)
+	if err != nil {
+		return nil, err
+	}
+
+	return exporterhelper.NewMetricsExporter(
+		ctx,
+		set,
+		cfg,
+		gcsExporter.ConsumeMetrics,
+	)
+}