[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 processing based on source.address resource attribute
+
+# 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,8 @@
 [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 address. To add geographical information, the IP address must be included in the resource attributes using the [`source.address` semantic conventions key attribute](https://github.com/open-telemetry/semantic-conventions/blob/v1.26.0/docs/general/attributes.md#source).

processor/geoipprocessor/config_test.go

@@ -19,8 +19,9 @@ 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),
@@ -40,11 +41,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

@@ -10,11 +10,20 @@ import (
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/processor/processorhelper"
+	"go.opentelemetry.io/otel/attribute"
+	semconv "go.opentelemetry.io/otel/semconv/v1.25.0"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/metadata"
 )
 
-var processorCapabilities = consumer.Capabilities{MutatesData: true}
+var (
+	processorCapabilities = consumer.Capabilities{MutatesData: true}
+	// defaultResourceAttributes holds a list of default resource attribute keys.
+	// These keys are used to identify an IP address attribute associated with the resource.
+	defaultResourceAttributes = []attribute.Key{
+		semconv.SourceAddressKey, // This key represents the standard source address attribute as defined in the OpenTelemetry semantic conventions.
+	}
+)
 
 // NewFactory creates a new processor factory with default configuration,
 // and registers the processors for metrics, traces, and logs.
@@ -28,13 +37,13 @@ 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))
+	return processorhelper.NewMetricsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(defaultResourceAttributes).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))
+	return processorhelper.NewTracesProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(defaultResourceAttributes).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))
+	return processorhelper.NewLogsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(defaultResourceAttributes).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"
+	"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
+	resourceAttributes []attribute.Key
+}
+
+func newGeoIPProcessor(resourceAttributes []attribute.Key) *geoIPProcessor {
+	return &geoIPProcessor{
+		resourceAttributes: resourceAttributes,
+	}
+}
+
+// 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(attributes []attribute.Key, resource pcommon.Resource) (net.IP, error) {
+	for _, attr := range attributes {
+		if ipField, found := resource.Attributes().Get(string(attr)); found {
+			ipAttribute := net.ParseIP(ipField.AsString())
+			// The attribute might contain a domain name. Skip any net.ParseIP error until we have a fine-grained error propagation strategy.
+			// TODO: propagate an error once error_mode configuration option is available (e.g. transformprocessor)
+			if ipAttribute != nil {
+				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.resourceAttributes, 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
 }