[cisco_meraki_metrics] Add pagination support for device_health metricset (#46938) (#47149)

* add pagination logic to device_health metricset

* add tests

* fix imports

* fix imports

* improve debug logs

* add nil check in getDevices

* add changelog entry

(cherry picked from commit 9db5eeb)

Co-authored-by: Gabriel Pop <[email protected]>

Author: mergify[bot]

changelog/fragments/1760358970-add-meraki-device-health-pagination-support.yaml

@@ -0,0 +1,45 @@
+# REQUIRED
+# Kind can be one of:
+# - breaking-change: a change to previously-documented behavior
+# - deprecation: functionality that is being removed in a later release
+# - bug-fix: fixes a problem in a previous version
+# - enhancement: extends functionality but does not break or fix existing behavior
+# - feature: new functionality
+# - known-issue: problems that we are aware of in a given version
+# - security: impacts on the security of a product or a user’s deployment.
+# - upgrade: important information for someone upgrading from a prior version
+# - other: does not fit into any of the other categories
+kind: bug-fix
+
+# REQUIRED for all kinds
+# Change summary; a 80ish characters long description of the change.
+summary: Add pagination support to the device health metricset in the meraki module
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# Long description; in case the summary is not enough to describe the change
+# this field accommodate a description without length limits.
+# description:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# impact:
+
+# REQUIRED for breaking-change, deprecation, known-issue
+# action:
+
+# REQUIRED for all kinds
+# Affected component; usually one of "elastic-agent", "fleet-server", "filebeat", "metricbeat", "auditbeat", "all", etc.
+component: "metricbeat"
+
+# AUTOMATED
+# OPTIONAL to manually add other PR URLs
+# PR URL: A link the PR that added the changeset.
+# If not present is automatically filled by the tooling finding the PR where this changelog fragment has been added.
+# NOTE: the tooling supports backports, so it's able to fill the original PR number instead of the backport PR number.
+# Please provide it if you are adding a fragment for a different PR.
+# pr: https://github.com/owner/repo/1234
+
+# AUTOMATED
+# OPTIONAL to manually add other issue URLs
+# Issue URL; optional; the GitHub issue related to this changeset (either closes or is part of).
+# If not present is automatically filled by the tooling with the issue linked to the PR number.
+# issue: https://github.com/owner/repo/1234

x-pack/metricbeat/module/meraki/device_health/device_health.go

@@ -64,14 +64,19 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 		// First we get the list of all devices for this org (and their metadata).
 		// Devices are uniquely identified by their serial number, which are used to
 		// associate the metrics we collect later with the devices returned here.
-		devices, err := getDevices(m.client, org)
+
+		organizationsService := &OrganizationsServiceWrapper{
+			service: m.client.Organizations,
+		}
+
+		devices, err := getDevices(organizationsService, org, m.logger)
 		if err != nil {
 			return fmt.Errorf("getDevices failed; %w", err)
 		}
 
 		// Now we continue to populate the device data structure with health
 		// attributes/statuses/metrics etc in the following functions...
-		err = getDeviceStatuses(m.client, org, devices)
+		err = getDeviceStatuses(m.client, org, devices, m.logger)
 		if err != nil {
 			return fmt.Errorf("getDeviceStatuses failed; %w", err)
 		}
@@ -87,17 +92,17 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 			return fmt.Errorf("getDeviceChannelUtilization failed; %w", err)
 		}
 
-		err = getDeviceLicenses(m.client, org, devices)
+		err = getDeviceLicenses(m.client, org, devices, m.logger)
 		if err != nil {
 			return fmt.Errorf("getDeviceLicenses failed; %w", err)
 		}
 
-		err = getDeviceUplinks(m.client, org, devices, collectionPeriod)
+		err = getDeviceUplinks(m.client, org, devices, collectionPeriod, m.logger)
 		if err != nil {
 			return fmt.Errorf("getDeviceUplinks failed; %w", err)
 		}
 
