359 строки
11 KiB
Go
359 строки
11 KiB
Go
package influx
|
|
|
|
import (
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/http/httptest"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
)
|
|
|
|
func TestEscapeTagValue(t *testing.T) {
|
|
tests := []struct {
|
|
input string
|
|
expected string
|
|
}{
|
|
{"OK", "OK"},
|
|
{"simple", "simple"},
|
|
{"has space", `has\ space`},
|
|
{"has,comma", `has\,comma`},
|
|
{"has=equals", `has\=equals`},
|
|
{`has\backslash`, `has\\backslash`},
|
|
// Real-world error messages that were causing TSDB write failures
|
|
{`response check: expected keyword Каталония not found`, `response\ check:\ expected\ keyword\ Каталония\ not\ found`},
|
|
{`request exec: Get "https://example.ru": dial tcp: lookup example.ru: no such host`, `request\ exec:\ Get\ "https://example.ru":\ dial\ tcp:\ lookup\ example.ru:\ no\ such\ host`},
|
|
{`redirect: https://example.ru/path`, `redirect:\ https://example.ru/path`},
|
|
{`Bad status code: 200 (expected 403)`, `Bad\ status\ code:\ 200\ (expected\ 403)`},
|
|
{"", ""},
|
|
// Multiple special chars together
|
|
{`a=b,c d\e`, `a\=b\,c\ d\\e`},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.input, func(t *testing.T) {
|
|
got := escapeTagValue(tt.input)
|
|
if got != tt.expected {
|
|
t.Errorf("escapeTagValue(%q) = %q, want %q", tt.input, got, tt.expected)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestFormatInfluxLine(t *testing.T) {
|
|
ts := time.Unix(0, 1771961128540561786)
|
|
|
|
tests := []struct {
|
|
name string
|
|
measurement string
|
|
tags map[string]string
|
|
fields map[string]interface{}
|
|
wantPrefix string // Check the line starts with this (before timestamp)
|
|
}{
|
|
{
|
|
name: "simple OK check",
|
|
measurement: "chttp",
|
|
tags: map[string]string{"check": "457", "code": "200", "state": "OK", "warnings": ""},
|
|
fields: map[string]interface{}{"took": int64(294)},
|
|
wantPrefix: "chttp,check=457,code=200,state=OK,warnings= took=294i",
|
|
},
|
|
{
|
|
name: "check with error containing spaces and colons",
|
|
measurement: "chttp",
|
|
tags: map[string]string{"check": "791", "code": "200", "error": "response check: expected keyword not found", "state": "ERR", "warnings": ""},
|
|
fields: map[string]interface{}{"took": int64(294)},
|
|
wantPrefix: `chttp,check=791,code=200,error=response\ check:\ expected\ keyword\ not\ found,state=ERR,warnings= took=294i`,
|
|
},
|
|
{
|
|
name: "check with redirect URL in warnings",
|
|
measurement: "chttp",
|
|
tags: map[string]string{"check": "672", "code": "301", "state": "WARN", "warnings": "redirect: https://example.ru/"},
|
|
fields: map[string]interface{}{"took": int64(276)},
|
|
wantPrefix: `chttp,check=672,code=301,state=WARN,warnings=redirect:\ https://example.ru/ took=276i`,
|
|
},
|
|
{
|
|
name: "check with equals in error",
|
|
measurement: "chttp",
|
|
tags: map[string]string{"check": "100", "state": "ERR", "error": "key=value problem"},
|
|
fields: map[string]interface{}{"took": int64(0)},
|
|
wantPrefix: `chttp,check=100,error=key\=value\ problem,state=ERR took=0i`,
|
|
},
|
|
}
|
|
|
|
for _, tt := range tests {
|
|
t.Run(tt.name, func(t *testing.T) {
|
|
got := formatInfluxLine(tt.measurement, tt.tags, tt.fields, ts)
|
|
// The line should end with the timestamp
|
|
wantSuffix := " 1771961128540561786"
|
|
want := tt.wantPrefix + wantSuffix
|
|
if got != want {
|
|
t.Errorf("formatInfluxLine() =\n %q\nwant:\n %q", got, want)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestQueryVM_UsesMatchParam(t *testing.T) {
|
|
var receivedQuery string
|
|
var receivedStart string
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
receivedQuery = r.URL.Query().Get("match[]")
|
|
receivedStart = r.URL.Query().Get("start")
|
|
|
|
// Verify it's NOT using the old "query" param
|
|
if q := r.URL.Query().Get("query"); q != "" {
|
|
t.Errorf("QueryVM sent deprecated 'query' param: %s", q)
|
|
}
|
|
|
|
w.WriteHeader(http.StatusOK)
|
|
// Return empty JSON-lines response
|
|
}))
|
|
defer server.Close()
|
|
|
|
// Override addr for test
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
selector := `chttp_took{check="123"}`
|
|
_, err := QueryVM(selector, "1771960000")
|
|
if err != nil {
|
|
t.Fatalf("QueryVM returned error: %v", err)
|
|
}
|
|
|
|
if receivedQuery != selector {
|
|
t.Errorf("match[] = %q, want %q", receivedQuery, selector)
|
|
}
|
|
if receivedStart != "1771960000" {
|
|
t.Errorf("start = %q, want %q", receivedStart, "1771960000")
|
|
}
|
|
}
|
|
|
|
func TestQueryVM_ParsesResponse(t *testing.T) {
|
|
response := `{"metric":{"__name__":"chttp_took","check":"123","state":"OK"},"values":[294,305],"timestamps":[1771961128000,1771961188000]}
|
|
{"metric":{"__name__":"chttp_took","check":"123","state":"ERR","error":"timeout"},"values":[0],"timestamps":[1771961248000]}`
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusOK)
|
|
io.WriteString(w, response)
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
results, err := QueryVM(`chttp_took{check="123"}`, "")
|
|
if err != nil {
|
|
t.Fatalf("QueryVM returned error: %v", err)
|
|
}
|
|
|
|
if len(results) != 2 {
|
|
t.Fatalf("expected 2 results, got %d", len(results))
|
|
}
|
|
|
|
if results[0].Metric["state"] != "OK" {
|
|
t.Errorf("first result state = %q, want OK", results[0].Metric["state"])
|
|
}
|
|
if len(results[0].Values) != 2 {
|
|
t.Errorf("first result values count = %d, want 2", len(results[0].Values))
|
|
}
|
|
if results[1].Metric["error"] != "timeout" {
|
|
t.Errorf("second result error = %q, want timeout", results[1].Metric["error"])
|
|
}
|
|
}
|
|
|
|
func TestGetLast(t *testing.T) {
|
|
response := `{"metric":{"__name__":"chttp_took","check":"123","state":"OK","error":"","warnings":""},"values":[294,305],"timestamps":[1771961128000,1771961188000]}
|
|
{"metric":{"__name__":"chttp_took","check":"123","state":"ERR","error":"timeout","warnings":""},"values":[0],"timestamps":[1771961248000]}`
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
// Verify correct params
|
|
match := r.URL.Query().Get("match[]")
|
|
if match != `chttp_took{check="123"}` {
|
|
t.Errorf("match[] = %q, want chttp_took{check=\"123\"}", match)
|
|
}
|
|
start := r.URL.Query().Get("start")
|
|
if start == "" {
|
|
t.Error("expected start parameter")
|
|
}
|
|
|
|
w.WriteHeader(http.StatusOK)
|
|
io.WriteString(w, response)
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
data, err := GetLast("chttp", 123, 6)
|
|
if err != nil {
|
|
t.Fatalf("GetLast returned error: %v", err)
|
|
}
|
|
|
|
if len(data) != 3 {
|
|
t.Fatalf("expected 3 data points, got %d", len(data))
|
|
}
|
|
|
|
// Should be sorted newest first
|
|
if data[0].State != "ERR" {
|
|
t.Errorf("first (newest) data point state = %q, want ERR", data[0].State)
|
|
}
|
|
if data[0].Duration != 0 {
|
|
t.Errorf("first data point duration = %d, want 0", data[0].Duration)
|
|
}
|
|
|
|
if data[2].State != "OK" {
|
|
t.Errorf("last (oldest) data point state = %q, want OK", data[2].State)
|
|
}
|
|
if data[2].Duration != 294 {
|
|
t.Errorf("last data point duration = %d, want 294", data[2].Duration)
|
|
}
|
|
}
|
|
|
|
func TestWriteOne_EscapesSpecialChars(t *testing.T) {
|
|
var receivedBody string
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
body, _ := io.ReadAll(r.Body)
|
|
receivedBody = string(body)
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
tags := map[string]string{
|
|
"check": "791",
|
|
"code": "200",
|
|
"error": "response check: expected keyword not found",
|
|
"state": "ERR",
|
|
"warnings": "",
|
|
}
|
|
fields := map[string]interface{}{
|
|
"took": int64(294),
|
|
}
|
|
|
|
err := WriteOne("chttp", tags, fields)
|
|
if err != nil {
|
|
t.Fatalf("WriteOne returned error: %v", err)
|
|
}
|
|
|
|
// The body should have properly escaped tag values
|
|
if strings.Contains(receivedBody, "error=response check:") {
|
|
t.Error("error tag value was not escaped - spaces should be escaped")
|
|
}
|
|
if !strings.Contains(receivedBody, `error=response\ check:\ expected\ keyword\ not\ found`) {
|
|
t.Errorf("expected escaped error tag, got: %s", receivedBody)
|
|
}
|
|
|
|
// Should contain the field
|
|
if !strings.Contains(receivedBody, "took=294i") {
|
|
t.Errorf("body doesn't contain took=294i: %s", receivedBody)
|
|
}
|
|
|
|
// Should start with measurement name
|
|
if !strings.HasPrefix(receivedBody, "chttp,") {
|
|
t.Errorf("body doesn't start with chttp,: %s", receivedBody)
|
|
}
|
|
}
|
|
|
|
func TestWriteOne_SendsToCorrectEndpoint(t *testing.T) {
|
|
var receivedPath string
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
receivedPath = r.URL.Path
|
|
w.WriteHeader(http.StatusNoContent)
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
err := WriteOne("chttp", map[string]string{"check": "1"}, map[string]interface{}{"took": int64(100)})
|
|
if err != nil {
|
|
t.Fatalf("WriteOne returned error: %v", err)
|
|
}
|
|
|
|
if receivedPath != "/api/v2/write" {
|
|
t.Errorf("WriteOne sent to %q, want /api/v2/write", receivedPath)
|
|
}
|
|
}
|
|
|
|
func TestQueryVM_Error(t *testing.T) {
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
w.WriteHeader(http.StatusBadRequest)
|
|
io.WriteString(w, "missing `match[]` arg")
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
_, err := QueryVM("bad_query", "")
|
|
if err == nil {
|
|
t.Fatal("expected error, got nil")
|
|
}
|
|
if !strings.Contains(err.Error(), "400") {
|
|
t.Errorf("error should contain status code 400: %v", err)
|
|
}
|
|
}
|
|
|
|
func TestGetLast_RejectsHyphenatedMetric(t *testing.T) {
|
|
var receivedMatch string
|
|
|
|
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
|
receivedMatch = r.URL.Query().Get("match[]")
|
|
w.WriteHeader(http.StatusOK)
|
|
}))
|
|
defer server.Close()
|
|
|
|
oldAddr := addr
|
|
addr = server.URL
|
|
defer func() { addr = oldAddr }()
|
|
|
|
_, err := GetLast("cllm_http", 123, 6)
|
|
if err != nil {
|
|
t.Fatalf("GetLast returned error: %v", err)
|
|
}
|
|
|
|
if receivedMatch != `cllm_http_took{check="123"}` {
|
|
t.Errorf("match[] = %q, want cllm_http_took{check=\"123\"}", receivedMatch)
|
|
}
|
|
|
|
if strings.Contains(receivedMatch, "-") {
|
|
t.Errorf("match[] should not contain hyphen: %q", receivedMatch)
|
|
}
|
|
}
|
|
|
|
func TestVMExportResponseParsing(t *testing.T) {
|
|
jsonStr := `{"metric":{"__name__":"chttp_took","check":"456","state":"WARN","warnings":"redirect: https://example.com/"},"values":[276],"timestamps":[1771961130522365547]}`
|
|
|
|
var resp VMExportResponse
|
|
err := json.Unmarshal([]byte(jsonStr), &resp)
|
|
if err != nil {
|
|
t.Fatalf("failed to parse: %v", err)
|
|
}
|
|
|
|
if resp.Metric["check"] != "456" {
|
|
t.Errorf("check = %q, want 456", resp.Metric["check"])
|
|
}
|
|
if resp.Metric["warnings"] != "redirect: https://example.com/" {
|
|
t.Errorf("warnings = %q", resp.Metric["warnings"])
|
|
}
|
|
if len(resp.Values) != 1 {
|
|
t.Fatalf("expected 1 value, got %d", len(resp.Values))
|
|
}
|
|
v, _ := resp.Values[0].Int64()
|
|
if v != 276 {
|
|
t.Errorf("value = %d, want 276", v)
|
|
}
|
|
}
|