fix(hubs): paginate Azure Resource Graph queries past 1,000 rows

docs-mslearn/toolkit/changelog.md

@@ -3,7 +3,7 @@ title: FinOps toolkit changelog
 description: Review the latest features and enhancements in the FinOps toolkit, including updates to FinOps hubs, Power BI reports, and more.
 author: MSBrett
 ms.author: brettwil
-ms.date: 09/11/2026
+ms.date: 10/14/2026
 ms.topic: reference
 ms.service: finops
 ms.subservice: finops-toolkit
@@ -41,6 +41,7 @@ The following section lists features and enhancements that are currently in deve
   - Fixed the SQL VMs without Azure Hybrid Benefit recommendation query to join on the SQL VM `virtualMachineResourceId` instead of a case-sensitive VM name match that skipped VMs with uppercase names and dropped duplicate names, and made all Azure Resource Graph join kinds explicit so no query relies on the `innerunique` default ([#2225](https://github.com/microsoft/finops-toolkit/pull/2225)).
   - Switched dimension enrichment in the v1_0/v1_2 ingestion transforms (`PricingUnits`, `Regions`, `ResourceTypes`, `Services`) from `join` to the broadcast-optimized `lookup` operator and deduplicated the `Services` mapping per resource type to prevent cost row fan-out ([#2225](https://github.com/microsoft/finops-toolkit/pull/2225)).
   - Fixed the v1_0 and v1_2 price transforms setting `x_CommitmentDiscountSpendEligibility` from reservation meters and `x_CommitmentDiscountUsageEligibility` from savings plan meters, the opposite of the FOCUS `CommitmentDiscountCategory` value the same transform assigns. The spend column now reports savings plan pricing and the usage column reports reservation pricing. Prices ingested after upgrading carry the corrected values; rows ingested earlier keep the old values until they are reingested ([#2279](https://github.com/microsoft/finops-toolkit/issues/2279)).
+  - Fixed the Azure Resource Graph query pipeline only ingesting the first 1,000 rows of any query (for example, `Recommendations-Microsoft-AdvisorCost`), since Azure Resource Graph caps a single response at 1,000 rows and returns a `$skipToken` for the rest. The pipeline now loops until Azure Resource Graph stops returning a token, writing each page to its own Parquet file ([#2320](https://github.com/microsoft/finops-toolkit/issues/2320)).
 
 ### [FinOps workbooks](workbooks/finops-workbooks-overview.md)
 

src/templates/finops-hub/modules/Microsoft.FinOpsHubs/AzureResourceGraph/README.md

@@ -5,13 +5,13 @@ Query engine for Azure Resource Graph (ARG). Implements the `queries_{engineName
 ## What it provides
 
 - **`azureResourceGraph` dataset** — ADF REST dataset pointing to the ARG API (`/providers/Microsoft.ResourceGraph/resources?api-version=2022-10-01`)
-- **`queries_ResourceGraph_ExecuteQuery` pipeline** — Executes a single ARG query via REST POST and writes results as Parquet to the ingestion container
+- **`queries_ResourceGraph_ExecuteQuery` pipeline** — Executes an ARG query, paging through all results, and writes each page as Parquet to the ingestion container
 
 ## How it works
 
 1. IngestionQueries dispatches to this pipeline via the ADF REST API
-2. The pipeline POSTs the query to the ARG endpoint, appending source metadata columns (`x_SourceName`, `x_SourceType`, `x_SourceProvider`, `x_SourceVersion`) directly in the query text
-3. ARG returns results as JSON; the ADF Copy activity uses the provided `translator` to map columns and writes Parquet to the `ingestionPath`
+2. The pipeline loops, POSTing the query to the ARG endpoint on each iteration, appending source metadata columns (`x_SourceName`, `x_SourceType`, `x_SourceProvider`, `x_SourceVersion`) directly in the query text. A Web activity runs the query first to check for results and read the `$skipToken` continuation token; when there are results, a Copy activity re-runs the same page and uses the provided `translator` to map columns and write it as Parquet to `ingestionPath` (with a `_{page number}` suffix so each page gets its own file)
+3. The loop continues until ARG stops returning a `$skipToken`, since a single ARG response is capped at 1,000 rows
 
 ## Dependencies
 
@@ -20,5 +20,5 @@ Query engine for Azure Resource Graph (ARG). Implements the `queries_{engineName
 
 ## Limitations
 
-- ARG queries are limited to 1,000 rows per page. The current implementation does not paginate, so queries returning more than 1,000 rows will be truncated. Pagination support can be added later if needed.
 - ARG query text has a 10 KB limit.
+- Each page is fetched twice (once to read `$skipToken`, once via Copy to write it), since the ADF REST connector's `paginationRules` can't inject a continuation token into a POST body. This doubles ARG API calls for queries with multiple pages.

src/templates/finops-hub/modules/Microsoft.FinOpsHubs/AzureResourceGraph/app.bicep

@@ -102,115 +102,245 @@ resource pipeline_ExecuteQuery 'Microsoft.DataFactory/factories/pipelines@2018-0
   properties: {
     activities: [
       {
-        name: 'Check Query Has Results'
-        description: 'Run the query with | count to check if there are any results before attempting the full copy.'
-        type: 'WebActivity'
+        name: 'Page Through ARG Results'
+        description: 'Azure Resource Graph caps a single response at 1000 rows and returns a $skipToken when more rows are available. Loop until no token comes back, writing each page to its own Parquet file.'
+        type: 'Until'
         dependsOn: []
-        policy: {
-          timeout: '0.00:05:00'
-          retry: 1
-          retryIntervalInSeconds: 30
-          secureOutput: false
-          secureInput: false
-        }
-        userProperties: []
-        typeProperties: {
-          url: '${environment().resourceManager}providers/Microsoft.ResourceGraph/resources?api-version=2022-10-01'
-          method: 'POST'
-          headers: {
-            'Content-Type': 'application/json'
-          }
-          body: {
-            value: '@concat(\'{ "query": "\', pipeline().parameters.query, \' | count" }\')'
-            type: 'Expression'
-          }
-          authentication: {
-            type: 'MSI'
-            resource: environment().resourceManager
-          }
-        }
-      }
-      {
-        name: 'If Query Has Results'
-        description: 'Only run the copy if the query returned results to avoid schema mapping errors on empty result sets.'
-        type: 'IfCondition'
-        dependsOn: [
-          {
-            activity: 'Check Query Has Results'
-            dependencyConditions: ['Succeeded']
-          }
-        ]
         userProperties: []
         typeProperties: {
           expression: {
-            value: '@greater(int(activity(\'Check Query Has Results\').output.data[0].Count), 0)'
+            value: '@equals(variables(\'hasMorePages\'), false)'
             type: 'Expression'
           }
-          ifTrueActivities: [
+          activities: [
             {
-              name: 'Execute ARG Query'
-              description: 'Execute a single ARG query and write the result to the ingestion container as Parquet.'
-              type: 'Copy'
+              name: 'Build Query Request Body'
+              description: 'Compose the ARG request body for the current page. $skipToken is only included once a continuation token is available.'
+              type: 'SetVariable'
               dependsOn: []
               policy: {
-                timeout: '0.00:10:00'
-                retry: 0
-                retryIntervalInSeconds: 60
                 secureOutput: false
                 secureInput: false
               }
               userProperties: []
               typeProperties: {
-                source: {
-                  type: 'RestSource'
-                  httpRequestTimeout: '00:02:00'
-                  requestInterval: '00.00:00:00.050'
-                  requestMethod: 'POST'
+                variableName: 'requestBody'
+                value: {
                   // Query text is from trusted config/queries/*.json files; no escaping needed.
-                  requestBody: {
-                    value: '@concat(\'{ "query": "\', pipeline().parameters.query, \' | extend x_SourceName=\\"\', pipeline().parameters.querySource, \'\\", x_SourceType=\\"\', pipeline().parameters.queryType, \'\\", x_SourceProvider=\\"\', pipeline().parameters.queryProvider, \'\\", x_SourceVersion=\\"\', pipeline().parameters.queryVersion, \'\\"" }\')'
-                    type: 'Expression'
-                  }
-                  additionalHeaders: {
-                    'Content-Type': 'application/json'
-                  }
+                  value: '@concat(\'{ "query": "\', pipeline().parameters.query, \' | order by x_RecommendationId asc | extend x_SourceName=\\"\', pipeline().parameters.querySource, \'\\", x_SourceType=\\"\', pipeline().parameters.queryType, \'\\", x_SourceProvider=\\"\', pipeline().parameters.queryProvider, \'\\", x_SourceVersion=\\"\', pipeline().parameters.queryVersion, \'\\""\', \', "options": { "$top": 500\', if(equals(variables(\'skipToken\'), \'\'), \'\', concat(\', "$skipToken": "\', variables(\'skipToken\'), \'"\')), \' } }\')'
+                  type: 'Expression'
                 }
-                sink: {
-                  type: 'ParquetSink'
-                  storeSettings: {
-                    type: 'AzureBlobFSWriteSettings'
-                  }
-                  formatSettings: {
-                    type: 'ParquetWriteSettings'
-                  }
+              }
+            }
+            {
+              name: 'Fetch ARG Page'
+              description: 'Run the current page of the query directly (not via the Copy activity) so the response body is available to read $skipToken from. The REST connector\'s paginationRules cannot inject a continuation token into a POST body, so the page is re-run below via Copy to write it.'
+              type: 'WebActivity'
+              dependsOn: [
+                {
+                  activity: 'Build Query Request Body'
+                  dependencyConditions: ['Succeeded']
+                }
+              ]
+              policy: {
+                timeout: '0.00:05:00'
+                retry: 1
+                retryIntervalInSeconds: 30
+                secureOutput: false
+                secureInput: false
+              }
+              userProperties: []
+              typeProperties: {
+                url: '${environment().resourceManager}providers/Microsoft.ResourceGraph/resources?api-version=2022-10-01'
+                method: 'POST'
+                headers: {
+                  'Content-Type': 'application/json'
                 }
-                enableStaging: false
-                translator: {
-                  value: '@pipeline().parameters.translator'
+                body: {
+                  value: '@variables(\'requestBody\')'
                   type: 'Expression'
                 }
+                authentication: {
+                  type: 'MSI'
+                  resource: environment().resourceManager
+                }
               }
-              inputs: [
+            }
+            {
+              name: 'Set Has Results'
+              type: 'SetVariable'
+              dependsOn: [
                 {
-                  referenceName: dataset_azureResourceGraph.name
-                  type: 'DatasetReference'
-                  parameters: {}
+                  activity: 'Fetch ARG Page'
+                  dependencyConditions: ['Succeeded']
                 }
               ]
-              outputs: [
+              policy: {
+                secureOutput: false
+                secureInput: false
+              }
+              userProperties: []
+              typeProperties: {
+                variableName: 'hasResults'
+                value: {
+                  value: '@greater(length(activity(\'Fetch ARG Page\').output.data), 0)'
+                  type: 'Expression'
+                }
+              }
+            }
+            {
+              name: 'Set Skip Token'
+              description: 'Read the continuation token for the next page, if any.'
+              type: 'SetVariable'
+              dependsOn: [
                 {
-                  referenceName: dataset_ingestion.name
-                  type: 'DatasetReference'
-                  parameters: {
-                    blobPath: {
-                      value: '@pipeline().parameters.ingestionPath'
-                      type: 'Expression'
-                    }
-                  }
+                  activity: 'Set Has Results'
+                  dependencyConditions: ['Succeeded']
+                }
+              ]
+              policy: {
+                secureOutput: false
+                secureInput: false
+              }
+              userProperties: []
+              typeProperties: {
+                variableName: 'skipToken'
+                value: {
+                  value: '@if(contains(activity(\'Fetch ARG Page\').output, \'$skipToken\'), activity(\'Fetch ARG Page\').output[\'$skipToken\'], \'\')'
+                  type: 'Expression'
+                }
+              }
+            }
+            {
+              name: 'Set Has More Pages'
+              description: 'Stop the loop once Resource Graph stops returning a continuation token. An empty page can still carry a token, so row count is not used.'
+              type: 'SetVariable'
+              dependsOn: [
+                {
+                  activity: 'Set Skip Token'
+                  dependencyConditions: ['Succeeded']
+                }
+              ]
+              policy: {
+                secureOutput: false
+                secureInput: false
+              }
+              userProperties: []
+              typeProperties: {
+                variableName: 'hasMorePages'
+                value: {
+                  value: '@not(equals(variables(\'skipToken\'), \'\'))'
+                  type: 'Expression'
+                }
+              }
+            }
+            {
+              name: 'If Page Has Results'
+              description: 'Only write the page if it returned results, to avoid schema mapping errors on empty result sets.'
+              type: 'IfCondition'
+              dependsOn: [
+                {
+                  activity: 'Set Has More Pages'
+                  dependencyConditions: ['Succeeded']
                 }
               ]
+              userProperties: []
+              typeProperties: {
+                expression: {
+                  value: '@variables(\'hasResults\')'
+                  type: 'Expression'
+                }
+                ifTrueActivities: [
+                  {
+                    name: 'Write ARG Page'
+                    description: 'Re-run the current page and write the result to the ingestion container as Parquet.'
+                    type: 'Copy'
+                    dependsOn: []
+                    policy: {
+                      timeout: '0.00:10:00'
+                      retry: 2
+                      retryIntervalInSeconds: 60
+                      secureOutput: false
+                      secureInput: false
+                    }
+                    userProperties: []
+                    typeProperties: {
+                      source: {
+                        type: 'RestSource'
+                        httpRequestTimeout: '00:02:00'
+                        requestInterval: '00.00:00:00.050'
+                        requestMethod: 'POST'
+                        requestBody: {
+                          value: '@variables(\'requestBody\')'
+                          type: 'Expression'
+                        }
+                        additionalHeaders: {
+                          'Content-Type': 'application/json'
+                        }
+                      }
+                      sink: {
+                        type: 'ParquetSink'
+                        storeSettings: {
+                          type: 'AzureBlobFSWriteSettings'
+                        }
+                        formatSettings: {
+                          type: 'ParquetWriteSettings'
+                        }
+                      }
+                      enableStaging: false
+                      translator: {
+                        value: '@pipeline().parameters.translator'
+                        type: 'Expression'
+                      }
+                    }
+                    inputs: [
+                      {
+                        referenceName: dataset_azureResourceGraph.name
+                        type: 'DatasetReference'
+                        parameters: {}
+                      }
+                    ]
+                    outputs: [
+                      {
+                        referenceName: dataset_ingestion.name
+                        type: 'DatasetReference'
+                        parameters: {
+                          blobPath: {
+                            // Insert a "_<page number>" suffix before the ".parquet" extension so each page gets a unique file in the same ingestion folder.
+                            value: '@replace(pipeline().parameters.ingestionPath, \'.parquet\', concat(\'_\', variables(\'pageNumber\'), \'.parquet\'))'
+                            type: 'Expression'
+                          }
+                        }
+                      }
+                    ]
+                  }
+                  {
+                    name: 'Increment Page Number'
+                    type: 'SetVariable'
+                    dependsOn: [
+                      {
+                        activity: 'Write ARG Page'
+                        dependencyConditions: ['Succeeded']
+                      }
+                    ]
+                    policy: {
+                      secureOutput: false
+                      secureInput: false
+                    }
+                    userProperties: []
+                    typeProperties: {
+                      variableName: 'pageNumber'
+                      value: {
+                        value: '@string(add(int(variables(\'pageNumber\')), 1))'
+                        type: 'Expression'
+                      }
+                    }
+                  }
+                ]
+              }
             }
           ]
+          timeout: '0.06:00:00'
         }
       }
     ]
@@ -237,6 +367,28 @@ resource pipeline_ExecuteQuery 'Microsoft.DataFactory/factories/pipelines@2018-0
         type: 'Object'
       }
     }
+    variables: {
+      skipToken: {
+        type: 'String'
+        defaultValue: ''
+      }
+      hasMorePages: {
+        type: 'Bool'
+        defaultValue: true
+      }
+      hasResults: {
+        type: 'Bool'
+        defaultValue: false
+      }
+      pageNumber: {
+        type: 'String'
+        defaultValue: '1'
+      }
+      requestBody: {
+        type: 'String'
+        defaultValue: ''
+      }
+    }
     policy: {
       elapsedTimeMetric: {}
     }