-		err = getDeviceSwitchports(m.client, org, devices, collectionPeriod)
+		err = getDeviceSwitchports(m.client, org, devices, collectionPeriod, m.logger)
 		if err != nil {
 			return fmt.Errorf("getDeviceSwitchports failed; %w", err)
 		}

x-pack/metricbeat/module/meraki/device_health/device_health_test.go

@@ -17,6 +17,8 @@ import (
 	meraki "github.com/meraki/dashboard-api-go/v3/sdk"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 )
 
 func TestGetDeviceChannelUtilization(t *testing.T) {
@@ -170,3 +172,99 @@ func (m *GenericErrorMockNetworkHealthService) GetOrganizationWirelessDevicesCha
 	}
 	return r, fmt.Errorf("mock API error")
 }
+
+func TestGetDevices_Pagination(t *testing.T) {
+	logger := logp.NewLogger("test")
+
+	tests := []struct {
+		name            string
+		client          OrganizationsClient
+		expectedDevices int
+		expectedCalls   int
+		wantErr         bool
+	}{
+		{
+			name:            "single page",
+			client:          newMockOrganizationsClient(1, 2), // 1 page, 2 devices
+			expectedDevices: 2,
+			expectedCalls:   1,
+		},
+		{
+			name:            "multiple pages",
+			client:          newMockOrganizationsClient(3, 2), // 3 pages, 2 devices per page
+			expectedDevices: 6,
+			expectedCalls:   3,
+		},
+		{
+			name:            "max pages limit",
+			client:          newMockOrganizationsClient(101, 1), // 101 pages (exceeds MAX_PAGES)
+			expectedDevices: 100,                                // Should stop at MAX_PAGES
+			expectedCalls:   100,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			devices, err := getDevices(tt.client, "org123", logger)
+
+			if tt.wantErr {
+				assert.Error(t, err)
+				return
+			}
+
+			require.NoError(t, err)
+			assert.Equal(t, tt.expectedDevices, len(devices))
+
+			// Verify the mock was called the expected number of times
+			if mock, ok := tt.client.(*mockOrganizationsClient); ok {
+				assert.Equal(t, tt.expectedCalls, mock.callCount)
+			}
+		})
+	}
+}
+
+// Mock implementation for pagination testing
+type mockOrganizationsClient struct {
+	totalPages     int
+	devicesPerPage int
+	callCount      int
+}
+
+func newMockOrganizationsClient(totalPages, devicesPerPage int) *mockOrganizationsClient {
+	return &mockOrganizationsClient{
+		totalPages:     totalPages,
+		devicesPerPage: devicesPerPage,
+	}
+}
+
+func (m *mockOrganizationsClient) GetOrganizationDevices(organizationID string, params *meraki.GetOrganizationDevicesQueryParams) (*meraki.ResponseOrganizationsGetOrganizationDevices, *resty.Response, error) {
+	m.callCount++
+
+	devices := make(meraki.ResponseOrganizationsGetOrganizationDevices, 0, m.devicesPerPage)
+	for i := 0; i < m.devicesPerPage; i++ {
+		serial := fmt.Sprintf("SERIAL-%d-%d", m.callCount, i)
+		devices = append(devices, meraki.ResponseItemOrganizationsGetOrganizationDevices{
+			Serial: serial,
+			Name:   fmt.Sprintf("Device %s", serial),
+		})
+	}
+
+	resp := &resty.Response{}
+	bodyBytes, _ := json.Marshal(devices)
+	resp.SetBody(bodyBytes)
+
+	headers := http.Header{}
+	if m.callCount < m.totalPages {
+		nextSerial := fmt.Sprintf("SERIAL-%d-%d", m.callCount, m.devicesPerPage-1)
+		linkHeader := fmt.Sprintf(`<https://api.meraki.com/api/v1/organizations/%s/devices?startingAfter=%s>; rel="next"`, organizationID, nextSerial)
+		headers.Set("Link", linkHeader)
+	}
+
+	resp.RawResponse = &http.Response{
+		StatusCode: 200,
+		Body:       io.NopCloser(bytes.NewBuffer(bodyBytes)),
+		Header:     headers,
+	}
+
+	return &devices, resp, nil
+}

