[filebeat][httpjson][bugfix] - root request pagination now working with chaining

CHANGELOG.next.asciidoc

@@ -18,7 +18,7 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 
 *Filebeat*
-
+ - [Httpjson]- Fix for pagination at root level not working when used with chaining.
 
 *Heartbeat*
 - Browser monitors (beta) no write to the `synthetics-*` index prefix. {pull}32064[32064]

x-pack/filebeat/input/httpjson/chain.go

@@ -69,10 +69,10 @@ type chainConfig struct {
 // will contain replace string with original URL to make a skeleton for the
 // call request.
 type stepConfig struct {
-	Auth     *authConfig         `config:"auth"`
-	Request  requestConfig       `config:"request" validate:"required"`
-	Response responseChainConfig `config:"response,omitempty"`
-	Replace  string              `config:"replace,omitempty"`
+	Auth     *authConfig          `config:"auth"`
+	Request  *requestConfig       `config:"request" validate:"required"`
+	Response *responseChainConfig `config:"response,omitempty"`
+	Replace  string               `config:"replace,omitempty"`
 }
 
 // whileConfig will contain basic properties like auth parameters, request parameters,
@@ -82,11 +82,11 @@ type stepConfig struct {
 // expression is evaluated to "true" or request.retry.max_attempts is exhausted. If
 // request.retry.max_attempts is not specified , the max_attempts is always 1.
 type whileConfig struct {
-	Auth     *authConfig         `config:"auth"`
-	Request  requestConfig       `config:"request" validate:"required"`
-	Response responseChainConfig `config:"response,omitempty"`
-	Replace  string              `config:"replace,omitempty"`
-	Until    *valueTpl           `config:"until" validate:"required"`
+	Auth     *authConfig          `config:"auth"`
+	Request  *requestConfig       `config:"request" validate:"required"`
+	Response *responseChainConfig `config:"response,omitempty"`
+	Replace  string               `config:"replace,omitempty"`
+	Until    *valueTpl            `config:"until" validate:"required"`
 }
 
 type responseChainConfig struct {
@@ -100,13 +100,13 @@ func defaultChainConfig() config {
 		{
 			While: &whileConfig{
 				Auth:     chaincfg.Auth,
-				Request:  *chaincfg.Request,
-				Response: responseChainConfig{},
+				Request:  chaincfg.Request,
+				Response: &responseChainConfig{},
 			},
 			Step: &stepConfig{
 				Auth:     chaincfg.Auth,
-				Request:  *chaincfg.Request,
-				Response: responseChainConfig{},
+				Request:  chaincfg.Request,
+				Response: &responseChainConfig{},
 			},
 		},
 	}

x-pack/filebeat/input/httpjson/input_test.go

@@ -578,6 +578,36 @@ func TestInput(t *testing.T) {
 				`{"space":{"cake":"pumpkin"}}`,
 			},
 		},
+		{
+			name:        "Test pagination when used with chaining",
+			setupServer: newChainPaginationTestServer(httptest.NewServer),
+			baseConfig: map[string]interface{}{
+				"interval":       1,
+				"request.method": http.MethodGet,
+				"response.pagination": []interface{}{
+					map[string]interface{}{
+						"set": map[string]interface{}{
+							"target":                 "url.value",
+							"value":                  "[[.last_response.body.nextLink]]",
+							"fail_on_template_error": true,
+						},
+					},
+				},
+				"chain": []interface{}{
+					map[string]interface{}{
+						"step": map[string]interface{}{
+							"request.method": http.MethodGet,
+							"replace":        "$.records[:].id",
+						},
+					},
+				},
+			},
+			handler: defaultHandler(http.MethodGet, ""),
+			expected: []string{
+				`{"hello":{"world":"moon"}}`,
+				`{"space":{"cake":"pumpkin"}}`,
+			},
+		},
 	}
 
 	for _, testCase := range testCases {
@@ -668,6 +698,34 @@ func newChainTestServer(
 	}
 }
 
+func newChainPaginationTestServer(
+	newServer func(http.Handler) *httptest.Server,
+) func(*testing.T, http.HandlerFunc, map[string]interface{}) {
+	return func(t *testing.T, h http.HandlerFunc, config map[string]interface{}) {
+		registerPaginationTransforms()
+		var serverURL string
+		r := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			switch r.URL.Path {
+			case "/":
+				link := serverURL + "/link2"
+				value := fmt.Sprintf(`{"records":[{"id":1}], "nextLink":"%s"}`, link)
+				fmt.Fprintln(w, value)
+			case "/1":
+				fmt.Fprintln(w, `{"hello":{"world":"moon"}}`)
+			case "/link2":
+				fmt.Fprintln(w, `{"records":[{"id":2}]}`)
+			case "/2":
+				fmt.Fprintln(w, `{"space":{"cake":"pumpkin"}}`)
+			}
+		})
+		server := httptest.NewServer(r)
+		config["request.url"] = server.URL
+		serverURL = server.URL
+		config["chain.0.step.request.url"] = server.URL + "/$.records[:].id"
+		t.Cleanup(func() { registeredTransforms = newRegistry() })
+	}
+}
+
 func newV2Context() (v2.Context, func()) {
 	ctx, cancel := context.WithCancel(context.Background())
 	return v2.Context{