Compare commits
3 Commits
tvats-quer
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9ba2a25eb9 | ||
|
|
30839d390f | ||
|
|
2483c41452 |
8
.github/CODEOWNERS
vendored
@@ -290,6 +290,14 @@ go.mod @therealpandey
|
||||
/frontend/src/components/MessagingQueueHealthCheck/ @SigNoz/events-frontend
|
||||
/frontend/src/hooks/messagingQueue/ @SigNoz/events-frontend
|
||||
|
||||
## Saved Views
|
||||
/frontend/src/container/SavedViews/ @aks07
|
||||
/frontend/src/container/Home/SavedViews/ @aks07
|
||||
/frontend/src/hooks/saveViews/ @aks07
|
||||
|
||||
## Bottom Strip
|
||||
/frontend/src/container/BottomStrip/ @aks07
|
||||
|
||||
## Storybook
|
||||
/frontend/.storybook/ @H4ad
|
||||
/frontend/src/storybook/ @H4ad
|
||||
|
||||
@@ -121,8 +121,8 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
|
||||
func(_ context.Context, _ factory.ProviderSettings, _ flagger.Flagger, _ licensing.Licensing, _ telemetrystore.TelemetryStore, _ retention.Getter, _ organization.Getter, _ zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string) {
|
||||
return signoz.NewMeterReporterProviderFactories(), "noop"
|
||||
},
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics, cfg querier.Config) querier.Handler {
|
||||
return querier.NewHandler(ps, q, a, cfg)
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
|
||||
return querier.NewHandler(ps, q, a)
|
||||
},
|
||||
func(_ sqlstore.SQLStore, _ dashboard.Module, _ global.Global, _ zeus.Zeus, _ gateway.Gateway, _ licensing.Licensing, _ serviceaccount.Module, _ cloudintegration.Config) (cloudintegration.Module, error) {
|
||||
return implcloudintegration.NewModule(), nil
|
||||
|
||||
@@ -172,9 +172,9 @@ func runServer(ctx context.Context, config signoz.Config, logger *slog.Logger) e
|
||||
|
||||
return factories, "http"
|
||||
},
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics, cfg querier.Config) querier.Handler {
|
||||
communityHandler := querier.NewHandler(ps, q, a, cfg)
|
||||
return eequerier.NewHandler(ps, q, communityHandler, cfg)
|
||||
func(ps factory.ProviderSettings, q querier.Querier, a analytics.Analytics) querier.Handler {
|
||||
communityHandler := querier.NewHandler(ps, q, a)
|
||||
return eequerier.NewHandler(ps, q, communityHandler)
|
||||
},
|
||||
func(sqlStore sqlstore.SQLStore, dashboardModule dashboard.Module, global global.Global, zeus zeus.Zeus, gateway gateway.Gateway, licensing licensing.Licensing, serviceAccount serviceaccount.Module, config cloudintegration.Config) (cloudintegration.Module, error) {
|
||||
defStore := pkgcloudintegration.NewServiceDefinitionStore()
|
||||
|
||||
@@ -178,8 +178,6 @@ querier:
|
||||
# range with padding to include slightly delayed log exports. Logs only; set
|
||||
# to 0 to disable.
|
||||
log_trace_id_window_padding: 5m
|
||||
# The maximum time range of a query range request. Set to 0 to disable.
|
||||
max_time_range: 17520h
|
||||
|
||||
##################### TelemetryStore #####################
|
||||
telemetrystore:
|
||||
|
||||
@@ -21,15 +21,13 @@ type handler struct {
|
||||
set factory.ProviderSettings
|
||||
querier querier.Querier
|
||||
community querier.Handler
|
||||
config querier.Config
|
||||
}
|
||||
|
||||
func NewHandler(set factory.ProviderSettings, querier querier.Querier, community querier.Handler, config querier.Config) querier.Handler {
|
||||
func NewHandler(set factory.ProviderSettings, querier querier.Querier, community querier.Handler) querier.Handler {
|
||||
return &handler{
|
||||
set: set,
|
||||
querier: querier,
|
||||
community: community,
|
||||
config: config,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -55,7 +53,7 @@ func (h *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
if err := queryRangeRequest.Validate(qbtypes.WithMaxTimeRange(h.config.MaxTimeRange)); err != nil {
|
||||
if err := queryRangeRequest.Validate(); err != nil {
|
||||
render.Error(rw, err)
|
||||
return
|
||||
}
|
||||
|
||||
BIN
frontend/src/assets/Images/users/anselm.jpg
Normal file
|
After Width: | Height: | Size: 21 KiB |
BIN
frontend/src/assets/Images/users/daniel.webp
Normal file
|
After Width: | Height: | Size: 15 KiB |
BIN
frontend/src/assets/Images/users/dhruv-garg.webp
Normal file
|
After Width: | Height: | Size: 23 KiB |
BIN
frontend/src/assets/Images/users/go-frendi.webp
Normal file
|
After Width: | Height: | Size: 15 KiB |
BIN
frontend/src/assets/Images/users/pranay-narang.webp
Normal file
|
After Width: | Height: | Size: 29 KiB |
BIN
frontend/src/assets/Images/users/shey.webp
Normal file
|
After Width: | Height: | Size: 20 KiB |
BIN
frontend/src/assets/Images/users/subomi-oluwalana.webp
Normal file
|
After Width: | Height: | Size: 15 KiB |
BIN
frontend/src/assets/Images/users/vivek-bhakta.webp
Normal file
|
After Width: | Height: | Size: 11 KiB |
@@ -1,3 +1,12 @@
|
||||
import anselmEickhoffUrl from '@/assets/Images/users/anselm.jpg';
|
||||
import danielSchellUrl from '@/assets/Images/users/daniel.webp';
|
||||
import dhruvGargUrl from '@/assets/Images/users/dhruv-garg.webp';
|
||||
import goFrendiGunawanUrl from '@/assets/Images/users/go-frendi.webp';
|
||||
import pranayNarangUrl from '@/assets/Images/users/pranay-narang.webp';
|
||||
import sheheryarSewaniUrl from '@/assets/Images/users/shey.webp';
|
||||
import subomiOluwalanaUrl from '@/assets/Images/users/subomi-oluwalana.webp';
|
||||
import vivekBhaktaUrl from '@/assets/Images/users/vivek-bhakta.webp';
|
||||
|
||||
export const infoData = [
|
||||
{
|
||||
id: 'infoBlock-1',
|
||||
@@ -43,7 +52,7 @@ export const enterpriseGradeValuesData = [
|
||||
export const customerStoriesData = [
|
||||
{
|
||||
key: 'story-subomi-oluwalana',
|
||||
avatar: 'https://signoz.io/img/users/subomi-oluwalana.webp',
|
||||
avatar: subomiOluwalanaUrl,
|
||||
personName: 'Subomi Oluwalana',
|
||||
role: 'Founder & CEO at Convoy',
|
||||
customerName: 'Convoy',
|
||||
@@ -54,7 +63,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-dhruv-garg',
|
||||
avatar: 'https://signoz.io/img/users/dhruv-garg.webp',
|
||||
avatar: dhruvGargUrl,
|
||||
personName: 'Dhruv Garg',
|
||||
role: 'Tech Lead at Nudge',
|
||||
customerName: 'Nudge',
|
||||
@@ -65,7 +74,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-vivek-bhakta',
|
||||
avatar: 'https://signoz.io/img/users/vivek-bhakta.webp',
|
||||
avatar: vivekBhaktaUrl,
|
||||
personName: 'Vivek Bhakta',
|
||||
role: 'CTO at Wombo AI',
|
||||
customerName: 'Wombo AI',
|
||||
@@ -75,7 +84,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-pranay-narang',
|
||||
avatar: 'https://signoz.io/img/users/pranay-narang.webp',
|
||||
avatar: pranayNarangUrl,
|
||||
personName: 'Pranay Narang',
|
||||
role: 'Engineering at Azodha',
|
||||
customerName: 'Azodha',
|
||||
@@ -85,7 +94,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-Sheheryar-Sewani',
|
||||
avatar: 'https://signoz.io/img/users/shey.webp',
|
||||
avatar: sheheryarSewaniUrl,
|
||||
personName: 'Sheheryar Sewani',
|
||||
role: 'Seasoned Rails Dev & Founder',
|
||||
customerName: '',
|
||||
@@ -96,7 +105,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-daniel-schell',
|
||||
avatar: 'https://signoz.io/img/users/daniel.webp',
|
||||
avatar: danielSchellUrl,
|
||||
personName: 'Daniel Schell',
|
||||
role: 'Founder & CTO at Airlockdigital',
|
||||
customerName: 'Airlockdigital',
|
||||
@@ -106,7 +115,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'c-story-6',
|
||||
avatar: 'https://signoz.io/img/users/go-frendi.webp',
|
||||
avatar: goFrendiGunawanUrl,
|
||||
personName: 'Go Frendi Gunawan',
|
||||
role: 'Data Engineer at Ctlyst.id',
|
||||
customerName: 'Ctlyst.id',
|
||||
@@ -116,7 +125,7 @@ export const customerStoriesData = [
|
||||
},
|
||||
{
|
||||
key: 'story-anselm-eickhoff',
|
||||
avatar: 'https://signoz.io/img/users/anselm.jpg',
|
||||
avatar: anselmEickhoffUrl,
|
||||
personName: 'Anselm Eickhoff',
|
||||
role: 'Software Architect',
|
||||
customerName: '',
|
||||
|
||||
@@ -22,7 +22,7 @@ func newConfig() factory.Config {
|
||||
Agent: AgentConfig{
|
||||
// we will maintain the latest version of cloud integration agent from here,
|
||||
// till we automate it externally or figure out a way to validate it.
|
||||
Version: "v0.0.15",
|
||||
Version: "v0.0.16",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -33,11 +33,10 @@ type handler struct {
|
||||
set factory.ProviderSettings
|
||||
analytics analytics.Analytics
|
||||
querier Querier
|
||||
config Config
|
||||
}
|
||||
|
||||
func NewHandler(set factory.ProviderSettings, querier Querier, analytics analytics.Analytics, config Config) Handler {
|
||||
return &handler{set: set, querier: querier, analytics: analytics, config: config}
|
||||
func NewHandler(set factory.ProviderSettings, querier Querier, analytics analytics.Analytics) Handler {
|
||||
return &handler{set: set, querier: querier, analytics: analytics}
|
||||
}
|
||||
|
||||
func (handler *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
|
||||
@@ -60,7 +59,7 @@ func (handler *handler) QueryRange(rw http.ResponseWriter, req *http.Request) {
|
||||
}
|
||||
|
||||
// Validate the query request
|
||||
if err := queryRangeRequest.Validate(qbtypes.WithMaxTimeRange(handler.config.MaxTimeRange)); err != nil {
|
||||
if err := queryRangeRequest.Validate(); err != nil {
|
||||
render.Error(rw, err)
|
||||
return
|
||||
}
|
||||
@@ -141,9 +140,7 @@ func (handler *handler) QueryRawStream(rw http.ResponseWriter, req *http.Request
|
||||
}
|
||||
// create the v5 request param
|
||||
queryRangeRequest := qbtypes.QueryRangeRequest{
|
||||
// Coerce the window to epoch milliseconds up front (End may be 0 for the
|
||||
// open-ended stream, which ToMilliSecs leaves untouched).
|
||||
Start: qbtypes.ToMilliSecs(start),
|
||||
Start: start,
|
||||
RequestType: qbtypes.RequestTypeRawStream,
|
||||
CompositeQuery: qbtypes.CompositeQuery{
|
||||
Queries: []qbtypes.QueryEnvelope{
|
||||
|
||||
@@ -10,8 +10,6 @@ import (
|
||||
|
||||
const DefaultMaxConcurrentQueries = 8
|
||||
|
||||
const DefaultMaxTimeRange = 2 * 365 * 24 * time.Hour
|
||||
|
||||
// Config represents the configuration for the querier.
|
||||
type Config struct {
|
||||
// CacheTTL is the TTL for cached query results
|
||||
@@ -22,8 +20,6 @@ type Config struct {
|
||||
MaxConcurrentQueries int `yaml:"max_concurrent_queries" mapstructure:"max_concurrent_queries"`
|
||||
// LogTraceIDWindowPadding is the padding added to narrowed down timerange from trace summary to logs with trace_id filter.
|
||||
LogTraceIDWindowPadding time.Duration `yaml:"log_trace_id_window_padding" mapstructure:"log_trace_id_window_padding"`
|
||||
// MaxTimeRange is the maximum allowed span between start and end of a query range request; 0 disables the limit.
|
||||
MaxTimeRange time.Duration `yaml:"max_time_range" mapstructure:"max_time_range"`
|
||||
|
||||
// Keys sit under querier.skip_resource_fingerprint.
|
||||
statementbuilder.Config `mapstructure:",squash" yaml:",squash"`
|
||||
@@ -41,7 +37,6 @@ func newConfig() factory.Config {
|
||||
FluxInterval: 5 * time.Minute,
|
||||
MaxConcurrentQueries: DefaultMaxConcurrentQueries,
|
||||
LogTraceIDWindowPadding: 5 * time.Minute,
|
||||
MaxTimeRange: DefaultMaxTimeRange,
|
||||
Config: statementbuilder.NewConfig(),
|
||||
}
|
||||
}
|
||||
@@ -60,9 +55,6 @@ func (c Config) Validate() error {
|
||||
if c.LogTraceIDWindowPadding < 0 {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "log_trace_id_window_padding must not be negative, got %v", c.LogTraceIDWindowPadding)
|
||||
}
|
||||
if c.MaxTimeRange < 0 {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "max_time_range must not be negative, got %v", c.MaxTimeRange)
|
||||
}
|
||||
// Embedded Validate is shadowed by this one; call it explicitly.
|
||||
if err := c.Config.Validate(); err != nil {
|
||||
return err
|
||||
|
||||
@@ -20,9 +20,9 @@ func makeBuckets(start, end uint64) []tsRange {
|
||||
curEnd := endNS
|
||||
|
||||
for {
|
||||
curStart := startNS
|
||||
if curEnd-startNS > bucket {
|
||||
curStart = curEnd - bucket
|
||||
curStart := curEnd - bucket
|
||||
if curStart < startNS {
|
||||
curStart = startNS
|
||||
}
|
||||
out = append(out, tsRange{fromNS: curStart, toNS: curEnd})
|
||||
|
||||
|
||||
@@ -23,7 +23,7 @@ func (q *querier) QueryRangePreview(
|
||||
opts qbtypes.QueryRangePreviewOptions,
|
||||
) (*qbtypes.QueryRangePreviewResponse, error) {
|
||||
|
||||
validationOpts, err := req.ValidateRequestScope(qbtypes.WithMaxTimeRange(q.maxTimeRange))
|
||||
validationOpts, err := req.ValidateRequestScope()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -62,7 +62,6 @@ type querier struct {
|
||||
liveDataRefresh time.Duration
|
||||
builderConfig builderConfig
|
||||
maxConcurrentQueries int
|
||||
maxTimeRange time.Duration
|
||||
}
|
||||
|
||||
var _ Querier = (*querier)(nil)
|
||||
@@ -83,7 +82,6 @@ func New(
|
||||
flagger flagger.Flagger,
|
||||
logTraceIDWindowPadding time.Duration,
|
||||
maxConcurrentQueries int,
|
||||
maxTimeRange time.Duration,
|
||||
) *querier {
|
||||
querierSettings := factory.NewScopedProviderSettings(settings, "github.com/SigNoz/signoz/pkg/querier")
|
||||
if maxConcurrentQueries <= 0 {
|
||||
@@ -108,12 +106,17 @@ func New(
|
||||
logTraceIDWindowPaddingMS: uint64(logTraceIDWindowPadding.Milliseconds()),
|
||||
},
|
||||
maxConcurrentQueries: maxConcurrentQueries,
|
||||
maxTimeRange: maxTimeRange,
|
||||
}
|
||||
}
|
||||
|
||||
func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest) (*qbtypes.QueryRangeResponse, error) {
|
||||
|
||||
// Coerce the window to epoch milliseconds up front so every downstream
|
||||
// consumer (TimeRange, narrowWindowByTraceID, step interval, etc.) can
|
||||
// safely assume ms regardless of the resolution the caller sent.
|
||||
req.Start = querybuilder.ToMilliSecs(req.Start)
|
||||
req.End = querybuilder.ToMilliSecs(req.End)
|
||||
|
||||
event := &qbtypes.QBEvent{
|
||||
Version: "v5",
|
||||
NumberOfQueries: len(req.CompositeQuery.Queries),
|
||||
@@ -531,6 +534,11 @@ func familyTemporality(temporalities map[string]metrictypes.Temporality, names [
|
||||
|
||||
func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qbtypes.QueryRangeRequest, client *qbtypes.RawStream) {
|
||||
|
||||
// Coerce the window to epoch milliseconds up front (End may be 0 for the
|
||||
// open-ended stream, which ToMilliSecs leaves untouched).
|
||||
req.Start = querybuilder.ToMilliSecs(req.Start)
|
||||
req.End = querybuilder.ToMilliSecs(req.End)
|
||||
|
||||
event := &qbtypes.QBEvent{
|
||||
Version: "v5",
|
||||
NumberOfQueries: len(req.CompositeQuery.Queries),
|
||||
|
||||
@@ -59,7 +59,6 @@ func TestQueryRange_MetricTypeMissing(t *testing.T) {
|
||||
flaggertest.New(t), // flagger
|
||||
0, // logTraceIDWindowPadding
|
||||
0, // maxConcurrentQueries
|
||||
0, // maxTimeRange
|
||||
)
|
||||
|
||||
req := &qbtypes.QueryRangeRequest{
|
||||
@@ -133,7 +132,6 @@ func TestQueryRange_MetricTypeFromStore(t *testing.T) {
|
||||
flaggertest.New(t), // flagger
|
||||
0, // logTraceIDWindowPadding
|
||||
0, // maxConcurrentQueries
|
||||
0, // maxTimeRange
|
||||
)
|
||||
|
||||
req := &qbtypes.QueryRangeRequest{
|
||||
|
||||
@@ -52,7 +52,6 @@ func NewFactory(
|
||||
flagger,
|
||||
cfg.LogTraceIDWindowPadding,
|
||||
cfg.MaxConcurrentQueries,
|
||||
cfg.MaxTimeRange,
|
||||
), nil
|
||||
},
|
||||
)
|
||||
|
||||
@@ -52,7 +52,6 @@ func prepareQuerierForMetrics(t *testing.T, telemetryStore telemetrystore.Teleme
|
||||
fl,
|
||||
0,
|
||||
0, // maxConcurrentQueries (0 means default)
|
||||
0, // maxTimeRange (0 means no limit)
|
||||
), metadataStore
|
||||
}
|
||||
|
||||
@@ -88,7 +87,6 @@ func prepareQuerierForLogs(t *testing.T, telemetryStore telemetrystore.Telemetry
|
||||
fl,
|
||||
5*time.Minute, // logTraceIDWindowPadding
|
||||
0, // maxConcurrentQueries (0 means default)
|
||||
0, // maxTimeRange (0 means no limit)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -125,7 +123,6 @@ func prepareQuerierForTraces(t *testing.T, telemetryStore telemetrystore.Telemet
|
||||
fl,
|
||||
0,
|
||||
0, // maxConcurrentQueries (0 means default)
|
||||
0, // maxTimeRange (0 means no limit)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -162,6 +159,5 @@ func prepareQuerierForAITraces(t *testing.T, telemetryStore telemetrystore.Telem
|
||||
fl,
|
||||
0,
|
||||
0, // maxConcurrentQueries (0 means default)
|
||||
0, // maxTimeRange (0 means no limit)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -33,6 +33,28 @@ func ToNanoSecs(epoch uint64) uint64 {
|
||||
return temp * uint64(math.Pow(10, float64(19-count)))
|
||||
}
|
||||
|
||||
// ToMilliSecs takes an epoch whose resolution is inferred from its magnitude
|
||||
// (s/ms/µs/ns) and returns it in milliseconds. A millisecond epoch for the
|
||||
// current era has 13 digits (e.g. ~1.7e12 in 2026), so the value is scaled so
|
||||
// its digit-width matches: smaller values (seconds) are scaled up, larger ones
|
||||
// (micro/nanoseconds) are scaled down. Zero is returned unchanged.
|
||||
func ToMilliSecs(epoch uint64) uint64 {
|
||||
if epoch == 0 {
|
||||
return 0
|
||||
}
|
||||
temp := epoch
|
||||
count := 0
|
||||
for epoch != 0 {
|
||||
epoch /= 10
|
||||
count++
|
||||
}
|
||||
const msDigits = 13
|
||||
if count < msDigits {
|
||||
return temp * uint64(math.Pow(10, float64(msDigits-count)))
|
||||
}
|
||||
return temp / uint64(math.Pow(10, float64(count-msDigits)))
|
||||
}
|
||||
|
||||
// TODO(srikanthccv): should these be rounded to nearest multiple of 60 instead of 5 if step > 60?
|
||||
// That would make graph look nice but "nice" but should be less important than the usefulness.
|
||||
func RecommendedStepInterval(start, end uint64) uint64 {
|
||||
|
||||
@@ -60,3 +60,51 @@ func TestToNanoSecs(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestToMilliSecs(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
epoch uint64
|
||||
expected uint64
|
||||
}{
|
||||
{
|
||||
name: "10-digit Unix timestamp (seconds) - 2023-01-01 00:00:00 UTC",
|
||||
epoch: 1672531200, // seconds
|
||||
expected: 1672531200000, // * 10^3
|
||||
},
|
||||
{
|
||||
name: "13-digit Unix timestamp (milliseconds) - already ms",
|
||||
epoch: 1672531200000,
|
||||
expected: 1672531200000, // unchanged
|
||||
},
|
||||
{
|
||||
name: "16-digit Unix timestamp (microseconds)",
|
||||
epoch: 1672531200000000, // microseconds
|
||||
expected: 1672531200000, // / 10^3
|
||||
},
|
||||
{
|
||||
name: "19-digit Unix timestamp (nanoseconds)",
|
||||
epoch: 1672531200000000000, // nanoseconds
|
||||
expected: 1672531200000, // / 10^6
|
||||
},
|
||||
{
|
||||
name: "Unix epoch start - zero is unchanged",
|
||||
epoch: 0,
|
||||
expected: 0,
|
||||
},
|
||||
{
|
||||
name: "Recent timestamp in seconds - 2024-05-25 12:00:00 UTC",
|
||||
epoch: 1716638400,
|
||||
expected: 1716638400000,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
result := ToMilliSecs(tt.epoch)
|
||||
if result != tt.expected {
|
||||
t.Errorf("ToMilliSecs(%d) = %d, want %d", tt.epoch, result, tt.expected)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,7 +67,7 @@ func TestNewHandlers(t *testing.T) {
|
||||
spanMapperModule := implspanmapper.NewModule(implspanmapper.NewStore(sqlstore), spanMapperRegistry, providerSettings)
|
||||
modules := NewModules(sqlstore, tokenizer, emailing, providerSettings, orgGetter, alertmanager, nil, nil, nil, nil, nil, nil, nil, queryParser, Config{}, dashboardModule, userGetter, userRoleStore, nil, nil, nil, retentionGetter, flagger, tagModule, nil, spanMapperModule)
|
||||
|
||||
querierHandler := querier.NewHandler(providerSettings, nil, nil, querier.Config{})
|
||||
querierHandler := querier.NewHandler(providerSettings, nil, nil)
|
||||
registryHandler := factory.NewHandler(nil)
|
||||
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, nil, registryHandler, alertmanager, nil, nil, nil)
|
||||
reflectVal := reflect.ValueOf(handlers)
|
||||
|
||||
@@ -182,7 +182,7 @@ func New(
|
||||
gatewayProviderFactory func(licensing.Licensing) factory.ProviderFactory[gateway.Gateway, gateway.Config],
|
||||
auditorProviderFactories func(licensing.Licensing) factory.NamedMap[factory.ProviderFactory[auditor.Auditor, auditor.Config]],
|
||||
meterReporterProviderFactories func(context.Context, factory.ProviderSettings, flagger.Flagger, licensing.Licensing, telemetrystore.TelemetryStore, retention.Getter, organization.Getter, zeus.Zeus) (factory.NamedMap[factory.ProviderFactory[meterreporter.Reporter, meterreporter.Config]], string),
|
||||
querierHandlerCallback func(factory.ProviderSettings, querier.Querier, analytics.Analytics, querier.Config) querier.Handler,
|
||||
querierHandlerCallback func(factory.ProviderSettings, querier.Querier, analytics.Analytics) querier.Handler,
|
||||
cloudIntegrationCallback func(sqlstore.SQLStore, dashboard.Module, global.Global, zeus.Zeus, gateway.Gateway, licensing.Licensing, serviceaccount.Module, cloudintegration.Config) (cloudintegration.Module, error),
|
||||
metricReductionRuleModuleCallback func(sqlstore.SQLStore, telemetrystore.TelemetryStore, dashboard.Module, queryparser.QueryParser, licensing.Licensing, flagger.Flagger, telemetrytypes.MetadataStore, factory.ProviderSettings, int) metricreductionrule.Module,
|
||||
rulerProviderFactories func(cache.Cache, alertmanager.Alertmanager, sqlstore.SQLStore, telemetrystore.TelemetryStore, telemetrytypes.MetadataStore, prometheus.Prometheus, organization.Getter, rulestatehistory.Module, querier.Querier, queryparser.QueryParser) factory.NamedMap[factory.ProviderFactory[ruler.Ruler, ruler.Config]],
|
||||
@@ -553,7 +553,7 @@ func New(
|
||||
userService := impluser.NewService(providerSettings, impluser.NewStore(sqlstore, providerSettings), modules.UserGetter, modules.UserSetter, orgGetter, authz, config.User.Root)
|
||||
|
||||
// Initialize the querier handler via callback (allows EE to decorate with anomaly detection)
|
||||
querierHandler := querierHandlerCallback(providerSettings, querier, analytics, config.Querier)
|
||||
querierHandler := querierHandlerCallback(providerSettings, querier, analytics)
|
||||
|
||||
// Create a list of all stats collectors
|
||||
statsCollectors := []statsreporter.StatsCollector{
|
||||
|
||||
@@ -673,12 +673,6 @@ func (r *QueryRangeRequest) UnmarshalJSON(data []byte) error {
|
||||
// Copy the decoded values back to the original struct
|
||||
*r = QueryRangeRequest(temp)
|
||||
|
||||
// Coerce the window to epoch milliseconds up front so every downstream
|
||||
// consumer (TimeRange, narrowWindowByTraceID, step interval, etc.) can
|
||||
// safely assume ms regardless of the resolution the caller sent.
|
||||
r.Start = ToMilliSecs(r.Start)
|
||||
r.End = ToMilliSecs(r.End)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -1,25 +0,0 @@
|
||||
package querybuildertypesv5
|
||||
|
||||
import "math"
|
||||
|
||||
// ToMilliSecs takes an epoch whose resolution is inferred from its magnitude
|
||||
// (s/ms/µs/ns) and returns it in milliseconds. A millisecond epoch for the
|
||||
// current era has 13 digits (e.g. ~1.7e12 in 2026), so the value is scaled so
|
||||
// its digit-width matches: smaller values (seconds) are scaled up, larger ones
|
||||
// (micro/nanoseconds) are scaled down. Zero is returned unchanged.
|
||||
func ToMilliSecs(epoch uint64) uint64 {
|
||||
if epoch == 0 {
|
||||
return 0
|
||||
}
|
||||
temp := epoch
|
||||
count := 0
|
||||
for epoch != 0 {
|
||||
epoch /= 10
|
||||
count++
|
||||
}
|
||||
const msDigits = 13
|
||||
if count < msDigits {
|
||||
return temp * uint64(math.Pow(10, float64(msDigits-count)))
|
||||
}
|
||||
return temp / uint64(math.Pow(10, float64(count-msDigits)))
|
||||
}
|
||||
@@ -1,51 +0,0 @@
|
||||
package querybuildertypesv5
|
||||
|
||||
import "testing"
|
||||
|
||||
func TestToMilliSecs(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
epoch uint64
|
||||
expected uint64
|
||||
}{
|
||||
{
|
||||
name: "10-digit Unix timestamp (seconds) - 2023-01-01 00:00:00 UTC",
|
||||
epoch: 1672531200, // seconds
|
||||
expected: 1672531200000, // * 10^3
|
||||
},
|
||||
{
|
||||
name: "13-digit Unix timestamp (milliseconds) - already ms",
|
||||
epoch: 1672531200000,
|
||||
expected: 1672531200000, // unchanged
|
||||
},
|
||||
{
|
||||
name: "16-digit Unix timestamp (microseconds)",
|
||||
epoch: 1672531200000000, // microseconds
|
||||
expected: 1672531200000, // / 10^3
|
||||
},
|
||||
{
|
||||
name: "19-digit Unix timestamp (nanoseconds)",
|
||||
epoch: 1672531200000000000, // nanoseconds
|
||||
expected: 1672531200000, // / 10^6
|
||||
},
|
||||
{
|
||||
name: "Unix epoch start - zero is unchanged",
|
||||
epoch: 0,
|
||||
expected: 0,
|
||||
},
|
||||
{
|
||||
name: "Recent timestamp in seconds - 2024-05-25 12:00:00 UTC",
|
||||
epoch: 1716638400,
|
||||
expected: 1716638400000,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
result := ToMilliSecs(tt.epoch)
|
||||
if result != tt.expected {
|
||||
t.Errorf("ToMilliSecs(%d) = %d, want %d", tt.epoch, result, tt.expected)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -5,16 +5,12 @@ import (
|
||||
"math"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/errors"
|
||||
"github.com/SigNoz/signoz/pkg/types/metrictypes"
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
|
||||
// maxEndTimeAhead tolerates client clock skew; anything further ahead is a malformed timestamp.
|
||||
const maxEndTimeAhead = 24 * time.Hour
|
||||
|
||||
// getQueryIdentifier returns a friendly identifier for a query based on its type and name/content.
|
||||
func getQueryIdentifier(envelope QueryEnvelope, index int) string {
|
||||
name := envelope.GetQueryName()
|
||||
@@ -86,7 +82,6 @@ type validationConfig struct {
|
||||
skipGroupByValidation bool
|
||||
withTimestampGroupByValidation bool
|
||||
withReduceToValidation bool
|
||||
maxTimeRange time.Duration
|
||||
}
|
||||
|
||||
func applyValidationOptions(opts []ValidationOption) validationConfig {
|
||||
@@ -97,13 +92,6 @@ func applyValidationOptions(opts []ValidationOption) validationConfig {
|
||||
return cfg
|
||||
}
|
||||
|
||||
// WithMaxTimeRange rejects requests whose window is longer than maxTimeRange; 0 disables the limit.
|
||||
func WithMaxTimeRange(maxTimeRange time.Duration) ValidationOption {
|
||||
return func(cfg *validationConfig) {
|
||||
cfg.maxTimeRange = maxTimeRange
|
||||
}
|
||||
}
|
||||
|
||||
// SkipLimitOffsetValidation returns a ValidationOption that skips the limit and offset range checks.
|
||||
// Use this when the caller has already validated limits and offsets with different constraints.
|
||||
func WithSkipLimitOffsetValidation() ValidationOption {
|
||||
@@ -586,8 +574,12 @@ func (q *QueryBuilderQuery[T]) validateOrderByForAggregation() error {
|
||||
|
||||
// Validate validates the entire query range request.
|
||||
func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
|
||||
if err := r.validateTimeRange(applyValidationOptions(opts)); err != nil {
|
||||
return err
|
||||
// Validate time range
|
||||
if r.RequestType != RequestTypeRawStream && r.Start >= r.End {
|
||||
return errors.NewInvalidInputf(
|
||||
errors.CodeInvalidInput,
|
||||
"start time must be before end time",
|
||||
)
|
||||
}
|
||||
|
||||
// Validate request type
|
||||
@@ -638,14 +630,15 @@ func (r *QueryRangeRequest) Validate(opts ...ValidationOption) error {
|
||||
// specs) and returns the request type's ValidationOptions. The dry-run path uses
|
||||
// this so per-query errors can be attributed individually via QueryEnvelope.Validate
|
||||
// instead of failing fast like Validate does.
|
||||
func (r *QueryRangeRequest) ValidateRequestScope(opts ...ValidationOption) ([]ValidationOption, error) {
|
||||
if err := r.validateTimeRange(applyValidationOptions(opts)); err != nil {
|
||||
return nil, err
|
||||
func (r *QueryRangeRequest) ValidateRequestScope() ([]ValidationOption, error) {
|
||||
if r.RequestType != RequestTypeRawStream && r.Start >= r.End {
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "start time must be before end time")
|
||||
}
|
||||
|
||||
var opts []ValidationOption
|
||||
switch r.RequestType {
|
||||
case RequestTypeRaw, RequestTypeRawStream, RequestTypeTrace, RequestTypeTimeSeries, RequestTypeScalar, RequestTypeHeatmap:
|
||||
opts = append(opts, GetValidationOptions(r.RequestType)...)
|
||||
opts = GetValidationOptions(r.RequestType)
|
||||
default:
|
||||
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid request type: %s", r.RequestType).
|
||||
WithAdditional("Valid request types are: raw, timeseries, scalar, heatmap")
|
||||
@@ -706,25 +699,6 @@ func (e QueryEnvelope) Validate(opts ...ValidationOption) error {
|
||||
return validateQueryEnvelope(e, opts...)
|
||||
}
|
||||
|
||||
func (r *QueryRangeRequest) validateTimeRange(cfg validationConfig) error {
|
||||
if r.RequestType == RequestTypeRawStream {
|
||||
return nil
|
||||
}
|
||||
if r.Start == 0 {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "start time is required")
|
||||
}
|
||||
if r.Start >= r.End {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "start time must be before end time")
|
||||
}
|
||||
if r.End > uint64(time.Now().Add(maxEndTimeAhead).UnixMilli()) {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "end time %s is too far in the future", time.UnixMilli(int64(r.End)).UTC().Format(time.RFC3339))
|
||||
}
|
||||
if cfg.maxTimeRange > 0 && time.Duration(r.End-r.Start)*time.Millisecond > cfg.maxTimeRange {
|
||||
return errors.NewInvalidInputf(errors.CodeInvalidInput, "query time range must not exceed %d days", int64(cfg.maxTimeRange/(24*time.Hour)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// validateAllQueriesNotDisabled validates that at least one query in the composite query is enabled.
|
||||
func (r *QueryRangeRequest) validateAllQueriesNotDisabled() error {
|
||||
for _, envelope := range r.CompositeQuery.Queries {
|
||||
|
||||
@@ -1,166 +0,0 @@
|
||||
from collections.abc import Callable
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from http import HTTPStatus
|
||||
|
||||
import pytest
|
||||
|
||||
from fixtures import types
|
||||
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
|
||||
from fixtures.logs import Logs
|
||||
from fixtures.metrics import Metrics
|
||||
from fixtures.querier import (
|
||||
Aggregation,
|
||||
BuilderQuery,
|
||||
MetricAggregation,
|
||||
get_all_warnings,
|
||||
get_error_message,
|
||||
get_series_values,
|
||||
make_query_request,
|
||||
)
|
||||
from fixtures.traces import TraceIdGenerator, Traces
|
||||
|
||||
TWO_YEARS = timedelta(days=730)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"start_ms,end_ms,message",
|
||||
[
|
||||
pytest.param(0, 1_700_000_000_000, "start time is required", id="zero_start"),
|
||||
pytest.param(1_700_000_000_000, 1_700_000_000_000, "start time must be before end time", id="start_equals_end"),
|
||||
# 2592000000 is 30 days in ms; read as an epoch in seconds it is year 2052.
|
||||
pytest.param(1_700_000_000_000, 2_592_000_000, "end time 2052-02-20T00:00:00Z is too far in the future", id="end_in_2052"),
|
||||
pytest.param(1_700_000_000_000 - int((TWO_YEARS + timedelta(days=1)).total_seconds() * 1000), 1_700_000_000_000, "query time range must not exceed 730 days", id="range_over_two_years"),
|
||||
],
|
||||
)
|
||||
def test_time_range_rejected(
|
||||
start_ms: int,
|
||||
end_ms: int,
|
||||
message: str,
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
) -> None:
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=start_ms,
|
||||
end_ms=end_ms,
|
||||
queries=[BuilderQuery(signal="logs", aggregations=[Aggregation(expression="count()")], step_interval=60).to_dict()],
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.BAD_REQUEST
|
||||
assert get_error_message(response.json()) == message
|
||||
|
||||
|
||||
def test_two_year_logs_timeseries_step_clamped(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_logs: Callable[[list[Logs]], None],
|
||||
) -> None:
|
||||
# A 1s step over 2 years is clamped to 42045s (at most ~1500 points); end is aligned to it so the bucket count is stable.
|
||||
step_seconds = 42045
|
||||
end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
|
||||
insert_logs([Logs(timestamp=end - timedelta(minutes=1), resources={"service.name": "time-range-logs"}, body="two year span")])
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((end - TWO_YEARS).timestamp() * 1000),
|
||||
end_ms=int(end.timestamp() * 1000),
|
||||
queries=[
|
||||
BuilderQuery(
|
||||
signal="logs",
|
||||
filter_expression="service.name = 'time-range-logs'",
|
||||
aggregations=[Aggregation(expression="count()")],
|
||||
step_interval=1,
|
||||
).to_dict()
|
||||
],
|
||||
format_options={"formatTableResultForUI": False, "fillGaps": True},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
|
||||
assert get_all_warnings(response.json()) == [
|
||||
{"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
|
||||
]
|
||||
values = get_series_values(response.json(), "A")
|
||||
assert len(values) == 1502
|
||||
assert sum(v["value"] for v in values) == 1
|
||||
|
||||
|
||||
def test_two_year_traces_timeseries_step_clamped(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_traces: Callable[[list[Traces]], None],
|
||||
) -> None:
|
||||
# Same clamp as logs: 42045s; end is aligned to it so the bucket count is stable.
|
||||
step_seconds = 42045
|
||||
end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
|
||||
insert_traces([Traces(timestamp=end - timedelta(minutes=1), trace_id=TraceIdGenerator.trace_id(), span_id=TraceIdGenerator.span_id(), resources={"service.name": "time-range-traces"})])
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((end - TWO_YEARS).timestamp() * 1000),
|
||||
end_ms=int(end.timestamp() * 1000),
|
||||
queries=[
|
||||
BuilderQuery(
|
||||
signal="traces",
|
||||
filter_expression="service.name = 'time-range-traces'",
|
||||
aggregations=[Aggregation(expression="count()")],
|
||||
step_interval=1,
|
||||
).to_dict()
|
||||
],
|
||||
format_options={"formatTableResultForUI": False, "fillGaps": True},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
|
||||
assert get_all_warnings(response.json()) == [
|
||||
{"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
|
||||
]
|
||||
values = get_series_values(response.json(), "A")
|
||||
assert len(values) == 1502
|
||||
assert sum(v["value"] for v in values) == 1
|
||||
|
||||
|
||||
def test_two_year_metrics_timeseries_step_clamped(
|
||||
signoz: types.SigNoz,
|
||||
create_user_admin: None, # pylint: disable=unused-argument
|
||||
get_token: Callable[[str, str], str],
|
||||
insert_metrics: Callable[[list[Metrics]], None],
|
||||
) -> None:
|
||||
# A 1s step over 2 years is clamped to 210600s (~300 points); end is aligned to it so the bucket count is stable.
|
||||
step_seconds = 210600
|
||||
metric_name = "time_range_two_year_gauge"
|
||||
end = datetime.fromtimestamp(int(datetime.now(tz=UTC).timestamp()) // step_seconds * step_seconds, tz=UTC)
|
||||
insert_metrics([Metrics(metric_name=metric_name, labels={"service": "time-range-metrics"}, timestamp=end - timedelta(minutes=1), value=1.0, temporality="Unspecified", type_="Gauge", is_monotonic=False)])
|
||||
|
||||
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
|
||||
response = make_query_request(
|
||||
signoz,
|
||||
token,
|
||||
start_ms=int((end - TWO_YEARS).timestamp() * 1000),
|
||||
end_ms=int(end.timestamp() * 1000),
|
||||
queries=[
|
||||
BuilderQuery(
|
||||
signal="metrics",
|
||||
aggregations=[MetricAggregation(metric_name=metric_name, time_aggregation="latest", space_aggregation="sum", temporality="unspecified")],
|
||||
step_interval=1,
|
||||
).to_dict()
|
||||
],
|
||||
format_options={"formatTableResultForUI": False, "fillGaps": True},
|
||||
)
|
||||
|
||||
assert response.status_code == HTTPStatus.OK
|
||||
assert response.json()["data"]["meta"]["stepIntervals"] == {"A": step_seconds}
|
||||
assert get_all_warnings(response.json()) == [
|
||||
{"message": f"Query A is requesting aggregation interval 1 seconds, which is smaller than the minimum allowed interval of {step_seconds} seconds for selected time range. Using the minimum instead"},
|
||||
]
|
||||
assert len(get_series_values(response.json(), "A")) == 301
|
||||