Add tailing to CLI. (#21)

* Add tailing to CLI. Signed-off-by: Tom Wilkie <tom.wilkie@gmail.com> * Review feedback. Signed-off-by: Tom Wilkie <tom.wilkie@gmail.com>
8 years ago · c0b153e4a4
parent 9b1dc8062e
commit c0b153e4a4
5 changed files with 325 additions and 182 deletions
--- a/cmd/logcli/client.go
+++ b/cmd/logcli/client.go
@ -0,0 +1,71 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/grafana/logish/pkg/logproto"
+)
+
+const (
+	queryPath       = "/api/prom/query?query=%s&limit=%d&start=%d&end=%d&direction=%s&regexp=%s"
+	labelsPath      = "/api/prom/label"
+	labelValuesPath = "/api/prom/label/%s/values"
+)
+
+func query(from, through time.Time, direction logproto.Direction) (*logproto.QueryResponse, error) {
+	path := fmt.Sprintf(queryPath, url.QueryEscape(*queryStr), *limit, from.UnixNano(),
+		through.UnixNano(), direction.String(), url.QueryEscape(*regexpStr))
+
+	var resp logproto.QueryResponse
+	if err := doRequest(path, &resp); err != nil {
+		return nil, err
+	}
+
+	return &resp, nil
+}
+
+func listLabelNames() (*logproto.LabelResponse, error) {
+	var labelResponse logproto.LabelResponse
+	if err := doRequest(labelsPath, &labelResponse); err != nil {
+		return nil, err
+	}
+	return &labelResponse, nil
+}
+
+func listLabelValues(name string) (*logproto.LabelResponse, error) {
+	path := fmt.Sprintf(labelValuesPath, url.PathEscape(name))
+	var labelResponse logproto.LabelResponse
+	if err := doRequest(path, &labelResponse); err != nil {
+		return nil, err
+	}
+	return &labelResponse, nil
+}
+
+func doRequest(path string, out interface{}) error {
+	url := *addr + path
+	fmt.Println(url)
+
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return err
+	}
+	req.SetBasicAuth(*username, *password)
+
+	resp, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode/100 != 2 {
+		buf, _ := ioutil.ReadAll(resp.Body)
+		return fmt.Errorf("Error response from server: %s (%v)", string(buf), err)
+	}
+
+	return json.NewDecoder(resp.Body).Decode(out)
+}
--- a/cmd/logcli/labels.go
+++ b/cmd/logcli/labels.go
@ -0,0 +1,32 @@
+package main
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/grafana/logish/pkg/logproto"
+)
+
+func doLabels() {
+	var labelResponse *logproto.LabelResponse
+	var err error
+	if len(*labelName) > 0 {
+		labelResponse, err = listLabelValues(*labelName)
+	} else {
+		labelResponse, err = listLabelNames()
+	}
+	if err != nil {
+		log.Fatalf("Error doing request: %+v", err)
+	}
+	for _, value := range labelResponse.Values {
+		fmt.Println(value)
+	}
+}
+
+func listLabels() []string {
+	labelResponse, err := listLabelNames()
+	if err != nil {
+		log.Fatalf("Error fetching labels: %+v", err)
+	}
+	return labelResponse.Values
+}
--- a/cmd/logcli/main.go
+++ b/cmd/logcli/main.go
@ -1,23 +1,9 @@
 package main

 import (
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"log"
-	"net/http"
-	"net/url"
 	"os"
-	"strings"
-	"time"

-	"github.com/fatih/color"
-	"github.com/prometheus/prometheus/pkg/labels"
 	kingpin "gopkg.in/alecthomas/kingpin.v2"
-
-	"github.com/grafana/logish/pkg/iter"
-	"github.com/grafana/logish/pkg/logproto"
-	"github.com/grafana/logish/pkg/parser"
 )

 var (
@ -32,181 +18,17 @@ var (
 	limit     = queryCmd.Flag("limit", "Limit on number of entries to print.").Default("30").Int()
 	since     = queryCmd.Flag("since", "Lookback window.").Default("1h").Duration()
 	forward   = queryCmd.Flag("forward", "Scan forwards through logs.").Default("false").Bool()
+	tail      = queryCmd.Flag("tail", "Tail the logs").Short('t').Default("false").Bool()

 	labelsCmd = app.Command("labels", "Find values for a given label.")
-	labelName = labelsCmd.Arg("label", "The name of the label.").String()
+	labelName = labelsCmd.Arg("label", "The name of the label.").HintAction(listLabels).String()
 )

 func main() {
 	switch kingpin.MustParse(app.Parse(os.Args[1:])) {
-
 	case queryCmd.FullCommand():
-		query()
-
+		doQuery()
 	case labelsCmd.FullCommand():
-		label()
-	}
-}
-
-func label() {
-	var path string
-	if len(*labelName) > 0 {
-		path = fmt.Sprintf("/api/prom/label/%s/values", url.PathEscape(*labelName))
-	} else {
-		path = "/api/prom/label"
-	}
-	var labelResponse logproto.LabelResponse
-	doRequest(path, &labelResponse)
-	for _, value := range labelResponse.Values {
-		fmt.Println(value)
-	}
-}
-
-func query() {
-	end := time.Now()
-	start := end.Add(-*since)
-
-	directionStr := "backward"
-	if *forward {
-		directionStr = "forward"
-	}
-
-	path := fmt.Sprintf("/api/prom/query?query=%s&limit=%d&start=%d&end=%d&direction=%s&regexp=%s",
-		url.QueryEscape(*queryStr), *limit, start.Unix(), end.Unix(), directionStr, url.QueryEscape(*regexpStr))
-	var queryResponse logproto.QueryResponse
-	doRequest(path, &queryResponse)
-
-	if len(queryResponse.Streams) == 0 {
-		return
-	}
-
-	labelsCache := make(map[string]labels.Labels, len(queryResponse.Streams))
-	lss := make([]labels.Labels, 0, len(queryResponse.Streams))
-	for _, stream := range queryResponse.Streams {
-		ls, err := parser.Labels(stream.Labels)
-		if err != nil {
-			log.Fatalf("Error parsing labels: %v", err)
-		}
-		labelsCache[stream.Labels] = ls
-		lss = append(lss, ls)
-	}
-
-	commonLabels, err := commonLabels(lss)
-	if err != nil {
-		log.Fatalf("Error parsing labels: %v", err)
-	}
-
-	if len(commonLabels) > 0 {
-		fmt.Println("Common labels:", color.RedString(commonLabels.String()))
-	}
-
-	maxLabelsLen := 0
-	for _, ls := range lss {
-		ls = subtract(commonLabels, ls)
-		len := len(ls.String())
-		if maxLabelsLen < len {
-			maxLabelsLen = len
-		}
-	}
-
-	d := logproto.BACKWARD
-	if *forward {
-		d = logproto.FORWARD
-	}
-	iter := iter.NewQueryResponseIterator(&queryResponse, d)
-	for iter.Next() {
-		ls := labelsCache[iter.Labels()]
-		ls = subtract(commonLabels, ls)
-		labels := ls.String()
-		labels += strings.Repeat(" ", maxLabelsLen-len(labels))
-
-		fmt.Println(
-			color.BlueString(iter.Entry().Timestamp.Format(time.RFC3339)),
-			color.RedString(labels),
-			strings.TrimSpace(iter.Entry().Line),
-		)
-	}
-
-	if err := iter.Error(); err != nil {
-		log.Fatalf("Error from iterator: %v", err)
-	}
-}
-
-func doRequest(path string, out interface{}) {
-	url := *addr + path
-	fmt.Println(url)
-
-	req, err := http.NewRequest("GET", url, nil)
-	if err != nil {
-		log.Fatalf("Error creating request: %v", err)
-	}
-	req.SetBasicAuth(*username, *password)
-
-	resp, err := http.DefaultClient.Do(req)
-	if err != nil {
-		log.Fatalf("Error doing request: %v", err)
-	}
-	defer resp.Body.Close()
-
-	if resp.StatusCode/100 != 2 {
-		buf, err := ioutil.ReadAll(resp.Body)
-		log.Fatalf("Error response from server: %s (%v)", string(buf), err)
-	}
-
-	if err := json.NewDecoder(resp.Body).Decode(out); err != nil {
-		log.Fatalf("Error decoding response: %v", err)
-	}
-}
-
-func commonLabels(lss []labels.Labels) (labels.Labels, error) {
-	result := lss[0]
-	for i := 1; i < len(lss); i++ {
-		result = intersect(result, lss[i])
-	}
-	return result, nil
-}
-
-func intersect(a, b labels.Labels) labels.Labels {
-	var result labels.Labels
-	for i, j := 0, 0; i < len(a) && j < len(b); {
-		k := strings.Compare(a[i].Name, b[j].Name)
-		switch {
-		case k == 0:
-			if a[i].Value == b[j].Value {
-				result = append(result, a[i])
-			}
-			i++
-			j++
-		case k < 0:
-			i++
-		case k > 0:
-			j++
-		}
-	}
-	return result
-}
-
-// substract a from b
-func subtract(a, b labels.Labels) labels.Labels {
-	var result labels.Labels
-	i, j := 0, 0
-	for i < len(a) && j < len(b) {
-		k := strings.Compare(a[i].Name, b[j].Name)
-		if k != 0 || a[i].Value != b[j].Value {
-			result = append(result, b[j])
-		}
-		switch {
-		case k == 0:
-			i++
-			j++
-		case k < 0:
-			i++
-		case k > 0:
-			j++
-		}
-	}
-	for ; j < len(b); j++ {
-		result = append(result, b[j])
+		doLabels()
 	}
-	return result
 }
--- a/cmd/logcli/query.go
+++ b/cmd/logcli/query.go
@ -0,0 +1,163 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/fatih/color"
+	"github.com/prometheus/prometheus/pkg/labels"
+
+	"github.com/grafana/logish/pkg/iter"
+	"github.com/grafana/logish/pkg/logproto"
+	"github.com/grafana/logish/pkg/parser"
+)
+
+type labelsCache func(labels string) labels.Labels
+
+func doQuery() {
+	var (
+		i            iter.EntryIterator
+		labelsCache  = mustParseLabels
+		common       labels.Labels
+		maxLabelsLen = 100
+	)
+
+	if *tail {
+		i = tailQuery()
+	} else {
+		end := time.Now()
+		start := end.Add(-*since)
+		d := logproto.BACKWARD
+		if *forward {
+			d = logproto.FORWARD
+		}
+
+		resp, err := query(start, end, d)
+		if err != nil {
+			log.Fatalf("Query failed: %+v", err)
+		}
+
+		cache, lss, err := parseLabels(resp)
+		if err != nil {
+			log.Fatalf("Failed parsing labels: %+v", err)
+		}
+
+		labelsCache = func(labels string) labels.Labels {
+			return cache[labels]
+		}
+		common = commonLabels(lss)
+		i = iter.NewQueryResponseIterator(resp, d)
+
+		if len(common) > 0 {
+			fmt.Println("Common labels:", color.RedString(common.String()))
+		}
+
+		for _, ls := range cache {
+			ls = subtract(common, ls)
+			len := len(ls.String())
+			if maxLabelsLen < len {
+				maxLabelsLen = len
+			}
+		}
+	}
+
+	for i.Next() {
+		ls := labelsCache(i.Labels())
+		ls = subtract(ls, common)
+		fmt.Println(
+			color.BlueString(i.Entry().Timestamp.Format(time.RFC3339)),
+			color.RedString(padLabel(ls, maxLabelsLen)),
+			strings.TrimSpace(i.Entry().Line),
+		)
+	}
+
+	if err := i.Error(); err != nil {
+		log.Fatalf("Error from iterator: %v", err)
+	}
+}
+
+func padLabel(ls labels.Labels, maxLabelsLen int) string {
+	labels := ls.String()
+	if len(labels) < maxLabelsLen {
+		labels += strings.Repeat(" ", maxLabelsLen-len(labels))
+	}
+	return labels
+}
+
+func mustParseLabels(labels string) labels.Labels {
+	ls, err := parser.Labels(labels)
+	if err != nil {
+		log.Fatalf("Failed to parse labels: %+v", err)
+	}
+	return ls
+}
+
+func parseLabels(resp *logproto.QueryResponse) (map[string]labels.Labels, []labels.Labels, error) {
+	cache := make(map[string]labels.Labels, len(resp.Streams))
+	lss := make([]labels.Labels, 0, len(resp.Streams))
+	for _, stream := range resp.Streams {
+		ls := mustParseLabels(stream.Labels)
+		cache[stream.Labels] = ls
+		lss = append(lss, ls)
+	}
+	return cache, lss, nil
+}
+
+func commonLabels(lss []labels.Labels) labels.Labels {
+	if len(lss) == 0 {
+		return nil
+	}
+
+	result := lss[0]
+	for i := 1; i < len(lss); i++ {
+		result = intersect(result, lss[i])
+	}
+	return result
+}
+
+func intersect(a, b labels.Labels) labels.Labels {
+	var result labels.Labels
+	for i, j := 0, 0; i < len(a) && j < len(b); {
+		k := strings.Compare(a[i].Name, b[j].Name)
+		switch {
+		case k == 0:
+			if a[i].Value == b[j].Value {
+				result = append(result, a[i])
+			}
+			i++
+			j++
+		case k < 0:
+			i++
+		case k > 0:
+			j++
+		}
+	}
+	return result
+}
+
+// substract b from a
+func subtract(a, b labels.Labels) labels.Labels {
+	var result labels.Labels
+	i, j := 0, 0
+	for i < len(a) && j < len(b) {
+		k := strings.Compare(a[i].Name, b[j].Name)
+		if k != 0 || a[i].Value != b[j].Value {
+			result = append(result, a[i])
+		}
+		switch {
+		case k == 0:
+			i++
+			j++
+		case k < 0:
+			i++
+		case k > 0:
+			j++
+		}
+	}
+	for ; i < len(a); i++ {
+		result = append(result, a[i])
+	}
+	return result
+}
--- a/cmd/logcli/tail.go
+++ b/cmd/logcli/tail.go
@ -0,0 +1,55 @@
+package main
+
+import (
+	"time"
+
+	"github.com/grafana/logish/pkg/iter"
+	"github.com/grafana/logish/pkg/logproto"
+)
+
+const tailIteratorIncrement = 10 * time.Second
+
+func tailQuery() iter.EntryIterator {
+	return &tailIterator{
+		from: time.Now().Add(-tailIteratorIncrement),
+	}
+}
+
+type tailIterator struct {
+	from time.Time
+	err  error
+	iter.EntryIterator
+}
+
+func (t *tailIterator) Next() bool {
+	for t.EntryIterator == nil || !t.EntryIterator.Next() {
+		through, now := t.from.Add(tailIteratorIncrement), time.Now()
+		if through.After(now) {
+			time.Sleep(through.Sub(now))
+		}
+
+		resp, err := query(t.from, through, logproto.FORWARD)
+		if err != nil {
+			t.err = err
+			return false
+		}
+
+		// We store the through time such that if we don't see any entries, we will
+		// still make forward progress. This is overwritten by any entries we might
+		// see to ensure pagination works.
+		t.from = through
+		t.EntryIterator = iter.NewQueryResponseIterator(resp, logproto.FORWARD)
+	}
+
+	return true
+}
+
+func (t *tailIterator) Entry() logproto.Entry {
+	entry := t.EntryIterator.Entry()
+	t.from = entry.Timestamp.Add(1 * time.Nanosecond)
+	return entry
+}
+
+func (t *tailIterator) Error() error {
+	return t.err
+}