unkin-agent
4d78ed534b
golangci-lint errcheck flagged six unchecked error returns, failing the pr/test workflow's lint step and skipping tests. - Blank-assign hash writes in Row.Key and MarkHidden calls - Close response body via deferred func matching node-lookup convention - Read test request body with io.ReadAll instead of a single Body.Read
110 lines
2.7 KiB
Go
110 lines
2.7 KiB
Go
package chlog
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"hash/fnv"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
const chTimeLayout = "2006-01-02 15:04:05.999"
|
|
|
|
type Row struct {
|
|
Timestamp string `json:"timestamp"`
|
|
Host string `json:"host"`
|
|
Source string `json:"source"`
|
|
Namespace string `json:"namespace"`
|
|
Pod string `json:"pod"`
|
|
Container string `json:"container"`
|
|
Stream string `json:"stream"`
|
|
Severity string `json:"severity"`
|
|
Message string `json:"message"`
|
|
Labels map[string]string `json:"labels"`
|
|
Fields map[string]string `json:"fields"`
|
|
}
|
|
|
|
func (r Row) Time() time.Time {
|
|
t, err := time.Parse(chTimeLayout, r.Timestamp)
|
|
if err != nil {
|
|
return time.Time{}
|
|
}
|
|
return t.UTC()
|
|
}
|
|
|
|
// Key identifies a row for overlap dedupe during paging and tailing.
|
|
func (r Row) Key() uint64 {
|
|
h := fnv.New64a()
|
|
for _, s := range []string{r.Timestamp, r.Host, r.Source, r.Namespace, r.Pod, r.Container, r.Stream, r.Message} {
|
|
_, _ = io.WriteString(h, s)
|
|
h.Write([]byte{0})
|
|
}
|
|
return h.Sum64()
|
|
}
|
|
|
|
type Client struct {
|
|
cfg Config
|
|
http *http.Client
|
|
}
|
|
|
|
func NewClient(cfg Config) *Client {
|
|
return &Client{cfg: cfg, http: &http.Client{Timeout: 130 * time.Second}}
|
|
}
|
|
|
|
// Run executes the query and streams each result row to fn. The SQL travels
|
|
// in the request body; every value goes as a param_* HTTP parameter.
|
|
func (c *Client) Run(ctx context.Context, q Query, fn func(Row) error) error {
|
|
v := url.Values{}
|
|
v.Set("default_format", "JSONEachRow")
|
|
for name, value := range q.Params {
|
|
v.Set("param_"+name, value)
|
|
}
|
|
u := strings.TrimRight(c.cfg.URL, "/") + "/?" + v.Encode()
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, strings.NewReader(q.SQL))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req.Header.Set("Content-Type", "text/plain")
|
|
req.Header.Set("X-ClickHouse-User", c.cfg.User)
|
|
if c.cfg.Password != "" {
|
|
req.Header.Set("X-ClickHouse-Key", c.cfg.Password)
|
|
}
|
|
|
|
resp, err := c.http.Do(req)
|
|
if err != nil {
|
|
return fmt.Errorf("clickhouse request: %w", err)
|
|
}
|
|
defer func() { _ = resp.Body.Close() }()
|
|
|
|
if resp.StatusCode != http.StatusOK {
|
|
body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
|
return fmt.Errorf("clickhouse HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
|
|
}
|
|
|
|
sc := bufio.NewScanner(resp.Body)
|
|
sc.Buffer(make([]byte, 0, 64*1024), 16*1024*1024)
|
|
line := 0
|
|
for sc.Scan() {
|
|
line++
|
|
b := sc.Bytes()
|
|
if len(b) == 0 {
|
|
continue
|
|
}
|
|
var r Row
|
|
if err := json.Unmarshal(b, &r); err != nil {
|
|
return fmt.Errorf("parse result row %s: %w", strconv.Itoa(line), err)
|
|
}
|
|
if err := fn(r); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return sc.Err()
|
|
}
|