forked from grafana/loki
/
tail.go
88 lines (71 loc) · 2.16 KB
/
tail.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
package query
import (
"log"
"os"
"os/signal"
"strings"
"syscall"
"time"
"github.com/fatih/color"
"github.com/gorilla/websocket"
"github.com/grafana/loki/pkg/logcli/client"
"github.com/grafana/loki/pkg/logcli/output"
"github.com/grafana/loki/pkg/loghttp"
"github.com/grafana/loki/pkg/util/unmarshal"
)
// TailQuery connects to the Loki websocket endpoint and tails logs
func (q *Query) TailQuery(delayFor time.Duration, c client.Client, out output.LogOutput) {
conn, err := c.LiveTailQueryConn(q.QueryString, delayFor, q.Limit, q.Start, q.Quiet)
if err != nil {
log.Fatalf("Tailing logs failed: %+v", err)
}
go func() {
stopChan := make(chan os.Signal, 1)
signal.Notify(stopChan, os.Interrupt, syscall.SIGTERM)
<-stopChan
if err := conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")); err != nil {
log.Println("Error closing websocket:", err)
}
os.Exit(0)
}()
tailResponse := new(loghttp.TailResponse)
if len(q.IgnoreLabelsKey) > 0 && !q.Quiet {
log.Println("Ignoring labels key:", color.RedString(strings.Join(q.IgnoreLabelsKey, ",")))
}
if len(q.ShowLabelsKey) > 0 && !q.Quiet {
log.Println("Print only labels key:", color.RedString(strings.Join(q.ShowLabelsKey, ",")))
}
for {
err := unmarshal.ReadTailResponseJSON(tailResponse, conn)
if err != nil {
log.Println("Error reading stream:", err)
return
}
labels := loghttp.LabelSet{}
for _, stream := range tailResponse.Streams {
if !q.NoLabels {
if len(q.IgnoreLabelsKey) > 0 || len(q.ShowLabelsKey) > 0 {
ls := stream.Labels
if len(q.ShowLabelsKey) > 0 {
ls = matchLabels(true, ls, q.ShowLabelsKey)
}
if len(q.IgnoreLabelsKey) > 0 {
ls = matchLabels(false, ls, q.ShowLabelsKey)
}
labels = ls
} else {
labels = stream.Labels
}
}
for _, entry := range stream.Entries {
out.FormatAndPrintln(entry.Timestamp, labels, 0, entry.Line)
}
}
if len(tailResponse.DroppedStreams) != 0 {
log.Println("Server dropped following entries due to slow client")
for _, d := range tailResponse.DroppedStreams {
log.Println(d.Timestamp, d.Labels)
}
}
}
}