[processor/transform] Add business logic for handling traces queries.

processor/transformprocessor/config.go

@@ -16,10 +16,15 @@ package transformprocessor // import "github.com/open-telemetry/opentelemetry-co
 
 import (
 	"go.opentelemetry.io/collector/config"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
 type TracesConfig struct {
 	Queries []string `mapstructure:"queries"`
+
+	// The functions that have been registered in the extension for traces processing.
+	functions map[string]interface{} `mapstructure:"-"`
 }
 
 type Config struct {
@@ -29,3 +34,8 @@ type Config struct {
 }
 
 var _ config.Processor = (*Config)(nil)
+
+func (c *Config) Validate() error {
+	_, err := traces.Parse(c.Traces.Queries, c.Traces.functions)
+	return err
+}

processor/transformprocessor/config_test.go

@@ -23,6 +23,8 @@ import (
 	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/service/servicetest"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
 func TestLoadingConfig(t *testing.T) {
@@ -41,8 +43,26 @@ func TestLoadingConfig(t *testing.T) {
 		Traces: TracesConfig{
 			Queries: []string{
 				`set(name, "bear") where attributes["http.path"] == "/animal"`,
-				`keep(attributes, "http.method", "http.path")`,
+				`keep_keys(attributes, "http.method", "http.path")`,
 			},
+
+			functions: traces.DefaultFunctions(),
 		},
 	})
 }
+
+func TestLoadInvalidConfig(t *testing.T) {
+	factories, err := componenttest.NopFactories()
+	assert.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Processors[typeStr] = factory
+
+	cfg, err := servicetest.LoadConfigAndValidate(filepath.Join("testdata", "invalid_config_bad_syntax.yaml"), factories)
+	assert.Error(t, err)
+	assert.NotNil(t, cfg)
+
+	cfg, err = servicetest.LoadConfigAndValidate(filepath.Join("testdata", "invalid_config_unknown_function.yaml"), factories)
+	assert.Error(t, err)
+	assert.NotNil(t, cfg)
+}

processor/transformprocessor/factory.go

@@ -16,12 +16,14 @@ package transformprocessor // import "github.com/open-telemetry/opentelemetry-co
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/consumer"
-	"go.opentelemetry.io/collector/model/pdata"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
 const (
@@ -43,22 +45,27 @@ func createDefaultConfig() config.Processor {
 		ProcessorSettings: config.NewProcessorSettings(config.NewComponentID(typeStr)),
 		Traces: TracesConfig{
 			Queries: []string{},
+
+			functions: traces.DefaultFunctions(),
 		},
 	}
 }
 
 func createTracesProcessor(
 	_ context.Context,
-	_ component.ProcessorCreateSettings,
+	settings component.ProcessorCreateSettings,
 	cfg config.Processor,
 	nextConsumer consumer.Traces,
 ) (component.TracesProcessor, error) {
+	oCfg := cfg.(*Config)
+
+	proc, err := traces.NewProcessor(oCfg.Traces.Queries, oCfg.Traces.functions, settings)
+	if err != nil {
+		return nil, fmt.Errorf("invalid config for \"transform\" processor %w", err)
+	}
 	return processorhelper.NewTracesProcessor(
 		cfg,
 		nextConsumer,
-		// TODO(anuraaga): Replace with business logic.
-		func(ctx context.Context, p pdata.Traces) (pdata.Traces, error) {
-			return p, nil
-		},
+		proc.ProcessTraces,
 		processorhelper.WithCapabilities(processorCapabilities))
 }

processor/transformprocessor/factory_test.go

@@ -15,11 +15,17 @@
 package transformprocessor
 
 import (
+	"context"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/config"
 	"go.opentelemetry.io/collector/config/configtest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/model/pdata"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/transformprocessor/internal/traces"
 )
 
 func TestFactory_Type(t *testing.T) {
@@ -34,6 +40,8 @@ func TestFactory_CreateDefaultConfig(t *testing.T) {
 		ProcessorSettings: config.NewProcessorSettings(config.NewComponentID(typeStr)),
 		Traces: TracesConfig{
 			Queries: []string{},
+
+			functions: traces.DefaultFunctions(),
 		},
 	})
 	assert.NoError(t, configtest.CheckConfigStruct(cfg))
@@ -45,3 +53,38 @@ func TestFactoryCreateTracesProcessor_Empty(t *testing.T) {
 	err := cfg.Validate()
 	assert.NoError(t, err)
 }
+
+func TestFactoryCreateTracesProcessor_InvalidActions(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	oCfg := cfg.(*Config)
+	oCfg.Traces.Queries = []string{`set(123`}
+	ap, err := factory.CreateTracesProcessor(context.Background(), componenttest.NewNopProcessorCreateSettings(), cfg, consumertest.NewNop())
+	assert.Error(t, err)
+	assert.Nil(t, ap)
+}
+
+func TestFactoryCreateTracesProcessor(t *testing.T) {
+	factory := NewFactory()
+	cfg := factory.CreateDefaultConfig()
+	oCfg := cfg.(*Config)
+	oCfg.Traces.Queries = []string{`set(attributes["test"], "pass") where name == "operationA"`}
+
+	tp, err := factory.CreateTracesProcessor(context.Background(), componenttest.NewNopProcessorCreateSettings(), cfg, consumertest.NewNop())
+	assert.NotNil(t, tp)
+	assert.NoError(t, err)
+
+	td := pdata.NewTraces()
+	span := td.ResourceSpans().AppendEmpty().InstrumentationLibrarySpans().AppendEmpty().Spans().AppendEmpty()
+	span.SetName("operationA")
+
+	_, ok := span.Attributes().Get("test")
+	assert.False(t, ok)
+
+	err = tp.ConsumeTraces(context.Background(), td)
+	assert.NoError(t, err)
+
+	val, ok := span.Attributes().Get("test")
+	assert.True(t, ok)
+	assert.Equal(t, "pass", val.StringVal())
+}

processor/transformprocessor/go.mod

@@ -7,6 +7,8 @@ require (
 	github.com/stretchr/testify v1.7.0
 	go.opentelemetry.io/collector v0.44.1-0.20220210184720-ea897a6906a5
 	go.opentelemetry.io/collector/model v0.44.1-0.20220210184720-ea897a6906a5
+	go.uber.org/multierr v1.7.0
+	go.uber.org/zap v1.21.0
 )
 
 require (
@@ -28,8 +30,6 @@ require (
 	go.opentelemetry.io/otel/metric v0.27.0 // indirect
 	go.opentelemetry.io/otel/trace v1.4.0 // indirect
 	go.uber.org/atomic v1.9.0 // indirect
-	go.uber.org/multierr v1.7.0 // indirect
-	go.uber.org/zap v1.21.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect

processor/transformprocessor/internal/common/parser.go

@@ -19,9 +19,9 @@ import (
 	"github.com/alecthomas/participle/v2/lexer"
 )
 
-// Query represents a parsed query. It is the entry point into the query DSL.
+// ParsedQuery represents a parsed query. It is the entry point into the query DSL.
 // nolint:govet
-type Query struct {
+type ParsedQuery struct {
 	Invocation Invocation `@@`
 	Condition  *Condition `( "where" @@ )?`
 }
@@ -65,29 +65,20 @@ type Field struct {
 	MapKey *string `( "[" @String "]" )?`
 }
 
-func Parse(rawQueries []string) ([]Query, error) {
-	parser, err := newParser()
-	if err != nil {
-		return []Query{}, err
-	}
+var parser = newParser()
 
-	parsed := make([]Query, 0)
-
-	for _, raw := range rawQueries {
-		query := Query{}
-		err = parser.ParseString("", raw, &query)
-		if err != nil {
-			return []Query{}, err
-		}
-		parsed = append(parsed, query)
+func Parse(raw string) (*ParsedQuery, error) {
+	parsed := &ParsedQuery{}
+	err := parser.ParseString("", raw, parsed)
+	if err != nil {
+		return nil, err
 	}
-
 	return parsed, nil
 }
 
-// newParser returns a parser that can be used to read a string into a Query. An error will be returned if the string
+// newParser returns a parser that can be used to read a string into a ParsedQuery. An error will be returned if the string
 // is not formatted for the DSL.
-func newParser() (*participle.Parser, error) {
+func newParser() *participle.Parser {
 	lex := lexer.MustSimple([]lexer.Rule{
 		{Name: `Ident`, Pattern: `[a-zA-Z_][a-zA-Z0-9_]*`, Action: nil},
 		{Name: `Float`, Pattern: `[-+]?\d*\.\d+([eE][-+]?\d+)?`, Action: nil},
@@ -96,9 +87,13 @@ func newParser() (*participle.Parser, error) {
 		{Name: `Operators`, Pattern: `==|!=|[,.()\[\]]`, Action: nil},
 		{Name: "whitespace", Pattern: `\s+`, Action: nil},
 	})
-	return participle.Build(&Query{},
+	parser, err := participle.Build(&ParsedQuery{},
 		participle.Lexer(lex),
 		participle.Unquote("String"),
 		participle.Elide("whitespace"),
 	)
+	if err != nil {
+		panic("Unable to initialize parser, this is a programming error in the transformprocesor")
+	}
+	return parser
 }

processor/transformprocessor/internal/common/parser_test.go

@@ -23,11 +23,11 @@ import (
 func Test_parse(t *testing.T) {
 	tests := []struct {
 		query    string
-		expected Query
+		expected *ParsedQuery
 	}{
 		{
 			query: `set("foo")`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -41,7 +41,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `met(1.2)`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "met",
 					Arguments: []Value{
@@ -55,7 +55,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `fff(12)`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "fff",
 					Arguments: []Value{
@@ -69,7 +69,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set("foo", get(bear.honey))`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -102,7 +102,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set(foo.attributes["bar"].cat, "dog")`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -132,7 +132,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set(foo.attributes["bar"].cat, "dog") where name == "fido"`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -176,7 +176,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set(foo.attributes["bar"].cat, "dog") where name != "fido"`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -220,7 +220,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set  ( foo.attributes[ "bar"].cat,   "dog")   where name=="fido"`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -264,7 +264,7 @@ func Test_parse(t *testing.T) {
 		},
 		{
 			query: `set("fo\"o")`,
-			expected: Query{
+			expected: &ParsedQuery{
 				Invocation: Invocation{
 					Function: "set",
 					Arguments: []Value{
@@ -280,10 +280,9 @@ func Test_parse(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.query, func(t *testing.T) {
-			parsed, err := Parse([]string{tt.query})
+			parsed, err := Parse(tt.query)
 			assert.NoError(t, err)
-			assert.Len(t, parsed, 1)
-			assert.Equal(t, tt.expected, parsed[0])
+			assert.Equal(t, tt.expected, parsed)
 		})
 	}
 }
@@ -298,7 +297,7 @@ func Test_parse_failure(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt, func(t *testing.T) {
-			_, err := Parse([]string{tt})
+			_, err := Parse(tt)
 			assert.Error(t, err)
 		})
 	}