[elastic/beats] [HttpJson] - Improves request chaining with new 'while' step & client creation per step

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

@@ -93,3 +93,26 @@ type responseChainConfig struct {
 	Transforms transformsConfig `config:"transforms"`
 	Split      *splitConfig     `config:"split"`
 }
+
+func defaultChainConfig() config {
+	chaincfg := defaultConfig()
+	chaincfg.Chain = []chainConfig{
+		{
+			While: &whileConfig{
+				Auth:     chaincfg.Auth,
+				Request:  *chaincfg.Request,
+				Response: responseChainConfig{},
+				Replace:  "",
+				Till:     nil,
+			},
+			Step: &stepConfig{
+				Auth:     chaincfg.Auth,
+				Request:  *chaincfg.Request,
+				Response: responseChainConfig{},
+				Replace:  "",
+			},
+		},
+	}
+
+	return chaincfg
+}

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

@@ -0,0 +1,153 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package httpjson
+
+import (
+	"bytes"
+	"context"
+	"crypto/x509"
+	"io"
+	"net/http"
+	"net/url"
+	"testing"
+
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPolicy_CustomRetryPolicy(t *testing.T) {
+	statusCompleted := `{"status":"completed"}`
+	statusInitiated := `{"status":"cmnsmc"}`
+
+	exp := &valueTpl{}
+	err := exp.Unpack(`[[ eq .last_response.body.status "completed" ]]`)
+	assert.NoError(t, err)
+
+	expErr := &valueTpl{}
+	err = exp.Unpack("")
+	assert.NoError(t, err)
+	type fields struct {
+		fn         Evaluate
+		expression *valueTpl
+		log        *logp.Logger
+	}
+	type args struct {
+		ctx  context.Context
+		resp *http.Response
+		err  error
+	}
+	tests := []struct {
+		name          string
+		fields        fields
+		args          args
+		want          bool
+		expectedError string
+	}{
+		{
+			name: "customRetryPolicy_doNotRetryFurther",
+			fields: fields{
+				fn:         evaluateResponse,
+				expression: exp,
+				log:        logp.NewLogger(""),
+			},
+			args: args{
+				ctx:  context.Background(),
+				resp: getTestResponse(statusCompleted, 200),
+				err:  nil,
+			},
+			want:          false,
+			expectedError: "",
+		},
+		{
+			name: "customRetryPolicy_keepRetrying",
+			fields: fields{
+				fn:         evaluateResponse,
+				expression: exp,
+				log:        logp.NewLogger(""),
+			},
+			args: args{
+				ctx:  context.Background(),
+				resp: getTestResponse(statusInitiated, 200),
+				err:  nil,
+			},
+			want:          true,
+			expectedError: "",
+		},
+		{
+			name: "customRetryPolicy_emptyTemplateError",
+			fields: fields{
+				fn:         evaluateResponse,
+				expression: expErr,
+				log:        logp.NewLogger(""),
+			},
+			args: args{
+				ctx:  context.Background(),
+				resp: getTestResponse(statusCompleted, 200),
+				err:  nil,
+			},
+			want:          false,
+			expectedError: "error while evaluating expression : the template result is empty",
+		},
+		{
+			name: "customRetryPolicy_internalServerError",
+			fields: fields{
+				fn:         evaluateResponse,
+				expression: exp,
+				log:        logp.NewLogger(""),
+			},
+			args: args{
+				ctx:  context.Background(),
+				resp: getTestResponse(statusCompleted, 500),
+				err:  nil,
+			},
+			want:          true,
+			expectedError: "",
+		},
+		{
+			name: "customRetryPolicy_unknownCertError",
+			fields: fields{
+				fn:         evaluateResponse,
+				expression: exp,
+				log:        logp.NewLogger(""),
+			},
+			args: args{
+				ctx:  context.Background(),
+				resp: getTestResponse(statusCompleted, 200),
+				err:  &url.Error{Err: x509.UnknownAuthorityError{}},
+			},
+			want:          false,
+			expectedError: "",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := &Policy{
+				fn:         tt.fields.fn,
+				expression: tt.fields.expression,
+				log:        tt.fields.log,
+			}
+			got, err := p.CustomRetryPolicy(tt.args.ctx, tt.args.resp, tt.args.err)
+			if err != nil {
+				assert.Error(t, err, tt.expectedError)
+			} else {
+				assert.Equal(t, tt.want, got)
+			}
+		})
+	}
+}
+
+func getTestResponse(exprStr string, statusCode int) *http.Response {
+	return &http.Response{
+		StatusCode:    statusCode,
+		Proto:         "HTTP/1.1",
+		ProtoMajor:    1,
+		ProtoMinor:    1,
+		Body:          io.NopCloser(bytes.NewBufferString(exprStr)),
+		ContentLength: int64(len(string(exprStr))),
+		Request:       nil,
+		Header:        make(http.Header, 0),
+	}
+}

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

