Feat: logspipelineprocessor: read/address body fields in pipeline operators directly if body is json

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/apache/thrift v0.20.0
 	github.com/aws/aws-sdk-go v1.53.11
 	github.com/cenkalti/backoff/v4 v4.3.0
+	github.com/expr-lang/expr v1.16.9
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang-migrate/migrate/v4 v4.15.1
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
@@ -181,6 +182,7 @@ require (
 	google.golang.org/grpc v1.64.1
 	google.golang.org/protobuf v1.34.1
 	gopkg.in/yaml.v2 v2.4.0
+	gotest.tools/v3 v3.2.0
 )
 
 require (
@@ -274,7 +276,6 @@ require (
 	github.com/envoyproxy/go-control-plane v0.12.0 // indirect
 	github.com/envoyproxy/protoc-gen-validate v1.0.4 // indirect
 	github.com/euank/go-kmsg-parser v2.0.0+incompatible // indirect
-	github.com/expr-lang/expr v1.16.9 // indirect
 	github.com/fatih/color v1.16.0 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.2 // indirect
 	github.com/go-faster/city v1.0.1 // indirect
@@ -497,7 +498,7 @@ require (
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
-	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/json-iterator/go v1.1.12
 	github.com/klauspost/compress v1.17.8 // indirect
 	github.com/leodido/ragel-machinery v0.0.0-20190525184631-5f46317e436b // indirect
 	github.com/leoluk/perflib_exporter v0.2.1 // indirect

go.sum

@@ -1411,6 +1411,7 @@ github.com/spf13/cobra v1.8.0 h1:7aJaZx1B85qltLMc546zn58BxxfZdR/W22ej9CFoEf0=
 github.com/spf13/cobra v1.8.0/go.mod h1:WXLWApfZ71AjXPya3WOlMsY9yMs7YeiHhFVlvLyhcho=
 github.com/spf13/jwalterweatherman v1.1.0/go.mod h1:aNWZUN0dPAAO/Ljvb5BEdw96iTZ0EXowPYD95IqWIGo=
 github.com/spf13/pflag v0.0.0-20170130214245-9ff6c6923cff/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
+github.com/spf13/pflag v1.0.3/go.mod h1:DYY7MBk1bdzusC3SYhjObp+wFpr4gzcvqqNjLnInEg4=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.8.1/go.mod h1:o0Pch8wJ9BVSWGQMbra6iw0oQ5oktSIBaujf1rJH9Ns=

pkg/parser/grok/config_test.go

@@ -17,8 +17,9 @@ import (
 	"path/filepath"
 	"testing"
 
+	signozstanzaentry "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/entry"
+	signozstanzahelper "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/operatortest"
 )
 
@@ -43,15 +44,15 @@ func TestParserGoldenConfig(t *testing.T) {
 				Name: "parse_from_simple",
 				Expect: func() *Config {
 					cfg := NewConfig()
-					cfg.ParseFrom = entry.NewBodyField("from")
+					cfg.ParseFrom = signozstanzaentry.Field{signozstanzaentry.NewBodyField("from")}
 					return cfg
 				}(),
 			},
 			{
 				Name: "parse_to_simple",
 				Expect: func() *Config {
 					cfg := NewConfig()
-					cfg.ParseTo = entry.RootableField{Field: entry.NewBodyField("log")}
+					cfg.ParseTo = signozstanzaentry.RootableField{Field: signozstanzaentry.Field{signozstanzaentry.NewBodyField("log")}}
 					return cfg
 				}(),
 			},
@@ -67,8 +68,8 @@ func TestParserGoldenConfig(t *testing.T) {
 				Name: "timestamp",
 				Expect: func() *Config {
 					cfg := NewConfig()
-					parseField := entry.NewBodyField("timestamp_field")
-					newTime := helper.TimeParser{
+					parseField := signozstanzaentry.Field{signozstanzaentry.NewBodyField("timestamp_field")}
+					newTime := signozstanzahelper.TimeParser{
 						LayoutType: "strptime",
 						Layout:     "%Y-%m-%d",
 						ParseFrom:  &parseField,
@@ -81,8 +82,8 @@ func TestParserGoldenConfig(t *testing.T) {
 				Name: "severity",
 				Expect: func() *Config {
 					cfg := NewConfig()
-					parseField := entry.NewBodyField("severity_field")
-					severityParser := helper.NewSeverityConfig()
+					parseField := signozstanzaentry.Field{signozstanzaentry.NewBodyField("severity_field")}
+					severityParser := signozstanzahelper.NewSeverityConfig()
 					severityParser.ParseFrom = &parseField
 					mapping := map[string]interface{}{
 						"critical": "5xx",
@@ -108,8 +109,8 @@ func TestParserGoldenConfig(t *testing.T) {
 				Expect: func() *Config {
 					cfg := NewConfig()
 					cfg.Pattern = "a=%{NOTSPACE:data}"
-					parseField := entry.NewBodyField("logger_name_field")
-					loggerNameParser := helper.NewScopeNameParser()
+					parseField := signozstanzaentry.Field{signozstanzaentry.NewBodyField("logger_name_field")}
+					loggerNameParser := signozstanzahelper.NewScopeNameParser()
 					loggerNameParser.ParseFrom = parseField
 					cfg.ScopeNameParser = &loggerNameParser
 					return cfg
@@ -119,23 +120,23 @@ func TestParserGoldenConfig(t *testing.T) {
 				Name: "parse_to_attributes",
 				Expect: func() *Config {
 					p := NewConfig()
-					p.ParseTo = entry.RootableField{Field: entry.NewAttributeField()}
+					p.ParseTo = signozstanzaentry.RootableField{Field: signozstanzaentry.Field{entry.NewAttributeField()}}
 					return p
 				}(),
 			},
 			{
 				Name: "parse_to_body",
 				Expect: func() *Config {
 					p := NewConfig()
-					p.ParseTo = entry.RootableField{Field: entry.NewBodyField()}
+					p.ParseTo = signozstanzaentry.RootableField{Field: signozstanzaentry.Field{signozstanzaentry.NewBodyField()}}
 					return p
 				}(),
 			},
 			{
 				Name: "parse_to_resource",
 				Expect: func() *Config {
 					p := NewConfig()
-					p.ParseTo = entry.RootableField{Field: entry.NewResourceField()}
+					p.ParseTo = signozstanzaentry.RootableField{Field: signozstanzaentry.Field{entry.NewResourceField()}}
 					return p
 				}(),
 			},

pkg/parser/grok/grok.go

@@ -5,9 +5,9 @@ import (
 	"fmt"
 
 	signozlogspipelinestanzaoperator "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator"
+	signozstanzahelper "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/helper"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 	"github.com/vjeantet/grok"
 	"go.opentelemetry.io/collector/component"
 )
@@ -27,13 +27,13 @@ func NewConfig() *Config {
 // NewConfigWithID creates a new grok parser config with default values
 func NewConfigWithID(operatorID string) *Config {
 	return &Config{
-		ParserConfig: helper.NewParserConfig(operatorID, operatorType),
+		ParserConfig: signozstanzahelper.NewParserConfig(operatorID, operatorType),
 	}
 }
 
 // Config is the configuration of a grok parser operator.
 type Config struct {
-	helper.ParserConfig `mapstructure:",squash"`
+	signozstanzahelper.ParserConfig `mapstructure:",squash"`
 
 	// grok pattern
 	Pattern string `mapstructure:"pattern"`
@@ -102,7 +102,7 @@ func (c Config) Build(set component.TelemetrySettings) (operator.Operator, error
 
 // Parser is an operator that parses grok in an entry.
 type Parser struct {
-	helper.ParserOperator
+	signozstanzahelper.ParserOperator
 	grok               *grok.Grok
 	pattern            string
 	cache              cache

processor/signozlogspipelineprocessor/config_test.go

@@ -11,10 +11,11 @@ import (
 	"go.opentelemetry.io/collector/confmap/confmaptest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/entry"
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza/operator/helper"
 
 	signozlogspipelinestanzaadapter "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/adapter"
+	signozstanzaentry "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/entry"
 	signozlogspipelinestanzaoperator "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator"
+	signozstanzahelper "github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/helper"
 	"github.com/SigNoz/signoz-otel-collector/processor/signozlogspipelineprocessor/stanza/operator/operators/regex"
 )
 
@@ -31,12 +32,12 @@ func TestLoadConfig(t *testing.T) {
 					Builder: func() *regex.Config {
 						cfg := regex.NewConfig()
 						cfg.Regex = "^(?P<time>\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (?P<sev>[A-Z]*) (?P<msg>.*)$"
-						sevField := entry.NewAttributeField("sev")
-						sevCfg := helper.NewSeverityConfig()
+						sevField := signozstanzaentry.Field{entry.NewAttributeField("sev")}
+						sevCfg := signozstanzahelper.NewSeverityConfig()
 						sevCfg.ParseFrom = &sevField
 						cfg.SeverityConfig = &sevCfg
-						timeField := entry.NewAttributeField("time")
-						timeCfg := helper.NewTimeParser()
+						timeField := signozstanzaentry.Field{entry.NewAttributeField("time")}
+						timeCfg := signozstanzahelper.NewTimeParser()
 						timeCfg.Layout = "%Y-%m-%d %H:%M:%S"
 						timeCfg.ParseFrom = &timeField
 						cfg.TimeParser = &timeCfg