[processor/geoip] Initial implementation (#33319)

**Description:** <Describe what has changed.>
This is the initial implementation of the GeoIP processor. Note that it
will still be a NOP (No Operation) from the user's perspective, as no
Geo data providers are implemented [nor can be configured
yet](#33268).
However, this implementation aims to provide an overview of the
processor's logic.

The logic is the same for all signals (traces, metrics, and logs). For
each signal, the processor follows these steps:

1. Locate an IP address attribute within the resource attributes. If not
found, continue; if parsing the IP fails, return an error. By default,
the IP address must be included within the `source.address` attribute
key, see:
https://github.com/open-telemetry/semantic-conventions/blob/v1.26.0/docs/general/attributes.md#source
2. Retrieve the geo-location metadata associated with the IP address
using the available GeoProviders. At the moment, no additional check is
done on the attributes returned by the providers. In a future PR, the
processor will need to validate that all attributes follow the Geo
semantic conventions.
 3. Append the set of geo attributes to the resource attributes. 

The `processor/geoipprocessor/internal/provider/geoipprovider.go` file
contains an initial interface proposal for the Geo providers. A mock
implementation can be found in the geoip_processor_test.go file.

**Link to tracking Issue:**
#32663

**Testing:** Unit tests for the signals process

**Documentation:** Should we update the README with the available
configuration once it becomes operational?

---------

Co-authored-by: Tiffany Hrabusa <[email protected]>

.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
 }