Circuit breaker in http output plugins

e2e/file_elasticsearch/config.yml

@@ -9,7 +9,7 @@ pipelines:
     output:
       type: elasticsearch
       endpoints:
-        - http://localhost:9200
+        - http://localhost:19200
       username: SOME_USERNAME
       password: SOME_PASSWORD
       index_format: SOME_INDEX

e2e/file_elasticsearch/config_cb.yml

@@ -0,0 +1,19 @@
+pipelines:
+  file_elasticsearch_cb:
+    input:
+      type: file
+      persistence_mode: async
+      watching_dir: SOME_DIR
+      offsets_file: SOME_FILE
+      offsets_op: reset
+    output:
+      type: elasticsearch
+      endpoints:
+        - http://localhost:19200
+        - http://localhost:19201
+      username: SOME_USERNAME
+      password: SOME_PASSWORD
+      index_format: SOME_INDEX
+      batch_size: 2
+      ban_period: 6s
+      reconnect_interval: 3s

e2e/file_elasticsearch/docker-compose.yml

@@ -1,4 +1,3 @@
-# https://github.com/elastic/start-local/tree/main
 services:
   elasticsearch:
     image: elasticsearch:8.17.0
@@ -20,3 +19,23 @@ services:
       interval: 10s
       timeout: 10s
       retries: 10
+  elasticsearch2:
+    image: elasticsearch:8.17.0
+    container_name: es-local-test-2
+    ports:
+      - "19201:9200"
+    environment:
+      - discovery.type=single-node
+      - ELASTIC_PASSWORD=elastic
+      - xpack.security.enabled=true
+      - xpack.security.http.ssl.enabled=false
+    mem_limit: 1073741824
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "curl --output /dev/null --silent --head --fail -u elastic:elastic http://elasticsearch2:19201",
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 10

e2e/file_elasticsearch/file_elasticsearch.go

@@ -18,13 +18,13 @@ import (
 
 // Config for file-elasticsearch plugin e2e test
 type Config struct {
-	Count    int
-	Endpoint string
-	Pipeline string
-	Username string
-	Password string
-	dir      string
-	index    string
+	Count     int
+	Endpoints []string
+	Pipeline  string
+	Username  string
+	Password  string
+	dir       string
+	index     string
 }
 
 // Configure sets additional fields for input and output plugins
@@ -43,10 +43,12 @@ func (c *Config) Configure(t *testing.T, conf *cfg.Config, pipelineName string)
 	output.Set("ingest_pipeline", c.Pipeline)
 	output.Set("username", c.Username)
 	output.Set("password", c.Password)
-	output.Set("endpoints", []string{c.Endpoint})
+	output.Set("endpoints", c.Endpoints)
 
-	err := createIngestPipeline(c.Endpoint, c.Pipeline, c.Username, c.Password)
-	require.NoError(t, err)
+	for _, endpoint := range c.Endpoints {
+		err := createIngestPipeline(endpoint, c.Pipeline, c.Username, c.Password, c.Count)
+		require.NoError(t, err)
+	}
 }
 
 // Send creates file and writes messages
