add otel to pipeline builder (#672)

README.md

@@ -49,15 +49,16 @@ Usage:
   flowlogs-pipeline [flags]  
 
 Flags:  
-      --config string            config file (default is $HOME/.flowlogs-pipeline)  
-      --health.address string    Health server address (default "0.0.0.0")  
-      --health.port string       Health server port (default "8080")  
-  -h, --help                     help for flowlogs-pipeline  
-      --log-level string         Log level: debug, info, warning, error (default "error")  
-      --metricsSettings string   json for global metrics settings  
-      --parameters string        json of config file parameters field  
-      --pipeline string          json of config file pipeline field  
-      --profile.port int         Go pprof tool port (default: disabled)
+      --config string              config file (default is $HOME/.flowlogs-pipeline)  
+      --dynamicParameters string   json of configmap location for dynamic parameters  
+      --health.address string      Health server address (default "0.0.0.0")  
+      --health.port string         Health server port (default "8080")  
+  -h, --help                       help for flowlogs-pipeline  
+      --log-level string           Log level: debug, info, warning, error (default "error")  
+      --metricsSettings string     json for global metrics settings  
+      --parameters string          json of config file parameters field  
+      --pipeline string            json of config file pipeline field  
+      --profile.port int           Go pprof tool port (default: disabled)
 ```
 <!---END-AUTO-flowlogs-pipeline_help--->
 
@@ -927,6 +928,8 @@ Develop
   docs                  Update flowlogs-pipeline documentation  
   clean                 Clean  
   tests-unit            Unit tests  
+  coverage-report       Generate coverage report  
+  coverage-report-html  Generate HTML coverage report  
   tests-fast            Fast unit tests (no race tests / coverage)  
   tests-e2e             End-to-end tests  
   tests-all             All tests  

pkg/config/pipeline_builder.go

@@ -165,6 +165,27 @@ func (b *PipelineBuilderStage) EncodeS3(name string, s3 api.EncodeS3) PipelineBu
 	return b.next(name, NewEncodeS3Params(name, s3))
 }
 
+// EncodeOtelLogs chains the current stage with an EncodeOtelLogs stage (writing logs to open telemetry) and returns that new stage
+//
+//nolint:golint,gocritic
+func (b *PipelineBuilderStage) EncodeOtelLogs(name string, logs api.EncodeOtlpLogs) PipelineBuilderStage {
+	return b.next(name, NewEncodeOtelLogsParams(name, logs))
+}
+
+// EncodeOtelMetrics chains the current stage with an EncodeOtelMetrics stage (writing metrics to open telemetry) and returns that new stage
+//
+//nolint:golint,gocritic
+func (b *PipelineBuilderStage) EncodeOtelMetrics(name string, metrics api.EncodeOtlpMetrics) PipelineBuilderStage {
+	return b.next(name, NewEncodeOtelMetricsParams(name, metrics))
+}
+
+// EncodeOtelTraces chains the current stage with an EncodeOtelTraces stage (writing traces to open telemetry) and returns that new stage
+//
+//nolint:golint,gocritic
+func (b *PipelineBuilderStage) EncodeOtelTraces(name string, traces api.EncodeOtlpTraces) PipelineBuilderStage {
+	return b.next(name, NewEncodeOtelTracesParams(name, traces))
+}
+
 // WriteStdout chains the current stage with a WriteStdout stage and returns that new stage
 func (b *PipelineBuilderStage) WriteStdout(name string, stdout api.WriteStdout) PipelineBuilderStage {
 	return b.next(name, NewWriteStdoutParams(name, stdout))

pkg/config/stage_params.go

@@ -74,6 +74,21 @@ func NewEncodeS3Params(name string, s3 api.EncodeS3) StageParam {
 	return StageParam{Name: name, Encode: &Encode{Type: api.S3Type, S3: &s3}}
 }
 
+//nolint:golint,gocritic
+func NewEncodeOtelLogsParams(name string, otelLogs api.EncodeOtlpLogs) StageParam {
+	return StageParam{Name: name, Encode: &Encode{Type: api.OtlpLogsType, OtlpLogs: &otelLogs}}
+}
+
+//nolint:golint,gocritic
+func NewEncodeOtelMetricsParams(name string, otelMetrics api.EncodeOtlpMetrics) StageParam {
+	return StageParam{Name: name, Encode: &Encode{Type: api.OtlpMetricsType, OtlpMetrics: &otelMetrics}}
+}
+
+//nolint:golint,gocritic
+func NewEncodeOtelTracesParams(name string, otelTraces api.EncodeOtlpTraces) StageParam {
+	return StageParam{Name: name, Encode: &Encode{Type: api.OtlpTracesType, OtlpTraces: &otelTraces}}
+}
+
 func NewWriteStdoutParams(name string, stdout api.WriteStdout) StageParam {
 	return StageParam{Name: name, Write: &Write{Type: api.StdoutType, Stdout: &stdout}}
 }