Create custon url config type

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

@@ -5,6 +5,7 @@
 package httpjson
 
 import (
+	"net/url"
 	"regexp"
 	"strings"
 	"text/template"
@@ -35,7 +36,7 @@ type config struct {
 	RetryWaitMin         time.Duration     `config:"retry.wait_min"`
 	RetryWaitMax         time.Duration     `config:"retry.wait_max"`
 	TLS                  *tlscommon.Config `config:"ssl"`
-	URL                  string            `config:"url" validate:"required"`
+	URL                  *URL              `config:"url" validate:"required"`
 	DateCursor           *DateCursor       `config:"date_cursor"`
 }
 
@@ -92,6 +93,21 @@ func (t *Template) Unpack(in string) error {
 	return nil
 }
 
+type URL struct {
+	*url.URL
+}
+
+func (u *URL) Unpack(in string) error {
+	parsed, err := url.Parse(in)
+	if err != nil {
+		return err
+	}
+
+	*u = URL{URL: parsed}
+
+	return nil
+}
+
 // IsEnabled returns true if the `enable` field is set to true in the yaml.
 func (dc *DateCursor) IsEnabled() bool {
 	return dc != nil && (dc.Enabled == nil || *dc.Enabled)

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

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 	"golang.org/x/oauth2/google"
 
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -110,6 +111,16 @@ func TestConfigValidationCase7(t *testing.T) {
 	}
 }
 
+func TestConfigMustFailWithInvalidURL(t *testing.T) {
+	m := map[string]interface{}{
+		"url": "::invalid::",
+	}
+	cfg := common.MustNewConfigFrom(m)
+	conf := defaultConfig()
+	err := cfg.Unpack(&conf)
+	assert.EqualError(t, err, `parse "::invalid::": missing protocol scheme accessing 'url'`)
+}
+
 func TestConfigOauth2Validation(t *testing.T) {
 	cases := []struct {
 		name        string

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

@@ -17,7 +17,7 @@ type dateCursor struct {
 	log             *logp.Logger
 	enabled         bool
 	field           string
-	url             string
+	url             url.URL
 	urlField        string
 	initialInterval time.Duration
 	dateFormat      string
@@ -29,7 +29,7 @@ type dateCursor struct {
 func newDateCursorFromConfig(config config, log *logp.Logger) *dateCursor {
 	c := &dateCursor{
 		enabled: config.DateCursor.IsEnabled(),
-		url:     config.URL,
+		url:     *config.URL.URL,
 	}
 
 	if !c.enabled {
@@ -38,7 +38,6 @@ func newDateCursorFromConfig(config config, log *logp.Logger) *dateCursor {
 
 	c.log = log
 	c.field = config.DateCursor.Field
-	c.url = config.URL
 	c.urlField = config.DateCursor.URLField
 	c.initialInterval = config.DateCursor.InitialInterval
 	c.dateFormat = config.DateCursor.GetDateFormat()
@@ -49,7 +48,7 @@ func newDateCursorFromConfig(config config, log *logp.Logger) *dateCursor {
 
 func (c *dateCursor) getURL() string {
 	if !c.enabled {
-		return c.url
+		return c.url.String()
 	}
 
 	var dateStr string
@@ -60,29 +59,24 @@ func (c *dateCursor) getURL() string {
 		dateStr = c.value
 	}
 
-	url, err := url.Parse(c.url)
-	if err != nil {
-		return c.url
-	}
-
-	q := url.Query()
+	q := c.url.Query()
 
 	var value string
 	if c.valueTpl == nil {
 		value = dateStr
 	} else {
 		buf := new(bytes.Buffer)
 		if err := c.valueTpl.Template.Execute(buf, dateStr); err != nil {
-			return c.url
+			return c.url.String()
 		}
 		value = buf.String()
 	}
 
 	q.Set(c.urlField, value)
 
-	url.RawQuery = q.Encode()
+	c.url.RawQuery = q.Encode()
 
-	return url.String()
+	return c.url.String()
 }
 
 func (c *dateCursor) advance(m common.MapStr) {

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

@@ -178,6 +178,7 @@ func TestHTTPJSONInput(t *testing.T) {
 			expected: []string{
 				`{"@timestamp":"2002-10-02T15:00:00Z","foo":"bar"}`,
 				`{"@timestamp":"2002-10-02T15:00:01Z","foo":"bar"}`,
+				`{"@timestamp":"2002-10-02T15:00:02Z","foo":"bar"}`,
 			},
 		},
 		{
@@ -453,6 +454,13 @@ func dateCursorHandler() http.HandlerFunc {
 				return
 			}
 			_, _ = w.Write([]byte(`{"@timestamp":"2002-10-02T15:00:01Z","foo":"bar"}`))
+		case 2:
+			if r.URL.Query().Get("$filter") != "alertCreationTime ge 2002-10-02T15:00:01Z" {
+				w.WriteHeader(http.StatusBadRequest)
+				_, _ = w.Write([]byte(`{"error":"wrong cursor value"`))
+				return
+			}
+			_, _ = w.Write([]byte(`{"@timestamp":"2002-10-02T15:00:02Z","foo":"bar"}`))
 		}
 		count += 1
 	}

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

@@ -9,7 +9,6 @@ import (
 	"fmt"
 	"net"
 	"net/http"
-	"net/url"
 	"time"
 
 	"github.com/hashicorp/go-retryablehttp"
@@ -107,23 +106,18 @@ func newHTTPJSONInput(config config) (*httpJSONInput, error) {
 func (*httpJSONInput) Name() string { return inputName }
 
 func (in *httpJSONInput) Test(v2.TestContext) error {
-	url, err := url.Parse(in.config.URL)
-	if err != nil {
-		return err
-	}
-
 	port := func() string {
-		if url.Port() != "" {
-			return url.Port()
+		if in.config.URL.Port() != "" {
+			return in.config.URL.Port()
 		}
-		switch url.Scheme {
+		switch in.config.URL.Scheme {
 		case "https":
 			return "443"
 		}
 		return "80"
 	}()
 
-	_, err = net.DialTimeout("tcp", net.JoinHostPort(url.Hostname(), port), time.Second)
+	_, err := net.DialTimeout("tcp", net.JoinHostPort(in.config.URL.Hostname(), port), time.Second)
 	if err != nil {
 		return fmt.Errorf("url %q is unreachable", in.config.URL)
 	}