[9.1](backport #46018) Remove all remaining NewRegistry method calls (#46222)

* Remove all remaining NewRegistry method calls (#46018)

The `NewRegistry` method in `elastic-agent-libs` is inherently unsafe, especially in concurrent code: calling `NewRegistry` on a name that already exists causes a panic, but checking whether the name exists first is not atomic. The recently added `GetOrCreateRegistry` method is a safe alternative that checks the name's existence and initializes it atomically, and never panics. This PR replaces all remaining `NewRegistry` calls with `GetOrCreateRegistry`.

(cherry picked from commit bf7a774)

# Conflicts:
#	libbeat/monitoring/inputmon/input_test.go

* fix merge

---------

Co-authored-by: Fae Charlton <[email protected]>

Author: mergify[bot]

auditbeat/module/auditd/audit_linux.go

@@ -65,7 +65,7 @@ const (
 )
 
 var (
-	auditdMetrics         = monitoring.Default.NewRegistry(moduleName)
+	auditdMetrics         = monitoring.Default.GetOrCreateRegistry(moduleName)
 	reassemblerGapsMetric = monitoring.NewInt(auditdMetrics, "reassembler_seq_gaps")
 	kernelLostMetric      = monitoring.NewInt(auditdMetrics, "kernel_lost")
 	userspaceLostMetric   = monitoring.NewInt(auditdMetrics, "userspace_lost")

filebeat/input/log/harvester.go

@@ -60,7 +60,7 @@ import (
 )
 
 var (
-	harvesterMetrics = monitoring.Default.NewRegistry("filebeat.harvester")
+	harvesterMetrics = monitoring.Default.GetOrCreateRegistry("filebeat.harvester")
 	filesMetrics     = monitoring.GetNamespace("dataset").GetRegistry()
 
 	harvesterStarted   = monitoring.NewInt(harvesterMetrics, "started")
@@ -178,7 +178,7 @@ func (h *Harvester) open() error {
 	case harvester.LogType, harvester.DockerType, harvester.ContainerType:
 		return h.openFile()
 	default:
-		return fmt.Errorf("Invalid harvester type: %+v", h.config)
+		return fmt.Errorf("invalid harvester type: %+v", h.config)
 	}
 }
 
@@ -217,7 +217,7 @@ func (h *Harvester) Setup() error {
 }
 
 func newHarvesterProgressMetrics(id string) *harvesterProgressMetrics {
-	r := filesMetrics.NewRegistry(id)
+	r := filesMetrics.GetOrCreateRegistry(id)
 	return &harvesterProgressMetrics{
 		metricsRegistry:             r,
 		filename:                    monitoring.NewString(r, "name"),
@@ -444,7 +444,7 @@ func (h *Harvester) onMessage(
 	// Check if json fields exist
 	var jsonFields mapstr.M
 	if f, ok := fields["json"]; ok {
-		jsonFields = f.(mapstr.M)
+		jsonFields, _ = f.(mapstr.M)
 	}
 
 	var meta mapstr.M

heartbeat/hbregistry/hbregistry.go

@@ -20,10 +20,10 @@ package hbregistry
 import "github.com/elastic/elastic-agent-libs/monitoring"
 
 // StatsRegistry contains a singleton instance of the heartbeat stats registry
-var StatsRegistry = monitoring.Default.NewRegistry("heartbeat")
+var StatsRegistry = monitoring.Default.GetOrCreateRegistry("heartbeat")
 
 // SchedulerRegistry holds scheduler stats
-var SchedulerRegistry = StatsRegistry.NewRegistry("scheduler")
+var SchedulerRegistry = StatsRegistry.GetOrCreateRegistry("scheduler")
 
 // TelemetryRegistry contains a singleton instance of the heartbeat telemetry registry
-var TelemetryRegistry = monitoring.GetNamespace("state").GetRegistry().NewRegistry("heartbeat")
+var TelemetryRegistry = monitoring.GetNamespace("state").GetRegistry().GetOrCreateRegistry("heartbeat")

heartbeat/monitors/plugin/regrecord.go

@@ -53,7 +53,7 @@ type CountersRecorder struct {
 }
 
 func NewPluginCountersRecorder(pluginName string, rootRegistry *monitoring.Registry) RegistryRecorder {
-	pluginRegistry := rootRegistry.NewRegistry(pluginName)
+	pluginRegistry := rootRegistry.GetOrCreateRegistry(pluginName)
 	return CountersRecorder{
 		monitoring.NewInt(pluginRegistry, "monitor_starts"),
 		monitoring.NewInt(pluginRegistry, "monitor_stops"),
@@ -87,7 +87,7 @@ func newRootGaugeRecorder(r *monitoring.Registry) RegistryRecorder {
 }
 
 func newPluginGaugeRecorder(pluginName string, rootRegistry *monitoring.Registry) RegistryRecorder {
-	pluginRegistry := rootRegistry.NewRegistry(pluginName)
+	pluginRegistry := rootRegistry.GetOrCreateRegistry(pluginName)
 	return newRootGaugeRecorder(pluginRegistry)
 }
 

libbeat/cmd/instance/beat.go

@@ -358,7 +358,7 @@ func (b *Beat) createBeater(bt beat.Creator) (beat.Beater, error) {
 	}
 
 	// Report central management state
-	mgmt := b.Monitoring.StateRegistry().NewRegistry("management")
+	mgmt := b.Monitoring.StateRegistry().GetOrCreateRegistry("management")
 	monitoring.NewBool(mgmt, "enabled").Set(b.Manager.Enabled())
 
 	log.Debug("Initializing output plugins")
@@ -1287,7 +1287,7 @@ func (b *Beat) registerClusterUUIDFetching() {
 
 // Build and return a callback to fetch the Elasticsearch cluster_uuid for monitoring
 func (b *Beat) clusterUUIDFetchingCallback() elasticsearch.ConnectCallback {
-	elasticsearchRegistry := b.Monitoring.StateRegistry().NewRegistry("outputs.elasticsearch")
+	elasticsearchRegistry := b.Monitoring.StateRegistry().GetOrCreateRegistry("outputs.elasticsearch")
 	clusterUUIDRegVar := monitoring.NewString(elasticsearchRegistry, "cluster_uuid")
 
 	callback := func(esClient *eslegclient.Connection) error {
@@ -1324,7 +1324,7 @@ func (b *Beat) setupMonitoring(settings Settings) (report.Reporter, error) {
 
 	// Expose monitoring.cluster_uuid in state API
 	if monitoringClusterUUID != "" {
-		monitoringRegistry := b.Monitoring.StateRegistry().NewRegistry("monitoring")
+		monitoringRegistry := b.Monitoring.StateRegistry().GetOrCreateRegistry("monitoring")
 		clusterUUIDRegVar := monitoring.NewString(monitoringRegistry, "cluster_uuid")
 		clusterUUIDRegVar.Set(monitoringClusterUUID)
 	}

libbeat/monitoring/inputmon/httphandler_test.go

@@ -59,7 +59,7 @@ func TestHandler(t *testing.T) {
 
 	// Register legacy metrics without id or input. This must be ignored.
 	{
-		legacy := parent.NewRegistry("f49c0680-fc5f-4b78-bd98-7b16628f9a77")
+		legacy := parent.GetOrCreateRegistry("f49c0680-fc5f-4b78-bd98-7b16628f9a77")
 		monitoring.NewString(legacy, "name").Set("/var/log/wifi.log")
 		monitoring.NewTimestamp(legacy, "last_event_published_time").Set(time.Now())
 	}

libbeat/monitoring/inputmon/input.go

@@ -70,7 +70,7 @@ func NewInputRegistry(inputType, inputID string, optionalParent *monitoring.Regi
 	// that isn't at the top-level.
 	reg = findInputRegistryWithID(parentRegistry, inputID)
 	if reg == nil {
-		reg = parentRegistry.NewRegistry(registryID)
+		reg = parentRegistry.GetOrCreateRegistry(registryID)
 	} else {
 		log.Warnw(fmt.Sprintf(
 			"parent metrics registry already contains a %q registry, reusing it",
@@ -143,7 +143,15 @@ func NewMetricsRegistry(
 	registryID := sanitizeID(inputID)
 	reg := parent.GetRegistry(registryID)
 	if reg == nil {
-		reg = parent.NewRegistry(registryID)
+		// Technically this is a race: the previous GetRegistry is not atomic with
+		// this GetOrCreateRegistry, so if someone else is adding the same id at this
+		// exact moment we might take this branch inappropriately. However, the
+		// consequences in that case are metrics for two inputs with the same id
+		// being inappropriately reported under the same registry, whereas the
+		// consequences if we called NewRegistry here are a panic.
+		// (We should also never have two inputs with the same id in the first
+		// place, but it's worth being cautious.)
+		reg = parent.GetOrCreateRegistry(registryID)
 	} else {
 		// Null route metrics for duplicated ID.
 		reg = monitoring.NewRegistry()

libbeat/monitoring/inputmon/input_test.go

@@ -105,7 +105,7 @@ func TestMetricSnapshotJSON(t *testing.T) {
 	monitoring.NewInt(reg, "events_pipeline_total").Set(10)
 
 	// simulate a duplicated ID in the local and global namespace.
-	reg = globalRegistry().NewRegistry(inputID)
+	reg = globalRegistry().GetOrCreateRegistry(inputID)
 	monitoring.NewString(reg, "id").Set(inputID)
 	monitoring.NewString(reg, "input").Set(inputType)
 	monitoring.NewBool(reg, "should_be_overwritten").Set(true)
@@ -139,12 +139,12 @@ func TestMetricSnapshotJSON(t *testing.T) {
 
 	// ==== registries in the global registries which aren't input metrics ===
 	// unrelated registry in the global namespace, should be ignored.
-	reg = globalRegistry().NewRegistry("another-registry")
+	reg = globalRegistry().GetOrCreateRegistry("another-registry")
 	monitoring.NewInt(reg, "foo3_total").Set(100)
 	defer globalRegistry().Remove("another-registry")
 
 	// another input registry missing required information.
-	reg = globalRegistry().NewRegistry("yet-another-registry")
+	reg = globalRegistry().GetOrCreateRegistry("yet-another-registry")
 	monitoring.NewString(reg, "id").Set("some-id")
 	monitoring.NewInt(reg, "foo3_total").Set(100)
 	defer globalRegistry().Remove("yet-another-registry")
@@ -339,7 +339,7 @@ func TestCancelMetricsRegistry(t *testing.T) {
 	inputID := "input-ID"
 	inputType := "input-type"
 
-	_ = parent.NewRegistry(inputID)
+	_ = parent.GetOrCreateRegistry(inputID)
 	got := parent.GetRegistry(inputID)
 	require.NotNil(t, got, "metrics registry not found on parent")
 

libbeat/processors/add_host_metadata/add_host_metadata.go

@@ -50,7 +50,7 @@ func init() {
 	processors.RegisterPlugin(processorName, New)
 	jsprocessor.RegisterPlugin("AddHostMetadata", New)
 
-	reg = monitoring.Default.NewRegistry(logName, monitoring.DoNotReport)
+	reg = monitoring.Default.GetOrCreateRegistry(logName, monitoring.DoNotReport)
 }
 
 type metrics struct {

libbeat/processors/dns/dns.go

@@ -59,7 +59,7 @@ func New(cfg *conf.C, log *logp.Logger) (beat.Processor, error) {
 	// Logging and metrics (each processor instance has a unique ID).
 	var (
 		id      = int(instanceID.Add(1))
-		metrics = monitoring.Default.NewRegistry(logName+"."+strconv.Itoa(id), monitoring.DoNotReport)
+		metrics = monitoring.Default.GetOrCreateRegistry(logName+"."+strconv.Itoa(id), monitoring.DoNotReport)
 	)
 
 	log = log.Named(logName).With("instance_id", id)
@@ -69,7 +69,7 @@ func New(cfg *conf.C, log *logp.Logger) (beat.Processor, error) {
 		return nil, err
 	}
 
-	cache, err := newLookupCache(metrics.NewRegistry("cache"), c.cacheConfig, resolver)
+	cache, err := newLookupCache(metrics.GetOrCreateRegistry("cache"), c.cacheConfig, resolver)
 	if err != nil {
 		return nil, err
 	}