[processor/attributesprocessor] Add a convert action

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### 💡 Enhancements 💡
 
+- `attributesprocessor`: Add convert action (#7930)
 - `attributesprocessor`: Add metric support (#8111)
 - `prometheusremotewriteexporter`: Write-Ahead Log support enabled (#7304)
 - `hostreceiver/filesystemscraper`: Add filesystem utilization (#8027)

internal/coreinternal/attraction/attraction.go

@@ -22,14 +22,15 @@ import (
 
 	"go.opentelemetry.io/collector/client"
 	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterhelper"
 )
 
 // Settings specifies the processor settings.
 type Settings struct {
 	// Actions specifies the list of attributes to act on.
-	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT}.
+	// The set of actions are {INSERT, UPDATE, UPSERT, DELETE, HASH, EXTRACT, CONVERT}.
 	// This is a required field.
 	Actions []ActionKeyValue `mapstructure:"actions"`
 }
@@ -64,6 +65,11 @@ type ActionKeyValue struct {
 	// If the key has multiple values the values will be joined with `;` separator.
 	FromContext string `mapstructure:"from_context"`
 
+	// ConvertedType specifies the target type of an attribute to be converted
+	// If the key doesn't exist, no action is performed.
+	// If the value cannot be converted, the original value will be left as-is
+	ConvertedType string `mapstructure:"converted_type"`
+
 	// Action specifies the type of action to perform.
 	// The set of values are {INSERT, UPDATE, UPSERT, DELETE, HASH}.
 	// Both lower case and upper case are supported.
@@ -85,6 +91,7 @@ type ActionKeyValue struct {
 	// EXTRACT - Extracts values using a regular expression rule from the input
 	//           'key' to target keys specified in the 'rule'. If a target key
 	//           already exists, it will be overridden.
+	// CONVERT  - converts the type of an existing attribute, if convertable
 	// This is a required field.
 	Action Action `mapstructure:"action"`
 }
@@ -134,12 +141,16 @@ const (
 	// 'key' to target keys specified in the 'rule'. If a target key already
 	// exists, it will be overridden.
 	EXTRACT Action = "extract"
+
+	// CONVERT converts the type of an existing attribute, if convertable
+	CONVERT Action = "convert"
 )
 
 type attributeAction struct {
 	Key           string
 	FromAttribute string
 	FromContext   string
+	ConvertedType string
 	// Compiled regex if provided
 	Regex *regexp.Regexp
 	// Attribute names extracted from the regexp's subexpressions.
@@ -190,7 +201,9 @@ func NewAttrProc(settings *Settings) (*AttrProc, error) {
 			}
 			if a.RegexPattern != "" {
 				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use the \"pattern\" field. This must not be specified for %d-th action", a.Action, i)
-
+			}
+			if a.ConvertedType != "" {
+				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use the \"converted_type\" field. This must not be specified for %d-th action", a.Action, i)
 			}
 			// Convert the raw value from the configuration to the internal trace representation of the value.
 			if a.Value != nil {
@@ -207,13 +220,18 @@ func NewAttrProc(settings *Settings) (*AttrProc, error) {
 			if valueSourceCount > 0 || a.RegexPattern != "" {
 				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use value sources or \"pattern\" field. These must not be specified for %d-th action", a.Action, i)
 			}
+			if a.ConvertedType != "" {
+				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use the \"converted_type\" field. This must not be specified for %d-th action", a.Action, i)
+			}
 		case EXTRACT:
 			if valueSourceCount > 0 {
 				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use a value source field. These must not be specified for %d-th action", a.Action, i)
 			}
 			if a.RegexPattern == "" {
 				return nil, fmt.Errorf("error creating AttrProc due to missing required field \"pattern\" for action \"%s\" at the %d-th action", a.Action, i)
-
+			}
+			if a.ConvertedType != "" {
+				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use the \"converted_type\" field. This must not be specified for %d-th action", a.Action, i)
 			}
 			re, err := regexp.Compile(a.RegexPattern)
 			if err != nil {
@@ -231,6 +249,20 @@ func NewAttrProc(settings *Settings) (*AttrProc, error) {
 			}
 			action.Regex = re
 			action.AttrNames = attrNames
+		case CONVERT:
+			if valueSourceCount > 0 || a.RegexPattern != "" {
+				return nil, fmt.Errorf("error creating AttrProc. Action \"%s\" does not use value sources or \"pattern\" field. These must not be specified for %d-th action", a.Action, i)
+			}
+			switch a.ConvertedType {
+			case stringConversionTarget:
+			case intConversionTarget:
+			case doubleConversionTarget:
+			case "":
+				return nil, fmt.Errorf("error creating AttrProc due to missing required field \"converted_type\" for action \"%s\" at the %d-th action", a.Action, i)
+			default:
+				return nil, fmt.Errorf("error creating AttrProc due to invalid value \"%s\" in field \"converted_type\" for action \"%s\" at the %d-th action", a.ConvertedType, a.Action, i)
+			}
+			action.ConvertedType = a.ConvertedType
 		default:
 			return nil, fmt.Errorf("error creating AttrProc due to unsupported action %q at the %d-th actions", a.Action, i)
 		}
@@ -241,7 +273,7 @@ func NewAttrProc(settings *Settings) (*AttrProc, error) {
 }
 
 // Process applies the AttrProc to an attribute map.
-func (ap *AttrProc) Process(ctx context.Context, attrs pdata.AttributeMap) {
+func (ap *AttrProc) Process(ctx context.Context, logger *zap.Logger, attrs pdata.AttributeMap) {
 	for _, action := range ap.actions {
 		// TODO https://go.opentelemetry.io/collector/issues/296
 		// Do benchmark testing between having action be of type string vs integer.
@@ -272,6 +304,8 @@ func (ap *AttrProc) Process(ctx context.Context, attrs pdata.AttributeMap) {
 			hashAttribute(action, attrs)
 		case EXTRACT:
 			extractAttributes(action, attrs)
+		case CONVERT:
+			convertAttribute(logger, action, attrs)
 		}
 	}
 }
@@ -306,6 +340,12 @@ func hashAttribute(action attributeAction, attrs pdata.AttributeMap) {
 	}
 }
 
+func convertAttribute(logger *zap.Logger, action attributeAction, attrs pdata.AttributeMap) {
+	if value, exists := attrs.Get(action.Key); exists {
+		convertValue(logger, action.Key, action.ConvertedType, value)
+	}
+}
+
 func extractAttributes(action attributeAction, attrs pdata.AttributeMap) {
 	value, found := attrs.Get(action.Key)
 

internal/coreinternal/attraction/attraction_test.go

@@ -41,7 +41,7 @@ type testCase struct {
 func runIndividualTestCase(t *testing.T, tt testCase, ap *AttrProc) {
 	t.Run(tt.name, func(t *testing.T) {
 		attrMap := pdata.NewAttributeMapFromMap(tt.inputAttributes)
-		ap.Process(context.TODO(), attrMap)
+		ap.Process(context.TODO(), nil, attrMap)
 		attrMap.Sort()
 		require.Equal(t, pdata.NewAttributeMapFromMap(tt.expectedAttributes).Sort(), attrMap)
 	})
@@ -930,7 +930,7 @@ func TestFromContext(t *testing.T) {
 			require.Nil(t, err)
 			require.NotNil(t, ap)
 			attrMap := pdata.NewAttributeMap()
-			ap.Process(tc.ctx, attrMap)
+			ap.Process(tc.ctx, nil, attrMap)
 			attrMap.Sort()
 			require.Equal(t, pdata.NewAttributeMapFromMap(tc.expectedAttributes).Sort(), attrMap)
 		})

internal/coreinternal/attraction/type_converter.go

@@ -0,0 +1,84 @@
+// Copyright  The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package attraction // import "github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/attraction"
+
+import (
+	"strconv"
+
+	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
+)
+
+const (
+	stringConversionTarget = "string"
+	intConversionTarget    = "int"
+	doubleConversionTarget = "double"
+)
+
+func convertValue(logger *zap.Logger, key string, to string, v pdata.AttributeValue) {
+	switch to {
+	case stringConversionTarget:
+		switch v.Type() {
+		case pdata.AttributeValueTypeString:
+		default:
+			v.SetStringVal(v.AsString())
+		}
+	case intConversionTarget:
+		switch v.Type() {
+		case pdata.AttributeValueTypeInt:
+		case pdata.AttributeValueTypeDouble:
+			v.SetIntVal(int64(v.DoubleVal()))
+		case pdata.AttributeValueTypeBool:
+			if v.BoolVal() {
+				v.SetIntVal(1)
+			} else {
+				v.SetIntVal(0)
+			}
+		case pdata.AttributeValueTypeString:
+			s := v.StringVal()
+			n, err := strconv.ParseInt(s, 10, 64)
+			if err == nil {
+				v.SetIntVal(n)
+			} else {
+				logger.Debug("String could not be converted to int", zap.String("key", key), zap.String("value", s), zap.Error(err))
+			}
+		default:
+			logger.Debug("Unable to convert type", zap.String("key", key), zap.String("from", v.Type().String()), zap.String("to", intConversionTarget))
+		}
+	case doubleConversionTarget:
+		switch v.Type() {
+		case pdata.AttributeValueTypeInt:
+			v.SetDoubleVal(float64(v.IntVal()))
+		case pdata.AttributeValueTypeDouble:
+		case pdata.AttributeValueTypeBool:
+			if v.BoolVal() {
+				v.SetDoubleVal(1)
+			} else {
+				v.SetDoubleVal(0)
+			}
+		case pdata.AttributeValueTypeString:
+			s := v.StringVal()
+			n, err := strconv.ParseFloat(s, 64)
+			if err == nil {
+				v.SetDoubleVal(n)
+			} else {
+				logger.Debug("String could not be converted to double", zap.String("key", key), zap.String("value", s), zap.Error(err))
+			}
+		default:
+			logger.Debug("Unable to convert type", zap.String("key", key), zap.String("from", v.Type().String()), zap.String("to", doubleConversionTarget))
+		}
+	default: // No-op
+	}
+}

processor/attributesprocessor/README.md

@@ -21,6 +21,7 @@ The supported actions are:
   to target keys specified in the rule. If a target key already exists, it will
   be overridden. Note: It behaves similar to the Span Processor `to_attributes`
   setting with the existing attribute as the source.
+- `convert`: Converts an existing attribute to a specified type.
 
 For the actions `insert`, `update` and `upsert`,
  - `key`  is required
@@ -88,6 +89,18 @@ For the `extract` action,
 
  ```
 
+
+For the `convert` action,
+ - `key` is required
+ - `action: convert` is required.
+ - `converted_type` is required and must be one of int, double or string
+```yaml
+# Key specifies the attribute to act upon.
+- key: <key>
+  action: convert
+  converted_type: <int|double|string>
+```
+
 The list of actions can be composed to create rich scenarios, such as
 back filling attribute, copying values to a new key, redacting sensitive information.
 The following is a sample configuration.
@@ -110,6 +123,9 @@ processors:
         action: delete
       - key: account_email
         action: hash
+      - key: http.status_code
+        action: convert
+        converted_type: int
 
 ```
 

processor/attributesprocessor/attributes_log.go

@@ -18,12 +18,14 @@ import (
 	"context"
 
 	"go.opentelemetry.io/collector/model/pdata"
+	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/attraction"
 	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal/processor/filterlog"
 )
 
 type logAttributesProcessor struct {
+	logger   *zap.Logger
 	attrProc *attraction.AttrProc
 	include  filterlog.Matcher
 	exclude  filterlog.Matcher
@@ -32,8 +34,9 @@ type logAttributesProcessor struct {
 // newLogAttributesProcessor returns a processor that modifies attributes of a
 // log record. To construct the attributes processors, the use of the factory
 // methods are required in order to validate the inputs.
-func newLogAttributesProcessor(attrProc *attraction.AttrProc, include, exclude filterlog.Matcher) *logAttributesProcessor {
+func newLogAttributesProcessor(logger *zap.Logger, attrProc *attraction.AttrProc, include, exclude filterlog.Matcher) *logAttributesProcessor {
 	return &logAttributesProcessor{
+		logger:   logger,
 		attrProc: attrProc,
 		include:  include,
 		exclude:  exclude,
@@ -56,7 +59,7 @@ func (a *logAttributesProcessor) processLogs(ctx context.Context, ld pdata.Logs)
 					continue
 				}
 
-				a.attrProc.Process(ctx, lr.Attributes())
+				a.attrProc.Process(ctx, a.logger, lr.Attributes())
 			}
 		}
 	}