feat(loki-canary): Add support to push logs directly to Loki.

cmd/loki-canary/main.go

@@ -15,6 +15,7 @@ import (
 	"os/signal"
 
 	"github.com/go-kit/log"
+	"github.com/grafana/dskit/backoff"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"github.com/prometheus/common/config"
 	"github.com/prometheus/common/version"
@@ -25,6 +26,12 @@ import (
 	_ "github.com/grafana/loki/pkg/util/build"
 )
 
+const (
+	defaultMinBackoff = 500 * time.Millisecond
+	defaultMaxBackoff = 5 * time.Minute
+	defaultMaxRetries = 10
+)
+
 type canary struct {
 	lock sync.Mutex
 
@@ -50,6 +57,9 @@ func main() {
 	pass := flag.String("pass", "", "Loki password. This credential should have both read and write permissions to Loki endpoints")
 	tenantID := flag.String("tenant-id", "", "Tenant ID to be set in X-Scope-OrgID header.")
 	writeTimeout := flag.Duration("write-timeout", 10*time.Second, "How long to wait write response from Loki")
+	writeMinBackoff := flag.Duration("write-min-backoff", defaultMinBackoff, "Initial backoff time before first retry ")
+	writeMaxBackoff := flag.Duration("write-max-backoff", defaultMaxBackoff, "Maximum backoff time between retries ")
+	writeMaxRetries := flag.Int("write-max-retries", defaultMaxRetries, "Maximum number of retries when push a log entry ")
 	queryTimeout := flag.Duration("query-timeout", 10*time.Second, "How long to wait for a query response from Loki")
 
 	interval := flag.Duration("interval", 1000*time.Millisecond, "Duration between log entries")
@@ -134,6 +144,11 @@ func main() {
 		w = os.Stdout
 
 		if *push {
+			backoffCfg := backoff.Config{
+				MinBackoff: *writeMinBackoff,
+				MaxBackoff: *writeMaxBackoff,
+				MaxRetries: *writeMaxRetries,
+			}
 			push, err := writer.NewPush(
 				*addr,
 				*tenantID,
@@ -144,6 +159,7 @@ func main() {
 				tlsConfig,
 				*caFile,
 				*user, *pass,
+				&backoffCfg,
 				log.NewLogfmtLogger(os.Stdout),
 			)
 			if err != nil {

docs/sources/operations/loki-canary.md

@@ -367,4 +367,10 @@ All options:
         Print this build's version information
   -wait duration
         Duration to wait for log entries before reporting them as lost (default 1m0s)
+  -write-max-backoff duration
+    	Maximum backoff time between retries  (default 5m0s)
+  -write-max-retries int
+    	Maximum number of retries when push a log entry  (default 10)
+  -write-min-backoff duration
+    	Initial backoff time before first retry  (default 500ms)
 ```

pkg/canary/writer/push.go

@@ -19,6 +19,7 @@ import (
 	"github.com/prometheus/common/config"
 	"github.com/prometheus/common/model"
 
+	"github.com/grafana/dskit/backoff"
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/util/build"
 )
@@ -50,6 +51,9 @@ type Push struct {
 
 	// Will add these label to the logs pushed to loki
 	labelName, labelValue, streamName, streamValue string
+
+	// push retry and backoff
+	backoff *backoff.Config
 }
 
 // NewPush creates an instance of `Push` which writes logs directly to given `lokiAddr`
@@ -62,6 +66,7 @@ func NewPush(
 	tlsCfg *tls.Config,
 	caFile string,
 	username, password string,
+	backoffCfg *backoff.Config,
 	logger log.Logger,
 ) (*Push, error) {
 
@@ -104,6 +109,7 @@ func NewPush(
 		streamValue: streamValue,
 		username:    username,
 		password:    password,
+		backoff:     backoffCfg,
 	}, nil
 }
 
@@ -185,24 +191,38 @@ func (p *Push) send(ctx context.Context, payload []byte) error {
 		req.SetBasicAuth(p.username, p.password)
 	}
 
-	resp, err := p.httpClient.Do(req)
-	if err != nil {
-		return fmt.Errorf("failed to push payload: %w", err)
-	}
-	defer func() {
-		if err := resp.Body.Close(); err != nil {
+	backoff := backoff.New(ctx, *p.backoff)
+	var resp *http.Response
+
+	// send log with retry
+	for {
+		resp, err = p.httpClient.Do(req)
+		if err != nil {
+			return fmt.Errorf("failed to push payload: %w", err)
+		}
+		status := resp.StatusCode
+
+		if status/100 != 2 {
+			scanner := bufio.NewScanner(io.LimitReader(resp.Body, defaultMaxReponseBufferLen))
+			line := ""
+			if scanner.Scan() {
+				line = scanner.Text()
+			}
+			err = fmt.Errorf("server returned HTTP status %s (%d): %s", resp.Status, status, line)
+		}
+
+		if err = resp.Body.Close(); err != nil {
 			level.Error(p.logger).Log("msg", "failed to close response body", "error", err)
 		}
-	}()
 
-	if resp.StatusCode/100 != 2 {
-		scanner := bufio.NewScanner(io.LimitReader(resp.Body, defaultMaxReponseBufferLen))
-		line := ""
-		if scanner.Scan() {
-			line = scanner.Text()
+		if status > 0 && status != 429 && status/100 != 5 {
+			break
 		}
-		return fmt.Errorf("server returned HTTP status %s (%d): %s", resp.Status, resp.StatusCode, line)
-	}
 
-	return nil
+		if !backoff.Ongoing() {
+			break
+		}
+		level.Info(p.logger).Log("msg", "retrying as server returned non successful error", "status", status)
+	}
+	return err
 }

pkg/canary/writer/push_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/grafana/dskit/backoff"
 	"github.com/grafana/loki/pkg/logproto"
 	"github.com/grafana/loki/pkg/util"
 )
@@ -29,14 +30,19 @@ const (
 func Test_Push(t *testing.T) {
 	// create dummy loki server
 	responses := make(chan response, 1) // buffered not to block the response handler
+	backoff := backoff.Config{
+		MinBackoff: 300 * time.Millisecond,
+		MaxBackoff: 5 * time.Minute,
+		MaxRetries: 10,
+	}
 
 	// mock loki server
 	mock := httptest.NewServer(createServerHandler(responses))
 	require.NotNil(t, mock)
 	defer mock.Close()
 
 	// without TLS
-	push, err := NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "stream", "stdout", nil, "", "", "", log.NewNopLogger())
+	push, err := NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "stream", "stdout", nil, "", "", "", &backoff, log.NewNopLogger())
 	require.NoError(t, err)
 	ts, payload := testPayload()
 	n, err := push.Write([]byte(payload))
@@ -46,7 +52,7 @@ func Test_Push(t *testing.T) {
 	assertResponse(t, resp, false, labelSet("name", "loki-canary", "stream", "stdout"), ts, payload)
 
 	// with basic Auth
-	push, err = NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "stream", "stdout", nil, "", testUsername, testPassword, log.NewNopLogger())
+	push, err = NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "stream", "stdout", nil, "", testUsername, testPassword, &backoff, log.NewNopLogger())
 	require.NoError(t, err)
 	ts, payload = testPayload()
 	n, err = push.Write([]byte(payload))
@@ -56,7 +62,7 @@ func Test_Push(t *testing.T) {
 	assertResponse(t, resp, true, labelSet("name", "loki-canary", "stream", "stdout"), ts, payload)
 
 	// with custom labels
-	push, err = NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "pod", "abc", nil, "", testUsername, testPassword, log.NewNopLogger())
+	push, err = NewPush(mock.Listener.Addr().String(), "test1", 2*time.Second, config.DefaultHTTPClientConfig, "name", "loki-canary", "pod", "abc", nil, "", testUsername, testPassword, &backoff, log.NewNopLogger())
 	require.NoError(t, err)
 	ts, payload = testPayload()
 	n, err = push.Write([]byte(payload))