fix: close geoip providers database readers (#39379)

<!--Ex. Fixing a bug - Describe the bug and how this fixes the issue.
Ex. Adding a feature - Explain what this achieves.-->
#### Description

Adds a new `Close` method to the geoIP providers interface, so resources
can be freed during processor's shutdown.

<!-- Issue number (e.g. #1234) or full URL to issue, if applicable. -->
#### Link to tracking issue
Fixes
#38961

<!--Describe what testing was performed and which tests were added.-->
#### Testing

Flaky windows tests uncommented. 

<!--Describe the documentation added.-->
#### Documentation

<!--Please delete paragraphs that you did not use before submitting.-->

---------

Co-authored-by: Andrzej Stencel <[email protected]>

Author: rogercoll

.chloggen/fix_geoip_close.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: bug_fix
+
+# 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: Close providers readers on shutdown
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [38961]
+
+# (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/factory.go

@@ -92,7 +92,8 @@ func createMetricsProcessor(ctx context.Context, set processor.Settings, cfg com
 	if err != nil {
 		return nil, err
 	}
-	return processorhelper.NewMetrics(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg, providers, set).processMetrics, processorhelper.WithCapabilities(processorCapabilities))
+	geoProcessor := newGeoIPProcessor(geoCfg, providers, set)
+	return processorhelper.NewMetrics(ctx, set, cfg, nextConsumer, geoProcessor.processMetrics, processorhelper.WithShutdown(geoProcessor.shutdown), processorhelper.WithCapabilities(processorCapabilities))
 }
 
 func createTracesProcessor(ctx context.Context, set processor.Settings, cfg component.Config, nextConsumer consumer.Traces) (processor.Traces, error) {
@@ -101,7 +102,8 @@ func createTracesProcessor(ctx context.Context, set processor.Settings, cfg comp
 	if err != nil {
 		return nil, err
 	}
-	return processorhelper.NewTraces(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg, providers, set).processTraces, processorhelper.WithCapabilities(processorCapabilities))
+	geoProcessor := newGeoIPProcessor(geoCfg, providers, set)
+	return processorhelper.NewTraces(ctx, set, cfg, nextConsumer, geoProcessor.processTraces, processorhelper.WithShutdown(geoProcessor.shutdown), processorhelper.WithCapabilities(processorCapabilities))
 }
 
 func createLogsProcessor(ctx context.Context, set processor.Settings, cfg component.Config, nextConsumer consumer.Logs) (processor.Logs, error) {
@@ -110,5 +112,6 @@ func createLogsProcessor(ctx context.Context, set processor.Settings, cfg compon
 	if err != nil {
 		return nil, err
 	}
-	return processorhelper.NewLogs(ctx, set, cfg, nextConsumer, newGeoIPProcessor(geoCfg, providers, set).processLogs, processorhelper.WithCapabilities(processorCapabilities))
+	geoProcessor := newGeoIPProcessor(geoCfg, providers, set)
+	return processorhelper.NewLogs(ctx, set, cfg, nextConsumer, geoProcessor.processLogs, processorhelper.WithShutdown(geoProcessor.shutdown), processorhelper.WithCapabilities(processorCapabilities))
 }

processor/geoipprocessor/geoip_processor.go

@@ -12,6 +12,7 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/processor"
 	"go.opentelemetry.io/otel/attribute"
+	"go.uber.org/multierr"
 	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector-contrib/processor/geoipprocessor/internal/provider"
@@ -115,3 +116,15 @@ func (g *geoIPProcessor) processAttributes(ctx context.Context, metadata pcommon
 
 	return nil
 }
+
+func (g *geoIPProcessor) shutdown(ctx context.Context) error {
+	var errs error
+	for _, geoProvider := range g.providers {
+		err := geoProvider.Close(ctx)
+		if err != nil {
+			errs = multierr.Append(errs, err)
+		}
+	}
+
+	return errs
+}

processor/geoipprocessor/geoip_processor_test.go

@@ -5,10 +5,12 @@ package geoipprocessor
 
 import (
 	"context"
+	"errors"
 	"net"
 	"path/filepath"
 	"testing"
 
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"go.opentelemetry.io/collector/component"
 	"go.opentelemetry.io/collector/consumer/consumertest"
@@ -36,10 +38,10 @@ type providerFactoryMock struct {
 
 type providerMock struct {
 	LocationF func(context.Context, net.IP) (attribute.Set, error)
+	CloseF    func(context.Context) error
 }
 
 var (
-	_ provider.GeoIPProvider        = (*providerMock)(nil)
 	_ provider.GeoIPProvider        = (*providerMock)(nil)
 	_ provider.GeoIPProviderFactory = (*providerFactoryMock)(nil)
 )
@@ -60,10 +62,17 @@ func (pm *providerMock) Location(ctx context.Context, ip net.IP) (attribute.Set,
 	return pm.LocationF(ctx, ip)
 }
 
+func (pm *providerMock) Close(ctx context.Context) error {
+	return pm.CloseF(ctx)
+}
+
 var baseMockProvider = providerMock{
 	LocationF: func(context.Context, net.IP) (attribute.Set, error) {
 		return attribute.Set{}, nil
 	},
+	CloseF: func(context.Context) error {
+		return nil
+	},
 }
 
 var baseMockFactory = providerFactoryMock{
@@ -79,6 +88,9 @@ var baseProviderMock = providerMock{
 	LocationF: func(context.Context, net.IP) (attribute.Set, error) {
 		return attribute.Set{}, nil
 	},
+	CloseF: func(context.Context) error {
+		return nil
+	},
 }
 
 var testCases = []struct {
@@ -148,6 +160,7 @@ func compareAllSignals(cfg component.Config, goldenDir string) func(t *testing.T
 
 		err = metricsProcessor.ConsumeMetrics(context.Background(), inputMetrics)
 		require.NoError(t, err)
+		require.NoError(t, metricsProcessor.Shutdown(context.Background()))
 
 		actualMetrics := nextMetrics.AllMetrics()
 		require.Len(t, actualMetrics, 1)
@@ -167,6 +180,7 @@ func compareAllSignals(cfg component.Config, goldenDir string) func(t *testing.T
 
 		err = tracesProcessor.ConsumeTraces(context.Background(), inputTraces)
 		require.NoError(t, err)
+		require.NoError(t, tracesProcessor.Shutdown(context.Background()))
 
 		actualTraces := nextTraces.AllTraces()
 		require.Len(t, actualTraces, 1)
@@ -191,6 +205,7 @@ func compareAllSignals(cfg component.Config, goldenDir string) func(t *testing.T
 		require.Len(t, actualLogs, 1)
 		// golden.WriteLogs(t, filepath.Join(dir, "output-logs.yaml"), actualLogs[0])
 		require.NoError(t, plogtest.CompareLogs(expectedLogs, actualLogs[0]))
+		require.NoError(t, logsProcessor.Shutdown(context.Background()))
 	}
 }
 
@@ -233,3 +248,23 @@ func TestProcessor(t *testing.T) {
 		})
 	}
 }
+
+func TestProcessorShutdownError(t *testing.T) {
+	// processor with two mocked providers that return error on close
+	processor := geoIPProcessor{
+		providers: []provider.GeoIPProvider{
+			&providerMock{
+				CloseF: func(context.Context) error {
+					return errors.New("test error 1")
+				},
+			},
+			&providerMock{
+				CloseF: func(context.Context) error {
+					return errors.New("test error 2")
+				},
+			},
+		},
+	}
+
+	assert.EqualError(t, processor.shutdown(context.Background()), "test error 1; test error 2")
+}

processor/geoipprocessor/go.mod

@@ -21,6 +21,7 @@ require (
 	go.opentelemetry.io/collector/processor/processortest v0.124.0
 	go.opentelemetry.io/otel v1.35.0
 	go.uber.org/goleak v1.3.0
+	go.uber.org/multierr v1.11.0
 	go.uber.org/zap v1.27.0
 )
 
@@ -120,7 +121,6 @@ require (
 	go.opentelemetry.io/otel/sdk/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.5.0 // indirect
-	go.uber.org/multierr v1.11.0 // indirect
 	go4.org/netipx v0.0.0-20230824141953-6213f710f925 // indirect
 	golang.org/x/exp v0.0.0-20240506185415-9bf2ced13842 // indirect
 	golang.org/x/net v0.39.0 // indirect

processor/geoipprocessor/internal/provider/geoipprovider.go

@@ -25,6 +25,9 @@ type Config interface {
 type GeoIPProvider interface {
 	// Location returns a set of attributes representing the geographical location for the given IP address. It requires a context for managing request lifetime.
 	Location(context.Context, net.IP) (attribute.Set, error)
+	// Close releases any resources held by the provider. It should be called when the
+	// provider is no longer needed to ensure proper cleanup.
+	Close(context.Context) error
 }
 
 // GeoIPProviderFactory can create GeoIPProvider instances.

processor/geoipprocessor/internal/provider/maxmindprovider/provider.go

@@ -58,6 +58,15 @@ func (g *maxMindProvider) Location(_ context.Context, ipAddress net.IP) (attribu
 	}
 }
 
+// Close unmaps the geo database file from virtual memory and returns the
+// resources to the system.
+func (g *maxMindProvider) Close(context.Context) error {
+	if g.geoReader != nil {
+		return g.geoReader.Close()
+	}
+	return nil
+}
+
 // cityAttributes returns a list of key-values containing geographical metadata associated to the provided IP. The key names are populated using the internal geo IP conventions package. If an invalid or nil IP is provided, an error is returned.
 func (g *maxMindProvider) cityAttributes(ipAddress net.IP) (*[]attribute.KeyValue, error) {
 	attributes := make([]attribute.KeyValue, 0, 11)

processor/geoipprocessor/internal/provider/maxmindprovider/provider_test.go

@@ -32,9 +32,6 @@ func TestInvalidNewProvider(t *testing.T) {
 
 // TestProviderLocation asserts that the MaxMind provider adds the geo location data given an IP.
 func TestProviderLocation(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		t.Skip("https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/38961")
-	}
 	tmpDBfiles := testdata.GenerateLocalDB(t, "./testdata")
 	defer os.RemoveAll(tmpDBfiles)
 
@@ -108,10 +105,12 @@ func TestProviderLocation(t *testing.T) {
 			actualAttributes, err := provider.Location(context.Background(), tt.sourceIP)
 			if tt.expectedErrMsg != "" {
 				assert.EqualError(t, err, tt.expectedErrMsg)
+				assert.NoError(t, provider.Close(context.Background()))
 				return
 			}
 
 			assert.True(t, tt.expectedAttributes.Equals(&actualAttributes))
+			assert.NoError(t, provider.Close(context.Background()))
 		})
 	}
 }