Fixes tail api marshalling for v1. (#3211)

Introduce a bug by removing the default marshalling (#3163) but the tail api was using the default json. This fixes it by forcing the usage of jsoniter package.
Added a missing test so that it doesn't happen again.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

pkg/logql/marshal/marshal.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/gorilla/websocket"
 	json "github.com/json-iterator/go"
+	jsoniter "github.com/json-iterator/go"
 
 	"github.com/grafana/loki/pkg/loghttp"
 	legacy "github.com/grafana/loki/pkg/loghttp/legacy"
@@ -46,15 +47,23 @@ func WriteLabelResponseJSON(l logproto.LabelResponse, w io.Writer) error {
 	return json.NewEncoder(w).Encode(v1Response)
 }
 
+// WebsocketWriter knows how to write message to a websocket connection.
+type WebsocketWriter interface {
+	WriteMessage(int, []byte) error
+}
+
 // WriteTailResponseJSON marshals the legacy.TailResponse to v1 loghttp JSON and
 // then writes it to the provided connection.
-func WriteTailResponseJSON(r legacy.TailResponse, c *websocket.Conn) error {
+func WriteTailResponseJSON(r legacy.TailResponse, c WebsocketWriter) error {
 	v1Response, err := NewTailResponse(r)
 	if err != nil {
 		return err
 	}
-
-	return c.WriteJSON(v1Response)
+	data, err := jsoniter.Marshal(v1Response)
+	if err != nil {
+		return err
+	}
+	return c.WriteMessage(websocket.TextMessage, data)
 }
 
 // WriteSeriesResponseJSON marshals a logproto.SeriesResponse to v1 loghttp JSON and then

pkg/logql/marshal/marshal_test.go

@@ -521,3 +521,25 @@ func Benchmark_Encode(b *testing.B) {
 		}
 	}
 }
+
+type WebsocketWriterFunc func(int, []byte) error
+
+func (w WebsocketWriterFunc) WriteMessage(t int, d []byte) error { return w(t, d) }
+
+func Test_WriteTailResponseJSON(t *testing.T) {
+	require.NoError(t,
+		WriteTailResponseJSON(legacy.TailResponse{
+			Streams: []logproto.Stream{
+				{Labels: `{app="foo"}`, Entries: []logproto.Entry{{Timestamp: time.Unix(0, 1), Line: `foobar`}}},
+			},
+			DroppedEntries: []legacy.DroppedEntry{
+				{Timestamp: time.Unix(0, 2), Labels: `{app="dropped"}`},
+			},
+		},
+			WebsocketWriterFunc(func(i int, b []byte) error {
+				require.Equal(t, `{"streams":[{"stream":{"app":"foo"},"values":[["1","foobar"]]}],"dropped_entries":[{"timestamp":"2","labels":{"app":"dropped"}}]}`, string(b))
+				return nil
+			}),
+		),
+	)
+}