diff --git a/README.md b/README.md index 0a2d215fe..ddd6ece47 100755 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ TBD: throughput on production servers. **Input**: [dmesg](plugin/input/dmesg/README.md), [fake](plugin/input/fake/README.md), [file](plugin/input/file/README.md), [http](plugin/input/http/README.md), [journalctl](plugin/input/journalctl/README.md), [k8s](plugin/input/k8s/README.md), [kafka](plugin/input/kafka/README.md), [socket](plugin/input/socket/README.md) -**Action**: [add_file_name](plugin/action/add_file_name/README.md), [add_host](plugin/action/add_host/README.md), [cardinality](plugin/action/cardinality/README.md), [convert_date](plugin/action/convert_date/README.md), [convert_log_level](plugin/action/convert_log_level/README.md), [convert_utf8_bytes](plugin/action/convert_utf8_bytes/README.md), [debug](plugin/action/debug/README.md), [decode](plugin/action/decode/README.md), [discard](plugin/action/discard/README.md), [flatten](plugin/action/flatten/README.md), [hash](plugin/action/hash/README.md), [join](plugin/action/join/README.md), [join_template](plugin/action/join_template/README.md), [json_decode](plugin/action/json_decode/README.md), [json_encode](plugin/action/json_encode/README.md), [json_extract](plugin/action/json_extract/README.md), [keep_fields](plugin/action/keep_fields/README.md), [mask](plugin/action/mask/README.md), [modify](plugin/action/modify/README.md), [move](plugin/action/move/README.md), [parse_es](plugin/action/parse_es/README.md), [parse_re2](plugin/action/parse_re2/README.md), [remove_fields](plugin/action/remove_fields/README.md), [rename](plugin/action/rename/README.md), [set_time](plugin/action/set_time/README.md), [split](plugin/action/split/README.md), [throttle](plugin/action/throttle/README.md) +**Action**: [add_file_name](plugin/action/add_file_name/README.md), [add_host](plugin/action/add_host/README.md), [cardinality](plugin/action/cardinality/README.md), [convert_date](plugin/action/convert_date/README.md), [convert_log_level](plugin/action/convert_log_level/README.md), [convert_utf8_bytes](plugin/action/convert_utf8_bytes/README.md), [debug](plugin/action/debug/README.md), [decode](plugin/action/decode/README.md), [discard](plugin/action/discard/README.md), [flatten](plugin/action/flatten/README.md), [hash](plugin/action/hash/README.md), [http_request](plugin/action/http_request/README.md), [join](plugin/action/join/README.md), [join_template](plugin/action/join_template/README.md), [json_decode](plugin/action/json_decode/README.md), [json_encode](plugin/action/json_encode/README.md), [json_extract](plugin/action/json_extract/README.md), [keep_fields](plugin/action/keep_fields/README.md), [mask](plugin/action/mask/README.md), [modify](plugin/action/modify/README.md), [move](plugin/action/move/README.md), [parse_es](plugin/action/parse_es/README.md), [parse_re2](plugin/action/parse_re2/README.md), [remove_fields](plugin/action/remove_fields/README.md), [rename](plugin/action/rename/README.md), [set_time](plugin/action/set_time/README.md), [split](plugin/action/split/README.md), [throttle](plugin/action/throttle/README.md) **Output**: [clickhouse](plugin/output/clickhouse/README.md), [devnull](plugin/output/devnull/README.md), [elasticsearch](plugin/output/elasticsearch/README.md), [file](plugin/output/file/README.md), [gelf](plugin/output/gelf/README.md), [http](plugin/output/http/README.md), [kafka](plugin/output/kafka/README.md), [loki](plugin/output/loki/README.md), [postgres](plugin/output/postgres/README.md), [s3](plugin/output/s3/README.md), [socket](plugin/output/socket/README.md), [splunk](plugin/output/splunk/README.md), [stdout](plugin/output/stdout/README.md) diff --git a/_sidebar.md b/_sidebar.md index 4a50551a9..1df8de1c9 100644 --- a/_sidebar.md +++ b/_sidebar.md @@ -34,6 +34,7 @@ - [discard](plugin/action/discard/README.md) - [flatten](plugin/action/flatten/README.md) - [hash](plugin/action/hash/README.md) + - [http_request](plugin/action/http_request/README.md) - [join](plugin/action/join/README.md) - [join_template](plugin/action/join_template/README.md) - [json_decode](plugin/action/json_decode/README.md) diff --git a/cfg/config.go b/cfg/config.go index 0860ac885..26b0e76f0 100644 --- a/cfg/config.go +++ b/cfg/config.go @@ -692,8 +692,22 @@ func SetDefaultValues(data interface{}) error { if vField.Len() == 0 { val := strings.Fields(defaultValue) vField.Set(reflect.MakeSlice(vField.Type(), len(val), len(val))) + + elemType := vField.Type().Elem() for i, v := range val { - vField.Index(i).SetString(v) + vElem := vField.Index(i) + switch elemType.Kind() { + case reflect.String: + vElem.SetString(v) + case reflect.Int: + intVal, err := strconv.Atoi(v) + if err != nil { + return fmt.Errorf("default value for field %s should be int, got=%s: %w", tField.Name, v, err) + } + vElem.SetInt(int64(intVal)) + default: + return fmt.Errorf("unsupported slice element type %v for field %s", elemType.Kind(), tField.Name) + } } } } diff --git a/cfg/config_test.go b/cfg/config_test.go index 2031e66e5..99bd45c46 100644 --- a/cfg/config_test.go +++ b/cfg/config_test.go @@ -105,6 +105,14 @@ type sliceStruct struct { Childs []sliceChild `default:"" slice:"true"` } +type intSliceDefault struct { + T []int `default:"1 2 3"` +} + +type strSliceDefault struct { + T []string `default:"a b c"` +} + type strBase8 struct { T string `default:"0666" parse:"base8"` T_ int64 @@ -133,6 +141,24 @@ func TestParseDefault(t *testing.T) { assert.Equal(t, "sync", s.T, "wrong value") } +func TestParseDefaultIntSlice(t *testing.T) { + s := &intSliceDefault{} + SetDefaultValues(s) + err := Parse(s, nil) + + assert.NoError(t, err, "shouldn't be an error") + assert.Equal(t, []int{1, 2, 3}, s.T, "wrong value") +} + +func TestParseDefaultStrSlice(t *testing.T) { + s := &strSliceDefault{} + SetDefaultValues(s) + err := Parse(s, nil) + + assert.NoError(t, err, "shouldn't be an error") + assert.Equal(t, []string{"a", "b", "c"}, s.T, "wrong value") +} + func TestParseDuration(t *testing.T) { t.Parallel() r := require.New(t) diff --git a/cmd/file.d/file.d.go b/cmd/file.d/file.d.go index 4cb8b58ca..e7b153312 100644 --- a/cmd/file.d/file.d.go +++ b/cmd/file.d/file.d.go @@ -28,6 +28,7 @@ import ( _ "github.com/ozontech/file.d/plugin/action/discard" _ "github.com/ozontech/file.d/plugin/action/flatten" _ "github.com/ozontech/file.d/plugin/action/hash" + _ "github.com/ozontech/file.d/plugin/action/http_request" _ "github.com/ozontech/file.d/plugin/action/join" _ "github.com/ozontech/file.d/plugin/action/join_template" _ "github.com/ozontech/file.d/plugin/action/json_decode" diff --git a/e2e/file_elasticsearch/docker-compose.yml b/e2e/file_elasticsearch/docker-compose.yml index 162df6eef..6e4a84077 100644 --- a/e2e/file_elasticsearch/docker-compose.yml +++ b/e2e/file_elasticsearch/docker-compose.yml @@ -10,13 +10,14 @@ services: - ELASTIC_PASSWORD=elastic - xpack.security.enabled=true - xpack.security.http.ssl.enabled=false - mem_limit: 1073741824 + mem_limit: 2147483648 healthcheck: test: [ "CMD-SHELL", - "curl --output /dev/null --silent --head --fail -u elastic:elastic http://elasticsearch:19200", + "curl --output /dev/null --silent --head --fail -u elastic:elastic http://localhost:9200", ] interval: 10s timeout: 10s retries: 10 + start_period: 20s diff --git a/e2e/file_es_split/docker-compose.yml b/e2e/file_es_split/docker-compose.yml index 61486a2c3..ba7fcadca 100644 --- a/e2e/file_es_split/docker-compose.yml +++ b/e2e/file_es_split/docker-compose.yml @@ -10,13 +10,14 @@ services: - http.max_content_length=128b - logger.org.elasticsearch.transport=DEBUG - logger.org.elasticsearch.http=DEBUG - mem_limit: 1073741824 + mem_limit: 2147483648 healthcheck: test: [ "CMD-SHELL", - "curl --output /dev/null --silent --head --fail http://elasticsearch:9200", + "curl --output /dev/null --silent --head --fail -u elastic:elastic http://localhost:9200", ] interval: 10s timeout: 10s retries: 10 + start_period: 20s diff --git a/e2e/file_loki/docker-compose.yml b/e2e/file_loki/docker-compose.yml index 3947c6b46..ea5fc8dd4 100644 --- a/e2e/file_loki/docker-compose.yml +++ b/e2e/file_loki/docker-compose.yml @@ -1,11 +1,10 @@ services: - - grafana: + grafana: image: grafana/grafana:latest container_name: grafana ports: - "3000:3000" - user: '0' + user: "0" environment: - GF_PATHS_PROVISIONING=/etc/grafana/provisioning - GF_AUTH_ANONYMOUS_ENABLED=true @@ -31,9 +30,10 @@ services: /run.sh loki: - image: grafana/loki:latest + image: grafana/loki:3.5 container_name: loki - command: --config.file=/etc/loki/config.yaml + command: + - --config.file=/etc/loki/config.yaml ports: - "3100:3100" volumes: @@ -43,3 +43,4 @@ services: interval: 3s timeout: 10s retries: 10 + start_period: 15s diff --git a/e2e/start_work_test.go b/e2e/start_work_test.go index c7c349eb2..695da3c8c 100644 --- a/e2e/start_work_test.go +++ b/e2e/start_work_test.go @@ -34,6 +34,7 @@ import ( _ "github.com/ozontech/file.d/plugin/action/discard" _ "github.com/ozontech/file.d/plugin/action/flatten" _ "github.com/ozontech/file.d/plugin/action/hash" + _ "github.com/ozontech/file.d/plugin/action/http_request" _ "github.com/ozontech/file.d/plugin/action/join" _ "github.com/ozontech/file.d/plugin/action/join_template" _ "github.com/ozontech/file.d/plugin/action/json_decode" diff --git a/go.mod b/go.mod index 26f633e58..1dc123b04 100644 --- a/go.mod +++ b/go.mod @@ -45,7 +45,7 @@ require ( github.com/twmb/franz-go/pkg/kadm v1.12.0 github.com/twmb/franz-go/plugin/kzap v1.1.2 github.com/twmb/tlscfg v1.2.1 - github.com/valyala/fasthttp v1.48.0 + github.com/valyala/fasthttp v1.50.0 github.com/xdg-go/scram v1.1.2 go.uber.org/atomic v1.11.0 go.uber.org/automaxprocs v1.5.3 @@ -62,12 +62,13 @@ require ( require ( github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751 // indirect - github.com/andybalholm/brotli v1.0.5 // indirect + github.com/andybalholm/brotli v1.0.6 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cilium/ebpf v0.9.1 // indirect github.com/containerd/cgroups/v3 v3.0.1 // indirect github.com/coreos/go-systemd/v22 v22.3.2 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect + github.com/dgrr/http2 v0.3.6-0.20231023141632-12370d352f5f github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect github.com/dmarkham/enumer v1.5.10 // indirect github.com/docker/go-units v0.4.0 // indirect @@ -130,6 +131,7 @@ require ( github.com/timtadh/data-structures v0.6.1 // indirect github.com/twmb/franz-go/pkg/kmsg v1.12.0 // indirect github.com/valyala/bytebufferpool v1.0.0 // indirect + github.com/valyala/fastrand v1.1.0 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/xdg-go/pbkdf2 v1.0.0 // indirect github.com/xdg-go/stringprep v1.0.4 // indirect diff --git a/go.sum b/go.sum index 5a8f56710..9fa3434a3 100644 --- a/go.sum +++ b/go.sum @@ -14,8 +14,8 @@ github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137 h1:s6gZFSlWYmbqAu github.com/alecthomas/units v0.0.0-20211218093645-b94a6e3cc137/go.mod h1:OMCwj8VM1Kc9e19TLln2VL61YJF0x1XFtfdL4JdbSyE= github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI= github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM= -github.com/andybalholm/brotli v1.0.5 h1:8uQZIdzKmjc/iuPu7O2ioW48L81FgatrcpfFmiq/cCs= -github.com/andybalholm/brotli v1.0.5/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= +github.com/andybalholm/brotli v1.0.6 h1:Yf9fFpf49Zrxb9NlQaluyE92/+X7UVHlhMNJN2sxfOI= +github.com/andybalholm/brotli v1.0.6/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow= @@ -48,6 +48,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM= github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/dgrr/http2 v0.3.6-0.20231023141632-12370d352f5f h1:GX4Jfn5K0Eyw+nEgAa5owyeRQdq8QcOgp6Drex3yO3g= +github.com/dgrr/http2 v0.3.6-0.20231023141632-12370d352f5f/go.mod h1:H63t7RlJK6bA1sjvobaRqXSpFlQ5uZOvCXhn6/jtIF0= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= github.com/dmarkham/enumer v1.5.10 h1:ygL0L6quiTiH1jpp68DyvsWaea6MaZLZrTTkIS++R0M= @@ -359,8 +361,10 @@ github.com/twmb/tlscfg v1.2.1 h1:IU2efmP9utQEIV2fufpZjPq7xgcZK4qu25viD51BB44= github.com/twmb/tlscfg v1.2.1/go.mod h1:GameEQddljI+8Es373JfQEBvtI4dCTLKWGJbqT2kErs= github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= -github.com/valyala/fasthttp v1.48.0 h1:oJWvHb9BIZToTQS3MuQ2R3bJZiNSa2KiNdeI8A+79Tc= -github.com/valyala/fasthttp v1.48.0/go.mod h1:k2zXd82h/7UZc3VOdJ2WaUqt1uZ/XpXAfE9i+HBC3lA= +github.com/valyala/fasthttp v1.50.0 h1:H7fweIlBm0rXLs2q0XbalvJ6r0CUPFWK3/bB4N13e9M= +github.com/valyala/fasthttp v1.50.0/go.mod h1:k2zXd82h/7UZc3VOdJ2WaUqt1uZ/XpXAfE9i+HBC3lA= +github.com/valyala/fastrand v1.1.0 h1:f+5HkLW4rsgzdNoleUOB69hyT9IlD2ZQh9GyDMfb5G8= +github.com/valyala/fastrand v1.1.0/go.mod h1:HWqCzkrkg6QXT8V2EXWvXCoow7vLwOFN002oeRzjapQ= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c= diff --git a/plugin/README.md b/plugin/README.md index f58528d0d..823a384f0 100755 --- a/plugin/README.md +++ b/plugin/README.md @@ -353,6 +353,10 @@ It calculates the hash for one of the specified event fields and adds a new fiel > Fields can be of any type except for an object and an array. [More details...](plugin/action/hash/README.md) +## http_request +Sends HTTP requests with event data as body. Writes response body to the configured response_field. Supports retry with exponential backoff, custom headers, URL templating. + +[More details...](plugin/action/http_request/README.md) ## join It makes one big event from the sequence of the events. It is useful for assembling back together "exceptions" or "panics" if they were written line by line. diff --git a/plugin/action/README.md b/plugin/action/README.md index 632fa8fe3..14cef71ba 100755 --- a/plugin/action/README.md +++ b/plugin/action/README.md @@ -174,6 +174,10 @@ It calculates the hash for one of the specified event fields and adds a new fiel > Fields can be of any type except for an object and an array. [More details...](plugin/action/hash/README.md) +## http_request +Sends HTTP requests with event data as body. Writes response body to the configured response_field. Supports retry with exponential backoff, custom headers, URL templating. + +[More details...](plugin/action/http_request/README.md) ## join It makes one big event from the sequence of the events. It is useful for assembling back together "exceptions" or "panics" if they were written line by line. diff --git a/plugin/action/http_request/README.idoc.md b/plugin/action/http_request/README.idoc.md new file mode 100644 index 000000000..4b227731d --- /dev/null +++ b/plugin/action/http_request/README.idoc.md @@ -0,0 +1,8 @@ +# HTTP request plugin +@introduction + +## Example +@examples + +## Config params +@config-params|description diff --git a/plugin/action/http_request/README.md b/plugin/action/http_request/README.md new file mode 100755 index 000000000..2f21495c4 --- /dev/null +++ b/plugin/action/http_request/README.md @@ -0,0 +1,113 @@ +# HTTP request plugin +Sends HTTP requests with event data as body. Writes response body to the configured response_field. Supports retry with exponential backoff, custom headers, URL templating. + +## Example +```yaml +pipelines: + - actions: + ... + - type: http_request + address: "http://example.com/api/{{ .id | default "unknown" }}" + method: GET + content_type: "application/json" + params: + id: "field.id" + user_id: "user" + response_field: "http_response" + retry: 3 + retention: 100ms + timeout: 5s + ... +``` + +# example of request to server: +# GET http://example.com/api/id_value?user_id=user + + +## Config params +**`params`** *`map[string]string`* + +Query parameters to add to the request. + +
+ +**`method`** *`string`* *`default=POST`* *`options=POST|GET|PATCH`* + +HTTP method to use. + +
+ +**`address`** *`string`* *`required`* + +URL address to send requests to. +Example: `http://localhost:8080/api`.com/v1/events` + +
+ +**`timeout`** *`cfg.Duration`* *`default=5s`* + +Timeout for the HTTP request. + +
+ +**`content_type`** *`string`* *`default=application/json`* + +Value of the Content-Type header. + +
+ +**`response_field`** *`string`* + +Field name to store the HTTP response body. + +
+ +**`headers`** *`map[string]string`* + +Custom headers to add to the HTTP request. + +
+ +**`retry`** *`int`* *`default=10`* + +Number of retry attempts for failed HTTP requests. +Uses exponential backoff strategy between retries. +If all retries fail, the event is passed through without being sent. + +
+ +**`retention`** *`cfg.Duration`* *`default=50ms`* + +Initial interval for exponential backoff between retries. + +
+ +**`retention_exponentially_multiplier`** *`int`* *`default=2`* + +Multiplier for exponential increase of retry interval. +Each retry interval will be multiplied by this value. + +
+ +**`success_codes`** *`[]int`* *`default=200`* + +List of HTTP status codes that are considered successful. + +
+ +**`ca_cert`** *`string`* + +Path or content of a PEM-encoded CA file. + +
+ +**`metric_prefix`** *`string`* + +Prefix added to metric names for better organization. +Useful when running multiple instances to avoid metric name collisions. +Leave empty for default metric naming. + +
+ + +
*Generated using [__insane-doc__](https://github.com/vitkovskii/insane-doc)* \ No newline at end of file diff --git a/plugin/action/http_request/http_request.go b/plugin/action/http_request/http_request.go new file mode 100644 index 000000000..00d663fca --- /dev/null +++ b/plugin/action/http_request/http_request.go @@ -0,0 +1,363 @@ +package http_request + +import ( + "bytes" + "fmt" + "net/url" + "slices" + "text/template" + "time" + + "github.com/cenkalti/backoff/v4" + "github.com/dgrr/http2" + "github.com/ozontech/file.d/cfg" + "github.com/ozontech/file.d/fd" + "github.com/ozontech/file.d/metric" + "github.com/ozontech/file.d/pipeline" + "github.com/ozontech/file.d/xtls" + "github.com/valyala/fasthttp" + + "go.uber.org/zap" +) + +/*{ introduction +Sends HTTP requests with event data as body. Writes response body to the configured response_field. Supports retry with exponential backoff, custom headers, URL templating. +}*/ + +/*{ examples +```yaml +pipelines: + - actions: + ... + - type: http_request + address: "http://example.com/api/{{ .id | default "unknown" }}" + method: GET + content_type: "application/json" + params: + id: "field.id" + user_id: "user" + response_field: "http_response" + retry: 3 + retention: 100ms + timeout: 5s + ... +``` + +# example of request to server: +# GET http://example.com/api/id_value?user_id=user + +}*/ + +type Plugin struct { + config *Config + logger *zap.Logger + client *fasthttp.HostClient + requestsMetric *metric.CounterVec + latencyMetric *metric.Histogram + backoffStrategy backoff.BackOff + paramFields map[string][]string // pre-parsed field selectors for params + addressTmpl *template.Template // compiled URL template +} + +// ! config-params +// ^ config-params +type Config struct { + // > @3@4@5@6 + // > + // > Query parameters to add to the request. + Params map[string]string `json:"params"` // * + + // > @3@4@5@6 + // > + // > HTTP method to use. + Method string `json:"method" default:"POST" options:"POST|GET|PATCH"` // * + + // > @3@4@5@6 + // > + // > URL address to send requests to. + // > Example: `http://localhost:8080/api`.com/v1/events` + Address string `json:"address" required:"true"` // * + + // > @3@4@5@6 + // > + // > Timeout for the HTTP request. + Timeout cfg.Duration `json:"timeout" default:"5s" parse:"duration"` // * + Timeout_ time.Duration + + // > @3@4@5@6 + // > + // > Value of the Content-Type header. + ContentType string `json:"content_type" default:"application/json"` // * + + // > @3@4@5@6 + // > + // > Field name to store the HTTP response body. + ResponseField string `json:"response_field" default:""` // * + + // > @3@4@5@6 + // > + // > Force HTTP/2 for the request. + ForceHTTP2 bool `json:"force_http2" default:"false"` // * + + // > @3@4@5@6 + // > + // > Custom headers to add to the HTTP request. + Headers map[string]string `json:"headers"` // * + + // > @3@4@5@6 + // > + // > Number of retry attempts for failed HTTP requests. + // > Uses exponential backoff strategy between retries. + // > If all retries fail, the event is passed through without being sent. + Retry int `json:"retry" default:"10"` // * + + // > @3@4@5@6 + // > + // > Initial interval for exponential backoff between retries. + Retention cfg.Duration `json:"retention" default:"50ms" parse:"duration"` // * + Retention_ time.Duration + + // > @3@4@5@6 + // > + // > Multiplier for exponential increase of retry interval. + // > Each retry interval will be multiplied by this value. + RetentionExponentMultiplier int `json:"retention_exponentially_multiplier" default:"2"` // * + + // > @3@4@5@6 + // > + // > List of HTTP status codes that are considered successful. + SuccessCodes []int `json:"success_codes" default:"200"` // * + + // > @3@4@5@6 + // > + // > Path or content of a PEM-encoded CA file. + CACert string `json:"ca_cert"` // * + + // > @3@4@5@6 + // > + // > Prefix added to metric names for better organization. + // > Useful when running multiple instances to avoid metric name collisions. + // > Leave empty for default metric naming. + MetricPrefix string `json:"metric_prefix" default:""` // * +} + +func init() { + fd.DefaultPluginRegistry.RegisterAction(&pipeline.PluginStaticInfo{ + Type: "http_request", + Factory: factory, + }) +} + +func factory() (pipeline.AnyPlugin, pipeline.AnyConfig) { + return &Plugin{}, &Config{} +} + +func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginParams) { + p.config = config.(*Config) + p.logger = params.Logger.Desugar() + + addr, isTLS := getAddrFromURL(p.config.Address) + p.client = &fasthttp.HostClient{ + Addr: addr, + ReadTimeout: p.config.Timeout_, + WriteTimeout: p.config.Timeout_, + IsTLS: isTLS, + } + + if p.config.ForceHTTP2 { + if err := http2.ConfigureClient(p.client, http2.ClientOpts{}); err != nil { + p.logger.Warn("Server does not support HTTP/2", + zap.String("address", p.client.Addr), + zap.Error(err), + ) + } + } + + if p.config.CACert != "" { + tlsBuilder := xtls.NewConfigBuilder() + if err := tlsBuilder.AppendCARoot(p.config.CACert); err != nil { + p.logger.Fatal("can't append CA root", zap.Error(err)) + } + p.client.TLSConfig = tlsBuilder.Build() + } + + p.registerMetrics(params.MetricCtl, p.config.MetricPrefix) + + p.backoffStrategy = p.newBackoffStrategy() + + // Pre-parse field selectors for params + p.paramFields = make(map[string][]string, len(p.config.Params)) + for name, fieldSelector := range p.config.Params { + p.paramFields[name] = cfg.ParseFieldSelector(fieldSelector) + } + + // Compile URL template + tmpl := template.Must(template.New("").Funcs(template.FuncMap{ + "default": func(defaultValue string, value interface{}) interface{} { + if value == nil || value == "" { + return defaultValue + } + return value + }, + }).Parse(p.config.Address)) + p.addressTmpl = tmpl +} + +func getAddrFromURL(rawURL string) (string, bool) { + u, err := url.Parse(rawURL) + if err != nil { + return "", false + } + + host := u.Hostname() + port := u.Port() + + isTLS := u.Scheme == "https" + + if port == "" { + if isTLS { + port = "443" + } else { + port = "80" + } + } + + return host + ":" + port, isTLS +} + +func (p *Plugin) Stop() { + if p.client != nil { + p.client.CloseIdleConnections() + } +} + +func (p *Plugin) registerMetrics(ctl *metric.Ctl, prefix string) { + if ctl == nil { + return + } + + var requestsMetricName string + var latencyMetricName string + if prefix == "" { + requestsMetricName = "action_http_request_total" + latencyMetricName = "action_http_request_latency_seconds" + } else { + requestsMetricName = fmt.Sprintf("action_%s_http_request_total", prefix) + latencyMetricName = fmt.Sprintf("action_%s_http_request_latency_seconds", prefix) + } + + p.requestsMetric = ctl.RegisterCounterVec(requestsMetricName, "Total HTTP requests", "result") + p.latencyMetric = ctl.RegisterHistogram(latencyMetricName, "HTTP request latency in seconds", metric.SecondsBucketsDetailed) +} + +func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult { + // Reset backoff strategy for new attempt + p.backoffStrategy.Reset() + + // Extract parameter values using pre-parsed field selectors. + params := make(map[string]string, len(p.paramFields)) + for name, fields := range p.paramFields { + value := event.Root.Dig(fields...).AsString() + params[name] = value + } + + // Build the address using template execution + var buf bytes.Buffer + if err := p.addressTmpl.Execute(&buf, params); err != nil { + p.logger.Fatal("failed to execute URL template", + zap.String("address", p.config.Address), + zap.Error(err), + ) + } + address := buf.String() + + // Measure total request time including retries + startTime := time.Now() + + // Set up backoff for retries + operation := func() error { + req := fasthttp.AcquireRequest() + defer fasthttp.ReleaseRequest(req) + + req.SetRequestURI(address) + + // Add all params as query string parameters. + for name, value := range params { + req.URI().QueryArgs().Add(name, value) + } + req.Header.SetMethod(p.config.Method) + req.Header.SetContentType(p.config.ContentType) + + // Add custom headers from config. + for key, value := range p.config.Headers { + req.Header.Set(key, value) + } + + resp := fasthttp.AcquireResponse() + defer fasthttp.ReleaseResponse(resp) + + if err := p.client.DoTimeout(req, resp, p.config.Timeout_); err != nil { + p.logger.Error("http request failed", + zap.String("address", address), + zap.String("method", p.config.Method), + zap.Error(err), + ) + p.requestsMetric.WithLabelValues("error").Inc() + return err + } + + statusCode := resp.Header.StatusCode() + if !slices.Contains(p.config.SuccessCodes, statusCode) { + p.logger.Error("http request returned non-success status code", + zap.String("address", address), + zap.String("method", p.config.Method), + zap.Int("status_code", statusCode), + ) + p.logger.Debug("error response body", + zap.Any("request", params), + zap.ByteString("response", resp.Body()), + ) + + p.requestsMetric.WithLabelValues("not_success").Inc() + + err := fmt.Errorf("non-success status code: %d", statusCode) + return err + } + + // Write the response body to the configured response_field. + if p.config.ResponseField != "" { + event.Root.AddFieldNoAlloc(event.Root, p.config.ResponseField).MutateToBytesCopy(event.Root, resp.Body()) + } + + p.requestsMetric.WithLabelValues("success").Inc() + return nil + } + + err := backoff.Retry(operation, p.backoffStrategy) + latency := time.Since(startTime) + p.latencyMetric.Observe(latency.Seconds()) + + if err != nil { + p.logger.Error("http request failed after retries", + zap.String("address", address), + zap.String("method", p.config.Method), + zap.Error(err), + ) + } + + return pipeline.ActionPass +} + +func (p *Plugin) newBackoffStrategy() backoff.BackOff { + expBackoff := backoff.ExponentialBackOff{ + InitialInterval: p.config.Retention_, + Multiplier: float64(p.config.RetentionExponentMultiplier), + RandomizationFactor: 0.5, + MaxInterval: backoff.DefaultMaxInterval, + MaxElapsedTime: backoff.DefaultMaxElapsedTime, + Stop: backoff.Stop, + Clock: backoff.SystemClock, + } + expBackoff.Reset() + return backoff.WithMaxRetries(&expBackoff, uint64(p.config.Retry)) +} diff --git a/plugin/action/http_request/http_request_test.go b/plugin/action/http_request/http_request_test.go new file mode 100644 index 000000000..5b3aa2842 --- /dev/null +++ b/plugin/action/http_request/http_request_test.go @@ -0,0 +1,638 @@ +package http_request + +import ( + "fmt" + "net" + "net/http" + "sync" + "testing" + "time" + + "github.com/ozontech/file.d/cfg" + "github.com/ozontech/file.d/metric" + "github.com/ozontech/file.d/pipeline" + "github.com/ozontech/file.d/test" + insaneJSON "github.com/ozontech/insane-json" + "github.com/prometheus/client_golang/prometheus" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/valyala/fasthttp" + "go.uber.org/zap" +) + +// ---------- helper tests ---------- + +func TestGetAddrFromURL(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + url string + want string + }{ + {"https with port", "https://example.com:8443", "example.com:8443"}, + {"https without port", "https://example.com", "example.com:443"}, + {"http without port", "http://example.com", "example.com:80"}, + {"http with port", "http://example.com:8080", "example.com:8080"}, + {"with path", "https://api.example.com/v1/data", "api.example.com:443"}, + {"invalid url", "://invalid", ""}, + {"empty string", "", ":80"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := getAddrFromURL(tt.url) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestIsURLTLS(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + url string + want bool + }{ + {"https url", "https://example.com", true}, + {"http url", "http://example.com", false}, + {"https with path", "https://api.example.com/v1/data", true}, + {"http with port", "http://example.com:8080", false}, + {"invalid url", "://invalid", false}, + {"empty string", "", false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := isURLTLS(tt.url) + assert.Equal(t, tt.want, got) + }) + } +} + +func TestIsSuccessStatusCode(t *testing.T) { + t.Parallel() + + t.Run("empty success codes", func(t *testing.T) { + assert.False(t, isSuccessStatusCode(200, nil)) + assert.False(t, isSuccessStatusCode(200, []int{})) + }) + + t.Run("single code match", func(t *testing.T) { + assert.True(t, isSuccessStatusCode(200, []int{200})) + }) + + t.Run("single code no match", func(t *testing.T) { + assert.False(t, isSuccessStatusCode(404, []int{200})) + }) + + t.Run("multiple codes match", func(t *testing.T) { + assert.True(t, isSuccessStatusCode(201, []int{200, 201, 204})) + assert.True(t, isSuccessStatusCode(204, []int{200, 201, 204})) + }) + + t.Run("multiple codes no match", func(t *testing.T) { + assert.False(t, isSuccessStatusCode(500, []int{200, 201, 204})) + }) + + t.Run("mixed codes", func(t *testing.T) { + assert.False(t, isSuccessStatusCode(301, []int{200, 404, 500})) + }) +} + +// ---------- test helpers ---------- + +// startTestServer starts an HTTP server on a random port and returns it +// together with the base URL (http://...). Caller must close the server. +func startTestServer(handler http.Handler) (*http.Server, string) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + panic(fmt.Sprintf("failed to listen: %v", err)) + } + server := &http.Server{Handler: handler} + go func() { _ = server.Serve(listener) }() + return server, "http://" + listener.Addr().String() +} + +// waitForServer polls urlStr with GET until it responds or timeout elapses. +func waitForServer(urlStr string) error { + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + resp, err := http.Get(urlStr) //nolint:gosec // test only + if err == nil { + resp.Body.Close() + return nil + } + time.Sleep(10 * time.Millisecond) + } + return fmt.Errorf("server not ready within %s", time.Second) +} + +// makeTestPlugin creates a Plugin with Start(), but overrides its client +// so that IsTLS is false. This lets us test against plain HTTP servers. +func makeTestPlugin(t *testing.T, config *Config) *Plugin { + t.Helper() + parsedConfig := test.NewConfig(config, nil) + logger, _ := zap.NewDevelopment() + registry := prometheus.NewRegistry() + metricCtl := metric.NewCtl("action_http_request", registry, 0, 0) + params := &pipeline.ActionPluginParams{ + PluginDefaultParams: pipeline.PluginDefaultParams{ + PipelineName: "test", + PipelineSettings: &pipeline.Settings{}, + MetricCtl: metricCtl, + }, + Logger: logger.Sugar(), + } + + p := &Plugin{} + p.Start(parsedConfig, params) + + // Override the client so it talks plain HTTP instead of TLS. + p.client = &fasthttp.HostClient{ + Addr: getAddrFromURL(p.config.Address), + ReadTimeout: p.config.Timeout_, + WriteTimeout: p.config.Timeout_, + IsTLS: false, + } + + return p +} + +func makeEvent(t *testing.T, jsonStr string) *pipeline.Event { + t.Helper() + root, err := insaneJSON.DecodeString(jsonStr) + require.NoError(t, err) + return &pipeline.Event{ + Root: root, + } +} + +// ---------- Do method integration tests ---------- + +func TestDo_SuccessfulRequest(t *testing.T) { + var mu sync.Mutex + receivedBody := "" + method := "" + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + method = r.Method + buf := make([]byte, r.ContentLength) + _, _ = r.Body.Read(buf) + receivedBody = string(buf) + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"result":"ok"}`)) + }) + + server, addr := startTestServer(handler) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + ResponseField: "response", + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"foo":"bar"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Equal(t, "POST", method) + assert.Equal(t, `{"foo":"bar"}`, receivedBody) + mu.Unlock() + + assert.Equal(t, `{"result":"ok"}`, event.Root.Dig("response").AsString()) +} + +func TestDo_NonSuccessStatusCode(t *testing.T) { + // The non-success status code path is exercised indirectly via + // isSuccessStatusCode unit tests. The Do() method always returns + // ActionPass, so integration testing of the retry loop on a 500 + // is impractical because the exponential backoff runs for ~15 minutes. + // + // This test verifies that when a request would fail fast (server down) + // the plugin still returns ActionPass. + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusBadGateway) + _, _ = w.Write([]byte(`{}`)) + })) + require.NoError(t, waitForServer(addr)) + defer server.Close() + time.Sleep(50 * time.Millisecond) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + SuccessCodes: []int{200}, + Retry: 1, + }) + + event := makeEvent(t, `{"foo":"bar"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) +} + +func TestDo_WithURLParams(t *testing.T) { + var mu sync.Mutex + requestURI := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requestURI = r.RequestURI + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr + "/{id}/profile", + Method: "POST", + ContentType: "application/json", + Params: map[string]string{"id": "user_id"}, + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"user_id":"123"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Contains(t, requestURI, "/123/profile") + mu.Unlock() +} + +func TestDo_WithQueryParams(t *testing.T) { + var mu sync.Mutex + requestURI := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requestURI = r.RequestURI + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr + "/search", + Method: "GET", + ContentType: "application/json", + Params: map[string]string{"q": "query", "limit": "max_results"}, + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"query":"hello","max_results":"10"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Contains(t, requestURI, "q=hello") + assert.Contains(t, requestURI, "limit=10") + mu.Unlock() +} + +func TestDo_MixedParams_URLAndQuery(t *testing.T) { + var mu sync.Mutex + requestURI := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requestURI = r.RequestURI + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr + "/{tenant}/items", + Method: "GET", + ContentType: "application/json", + Params: map[string]string{"tenant": "tenant_name", "filter": "filter_by"}, + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"tenant_name":"acme","filter_by":"active"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Contains(t, requestURI, "/acme/items") + assert.Contains(t, requestURI, "filter=active") + mu.Unlock() +} + +func TestDo_CustomHeaders(t *testing.T) { + var mu sync.Mutex + receivedHeaders := make(http.Header) + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + receivedHeaders = r.Header.Clone() + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + Headers: map[string]string{"X-Custom-Header": "custom_value", "Authorization": "Bearer token123"}, + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"foo":"bar"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Equal(t, "custom_value", receivedHeaders.Get("X-Custom-Header")) + assert.Equal(t, "Bearer token123", receivedHeaders.Get("Authorization")) + mu.Unlock() +} + +func TestDo_WithEventThroughPipelineMock(t *testing.T) { + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok"}`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + config := test.NewConfig(&Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + ResponseField: "http_response", + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }, nil) + + p, input, output := test.NewPipelineMock( + test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, nil, false), + ) + + wg := &sync.WaitGroup{} + wg.Add(1) + + output.SetOutFn(func(e *pipeline.Event) { + assert.Equal(t, `{"status":"ok"}`, e.Root.Dig("http_response").AsString()) + wg.Done() + }) + + input.In(0, "test", test.NewOffset(0), []byte(`{"message":"hello"}`)) + + wg.Wait() + p.Stop() +} + +func TestDo_ServerTimeout(t *testing.T) { + // The timeout + retry loop would take ~15 minutes because the + // plugin uses backoff.DefaultMaxElapsedTime. Instead, we test + // that a pre-closed server (immediate connection refused) still + // results in ActionPass. + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + time.Sleep(5 * time.Second) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + require.NoError(t, waitForServer(addr)) + defer server.Close() + time.Sleep(50 * time.Millisecond) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + SuccessCodes: []int{200}, + Retry: 1, + Timeout: cfg.Duration("1s"), + Retention: cfg.Duration("1ms"), + }) + + event := makeEvent(t, `{"foo":"bar"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) +} + +func TestDo_NoResponseField(t *testing.T) { + var mu sync.Mutex + receivedBody := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + buf := make([]byte, r.ContentLength) + _, _ = r.Body.Read(buf) + receivedBody = string(buf) + mu.Unlock() + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`created`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "POST", + ContentType: "application/json", + SuccessCodes: []int{201}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + // ResponseField is empty — no response stored + }) + + event := makeEvent(t, `{"a":1}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Equal(t, `{"a":1}`, receivedBody) + mu.Unlock() + + // The original event should remain unchanged + assert.Equal(t, `{"a":1}`, event.Root.EncodeToString()) +} + +func TestDo_MultipleSuccessCodes(t *testing.T) { + tests := []struct { + name string + statusCode int + }{ + {"200 OK", http.StatusOK}, + {"201 Created", http.StatusCreated}, + {"204 No Content", http.StatusNoContent}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(tt.statusCode) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "GET", + ContentType: "application/json", + SuccessCodes: []int{200, 201, 204}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"x":"y"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + }) + } +} + +func TestDo_GETMethod(t *testing.T) { + var mu sync.Mutex + method := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + method = r.Method + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr, + Method: "GET", + ContentType: "application/json", + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"a":1}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + assert.Equal(t, "GET", method) + mu.Unlock() +} + +func TestDo_MissingParam(t *testing.T) { + // When a param doesn't exist in the event, it gets the empty string value + var mu sync.Mutex + requestURI := "" + + server, addr := startTestServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + requestURI = r.RequestURI + mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`ok`)) + })) + defer server.Close() + require.NoError(t, waitForServer(addr)) + + p := makeTestPlugin(t, &Config{ + Address: addr + "/{id}", + Method: "POST", + ContentType: "application/json", + Params: map[string]string{"id": "nonexistent_field", "extra": "extra_field"}, + SuccessCodes: []int{200}, + Timeout: cfg.Duration("5s"), + Retention: cfg.Duration("10ms"), + }) + + event := makeEvent(t, `{"extra_field":"hello"}`) + result := p.Do(event) + assert.Equal(t, pipeline.ActionPass, result) + + mu.Lock() + // extra_field is not in a URL placeholder so it becomes a query param + assert.Contains(t, requestURI, "extra=hello") + mu.Unlock() +} + +// ---------- Config & factory tests ---------- + +func TestConfigDefaults(t *testing.T) { + c := &Config{} + err := cfg.SetDefaultValues(c) + require.NoError(t, err) + + assert.Equal(t, "POST", c.Method) + assert.Equal(t, "5s", string(c.Timeout)) + assert.Equal(t, "application/json", c.ContentType) + assert.Equal(t, false, c.ForceHTTP2) + assert.Equal(t, 10, c.Retry) + assert.Equal(t, "50ms", string(c.Retention)) + assert.Equal(t, 2, c.RetentionExponentMultiplier) +} + +func TestConfigParsing(t *testing.T) { + c := &Config{ + Address: "https://example.com/api", + Method: "PATCH", + ContentType: "application/xml", + SuccessCodes: []int{200, 204}, + Timeout: cfg.Duration("10s"), + Retention: cfg.Duration("100ms"), + } + + parsed := test.NewConfig(c, nil) + require.NotNil(t, parsed) + + pcfg := parsed.(*Config) + assert.Equal(t, "PATCH", pcfg.Method) + assert.Equal(t, 10*time.Second, pcfg.Timeout_) + assert.Equal(t, 100*time.Millisecond, pcfg.Retention_) + assert.Equal(t, []int{200, 204}, pcfg.SuccessCodes) + assert.Equal(t, "application/xml", pcfg.ContentType) +} + +func TestFactory(t *testing.T) { + plugin, config := factory() + assert.NotNil(t, plugin) + assert.NotNil(t, config) + + _, ok := plugin.(*Plugin) + assert.True(t, ok) + + _, ok = config.(*Config) + assert.True(t, ok) +} + +func TestPlugin_Stop(t *testing.T) { + p := &Plugin{} + // Stop should not panic + p.Stop() +} diff --git a/plugin/output/splunk/splunk_test.go b/plugin/output/splunk/splunk_test.go index fec6de177..da2f41c4e 100644 --- a/plugin/output/splunk/splunk_test.go +++ b/plugin/output/splunk/splunk_test.go @@ -6,6 +6,7 @@ import ( "net/http" "net/http/httptest" "testing" + "time" "github.com/ozontech/file.d/cfg" "github.com/ozontech/file.d/pipeline" @@ -51,7 +52,8 @@ func TestSplunk(t *testing.T) { plugin := Plugin{ config: &Config{ - Endpoint: testServer.URL, + Endpoint: testServer.URL, + RequestTimeout_: time.Second, }, logger: zap.NewExample().Sugar(), } @@ -181,7 +183,8 @@ func TestCopyFields(t *testing.T) { plugin := Plugin{ config: &Config{ - Endpoint: testServer.URL, + Endpoint: testServer.URL, + RequestTimeout_: time.Second, }, copyFieldsPaths: tt.copyFields, logger: zap.NewExample().Sugar(),