mirror of
https://github.com/openobserve/goflow2.git
synced 2025-11-03 05:23:32 +00:00
add retry for http send in case of failure
This commit is contained in:
@@ -4,8 +4,11 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"flag"
|
"flag"
|
||||||
|
"fmt"
|
||||||
|
"math"
|
||||||
"net/http"
|
"net/http"
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/netsampler/goflow2/v2/transport"
|
"github.com/netsampler/goflow2/v2/transport"
|
||||||
)
|
)
|
||||||
@@ -54,23 +57,40 @@ func (d *HTTPDriver) Send(key, data []byte) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
req, err := http.NewRequest("POST", d.httpDestination, bytes.NewBuffer(jsonData))
|
maxRetries := 3
|
||||||
if err != nil {
|
delay := time.Millisecond * 500 // initial delay
|
||||||
return err
|
for i := 0; i < maxRetries; i++ {
|
||||||
|
req, err := http.NewRequest("POST", d.httpDestination, bytes.NewBuffer(jsonData))
|
||||||
|
if err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
req.Header.Set("Content-Type", "application/json")
|
||||||
|
req.Header.Set(d.httpAuthHeader, d.httpAuthCredentials)
|
||||||
|
|
||||||
|
client := &http.Client{}
|
||||||
|
resp, err := client.Do(req)
|
||||||
|
if err != nil {
|
||||||
|
if i == maxRetries-1 {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
defer resp.Body.Close()
|
||||||
|
|
||||||
|
// If the status code is not in the 2xx range, retry
|
||||||
|
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||||
|
if i == maxRetries-1 {
|
||||||
|
return fmt.Errorf("failed to send data, status code: %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
time.Sleep(delay * time.Duration(math.Pow(2, float64(i)))) // exponential backoff
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
|
||||||
|
// reset batchData
|
||||||
|
d.batchData = d.batchData[:0]
|
||||||
|
break
|
||||||
}
|
}
|
||||||
|
|
||||||
req.Header.Set("Content-Type", "application/json")
|
|
||||||
req.Header.Set(d.httpAuthHeader, d.httpAuthCredentials)
|
|
||||||
|
|
||||||
client := &http.Client{}
|
|
||||||
resp, err := client.Do(req)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
defer resp.Body.Close()
|
|
||||||
|
|
||||||
// reset batchData
|
|
||||||
d.batchData = d.batchData[:0]
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
Reference in New Issue
Block a user