mirror of https://github.com/grafana/loki
Logcli: Add Support for New Query Path (#987)
Signed-off-by: Joe Elliott <number101010@gmail.com>pull/1013/head
Joe Elliott
6 years ago
committed by
Robert Fratto
parent
364b5bc4f6
commit
47637852fb
@ -1,167 +0,0 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"encoding/base64" |
||||
"encoding/json" |
||||
"fmt" |
||||
"io/ioutil" |
||||
"log" |
||||
"net/http" |
||||
"net/url" |
||||
"strings" |
||||
"time" |
||||
|
||||
"github.com/gorilla/websocket" |
||||
"github.com/prometheus/common/config" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
const ( |
||||
queryPath = "/api/prom/query?query=%s&limit=%d&start=%d&end=%d&direction=%s" |
||||
labelsPath = "/api/prom/label" |
||||
labelValuesPath = "/api/prom/label/%s/values" |
||||
tailPath = "/api/prom/tail?query=%s&delay_for=%d&limit=%d&start=%d" |
||||
) |
||||
|
||||
func query(from, through time.Time, direction logproto.Direction) (*logproto.QueryResponse, error) { |
||||
path := fmt.Sprintf(queryPath, |
||||
url.QueryEscape(*queryStr), // query
|
||||
*limit, // limit
|
||||
from.UnixNano(), // start
|
||||
through.UnixNano(), // end
|
||||
direction.String(), // direction
|
||||
) |
||||
|
||||
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 { |
||||
us := *addr + path |
||||
if !*quiet { |
||||
log.Print(us) |
||||
} |
||||
|
||||
req, err := http.NewRequest("GET", us, nil) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
req.SetBasicAuth(*username, *password) |
||||
|
||||
// Parse the URL to extract the host
|
||||
u, err := url.Parse(us) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
clientConfig := config.HTTPClientConfig{ |
||||
TLSConfig: config.TLSConfig{ |
||||
CAFile: *tlsCACertPath, |
||||
CertFile: *tlsClientCertPath, |
||||
KeyFile: *tlsClientCertKeyPath, |
||||
ServerName: u.Host, |
||||
InsecureSkipVerify: *tlsSkipVerify, |
||||
}, |
||||
} |
||||
|
||||
client, err := config.NewClientFromConfig(clientConfig, "logcli") |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
resp, err := client.Do(req) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer func() { |
||||
if err := resp.Body.Close(); err != nil { |
||||
log.Println("error closing body", err) |
||||
} |
||||
}() |
||||
|
||||
if resp.StatusCode/100 != 2 { |
||||
buf, _ := ioutil.ReadAll(resp.Body) // nolint
|
||||
return fmt.Errorf("Error response from server: %s (%v)", string(buf), err) |
||||
} |
||||
|
||||
return json.NewDecoder(resp.Body).Decode(out) |
||||
} |
||||
|
||||
func liveTailQueryConn() (*websocket.Conn, error) { |
||||
path := fmt.Sprintf(tailPath, |
||||
url.QueryEscape(*queryStr), // query
|
||||
*delayFor, // delay_for
|
||||
*limit, // limit
|
||||
getStart(time.Now()).UnixNano(), // start
|
||||
) |
||||
return wsConnect(path) |
||||
} |
||||
|
||||
func wsConnect(path string) (*websocket.Conn, error) { |
||||
us := *addr + path |
||||
|
||||
// Parse the URL to extract the host
|
||||
u, err := url.Parse(us) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
tlsConfig, err := config.NewTLSConfig(&config.TLSConfig{ |
||||
CAFile: *tlsCACertPath, |
||||
CertFile: *tlsClientCertPath, |
||||
KeyFile: *tlsClientCertKeyPath, |
||||
ServerName: u.Host, |
||||
InsecureSkipVerify: *tlsSkipVerify, |
||||
}) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
if strings.HasPrefix(us, "https") { |
||||
us = strings.Replace(us, "https", "wss", 1) |
||||
} else if strings.HasPrefix(us, "http") { |
||||
us = strings.Replace(us, "http", "ws", 1) |
||||
} |
||||
if !*quiet { |
||||
log.Println(us) |
||||
} |
||||
|
||||
h := http.Header{"Authorization": {"Basic " + base64.StdEncoding.EncodeToString([]byte(*username+":"+*password))}} |
||||
|
||||
ws := websocket.Dialer{ |
||||
TLSClientConfig: tlsConfig, |
||||
} |
||||
|
||||
c, resp, err := ws.Dial(us, h) |
||||
|
||||
if err != nil { |
||||
if resp == nil { |
||||
return nil, err |
||||
} |
||||
buf, _ := ioutil.ReadAll(resp.Body) // nolint
|
||||
return nil, fmt.Errorf("Error response from server: %s (%v)", string(buf), err) |
||||
} |
||||
|
||||
return c, nil |
||||
} |
||||
@ -1,32 +0,0 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"fmt" |
||||
"log" |
||||
|
||||
"github.com/grafana/loki/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 |
||||
} |
||||
@ -1,114 +0,0 @@ |
||||
package main |
||||
|
||||
import ( |
||||
"fmt" |
||||
"log" |
||||
"strings" |
||||
"time" |
||||
|
||||
"github.com/fatih/color" |
||||
"github.com/prometheus/prometheus/pkg/labels" |
||||
|
||||
"github.com/grafana/loki/pkg/iter" |
||||
"github.com/grafana/loki/pkg/logcli/output" |
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
func getStart(end time.Time) time.Time { |
||||
start := end.Add(-*since) |
||||
if *from != "" { |
||||
var err error |
||||
start, err = time.Parse(time.RFC3339Nano, *from) |
||||
if err != nil { |
||||
log.Fatalf("error parsing date '%s': %s", *from, err) |
||||
} |
||||
} |
||||
return start |
||||
} |
||||
|
||||
func doQuery(out output.LogOutput) { |
||||
if *tail { |
||||
tailQuery(out) |
||||
return |
||||
} |
||||
|
||||
var ( |
||||
i iter.EntryIterator |
||||
common labels.Labels |
||||
) |
||||
|
||||
end := time.Now() |
||||
start := getStart(end) |
||||
|
||||
if *to != "" { |
||||
var err error |
||||
end, err = time.Parse(time.RFC3339Nano, *to) |
||||
if err != nil { |
||||
log.Fatalf("error parsing --to date '%s': %s", *to, err) |
||||
} |
||||
} |
||||
|
||||
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 := parseLabels(resp) |
||||
|
||||
labelsCache := func(labels string) labels.Labels { |
||||
return cache[labels] |
||||
} |
||||
|
||||
common = commonLabels(lss) |
||||
|
||||
// Remove the labels we want to show from common
|
||||
if len(*showLabelsKey) > 0 { |
||||
common = common.MatchLabels(false, *showLabelsKey...) |
||||
} |
||||
|
||||
if len(common) > 0 && !*quiet { |
||||
log.Println("Common labels:", color.RedString(common.String())) |
||||
} |
||||
|
||||
if len(*ignoreLabelsKey) > 0 && !*quiet { |
||||
log.Println("Ignoring labels key:", color.RedString(strings.Join(*ignoreLabelsKey, ","))) |
||||
} |
||||
|
||||
// Remove ignored and common labels from the cached labels and
|
||||
// calculate the max labels length
|
||||
maxLabelsLen := *fixedLabelsLen |
||||
for key, ls := range cache { |
||||
// Remove common labels
|
||||
ls = subtract(ls, common) |
||||
|
||||
// Remove ignored labels
|
||||
if len(*ignoreLabelsKey) > 0 { |
||||
ls = ls.MatchLabels(false, *ignoreLabelsKey...) |
||||
} |
||||
|
||||
// Update cached labels
|
||||
cache[key] = ls |
||||
|
||||
// Update max labels length
|
||||
len := len(ls.String()) |
||||
if maxLabelsLen < len { |
||||
maxLabelsLen = len |
||||
} |
||||
} |
||||
|
||||
i = iter.NewQueryResponseIterator(resp, d) |
||||
|
||||
for i.Next() { |
||||
ls := labelsCache(i.Labels()) |
||||
fmt.Println(out.Format(i.Entry().Timestamp, &ls, maxLabelsLen, i.Entry().Line)) |
||||
} |
||||
|
||||
if err := i.Error(); err != nil { |
||||
log.Fatalf("Error from iterator: %v", err) |
||||
} |
||||
} |
||||
@ -0,0 +1,223 @@ |
||||
package client |
||||
|
||||
import ( |
||||
"encoding/base64" |
||||
"encoding/json" |
||||
"fmt" |
||||
"io/ioutil" |
||||
"log" |
||||
"net/http" |
||||
"net/url" |
||||
"strings" |
||||
"time" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
|
||||
"github.com/grafana/loki/pkg/logql" |
||||
|
||||
"github.com/gorilla/websocket" |
||||
"github.com/prometheus/common/config" |
||||
"github.com/prometheus/prometheus/promql" |
||||
|
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
const ( |
||||
queryPath = "/loki/api/v1/query?query=%s&limit=%d&time=%d&direction=%s" |
||||
queryRangePath = "/loki/api/v1/query_range?query=%s&limit=%d&start=%d&end=%d&direction=%s" |
||||
labelsPath = "/loki/api/v1/label" |
||||
labelValuesPath = "/loki/api/v1/label/%s/values" |
||||
tailPath = "/loki/api/v1/tail?query=%s&delay_for=%d&limit=%d&start=%d" |
||||
) |
||||
|
||||
// Client contains fields necessary to query a Loki instance
|
||||
type Client struct { |
||||
TLSConfig config.TLSConfig |
||||
Username string |
||||
Password string |
||||
Address string |
||||
} |
||||
|
||||
// QueryResult contains fields necessary to return data from Loki endpoints
|
||||
type QueryResult struct { |
||||
ResultType promql.ValueType |
||||
Result interface{} |
||||
} |
||||
|
||||
// Query uses the /api/v1/query endpoint to execute an instant query
|
||||
// excluding interfacer b/c it suggests taking the interface promql.Node instead of logproto.Direction b/c it happens to have a String() method
|
||||
// nolint:interfacer
|
||||
func (c *Client) Query(queryStr string, limit int, time time.Time, direction logproto.Direction, quiet bool) (*QueryResult, error) { |
||||
path := fmt.Sprintf(queryPath, |
||||
url.QueryEscape(queryStr), // query
|
||||
limit, // limit
|
||||
time.UnixNano(), // start
|
||||
direction.String(), // direction
|
||||
) |
||||
|
||||
return c.doQuery(path, quiet) |
||||
} |
||||
|
||||
// QueryRange uses the /api/v1/query_range endpoint to execute a range query
|
||||
// excluding interfacer b/c it suggests taking the interface promql.Node instead of logproto.Direction b/c it happens to have a String() method
|
||||
// nolint:interfacer
|
||||
func (c *Client) QueryRange(queryStr string, limit int, from, through time.Time, direction logproto.Direction, quiet bool) (*QueryResult, error) { |
||||
path := fmt.Sprintf(queryRangePath, |
||||
url.QueryEscape(queryStr), // query
|
||||
limit, // limit
|
||||
from.UnixNano(), // start
|
||||
through.UnixNano(), // end
|
||||
direction.String(), // direction
|
||||
) |
||||
|
||||
return c.doQuery(path, quiet) |
||||
} |
||||
|
||||
// ListLabelNames uses the /api/v1/label endpoint to list label names
|
||||
func (c *Client) ListLabelNames(quiet bool) (*logproto.LabelResponse, error) { |
||||
var labelResponse logproto.LabelResponse |
||||
if err := c.doRequest(labelsPath, quiet, &labelResponse); err != nil { |
||||
return nil, err |
||||
} |
||||
return &labelResponse, nil |
||||
} |
||||
|
||||
// ListLabelValues uses the /api/v1/label endpoint to list label values
|
||||
func (c *Client) ListLabelValues(name string, quiet bool) (*logproto.LabelResponse, error) { |
||||
path := fmt.Sprintf(labelValuesPath, url.PathEscape(name)) |
||||
var labelResponse logproto.LabelResponse |
||||
if err := c.doRequest(path, quiet, &labelResponse); err != nil { |
||||
return nil, err |
||||
} |
||||
return &labelResponse, nil |
||||
} |
||||
|
||||
func (c *Client) doQuery(path string, quiet bool) (*QueryResult, error) { |
||||
var err error |
||||
|
||||
unmarshal := struct { |
||||
Type promql.ValueType `json:"resultType"` |
||||
Result json.RawMessage `json:"result"` |
||||
}{} |
||||
|
||||
if err = c.doRequest(path, quiet, &unmarshal); err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
var value interface{} |
||||
|
||||
// unmarshal results
|
||||
switch unmarshal.Type { |
||||
case logql.ValueTypeStreams: |
||||
var s logql.Streams |
||||
err = json.Unmarshal(unmarshal.Result, &s) |
||||
value = s |
||||
case promql.ValueTypeMatrix: |
||||
var m model.Matrix |
||||
err = json.Unmarshal(unmarshal.Result, &m) |
||||
value = m |
||||
case promql.ValueTypeVector: |
||||
var m model.Vector |
||||
err = json.Unmarshal(unmarshal.Result, &m) |
||||
value = m |
||||
default: |
||||
return nil, fmt.Errorf("Unknown type: %s", unmarshal.Type) |
||||
} |
||||
|
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
return &QueryResult{ |
||||
ResultType: unmarshal.Type, |
||||
Result: value, |
||||
}, nil |
||||
} |
||||
|
||||
func (c *Client) doRequest(path string, quiet bool, out interface{}) error { |
||||
us := c.Address + path |
||||
if !quiet { |
||||
log.Print(us) |
||||
} |
||||
|
||||
req, err := http.NewRequest("GET", us, nil) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
req.SetBasicAuth(c.Username, c.Password) |
||||
|
||||
// Parse the URL to extract the host
|
||||
clientConfig := config.HTTPClientConfig{ |
||||
TLSConfig: c.TLSConfig, |
||||
} |
||||
|
||||
client, err := config.NewClientFromConfig(clientConfig, "logcli") |
||||
if err != nil { |
||||
return err |
||||
} |
||||
|
||||
resp, err := client.Do(req) |
||||
if err != nil { |
||||
return err |
||||
} |
||||
defer func() { |
||||
if err := resp.Body.Close(); err != nil { |
||||
log.Println("error closing body", err) |
||||
} |
||||
}() |
||||
|
||||
if resp.StatusCode/100 != 2 { |
||||
buf, _ := ioutil.ReadAll(resp.Body) // nolint
|
||||
return fmt.Errorf("Error response from server: %s (%v)", string(buf), err) |
||||
} |
||||
|
||||
return json.NewDecoder(resp.Body).Decode(out) |
||||
} |
||||
|
||||
// LiveTailQueryConn uses /api/prom/tail to set up a websocket connection and returns it
|
||||
func (c *Client) LiveTailQueryConn(queryStr string, delayFor int, limit int, from int64, quiet bool) (*websocket.Conn, error) { |
||||
path := fmt.Sprintf(tailPath, |
||||
url.QueryEscape(queryStr), // query
|
||||
delayFor, // delay_for
|
||||
limit, // limit
|
||||
from, // start
|
||||
) |
||||
return c.wsConnect(path, quiet) |
||||
} |
||||
|
||||
func (c *Client) wsConnect(path string, quiet bool) (*websocket.Conn, error) { |
||||
us := c.Address + path |
||||
|
||||
tlsConfig, err := config.NewTLSConfig(&c.TLSConfig) |
||||
if err != nil { |
||||
return nil, err |
||||
} |
||||
|
||||
if strings.HasPrefix(us, "https") { |
||||
us = strings.Replace(us, "https", "wss", 1) |
||||
} else if strings.HasPrefix(us, "http") { |
||||
us = strings.Replace(us, "http", "ws", 1) |
||||
} |
||||
if !quiet { |
||||
log.Println(us) |
||||
} |
||||
|
||||
h := http.Header{"Authorization": {"Basic " + base64.StdEncoding.EncodeToString([]byte(c.Username+":"+c.Password))}} |
||||
|
||||
ws := websocket.Dialer{ |
||||
TLSClientConfig: tlsConfig, |
||||
} |
||||
|
||||
conn, resp, err := ws.Dial(us, h) |
||||
|
||||
if err != nil { |
||||
if resp == nil { |
||||
return nil, err |
||||
} |
||||
buf, _ := ioutil.ReadAll(resp.Body) // nolint
|
||||
return nil, fmt.Errorf("Error response from server: %s (%v)", string(buf), err) |
||||
} |
||||
|
||||
return conn, nil |
||||
} |
||||
@ -0,0 +1,39 @@ |
||||
package labelquery |
||||
|
||||
import ( |
||||
"fmt" |
||||
"log" |
||||
|
||||
"github.com/grafana/loki/pkg/logcli/client" |
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
// LabelQuery contains all necessary fields to execute label queries and print out the resutls
|
||||
type LabelQuery struct { |
||||
LabelName string |
||||
Quiet bool |
||||
} |
||||
|
||||
// DoLabels prints out label results
|
||||
func (q *LabelQuery) DoLabels(c *client.Client) { |
||||
values := q.ListLabels(c) |
||||
|
||||
for _, value := range values { |
||||
fmt.Println(value) |
||||
} |
||||
} |
||||
|
||||
// ListLabels returns an array of label strings
|
||||
func (q *LabelQuery) ListLabels(c *client.Client) []string { |
||||
var labelResponse *logproto.LabelResponse |
||||
var err error |
||||
if len(q.LabelName) > 0 { |
||||
labelResponse, err = c.ListLabelValues(q.LabelName, q.Quiet) |
||||
} else { |
||||
labelResponse, err = c.ListLabelNames(q.Quiet) |
||||
} |
||||
if err != nil { |
||||
log.Fatalf("Error doing request: %+v", err) |
||||
} |
||||
return labelResponse.Values |
||||
} |
||||
@ -0,0 +1,164 @@ |
||||
package query |
||||
|
||||
import ( |
||||
"encoding/json" |
||||
"fmt" |
||||
"log" |
||||
"strings" |
||||
"time" |
||||
|
||||
"github.com/prometheus/common/model" |
||||
"github.com/prometheus/prometheus/promql" |
||||
|
||||
"github.com/fatih/color" |
||||
"github.com/grafana/loki/pkg/iter" |
||||
"github.com/grafana/loki/pkg/logql" |
||||
"github.com/prometheus/prometheus/pkg/labels" |
||||
|
||||
"github.com/grafana/loki/pkg/logcli/client" |
||||
"github.com/grafana/loki/pkg/logcli/output" |
||||
"github.com/grafana/loki/pkg/logproto" |
||||
) |
||||
|
||||
// Query contains all necessary fields to execute instant and range queries and print the results.
|
||||
type Query struct { |
||||
QueryString string |
||||
Start time.Time |
||||
End time.Time |
||||
Limit int |
||||
Forward bool |
||||
Quiet bool |
||||
NoLabels bool |
||||
IgnoreLabelsKey []string |
||||
ShowLabelsKey []string |
||||
FixedLabelsLen int |
||||
} |
||||
|
||||
// DoQuery executes the query and prints out the results
|
||||
func (q *Query) DoQuery(c *client.Client, out output.LogOutput) { |
||||
d := q.resultsDirection() |
||||
|
||||
var resp *client.QueryResult |
||||
var err error |
||||
|
||||
if q.isInstant() { |
||||
resp, err = c.Query(q.QueryString, q.Limit, q.Start, d, q.Quiet) |
||||
} else { |
||||
resp, err = c.QueryRange(q.QueryString, q.Limit, q.Start, q.End, d, q.Quiet) |
||||
} |
||||
|
||||
if err != nil { |
||||
log.Fatalf("Query failed: %+v", err) |
||||
} |
||||
|
||||
switch resp.ResultType { |
||||
case logql.ValueTypeStreams: |
||||
streams := resp.Result.(logql.Streams) |
||||
q.printStream(streams, out) |
||||
case promql.ValueTypeMatrix: |
||||
matrix := resp.Result.(model.Matrix) |
||||
q.printMatrix(matrix) |
||||
case promql.ValueTypeVector: |
||||
vector := resp.Result.(model.Vector) |
||||
q.printVector(vector) |
||||
default: |
||||
log.Fatalf("Unable to print unsupported type: %v", resp.ResultType) |
||||
} |
||||
} |
||||
|
||||
// SetInstant makes the Query an instant type
|
||||
func (q *Query) SetInstant(time time.Time) { |
||||
q.Start = time |
||||
q.End = time |
||||
} |
||||
|
||||
func (q *Query) isInstant() bool { |
||||
return q.Start == q.End |
||||
} |
||||
|
||||
func (q *Query) printStream(streams logql.Streams, out output.LogOutput) { |
||||
cache, lss := parseLabels(streams) |
||||
|
||||
labelsCache := func(labels string) labels.Labels { |
||||
return cache[labels] |
||||
} |
||||
|
||||
common := commonLabels(lss) |
||||
|
||||
// Remove the labels we want to show from common
|
||||
if len(q.ShowLabelsKey) > 0 { |
||||
common = common.MatchLabels(false, q.ShowLabelsKey...) |
||||
} |
||||
|
||||
if len(common) > 0 && !q.Quiet { |
||||
log.Println("Common labels:", color.RedString(common.String())) |
||||
} |
||||
|
||||
if len(q.IgnoreLabelsKey) > 0 && !q.Quiet { |
||||
log.Println("Ignoring labels key:", color.RedString(strings.Join(q.IgnoreLabelsKey, ","))) |
||||
} |
||||
|
||||
// Remove ignored and common labels from the cached labels and
|
||||
// calculate the max labels length
|
||||
maxLabelsLen := q.FixedLabelsLen |
||||
for key, ls := range cache { |
||||
// Remove common labels
|
||||
ls = subtract(ls, common) |
||||
|
||||
// Remove ignored labels
|
||||
if len(q.IgnoreLabelsKey) > 0 { |
||||
ls = ls.MatchLabels(false, q.IgnoreLabelsKey...) |
||||
} |
||||
|
||||
// Update cached labels
|
||||
cache[key] = ls |
||||
|
||||
// Update max labels length
|
||||
len := len(ls.String()) |
||||
if maxLabelsLen < len { |
||||
maxLabelsLen = len |
||||
} |
||||
} |
||||
|
||||
d := q.resultsDirection() |
||||
i := iter.NewStreamsIterator(streams, d) |
||||
|
||||
for i.Next() { |
||||
ls := labelsCache(i.Labels()) |
||||
fmt.Println(out.Format(i.Entry().Timestamp, &ls, maxLabelsLen, i.Entry().Line)) |
||||
} |
||||
|
||||
if err := i.Error(); err != nil { |
||||
log.Fatalf("Error from iterator: %v", err) |
||||
} |
||||
} |
||||
|
||||
func (q *Query) printMatrix(matrix model.Matrix) { |
||||
// yes we are effectively unmarshalling and then immediately marshalling this object back to json. we are doing this b/c
|
||||
// it gives us more flexibility with regard to output types in the future. initially we are supporting just formatted json but eventually
|
||||
// we might add output options such as render to an image file on disk
|
||||
bytes, err := json.MarshalIndent(matrix, "", " ") |
||||
|
||||
if err != nil { |
||||
log.Fatalf("Error marshalling matrix: %v", err) |
||||
} |
||||
|
||||
fmt.Print(string(bytes)) |
||||
} |
||||
|
||||
func (q *Query) printVector(vector model.Vector) { |
||||
bytes, err := json.MarshalIndent(vector, "", " ") |
||||
|
||||
if err != nil { |
||||
log.Fatalf("Error marshalling vector: %v", err) |
||||
} |
||||
|
||||
fmt.Print(string(bytes)) |
||||
} |
||||
|
||||
func (q *Query) resultsDirection() logproto.Direction { |
||||
if q.Forward { |
||||
return logproto.FORWARD |
||||
} |
||||
return logproto.BACKWARD |
||||
} |
||||
@ -1,4 +1,4 @@ |
||||
package main |
||||
package query |
||||
|
||||
import ( |
||||
"reflect" |
||||
Loading…
Reference in new issue