[processor/geoip] Initial implementation

.chloggen/geoipprocessor_add_field.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: 'enhancement'
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: geoipprocessor
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add initial attributes processing based on fields configuration
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32663]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: []

processor/geoipprocessor/README.md

@@ -11,6 +11,25 @@
 [development]: https://github.com/open-telemetry/opentelemetry-collector#development
 <!-- end autogenerated section -->
 
-## Overview
+**This processor is currently under development and is presently a NOP (No Operation) processor. Further features and functionalities will be added in upcoming versions.**
 
-This processor is currently under development and is presently a **NOP (No Operation) processor**. Further features and functionalities will be added in upcoming versions.
+## Description
+
+The geoIP processor `geoipprocessor` enhances resource attributes by appending information about the geographical location of an IP addresses. The IP address must be included in the resource attributes for the geographical information to be added accordingly.
+
+## Configuration
+
+The following settings must configured:
+
+- `fields`: A list of resource attribute keys where the IP address will be searched for. The first matched key will be used to append the geographical location.
+
+#### Examples
+
+```yaml
+processors:
+    # processor name: geoip
+    geoip:
+        fields:
+          - host.ip
+          - ip
+```

processor/geoipprocessor/config.go

@@ -3,9 +3,21 @@
 
 package geoipprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor"
 
+import "errors"
+
 // Config holds the configuration for the GeoIP processor.
-type Config struct{}
+type Config struct {
+	// Fields defines a list of resource attributes keys to look for the IP value.
+	// The first matching attribute will be used to gather the IP geographical metadata.
+	// Example:
+	//   fields:
+	//     - client.ip
+	Fields []string `mapstructure:"fields"`
+}
 
 func (cfg *Config) Validate() error {
+	if len(cfg.Fields) < 1 {
+		return errors.New("must specify at least a field to look for the IP when using the geoip processor")
+	}
 	return nil
 }

processor/geoipprocessor/config_test.go

@@ -19,12 +19,18 @@ func TestLoadConfig(t *testing.T) {
 	t.Parallel()
 
 	tests := []struct {
-		id       component.ID
-		expected component.Config
+		id           component.ID
+		expected     component.Config
+		errorMessage string
 	}{
 		{
 			id:       component.NewID(metadata.Type),
-			expected: &Config{},
+			expected: &Config{Fields: []string{"host.ip"}},
+		},
+		{
+			id:           component.NewIDWithName(metadata.Type, "no_fields"),
+			expected:     &Config{Fields: []string{"host.ip"}},
+			errorMessage: "must specify at least a field to look for the IP when using the geoip processor",
 		},
 	}
 
@@ -40,11 +46,11 @@ func TestLoadConfig(t *testing.T) {
 			require.NoError(t, err)
 			require.NoError(t, component.UnmarshalConfig(sub, cfg))
 
-			if tt.expected == nil {
-				err = component.ValidateConfig(cfg)
-				assert.Error(t, err)
+			if tt.errorMessage != "" {
+				assert.EqualError(t, component.ValidateConfig(cfg), tt.errorMessage)
 				return
 			}
+
 			assert.NoError(t, component.ValidateConfig(cfg))
 			assert.Equal(t, tt.expected, cfg)
 		})

processor/geoipprocessor/factory.go

@@ -28,13 +28,16 @@ func createDefaultConfig() component.Config {
 }
 
 func createMetricsProcessor(ctx context.Context, set processor.CreateSettings, cfg component.Config, nextConsumer consumer.Metrics) (processor.Metrics, error) {
-	return processorhelper.NewMetricsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor().processMetrics, processorhelper.WithCapabilities(processorCapabilities))
+	geoCfg := cfg.(*Config)
+	return processorhelper.NewMetricsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg.Fields).processMetrics, processorhelper.WithCapabilities(processorCapabilities))
 }
 
 func createTracesProcessor(ctx context.Context, set processor.CreateSettings, cfg component.Config, nextConsumer consumer.Traces) (processor.Traces, error) {
-	return processorhelper.NewTracesProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor().processTraces, processorhelper.WithCapabilities(processorCapabilities))
+	geoCfg := cfg.(*Config)
+	return processorhelper.NewTracesProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg.Fields).processTraces, processorhelper.WithCapabilities(processorCapabilities))
 }
 
 func createLogsProcessor(ctx context.Context, set processor.CreateSettings, cfg component.Config, nextConsumer consumer.Logs) (processor.Logs, error) {
-	return processorhelper.NewLogsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor().processLogs, processorhelper.WithCapabilities(processorCapabilities))
+	geoCfg := cfg.(*Config)
+	return processorhelper.NewLogsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg.Fields).processLogs, processorhelper.WithCapabilities(processorCapabilities))
 }

