http request action

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)
 

_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)

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)
+						}
 					}
 				}
 			}

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)

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"

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

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

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

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"

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

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=

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.

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.

plugin/action/http_request/README.idoc.md

@@ -0,0 +1,8 @@
+# HTTP request plugin
+@introduction
+
+## Example
+@examples
+
+## Config params
+@config-params|description

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.
+
+<br>
+
+**`method`** *`string`* *`default=POST`* *`options=POST|GET|PATCH`* 
+
+HTTP method to use.
+
+<br>
+
+**`address`** *`string`* *`required`* 
+
+URL address to send requests to.
+Example: `http://localhost:8080/api`.com/v1/events`
+
+<br>
+
+**`timeout`** *`cfg.Duration`* *`default=5s`* 
+
+Timeout for the HTTP request.
+
+<br>
+
+**`content_type`** *`string`* *`default=application/json`* 
+
+Value of the Content-Type header.
+
+<br>
+
+**`response_field`** *`string`* 
+
+Field name to store the HTTP response body.
+
+<br>
+
+**`headers`** *`map[string]string`* 
+
+Custom headers to add to the HTTP request.
+
+<br>
+
+**`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.
+
+<br>
+
+**`retention`** *`cfg.Duration`* *`default=50ms`* 
+
+Initial interval for exponential backoff between retries.
+
+<br>
+
+**`retention_exponentially_multiplier`** *`int`* *`default=2`* 
+
+Multiplier for exponential increase of retry interval.
+Each retry interval will be multiplied by this value.
+
+<br>
+
+**`success_codes`** *`[]int`* *`default=200`* 
+
+List of HTTP status codes that are considered successful.
+
+<br>
+
+**`ca_cert`** *`string`* 
+
+Path or content of a PEM-encoded CA file.
+
+<br>
+
+**`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.
+
+<br>
+
+
+<br>*Generated using [__insane-doc__](https://github.com/vitkovskii/insane-doc)*