x-pack/metricbeat/module/meraki/device_health/devices.go

@@ -38,49 +38,102 @@ type Device struct {
 	switchports []*switchport
 }
 
-func getDevices(client *sdk.Client, organizationID string) (map[Serial]*Device, error) {
-	val, res, err := client.Organizations.GetOrganizationDevices(organizationID, &sdk.GetOrganizationDevicesQueryParams{})
+type OrganizationsClient interface {
+	GetOrganizationDevices(organizationID string, params *sdk.GetOrganizationDevicesQueryParams) (*sdk.ResponseOrganizationsGetOrganizationDevices, *resty.Response, error)
+}
+
+var _ OrganizationsClient = (*sdk.OrganizationsService)(nil)
+
+type OrganizationsServiceWrapper struct {
+	service *sdk.OrganizationsService
+}
+
+func (w *OrganizationsServiceWrapper) GetOrganizationDevices(organizationID string, params *sdk.GetOrganizationDevicesQueryParams) (*sdk.ResponseOrganizationsGetOrganizationDevices, *resty.Response, error) {
+	return w.service.GetOrganizationDevices(organizationID, params)
+}
+
+func getDevices(client OrganizationsClient, organizationID string, logger *logp.Logger) (map[Serial]*Device, error) {
+	devices := make(map[Serial]*Device)
 
-	if err != nil {
+	params := &sdk.GetOrganizationDevicesQueryParams{}
+	setStart := func(s string) { params.StartingAfter = s }
+
+	doRequest := func() (*sdk.ResponseOrganizationsGetOrganizationDevices, *resty.Response, error) {
+		logger.Debugf("calling GetOrganizationDevices with params: %+v", params)
+		return client.GetOrganizationDevices(organizationID, params)
+	}
+
+	onError := func(err error, res *resty.Response) error {
 		if res != nil {
-			return nil, fmt.Errorf("GetOrganizationDevices failed; [%d] %s. %w", res.StatusCode(), res.Body(), err)
+			return fmt.Errorf("GetOrganizationDevices failed; [%d] %s. %w", res.StatusCode(), res.Body(), err)
 		}
-		return nil, fmt.Errorf("GetOrganizationDevices failed; %w", err)
+		return fmt.Errorf("GetOrganizationDevices failed; %w", err)
 	}
 
-	devices := make(map[Serial]*Device)
-	for i := range *val {
-		device := (*val)[i]
-		devices[Serial(device.Serial)] = &Device{
-			details: &device,
+	onSuccess := func(val *sdk.ResponseOrganizationsGetOrganizationDevices) error {
+		if val == nil {
+			return errors.New("GetOrganizationDevices returned nil response")
 		}
+
+		for i := range *val {
+			device := (*val)[i]
+			devices[Serial(device.Serial)] = &Device{
+				details: &device,
+			}
+		}
+		return nil
 	}
 
-	return devices, nil
+	err := meraki.NewPaginator(
+		setStart,
+		doRequest,
+		onError,
+		onSuccess,
+		logger,
+	).GetAllPages()
+
+	return devices, err
 }
 
-func getDeviceStatuses(client *sdk.Client, organizationID string, devices map[Serial]*Device) error {
-	val, res, err := client.Organizations.GetOrganizationDevicesStatuses(organizationID, &sdk.GetOrganizationDevicesStatusesQueryParams{})
+func getDeviceStatuses(client *sdk.Client, organizationID string, devices map[Serial]*Device, logger *logp.Logger) error {
+	params := &sdk.GetOrganizationDevicesStatusesQueryParams{}
+	setStart := func(s string) { params.StartingAfter = s }
 
-	if err != nil {
+	doRequest := func() (*sdk.ResponseOrganizationsGetOrganizationDevicesStatuses, *resty.Response, error) {
+		logger.Debugf("calling GetOrganizationDevicesStatuses with params: %+v", params)
+		return client.Organizations.GetOrganizationDevicesStatuses(organizationID, params)
+	}
+
+	onError := func(err error, res *resty.Response) error {
 		if res != nil {
 			return fmt.Errorf("GetOrganizationDevicesStatuses failed; [%d] %s. %w", res.StatusCode(), res.Body(), err)
 		}
 		return fmt.Errorf("GetOrganizationDevicesStatuses failed; %w", err)
 	}
 
-	if val == nil {
-		return errors.New("GetOrganizationDevicesStatuses returned nil response")
-	}
+	onSuccess := func(val *sdk.ResponseOrganizationsGetOrganizationDevicesStatuses) error {
+		if val == nil {
+			return errors.New("GetOrganizationDevicesStatuses returned nil response")
+		}
 
-	for i := range *val {
-		status := (*val)[i]
-		if device, ok := devices[Serial(status.Serial)]; ok {
-			device.status = &status
+		for i := range *val {
+			status := (*val)[i]
+			if device, ok := devices[Serial(status.Serial)]; ok {
+				device.status = &status
+			}
 		}
+		return nil
 	}
 
-	return nil
+	err := meraki.NewPaginator(
+		setStart,
+		doRequest,
+		onError,
+		onSuccess,
+		logger,
+	).GetAllPages()
+
+	return err
 }
 
 func getDevicePerformanceScores(logger *logp.Logger, client *sdk.Client, devices map[Serial]*Device) {
@@ -186,9 +239,16 @@ func getDeviceChannelUtilization(client DeviceService, devices map[Serial]*Devic
 	return nil
 }
 
-func getDeviceLicenses(client *sdk.Client, organizationID string, devices map[Serial]*Device) error {
-	val, res, err := client.Organizations.GetOrganizationLicenses(organizationID, &sdk.GetOrganizationLicensesQueryParams{})
-	if err != nil {
+func getDeviceLicenses(client *sdk.Client, organizationID string, devices map[Serial]*Device, logger *logp.Logger) error {
+	params := &sdk.GetOrganizationLicensesQueryParams{}
+	setStart := func(s string) { params.StartingAfter = s }
+
+	doRequest := func() (*sdk.ResponseOrganizationsGetOrganizationLicenses, *resty.Response, error) {
+		logger.Debugf("calling GetOrganizationLicenses with params: %+v", params)
+		return client.Organizations.GetOrganizationLicenses(organizationID, params)
+	}
+
+	onError := func(err error, res *resty.Response) error {
 		// Ignore 400 error for per-device licensing not supported
 		if res != nil && res.StatusCode() == 400 && strings.Contains(string(res.Body()), "does not support per-device licensing") {
 			return nil
@@ -204,18 +264,29 @@ func getDeviceLicenses(client *sdk.Client, organizationID string, devices map[Se
 		return fmt.Errorf("GetOrganizationLicenses failed; %w", err)
 	}
 
-	if val == nil {
-		return errors.New("GetOrganizationLicenses returned nil response")
-	}
+	onSuccess := func(val *sdk.ResponseOrganizationsGetOrganizationLicenses) error {
+		if val == nil {
+			return errors.New("GetOrganizationLicenses returned nil response")
+		}
 
-	for i := range *val {
-		license := (*val)[i]
-		if device, ok := devices[Serial(license.DeviceSerial)]; ok {
-			device.license = &license
+		for i := range *val {
+			license := (*val)[i]
+			if device, ok := devices[Serial(license.DeviceSerial)]; ok {
+				device.license = &license
+			}
 		}
+		return nil
 	}
 
-	return nil
+	err := meraki.NewPaginator(
+		setStart,
+		doRequest,
+		onError,
+		onSuccess,
+		logger,
+	).GetAllPages()
+
+	return err
 }
 
 func deviceDetailsToMapstr(details *sdk.ResponseItemOrganizationsGetOrganizationDevices) mapstr.M {