Add Filebeat input httpjson

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

@@ -0,0 +1,50 @@
+// 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 (
+	"fmt"
+	"strings"
+
+	"github.com/pkg/errors"
+)
+
+type config struct {
+	APIKey                     string      `config:"api_key"`
+	HTTPClientTimeout          int64       `config:"http_client_timeout"`
+	HTTPMethod                 string      `config:"http_method" validate:"required"`
+	HTTPRequestBody            interface{} `config:"http_request_body"`
+	Interval                   int64       `config:"interval_in_seconds" validate:"required"`
+	JSONObjects                string      `config:"json_objects_array"`
+	PaginationEnable           bool        `config:"pagination_enable"`
+	PaginationExtraBodyContent interface{} `config:"pagination_extra_body_content"`
+	PaginationIDField          string      `config:"pagination_id_field"`
+	PaginationRequestField     string      `config:"pagination_req_field"`
+	PaginationURL              string      `config:"pagination_url"`
+	ServerName                 string      `config:"server_name"`
+	URL                        string      `config:"url" validate:"required"`
+}
+
+func (c *config) Validate() error {
+	if c.Interval < 3600 && c.Interval != 0 {
+		return errors.New("httpjson input: interval_in_seconds must not be less than 3600 - ")
+	}
+	switch strings.ToUpper(c.HTTPMethod) {
+	case "GET":
+		break
+	case "POST":
+		break
+	default:
+		return errors.New(fmt.Sprintf("httpjson input: Invalid http_method, %s - ", c.HTTPMethod))
+	}
+	return nil
+}
+
+func defaultConfig() config {
+	var c config
+	c.HTTPMethod = "GET"
+	c.HTTPClientTimeout = 30
+	return c
+}

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

@@ -0,0 +1,216 @@
+// 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"
+	"os"
+	"sync"
+	"testing"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/elastic/beats/filebeat/channel"
+	"github.com/elastic/beats/filebeat/input"
+	"github.com/elastic/beats/libbeat/beat"
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/tests/resources"
+)
+
+var (
+	once sync.Once
+	host = "httpbin.org"
+)
+
+func testSetup(t *testing.T) {
+	t.Helper()
+
+	once.Do(func() {
+		logp.TestingSetup()
+	})
+}
+
+func isInDockerIntegTestEnv() bool {
+	return os.Getenv("BEATS_DOCKER_INTEGRATION_TEST_ENV") != ""
+}
+
+func runTest(t *testing.T, cfg *common.Config, run func(input *httpjsonInput, out *stubOutleter, t *testing.T)) {
+	if !isInDockerIntegTestEnv() {
+		// Don't test goroutines when using our compose.EnsureUp.
+		defer resources.NewGoroutinesChecker().Check(t)
+	}
+
+	// Setup httpbin environment
+	testSetup(t)
+
+	// Simulate input.Context from Filebeat input runner.
+	inputCtx := newInputContext()
+	defer close(inputCtx.Done)
+
+	// Stub outlet for receiving events generated by the input.
+	eventOutlet := newStubOutlet()
+	defer eventOutlet.Close()
+
+	connector := channel.ConnectorFunc(func(_ *common.Config, _ beat.ClientConfig) (channel.Outleter, error) {
+		return eventOutlet, nil
+	})
+
+	in, err := NewInput(cfg, connector, inputCtx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	input := in.(*httpjsonInput)
+	defer input.Stop()
+
+	run(input, eventOutlet, t)
+}
+
+func newInputContext() input.Context {
+	return input.Context{
+		Done: make(chan struct{}),
+	}
+}
+
+type stubOutleter struct {
+	sync.Mutex
+	cond   *sync.Cond
+	done   bool
+	Events []beat.Event
+}
+
+func newStubOutlet() *stubOutleter {
+	o := &stubOutleter{}
+	o.cond = sync.NewCond(o)
+	return o
+}
+
+func (o *stubOutleter) waitForEvents(numEvents int) ([]beat.Event, bool) {
+	o.Lock()
+	defer o.Unlock()
+
+	for len(o.Events) < numEvents && !o.done {
+		o.cond.Wait()
+	}
+
+	size := numEvents
+	if size >= len(o.Events) {
+		size = len(o.Events)
+	}
+
+	out := make([]beat.Event, size)
+	copy(out, o.Events)
+	return out, len(out) == numEvents
+}
+
+func (o *stubOutleter) Close() error {
+	o.Lock()
+	defer o.Unlock()
+	o.done = true
+	return nil
+}
+
+func (o *stubOutleter) Done() <-chan struct{} { return nil }
+
+func (o *stubOutleter) OnEvent(event beat.Event) bool {
+	o.Lock()
+	defer o.Unlock()
+	o.Events = append(o.Events, event)
+	o.cond.Broadcast()
+	return !o.done
+}
+
+// --- Test Cases
+
+func TestGET(t *testing.T) {
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"api_key":                       "",
+		"http_client_timeout":           30,
+		"http_method":                   "GET",
+		"http_request_body":             nil,
+		"interval_in_seconds":           0,
+		"json_objects_array":            "",
+		"pagination_enable":             false,
+		"pagination_extra_body_content": nil,
+		"pagination_id_field":           "",
+		"pagination_req_field":          "",
+		"pagination_url":                "",
+		"server_name":                   "",
+		"url":                           "http://" + host + "/json",
+	})
+
+	runTest(t, cfg, func(input *httpjsonInput, out *stubOutleter, t *testing.T) {
+		group, _ := errgroup.WithContext(context.Background())
+		group.Go(input.run)
+
+		events, ok := out.waitForEvents(1)
+		if !ok {
+			t.Fatalf("Expected 1 events, but got %d.", len(events))
+		}
+		input.Stop()
+
+		if err := group.Wait(); err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+func TestPOST(t *testing.T) {
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"api_key":                       "",
+		"http_client_timeout":           30,
+		"http_method":                   "POST",
+		"http_request_body":             map[string]interface{}{"test": "abc", "testNested": map[string]interface{}{"testNested1": 123}},
+		"interval_in_seconds":           0,
+		"json_objects_array":            "",
+		"pagination_enable":             false,
+		"pagination_extra_body_content": nil,
+		"pagination_id_field":           "",
+		"pagination_req_field":          "",
+		"pagination_url":                "",
+		"server_name":                   "",
+		"url":                           "http://" + host + "/post",
+	})
+
+	runTest(t, cfg, func(input *httpjsonInput, out *stubOutleter, t *testing.T) {
+		group, _ := errgroup.WithContext(context.Background())
+		group.Go(input.run)
+
+		events, ok := out.waitForEvents(1)
+		if !ok {
+			t.Fatalf("Expected 1 events, but got %d.", len(events))
+		}
+		input.Stop()
+
+		if err := group.Wait(); err != nil {
+			t.Fatal(err)
+		}
+	})
+}
+
+func TestRunStop(t *testing.T) {
+	cfg := common.MustNewConfigFrom(map[string]interface{}{
+		"api_key":                       "",
+		"http_client_timeout":           30,
+		"http_method":                   "GET",
+		"http_request_body":             nil,
+		"interval_in_seconds":           0,
+		"json_objects_array":            "",
+		"pagination_enable":             false,
+		"pagination_extra_body_content": nil,
+		"pagination_id_field":           "",
+		"pagination_req_field":          "",
+		"pagination_url":                "",
+		"server_name":                   "",
+		"url":                           "http://" + host + "/json",
+	})
+
+	runTest(t, cfg, func(input *httpjsonInput, out *stubOutleter, t *testing.T) {
+		input.Run()
+		input.Stop()
+		input.Run()
+		input.Stop()
+	})
+}