[processor/geoip] Add GeoIP providers configuration and maxmind factory

.chloggen/geoipprocessor_provider.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 providers configuration and metadata retrival abstraction
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [33269]
+
+# (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/config.go

@@ -3,9 +3,77 @@
 
 package geoipprocessor // import "github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor"
 
+import (
+	"errors"
+	"fmt"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/confmap"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/provider"
+)
+
+const (
+	providerKey = "providers"
+)
+
 // Config holds the configuration for the GeoIP processor.
-type Config struct{}
+type Config struct {
+	// Providers specifies the sources to extract geographical information about a given IP.
+	Providers map[string]provider.Config `mapstructure:"-"`
+}
+
+var (
+	_ component.Config    = (*Config)(nil)
+	_ confmap.Unmarshaler = (*Config)(nil)
+)
 
 func (cfg *Config) Validate() error {
+	if len(cfg.Providers) == 0 {
+		return errors.New("must specify at least one geo IP data provider when using the geoip processor")
+	}
+	return nil
+}
+
+// Unmarshal a config.Parser into the config struct.
+func (cfg *Config) Unmarshal(componentParser *confmap.Conf) error {
+	if componentParser == nil {
+		return nil
+	}
+
+	// load the non-dynamic config normally
+	err := componentParser.Unmarshal(cfg, confmap.WithIgnoreUnused())
+	if err != nil {
+		return err
+	}
+
+	// dynamically load the individual providers configs based on the key name
+
+	cfg.Providers = map[string]provider.Config{}
+
+	providersSection, err := componentParser.Sub(providerKey)
+	if err != nil {
+		return err
+	}
+
+	for key := range providersSection.ToStringMap() {
+		factory, ok := getProviderFactory(key)
+		if !ok {
+			return fmt.Errorf("invalid provider key: %s", key)
+		}
+
+		providerCfg := factory.CreateDefaultConfig()
+		providerSection, err := providersSection.Sub(key)
+		if err != nil {
+			return err
+		}
+		err = providerSection.Unmarshal(providerCfg)
+		if err != nil {
+			return fmt.Errorf("error reading settings for provider type %q: %w", key, err)
+		}
+
+		cfg.Providers[key] = providerCfg
+	}
+
 	return nil
 }

processor/geoipprocessor/config_test.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/confmap/confmaptest"
+	"go.opentelemetry.io/collector/otelcol/otelcoltest"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/metadata"
 )
@@ -19,12 +20,13 @@ 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{},
+			id:           component.NewID(metadata.Type),
+			errorMessage: "must specify at least one geo IP data provider when using the geoip processor",
 		},
 	}
 
@@ -40,13 +42,24 @@ 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)
 		})
 	}
 }
+
+func TestLoadInvalidConfig_InvalidProviderKey(t *testing.T) {
+	factories, err := otelcoltest.NopFactories()
+	require.NoError(t, err)
+
+	factory := NewFactory()
+	factories.Processors[metadata.Type] = factory
+	_, err = otelcoltest.LoadConfigAndValidate(filepath.Join("testdata", "config-invalidProviderKey.yaml"), factories)
+
+	require.Contains(t, err.Error(), "error reading configuration for \"geoip\": invalid provider key: invalidProviderKey")
+}

processor/geoipprocessor/factory.go

@@ -5,36 +5,89 @@ package geoipprocessor // import "github.com/open-telemetry/opentelemetry-collec
 
 import (
 	"context"
+	"fmt"
 
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/collector/processor/processorhelper"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/provider"
 )
 
 var processorCapabilities = consumer.Capabilities{MutatesData: true}
 
+var providerFactories = map[string]provider.GeoIPProviderFactory{}
+
 // NewFactory creates a new processor factory with default configuration,
 // and registers the processors for metrics, traces, and logs.
 func NewFactory() processor.Factory {
 	return processor.NewFactory(metadata.Type, createDefaultConfig, processor.WithMetrics(createMetricsProcessor, metadata.MetricsStability), processor.WithLogs(createLogsProcessor, metadata.LogsStability), processor.WithTraces(createTracesProcessor, metadata.TracesStability))
 }
 
+func getProviderFactory(key string) (provider.GeoIPProviderFactory, bool) {
+	if factory, ok := providerFactories[key]; ok {
+		return factory, true
+	}
+
+	return nil, false
+}
+
 // createDefaultConfig returns a default configuration for the processor.
 func createDefaultConfig() component.Config {
 	return &Config{}
 }
 
+func createGeoIPProviders(
+	ctx context.Context,
+	set processor.CreateSettings,
+	config *Config,
+	factories map[string]provider.GeoIPProviderFactory,
+) ([]provider.GeoIPProvider, error) {
+	providers := make([]provider.GeoIPProvider, 0, len(config.Providers))
+
+	for key, cfg := range config.Providers {
+		factory := factories[key]
+		if factory == nil {
+			return nil, fmt.Errorf("geoIP provider factory not found for key: %q", key)
+		}
+
+		provider, err := factory.CreateGeoIPProvider(ctx, set, cfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to create provider for key %q: %w", key, err)
+		}
+
+		providers = append(providers, provider)
+
+	}
+
+	return providers, nil
+}
+
 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)