@@ -64,7 +64,7 @@ func evaluateResponse(expression *valueTpl, data []byte, log *logp.Logger) (bool
 
 	err := json.Unmarshal(data, &dataMap)
 	if err != nil {
-		return false, fmt.Errorf("error while unmarshalling data %w", err)
+		return false, fmt.Errorf("error while unmarshalling data : %w", err)
 	}
 	tr := transformable{}
 	paramCtx := &transformContext{
@@ -75,11 +75,11 @@ func evaluateResponse(expression *valueTpl, data []byte, log *logp.Logger) (bool
 
 	val, err := expression.Execute(paramCtx, tr, nil, log)
 	if err != nil {
-		return false, fmt.Errorf("error while evaluating expression %w", err)
+		return false, fmt.Errorf("error while evaluating expression : %w", err)
 	}
 	result, err := strconv.ParseBool(val)
 	if err != nil {
-		return false, fmt.Errorf("error while parsing boolean value of string %w", err)
+		return false, fmt.Errorf("error while parsing boolean value of string : %w", err)
 	}
 
 	return result, nil

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

@@ -0,0 +1,140 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package httpjson
+
+import (
+	"context"
+	"testing"
+
+	"github.com/elastic/elastic-agent-libs/logp"
+	"github.com/stretchr/testify/assert"
+	"k8s.io/apimachinery/pkg/util/json"
+)
+
+func Test_newChainHTTPClient(t *testing.T) {
+	cfg := defaultChainConfig()
+	ctx := context.Background()
+	log := logp.NewLogger("newChainClientTestLogger")
+
+	type args struct {
+		ctx        context.Context
+		authCfg    *authConfig
+		requestCfg *requestConfig
+		log        *logp.Logger
+		p          []*Policy
+	}
+	tests := []struct {
+		name string
+		args args
+	}{
+		{
+			name: "newChainClientTest",
+			args: args{
+				ctx:        ctx,
+				authCfg:    cfg.Auth,
+				requestCfg: cfg.Request,
+				log:        log,
+				p:          nil,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := newChainHTTPClient(tt.args.ctx, tt.args.authCfg, tt.args.requestCfg, tt.args.log, tt.args.p...)
+			assert.NoError(t, err)
+			assert.NotNil(t, got)
+		})
+	}
+}
+
+func Test_evaluateResponse(t *testing.T) {
+	log := logp.NewLogger("newEvaluateResponseTestLogger")
+	responseTrue, err := json.Marshal(map[string]interface{}{
+		"status": "completed",
+	})
+	assert.NoError(t, err)
+	responseFalse, err := json.Marshal(map[string]interface{}{
+		"status": "initiated",
+	})
+	assert.NoError(t, err)
+
+	type args struct {
+		expression string
+		data       []byte
+		log        *logp.Logger
+	}
+	tests := []struct {
+		name          string
+		args          args
+		expectedError string
+		want          bool
+	}{
+		{
+			name: "newEvaluateResponse_resultIsTrue",
+			args: args{
+				expression: `[[ eq .last_response.body.status "completed" ]]`,
+				data:       responseTrue,
+				log:        log,
+			},
+			want:          true,
+			expectedError: "",
+		},
+		{
+			name: "newEvaluateResponse_resultIsFalse",
+			args: args{
+				expression: `[[ eq .last_response.body.status "completed" ]]`,
+				data:       responseFalse,
+				log:        log,
+			},
+			want:          false,
+			expectedError: "",
+		},
+		{
+			name: "newEvaluateResponse_invalidExpressionError",
+			args: args{
+				expression: `eq .last_response.body.status "completed" ]]`,
+				data:       responseFalse,
+				log:        log,
+			},
+			want:          false,
+			expectedError: "error while parsing boolean value of string : strconv.ParseBool: parsing \"eq .last_response.body.status \\\"completed\\\" ]]\": invalid syntax",
+		},
+		{
+			name: "newEvaluateResponse_emptyExpressionError",
+			args: args{
+				expression: "",
+				data:       responseFalse,
+				log:        log,
+			},
+			want:          false,
+			expectedError: "error while evaluating expression : the template result is empty",
+		},
+		{
+			name: "newEvaluateResponse_incompleteExpressionError",
+			args: args{
+				expression: `[[.last_response.body.status]]`,
+				data:       responseFalse,
+				log:        log,
+			},
+			want:          false,
+			expectedError: "error while parsing boolean value of string : strconv.ParseBool: parsing \"initiated\": invalid syntax",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+
+			expression := &valueTpl{}
+			err := expression.Unpack(tt.args.expression)
+			assert.NoError(t, err)
+
+			got, err := evaluateResponse(expression, tt.args.data, tt.args.log)
+			if err != nil {
+				assert.EqualError(t, err, tt.expectedError)
+			} else {
+				assert.Equal(t, tt.want, got)
+			}
+		})
+	}
+}