mirror of
https://github.com/SigNoz/signoz.git
synced 2026-11-10 02:10:42 +00:00
Compare commits
5 Commits
nv/exec-st
...
tvats-stre
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
86ecccd1f4 | ||
|
|
f464a1bfba | ||
|
|
1fef77fa93 | ||
|
|
ffc63dc0e4 | ||
|
|
313ec2d657 |
2
.github/workflows/build-community.yaml
vendored
2
.github/workflows/build-community.yaml
vendored
@@ -63,7 +63,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
GO_NAME: signoz-community
|
||||
GO_INPUT_ARTIFACT_CACHE_KEY: community-jsbuild-${{ github.sha }}
|
||||
GO_INPUT_ARTIFACT_PATH: frontend/build
|
||||
|
||||
2
.github/workflows/build-enterprise.yaml
vendored
2
.github/workflows/build-enterprise.yaml
vendored
@@ -89,7 +89,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
GO_INPUT_ARTIFACT_CACHE_KEY: enterprise-jsbuild-${{ github.sha }}
|
||||
GO_INPUT_ARTIFACT_PATH: frontend/build
|
||||
GO_BUILD_CONTEXT: ./cmd/enterprise
|
||||
|
||||
2
.github/workflows/build-staging.yaml
vendored
2
.github/workflows/build-staging.yaml
vendored
@@ -95,7 +95,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
GO_INPUT_ARTIFACT_CACHE_KEY: staging-jsbuild-${{ github.sha }}
|
||||
GO_INPUT_ARTIFACT_PATH: frontend/build
|
||||
GO_BUILD_CONTEXT: ./cmd/enterprise
|
||||
|
||||
20
.github/workflows/goci.yaml
vendored
20
.github/workflows/goci.yaml
vendored
@@ -22,7 +22,7 @@ jobs:
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_TEST_CONTEXT: ./...
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
fmt:
|
||||
if: |
|
||||
github.event_name == 'merge_group' ||
|
||||
@@ -32,7 +32,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
lint:
|
||||
if: |
|
||||
github.event_name == 'merge_group' ||
|
||||
@@ -42,7 +42,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
deps:
|
||||
if: |
|
||||
github.event_name == 'merge_group' ||
|
||||
@@ -52,7 +52,7 @@ jobs:
|
||||
secrets: inherit
|
||||
with:
|
||||
PRIMUS_REF: main
|
||||
GO_VERSION: 1.24
|
||||
GO_VERSION: 1.25
|
||||
semconv-generated:
|
||||
if: |
|
||||
github.event_name == 'merge_group' ||
|
||||
@@ -65,7 +65,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: check-semconv-generated-files
|
||||
run: make semconv-check
|
||||
build:
|
||||
@@ -80,7 +80,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: qemu-install
|
||||
uses: docker/setup-qemu-action@v3
|
||||
- name: aarch64-install
|
||||
@@ -116,7 +116,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: generate-openapi
|
||||
run: |
|
||||
go run cmd/enterprise/*.go generate openapi
|
||||
@@ -133,7 +133,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: generate-authz
|
||||
run: |
|
||||
go run cmd/enterprise/*.go generate authz
|
||||
@@ -150,7 +150,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: generate-web-settings
|
||||
run: |
|
||||
go run cmd/enterprise/*.go generate config web-settings
|
||||
@@ -167,7 +167,7 @@ jobs:
|
||||
- name: go-install
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: generate-transaction-groups
|
||||
run: |
|
||||
go run cmd/enterprise/*.go generate config transaction-groups
|
||||
|
||||
4
.github/workflows/gor-signoz-community.yaml
vendored
4
.github/workflows/gor-signoz-community.yaml
vendored
@@ -66,7 +66,7 @@ jobs:
|
||||
- name: setup-go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: cross-compilation-tools
|
||||
if: matrix.os == 'ubuntu-latest'
|
||||
run: |
|
||||
@@ -130,7 +130,7 @@ jobs:
|
||||
- name: setup-go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
|
||||
# copy the caches from build
|
||||
- name: get-sha
|
||||
|
||||
4
.github/workflows/gor-signoz.yaml
vendored
4
.github/workflows/gor-signoz.yaml
vendored
@@ -75,7 +75,7 @@ jobs:
|
||||
- name: setup-go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
- name: cross-compilation-tools
|
||||
if: matrix.os == 'ubuntu-latest'
|
||||
run: |
|
||||
@@ -138,7 +138,7 @@ jobs:
|
||||
- name: setup-go
|
||||
uses: actions/setup-go@v5
|
||||
with:
|
||||
go-version: "1.24"
|
||||
go-version: "1.25"
|
||||
|
||||
# copy the caches from build
|
||||
- name: get-sha
|
||||
|
||||
1
go.mod
1
go.mod
@@ -20,6 +20,7 @@ require (
|
||||
github.com/evanphx/json-patch/v5 v5.9.11
|
||||
github.com/gin-gonic/gin v1.11.0
|
||||
github.com/go-co-op/gocron v1.30.1
|
||||
github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875
|
||||
github.com/go-openapi/runtime v0.29.2
|
||||
github.com/go-openapi/strfmt v0.26.1
|
||||
github.com/go-playground/validator/v10 v10.27.0
|
||||
|
||||
2
go.sum
2
go.sum
@@ -328,6 +328,8 @@ github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20200222043503-6f7a984d4dc4/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA=
|
||||
github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08=
|
||||
github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875 h1:XGPtjxqp0FS8jj1Z7QGcSOSgyq1fUPjNyJXvi3KfI4A=
|
||||
github.com/go-json-experiment/json v0.0.0-20260213210345-44df1a37e875/go.mod h1:uNVvRXArCGbZ508SxYYTC5v1JWoz2voff5pm25jU1Ok=
|
||||
github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
|
||||
github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as=
|
||||
github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY=
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"log/slog"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/textproto"
|
||||
"net/url"
|
||||
"os"
|
||||
"strconv"
|
||||
@@ -706,7 +707,9 @@ func TestEmailRejected(t *testing.T) {
|
||||
|
||||
// Send the alert to mock SMTP server.
|
||||
retry, err := e.Notify(context.Background(), firingAlert)
|
||||
require.ErrorContains(t, err, "501 5.5.4 Rejected!")
|
||||
var smtpErr *textproto.Error
|
||||
require.ErrorAs(t, err, &smtpErr)
|
||||
assert.Equal(t, 501, smtpErr.Code)
|
||||
assert.True(t, retry)
|
||||
require.NoError(t, srv.Shutdown(ctx))
|
||||
|
||||
|
||||
@@ -5,9 +5,12 @@ import (
|
||||
"net/http"
|
||||
"strconv"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
jsonv2 "github.com/go-json-experiment/json"
|
||||
jsonv1 "github.com/go-json-experiment/json/v1"
|
||||
jsoniter "github.com/json-iterator/go"
|
||||
"github.com/tidwall/gjson"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -29,6 +32,11 @@ type ErrorResponse struct {
|
||||
}
|
||||
|
||||
func Success(rw http.ResponseWriter, httpCode int, data interface{}) {
|
||||
if _, ok := data.(jsonv2.MarshalerTo); ok {
|
||||
successStream(rw, httpCode, data)
|
||||
return
|
||||
}
|
||||
|
||||
body, err := json.Marshal(&SuccessResponse{Status: StatusSuccess.s, Data: data})
|
||||
if err != nil {
|
||||
Error(rw, err)
|
||||
@@ -134,3 +142,32 @@ func Error(rw http.ResponseWriter, cause error) {
|
||||
rw.WriteHeader(httpCode)
|
||||
_, _ = rw.Write(body)
|
||||
}
|
||||
|
||||
// successStream writes the body as it is encoded. It can still render an error if encoding fails
|
||||
// before the first write; after that the status is sent and a failure leaves the body truncated.
|
||||
func successStream(rw http.ResponseWriter, httpCode int, data any) {
|
||||
if httpCode == 0 {
|
||||
httpCode = http.StatusOK
|
||||
}
|
||||
|
||||
writer := &successStreamWriter{rw: rw, httpCode: httpCode}
|
||||
if err := jsonv2.MarshalWrite(writer, &SuccessResponse{Status: StatusSuccess.s, Data: data}, jsonv1.DefaultOptionsV1()); err != nil && !writer.started {
|
||||
Error(rw, err)
|
||||
}
|
||||
}
|
||||
|
||||
// successStreamWriter sends the headers on the first write.
|
||||
type successStreamWriter struct {
|
||||
rw http.ResponseWriter
|
||||
httpCode int
|
||||
started bool
|
||||
}
|
||||
|
||||
func (writer *successStreamWriter) Write(p []byte) (int, error) {
|
||||
if !writer.started {
|
||||
writer.started = true
|
||||
writer.rw.Header().Set("Content-Type", "application/json")
|
||||
writer.rw.WriteHeader(writer.httpCode)
|
||||
}
|
||||
return writer.rw.Write(p)
|
||||
}
|
||||
|
||||
@@ -11,6 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/go-json-experiment/json/jsontext"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
@@ -210,3 +211,26 @@ func TestErrorRetryAfterHeader(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
type streamedData []int
|
||||
|
||||
func (data streamedData) MarshalJSONTo(enc *jsontext.Encoder) error {
|
||||
if err := enc.WriteToken(jsontext.BeginArray); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, v := range data {
|
||||
if err := enc.WriteToken(jsontext.Int(int64(v))); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return enc.WriteToken(jsontext.EndArray)
|
||||
}
|
||||
|
||||
func TestSuccess_Stream(t *testing.T) {
|
||||
recorder := httptest.NewRecorder()
|
||||
Success(recorder, http.StatusAccepted, streamedData{1, 2, 3})
|
||||
|
||||
assert.Equal(t, http.StatusAccepted, recorder.Code)
|
||||
assert.Equal(t, "application/json", recorder.Header().Get("Content-Type"))
|
||||
assert.Equal(t, `{"status":"success","data":[1,2,3]}`, recorder.Body.String())
|
||||
}
|
||||
|
||||
@@ -146,8 +146,6 @@ func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, res
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
funcs := []qbtypes.Function{{Name: qbtypes.FunctionNameFillZero}}
|
||||
funcs = q.prepareFillZeroArgsWithStep(funcs, req, stepInterval)
|
||||
// empty time series if it doesn't exist
|
||||
tsData, ok := typedResults[name].Value.(*qbtypes.TimeSeriesData)
|
||||
if !ok {
|
||||
@@ -170,7 +168,9 @@ func (q *querier) postProcessResults(ctx context.Context, orgID valuer.UUID, res
|
||||
}
|
||||
}
|
||||
|
||||
typedResults[name] = q.applyFunctions(typedResults[name], funcs)
|
||||
startMs := querybuilder.ToNanoSecs(req.Start) / 1_000_000
|
||||
endMs := querybuilder.ToNanoSecs(req.End) / 1_000_000
|
||||
tsData.SetZeroFill(int64(startMs), int64(endMs), stepInterval)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -136,6 +136,9 @@ func (q *QueryRangeResponse) PrepareJSONSchema(schema *jsonschema.Schema) error
|
||||
type TimeSeriesData struct {
|
||||
QueryName string `json:"queryName" description:"The name of the query that produced this result."`
|
||||
Aggregations []*AggregationBucket `json:"aggregations" description:"One bucket per aggregation of the query. The order is not guaranteed; match buckets to aggregations by index."`
|
||||
|
||||
// zeroFill is applied only when encoding; in-memory series stay sparse.
|
||||
zeroFill *zeroFill
|
||||
}
|
||||
|
||||
// FilterToTimeRange copies the aggregation and series structure keeping only
|
||||
@@ -562,17 +565,6 @@ func sanitizeValue(v any) any {
|
||||
}
|
||||
}
|
||||
|
||||
func (q QueryRangeResponse) MarshalJSON() ([]byte, error) {
|
||||
type Alias QueryRangeResponse
|
||||
return json.Marshal(&struct {
|
||||
*Alias
|
||||
Data any `json:"data"`
|
||||
}{
|
||||
Alias: (*Alias)(&q),
|
||||
Data: sanitizeValue(q.Data),
|
||||
})
|
||||
}
|
||||
|
||||
func (s ScalarData) MarshalJSON() ([]byte, error) {
|
||||
type Alias ScalarData
|
||||
sanitizedData := make([][]any, len(s.Data))
|
||||
@@ -615,36 +607,6 @@ func (r RawRow) MarshalJSON() ([]byte, error) {
|
||||
})
|
||||
}
|
||||
|
||||
func (t TimeSeriesValue) MarshalJSON() ([]byte, error) {
|
||||
type Alias TimeSeriesValue
|
||||
|
||||
var sanitizedValues any
|
||||
if t.Values != nil {
|
||||
sanitizedValues = sanitizeValue(t.Values)
|
||||
// If original was empty slice, ensure we return empty slice not nil
|
||||
if len(t.Values) == 0 {
|
||||
sanitizedValues = []any{}
|
||||
}
|
||||
}
|
||||
|
||||
// a heatmap point's counts are spread across Values, so there is no one
|
||||
// number Value could carry
|
||||
var sanitizedValue any
|
||||
if t.Values == nil {
|
||||
sanitizedValue = sanitizeValue(t.Value)
|
||||
}
|
||||
|
||||
return json.Marshal(&struct {
|
||||
*Alias
|
||||
Value any `json:"value,omitempty"`
|
||||
Values any `json:"values,omitempty"`
|
||||
}{
|
||||
Alias: (*Alias)(&t),
|
||||
Value: sanitizedValue,
|
||||
Values: sanitizedValues,
|
||||
})
|
||||
}
|
||||
|
||||
func (r RawData) MarshalJSON() ([]byte, error) {
|
||||
type Alias RawData
|
||||
return json.Marshal((*Alias)(&r))
|
||||
|
||||
233
pkg/types/querybuildertypes/querybuildertypesv5/resp_json.go
Normal file
233
pkg/types/querybuildertypes/querybuildertypesv5/resp_json.go
Normal file
@@ -0,0 +1,233 @@
|
||||
package querybuildertypesv5
|
||||
|
||||
import (
|
||||
"math"
|
||||
|
||||
jsonv2 "github.com/go-json-experiment/json"
|
||||
"github.com/go-json-experiment/json/jsontext"
|
||||
jsonv1 "github.com/go-json-experiment/json/v1"
|
||||
)
|
||||
|
||||
// SetZeroFill makes every series render one point per step over [startMs, endMs], writing 0 for
|
||||
// a step without a value, instead of storing those points. Points off the step grid are dropped.
|
||||
func (t *TimeSeriesData) SetZeroFill(startMs, endMs, stepMs int64) {
|
||||
if stepMs <= 0 {
|
||||
return
|
||||
}
|
||||
t.zeroFill = &zeroFill{start: startMs, end: endMs, step: stepMs}
|
||||
}
|
||||
|
||||
func (q QueryRangeResponse) MarshalJSON() ([]byte, error) {
|
||||
return jsonv2.Marshal(q, jsonv1.DefaultOptionsV1())
|
||||
}
|
||||
|
||||
// MarshalJSONTo lets a time series response stream to the writer instead of being held in
|
||||
// memory whole.
|
||||
func (q QueryRangeResponse) MarshalJSONTo(enc *jsontext.Encoder) error {
|
||||
type alias QueryRangeResponse
|
||||
return jsonv2.MarshalEncode(enc, (*alias)(&q))
|
||||
}
|
||||
|
||||
func (t TimeSeriesData) MarshalJSON() ([]byte, error) {
|
||||
return jsonv2.Marshal(t, jsonv1.DefaultOptionsV1())
|
||||
}
|
||||
|
||||
func (t TimeSeriesData) MarshalJSONTo(enc *jsontext.Encoder) error {
|
||||
writer := tokenWriter{enc: enc}
|
||||
writer.token(jsontext.BeginObject)
|
||||
writer.token(jsontext.String("queryName"))
|
||||
writer.token(jsontext.String(t.QueryName))
|
||||
writer.token(jsontext.String("aggregations"))
|
||||
if t.Aggregations == nil {
|
||||
writer.token(jsontext.Null)
|
||||
} else {
|
||||
writer.token(jsontext.BeginArray)
|
||||
for _, bucket := range t.Aggregations {
|
||||
if writer.err != nil {
|
||||
return writer.err
|
||||
}
|
||||
writer.aggregationBucket(bucket, t.zeroFill)
|
||||
}
|
||||
writer.token(jsontext.EndArray)
|
||||
}
|
||||
writer.token(jsontext.EndObject)
|
||||
return writer.err
|
||||
}
|
||||
|
||||
func (t TimeSeriesValue) MarshalJSON() ([]byte, error) {
|
||||
return jsonv2.Marshal(t, jsonv1.DefaultOptionsV1())
|
||||
}
|
||||
|
||||
func (t TimeSeriesValue) MarshalJSONTo(enc *jsontext.Encoder) error {
|
||||
writer := tokenWriter{enc: enc}
|
||||
writer.token(jsontext.BeginObject)
|
||||
writer.token(jsontext.String("timestamp"))
|
||||
writer.token(jsontext.Int(t.Timestamp))
|
||||
if t.Partial {
|
||||
writer.token(jsontext.String("partial"))
|
||||
writer.token(jsontext.True)
|
||||
}
|
||||
// a heatmap point's counts are spread across Values, so there is no one
|
||||
// number Value could carry
|
||||
if t.Values == nil {
|
||||
writer.token(jsontext.String("value"))
|
||||
writer.token(sanitizedFloatToken(t.Value))
|
||||
} else {
|
||||
writer.token(jsontext.String("values"))
|
||||
writer.token(jsontext.BeginArray)
|
||||
for _, v := range t.Values {
|
||||
writer.token(sanitizedFloatToken(v))
|
||||
}
|
||||
writer.token(jsontext.EndArray)
|
||||
}
|
||||
writer.token(jsontext.EndObject)
|
||||
return writer.err
|
||||
}
|
||||
|
||||
type zeroFill struct {
|
||||
start int64
|
||||
end int64
|
||||
step int64
|
||||
}
|
||||
|
||||
// tokenWriter keeps the first error so a sequence of writes needs a single check.
|
||||
type tokenWriter struct {
|
||||
enc *jsontext.Encoder
|
||||
err error
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) token(t jsontext.Token) {
|
||||
if writer.err == nil {
|
||||
writer.err = writer.enc.WriteToken(t)
|
||||
}
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) value(v any) {
|
||||
if writer.err == nil {
|
||||
writer.err = jsonv2.MarshalEncode(writer.enc, v)
|
||||
}
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) aggregationBucket(a *AggregationBucket, fill *zeroFill) {
|
||||
if a == nil {
|
||||
writer.token(jsontext.Null)
|
||||
return
|
||||
}
|
||||
|
||||
writer.token(jsontext.BeginObject)
|
||||
writer.token(jsontext.String("index"))
|
||||
writer.token(jsontext.Int(int64(a.Index)))
|
||||
writer.token(jsontext.String("alias"))
|
||||
writer.token(jsontext.String(a.Alias))
|
||||
writer.token(jsontext.String("meta"))
|
||||
writer.value(a.Meta)
|
||||
writer.token(jsontext.String("series"))
|
||||
writer.seriesList(a.Series, fill)
|
||||
writer.optionalSeriesList("predictedSeries", a.PredictedSeries)
|
||||
writer.optionalSeriesList("upperBoundSeries", a.UpperBoundSeries)
|
||||
writer.optionalSeriesList("lowerBoundSeries", a.LowerBoundSeries)
|
||||
writer.optionalSeriesList("anomalyScores", a.AnomalyScores)
|
||||
writer.token(jsontext.EndObject)
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) optionalSeriesList(name string, series []*TimeSeries) {
|
||||
if len(series) == 0 {
|
||||
return
|
||||
}
|
||||
writer.token(jsontext.String(name))
|
||||
writer.seriesList(series, nil)
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) seriesList(series []*TimeSeries, fill *zeroFill) {
|
||||
if series == nil {
|
||||
writer.token(jsontext.Null)
|
||||
return
|
||||
}
|
||||
|
||||
writer.token(jsontext.BeginArray)
|
||||
for _, s := range series {
|
||||
if writer.err != nil {
|
||||
return
|
||||
}
|
||||
writer.series(s, fill)
|
||||
}
|
||||
writer.token(jsontext.EndArray)
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) series(s *TimeSeries, fill *zeroFill) {
|
||||
if writer.err != nil {
|
||||
return
|
||||
}
|
||||
if s == nil {
|
||||
writer.token(jsontext.Null)
|
||||
return
|
||||
}
|
||||
|
||||
writer.token(jsontext.BeginObject)
|
||||
if len(s.Labels) > 0 {
|
||||
writer.token(jsontext.String("labels"))
|
||||
writer.value(s.Labels)
|
||||
}
|
||||
writer.token(jsontext.String("values"))
|
||||
switch {
|
||||
case fill != nil:
|
||||
writer.zeroFilledValues(s.Values, fill)
|
||||
case s.Values == nil:
|
||||
writer.token(jsontext.Null)
|
||||
default:
|
||||
writer.token(jsontext.BeginArray)
|
||||
for _, v := range s.Values {
|
||||
if writer.err != nil {
|
||||
return
|
||||
}
|
||||
writer.point(v)
|
||||
}
|
||||
writer.token(jsontext.EndArray)
|
||||
}
|
||||
writer.token(jsontext.EndObject)
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) zeroFilledValues(values []*TimeSeriesValue, fill *zeroFill) {
|
||||
byTimestamp := make(map[int64]*TimeSeriesValue, len(values))
|
||||
for _, v := range values {
|
||||
byTimestamp[v.Timestamp] = v
|
||||
}
|
||||
|
||||
writer.token(jsontext.BeginArray)
|
||||
for ts := fill.start - fill.start%fill.step; ts <= fill.end && writer.err == nil; ts += fill.step {
|
||||
if v, ok := byTimestamp[ts]; ok {
|
||||
writer.point(v)
|
||||
continue
|
||||
}
|
||||
writer.token(jsontext.BeginObject)
|
||||
writer.token(jsontext.String("timestamp"))
|
||||
writer.token(jsontext.Int(ts))
|
||||
writer.token(jsontext.String("value"))
|
||||
writer.token(jsontext.Int(0))
|
||||
writer.token(jsontext.EndObject)
|
||||
}
|
||||
writer.token(jsontext.EndArray)
|
||||
}
|
||||
|
||||
func (writer *tokenWriter) point(v *TimeSeriesValue) {
|
||||
if v == nil {
|
||||
writer.token(jsontext.Null)
|
||||
return
|
||||
}
|
||||
if writer.err == nil {
|
||||
writer.err = v.MarshalJSONTo(writer.enc)
|
||||
}
|
||||
}
|
||||
|
||||
// sanitizedFloatToken encodes f the way sanitizeValue renders it.
|
||||
func sanitizedFloatToken(f float64) jsontext.Token {
|
||||
switch {
|
||||
case math.IsNaN(f):
|
||||
return jsontext.String("NaN")
|
||||
case math.IsInf(f, 1):
|
||||
return jsontext.String("Inf")
|
||||
case math.IsInf(f, -1):
|
||||
return jsontext.String("-Inf")
|
||||
}
|
||||
return jsontext.Float(roundToNonZeroDecimals(f, 3))
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
package querybuildertypesv5
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
jsonv2 "github.com/go-json-experiment/json"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestTimeSeriesData_ZeroFill(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
values []*TimeSeriesValue
|
||||
start int64
|
||||
end int64
|
||||
step int64
|
||||
expected string
|
||||
}{
|
||||
{
|
||||
name: "Gaps",
|
||||
values: []*TimeSeriesValue{{Timestamp: 1000, Value: 1}, {Timestamp: 4000, Value: 4}},
|
||||
start: 1000,
|
||||
end: 4000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":1},{"timestamp":2000,"value":0},{"timestamp":3000,"value":0},{"timestamp":4000,"value":4}]`,
|
||||
},
|
||||
{
|
||||
name: "Empty",
|
||||
values: []*TimeSeriesValue{},
|
||||
start: 1000,
|
||||
end: 3000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":0},{"timestamp":3000,"value":0}]`,
|
||||
},
|
||||
{
|
||||
name: "Nil",
|
||||
values: nil,
|
||||
start: 1000,
|
||||
end: 2000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":0}]`,
|
||||
},
|
||||
{
|
||||
name: "UnalignedBounds",
|
||||
values: []*TimeSeriesValue{{Timestamp: 60000, Value: 1}, {Timestamp: 240000, Value: 4}},
|
||||
start: 50000,
|
||||
end: 250000,
|
||||
step: 60000,
|
||||
expected: `[{"timestamp":0,"value":0},{"timestamp":60000,"value":1},{"timestamp":120000,"value":0},{"timestamp":180000,"value":0},{"timestamp":240000,"value":4}]`,
|
||||
},
|
||||
{
|
||||
name: "OffGridAndOutOfRangeDropped",
|
||||
values: []*TimeSeriesValue{{Timestamp: 500, Value: 5}, {Timestamp: 1500, Value: 15}, {Timestamp: 2000, Value: 2}, {Timestamp: 3000, Value: 3}},
|
||||
start: 1000,
|
||||
end: 2000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"value":2}]`,
|
||||
},
|
||||
{
|
||||
name: "DuplicateTimestampKeepsLast",
|
||||
values: []*TimeSeriesValue{{Timestamp: 1000, Value: 1}, {Timestamp: 1000, Value: 9}},
|
||||
start: 1000,
|
||||
end: 1000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":9}]`,
|
||||
},
|
||||
{
|
||||
name: "PointFieldsKept",
|
||||
values: []*TimeSeriesValue{{Timestamp: 2000, Value: 2.12345, Partial: true}},
|
||||
start: 1000,
|
||||
end: 2000,
|
||||
step: 1000,
|
||||
expected: `[{"timestamp":1000,"value":0},{"timestamp":2000,"partial":true,"value":2.123}]`,
|
||||
},
|
||||
}
|
||||
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
series := &TimeSeries{Values: testCase.values}
|
||||
data := &TimeSeriesData{
|
||||
QueryName: "A",
|
||||
Aggregations: []*AggregationBucket{{
|
||||
Series: []*TimeSeries{series},
|
||||
PredictedSeries: []*TimeSeries{{Values: []*TimeSeriesValue{{Timestamp: 3000, Value: 3}}}},
|
||||
}},
|
||||
}
|
||||
data.SetZeroFill(testCase.start, testCase.end, testCase.step)
|
||||
|
||||
expected := `{"queryName":"A","aggregations":[{"index":0,"alias":"","meta":{},"series":[{"values":` + testCase.expected + `}],"predictedSeries":[{"values":[{"timestamp":3000,"value":3}]}]}]}`
|
||||
|
||||
marshalled, err := json.Marshal(data)
|
||||
require.NoError(t, err)
|
||||
assert.JSONEq(t, expected, string(marshalled))
|
||||
|
||||
response := QueryRangeResponse{Type: RequestTypeTimeSeries, Data: QueryData{Results: []any{data}}}
|
||||
viaMarshalJSON, err := json.Marshal(response)
|
||||
require.NoError(t, err)
|
||||
viaMarshalJSONTo, err := jsonv2.Marshal(response)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, string(viaMarshalJSON), string(viaMarshalJSONTo))
|
||||
|
||||
var decoded struct {
|
||||
Data struct {
|
||||
Results []json.RawMessage `json:"results"`
|
||||
} `json:"data"`
|
||||
}
|
||||
require.NoError(t, json.Unmarshal(viaMarshalJSONTo, &decoded))
|
||||
require.Len(t, decoded.Data.Results, 1)
|
||||
assert.JSONEq(t, expected, string(decoded.Data.Results[0]))
|
||||
|
||||
assert.Equal(t, testCase.values, series.Values)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestTimeSeriesData_ZeroFillMatchesFuncFillZero(t *testing.T) {
|
||||
values := []*TimeSeriesValue{{Timestamp: 500, Value: 5}, {Timestamp: 60000, Value: 1}, {Timestamp: 90000, Value: 9}, {Timestamp: 240000, Value: 4, Partial: true}}
|
||||
|
||||
filled := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: []*TimeSeries{funcFillZero(&TimeSeries{Values: values}, 50000, 250000, 60000)}}}}
|
||||
deferred := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: []*TimeSeries{{Values: values}}}}}
|
||||
deferred.SetZeroFill(50000, 250000, 60000)
|
||||
|
||||
expected, err := json.Marshal(filled)
|
||||
require.NoError(t, err)
|
||||
actual, err := json.Marshal(deferred)
|
||||
require.NoError(t, err)
|
||||
assert.Equal(t, string(expected), string(actual))
|
||||
}
|
||||
|
||||
// failingWriter accepts its first write and fails every one after it.
|
||||
type failingWriter struct {
|
||||
writes int
|
||||
}
|
||||
|
||||
func (writer *failingWriter) Write(p []byte) (int, error) {
|
||||
writer.writes++
|
||||
if writer.writes == 1 {
|
||||
return len(p), nil
|
||||
}
|
||||
return 0, assert.AnError
|
||||
}
|
||||
|
||||
func TestQueryRangeResponse_MarshalStopsOnWriteError(t *testing.T) {
|
||||
values := make([]*TimeSeriesValue, 16)
|
||||
for i := range values {
|
||||
values[i] = &TimeSeriesValue{Timestamp: int64(i) * 1000, Value: 1}
|
||||
}
|
||||
series := make([]*TimeSeries, 10000)
|
||||
for i := range series {
|
||||
series[i] = &TimeSeries{Values: values}
|
||||
}
|
||||
data := &TimeSeriesData{QueryName: "A", Aggregations: []*AggregationBucket{{Series: series}}}
|
||||
data.SetZeroFill(0, 1000*1000, 1000)
|
||||
|
||||
response := QueryRangeResponse{Type: RequestTypeTimeSeries, Data: QueryData{Results: []any{data}}}
|
||||
|
||||
writer := &failingWriter{}
|
||||
err := jsonv2.MarshalWrite(writer, response)
|
||||
assert.ErrorIs(t, err, assert.AnError)
|
||||
assert.Equal(t, 2, writer.writes)
|
||||
|
||||
allocs := testing.AllocsPerRun(1, func() {
|
||||
_ = jsonv2.MarshalWrite(&failingWriter{}, response)
|
||||
})
|
||||
assert.Less(t, allocs, float64(len(series)))
|
||||
}
|
||||
Reference in New Issue
Block a user