+	providers, err := createGeoIPProviders(ctx, set, geoCfg, providerFactories)
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewMetricsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(providers).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)
+	providers, err := createGeoIPProviders(ctx, set, geoCfg, providerFactories)
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewTracesProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(providers).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)
+	providers, err := createGeoIPProviders(ctx, set, geoCfg, providerFactories)
+	if err != nil {
+		return nil, err
+	}
+	return processorhelper.NewLogsProcessor(ctx, set, cfg, nextConsumer, newGeoIPProcessor(providers).processLogs, processorhelper.WithCapabilities(processorCapabilities))
 }

processor/geoipprocessor/geoip_processor.go

@@ -9,12 +9,21 @@ import (
 	"go.opentelemetry.io/collector/pdata/plog"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/provider"
 )
 
-type geoIPProcessor struct{}
+// GeoProviders will be used by the Processor retrieve geographical metadata given an IP address.
+type GeoProviders = []provider.GeoIPProvider
+
+type geoIPProcessor struct {
+	providers GeoProviders
+}
 
-func newGeoIPProcessor() *geoIPProcessor {
-	return &geoIPProcessor{}
+func newGeoIPProcessor(providers GeoProviders) *geoIPProcessor {
+	return &geoIPProcessor{
+		providers,
+	}
 }
 
 func (g *geoIPProcessor) processMetrics(_ context.Context, ms pmetric.Metrics) (pmetric.Metrics, error) {

processor/geoipprocessor/go.mod

@@ -7,47 +7,91 @@ require (
 	go.opentelemetry.io/collector/component v0.101.1-0.20240527192838-af4fdd4e342a
 	go.opentelemetry.io/collector/confmap v0.101.1-0.20240527192838-af4fdd4e342a
 	go.opentelemetry.io/collector/consumer v0.101.1-0.20240527192838-af4fdd4e342a
+	go.opentelemetry.io/collector/otelcol v0.101.0
 	go.opentelemetry.io/collector/pdata v1.8.1-0.20240527192838-af4fdd4e342a
 	go.opentelemetry.io/collector/processor v0.101.1-0.20240527192838-af4fdd4e342a
+	go.opentelemetry.io/otel v1.27.0
 	go.uber.org/goleak v1.3.0
 )
 
 require (
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
 	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/uuid v1.6.0 // indirect
+	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect
+	github.com/hashicorp/go-version v1.6.0 // indirect
+	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/knadh/koanf/maps v0.1.1 // indirect
 	github.com/knadh/koanf/providers/confmap v0.1.0 // indirect
 	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/prometheus/client_golang v1.19.1 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.53.0 // indirect
 	github.com/prometheus/procfs v0.15.0 // indirect
+	github.com/shirou/gopsutil/v3 v3.24.4 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/spf13/cobra v1.8.0 // indirect
+	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.4 // indirect
+	go.opencensus.io v0.24.0 // indirect
 	go.opentelemetry.io/collector v0.101.1-0.20240527192838-af4fdd4e342a // indirect
 	go.opentelemetry.io/collector/config/configtelemetry v0.101.1-0.20240527192838-af4fdd4e342a // indirect
+	go.opentelemetry.io/collector/confmap/converter/expandconverter v0.101.0 // indirect
+	go.opentelemetry.io/collector/confmap/provider/envprovider v0.101.0 // indirect
+	go.opentelemetry.io/collector/confmap/provider/fileprovider v0.101.0 // indirect
+	go.opentelemetry.io/collector/confmap/provider/httpprovider v0.101.0 // indirect
+	go.opentelemetry.io/collector/confmap/provider/httpsprovider v0.101.0 // indirect
+	go.opentelemetry.io/collector/confmap/provider/yamlprovider v0.101.0 // indirect
+	go.opentelemetry.io/collector/connector v0.101.0 // indirect
+	go.opentelemetry.io/collector/exporter v0.101.0 // indirect
+	go.opentelemetry.io/collector/extension v0.101.0 // indirect
+	go.opentelemetry.io/collector/featuregate v1.8.0 // indirect
 	go.opentelemetry.io/collector/pdata/testdata v0.101.0 // indirect
-	go.opentelemetry.io/otel v1.27.0 // indirect
+	go.opentelemetry.io/collector/receiver v0.101.0 // indirect
+	go.opentelemetry.io/collector/semconv v0.101.0 // indirect
+	go.opentelemetry.io/collector/service v0.101.0 // indirect
+	go.opentelemetry.io/contrib/config v0.6.0 // indirect
+	go.opentelemetry.io/contrib/propagators/b3 v1.26.0 // indirect
+	go.opentelemetry.io/otel/bridge/opencensus v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.26.0 // indirect
 	go.opentelemetry.io/otel/exporters/prometheus v0.49.0 // indirect
+	go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.26.0 // indirect
+	go.opentelemetry.io/otel/exporters/stdout/stdouttrace v1.26.0 // indirect
 	go.opentelemetry.io/otel/metric v1.27.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.27.0 // indirect
 	go.opentelemetry.io/otel/sdk/metric v1.27.0 // indirect
 	go.opentelemetry.io/otel/trace v1.27.0 // indirect
+	go.opentelemetry.io/proto/otlp v1.2.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
 	golang.org/x/net v0.24.0 // indirect
 	golang.org/x/sys v0.20.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
+	gonum.org/v1/gonum v0.15.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20240318140521-94a12d6c2237 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20240401170217-c3f982113cda // indirect
 	google.golang.org/grpc v1.64.0 // indirect
 	google.golang.org/protobuf v1.34.1 // indirect