[NDMII-3603] Dynamic batch size for SNMP check (#41689)

### What does this PR do?

This PR adds implements a dynamic batch size for the SNMP check, it works as follows:

At start, the batch size is the configured batch size (or default if not configured).

For a fetch in a SNMP check:
  - If the fetch fails:
    - The batch size is divided by a [`decreaseFactor`](https://github.com/DataDog/datadog-agent/pull/41689/files#diff-7e1c5e2d2bee2b336f1316cf0bba6c2f50d17461d9ff6b8997d4c8ef32ed202aR16), we retry the fetch with the new batch size in the same check. If it still fails, we decrease the batch size again.. until it reaches 1.
  - If the fetch succeeds:
    - The batch size is increased by an [`increaseValue`](https://github.com/DataDog/datadog-agent/pull/41689/files#diff-7e1c5e2d2bee2b336f1316cf0bba6c2f50d17461d9ff6b8997d4c8ef32ed202aR16) for the next check. The increased batch size cannot be more than the configured batch size.

We also keep a map that associates the number of fetch failures count by batch size. This map is used such that if a batch size have failed too much time ([`maxFailuresPerWindow`](https://github.com/DataDog/datadog-agent/pull/41689/files#diff-7e1c5e2d2bee2b336f1316cf0bba6c2f50d17461d9ff6b8997d4c8ef32ed202aR19)) during a certain time window ([failuresWindowDuration](https://github.com/DataDog/datadog-agent/pull/41689/files#diff-7e1c5e2d2bee2b336f1316cf0bba6c2f50d17461d9ff6b8997d4c8ef32ed202aR18)), we do not increase to this batch size, and keep the current one.
I think this is useful to not always retry batch sizes that will always fail for some devices, but also do retry it after the time window in case some devices didn't have enough capacity only for temporary moment.

I also chose to separate the batch size for each SNMP operation (Get, GetBulk, GetNext), because I think it's a possibility that a certain batch size will work for one operation but not for an other one (I do not have proof of that), let me know if you think keeping a single batch size for all operation would be better.

### Motivation

### Describe how you validated your changes

- [Unit tests](https://github.com/DataDog/datadog-agent/pull/41689/files#diff-1e048e3bd44ca70b6119fbcd02b2890d0eeb568c7ec448e557e227c289ddb933R1004) with multiple fetch iterations.
- Did a QA with a Python script that set up a SNMPSim device with custom max batch size for each SNMP operation, and it worked as expected.

### Additional Notes

Author: Pierre-L42

pkg/collector/corechecks/snmp/internal/devicecheck/devicecheck.go

@@ -142,6 +142,7 @@ type DeviceCheck struct {
 	sender                  *report.MetricSender
 	session                 session.Session
 	sessionFactory          session.Factory
+	oidBatchSizeOptimizers  *fetch.OidBatchSizeOptimizers
 	devicePinger            pinger.Pinger
 	sessionCloseErrorCount  *atomic.Uint64
 	savedDynamicTags        []string
@@ -173,6 +174,7 @@ func NewDeviceCheck(config *checkconfig.CheckConfig, ipAddress string, sessionFa
 	d := DeviceCheck{
 		config:                  newConfig,
 		sessionFactory:          sessionFactory,
+		oidBatchSizeOptimizers:  fetch.NewOidBatchSizeOptimizers(newConfig.OidBatchSize),
 		devicePinger:            devicePinger,
 		sessionCloseErrorCount:  atomic.NewUint64(0),
 		diagnoses:               diagnoses.NewDeviceDiagnoses(newConfig.DeviceID),
@@ -390,8 +392,8 @@ func (d *DeviceCheck) getValuesAndTags() (bool, profiledefinition.ProfileDefinit
 
 	tags = append(tags, profile.StaticTags...)
 
-	valuesStore, err := fetch.Fetch(d.session, d.profileCache.scalarOIDs, d.profileCache.columnOIDs, d.config.OidBatchSize,
-		d.config.BulkMaxRepetitions)
+	valuesStore, err := fetch.Fetch(d.session, d.profileCache.scalarOIDs, d.profileCache.columnOIDs,
+		d.oidBatchSizeOptimizers, d.config.BulkMaxRepetitions)
 	if log.ShouldLog(log.DebugLvl) {
 		log.Debugf("fetched values: %v", valuestore.ResultValueStoreAsString(valuesStore))
 	}

pkg/collector/corechecks/snmp/internal/fetch/batch_size_optimizers.go

@@ -0,0 +1,122 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025-present Datadog, Inc.
+
+package fetch
+
+import (
+	"time"
+
+	"github.com/DataDog/datadog-agent/pkg/util/log"
+)
+
+const (
+	onSuccessIncreaseValue  = 1
+	onFailureDecreaseFactor = 2
+
+	failuresWindowDuration = 60 * time.Minute
+	maxFailuresPerWindow   = 2
+)
+
+// OidBatchSizeOptimizers holds oidBatchSizeOptimizer for each SNMP request operation
+type OidBatchSizeOptimizers struct {
+	snmpGetOptimizer     *oidBatchSizeOptimizer
+	snmpGetBulkOptimizer *oidBatchSizeOptimizer
+	snmpGetNextOptimizer *oidBatchSizeOptimizer
+	lastRefreshTs        time.Time
+}
+
+// oidBatchSizeOptimizer holds data between check runs to be able to find an optimized batch size for SNMP requests
+type oidBatchSizeOptimizer struct {
+	snmpOperation           snmpOperation
+	configBatchSize         int
+	batchSize               int
+	failuresByBatchSize     map[int]int
+	lastSuccessfulBatchSize int
+}
+
+// NewOidBatchSizeOptimizers creates a OidBatchSizeOptimizers
+func NewOidBatchSizeOptimizers(configBatchSize int) *OidBatchSizeOptimizers {
+	now := time.Now()
+
+	return &OidBatchSizeOptimizers{
+		snmpGetOptimizer:     newOidBatchSizeOptimizer(snmpGet, configBatchSize),
+		snmpGetBulkOptimizer: newOidBatchSizeOptimizer(snmpGetBulk, configBatchSize),
+		snmpGetNextOptimizer: newOidBatchSizeOptimizer(snmpGetNext, configBatchSize),
+		lastRefreshTs:        now,
+	}
+}
+
+// Refresh refreshes each oidBatchSizeOptimizer in OidBatchSizeOptimizers when outdated
+func (o *OidBatchSizeOptimizers) refreshIfOutdated(now time.Time) {
+	if now.Sub(o.lastRefreshTs) < failuresWindowDuration {
+		return
+	}
+
+	o.snmpGetOptimizer.refreshFailures()
+	o.snmpGetBulkOptimizer.refreshFailures()
+	o.snmpGetNextOptimizer.refreshFailures()
+
+	o.lastRefreshTs = now
+
+	log.Debug("SNMP batch size optimizers have been refreshed")
+}
+
+// newOidBatchSizeOptimizer creates a oidBatchSizeOptimizer
+func newOidBatchSizeOptimizer(snmpOperation snmpOperation, configBatchSize int) *oidBatchSizeOptimizer {
+	return &oidBatchSizeOptimizer{
+		snmpOperation:       snmpOperation,
+		configBatchSize:     configBatchSize,
+		batchSize:           configBatchSize,
+		failuresByBatchSize: make(map[int]int),
+	}
+}
+
+// refreshFailures refreshes the failures count for each batch size in oidBatchSizeOptimizer
+func (o *oidBatchSizeOptimizer) refreshFailures() {
+	o.failuresByBatchSize = make(map[int]int)
+}
+
+// onBatchSizeFailure decreases the batch size and returns whether the batch size changed
+func (o *oidBatchSizeOptimizer) onBatchSizeFailure() bool {
+	o.failuresByBatchSize[o.batchSize]++
+
+	oldBatchSize := o.batchSize
+
+	newBatchSize := max(o.batchSize/onFailureDecreaseFactor, 1)
+	if oldBatchSize > o.lastSuccessfulBatchSize && newBatchSize < o.lastSuccessfulBatchSize {
+		newBatchSize = o.lastSuccessfulBatchSize
+	}
+
+	o.batchSize = newBatchSize
+
+	log.Debugf("SNMP fetch using %s with batch size %d failed, new batch size is %d",
+		o.snmpOperation, oldBatchSize, newBatchSize)
+
+	return oldBatchSize != newBatchSize
+}
+
+// onBatchSizeSuccess increases the batch size
+func (o *oidBatchSizeOptimizer) onBatchSizeSuccess() {
+	o.lastSuccessfulBatchSize = o.batchSize
+
+	if o.batchSize >= o.maxBatchSize() {
+		return
+	}
+
+	newBatchSize := min(o.batchSize+onSuccessIncreaseValue, o.maxBatchSize())
+	if o.failuresByBatchSize[newBatchSize] >= maxFailuresPerWindow {
+		return
+	}
+
+	log.Debugf("SNMP fetch using %s with batch size %d success, new batch size is %d",
+		o.snmpOperation, o.lastSuccessfulBatchSize, newBatchSize)
+
+	o.batchSize = newBatchSize
+}
+
+// maxBatchSize returns the max batch size
+func (o *oidBatchSizeOptimizer) maxBatchSize() int {
+	return o.configBatchSize
+}