processor/geoipprocessor/geoip_processor.go

@@ -5,26 +5,116 @@ package geoipprocessor // import "github.com/open-telemetry/opentelemetry-collec
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"net"
 
+	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+	"go.opentelemetry.io/otel/attribute"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/provider"
 )
 
-type geoIPProcessor struct{}
+var errIPNotFound = errors.New("no IP address found in the resource attributes")
+
+// newGeoIPProcessor creates a new instance of geoIPProcessor with the specified fields.
+type geoIPProcessor struct {
+	providers []provider.GeoIPProvider
+	fields    []string
+}
+
+func newGeoIPProcessor(fields []string) *geoIPProcessor {
+	return &geoIPProcessor{
+		fields: fields,
+	}
+}
+
+// ipFromResourceAttributes extracts an IP address from the given resource's attributes based on the specified fields.
+// It returns the first IP address if found, or an error if no valid IP address is found.
+func ipFromResourceAttributes(fields []string, resource pcommon.Resource) (net.IP, error) {
+	for _, field := range fields {
+		if ipField, found := resource.Attributes().Get(field); found {
+			ipAttribute := net.ParseIP(ipField.AsString())
+			if ipAttribute == nil {
+				return nil, fmt.Errorf("could not parse ip address %s", ipField.AsString())
+			}
+			return ipAttribute, nil
+		}
+	}
+
+	return nil, errIPNotFound
+}
+
+// geoLocation fetches geolocation information for the given IP address using the configured providers.
+// It returns a set of attributes containing the geolocation data, or an error if the location could not be determined.
+func (g *geoIPProcessor) geoLocation(ctx context.Context, ip net.IP) (attribute.Set, error) {
+	allAttributes := attribute.EmptySet()
+	for _, provider := range g.providers {
+		geoAttributes, err := provider.Location(ctx, ip)
+		if err != nil {
+			return attribute.Set{}, err
+		}
+		*allAttributes = attribute.NewSet(append(allAttributes.ToSlice(), geoAttributes.ToSlice()...)...)
+	}
+
+	return *allAttributes, nil
+}
+
+// processResource processes a single resource by adding geolocation attributes based on the found IP address.
+func (g *geoIPProcessor) processResource(ctx context.Context, resource pcommon.Resource) error {
+	ipAddr, err := ipFromResourceAttributes(g.fields, resource)
+	if err != nil {
+		// TODO: log IP error not found
+		if errors.Is(err, errIPNotFound) {
+			return nil
+		}
+		return err
+	}
+
+	attributes, err := g.geoLocation(ctx, ipAddr)
+	if err != nil {
+		return err
+	}
+
+	for _, geoAttr := range attributes.ToSlice() {
+		resource.Attributes().PutStr(string(geoAttr.Key), geoAttr.Value.AsString())
+	}
 
-func newGeoIPProcessor() *geoIPProcessor {
-	return &geoIPProcessor{}
+	return nil
 }
 
-func (g *geoIPProcessor) processMetrics(_ context.Context, ms pmetric.Metrics) (pmetric.Metrics, error) {
+func (g *geoIPProcessor) processMetrics(ctx context.Context, ms pmetric.Metrics) (pmetric.Metrics, error) {
+	rm := ms.ResourceMetrics()
+	for i := 0; i < rm.Len(); i++ {
+		err := g.processResource(ctx, rm.At(i).Resource())
+		if err != nil {
+			return ms, err
+		}
+	}
 	return ms, nil
 }
 
-func (g *geoIPProcessor) processTraces(_ context.Context, ts ptrace.Traces) (ptrace.Traces, error) {
+func (g *geoIPProcessor) processTraces(ctx context.Context, ts ptrace.Traces) (ptrace.Traces, error) {
+	rt := ts.ResourceSpans()
+	for i := 0; i < rt.Len(); i++ {
+		err := g.processResource(ctx, rt.At(i).Resource())
+		if err != nil {
+			return ts, err
+		}
+	}
 	return ts, nil
 }
 
-func (g *geoIPProcessor) processLogs(_ context.Context, ls plog.Logs) (plog.Logs, error) {
+func (g *geoIPProcessor) processLogs(ctx context.Context, ls plog.Logs) (plog.Logs, error) {
+	rl := ls.ResourceLogs()
+	for i := 0; i < rl.Len(); i++ {
+		err := g.processResource(ctx, rl.At(i).Resource())
+		if err != nil {
+			return ls, err
+		}
+	}
 	return ls, nil
 }