@@ -63,12 +65,19 @@ func (c *Config) Send(t *testing.T) {
 
 // Validate waits for the message processing to complete
 func (c *Config) Validate(t *testing.T) {
-	err := waitUntilIndexReady(c.Endpoint, c.index, c.Username, c.Password, c.Count, 10, 250*time.Millisecond)
-	require.NoError(t, err)
-	docs, err := getDocumentsFromIndex(c.Endpoint, c.index, c.Username, c.Password)
+	err := waitUntilIndexReady(c.Endpoints, c.index, c.Username, c.Password, c.Count, 10, 250*time.Millisecond)
 	require.NoError(t, err)
-	require.Len(t, docs, c.Count)
-	for _, doc := range docs {
+
+	allDocs := make([]map[string]any, 0, c.Count)
+	for _, endpoint := range c.Endpoints {
+		docs, err := getDocumentsFromIndex(endpoint, c.index, c.Username, c.Password)
+		require.NoError(t, err)
+		t.Logf("endpoint %s: %d docs", endpoint, len(docs))
+		allDocs = append(allDocs, docs...)
+	}
+
+	require.Len(t, allDocs, c.Count)
+	for _, doc := range allDocs {
 		if _, ok := doc["processed_at"]; !ok {
 			t.Errorf("doc %v doesn't have processed_at field", doc)
 		}

e2e/file_elasticsearch/helpers.go

@@ -9,41 +9,55 @@ import (
 	"time"
 )
 
-func createIngestPipeline(elasticURL, pipelineID, username, password string) error {
+func createIngestPipeline(elasticURL, pipelineID, username, password string, retries int) error {
 	url := fmt.Sprintf("%s/_ingest/pipeline/%s", elasticURL, pipelineID)
-
 	pipelineBody := `{"description":"test ingest pipeline","processors":[{"set":{"field":"processed_at","value":"{{_ingest.timestamp}}"}}]}`
 
-	req, err := http.NewRequest(http.MethodPut, url, strings.NewReader(pipelineBody))
-	if err != nil {
-		return fmt.Errorf("failed to create request: %w", err)
-	}
+	var err error
+	for i := range retries {
+		err = func() error {
+			req, err := http.NewRequest(http.MethodPut, url, strings.NewReader(pipelineBody))
+			if err != nil {
+				return fmt.Errorf("failed to create request: %w", err)
+			}
 
-	req.Header.Set("Content-Type", "application/json")
-	if username != "" && password != "" {
-		req.SetBasicAuth(username, password)
-	}
+			req.Header.Set("Content-Type", "application/json")
+			if username != "" && password != "" {
+				req.SetBasicAuth(username, password)
+			}
 
-	client := &http.Client{Timeout: time.Second}
-	resp, err := client.Do(req)
-	if err != nil {
-		return fmt.Errorf("failed to make HTTP request: %w", err)
-	}
-	defer func() { _ = resp.Body.Close() }()
+			client := &http.Client{Timeout: time.Second}
+			resp, err := client.Do(req)
+			if err != nil {
+				return fmt.Errorf("failed to make HTTP request: %w", err)
+			}
+			defer func() { _ = resp.Body.Close() }()
 
-	respBody, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return fmt.Errorf("failed to read body response: %w", err)
-	}
+			respBody, err := io.ReadAll(resp.Body)
+			if err != nil {
+				return fmt.Errorf("failed to read body response: %w", err)
+			}
+
+			if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
+				return fmt.Errorf("unexpected status: %d, body: %s", resp.StatusCode, string(respBody))
+			}
 
-	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
-		return fmt.Errorf("unexpected status: %d, body: %s", resp.StatusCode, string(respBody))
+			return nil
+		}()
+
+		if err == nil {
+			return nil
+		}
+
+		if i < retries-1 {
+			time.Sleep(time.Second)
+		}
 	}
 
-	return nil
+	return fmt.Errorf("can't create pipeline after %d retries: %w", retries, err)
 }
 
-func getDocumentsFromIndex(elasticURL, indexName, username, password string) ([]map[string]interface{}, error) {
+func getDocumentsFromIndex(elasticURL, indexName, username, password string) ([]map[string]any, error) {
 	url := fmt.Sprintf("%s/%s/_search", elasticURL, indexName)
 
 	body := `{"query":{"match_all":{}}}`
@@ -77,7 +91,7 @@ func getDocumentsFromIndex(elasticURL, indexName, username, password string) ([]
 	type searchResponse struct {
 		Hits struct {
 			Hits []struct {
-				Source map[string]interface{} `json:"_source"`
+				Source map[string]any `json:"_source"`
 			} `json:"hits"`
 		} `json:"hits"`
 	}
@@ -87,7 +101,7 @@ func getDocumentsFromIndex(elasticURL, indexName, username, password string) ([]
 		return nil, fmt.Errorf("failed to decode response: %w", err)
 	}
 
-	resultDocs := make([]map[string]interface{}, 0, len(result.Hits.Hits))
+	resultDocs := make([]map[string]any, 0, len(result.Hits.Hits))
 
 	for _, hit := range result.Hits.Hits {
 		resultDocs = append(resultDocs, hit.Source)
@@ -96,63 +110,66 @@ func getDocumentsFromIndex(elasticURL, indexName, username, password string) ([]
 	return resultDocs, nil
 }
 
-func waitUntilIndexReady(elasticURL, indexName, username, password string, minDocs, retries int, delay time.Duration) error {
-	client := &http.Client{
-		Timeout: time.Second,
-	}
-
+func getDocCount(client *http.Client, elasticURL, indexName, username, password string) (int, error) {
 	url := fmt.Sprintf("%s/%s/_count", elasticURL, indexName)
 	req, err := http.NewRequest(http.MethodGet, url, http.NoBody)
 	if err != nil {
-		return fmt.Errorf("failed to create request: %w", err)
+		return 0, fmt.Errorf("failed to create request: %w", err)
 	}
 
 	req.Header.Set("Content-Type", "application/json")
 	if username != "" && password != "" {
 		req.SetBasicAuth(username, password)
 	}
 
-	for i := 0; i < retries; i++ {
-		ok, err := func() (bool, error) {
-			resp, err := client.Do(req)
-			if err != nil {
-				return false, fmt.Errorf("failed to make request: %w", err)
-			}
-			defer func() { _ = resp.Body.Close() }()
+	resp, err := client.Do(req)
+	if err != nil {
+		return 0, fmt.Errorf("failed to make request: %w", err)
+	}
+	defer func() { _ = resp.Body.Close() }()
 
-			if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusServiceUnavailable {
-				return false, nil
-			}
+	if resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusServiceUnavailable {
+		return 0, nil
+	}
 
-			if resp.StatusCode != http.StatusOK {
-				return false, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
-			}
+	if resp.StatusCode != http.StatusOK {
+		return 0, fmt.Errorf("unexpected status code: %d", resp.StatusCode)
+	}
 
-			body, err := io.ReadAll(resp.Body)
-			if err != nil {
-				return false, fmt.Errorf("failed to read response: %w", err)
-			}
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return 0, fmt.Errorf("failed to read response: %w", err)
+	}
 
-			var result map[string]interface{}
-			if err := json.Unmarshal(body, &result); err != nil {
-				return false, fmt.Errorf("failed to decode response: %w", err)
-			}
+	var result map[string]any
+	if err := json.Unmarshal(body, &result); err != nil {
+		return 0, fmt.Errorf("failed to decode response: %w", err)
+	}
 
-			if count, ok := result["count"].(float64); ok {
-				if int(count) >= minDocs {
-					return true, nil
-				}
-			} else {
-				return false, fmt.Errorf("unexpected response structure")
-			}
+	count, ok := result["count"].(float64)
+	if !ok {
+		return 0, fmt.Errorf("unexpected response structure")
+	}
 
-			return false, nil
-		}()
+	return int(count), nil
+}
+
+func waitUntilIndexReady(elasticURLs []string, indexName, username, password string, minDocs, retries int, delay time.Duration) error {
+	client := &http.Client{
+		Timeout: time.Second,
+	}
 
-		if err != nil {
-			return err
+	for range retries {
+		total := 0
+		for _, elasticURL := range elasticURLs {
+			count, err := getDocCount(client, elasticURL, indexName, username, password)
+			if err != nil {
+				return err
+			}
+			total += count
 		}
-		if ok {
+
+		if total >= minDocs {
 			return nil
 		}
 		time.Sleep(delay)

e2e/start_work_test.go

@@ -166,14 +166,25 @@ func TestE2EStabilityWorkCase(t *testing.T) {
 		{
 			name: "file_elasticsearch",
 			e2eTest: &file_elasticsearch.Config{
-				Count:    10,
-				Pipeline: "test-ingest-pipeline",
-				Endpoint: "http://localhost:19200",
-				Username: "elastic",
-				Password: "elastic",
+				Count:     10,
+				Pipeline:  "test-ingest-pipeline",
+				Endpoints: []string{"http://localhost:19200"},
+				Username:  "elastic",
+				Password:  "elastic",
 			},
 			cfgPath: "./file_elasticsearch/config.yml",
 		},
+		{
+			name: "file_elasticsearch_cb",
+			e2eTest: &file_elasticsearch.Config{
+				Count:     10,
+				Pipeline:  "test-ingest-pipeline",
+				Endpoints: []string{"http://localhost:19200", "http://localhost:19201"},
+				Username:  "elastic",
+				Password:  "elastic",
+			},
+			cfgPath: "./file_elasticsearch/config_cb.yml",
+		},
 		{
 			name:    "file_es_split",
 			e2eTest: &file_es_split.Config{},

plugin/output/elasticsearch/README.md

@@ -170,5 +170,18 @@ Process ES response and report errors, if any.
 
 <br>
 
+**`ban_period`** *`cfg.Duration`* *`default=10s`* 
+
+Period for which addresses will be banned in case of unavailability.
+If set to 0, circuit breaker is disabled.
+
+<br>
+
+**`reconnect_interval`** *`cfg.Duration`* *`default=5s`* 
+
+Interval for checking banned endpoints availability.
+
+<br>
+
 
 <br>*Generated using [__insane-doc__](https://github.com/vitkovskii/insane-doc)*