Merge b8473c7 into 703970e

log/file_test.go

@@ -108,7 +108,7 @@ func TestFileHookFromConfigLine(t *testing.T) {
 			}
 
 			require.NoError(t, err)
-			assert.NotNil(t, res.(*fileHook).w)
+			assert.NotNil(t, res.(*fileHook).w) //nolint:forcetypeassert
 		})
 	}
 }

log/loki_test.go

@@ -1,8 +1,13 @@
 package log
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
 	"testing"
 	"time"
 
@@ -69,23 +74,27 @@ func TestSyslogFromConfigLine(t *testing.T) {
 	for _, test := range tests {
 		test := test
 		t.Run(test.line, func(t *testing.T) {
+			t.Parallel()
 			// no parallel because this is way too fast and parallel will only slow it down
 
 			res, err := LokiFromConfigLine(nil, test.line)
 			if test.err {
 				require.Error(t, err)
 				return
 			}
+			hook, ok := res.(*lokiHook)
+			require.True(t, ok)
 			require.NoError(t, err)
 
-			test.res.client = res.(*lokiHook).client
-			test.res.ch = res.(*lokiHook).ch
+			test.res.client = hook.client
+			test.res.ch = hook.ch
 			require.Equal(t, &test.res, res)
 		})
 	}
 }
 
 func TestLogEntryMarshal(t *testing.T) {
+	t.Parallel()
 	entry := logEntry{
 		t:   9223372036854775807, // the actual max
 		msg: "something",
@@ -98,6 +107,7 @@ func TestLogEntryMarshal(t *testing.T) {
 }
 
 func TestFilterLabels(t *testing.T) {
+	t.Parallel()
 	cases := []struct {
 		allowedLabels  []string
 		labels         map[string]string
@@ -131,6 +141,7 @@ func TestFilterLabels(t *testing.T) {
 	for i, c := range cases {
 		c := c
 		t.Run(fmt.Sprint(i), func(t *testing.T) {
+			t.Parallel()
 			h := &lokiHook{}
 			h.allowedLabels = c.allowedLabels
 			result := h.filterLabels(c.labels, c.msg)
@@ -139,3 +150,43 @@ func TestFilterLabels(t *testing.T) {
 		})
 	}
 }
+
+func TestLokiFlushingOnStop(t *testing.T) {
+	t.Parallel()
+	receivedData := make(chan string, 1)
+	srv := httptest.NewServer(
+		http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
+			b, err := io.ReadAll(req.Body)
+			if err != nil {
+				t.Fatal(err)
+			}
+			receivedData <- string(b)
+			close(receivedData) // this is mostly as this should never be called twice, so panicking here in that case
+		}),
+	)
+	configLine := fmt.Sprintf("loki=%s,pushPeriod=1h", srv.URL)
+	h, err := LokiFromConfigLine(nil, configLine)
+	require.NoError(t, err)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	wg := new(sync.WaitGroup)
+	now := time.Now()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		err = h.Fire(&logrus.Entry{Time: now, Level: logrus.InfoLevel, Message: "test message"})
+		time.Sleep(time.Millisecond * 10)
+		cancel()
+	}()
+	h.Listen(ctx)
+	wg.Wait()
+	select {
+	case data := <-receivedData:
+		require.JSONEq(t,
+			fmt.Sprintf(
+				`{"streams":[{"stream":{"level":"info"},"values":[["%d","test message"]]}]}`, now.UnixNano()),
+			data)
+	default:
+		t.Fatal("No logs were received from loki before hook has finished")
+	}
+}