Compare commits

..

3 Commits

Author SHA1 Message Date
Ashwin Bhatkal
ed85fed711 chore: resolve comments 2026-03-23 21:54:48 +05:30
Ashwin Bhatkal
194e8474b3 chore: fix tests 2026-03-23 21:54:48 +05:30
Ashwin Bhatkal
895ef98be4 fix: bugs in dashboard filter expression 2026-03-23 21:54:48 +05:30
118 changed files with 2423 additions and 11463 deletions

View File

@@ -17,7 +17,5 @@
},
"[html]": {
"editor.defaultFormatter": "vscode.html-language-features"
},
"python-envs.defaultEnvManager": "ms-python.python:system",
"python-envs.pythonProjects": []
}
}

View File

@@ -144,8 +144,6 @@ telemetrystore:
##################### Prometheus #####################
prometheus:
# The maximum time a PromQL query is allowed to run before being aborted.
timeout: 2m
active_query_tracker:
# Whether to enable the active query tracker.
enabled: true

View File

@@ -190,7 +190,7 @@ services:
# - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
signoz:
!!merge <<: *db-depend
image: signoz/signoz:v0.117.1
image: signoz/signoz:v0.116.1
ports:
- "8080:8080" # signoz port
# - "6060:6060" # pprof port

View File

@@ -117,7 +117,7 @@ services:
# - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
signoz:
!!merge <<: *db-depend
image: signoz/signoz:v0.117.1
image: signoz/signoz:v0.116.1
ports:
- "8080:8080" # signoz port
volumes:

View File

@@ -181,7 +181,7 @@ services:
# - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
signoz:
!!merge <<: *db-depend
image: signoz/signoz:${VERSION:-v0.117.1}
image: signoz/signoz:${VERSION:-v0.116.1}
container_name: signoz
ports:
- "8080:8080" # signoz port

View File

@@ -109,7 +109,7 @@ services:
# - ../common/clickhouse/storage.xml:/etc/clickhouse-server/config.d/storage.xml
signoz:
!!merge <<: *db-depend
image: signoz/signoz:${VERSION:-v0.117.1}
image: signoz/signoz:${VERSION:-v0.116.1}
container_name: signoz
ports:
- "8080:8080" # signoz port

File diff suppressed because it is too large Load Diff

View File

@@ -123,7 +123,6 @@ if err := router.Handle("/api/v1/things", handler.New(
Description: "This endpoint creates a thing",
Request: new(types.PostableThing),
RequestContentType: "application/json",
RequestQuery: new(types.QueryableThing),
Response: new(types.GettableThing),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusCreated,
@@ -156,8 +155,6 @@ The `handler.New` function ties the HTTP handler to OpenAPI metadata via `OpenAP
- **Request / RequestContentType**:
- `Request` is a Go type that describes the request body or form.
- `RequestContentType` is usually `"application/json"` or `"application/x-www-form-urlencoded"` (for callbacks like SAML).
- **RequestQuery**:
- `RequestQuery` is a Go type that descirbes query url params.
- **RequestExamples**: An array of `handler.OpenAPIExample` that provide concrete request payloads in the generated spec. See [Adding request examples](#adding-request-examples) below.
- **Response / ResponseContentType**:
- `Response` is the Go type for the successful response payload.

View File

@@ -273,7 +273,6 @@ Options can be simple (direct link) or nested (with another question):
- Place logo files in `public/Logos/`
- Use SVG format
- Reference as `"/Logos/your-logo.svg"`
- **Fetching Icons**: New icons can be easily fetched from [OpenBrand](https://openbrand.sh/). Use the pattern `https://openbrand.sh/?url=<TARGET_URL>`, where `<TARGET_URL>` is the URL-encoded link to the service's website. For example, to get Render's logo, use [https://openbrand.sh/?url=https%3A%2F%2Frender.com](https://openbrand.sh/?url=https%3A%2F%2Frender.com).
- **Optimize new SVGs**: Run any newly downloaded SVGs through an optimizer like [SVGOMG (svgo)](https://svgomg.net/) or use `npx svgo public/Logos/your-logo.svg` to minimise their size before committing.
### 4. Links

View File

@@ -57,10 +57,6 @@ func (provider *provider) Start(ctx context.Context) error {
return provider.openfgaServer.Start(ctx)
}
func (provider *provider) Healthy() <-chan struct{} {
return provider.openfgaServer.Healthy()
}
func (provider *provider) Stop(ctx context.Context) error {
return provider.openfgaServer.Stop(ctx)
}

View File

@@ -16,6 +16,7 @@ type Server struct {
}
func NewOpenfgaServer(ctx context.Context, pkgAuthzService authz.AuthZ) (*Server, error) {
return &Server{
pkgAuthzService: pkgAuthzService,
}, nil
@@ -25,10 +26,6 @@ func (server *Server) Start(ctx context.Context) error {
return server.pkgAuthzService.Start(ctx)
}
func (server *Server) Healthy() <-chan struct{} {
return server.pkgAuthzService.Healthy()
}
func (server *Server) Stop(ctx context.Context) error {
return server.pkgAuthzService.Stop(ctx)
}

View File

@@ -136,7 +136,6 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
logParsingPipelineController, err := logparsingpipeline.NewLogParsingPipelinesController(
signoz.SQLStore,
integrationsController.GetPipelinesForInstalledIntegrations,
reader,
)
if err != nil {
return nil, err

View File

@@ -7,21 +7,29 @@ import (
"log/slog"
"math"
"strings"
"sync"
"time"
"github.com/SigNoz/signoz/ee/query-service/anomaly"
"github.com/SigNoz/signoz/pkg/cache"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/model"
"github.com/SigNoz/signoz/pkg/transition"
"github.com/SigNoz/signoz/pkg/types/ruletypes"
"github.com/SigNoz/signoz/pkg/valuer"
querierV2 "github.com/SigNoz/signoz/pkg/query-service/app/querier/v2"
"github.com/SigNoz/signoz/pkg/query-service/app/queryBuilder"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
baserules "github.com/SigNoz/signoz/pkg/query-service/rules"
"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
"github.com/SigNoz/signoz/pkg/query-service/utils/times"
"github.com/SigNoz/signoz/pkg/query-service/utils/timestamp"
"github.com/SigNoz/signoz/pkg/units"
baserules "github.com/SigNoz/signoz/pkg/query-service/rules"
querierV5 "github.com/SigNoz/signoz/pkg/querier"
@@ -37,6 +45,16 @@ const (
type AnomalyRule struct {
*baserules.BaseRule
mtx sync.Mutex
reader interfaces.Reader
// querierV2 is used for alerts created after the introduction of new metrics query builder
querierV2 interfaces.Querier
// querierV5 is used for alerts migrated after the introduction of new query builder
querierV5 querierV5.Querier
provider anomaly.Provider
providerV2 anomalyV2.Provider
@@ -85,6 +103,14 @@ func NewAnomalyRule(
logger.Info("using seasonality", "seasonality", t.seasonality.String())
querierOptsV2 := querierV2.QuerierOptions{
Reader: reader,
Cache: cache,
KeyGenerator: queryBuilder.NewKeyGenerator(),
}
t.querierV2 = querierV2.NewQuerier(querierOptsV2)
t.reader = reader
if t.seasonality == anomaly.SeasonalityHourly {
t.provider = anomaly.NewHourlyProvider(
anomaly.WithCache[*anomaly.HourlyProvider](cache),
@@ -122,6 +148,7 @@ func NewAnomalyRule(
)
}
t.querierV5 = querierV5
t.version = p.Version
t.logger = logger
return &t, nil
@@ -306,8 +333,14 @@ func (r *AnomalyRule) buildAndRunQueryV5(ctx context.Context, orgID valuer.UUID,
}
func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
prevState := r.State()
valueFormatter := units.FormatterFromUnit(r.Unit())
var res ruletypes.Vector
var err error
if r.version == "v5" {
r.logger.InfoContext(ctx, "running v5 query")
res, err = r.buildAndRunQueryV5(ctx, r.OrgID(), ts)
@@ -319,8 +352,220 @@ func (r *AnomalyRule) Eval(ctx context.Context, ts time.Time) (int, error) {
return 0, err
}
opts := baserules.EvalVectorOptions{
DeleteLabels: []string{labels.MetricNameLabel, labels.TemporalityLabel},
r.mtx.Lock()
defer r.mtx.Unlock()
resultFPs := map[uint64]struct{}{}
var alerts = make(map[uint64]*ruletypes.Alert, len(res))
ruleReceivers := r.Threshold.GetRuleReceivers()
ruleReceiverMap := make(map[string][]string)
for _, value := range ruleReceivers {
ruleReceiverMap[value.Name] = value.Channels
}
return r.EvalVector(ctx, ts, res, opts)
for _, smpl := range res {
l := make(map[string]string, len(smpl.Metric))
for _, lbl := range smpl.Metric {
l[lbl.Name] = lbl.Value
}
value := valueFormatter.Format(smpl.V, r.Unit())
threshold := valueFormatter.Format(smpl.Target, smpl.TargetUnit)
r.logger.DebugContext(ctx, "Alert template data for rule", "rule_name", r.Name(), "formatter", valueFormatter.Name(), "value", value, "threshold", threshold)
tmplData := ruletypes.AlertTemplateData(l, value, threshold)
// Inject some convenience variables that are easier to remember for users
// who are not used to Go's templating system.
defs := "{{$labels := .Labels}}{{$value := .Value}}{{$threshold := .Threshold}}"
// utility function to apply go template on labels and annotations
expand := func(text string) string {
tmpl := ruletypes.NewTemplateExpander(
ctx,
defs+text,
"__alert_"+r.Name(),
tmplData,
times.Time(timestamp.FromTime(ts)),
nil,
)
result, err := tmpl.Expand()
if err != nil {
result = fmt.Sprintf("<error expanding template: %s>", err)
r.logger.ErrorContext(ctx, "Expanding alert template failed", errors.Attr(err), "data", tmplData, "rule_name", r.Name())
}
return result
}
lb := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel)
resultLabels := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel).Labels()
for name, value := range r.Labels().Map() {
lb.Set(name, expand(value))
}
lb.Set(labels.AlertNameLabel, r.Name())
lb.Set(labels.AlertRuleIdLabel, r.ID())
lb.Set(labels.RuleSourceLabel, r.GeneratorURL())
annotations := make(labels.Labels, 0, len(r.Annotations().Map()))
for name, value := range r.Annotations().Map() {
annotations = append(annotations, labels.Label{Name: name, Value: expand(value)})
}
if smpl.IsMissing {
lb.Set(labels.AlertNameLabel, "[No data] "+r.Name())
lb.Set(labels.NoDataLabel, "true")
}
lbs := lb.Labels()
h := lbs.Hash()
resultFPs[h] = struct{}{}
if _, ok := alerts[h]; ok {
r.logger.ErrorContext(ctx, "the alert query returns duplicate records", "rule_id", r.ID(), "alert", alerts[h])
err = fmt.Errorf("duplicate alert found, vector contains metrics with the same labelset after applying alert labels")
return 0, err
}
alerts[h] = &ruletypes.Alert{
Labels: lbs,
QueryResultLables: resultLabels,
Annotations: annotations,
ActiveAt: ts,
State: model.StatePending,
Value: smpl.V,
GeneratorURL: r.GeneratorURL(),
Receivers: ruleReceiverMap[lbs.Map()[ruletypes.LabelThresholdName]],
Missing: smpl.IsMissing,
IsRecovering: smpl.IsRecovering,
}
}
r.logger.InfoContext(ctx, "number of alerts found", "rule_name", r.Name(), "alerts_count", len(alerts))
// alerts[h] is ready, add or update active list now
for h, a := range alerts {
// Check whether we already have alerting state for the identifying label set.
// Update the last value and annotations if so, create a new alert entry otherwise.
if alert, ok := r.Active[h]; ok && alert.State != model.StateInactive {
alert.Value = a.Value
alert.Annotations = a.Annotations
// Update the recovering and missing state of existing alert
alert.IsRecovering = a.IsRecovering
alert.Missing = a.Missing
if v, ok := alert.Labels.Map()[ruletypes.LabelThresholdName]; ok {
alert.Receivers = ruleReceiverMap[v]
}
continue
}
r.Active[h] = a
}
itemsToAdd := []model.RuleStateHistory{}
// Check if any pending alerts should be removed or fire now. Write out alert timeseries.
for fp, a := range r.Active {
labelsJSON, err := json.Marshal(a.QueryResultLables)
if err != nil {
r.logger.ErrorContext(ctx, "error marshaling labels", errors.Attr(err), "labels", a.Labels)
}
if _, ok := resultFPs[fp]; !ok {
// If the alert was previously firing, keep it around for a given
// retention time so it is reported as resolved to the AlertManager.
if a.State == model.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
delete(r.Active, fp)
}
if a.State != model.StateInactive {
a.State = model.StateInactive
a.ResolvedAt = ts
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: model.StateInactive,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
continue
}
if a.State == model.StatePending && ts.Sub(a.ActiveAt) >= r.HoldDuration().Duration() {
a.State = model.StateFiring
a.FiredAt = ts
state := model.StateFiring
if a.Missing {
state = model.StateNoData
}
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
// We need to change firing alert to recovering if the returned sample meets recovery threshold
changeFiringToRecovering := a.State == model.StateFiring && a.IsRecovering
// We need to change recovering alerts to firing if the returned sample meets target threshold
changeRecoveringToFiring := a.State == model.StateRecovering && !a.IsRecovering && !a.Missing
// in any of the above case we need to update the status of alert
if changeFiringToRecovering || changeRecoveringToFiring {
state := model.StateRecovering
if changeRecoveringToFiring {
state = model.StateFiring
}
a.State = state
r.logger.DebugContext(ctx, "converting alert state", "name", r.Name(), "state", state)
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
}
currentState := r.State()
overallStateChanged := currentState != prevState
for idx, item := range itemsToAdd {
item.OverallStateChanged = overallStateChanged
item.OverallState = currentState
itemsToAdd[idx] = item
}
r.RecordRuleStateHistory(ctx, prevState, currentState, itemsToAdd)
return len(r.Active), nil
}
func (r *AnomalyRule) String() string {
ar := ruletypes.PostableRule{
AlertName: r.Name(),
RuleCondition: r.Condition(),
EvalWindow: r.EvalWindow(),
Labels: r.Labels().Map(),
Annotations: r.Annotations().Map(),
PreferredChannels: r.PreferredChannels(),
}
byt, err := json.Marshal(ar)
if err != nil {
return fmt.Sprintf("error marshaling alerting rule: %s", err.Error())
}
return string(byt)
}

View File

@@ -257,7 +257,7 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
WillReturnRows(samplesRows)
// Create Prometheus provider for this test
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, store)
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, store)
},
ManagerOptionsHook: func(opts *rules.ManagerOptions) {
// Set Prometheus provider for PromQL queries

View File

@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" fill="#fa520f" viewBox="0 0 24 24"><title>Mistral AI</title><path d="M17.143 3.429v3.428h-3.429v3.429h-3.428V6.857H6.857V3.43H3.43v13.714H0v3.428h10.286v-3.428H6.857v-3.429h3.429v3.429h3.429v-3.429h3.428v3.429h-3.428v3.428H24v-3.428h-3.43V3.429z"/></svg>

Before

Width:  |  Height:  |  Size: 294 B

View File

@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" fill="none" viewBox="0 0 120 120"><defs><linearGradient id="a" x1="0%" x2="100%" y1="0%" y2="100%"><stop offset="0%" stop-color="#ff4d4d"/><stop offset="100%" stop-color="#991b1b"/></linearGradient></defs><path fill="url(#a)" d="M60 10c-30 0-45 25-45 45s15 40 30 45v10h10v-10s5 2 10 0v10h10v-10c15-5 30-25 30-45S90 10 60 10"/><path fill="url(#a)" d="M20 45C5 40 0 50 5 60s15 5 20-5c3-7 0-10-5-10"/><path fill="url(#a)" d="M100 45c15-5 20 5 15 15s-15 5-20-5c-3-7 0-10 5-10"/><path stroke="#ff4d4d" stroke-linecap="round" stroke-width="3" d="M45 15Q35 5 30 8M75 15Q85 5 90 8"/><circle cx="45" cy="35" r="6" fill="#050810"/><circle cx="75" cy="35" r="6" fill="#050810"/><circle cx="46" cy="34" r="2.5" fill="#00e5cc"/><circle cx="76" cy="34" r="2.5" fill="#00e5cc"/></svg>

Before

Width:  |  Height:  |  Size: 809 B

View File

@@ -1 +0,0 @@
<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 24 24"><title>Render</title><path d="M18.263.007c-3.121-.147-5.744 2.109-6.192 5.082-.018.138-.045.272-.067.405-.696 3.703-3.936 6.507-7.827 6.507a7.9 7.9 0 0 1-3.825-.979.202.202 0 0 0-.302.178V24H12v-8.999c0-1.656 1.338-3 2.987-3h2.988c3.382 0 6.103-2.817 5.97-6.244-.12-3.084-2.61-5.603-5.682-5.75"/></svg>

Before

Width:  |  Height:  |  Size: 362 B

File diff suppressed because it is too large Load Diff

View File

@@ -1,250 +0,0 @@
/**
* ! Do not edit manually
* * The file has been auto-generated using Orval for SigNoz
* * regenerate with 'yarn generate:api'
* SigNoz
*/
import type {
InvalidateOptions,
QueryClient,
QueryFunction,
QueryKey,
UseQueryOptions,
UseQueryResult,
} from 'react-query';
import { useQuery } from 'react-query';
import type { ErrorType } from '../../../generatedAPIInstance';
import { GeneratedAPIInstance } from '../../../generatedAPIInstance';
import type {
Healthz200,
Healthz503,
Livez200,
Readyz200,
Readyz503,
RenderErrorResponseDTO,
} from '../sigNoz.schemas';
/**
* @summary Health check
*/
export const healthz = (signal?: AbortSignal) => {
return GeneratedAPIInstance<Healthz200>({
url: `/api/v2/healthz`,
method: 'GET',
signal,
});
};
export const getHealthzQueryKey = () => {
return [`/api/v2/healthz`] as const;
};
export const getHealthzQueryOptions = <
TData = Awaited<ReturnType<typeof healthz>>,
TError = ErrorType<Healthz503>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof healthz>>, TError, TData>;
}) => {
const { query: queryOptions } = options ?? {};
const queryKey = queryOptions?.queryKey ?? getHealthzQueryKey();
const queryFn: QueryFunction<Awaited<ReturnType<typeof healthz>>> = ({
signal,
}) => healthz(signal);
return { queryKey, queryFn, ...queryOptions } as UseQueryOptions<
Awaited<ReturnType<typeof healthz>>,
TError,
TData
> & { queryKey: QueryKey };
};
export type HealthzQueryResult = NonNullable<
Awaited<ReturnType<typeof healthz>>
>;
export type HealthzQueryError = ErrorType<Healthz503>;
/**
* @summary Health check
*/
export function useHealthz<
TData = Awaited<ReturnType<typeof healthz>>,
TError = ErrorType<Healthz503>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof healthz>>, TError, TData>;
}): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
const queryOptions = getHealthzQueryOptions(options);
const query = useQuery(queryOptions) as UseQueryResult<TData, TError> & {
queryKey: QueryKey;
};
query.queryKey = queryOptions.queryKey;
return query;
}
/**
* @summary Health check
*/
export const invalidateHealthz = async (
queryClient: QueryClient,
options?: InvalidateOptions,
): Promise<QueryClient> => {
await queryClient.invalidateQueries(
{ queryKey: getHealthzQueryKey() },
options,
);
return queryClient;
};
/**
* @summary Liveness check
*/
export const livez = (signal?: AbortSignal) => {
return GeneratedAPIInstance<Livez200>({
url: `/api/v2/livez`,
method: 'GET',
signal,
});
};
export const getLivezQueryKey = () => {
return [`/api/v2/livez`] as const;
};
export const getLivezQueryOptions = <
TData = Awaited<ReturnType<typeof livez>>,
TError = ErrorType<RenderErrorResponseDTO>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof livez>>, TError, TData>;
}) => {
const { query: queryOptions } = options ?? {};
const queryKey = queryOptions?.queryKey ?? getLivezQueryKey();
const queryFn: QueryFunction<Awaited<ReturnType<typeof livez>>> = ({
signal,
}) => livez(signal);
return { queryKey, queryFn, ...queryOptions } as UseQueryOptions<
Awaited<ReturnType<typeof livez>>,
TError,
TData
> & { queryKey: QueryKey };
};
export type LivezQueryResult = NonNullable<Awaited<ReturnType<typeof livez>>>;
export type LivezQueryError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary Liveness check
*/
export function useLivez<
TData = Awaited<ReturnType<typeof livez>>,
TError = ErrorType<RenderErrorResponseDTO>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof livez>>, TError, TData>;
}): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
const queryOptions = getLivezQueryOptions(options);
const query = useQuery(queryOptions) as UseQueryResult<TData, TError> & {
queryKey: QueryKey;
};
query.queryKey = queryOptions.queryKey;
return query;
}
/**
* @summary Liveness check
*/
export const invalidateLivez = async (
queryClient: QueryClient,
options?: InvalidateOptions,
): Promise<QueryClient> => {
await queryClient.invalidateQueries({ queryKey: getLivezQueryKey() }, options);
return queryClient;
};
/**
* @summary Readiness check
*/
export const readyz = (signal?: AbortSignal) => {
return GeneratedAPIInstance<Readyz200>({
url: `/api/v2/readyz`,
method: 'GET',
signal,
});
};
export const getReadyzQueryKey = () => {
return [`/api/v2/readyz`] as const;
};
export const getReadyzQueryOptions = <
TData = Awaited<ReturnType<typeof readyz>>,
TError = ErrorType<Readyz503>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof readyz>>, TError, TData>;
}) => {
const { query: queryOptions } = options ?? {};
const queryKey = queryOptions?.queryKey ?? getReadyzQueryKey();
const queryFn: QueryFunction<Awaited<ReturnType<typeof readyz>>> = ({
signal,
}) => readyz(signal);
return { queryKey, queryFn, ...queryOptions } as UseQueryOptions<
Awaited<ReturnType<typeof readyz>>,
TError,
TData
> & { queryKey: QueryKey };
};
export type ReadyzQueryResult = NonNullable<Awaited<ReturnType<typeof readyz>>>;
export type ReadyzQueryError = ErrorType<Readyz503>;
/**
* @summary Readiness check
*/
export function useReadyz<
TData = Awaited<ReturnType<typeof readyz>>,
TError = ErrorType<Readyz503>
>(options?: {
query?: UseQueryOptions<Awaited<ReturnType<typeof readyz>>, TError, TData>;
}): UseQueryResult<TData, TError> & { queryKey: QueryKey } {
const queryOptions = getReadyzQueryOptions(options);
const query = useQuery(queryOptions) as UseQueryResult<TData, TError> & {
queryKey: QueryKey;
};
query.queryKey = queryOptions.queryKey;
return query;
}
/**
* @summary Readiness check
*/
export const invalidateReadyz = async (
queryClient: QueryClient,
options?: InvalidateOptions,
): Promise<QueryClient> => {
await queryClient.invalidateQueries(
{ queryKey: getReadyzQueryKey() },
options,
);
return queryClient;
};

View File

@@ -20,113 +20,11 @@ import { useMutation, useQuery } from 'react-query';
import type { BodyType, ErrorType } from '../../../generatedAPIInstance';
import { GeneratedAPIInstance } from '../../../generatedAPIInstance';
import type {
HandleExportRawDataPOSTParams,
ListPromotedAndIndexedPaths200,
PromotetypesPromotePathDTO,
Querybuildertypesv5QueryRangeRequestDTO,
RenderErrorResponseDTO,
} from '../sigNoz.schemas';
/**
* This endpoints allows complex query exporting raw data for traces and logs
* @summary Export raw data
*/
export const handleExportRawDataPOST = (
querybuildertypesv5QueryRangeRequestDTO: BodyType<Querybuildertypesv5QueryRangeRequestDTO>,
params?: HandleExportRawDataPOSTParams,
signal?: AbortSignal,
) => {
return GeneratedAPIInstance<string>({
url: `/api/v1/export_raw_data`,
method: 'POST',
headers: { 'Content-Type': 'application/json' },
data: querybuildertypesv5QueryRangeRequestDTO,
params,
signal,
});
};
export const getHandleExportRawDataPOSTMutationOptions = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof handleExportRawDataPOST>>,
TError,
{
data: BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
params?: HandleExportRawDataPOSTParams;
},
TContext
>;
}): UseMutationOptions<
Awaited<ReturnType<typeof handleExportRawDataPOST>>,
TError,
{
data: BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
params?: HandleExportRawDataPOSTParams;
},
TContext
> => {
const mutationKey = ['handleExportRawDataPOST'];
const { mutation: mutationOptions } = options
? options.mutation &&
'mutationKey' in options.mutation &&
options.mutation.mutationKey
? options
: { ...options, mutation: { ...options.mutation, mutationKey } }
: { mutation: { mutationKey } };
const mutationFn: MutationFunction<
Awaited<ReturnType<typeof handleExportRawDataPOST>>,
{
data: BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
params?: HandleExportRawDataPOSTParams;
}
> = (props) => {
const { data, params } = props ?? {};
return handleExportRawDataPOST(data, params);
};
return { mutationFn, ...mutationOptions };
};
export type HandleExportRawDataPOSTMutationResult = NonNullable<
Awaited<ReturnType<typeof handleExportRawDataPOST>>
>;
export type HandleExportRawDataPOSTMutationBody = BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
export type HandleExportRawDataPOSTMutationError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary Export raw data
*/
export const useHandleExportRawDataPOST = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof handleExportRawDataPOST>>,
TError,
{
data: BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
params?: HandleExportRawDataPOSTParams;
},
TContext
>;
}): UseMutationResult<
Awaited<ReturnType<typeof handleExportRawDataPOST>>,
TError,
{
data: BodyType<Querybuildertypesv5QueryRangeRequestDTO>;
params?: HandleExportRawDataPOSTParams;
},
TContext
> => {
const mutationOptions = getHandleExportRawDataPOSTMutationOptions(options);
return useMutation(mutationOptions);
};
/**
* This endpoints promotes and indexes paths
* @summary Promote and index paths

View File

@@ -425,39 +425,6 @@ export interface AuthtypesSessionContextDTO {
orgs?: AuthtypesOrgSessionContextDTO[] | null;
}
export interface AuthtypesStorableRoleDTO {
/**
* @type string
* @format date-time
*/
createdAt?: Date;
/**
* @type string
*/
description?: string;
/**
* @type string
*/
id: string;
/**
* @type string
*/
name?: string;
/**
* @type string
*/
orgId?: string;
/**
* @type string
*/
type?: string;
/**
* @type string
* @format date-time
*/
updatedAt?: Date;
}
export interface AuthtypesTransactionDTO {
object: AuthtypesObjectDTO;
/**
@@ -470,504 +437,6 @@ export interface AuthtypesUpdateableAuthDomainDTO {
config?: AuthtypesAuthDomainConfigDTO;
}
export interface AuthtypesUserRoleDTO {
/**
* @type string
* @format date-time
*/
createdAt?: Date;
/**
* @type string
*/
id: string;
role?: AuthtypesStorableRoleDTO;
/**
* @type string
*/
roleId?: string;
/**
* @type string
* @format date-time
*/
updatedAt?: Date;
/**
* @type string
*/
userId?: string;
}
export interface AuthtypesUserWithRolesDTO {
/**
* @type string
* @format date-time
*/
createdAt?: Date;
/**
* @type string
*/
displayName?: string;
/**
* @type string
*/
email?: string;
/**
* @type string
*/
id: string;
/**
* @type boolean
*/
isRoot?: boolean;
/**
* @type string
*/
orgId?: string;
/**
* @type string
*/
status?: string;
/**
* @type string
* @format date-time
*/
updatedAt?: Date;
/**
* @type array
* @nullable true
*/
userRoles?: AuthtypesUserRoleDTO[] | null;
}
export interface CloudintegrationtypesAWSAccountConfigDTO {
/**
* @type array
*/
regions: string[];
}
export type CloudintegrationtypesAWSCollectionStrategyDTOS3Buckets = {
[key: string]: string[];
};
export interface CloudintegrationtypesAWSCollectionStrategyDTO {
aws_logs?: CloudintegrationtypesAWSLogsStrategyDTO;
aws_metrics?: CloudintegrationtypesAWSMetricsStrategyDTO;
/**
* @type object
*/
s3_buckets?: CloudintegrationtypesAWSCollectionStrategyDTOS3Buckets;
}
export interface CloudintegrationtypesAWSConnectionArtifactDTO {
/**
* @type string
*/
connectionURL: string;
}
export interface CloudintegrationtypesAWSConnectionArtifactRequestDTO {
/**
* @type string
*/
deploymentRegion: string;
/**
* @type array
*/
regions: string[];
}
export interface CloudintegrationtypesAWSIntegrationConfigDTO {
/**
* @type array
*/
enabledRegions: string[];
telemetry: CloudintegrationtypesAWSCollectionStrategyDTO;
}
export type CloudintegrationtypesAWSLogsStrategyDTOCloudwatchLogsSubscriptionsItem = {
/**
* @type string
*/
filter_pattern?: string;
/**
* @type string
*/
log_group_name_prefix?: string;
};
export interface CloudintegrationtypesAWSLogsStrategyDTO {
/**
* @type array
* @nullable true
*/
cloudwatch_logs_subscriptions?:
| CloudintegrationtypesAWSLogsStrategyDTOCloudwatchLogsSubscriptionsItem[]
| null;
}
export type CloudintegrationtypesAWSMetricsStrategyDTOCloudwatchMetricStreamFiltersItem = {
/**
* @type array
*/
MetricNames?: string[];
/**
* @type string
*/
Namespace?: string;
};
export interface CloudintegrationtypesAWSMetricsStrategyDTO {
/**
* @type array
* @nullable true
*/
cloudwatch_metric_stream_filters?:
| CloudintegrationtypesAWSMetricsStrategyDTOCloudwatchMetricStreamFiltersItem[]
| null;
}
export interface CloudintegrationtypesAWSServiceConfigDTO {
logs?: CloudintegrationtypesAWSServiceLogsConfigDTO;
metrics?: CloudintegrationtypesAWSServiceMetricsConfigDTO;
}
export type CloudintegrationtypesAWSServiceLogsConfigDTOS3Buckets = {
[key: string]: string[];
};
export interface CloudintegrationtypesAWSServiceLogsConfigDTO {
/**
* @type boolean
*/
enabled?: boolean;
/**
* @type object
*/
s3_buckets?: CloudintegrationtypesAWSServiceLogsConfigDTOS3Buckets;
}
export interface CloudintegrationtypesAWSServiceMetricsConfigDTO {
/**
* @type boolean
*/
enabled?: boolean;
}
export interface CloudintegrationtypesAccountDTO {
agentReport: CloudintegrationtypesAgentReportDTO;
config: CloudintegrationtypesAccountConfigDTO;
/**
* @type string
* @format date-time
*/
createdAt?: Date;
/**
* @type string
*/
id: string;
/**
* @type string
*/
orgId: string;
/**
* @type string
*/
provider: string;
/**
* @type string
* @nullable true
*/
providerAccountId: string | null;
/**
* @type string
* @format date-time
* @nullable true
*/
removedAt: Date | null;
/**
* @type string
* @format date-time
*/
updatedAt?: Date;
}
export interface CloudintegrationtypesAccountConfigDTO {
aws: CloudintegrationtypesAWSAccountConfigDTO;
}
/**
* @nullable
*/
export type CloudintegrationtypesAgentReportDTOData = {
[key: string]: unknown;
} | null;
/**
* @nullable
*/
export type CloudintegrationtypesAgentReportDTO = {
/**
* @type object
* @nullable true
*/
data: CloudintegrationtypesAgentReportDTOData;
/**
* @type integer
* @format int64
*/
timestampMillis: number;
} | null;
export interface CloudintegrationtypesAssetsDTO {
/**
* @type array
* @nullable true
*/
dashboards?: CloudintegrationtypesDashboardDTO[] | null;
}
export interface CloudintegrationtypesCollectedLogAttributeDTO {
/**
* @type string
*/
name?: string;
/**
* @type string
*/
path?: string;
/**
* @type string
*/
type?: string;
}
export interface CloudintegrationtypesCollectedMetricDTO {
/**
* @type string
*/
description?: string;
/**
* @type string
*/
name?: string;
/**
* @type string
*/
type?: string;
/**
* @type string
*/
unit?: string;
}
export interface CloudintegrationtypesCollectionStrategyDTO {
aws: CloudintegrationtypesAWSCollectionStrategyDTO;
}
export interface CloudintegrationtypesConnectionArtifactDTO {
aws: CloudintegrationtypesAWSConnectionArtifactDTO;
}
export interface CloudintegrationtypesConnectionArtifactRequestDTO {
aws: CloudintegrationtypesAWSConnectionArtifactRequestDTO;
}
export interface CloudintegrationtypesDashboardDTO {
definition?: DashboardtypesStorableDashboardDataDTO;
/**
* @type string
*/
description?: string;
/**
* @type string
*/
id?: string;
/**
* @type string
*/
title?: string;
}
export interface CloudintegrationtypesDataCollectedDTO {
/**
* @type array
* @nullable true
*/
logs?: CloudintegrationtypesCollectedLogAttributeDTO[] | null;
/**
* @type array
* @nullable true
*/
metrics?: CloudintegrationtypesCollectedMetricDTO[] | null;
}
export interface CloudintegrationtypesGettableAccountWithArtifactDTO {
connectionArtifact: CloudintegrationtypesConnectionArtifactDTO;
/**
* @type string
*/
id: string;
}
export interface CloudintegrationtypesGettableAccountsDTO {
/**
* @type array
*/
accounts: CloudintegrationtypesAccountDTO[];
}
export interface CloudintegrationtypesGettableAgentCheckInResponseDTO {
/**
* @type string
*/
account_id: string;
/**
* @type string
*/
cloud_account_id: string;
/**
* @type string
*/
cloudIntegrationId: string;
integration_config: CloudintegrationtypesIntegrationConfigDTO;
integrationConfig: CloudintegrationtypesProviderIntegrationConfigDTO;
/**
* @type string
*/
providerAccountId: string;
/**
* @type string
* @format date-time
* @nullable true
*/
removed_at: Date | null;
/**
* @type string
* @format date-time
* @nullable true
*/
removedAt: Date | null;
}
export interface CloudintegrationtypesGettableServicesMetadataDTO {
/**
* @type array
*/
services: CloudintegrationtypesServiceMetadataDTO[];
}
/**
* @nullable
*/
export type CloudintegrationtypesIntegrationConfigDTO = {
/**
* @type array
*/
enabled_regions: string[];
telemetry: CloudintegrationtypesAWSCollectionStrategyDTO;
} | null;
/**
* @nullable
*/
export type CloudintegrationtypesPostableAgentCheckInRequestDTOData = {
[key: string]: unknown;
} | null;
export interface CloudintegrationtypesPostableAgentCheckInRequestDTO {
/**
* @type string
*/
account_id?: string;
/**
* @type string
*/
cloud_account_id?: string;
/**
* @type string
*/
cloudIntegrationId?: string;
/**
* @type object
* @nullable true
*/
data: CloudintegrationtypesPostableAgentCheckInRequestDTOData;
/**
* @type string
*/
providerAccountId?: string;
}
export interface CloudintegrationtypesProviderIntegrationConfigDTO {
aws: CloudintegrationtypesAWSIntegrationConfigDTO;
}
export interface CloudintegrationtypesServiceDTO {
assets: CloudintegrationtypesAssetsDTO;
dataCollected: CloudintegrationtypesDataCollectedDTO;
/**
* @type string
*/
icon: string;
/**
* @type string
*/
id: string;
/**
* @type string
*/
overview: string;
serviceConfig?: CloudintegrationtypesServiceConfigDTO;
supported_signals: CloudintegrationtypesSupportedSignalsDTO;
telemetryCollectionStrategy: CloudintegrationtypesCollectionStrategyDTO;
/**
* @type string
*/
title: string;
}
export interface CloudintegrationtypesServiceConfigDTO {
aws: CloudintegrationtypesAWSServiceConfigDTO;
}
export interface CloudintegrationtypesServiceMetadataDTO {
/**
* @type boolean
*/
enabled: boolean;
/**
* @type string
*/
icon: string;
/**
* @type string
*/
id: string;
/**
* @type string
*/
title: string;
}
export interface CloudintegrationtypesSupportedSignalsDTO {
/**
* @type boolean
*/
logs?: boolean;
/**
* @type boolean
*/
metrics?: boolean;
}
export interface CloudintegrationtypesUpdatableAccountDTO {
config: CloudintegrationtypesAccountConfigDTO;
}
export interface CloudintegrationtypesUpdatableServiceDTO {
config: CloudintegrationtypesServiceConfigDTO;
}
export interface DashboardtypesDashboardDTO {
/**
* @type string
@@ -1074,23 +543,6 @@ export interface ErrorsResponseerroradditionalDTO {
message?: string;
}
/**
* @nullable
*/
export type FactoryResponseDTOServices = { [key: string]: string[] } | null;
export interface FactoryResponseDTO {
/**
* @type boolean
*/
healthy?: boolean;
/**
* @type object
* @nullable true
*/
services?: FactoryResponseDTOServices;
}
/**
* @nullable
*/
@@ -3180,13 +2632,6 @@ export interface TypesPostableResetPasswordDTO {
token?: string;
}
export interface TypesPostableRoleDTO {
/**
* @type string
*/
name: string;
}
export interface TypesResetPasswordTokenDTO {
/**
* @type string
@@ -3252,13 +2697,6 @@ export interface TypesStorableAPIKeyDTO {
userId?: string;
}
export interface TypesUpdatableUserDTO {
/**
* @type string
*/
displayName: string;
}
export interface TypesUserDTO {
/**
* @type string
@@ -3403,97 +2841,6 @@ export type AuthzResources200 = {
export type ChangePasswordPathParameters = {
id: string;
};
export type AgentCheckInDeprecatedPathParameters = {
cloudProvider: string;
};
export type AgentCheckInDeprecated200 = {
data: CloudintegrationtypesGettableAgentCheckInResponseDTO;
/**
* @type string
*/
status: string;
};
export type ListAccountsPathParameters = {
cloudProvider: string;
};
export type ListAccounts200 = {
data: CloudintegrationtypesGettableAccountsDTO;
/**
* @type string
*/
status: string;
};
export type CreateAccountPathParameters = {
cloudProvider: string;
};
export type CreateAccount200 = {
data: CloudintegrationtypesGettableAccountWithArtifactDTO;
/**
* @type string
*/
status: string;
};
export type DisconnectAccountPathParameters = {
cloudProvider: string;
id: string;
};
export type GetAccountPathParameters = {
cloudProvider: string;
id: string;
};
export type GetAccount200 = {
data: CloudintegrationtypesAccountDTO;
/**
* @type string
*/
status: string;
};
export type UpdateAccountPathParameters = {
cloudProvider: string;
id: string;
};
export type AgentCheckInPathParameters = {
cloudProvider: string;
};
export type AgentCheckIn200 = {
data: CloudintegrationtypesGettableAgentCheckInResponseDTO;
/**
* @type string
*/
status: string;
};
export type ListServicesMetadataPathParameters = {
cloudProvider: string;
};
export type ListServicesMetadata200 = {
data: CloudintegrationtypesGettableServicesMetadataDTO;
/**
* @type string
*/
status: string;
};
export type GetServicePathParameters = {
cloudProvider: string;
serviceId: string;
};
export type GetService200 = {
data: CloudintegrationtypesServiceDTO;
/**
* @type string
*/
status: string;
};
export type UpdateServicePathParameters = {
cloudProvider: string;
serviceId: string;
};
export type CreateSessionByGoogleCallback303 = {
data: AuthtypesGettableTokenDTO;
/**
@@ -3595,19 +2942,6 @@ export type DeleteAuthDomainPathParameters = {
export type UpdateAuthDomainPathParameters = {
id: string;
};
export type HandleExportRawDataPOSTParams = {
/**
* @enum csv,jsonl
* @type string
* @description The output format for the export.
*/
format?: HandleExportRawDataPOSTFormat;
};
export enum HandleExportRawDataPOSTFormat {
csv = 'csv',
jsonl = 'jsonl',
}
export type GetFieldsKeysParams = {
/**
* @description undefined
@@ -3965,7 +3299,7 @@ export type UpdateServiceAccountKeyPathParameters = {
export type UpdateServiceAccountStatusPathParameters = {
id: string;
};
export type ListUsersDeprecated200 = {
export type ListUsers200 = {
/**
* @type array
*/
@@ -3979,10 +3313,10 @@ export type ListUsersDeprecated200 = {
export type DeleteUserPathParameters = {
id: string;
};
export type GetUserDeprecatedPathParameters = {
export type GetUserPathParameters = {
id: string;
};
export type GetUserDeprecated200 = {
export type GetUser200 = {
data: TypesDeprecatedUserDTO;
/**
* @type string
@@ -3990,10 +3324,10 @@ export type GetUserDeprecated200 = {
status: string;
};
export type UpdateUserDeprecatedPathParameters = {
export type UpdateUserPathParameters = {
id: string;
};
export type UpdateUserDeprecated200 = {
export type UpdateUser200 = {
data: TypesDeprecatedUserDTO;
/**
* @type string
@@ -4001,7 +3335,7 @@ export type UpdateUserDeprecated200 = {
status: string;
};
export type GetMyUserDeprecated200 = {
export type GetMyUser200 = {
data: TypesDeprecatedUserDTO;
/**
* @type string
@@ -4123,30 +3457,6 @@ export type SearchIngestionKeys200 = {
status: string;
};
export type Healthz200 = {
data: FactoryResponseDTO;
/**
* @type string
*/
status: string;
};
export type Healthz503 = {
data: FactoryResponseDTO;
/**
* @type string
*/
status: string;
};
export type Livez200 = {
data: FactoryResponseDTO;
/**
* @type string
*/
status: string;
};
export type ListMetricsParams = {
/**
* @type integer
@@ -4282,36 +3592,6 @@ export type GetMyOrganization200 = {
status: string;
};
export type Readyz200 = {
data: FactoryResponseDTO;
/**
* @type string
*/
status: string;
};
export type Readyz503 = {
data: FactoryResponseDTO;
/**
* @type string
*/
status: string;
};
export type GetUsersByRoleIDPathParameters = {
id: string;
};
export type GetUsersByRoleID200 = {
/**
* @type array
*/
data: TypesUserDTO[];
/**
* @type string
*/
status: string;
};
export type GetSessionContext200 = {
data: AuthtypesSessionContextDTO;
/**
@@ -4336,60 +3616,6 @@ export type RotateSession200 = {
status: string;
};
export type ListUsers200 = {
/**
* @type array
*/
data: TypesUserDTO[];
/**
* @type string
*/
status: string;
};
export type GetUserPathParameters = {
id: string;
};
export type GetUser200 = {
data: AuthtypesUserWithRolesDTO;
/**
* @type string
*/
status: string;
};
export type UpdateUserPathParameters = {
id: string;
};
export type GetRolesByUserIDPathParameters = {
id: string;
};
export type GetRolesByUserID200 = {
/**
* @type array
*/
data: AuthtypesRoleDTO[];
/**
* @type string
*/
status: string;
};
export type SetRoleByUserIDPathParameters = {
id: string;
};
export type RemoveUserRoleByUserIDAndRoleIDPathParameters = {
id: string;
roleId: string;
};
export type GetMyUser200 = {
data: AuthtypesUserWithRolesDTO;
/**
* @type string
*/
status: string;
};
export type GetHosts200 = {
data: ZeustypesGettableHostDTO;
/**

File diff suppressed because it is too large Load Diff

View File

@@ -116,12 +116,7 @@ describe.each([
expect(screen.getByRole('dialog')).toBeInTheDocument();
expect(screen.getByText('FORMAT')).toBeInTheDocument();
expect(screen.getByText('Number of Rows')).toBeInTheDocument();
if (dataSource === DataSource.TRACES) {
expect(screen.queryByText('Columns')).not.toBeInTheDocument();
} else {
expect(screen.getByText('Columns')).toBeInTheDocument();
}
expect(screen.getByText('Columns')).toBeInTheDocument();
});
it('allows changing export format', () => {
@@ -151,17 +146,6 @@ describe.each([
});
it('allows changing columns scope', () => {
if (dataSource === DataSource.TRACES) {
renderWithStore(dataSource);
fireEvent.click(screen.getByTestId(testId));
expect(screen.queryByRole('radio', { name: 'All' })).not.toBeInTheDocument();
expect(
screen.queryByRole('radio', { name: 'Selected' }),
).not.toBeInTheDocument();
return;
}
renderWithStore(dataSource);
fireEvent.click(screen.getByTestId(testId));
@@ -226,12 +210,7 @@ describe.each([
mockUseQueryBuilder.mockReturnValue({ stagedQuery: mockQuery });
renderWithStore(dataSource);
fireEvent.click(screen.getByTestId(testId));
// For traces, column scope is always Selected and the radio is hidden
if (dataSource !== DataSource.TRACES) {
fireEvent.click(screen.getByRole('radio', { name: 'Selected' }));
}
fireEvent.click(screen.getByRole('radio', { name: 'Selected' }));
fireEvent.click(screen.getByText('Export'));
await waitFor(() => {
@@ -248,11 +227,6 @@ describe.each([
});
it('sends no selectFields when column scope is All', async () => {
// For traces, column scope is always Selected — this test only applies to other sources
if (dataSource === DataSource.TRACES) {
return;
}
renderWithStore(dataSource);
fireEvent.click(screen.getByTestId(testId));
fireEvent.click(screen.getByRole('radio', { name: 'All' }));

View File

@@ -1,6 +1,5 @@
import { useCallback, useMemo, useState } from 'react';
import { Button, Popover, Radio, Tooltip, Typography } from 'antd';
import { TelemetryFieldKey } from 'api/v5/v5';
import { useExportRawData } from 'hooks/useDownloadOptionsMenu/useDownloadOptionsMenu';
import { Download, DownloadIcon, Loader2 } from 'lucide-react';
import { DataSource } from 'types/common/queryBuilder';
@@ -15,12 +14,10 @@ import './DownloadOptionsMenu.styles.scss';
interface DownloadOptionsMenuProps {
dataSource: DataSource;
selectedColumns?: TelemetryFieldKey[];
}
export default function DownloadOptionsMenu({
dataSource,
selectedColumns,
}: DownloadOptionsMenuProps): JSX.Element {
const [exportFormat, setExportFormat] = useState<string>(DownloadFormats.CSV);
const [rowLimit, setRowLimit] = useState<number>(DownloadRowCounts.TEN_K);
@@ -38,19 +35,9 @@ export default function DownloadOptionsMenu({
await handleExportRawData({
format: exportFormat,
rowLimit,
clearSelectColumns:
dataSource !== DataSource.TRACES &&
columnsScope === DownloadColumnsScopes.ALL,
selectedColumns,
clearSelectColumns: columnsScope === DownloadColumnsScopes.ALL,
});
}, [
exportFormat,
rowLimit,
columnsScope,
selectedColumns,
handleExportRawData,
dataSource,
]);
}, [exportFormat, rowLimit, columnsScope, handleExportRawData]);
const popoverContent = useMemo(
() => (
@@ -85,22 +72,18 @@ export default function DownloadOptionsMenu({
</Radio.Group>
</div>
{dataSource !== DataSource.TRACES && (
<>
<div className="horizontal-line" />
<div className="horizontal-line" />
<div className="columns-scope">
<Typography.Text className="title">Columns</Typography.Text>
<Radio.Group
value={columnsScope}
onChange={(e): void => setColumnsScope(e.target.value)}
>
<Radio value={DownloadColumnsScopes.ALL}>All</Radio>
<Radio value={DownloadColumnsScopes.SELECTED}>Selected</Radio>
</Radio.Group>
</div>
</>
)}
<div className="columns-scope">
<Typography.Text className="title">Columns</Typography.Text>
<Radio.Group
value={columnsScope}
onChange={(e): void => setColumnsScope(e.target.value)}
>
<Radio value={DownloadColumnsScopes.ALL}>All</Radio>
<Radio value={DownloadColumnsScopes.SELECTED}>Selected</Radio>
</Radio.Group>
</div>
<Button
type="primary"
@@ -114,14 +97,7 @@ export default function DownloadOptionsMenu({
</Button>
</div>
),
[
exportFormat,
rowLimit,
columnsScope,
isDownloading,
handleExport,
dataSource,
],
[exportFormat, rowLimit, columnsScope, isDownloading, handleExport],
);
return (

View File

@@ -20,7 +20,7 @@ import { RenderErrorResponseDTO } from 'api/generated/services/sigNoz.schemas';
import {
getResetPasswordToken,
useDeleteUser,
useUpdateUserDeprecated,
useUpdateUser,
} from 'api/generated/services/users';
import { AxiosError } from 'axios';
import { MemberRow } from 'components/MembersTable/MembersTable';
@@ -60,7 +60,7 @@ function EditMemberDrawer({
const isInvited = member?.status === MemberStatus.Invited;
const { mutate: updateUser, isLoading: isSaving } = useUpdateUserDeprecated({
const { mutate: updateUser, isLoading: isSaving } = useUpdateUser({
mutation: {
onSuccess: (): void => {
toast.success('Member details updated successfully', { richColors: true });

View File

@@ -4,7 +4,7 @@ import { convertToApiError } from 'api/ErrorResponseHandlerForGeneratedAPIs';
import {
getResetPasswordToken,
useDeleteUser,
useUpdateUserDeprecated,
useUpdateUser,
} from 'api/generated/services/users';
import { MemberStatus } from 'container/MembersSettings/utils';
import {
@@ -50,7 +50,7 @@ jest.mock('@signozhq/dialog', () => ({
jest.mock('api/generated/services/users', () => ({
useDeleteUser: jest.fn(),
useUpdateUserDeprecated: jest.fn(),
useUpdateUser: jest.fn(),
getResetPasswordToken: jest.fn(),
}));
@@ -105,7 +105,7 @@ function renderDrawer(
describe('EditMemberDrawer', () => {
beforeEach(() => {
jest.clearAllMocks();
(useUpdateUserDeprecated as jest.Mock).mockReturnValue({
(useUpdateUser as jest.Mock).mockReturnValue({
mutate: mockUpdateMutate,
isLoading: false,
});
@@ -130,7 +130,7 @@ describe('EditMemberDrawer', () => {
const onComplete = jest.fn();
const user = userEvent.setup({ pointerEventsCheck: 0 });
(useUpdateUserDeprecated as jest.Mock).mockImplementation((options) => ({
(useUpdateUser as jest.Mock).mockImplementation((options) => ({
mutate: mockUpdateMutate.mockImplementation(() => {
options?.mutation?.onSuccess?.();
}),
@@ -239,7 +239,7 @@ describe('EditMemberDrawer', () => {
const onComplete = jest.fn();
const user = userEvent.setup({ pointerEventsCheck: 0 });
(useUpdateUserDeprecated as jest.Mock).mockImplementation((options) => ({
(useUpdateUser as jest.Mock).mockImplementation((options) => ({
mutate: mockUpdateMutate.mockImplementation(() => {
options?.mutation?.onSuccess?.();
}),
@@ -280,7 +280,7 @@ describe('EditMemberDrawer', () => {
const user = userEvent.setup({ pointerEventsCheck: 0 });
const mockToast = jest.mocked(toast);
(useUpdateUserDeprecated as jest.Mock).mockImplementation((options) => ({
(useUpdateUser as jest.Mock).mockImplementation((options) => ({
mutate: mockUpdateMutate.mockImplementation(() => {
options?.mutation?.onError?.({});
}),

View File

@@ -1191,6 +1191,49 @@ describe('removeKeysFromExpression', () => {
expect(pairs).toHaveLength(2);
});
});
describe('Parenthesised expressions', () => {
it('should not leave a dangling AND when removing the last filter inside parens', () => {
const expression =
'(deployment.environment = $deployment.environment AND service.name = $service.name AND operation IN $top_level_operation)';
const result = removeKeysFromExpression(expression, ['operation'], true);
expect(result).toBe(
'(deployment.environment = $deployment.environment AND service.name = $service.name)',
);
});
it('should not leave a dangling AND when removing the first filter inside parens', () => {
const expression =
'(deployment.environment = $deployment.environment AND service.name = $service.name AND operation IN $top_level_operation)';
const result = removeKeysFromExpression(
expression,
['deployment.environment'],
true,
);
expect(result).toBe(
'(service.name = $service.name AND operation IN $top_level_operation)',
);
});
it('should not leave a dangling AND when removing a middle filter inside parens', () => {
const expression =
'(deployment.environment = $deployment.environment AND service.name = $service.name AND operation IN $top_level_operation)';
const result = removeKeysFromExpression(expression, ['service.name'], true);
expect(result).toBe(
'(deployment.environment = $deployment.environment AND operation IN $top_level_operation)',
);
});
it('should return empty parens when removing the only filter inside parens', () => {
const expression = '(operation IN $top_level_operation)';
const result = removeKeysFromExpression(expression, ['operation'], true);
expect(result).toBe('()');
});
});
});
describe('formatValueForExpression', () => {

View File

@@ -569,7 +569,7 @@ export const removeKeysFromExpression = (
const currentQueryPair = queryPairsMap.get(`${key}`.trim().toLowerCase());
if (currentQueryPair && currentQueryPair.isComplete) {
// Determine the start index of the query pair (fallback order: key → operator → value)
const queryPairStart =
let queryPairStart =
currentQueryPair.position.keyStart ??
currentQueryPair.position.operatorStart ??
currentQueryPair.position.valueStart;
@@ -587,6 +587,15 @@ export const removeKeysFromExpression = (
// If match is found, extend the queryPairEnd to include the matched part
queryPairEnd += match[0].length;
}
// If no following conjunction was absorbed (e.g. removed pair is last in expression),
// absorb the preceding AND/OR instead to avoid leaving a dangling conjunction
if (!match?.[3]) {
const beforePair = updatedExpression.slice(0, queryPairStart);
const precedingConjunctionMatch = beforePair.match(/\s+(AND|OR)\s+$/i);
if (precedingConjunctionMatch) {
queryPairStart -= precedingConjunctionMatch[0].length;
}
}
// Remove the full query pair (including any conjunction/whitespace) from the expression
updatedExpression = `${updatedExpression.slice(
0,

View File

@@ -3,6 +3,7 @@ import { useCallback } from 'react';
import { useAddDynamicVariableToPanels } from 'hooks/dashboard/useAddDynamicVariableToPanels';
import { updateLocalStorageDashboardVariable } from 'hooks/dashboard/useDashboardFromLocalStorage';
import { useUpdateDashboard } from 'hooks/dashboard/useUpdateDashboard';
import { useNotifications } from 'hooks/useNotifications';
import { IDashboardVariables } from 'providers/Dashboard/store/dashboardVariables/dashboardVariablesStoreTypes';
import { useDashboardStore } from 'providers/Dashboard/store/useDashboardStore';
import { IDashboardVariable } from 'types/api/dashboard/getAll';
@@ -51,6 +52,7 @@ export const useDashboardVariableUpdate = (): UseDashboardVariableUpdateReturn =
);
const addDynamicVariableToPanels = useAddDynamicVariableToPanels();
const updateMutation = useUpdateDashboard();
const { notifications } = useNotifications();
const onValueUpdate = useCallback(
(
@@ -180,6 +182,15 @@ export const useDashboardVariableUpdate = (): UseDashboardVariableUpdateReturn =
// Get current dashboard variables
const currentVariables = selectedDashboard.data.variables || {};
// Prevent duplicate variable names
const nameExists = Object.values(currentVariables).some(
(v) => v.name === name,
);
if (nameExists) {
notifications.error({ message: `Variable "${name}" already exists` });
return;
}
// Create tableRowData like Dashboard Settings does
const tableRowData = [];
const variableOrderArr = [];
@@ -232,7 +243,8 @@ export const useDashboardVariableUpdate = (): UseDashboardVariableUpdateReturn =
// Convert to dashboard format and update
const updatedVariables = convertVariablesToDbFormat(tableRowData);
updateVariables(updatedVariables, newVariable.id, [], false);
// Don't pass currentRequestedId — variable creation should not modify widget filters.
updateVariables(updatedVariables);
},
[selectedDashboard, updateVariables],
);

View File

@@ -48,7 +48,6 @@ import DashboardEmptyState from './DashboardEmptyState/DashboardEmptyState';
import GridCard from './GridCard';
import { Card, CardContainer, ReactGridLayout } from './styles';
import {
applyRowCollapse,
hasColumnWidthsChanged,
removeUndefinedValuesFromLayout,
} from './utils';
@@ -269,10 +268,13 @@ function GraphLayout(props: GraphLayoutProps): JSX.Element {
return;
}
const updatedWidgets = selectedDashboard?.data?.widgets?.map((e) =>
e.id === currentSelectRowId ? { ...e, title: newTitle } : e,
currentWidget.title = newTitle;
const updatedWidgets = selectedDashboard?.data?.widgets?.filter(
(e) => e.id !== currentSelectRowId,
);
updatedWidgets?.push(currentWidget);
const updatedSelectedDashboard: Props = {
id: selectedDashboard.id,
data: {
@@ -314,13 +316,88 @@ function GraphLayout(props: GraphLayoutProps): JSX.Element {
if (!selectedDashboard) {
return;
}
const { updatedLayout, updatedPanelMap } = applyRowCollapse(
id,
dashboardLayout,
currentPanelMap,
);
setCurrentPanelMap((prev) => ({ ...prev, ...updatedPanelMap }));
setDashboardLayout(sortLayout(updatedLayout));
const rowProperties = { ...currentPanelMap[id] };
const updatedPanelMap = { ...currentPanelMap };
let updatedDashboardLayout = [...dashboardLayout];
if (rowProperties.collapsed === true) {
rowProperties.collapsed = false;
const widgetsInsideTheRow = rowProperties.widgets;
let maxY = 0;
widgetsInsideTheRow.forEach((w) => {
maxY = Math.max(maxY, w.y + w.h);
});
const currentRowWidget = dashboardLayout.find((w) => w.i === id);
if (currentRowWidget && widgetsInsideTheRow.length) {
maxY -= currentRowWidget.h + currentRowWidget.y;
}
const idxCurrentRow = dashboardLayout.findIndex((w) => w.i === id);
for (let j = idxCurrentRow + 1; j < dashboardLayout.length; j++) {
updatedDashboardLayout[j].y += maxY;
if (updatedPanelMap[updatedDashboardLayout[j].i]) {
updatedPanelMap[updatedDashboardLayout[j].i].widgets = updatedPanelMap[
updatedDashboardLayout[j].i
].widgets.map((w) => ({
...w,
y: w.y + maxY,
}));
}
}
updatedDashboardLayout = [...updatedDashboardLayout, ...widgetsInsideTheRow];
} else {
rowProperties.collapsed = true;
const currentIdx = dashboardLayout.findIndex((w) => w.i === id);
let widgetsInsideTheRow: Layout[] = [];
let isPanelMapUpdated = false;
for (let j = currentIdx + 1; j < dashboardLayout.length; j++) {
if (currentPanelMap[dashboardLayout[j].i]) {
rowProperties.widgets = widgetsInsideTheRow;
widgetsInsideTheRow = [];
isPanelMapUpdated = true;
break;
} else {
widgetsInsideTheRow.push(dashboardLayout[j]);
}
}
if (!isPanelMapUpdated) {
rowProperties.widgets = widgetsInsideTheRow;
}
let maxY = 0;
widgetsInsideTheRow.forEach((w) => {
maxY = Math.max(maxY, w.y + w.h);
});
const currentRowWidget = dashboardLayout[currentIdx];
if (currentRowWidget && widgetsInsideTheRow.length) {
maxY -= currentRowWidget.h + currentRowWidget.y;
}
for (let j = currentIdx + 1; j < updatedDashboardLayout.length; j++) {
updatedDashboardLayout[j].y += maxY;
if (updatedPanelMap[updatedDashboardLayout[j].i]) {
updatedPanelMap[updatedDashboardLayout[j].i].widgets = updatedPanelMap[
updatedDashboardLayout[j].i
].widgets.map((w) => ({
...w,
y: w.y + maxY,
}));
}
}
updatedDashboardLayout = updatedDashboardLayout.filter(
(widget) => !rowProperties.widgets.some((w: Layout) => w.i === widget.i),
);
}
setCurrentPanelMap((prev) => ({
...prev,
...updatedPanelMap,
[id]: {
...rowProperties,
},
}));
setDashboardLayout(sortLayout(updatedDashboardLayout));
};
const handleDragStop: ItemCallback = (_, oldItem, newItem): void => {

View File

@@ -1,181 +0,0 @@
import { Layout } from 'react-grid-layout';
import { applyRowCollapse, PanelMap } from '../utils';
// Helper to produce deeply-frozen objects that mimic what zustand/immer returns.
function freeze<T>(obj: T): T {
return JSON.parse(JSON.stringify(obj), (_, v) =>
typeof v === 'object' && v !== null ? Object.freeze(v) : v,
) as T;
}
// ─── fixtures ────────────────────────────────────────────────────────────────
const ROW_ID = 'row1';
/** A layout with one row followed by two widgets. */
function makeLayout(): Layout[] {
return [
{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 },
{ i: 'w1', x: 0, y: 1, w: 6, h: 4 },
{ i: 'w2', x: 6, y: 1, w: 6, h: 4 },
];
}
/** panelMap where the row is expanded (collapsed = false, widgets = []). */
function makeExpandedPanelMap(): PanelMap {
return {
[ROW_ID]: { collapsed: false, widgets: [] },
};
}
/** panelMap where the row is collapsed (widgets stored inside). */
function makeCollapsedPanelMap(): PanelMap {
return {
[ROW_ID]: {
collapsed: true,
widgets: [
{ i: 'w1', x: 0, y: 1, w: 6, h: 4 },
{ i: 'w2', x: 6, y: 1, w: 6, h: 4 },
],
},
};
}
// ─── frozen-input guard (regression for zustand/immer read-only bug) ──────────
describe('applyRowCollapse does not mutate frozen inputs', () => {
it('does not throw when collapsing a row with frozen layout + panelMap', () => {
expect(() =>
applyRowCollapse(
ROW_ID,
freeze(makeLayout()),
freeze(makeExpandedPanelMap()),
),
).not.toThrow();
});
it('does not throw when expanding a row with frozen layout + panelMap', () => {
// Collapsed layout only has the row item; widgets live in panelMap.
const collapsedLayout = freeze([{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 }]);
expect(() =>
applyRowCollapse(ROW_ID, collapsedLayout, freeze(makeCollapsedPanelMap())),
).not.toThrow();
});
it('leaves the original layout array untouched after collapse', () => {
const layout = makeLayout();
const originalY = layout[1].y; // w1.y before collapse
applyRowCollapse(ROW_ID, layout, makeExpandedPanelMap());
expect(layout[1].y).toBe(originalY);
});
it('leaves the original panelMap untouched after collapse', () => {
const panelMap = makeExpandedPanelMap();
applyRowCollapse(ROW_ID, makeLayout(), panelMap);
expect(panelMap[ROW_ID].collapsed).toBe(false);
});
});
// ─── collapse behaviour ───────────────────────────────────────────────────────
describe('applyRowCollapse collapsing a row', () => {
it('sets collapsed = true on the row entry', () => {
const { updatedPanelMap } = applyRowCollapse(
ROW_ID,
makeLayout(),
makeExpandedPanelMap(),
);
expect(updatedPanelMap[ROW_ID].collapsed).toBe(true);
});
it('stores the child widgets inside the panelMap entry', () => {
const { updatedPanelMap } = applyRowCollapse(
ROW_ID,
makeLayout(),
makeExpandedPanelMap(),
);
const ids = updatedPanelMap[ROW_ID].widgets.map((w) => w.i);
expect(ids).toContain('w1');
expect(ids).toContain('w2');
});
it('removes child widgets from the returned layout', () => {
const { updatedLayout } = applyRowCollapse(
ROW_ID,
makeLayout(),
makeExpandedPanelMap(),
);
const ids = updatedLayout.map((l) => l.i);
expect(ids).not.toContain('w1');
expect(ids).not.toContain('w2');
expect(ids).toContain(ROW_ID);
});
});
// ─── expand behaviour ─────────────────────────────────────────────────────────
describe('applyRowCollapse expanding a row', () => {
it('sets collapsed = false on the row entry', () => {
const collapsedLayout: Layout[] = [{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 }];
const { updatedPanelMap } = applyRowCollapse(
ROW_ID,
collapsedLayout,
makeCollapsedPanelMap(),
);
expect(updatedPanelMap[ROW_ID].collapsed).toBe(false);
});
it('restores child widgets to the returned layout', () => {
const collapsedLayout: Layout[] = [{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 }];
const { updatedLayout } = applyRowCollapse(
ROW_ID,
collapsedLayout,
makeCollapsedPanelMap(),
);
const ids = updatedLayout.map((l) => l.i);
expect(ids).toContain('w1');
expect(ids).toContain('w2');
});
it('restored child widgets appear in both the layout and the panelMap entry', () => {
const collapsedLayout: Layout[] = [{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 }];
const { updatedLayout, updatedPanelMap } = applyRowCollapse(
ROW_ID,
collapsedLayout,
makeCollapsedPanelMap(),
);
// The previously-stored widgets should now be back in the live layout.
expect(updatedLayout.map((l) => l.i)).toContain('w1');
// The panelMap entry still holds a reference to them (stale until next collapse).
expect(updatedPanelMap[ROW_ID].widgets.map((w) => w.i)).toContain('w1');
});
});
// ─── y-offset adjustment ──────────────────────────────────────────────────────
describe('applyRowCollapse y-offset adjustments for rows below', () => {
it('shifts items below a second row down when the first row expands', () => {
const ROW2 = 'row2';
// Layout: row1 (y=0,h=1) | w1 (y=1,h=4) | row2 (y=5,h=1) | w3 (y=6,h=2)
const layout: Layout[] = [
{ i: ROW_ID, x: 0, y: 0, w: 12, h: 1 },
{ i: 'w1', x: 0, y: 1, w: 12, h: 4 },
{ i: ROW2, x: 0, y: 5, w: 12, h: 1 },
{ i: 'w3', x: 0, y: 6, w: 12, h: 2 },
];
const panelMap: PanelMap = {
[ROW_ID]: {
collapsed: true,
widgets: [{ i: 'w1', x: 0, y: 1, w: 12, h: 4 }],
},
[ROW2]: { collapsed: false, widgets: [] },
};
// Expanding row1 should push row2 and w3 down by the height of w1 (4).
const collapsedLayout = layout.filter((l) => l.i !== 'w1');
const { updatedLayout } = applyRowCollapse(ROW_ID, collapsedLayout, panelMap);
const row2Item = updatedLayout.find((l) => l.i === ROW2);
expect(row2Item?.y).toBe(5 + 4); // shifted by maxY = 4
});
});

View File

@@ -4,122 +4,6 @@ import { isEmpty, isEqual } from 'lodash-es';
import { Dashboard, Widgets } from 'types/api/dashboard/getAll';
import { IBuilderQuery, Query } from 'types/api/queryBuilder/queryBuilderData';
export type PanelMap = Record<
string,
{ widgets: Layout[]; collapsed: boolean }
>;
export interface RowCollapseResult {
updatedLayout: Layout[];
updatedPanelMap: PanelMap;
}
/**
* Pure function that computes the new layout and panelMap after toggling a
* row's collapsed state. All inputs are treated as immutable — no input object
* is mutated, so it is safe to pass frozen objects from the zustand store.
*/
// eslint-disable-next-line sonarjs/cognitive-complexity
export function applyRowCollapse(
id: string,
dashboardLayout: Layout[],
currentPanelMap: PanelMap,
): RowCollapseResult {
// Deep-copy the row's own properties so we can mutate our local copy.
const rowProperties = {
...currentPanelMap[id],
widgets: [...(currentPanelMap[id]?.widgets ?? [])],
};
// Shallow-copy each entry's widgets array so inner .map() calls are safe.
const updatedPanelMap: PanelMap = Object.fromEntries(
Object.entries(currentPanelMap).map(([k, v]) => [
k,
{ ...v, widgets: [...v.widgets] },
]),
);
let updatedDashboardLayout = [...dashboardLayout];
if (rowProperties.collapsed === true) {
// ── EXPAND ──────────────────────────────────────────────────────────────
rowProperties.collapsed = false;
const widgetsInsideTheRow = rowProperties.widgets;
let maxY = 0;
widgetsInsideTheRow.forEach((w) => {
maxY = Math.max(maxY, w.y + w.h);
});
const currentRowWidget = dashboardLayout.find((w) => w.i === id);
if (currentRowWidget && widgetsInsideTheRow.length) {
maxY -= currentRowWidget.h + currentRowWidget.y;
}
const idxCurrentRow = dashboardLayout.findIndex((w) => w.i === id);
for (let j = idxCurrentRow + 1; j < dashboardLayout.length; j++) {
updatedDashboardLayout[j] = {
...updatedDashboardLayout[j],
y: updatedDashboardLayout[j].y + maxY,
};
if (updatedPanelMap[updatedDashboardLayout[j].i]) {
updatedPanelMap[updatedDashboardLayout[j].i].widgets = updatedPanelMap[
updatedDashboardLayout[j].i
].widgets.map((w) => ({ ...w, y: w.y + maxY }));
}
}
updatedDashboardLayout = [...updatedDashboardLayout, ...widgetsInsideTheRow];
} else {
// ── COLLAPSE ─────────────────────────────────────────────────────────────
rowProperties.collapsed = true;
const currentIdx = dashboardLayout.findIndex((w) => w.i === id);
let widgetsInsideTheRow: Layout[] = [];
let isPanelMapUpdated = false;
for (let j = currentIdx + 1; j < dashboardLayout.length; j++) {
if (currentPanelMap[dashboardLayout[j].i]) {
rowProperties.widgets = widgetsInsideTheRow;
widgetsInsideTheRow = [];
isPanelMapUpdated = true;
break;
} else {
widgetsInsideTheRow.push(dashboardLayout[j]);
}
}
if (!isPanelMapUpdated) {
rowProperties.widgets = widgetsInsideTheRow;
}
let maxY = 0;
widgetsInsideTheRow.forEach((w) => {
maxY = Math.max(maxY, w.y + w.h);
});
const currentRowWidget = dashboardLayout[currentIdx];
if (currentRowWidget && widgetsInsideTheRow.length) {
maxY -= currentRowWidget.h + currentRowWidget.y;
}
for (let j = currentIdx + 1; j < updatedDashboardLayout.length; j++) {
updatedDashboardLayout[j] = {
...updatedDashboardLayout[j],
y: updatedDashboardLayout[j].y + maxY,
};
if (updatedPanelMap[updatedDashboardLayout[j].i]) {
updatedPanelMap[updatedDashboardLayout[j].i].widgets = updatedPanelMap[
updatedDashboardLayout[j].i
].widgets.map((w) => ({ ...w, y: w.y + maxY }));
}
}
updatedDashboardLayout = updatedDashboardLayout.filter(
(widget) => !rowProperties.widgets.some((w: Layout) => w.i === widget.i),
);
}
updatedPanelMap[id] = { ...rowProperties };
return { updatedLayout: updatedDashboardLayout, updatedPanelMap };
}
export const removeUndefinedValuesFromLayout = (layout: Layout[]): Layout[] =>
layout.map((obj) =>
Object.fromEntries(

View File

@@ -62,6 +62,9 @@ export const getVolumeQueryPayload = (
const k8sPVCNameKey = dotMetricsEnabled
? 'k8s.persistentvolumeclaim.name'
: 'k8s_persistentvolumeclaim_name';
const legendTemplate = dotMetricsEnabled
? '{{k8s.namespace.name}}-{{k8s.pod.name}}'
: '{{k8s_namespace_name}}-{{k8s_pod_name}}';
return [
{
@@ -133,7 +136,7 @@ export const getVolumeQueryPayload = (
functions: [],
groupBy: [],
having: [],
legend: 'Available',
legend: legendTemplate,
limit: null,
orderBy: [],
queryName: 'A',
@@ -225,7 +228,7 @@ export const getVolumeQueryPayload = (
functions: [],
groupBy: [],
having: [],
legend: 'Capacity',
legend: legendTemplate,
limit: null,
orderBy: [],
queryName: 'A',
@@ -316,7 +319,7 @@ export const getVolumeQueryPayload = (
},
groupBy: [],
having: [],
legend: 'Inodes Used',
legend: legendTemplate,
limit: null,
orderBy: [],
queryName: 'A',
@@ -408,7 +411,7 @@ export const getVolumeQueryPayload = (
},
groupBy: [],
having: [],
legend: 'Total Inodes',
legend: legendTemplate,
limit: null,
orderBy: [],
queryName: 'A',
@@ -500,7 +503,7 @@ export const getVolumeQueryPayload = (
},
groupBy: [],
having: [],
legend: 'Inodes Free',
legend: legendTemplate,
limit: null,
orderBy: [],
queryName: 'A',

View File

@@ -92,10 +92,7 @@ function LogsActionsContainer({
/>
</div>
<div className="download-options-container">
<DownloadOptionsMenu
dataSource={DataSource.LOGS}
selectedColumns={options?.selectColumns}
/>
<DownloadOptionsMenu dataSource={DataSource.LOGS} />
</div>
<div className="format-options-container">
<LogsFormatOptionsMenu

View File

@@ -42,15 +42,8 @@ function LogsPanelComponent({
setPageSize(value);
setOffset(0);
setRequestData((prev) => {
const newQueryData = {
...prev.query,
builder: {
...prev.query.builder,
queryData: prev.query.builder.queryData.map((qd, i) =>
i === 0 ? { ...qd, pageSize: value } : qd,
),
},
};
const newQueryData = { ...prev.query };
newQueryData.builder.queryData[0].pageSize = value;
return {
...prev,
query: newQueryData,

View File

@@ -6122,95 +6122,5 @@
],
"id": "huggingface-observability",
"link": "/docs/huggingface-observability/"
},
{
"dataSource": "mistral-observability",
"label": "Mistral AI",
"imgUrl": "/Logos/mistral.svg",
"tags": [
"LLM Monitoring"
],
"module": "apm",
"relatedSearchKeywords": [
"llm",
"llm monitoring",
"mistral",
"mistral ai",
"monitoring",
"observability",
"otel mistral",
"traces",
"tracing"
],
"id": "mistral-observability",
"link": "/docs/mistral-observability/"
},
{
"dataSource": "openclaw-observability",
"label": "OpenClaw",
"imgUrl": "/Logos/openclaw.svg",
"tags": [
"LLM Monitoring"
],
"module": "apm",
"relatedSearchKeywords": [
"llm",
"llm monitoring",
"monitoring",
"observability",
"openclaw",
"otel openclaw",
"traces",
"tracing"
],
"id": "openclaw-observability",
"link": "/docs/openclaw-monitoring/"
},
{
"dataSource": "claude-agent-monitoring",
"label": "Claude Agent SDK",
"imgUrl": "/Logos/claude-code.svg",
"tags": [
"LLM Monitoring"
],
"module": "apm",
"relatedSearchKeywords": [
"anthropic",
"claude",
"claude agent",
"claude agent sdk",
"claude sdk",
"llm",
"llm monitoring",
"monitoring",
"observability",
"otel claude",
"traces",
"tracing"
],
"id": "claude-agent-monitoring",
"link": "/docs/claude-agent-monitoring/"
},
{
"dataSource": "render-metrics",
"label": "Render",
"imgUrl": "/Logos/render.svg",
"tags": [
"infrastructure monitoring",
"metrics"
],
"module": "metrics",
"relatedSearchKeywords": [
"infrastructure",
"metrics",
"monitoring",
"observability",
"paas",
"render",
"render metrics",
"render monitoring"
],
"id": "render-metrics",
"link": "/docs/metrics-management/render-metrics/"
}
]

View File

@@ -42,19 +42,11 @@ function Panel({
};
}
updatedQuery.builder.queryData[0].pageSize = 10;
const initialDataSource = updatedQuery.builder.queryData[0].dataSource;
const updatedQueryForList = {
...updatedQuery,
builder: {
...updatedQuery.builder,
queryData: updatedQuery.builder.queryData.map((qd, i) =>
i === 0 ? { ...qd, pageSize: 10 } : qd,
),
},
};
return {
query: updatedQueryForList,
query: updatedQuery,
graphType: PANEL_TYPES.LIST,
selectedTime: widget.timePreferance || 'GLOBAL_TIME',
tableParams: {

View File

@@ -239,10 +239,7 @@ function ListView({
/>
</div>
<DownloadOptionsMenu
dataSource={DataSource.TRACES}
selectedColumns={options?.selectColumns}
/>
<DownloadOptionsMenu dataSource={DataSource.TRACES} />
<TraceExplorerControls
isLoading={isFetching}

View File

@@ -52,44 +52,37 @@ export const useGetQueryRange: UseGetQueryRange = (
!firstQueryData?.filters?.items.some((filter) => filter.key?.key === 'id') &&
firstQueryData?.orderBy[0].columnName === 'timestamp';
if (
isListWithSingleTimestampOrder &&
firstQueryData?.dataSource === DataSource.LOGS
) {
return {
...requestData,
graphType:
requestData.graphType === PANEL_TYPES.BAR
? PANEL_TYPES.TIME_SERIES
: requestData.graphType,
query: {
...requestData.query,
builder: {
...requestData.query.builder,
queryData: [
{
...firstQueryData,
orderBy: [
...(firstQueryData?.orderBy || []),
{
columnName: 'id',
order: firstQueryData?.orderBy[0]?.order,
},
],
},
],
},
},
};
}
return {
const modifiedRequestData = {
...requestData,
graphType:
requestData.graphType === PANEL_TYPES.BAR
? PANEL_TYPES.TIME_SERIES
: requestData.graphType,
};
// If the query is a list with a single timestamp order, we need to add the id column to the order by clause
if (
isListWithSingleTimestampOrder &&
firstQueryData?.dataSource === DataSource.LOGS
) {
modifiedRequestData.query.builder = {
...requestData.query.builder,
queryData: [
{
...firstQueryData,
orderBy: [
...(firstQueryData?.orderBy || []),
{
columnName: 'id',
order: firstQueryData?.orderBy[0]?.order,
},
],
},
],
};
}
return modifiedRequestData;
}, [requestData]);
const queryKey = useMemo(() => {

View File

@@ -3,7 +3,7 @@ import { useCallback, useState } from 'react';
import { useSelector } from 'react-redux';
import { message } from 'antd';
import { downloadExportData } from 'api/v1/download/downloadExportData';
import { prepareQueryRangePayloadV5, TelemetryFieldKey } from 'api/v5/v5';
import { prepareQueryRangePayloadV5 } from 'api/v5/v5';
import { PANEL_TYPES } from 'constants/queryBuilder';
import { useQueryBuilder } from 'hooks/queryBuilder/useQueryBuilder';
import { AppState } from 'store/reducers';
@@ -14,7 +14,6 @@ interface ExportOptions {
format: string;
rowLimit: number;
clearSelectColumns: boolean;
selectedColumns?: TelemetryFieldKey[];
}
interface UseExportRawDataProps {
@@ -43,7 +42,6 @@ export function useExportRawData({
format,
rowLimit,
clearSelectColumns,
selectedColumns,
}: ExportOptions): Promise<void> => {
if (!stagedQuery) {
return;
@@ -52,12 +50,6 @@ export function useExportRawData({
try {
setIsDownloading(true);
const selectColumnsOverride = clearSelectColumns
? {}
: selectedColumns?.length
? { selectColumns: selectedColumns }
: {};
const exportQuery = {
...stagedQuery,
builder: {
@@ -67,7 +59,7 @@ export function useExportRawData({
groupBy: [],
having: { expression: '' },
limit: rowLimit,
...selectColumnsOverride,
...(clearSelectColumns && { selectColumns: [] }),
})),
queryTraceOperator: (stagedQuery.builder.queryTraceOperator || []).map(
(traceOp) => ({
@@ -75,7 +67,7 @@ export function useExportRawData({
groupBy: [],
having: { expression: '' },
limit: rowLimit,
...selectColumnsOverride,
...(clearSelectColumns && { selectColumns: [] }),
}),
),
},

4
go.mod
View File

@@ -81,8 +81,6 @@ require (
golang.org/x/oauth2 v0.34.0
golang.org/x/sync v0.19.0
golang.org/x/text v0.33.0
gonum.org/v1/gonum v0.17.0
google.golang.org/api v0.265.0
google.golang.org/protobuf v1.36.11
gopkg.in/yaml.v2 v2.4.0
gopkg.in/yaml.v3 v3.0.1
@@ -379,6 +377,8 @@ require (
golang.org/x/sys v0.40.0 // indirect
golang.org/x/time v0.14.0 // indirect
golang.org/x/tools v0.41.0 // indirect
gonum.org/v1/gonum v0.17.0 // indirect
google.golang.org/api v0.265.0
google.golang.org/genproto/googleapis/api v0.0.0-20260203192932-546029d2fa20 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20260128011058-8636f8732409 // indirect
google.golang.org/grpc v1.78.0 // indirect

View File

@@ -1,216 +0,0 @@
package signozapiserver
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/types"
citypes "github.com/SigNoz/signoz/pkg/types/cloudintegrationtypes"
"github.com/gorilla/mux"
)
func (provider *provider) addCloudIntegrationRoutes(router *mux.Router) error {
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.CreateAccount),
handler.OpenAPIDef{
ID: "CreateAccount",
Tags: []string{"cloudintegration"},
Summary: "Create account",
Description: "This endpoint creates a new cloud integration account for the specified cloud provider",
Request: new(citypes.PostableConnectionArtifact),
RequestContentType: "application/json",
Response: new(citypes.GettableAccountWithArtifact),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodPost).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.ListAccounts),
handler.OpenAPIDef{
ID: "ListAccounts",
Tags: []string{"cloudintegration"},
Summary: "List accounts",
Description: "This endpoint lists the accounts for the specified cloud provider",
Request: nil,
RequestContentType: "",
Response: new(citypes.GettableAccounts),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.GetAccount),
handler.OpenAPIDef{
ID: "GetAccount",
Tags: []string{"cloudintegration"},
Summary: "Get account",
Description: "This endpoint gets an account for the specified cloud provider",
Request: nil,
RequestContentType: "",
Response: new(citypes.GettableAccount),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.UpdateAccount),
handler.OpenAPIDef{
ID: "UpdateAccount",
Tags: []string{"cloudintegration"},
Summary: "Update account",
Description: "This endpoint updates an account for the specified cloud provider",
Request: new(citypes.UpdatableAccount),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodPut).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/{id}", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.DisconnectAccount),
handler.OpenAPIDef{
ID: "DisconnectAccount",
Tags: []string{"cloudintegration"},
Summary: "Disconnect account",
Description: "This endpoint disconnects an account for the specified cloud provider",
Request: nil,
RequestContentType: "",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodDelete).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/services", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.ListServicesMetadata),
handler.OpenAPIDef{
ID: "ListServicesMetadata",
Tags: []string{"cloudintegration"},
Summary: "List services metadata",
Description: "This endpoint lists the services metadata for the specified cloud provider",
Request: nil,
RequestContentType: "",
Response: new(citypes.GettableServicesMetadata),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/services/{service_id}", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.GetService),
handler.OpenAPIDef{
ID: "GetService",
Tags: []string{"cloudintegration"},
Summary: "Get service",
Description: "This endpoint gets a service for the specified cloud provider",
Request: nil,
RequestContentType: "",
Response: new(citypes.GettableService),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/services/{service_id}", handler.New(
provider.authZ.AdminAccess(provider.cloudIntegrationHandler.UpdateService),
handler.OpenAPIDef{
ID: "UpdateService",
Tags: []string{"cloudintegration"},
Summary: "Update service",
Description: "This endpoint updates a service for the specified cloud provider",
Request: new(citypes.UpdatableService),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
},
)).Methods(http.MethodPut).GetError(); err != nil {
return err
}
// Agent check-in endpoint is kept same as older one to maintain backward compatibility with already deployed agents.
// In the future, this endpoint will be deprecated and a new endpoint will be introduced for consistency with above endpoints.
if err := router.Handle("/api/v1/cloud-integrations/{cloud_provider}/agent-check-in", handler.New(
provider.authZ.ViewAccess(provider.cloudIntegrationHandler.AgentCheckIn),
handler.OpenAPIDef{
ID: "AgentCheckInDeprecated",
Tags: []string{"cloudintegration"},
Summary: "Agent check-in",
Description: "[Deprecated] This endpoint is called by the deployed agent to check in",
Request: new(citypes.PostableAgentCheckInRequest),
RequestContentType: "application/json",
Response: new(citypes.GettableAgentCheckInResponse),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: true, // this endpoint will be deprecated in future
SecuritySchemes: newSecuritySchemes(types.RoleViewer), // agent role is viewer
},
)).Methods(http.MethodPost).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/cloud_integrations/{cloud_provider}/accounts/check_in", handler.New(
provider.authZ.ViewAccess(provider.cloudIntegrationHandler.AgentCheckIn),
handler.OpenAPIDef{
ID: "AgentCheckIn",
Tags: []string{"cloudintegration"},
Summary: "Agent check-in",
Description: "This endpoint is called by the deployed agent to check in",
Request: new(citypes.PostableAgentCheckInRequest),
RequestContentType: "application/json",
Response: new(citypes.GettableAgentCheckInResponse),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleViewer), // agent role is viewer
},
)).Methods(http.MethodPost).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -12,14 +12,12 @@ import (
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/http/middleware"
"github.com/SigNoz/signoz/pkg/modules/authdomain"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
"github.com/SigNoz/signoz/pkg/modules/dashboard"
"github.com/SigNoz/signoz/pkg/modules/fields"
"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
"github.com/SigNoz/signoz/pkg/modules/organization"
"github.com/SigNoz/signoz/pkg/modules/preference"
"github.com/SigNoz/signoz/pkg/modules/promote"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
"github.com/SigNoz/signoz/pkg/modules/session"
"github.com/SigNoz/signoz/pkg/modules/user"
@@ -31,30 +29,27 @@ import (
)
type provider struct {
config apiserver.Config
settings factory.ScopedProviderSettings
router *mux.Router
authZ *middleware.AuthZ
orgHandler organization.Handler
userHandler user.Handler
sessionHandler session.Handler
authDomainHandler authdomain.Handler
preferenceHandler preference.Handler
globalHandler global.Handler
promoteHandler promote.Handler
flaggerHandler flagger.Handler
dashboardModule dashboard.Module
dashboardHandler dashboard.Handler
metricsExplorerHandler metricsexplorer.Handler
gatewayHandler gateway.Handler
fieldsHandler fields.Handler
authzHandler authz.Handler
rawDataExportHandler rawdataexport.Handler
zeusHandler zeus.Handler
querierHandler querier.Handler
serviceAccountHandler serviceaccount.Handler
factoryHandler factory.Handler
cloudIntegrationHandler cloudintegration.Handler
config apiserver.Config
settings factory.ScopedProviderSettings
router *mux.Router
authZ *middleware.AuthZ
orgHandler organization.Handler
userHandler user.Handler
sessionHandler session.Handler
authDomainHandler authdomain.Handler
preferenceHandler preference.Handler
globalHandler global.Handler
promoteHandler promote.Handler
flaggerHandler flagger.Handler
dashboardModule dashboard.Module
dashboardHandler dashboard.Handler
metricsExplorerHandler metricsexplorer.Handler
gatewayHandler gateway.Handler
fieldsHandler fields.Handler
authzHandler authz.Handler
zeusHandler zeus.Handler
querierHandler querier.Handler
serviceAccountHandler serviceaccount.Handler
}
func NewFactory(
@@ -74,12 +69,9 @@ func NewFactory(
gatewayHandler gateway.Handler,
fieldsHandler fields.Handler,
authzHandler authz.Handler,
rawDataExportHandler rawdataexport.Handler,
zeusHandler zeus.Handler,
querierHandler querier.Handler,
serviceAccountHandler serviceaccount.Handler,
factoryHandler factory.Handler,
cloudIntegrationHandler cloudintegration.Handler,
) factory.ProviderFactory[apiserver.APIServer, apiserver.Config] {
return factory.NewProviderFactory(factory.MustNewName("signoz"), func(ctx context.Context, providerSettings factory.ProviderSettings, config apiserver.Config) (apiserver.APIServer, error) {
return newProvider(
@@ -102,12 +94,9 @@ func NewFactory(
gatewayHandler,
fieldsHandler,
authzHandler,
rawDataExportHandler,
zeusHandler,
querierHandler,
serviceAccountHandler,
factoryHandler,
cloudIntegrationHandler,
)
})
}
@@ -132,40 +121,34 @@ func newProvider(
gatewayHandler gateway.Handler,
fieldsHandler fields.Handler,
authzHandler authz.Handler,
rawDataExportHandler rawdataexport.Handler,
zeusHandler zeus.Handler,
querierHandler querier.Handler,
serviceAccountHandler serviceaccount.Handler,
factoryHandler factory.Handler,
cloudIntegrationHandler cloudintegration.Handler,
) (apiserver.APIServer, error) {
settings := factory.NewScopedProviderSettings(providerSettings, "github.com/SigNoz/signoz/pkg/apiserver/signozapiserver")
router := mux.NewRouter().UseEncodedPath()
provider := &provider{
config: config,
settings: settings,
router: router,
orgHandler: orgHandler,
userHandler: userHandler,
sessionHandler: sessionHandler,
authDomainHandler: authDomainHandler,
preferenceHandler: preferenceHandler,
globalHandler: globalHandler,
promoteHandler: promoteHandler,
flaggerHandler: flaggerHandler,
dashboardModule: dashboardModule,
dashboardHandler: dashboardHandler,
metricsExplorerHandler: metricsExplorerHandler,
gatewayHandler: gatewayHandler,
fieldsHandler: fieldsHandler,
authzHandler: authzHandler,
rawDataExportHandler: rawDataExportHandler,
zeusHandler: zeusHandler,
querierHandler: querierHandler,
serviceAccountHandler: serviceAccountHandler,
factoryHandler: factoryHandler,
cloudIntegrationHandler: cloudIntegrationHandler,
config: config,
settings: settings,
router: router,
orgHandler: orgHandler,
userHandler: userHandler,
sessionHandler: sessionHandler,
authDomainHandler: authDomainHandler,
preferenceHandler: preferenceHandler,
globalHandler: globalHandler,
promoteHandler: promoteHandler,
flaggerHandler: flaggerHandler,
dashboardModule: dashboardModule,
dashboardHandler: dashboardHandler,
metricsExplorerHandler: metricsExplorerHandler,
gatewayHandler: gatewayHandler,
fieldsHandler: fieldsHandler,
authzHandler: authzHandler,
zeusHandler: zeusHandler,
querierHandler: querierHandler,
serviceAccountHandler: serviceAccountHandler,
}
provider.authZ = middleware.NewAuthZ(settings.Logger(), orgGetter, authz)
@@ -238,10 +221,6 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
if err := provider.addRawDataExportRoutes(router); err != nil {
return err
}
if err := provider.addZeusRoutes(router); err != nil {
return err
}
@@ -254,14 +233,6 @@ func (provider *provider) AddToRouter(router *mux.Router) error {
return err
}
if err := provider.addRegistryRoutes(router); err != nil {
return err
}
if err := provider.addCloudIntegrationRoutes(router); err != nil {
return err
}
return nil
}

View File

@@ -1,33 +0,0 @@
package signozapiserver
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/types"
"github.com/SigNoz/signoz/pkg/types/exporttypes"
v5 "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/gorilla/mux"
)
func (provider *provider) addRawDataExportRoutes(router *mux.Router) error {
if err := router.Handle("/api/v1/export_raw_data", handler.New(provider.authZ.ViewAccess(provider.rawDataExportHandler.ExportRawData), handler.OpenAPIDef{
ID: "HandleExportRawDataPOST",
Tags: []string{"logs", "traces"},
Summary: "Export raw data",
Description: "This endpoints allows complex query exporting raw data for traces and logs",
Request: new(v5.QueryRangeRequest),
RequestQuery: new(exporttypes.ExportRawDataFormatQueryParam),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest},
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
})).Methods(http.MethodPost).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -1,84 +0,0 @@
package signozapiserver
import (
"net/http"
"github.com/SigNoz/signoz/pkg/factory"
pkghandler "github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/gorilla/mux"
openapi "github.com/swaggest/openapi-go"
)
type healthOpenAPIHandler struct {
handlerFunc http.HandlerFunc
id string
summary string
}
func newHealthOpenAPIHandler(handlerFunc http.HandlerFunc, id, summary string) pkghandler.Handler {
return &healthOpenAPIHandler{
handlerFunc: handlerFunc,
id: id,
summary: summary,
}
}
func (handler *healthOpenAPIHandler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
handler.handlerFunc.ServeHTTP(rw, req)
}
func (handler *healthOpenAPIHandler) ServeOpenAPI(opCtx openapi.OperationContext) {
opCtx.SetID(handler.id)
opCtx.SetTags("health")
opCtx.SetSummary(handler.summary)
response := render.SuccessResponse{
Status: render.StatusSuccess.String(),
Data: new(factory.Response),
}
opCtx.AddRespStructure(
response,
openapi.WithContentType("application/json"),
openapi.WithHTTPStatus(http.StatusOK),
)
opCtx.AddRespStructure(
response,
openapi.WithContentType("application/json"),
openapi.WithHTTPStatus(http.StatusServiceUnavailable),
)
}
func (provider *provider) addRegistryRoutes(router *mux.Router) error {
if err := router.Handle("/api/v2/healthz", newHealthOpenAPIHandler(
provider.authZ.OpenAccess(provider.factoryHandler.Healthz),
"Healthz",
"Health check",
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/readyz", newHealthOpenAPIHandler(
provider.authZ.OpenAccess(provider.factoryHandler.Readyz),
"Readyz",
"Readiness check",
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/livez", pkghandler.New(provider.authZ.OpenAccess(provider.factoryHandler.Livez),
pkghandler.OpenAPIDef{
ID: "Livez",
Tags: []string{"health"},
Summary: "Liveness check",
Response: new(factory.Response),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
},
)).Methods(http.MethodGet).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -111,8 +111,8 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v1/user", handler.New(provider.authZ.AdminAccess(provider.userHandler.ListUsersDeprecated), handler.OpenAPIDef{
ID: "ListUsersDeprecated",
if err := router.Handle("/api/v1/user", handler.New(provider.authZ.AdminAccess(provider.userHandler.ListUsers), handler.OpenAPIDef{
ID: "ListUsers",
Tags: []string{"users"},
Summary: "List users",
Description: "This endpoint lists all users",
@@ -128,25 +128,8 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/users", handler.New(provider.authZ.AdminAccess(provider.userHandler.ListUsers), handler.OpenAPIDef{
ID: "ListUsers",
Tags: []string{"users"},
Summary: "List users v2",
Description: "This endpoint lists all users for the organization",
Request: nil,
RequestContentType: "",
Response: make([]*types.User, 0),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/user/me", handler.New(provider.authZ.OpenAccess(provider.userHandler.GetMyUserDeprecated), handler.OpenAPIDef{
ID: "GetMyUserDeprecated",
if err := router.Handle("/api/v1/user/me", handler.New(provider.authZ.OpenAccess(provider.userHandler.GetMyUser), handler.OpenAPIDef{
ID: "GetMyUser",
Tags: []string{"users"},
Summary: "Get my user",
Description: "This endpoint returns the user I belong to",
@@ -162,42 +145,8 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/users/me", handler.New(provider.authZ.OpenAccess(provider.userHandler.GetMyUser), handler.OpenAPIDef{
ID: "GetMyUser",
Tags: []string{"users"},
Summary: "Get my user v2",
Description: "This endpoint returns the user I belong to",
Request: nil,
RequestContentType: "",
Response: new(authtypes.UserWithRoles),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: []handler.OpenAPISecurityScheme{{Name: authtypes.IdentNProviderTokenizer.StringValue()}},
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/users/me", handler.New(provider.authZ.OpenAccess(provider.userHandler.UpdateMyUser), handler.OpenAPIDef{
ID: "UpdateMyUserV2",
Tags: []string{"users"},
Summary: "Update my user v2",
Description: "This endpoint updates the user I belong to",
Request: new(types.UpdatableUser),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{},
Deprecated: false,
SecuritySchemes: []handler.OpenAPISecurityScheme{{Name: authtypes.IdentNProviderTokenizer.StringValue()}},
})).Methods(http.MethodPut).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/user/{id}", handler.New(provider.authZ.SelfAccess(provider.userHandler.GetUserDeprecated), handler.OpenAPIDef{
ID: "GetUserDeprecated",
if err := router.Handle("/api/v1/user/{id}", handler.New(provider.authZ.SelfAccess(provider.userHandler.GetUser), handler.OpenAPIDef{
ID: "GetUser",
Tags: []string{"users"},
Summary: "Get user",
Description: "This endpoint returns the user by id",
@@ -213,25 +162,8 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/users/{id}", handler.New(provider.authZ.AdminAccess(provider.userHandler.GetUser), handler.OpenAPIDef{
ID: "GetUser",
Tags: []string{"users"},
Summary: "Get user by user id",
Description: "This endpoint returns the user by id",
Request: nil,
RequestContentType: "",
Response: new(authtypes.UserWithRoles),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/user/{id}", handler.New(provider.authZ.SelfAccess(provider.userHandler.UpdateUserDeprecated), handler.OpenAPIDef{
ID: "UpdateUserDeprecated",
if err := router.Handle("/api/v1/user/{id}", handler.New(provider.authZ.SelfAccess(provider.userHandler.UpdateUser), handler.OpenAPIDef{
ID: "UpdateUser",
Tags: []string{"users"},
Summary: "Update user",
Description: "This endpoint updates the user by id",
@@ -247,23 +179,6 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/users/{id}", handler.New(provider.authZ.AdminAccess(provider.userHandler.UpdateUser), handler.OpenAPIDef{
ID: "UpdateUser",
Tags: []string{"users"},
Summary: "Update user v2",
Description: "This endpoint updates the user by id",
Request: new(types.UpdatableUser),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodPut).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v1/user/{id}", handler.New(provider.authZ.AdminAccess(provider.userHandler.DeleteUser), handler.OpenAPIDef{
ID: "DeleteUser",
Tags: []string{"users"},
@@ -349,73 +264,5 @@ func (provider *provider) addUserRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/users/{id}/roles", handler.New(provider.authZ.AdminAccess(provider.userHandler.GetRolesByUserID), handler.OpenAPIDef{
ID: "GetRolesByUserID",
Tags: []string{"users"},
Summary: "Get user roles",
Description: "This endpoint returns the user roles by user id",
Request: nil,
RequestContentType: "",
Response: make([]*authtypes.Role, 0),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/users/{id}/roles", handler.New(provider.authZ.AdminAccess(provider.userHandler.SetRoleByUserID), handler.OpenAPIDef{
ID: "SetRoleByUserID",
Tags: []string{"users"},
Summary: "Set user roles",
Description: "This endpoint assigns the role to the user roles by user id",
Request: new(types.PostableRole),
RequestContentType: "application/json",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodPost).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/users/{id}/roles/{roleId}", handler.New(provider.authZ.AdminAccess(provider.userHandler.RemoveUserRoleByRoleID), handler.OpenAPIDef{
ID: "RemoveUserRoleByUserIDAndRoleID",
Tags: []string{"users"},
Summary: "Remove a role from user",
Description: "This endpoint removes a role from the user by user id and role id",
Request: nil,
RequestContentType: "",
Response: nil,
ResponseContentType: "",
SuccessStatusCode: http.StatusNoContent,
ErrorStatusCodes: []int{http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodDelete).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/roles/{id}/users", handler.New(provider.authZ.AdminAccess(provider.userHandler.GetUsersByRoleID), handler.OpenAPIDef{
ID: "GetUsersByRoleID",
Tags: []string{"users"},
Summary: "Get users by role id",
Description: "This endpoint returns the users having the role by role id",
Request: nil,
RequestContentType: "",
Response: make([]*types.User, 0),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusNotFound},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleAdmin),
})).Methods(http.MethodGet).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -11,7 +11,7 @@ import (
)
type AuthZ interface {
factory.ServiceWithHealthy
factory.Service
// CheckWithTupleCreation takes upon the responsibility for generating the tuples alongside everything Check does.
CheckWithTupleCreation(context.Context, authtypes.Claims, valuer.UUID, authtypes.Relation, authtypes.Typeable, []authtypes.Selector, []authtypes.Selector) error

View File

@@ -43,10 +43,6 @@ func (provider *provider) Start(ctx context.Context) error {
return provider.server.Start(ctx)
}
func (provider *provider) Healthy() <-chan struct{} {
return provider.server.Healthy()
}
func (provider *provider) Stop(ctx context.Context) error {
return provider.server.Stop(ctx)
}

View File

@@ -31,7 +31,6 @@ type Server struct {
modelID string
mtx sync.RWMutex
stopChan chan struct{}
healthyC chan struct{}
}
func NewOpenfgaServer(ctx context.Context, settings factory.ProviderSettings, config authz.Config, sqlstore sqlstore.SQLStore, openfgaSchema []openfgapkgtransformer.ModuleFile) (*Server, error) {
@@ -62,7 +61,6 @@ func NewOpenfgaServer(ctx context.Context, settings factory.ProviderSettings, co
openfgaSchema: openfgaSchema,
mtx: sync.RWMutex{},
stopChan: make(chan struct{}),
healthyC: make(chan struct{}),
}, nil
}
@@ -82,16 +80,10 @@ func (server *Server) Start(ctx context.Context) error {
server.storeID = storeID
server.mtx.Unlock()
close(server.healthyC)
<-server.stopChan
return nil
}
func (server *Server) Healthy() <-chan struct{} {
return server.healthyC
}
func (server *Server) Stop(ctx context.Context) error {
server.openfgaServer.Close()
close(server.stopChan)

View File

@@ -1,67 +0,0 @@
package factory
import (
"net/http"
"github.com/SigNoz/signoz/pkg/http/render"
)
// Handler provides HTTP handler functions for service health checks.
type Handler interface {
// Readyz reports whether services are ready.
Readyz(http.ResponseWriter, *http.Request)
// Livez reports whether services are alive.
Livez(http.ResponseWriter, *http.Request)
// Healthz reports overall service health.
Healthz(http.ResponseWriter, *http.Request)
}
type handler struct {
registry *Registry
}
func NewHandler(registry *Registry) Handler {
return &handler{
registry: registry,
}
}
type Response struct {
Healthy bool `json:"healthy"`
Services map[State][]Name `json:"services"`
}
func (handler *handler) Healthz(rw http.ResponseWriter, req *http.Request) {
byState := handler.registry.ServicesByState()
healthy := handler.registry.IsHealthy()
statusCode := http.StatusOK
if !healthy {
statusCode = http.StatusServiceUnavailable
}
render.Success(rw, statusCode, Response{
Healthy: healthy,
Services: byState,
})
}
func (handler *handler) Readyz(rw http.ResponseWriter, req *http.Request) {
healthy := handler.registry.IsHealthy()
statusCode := http.StatusOK
if !healthy {
statusCode = http.StatusServiceUnavailable
}
render.Success(rw, statusCode, Response{
Healthy: healthy,
Services: handler.registry.ServicesByState(),
})
}
func (handler *handler) Livez(rw http.ResponseWriter, req *http.Request) {
render.Success(rw, http.StatusOK, nil)
}

View File

@@ -5,11 +5,9 @@ import (
"regexp"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/swaggest/jsonschema-go"
)
var _ slog.LogValuer = (Name{})
var _ jsonschema.Exposer = (Name{})
var (
// nameRegex is a regex that matches a valid name.
@@ -29,21 +27,6 @@ func (n Name) String() string {
return n.name
}
// MarshalText implements encoding.TextMarshaler for JSON serialization.
func (n Name) MarshalText() ([]byte, error) {
return []byte(n.name), nil
}
// MarshalJSON implements json.Marshaler so Name serializes as a JSON string.
func (n Name) MarshalJSON() ([]byte, error) {
return []byte(`"` + n.name + `"`), nil
}
// JSONSchema implements jsonschema.Exposer so OpenAPI reflects Name as a string.
func (n Name) JSONSchema() (jsonschema.Schema, error) {
return *new(jsonschema.Schema).WithType(jsonschema.String.Type()), nil
}
// NewName creates a new name.
func NewName(name string) (Name, error) {
if !nameRegex.MatchString(name) {

View File

@@ -8,26 +8,21 @@ import (
"syscall"
"github.com/SigNoz/signoz/pkg/errors"
"gonum.org/v1/gonum/graph/simple"
"gonum.org/v1/gonum/graph/topo"
)
var (
ErrCodeInvalidRegistry = errors.MustNewCode("invalid_registry")
ErrCodeDependencyFailed = errors.MustNewCode("dependency_failed")
ErrCodeServiceFailed = errors.MustNewCode("service_failed")
ErrCodeInvalidRegistry = errors.MustNewCode("invalid_registry")
)
type Registry struct {
services []*serviceWithState
servicesByName map[Name]*serviceWithState
logger *slog.Logger
startC chan error
stopC chan error
services NamedMap[NamedService]
logger *slog.Logger
startCh chan error
stopCh chan error
}
// New creates a new registry of services. It needs at least one service in the input.
func NewRegistry(ctx context.Context, logger *slog.Logger, services ...NamedService) (*Registry, error) {
func NewRegistry(logger *slog.Logger, services ...NamedService) (*Registry, error) {
if logger == nil {
return nil, errors.Newf(errors.TypeInvalidInput, ErrCodeInvalidRegistry, "cannot build registry, logger is required")
}
@@ -36,131 +31,59 @@ func NewRegistry(ctx context.Context, logger *slog.Logger, services ...NamedServ
return nil, errors.Newf(errors.TypeInvalidInput, ErrCodeInvalidRegistry, "cannot build registry, at least one service is required")
}
servicesWithState := make([]*serviceWithState, len(services))
servicesByName := make(map[Name]*serviceWithState, len(services))
for i, s := range services {
if _, ok := servicesByName[s.Name()]; ok {
return nil, errors.Newf(errors.TypeInvalidInput, ErrCodeInvalidRegistry, "cannot build registry, duplicate service name %q", s.Name())
}
ss := newServiceWithState(s)
servicesWithState[i] = ss
servicesByName[s.Name()] = ss
}
registryLogger := logger.With(slog.String("pkg", "github.com/SigNoz/signoz/pkg/factory"))
for _, ss := range servicesWithState {
for _, dep := range ss.service.DependsOn() {
if dep == ss.service.Name() {
registryLogger.ErrorContext(ctx, "ignoring self-dependency", slog.Any("service", ss.service.Name()))
continue
}
if _, ok := servicesByName[dep]; !ok {
registryLogger.ErrorContext(ctx, "ignoring unknown dependency", slog.Any("service", ss.service.Name()), slog.Any("dependency", dep))
continue
}
ss.dependsOn = append(ss.dependsOn, dep)
}
}
if err := detectCyclicDeps(servicesWithState); err != nil {
m, err := NewNamedMap(services...)
if err != nil {
return nil, err
}
return &Registry{
logger: registryLogger,
services: servicesWithState,
servicesByName: servicesByName,
startC: make(chan error, 1),
stopC: make(chan error, len(services)),
logger: logger.With(slog.String("pkg", "go.signoz.io/pkg/factory")),
services: m,
startCh: make(chan error, 1),
stopCh: make(chan error, len(services)),
}, nil
}
func (registry *Registry) Start(ctx context.Context) {
for _, ss := range registry.services {
go func(ss *serviceWithState) {
// Wait for all dependencies to be healthy before starting.
for _, dep := range ss.dependsOn {
depState := registry.servicesByName[dep]
registry.logger.InfoContext(ctx, "service waiting for dependency", slog.Any("service", ss.service.Name()), slog.Any("dependency", dep))
select {
case <-ctx.Done():
ss.mu.Lock()
ss.state = StateFailed
ss.startErr = ctx.Err()
ss.mu.Unlock()
close(ss.startReturnedC)
registry.startC <- ctx.Err()
return
case <-depState.healthyC:
// Dependency is healthy, continue.
case <-depState.startReturnedC:
// Dependency failed before becoming healthy.
err := errors.Newf(errors.TypeInternal, ErrCodeDependencyFailed, "dependency %q of service %q failed", dep, ss.service.Name())
ss.mu.Lock()
ss.state = StateFailed
ss.startErr = err
ss.mu.Unlock()
close(ss.startReturnedC)
registry.startC <- err
return
}
}
registry.logger.InfoContext(ctx, "starting service", slog.Any("service", ss.service.Name()))
go func() {
select {
case <-ss.service.Healthy():
ss.setState(StateRunning)
case <-ss.startReturnedC:
}
}()
err := ss.service.Start(ctx)
if err != nil {
ss.mu.Lock()
ss.state = StateFailed
ss.startErr = err
ss.mu.Unlock()
}
close(ss.startReturnedC)
registry.startC <- err
}(ss)
func (r *Registry) Start(ctx context.Context) {
for _, s := range r.services.GetInOrder() {
go func(s NamedService) {
r.logger.InfoContext(ctx, "starting service", slog.Any("service", s.Name()))
err := s.Start(ctx)
r.startCh <- err
}(s)
}
}
func (registry *Registry) Wait(ctx context.Context) error {
func (r *Registry) Wait(ctx context.Context) error {
interrupt := make(chan os.Signal, 1)
signal.Notify(interrupt, syscall.SIGINT, syscall.SIGTERM)
select {
case <-ctx.Done():
registry.logger.InfoContext(ctx, "caught context error, exiting", errors.Attr(ctx.Err()))
r.logger.InfoContext(ctx, "caught context error, exiting", errors.Attr(ctx.Err()))
case s := <-interrupt:
registry.logger.InfoContext(ctx, "caught interrupt signal, exiting", slog.Any("signal", s))
case err := <-registry.startC:
registry.logger.ErrorContext(ctx, "caught service error, exiting", errors.Attr(err))
r.logger.InfoContext(ctx, "caught interrupt signal, exiting", slog.Any("signal", s))
case err := <-r.startCh:
r.logger.ErrorContext(ctx, "caught service error, exiting", errors.Attr(err))
return err
}
return nil
}
func (registry *Registry) Stop(ctx context.Context) error {
for _, ss := range registry.services {
go func(ss *serviceWithState) {
registry.logger.InfoContext(ctx, "stopping service", slog.Any("service", ss.service.Name()))
err := ss.service.Stop(ctx)
registry.stopC <- err
}(ss)
func (r *Registry) Stop(ctx context.Context) error {
for _, s := range r.services.GetInOrder() {
go func(s NamedService) {
r.logger.InfoContext(ctx, "stopping service", slog.Any("service", s.Name()))
err := s.Stop(ctx)
r.stopCh <- err
}(s)
}
errs := make([]error, len(registry.services))
for i := 0; i < len(registry.services); i++ {
err := <-registry.stopC
errs := make([]error, len(r.services.GetInOrder()))
for i := 0; i < len(r.services.GetInOrder()); i++ {
err := <-r.stopCh
if err != nil {
errs = append(errs, err)
}
@@ -168,83 +91,3 @@ func (registry *Registry) Stop(ctx context.Context) error {
return errors.Join(errs...)
}
// AwaitHealthy blocks until all services reach the RUNNING state or any service fails.
func (registry *Registry) AwaitHealthy(ctx context.Context) error {
for _, ss := range registry.services {
select {
case <-ctx.Done():
return ctx.Err()
case <-ss.healthyC:
case <-ss.startReturnedC:
ss.mu.RLock()
err := ss.startErr
ss.mu.RUnlock()
if err != nil {
return errors.Wrapf(err, errors.TypeInternal, ErrCodeServiceFailed, "service %q failed before becoming healthy", ss.service.Name())
}
return errors.Newf(errors.TypeInternal, ErrCodeServiceFailed, "service %q terminated before becoming healthy", ss.service.Name())
}
}
return nil
}
// ServicesByState returns a snapshot of the current state of all services.
func (registry *Registry) ServicesByState() map[State][]Name {
result := make(map[State][]Name)
for _, ss := range registry.services {
state := ss.getState()
result[state] = append(result[state], ss.service.Name())
}
return result
}
// IsHealthy returns true if all services are in the RUNNING state.
func (registry *Registry) IsHealthy() bool {
for _, ss := range registry.services {
if ss.getState() != StateRunning {
return false
}
}
return true
}
// detectCyclicDeps returns an error listing all dependency cycles found using
// gonum's Tarjan SCC algorithm.
func detectCyclicDeps(services []*serviceWithState) error {
nameToID := make(map[Name]int64, len(services))
idToName := make(map[int64]Name, len(services))
for i, ss := range services {
id := int64(i)
nameToID[ss.service.Name()] = id
idToName[id] = ss.service.Name()
}
g := simple.NewDirectedGraph()
for _, ss := range services {
g.AddNode(simple.Node(nameToID[ss.service.Name()]))
}
for _, ss := range services {
fromID := nameToID[ss.service.Name()]
for _, dep := range ss.dependsOn {
g.SetEdge(simple.Edge{F: simple.Node(fromID), T: simple.Node(nameToID[dep])})
}
}
if _, err := topo.Sort(g); err == nil {
return nil
}
var cycles [][]Name
for _, scc := range topo.TarjanSCC(g) {
if len(scc) > 1 {
cycle := make([]Name, len(scc))
for i, n := range scc {
cycle[i] = idToName[n.ID()]
}
cycles = append(cycles, cycle)
}
}
return errors.Newf(errors.TypeInvalidInput, ErrCodeInvalidRegistry, "dependency cycles detected: %v", cycles)
}

View File

@@ -5,10 +5,7 @@ import (
"log/slog"
"sync"
"testing"
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -31,46 +28,11 @@ func (s *tservice) Stop(_ context.Context) error {
return nil
}
type healthyTestService struct {
tservice
healthyC chan struct{}
}
func newHealthyTestService(t *testing.T) *healthyTestService {
t.Helper()
return &healthyTestService{
tservice: tservice{c: make(chan struct{})},
healthyC: make(chan struct{}),
}
}
func (s *healthyTestService) Healthy() <-chan struct{} {
return s.healthyC
}
// failingHealthyService implements Healthy but fails before signaling healthy.
type failingHealthyService struct {
healthyC chan struct{}
err error
}
func (s *failingHealthyService) Start(_ context.Context) error {
return s.err
}
func (s *failingHealthyService) Stop(_ context.Context) error {
return nil
}
func (s *failingHealthyService) Healthy() <-chan struct{} {
return s.healthyC
}
func TestRegistryWith2Services(t *testing.T) {
s1 := newTestService(t)
s2 := newTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
registry, err := NewRegistry(slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
require.NoError(t, err)
ctx, cancel := context.WithCancel(context.Background())
@@ -79,8 +41,8 @@ func TestRegistryWith2Services(t *testing.T) {
go func() {
defer wg.Done()
registry.Start(ctx)
assert.NoError(t, registry.Wait(ctx))
assert.NoError(t, registry.Stop(ctx))
require.NoError(t, registry.Wait(ctx))
require.NoError(t, registry.Stop(ctx))
}()
cancel()
@@ -91,7 +53,7 @@ func TestRegistryWith2ServicesWithoutWait(t *testing.T) {
s1 := newTestService(t)
s2 := newTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
registry, err := NewRegistry(slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
require.NoError(t, err)
ctx := context.Background()
@@ -100,245 +62,8 @@ func TestRegistryWith2ServicesWithoutWait(t *testing.T) {
go func() {
defer wg.Done()
registry.Start(ctx)
assert.NoError(t, registry.Stop(ctx))
require.NoError(t, registry.Stop(ctx))
}()
wg.Wait()
}
func TestServiceStateTransitions(t *testing.T) {
s1 := newTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
require.NoError(t, registry.AwaitHealthy(ctx))
byState := registry.ServicesByState()
assert.Len(t, byState[StateRunning], 1)
assert.True(t, registry.IsHealthy())
assert.NoError(t, registry.Stop(ctx))
}
func TestServiceStateWithHealthy(t *testing.T) {
s1 := newHealthyTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
// Poll until STARTING state is observed
require.Eventually(t, func() bool {
byState := registry.ServicesByState()
return len(byState[StateStarting]) == 1
}, time.Second, time.Millisecond)
assert.False(t, registry.IsHealthy())
// Signal healthy
close(s1.healthyC)
require.NoError(t, registry.AwaitHealthy(ctx))
assert.True(t, registry.IsHealthy())
byState := registry.ServicesByState()
assert.Len(t, byState[StateRunning], 1)
assert.NoError(t, registry.Stop(ctx))
}
func TestAwaitHealthy(t *testing.T) {
s1 := newTestService(t)
s2 := newTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
assert.NoError(t, registry.AwaitHealthy(ctx))
assert.True(t, registry.IsHealthy())
assert.NoError(t, registry.Stop(ctx))
}
func TestAwaitHealthyWithFailure(t *testing.T) {
s1 := &failingHealthyService{
healthyC: make(chan struct{}),
err: errors.Newf(errors.TypeInternal, errors.CodeInternal,"startup failed"),
}
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
err = registry.AwaitHealthy(ctx)
assert.Error(t, err)
assert.Contains(t, err.Error(), "startup failed")
}
func TestServicesByState(t *testing.T) {
s1 := newTestService(t)
s2 := newHealthyTestService(t)
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1), NewNamedService(MustNewName("s2"), s2))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
// Wait for s1 to be running (no Healthy interface) and s2 to be starting
require.Eventually(t, func() bool {
byState := registry.ServicesByState()
return len(byState[StateRunning]) == 1 && len(byState[StateStarting]) == 1
}, time.Second, time.Millisecond)
// Make s2 healthy
close(s2.healthyC)
require.NoError(t, registry.AwaitHealthy(ctx))
byState := registry.ServicesByState()
assert.Len(t, byState[StateRunning], 2)
assert.NoError(t, registry.Stop(ctx))
}
func TestDependsOnStartsAfterDependency(t *testing.T) {
s1 := newHealthyTestService(t)
s2 := newTestService(t)
// s2 depends on s1
registry, err := NewRegistry(
context.Background(),
slog.New(slog.DiscardHandler),
NewNamedService(MustNewName("s1"), s1),
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
)
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
// s2 should still be STARTING because s1 hasn't become healthy yet
require.Eventually(t, func() bool {
byState := registry.ServicesByState()
return len(byState[StateStarting]) == 2
}, time.Second, time.Millisecond)
// Make s1 healthy — s2 should then start and become RUNNING
close(s1.healthyC)
assert.NoError(t, registry.AwaitHealthy(ctx))
assert.True(t, registry.IsHealthy())
assert.NoError(t, registry.Stop(ctx))
}
func TestDependsOnFailsWhenDependencyFails(t *testing.T) {
s1 := &failingHealthyService{
healthyC: make(chan struct{}),
err: errors.Newf(errors.TypeInternal, errors.CodeInternal,"s1 crashed"),
}
s2 := newTestService(t)
// s2 depends on s1
registry, err := NewRegistry(
context.Background(),
slog.New(slog.DiscardHandler),
NewNamedService(MustNewName("s1"), s1),
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
)
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
// Both should eventually fail
assert.Eventually(t, func() bool {
byState := registry.ServicesByState()
return len(byState[StateFailed]) == 2
}, time.Second, time.Millisecond)
}
func TestDependsOnUnknownServiceIsIgnored(t *testing.T) {
s1 := newTestService(t)
// Unknown dependency is logged and ignored, not an error.
registry, err := NewRegistry(
context.Background(),
slog.New(slog.DiscardHandler),
NewNamedService(MustNewName("s1"), s1, MustNewName("unknown")),
)
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
assert.NoError(t, registry.AwaitHealthy(ctx))
assert.True(t, registry.IsHealthy())
assert.NoError(t, registry.Stop(ctx))
}
func TestServiceStateFailed(t *testing.T) {
s1 := &failingHealthyService{
healthyC: make(chan struct{}),
err: errors.Newf(errors.TypeInternal, errors.CodeInternal,"fatal error"),
}
registry, err := NewRegistry(context.Background(), slog.New(slog.DiscardHandler), NewNamedService(MustNewName("s1"), s1))
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
// Wait for the service to fail
assert.Eventually(t, func() bool {
byState := registry.ServicesByState()
return len(byState[StateFailed]) == 1
}, time.Second, time.Millisecond)
assert.False(t, registry.IsHealthy())
}
func TestDependsOnSelfDependencyIsIgnored(t *testing.T) {
s1 := newTestService(t)
// Self-dependency is logged and ignored.
registry, err := NewRegistry(
context.Background(),
slog.New(slog.DiscardHandler),
NewNamedService(MustNewName("s1"), s1, MustNewName("s1")),
)
require.NoError(t, err)
ctx := context.Background()
registry.Start(ctx)
assert.NoError(t, registry.AwaitHealthy(ctx))
assert.True(t, registry.IsHealthy())
assert.NoError(t, registry.Stop(ctx))
}
func TestDependsOnCycleReturnsError(t *testing.T) {
s1 := newTestService(t)
s2 := newTestService(t)
// A -> B and B -> A is a cycle.
_, err := NewRegistry(
context.Background(),
slog.New(slog.DiscardHandler),
NewNamedService(MustNewName("s1"), s1, MustNewName("s2")),
NewNamedService(MustNewName("s2"), s2, MustNewName("s1")),
)
assert.Error(t, err)
assert.Contains(t, err.Error(), "dependency cycles detected")
}

View File

@@ -2,81 +2,30 @@ package factory
import "context"
// Service is the core lifecycle interface for long-running services.
type Service interface {
// Starts a service. It should block and should not return until the service is stopped or it fails.
Start(context.Context) error
// Stops a service.
Stop(context.Context) error
}
// Healthy is an optional interface that services can implement to signal
// when they have completed startup and are ready to serve.
// Services that do not implement this interface are considered healthy
// immediately after Start() is called.
type Healthy interface {
// Healthy returns a channel that is closed when the service is healthy.
Healthy() <-chan struct{}
}
// ServiceWithHealthy is a Service that explicitly signals when it is healthy.
type ServiceWithHealthy interface {
Service
Healthy
}
// NamedService is a Service with a Name and optional dependencies.
type NamedService interface {
Named
ServiceWithHealthy
// DependsOn returns the names of services that must be healthy before this service starts.
DependsOn() []Name
Service
}
// closedC is a pre-closed channel returned for services that don't implement Healthy.
var closedC = func() chan struct{} {
c := make(chan struct{})
close(c)
return c
}()
type namedService struct {
name Name
dependsOn []Name
service Service
}
// NewNamedService wraps a Service with a Name and optional dependency names.
func NewNamedService(name Name, service Service, dependsOn ...Name) NamedService {
return &namedService{
name: name,
dependsOn: dependsOn,
service: service,
}
name Name
Service
}
func (s *namedService) Name() Name {
return s.name
}
func (s *namedService) DependsOn() []Name {
return s.dependsOn
}
func (s *namedService) Start(ctx context.Context) error {
return s.service.Start(ctx)
}
func (s *namedService) Stop(ctx context.Context) error {
return s.service.Stop(ctx)
}
// Healthy delegates to the underlying service if it implements Healthy,
// otherwise returns an already-closed channel (immediately healthy).
func (s *namedService) Healthy() <-chan struct{} {
if h, ok := s.service.(Healthy); ok {
return h.Healthy()
func NewNamedService(name Name, service Service) NamedService {
return &namedService{
name: name,
Service: service,
}
return closedC
}

View File

@@ -1,75 +0,0 @@
package factory
import "sync"
// State represents the lifecycle state of a service.
type State struct {
s string
}
func (s State) String() string {
return s.s
}
// MarshalText implements encoding.TextMarshaler so State can be used as a JSON map key.
func (s State) MarshalText() ([]byte, error) {
return []byte(s.s), nil
}
var (
StateStarting = State{"starting"}
StateRunning = State{"running"}
StateFailed = State{"failed"}
)
// serviceWithState wraps a NamedService with thread-safe state tracking.
type serviceWithState struct {
// service is the underlying named service.
service NamedService
// dependsOn is the validated subset of declared dependencies that exist in the registry.
dependsOn []Name
// mu protects state and startErr from concurrent access.
mu sync.RWMutex
// state is the current lifecycle state of the service.
state State
// healthyC is closed when the service transitions to StateRunning.
healthyC chan struct{}
// startReturnedC is closed when Start() returns, whether with nil or an error.
startReturnedC chan struct{}
// startErr is the error returned by Start(), or nil if it returned successfully.
startErr error
}
func newServiceWithState(service NamedService) *serviceWithState {
return &serviceWithState{
service: service,
state: StateStarting,
healthyC: make(chan struct{}),
startReturnedC: make(chan struct{}),
}
}
func (ss *serviceWithState) setState(state State) {
ss.mu.Lock()
defer ss.mu.Unlock()
ss.state = state
if state == StateRunning {
select {
case <-ss.healthyC:
default:
close(ss.healthyC)
}
}
}
func (ss *serviceWithState) getState() State {
ss.mu.RLock()
defer ss.mu.RUnlock()
return ss.state
}

View File

@@ -53,7 +53,7 @@ type Module interface {
}
type Handler interface {
CreateAccount(http.ResponseWriter, *http.Request)
GetConnectionArtifact(http.ResponseWriter, *http.Request)
ListAccounts(http.ResponseWriter, *http.Request)
GetAccount(http.ResponseWriter, *http.Request)
UpdateAccount(http.ResponseWriter, *http.Request)

View File

@@ -1,58 +0,0 @@
package implcloudintegration
import (
"net/http"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
)
type handler struct{}
func NewHandler() cloudintegration.Handler {
return &handler{}
}
func (handler *handler) CreateAccount(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) ListAccounts(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) GetAccount(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) UpdateAccount(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) DisconnectAccount(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) ListServicesMetadata(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) GetService(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) UpdateService(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}
func (handler *handler) AgentCheckIn(writer http.ResponseWriter, request *http.Request) {
// TODO implement me
panic("implement me")
}

View File

@@ -6,19 +6,20 @@ import (
"fmt"
"io"
"net/http"
"slices"
"net/url"
"strconv"
"strings"
"time"
"unicode"
"unicode/utf8"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/types/exporttypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
@@ -30,31 +31,129 @@ func NewHandler(module rawdataexport.Module) rawdataexport.Handler {
return &handler{module: module}
}
// ExportRawData handles data export requests.
//
// API Documentation:
// Endpoint: GET /api/v1/export_raw_data
//
// Query Parameters:
//
// - source (optional): Type of data to export ["logs" (default), "metrics", "traces"]
// Note: Currently only "logs" is fully supported
//
// - format (optional): Output format ["csv" (default), "jsonl"]
//
// - start (required): Start time for query (Unix timestamp in nanoseconds)
//
// - end (required): End time for query (Unix timestamp in nanoseconds)
//
// - limit (optional): Maximum number of rows to export
// Constraints: Must be positive and cannot exceed MAX_EXPORT_ROW_COUNT_LIMIT
//
// - filter (optional): Filter expression to apply to the query
//
// - columns (optional): Specific columns to include in export
// Default: all columns are returned
// Format: ["context.field:type", "context.field", "field"]
//
// - order_by (optional): Sorting specification ["column:direction" or "context.field:type:direction"]
// Direction: "asc" or "desc"
// Default: ["timestamp:desc", "id:desc"]
//
// Response Headers:
// - Content-Type: "text/csv" or "application/x-ndjson"
// - Content-Encoding: "gzip" (handled by HTTP middleware)
// - Content-Disposition: "attachment; filename=\"data_exported.[format]\""
// - Cache-Control: "no-cache"
// - Vary: "Accept-Encoding"
// - Transfer-Encoding: "chunked"
// - Trailers: X-Response-Complete
//
// Response Format:
//
// CSV: Headers in first row, data in subsequent rows
// JSONL: One JSON object per line
//
// Example Usage:
//
// Basic CSV export:
// GET /api/v1/export_raw_data?start=1693612800000000000&end=1693699199000000000
//
// Export with columns and format:
// GET /api/v1/export_raw_data?start=1693612800000000000&end=1693699199000000000&format=jsonl
// &columns=timestamp&columns=severity&columns=message
//
// Export with filter and ordering:
// GET /api/v1/export_raw_data?start=1693612800000000000&end=1693699199000000000
// &filter=severity="error"&order_by=timestamp:desc&limit=1000
func (handler *handler) ExportRawData(rw http.ResponseWriter, r *http.Request) {
var queryRangeRequest qbtypes.QueryRangeRequest
var formatParam exporttypes.ExportRawDataFormatQueryParam
if err := binding.Query.BindQuery(r.URL.Query(), &formatParam); err != nil {
render.Error(rw, err)
return
}
format := formatParam.Format
if err := binding.JSON.BindBody(r.Body, &queryRangeRequest); err != nil {
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid request body: %v", err))
return
}
if err := validateSpecForExport(&queryRangeRequest); err != nil {
source, err := getExportQuerySource(r.URL.Query())
if err != nil {
render.Error(rw, err)
return
}
if err := validateAndApplyDefaultExportLimits(queryRangeRequest.CompositeQuery.Queries); err != nil {
switch source {
case "logs":
handler.exportLogs(rw, r)
case "traces":
handler.exportTraces(rw, r)
case "metrics":
handler.exportMetrics(rw, r)
default:
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid source: must be logs"))
}
}
func (handler *handler) exportMetrics(rw http.ResponseWriter, r *http.Request) {
render.Error(rw, errors.Newf(errors.TypeUnsupported, errors.CodeUnsupported, "metrics export is not yet supported"))
}
func (handler *handler) exportTraces(rw http.ResponseWriter, r *http.Request) {
render.Error(rw, errors.Newf(errors.TypeUnsupported, errors.CodeUnsupported, "traces export is not yet supported"))
}
func (handler *handler) exportLogs(rw http.ResponseWriter, r *http.Request) {
// Set up response headers
rw.Header().Set("Cache-Control", "no-cache")
rw.Header().Set("Vary", "Accept-Encoding") // Indicate that response varies based on Accept-Encoding
rw.Header().Set("Access-Control-Expose-Headers", "Content-Disposition, X-Response-Complete")
rw.Header().Set("Trailer", "X-Response-Complete")
rw.Header().Set("Transfer-Encoding", "chunked")
queryParams := r.URL.Query()
startTime, endTime, err := getExportQueryTimeRange(queryParams)
if err != nil {
render.Error(rw, err)
return
}
queryRangeRequest.UseDefaultOrderBy()
limit, err := getExportQueryLimit(queryParams)
if err != nil {
render.Error(rw, err)
return
}
format, err := getExportQueryFormat(queryParams)
if err != nil {
render.Error(rw, err)
return
}
// Set appropriate content type and filename
filename := fmt.Sprintf("data_exported_%s.%s", time.Now().Format("2006-01-02_150405"), format)
rw.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", filename))
filterExpression := queryParams.Get("filter")
orderByExpression, err := getExportQueryOrderBy(queryParams)
if err != nil {
render.Error(rw, err)
return
}
columns := getExportQueryColumns(queryParams)
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
@@ -62,98 +161,76 @@ func (handler *handler) ExportRawData(rw http.ResponseWriter, r *http.Request) {
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(rw, errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "orgID is invalid"))
return
}
setExportResponseHeaders(rw, format)
queryRangeRequest := qbtypes.QueryRangeRequest{
Start: startTime,
End: endTime,
RequestType: qbtypes.RequestTypeRaw,
CompositeQuery: qbtypes.CompositeQuery{
Queries: []qbtypes.QueryEnvelope{
{
Type: qbtypes.QueryTypeBuilder,
Spec: nil,
},
},
},
}
spec := qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{
Signal: telemetrytypes.SignalLogs,
Name: "raw",
Filter: &qbtypes.Filter{
Expression: filterExpression,
},
Limit: limit,
Order: orderByExpression,
}
spec.SelectFields = columns
queryRangeRequest.CompositeQuery.Queries[0].Spec = spec
// This will signal Export module to stop sending data
doneChan := make(chan any)
defer close(doneChan)
rowChan, errChan := handler.module.ExportRawData(r.Context(), orgID, &queryRangeRequest, doneChan)
isComplete, err := handler.executeExport(rowChan, errChan, format, rw)
if err != nil {
render.Error(rw, err)
return
}
rw.Header().Set("X-Response-Complete", strconv.FormatBool(isComplete))
}
var isComplete bool
// validateSpecForExport validates query specs
func validateSpecForExport(req *qbtypes.QueryRangeRequest) error {
queries := req.CompositeQuery.Queries
// If the trace operator query is not present, and there are multiple queries, return an error
if req.TraceOperatorQueryIndex() == -1 && len(queries) > 1 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "multiple queries not allowed without a trace operator query")
}
for idx := range queries {
switch spec := queries[idx].Spec.(type) {
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation],
qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation],
qbtypes.QueryBuilderTraceOperator:
// Supported spec types
default:
return errors.NewInvalidInputf(errors.CodeInvalidInput, "unsupported query at index %d type: %T", idx, spec)
}
}
opts := append(qbtypes.GetValidationOptions(req.RequestType), qbtypes.WithSkipLimitOffsetValidation())
return req.Validate(opts...)
}
func validateAndApplyDefaultExportLimits(queries []qbtypes.QueryEnvelope) error {
for idx := range queries {
limit := queries[idx].GetLimit()
if limit == 0 {
limit = DefaultExportRowCountLimit
} else if limit < 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be positive")
} else if limit > MaxExportRowCountLimit {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit cannot be more than %d", MaxExportRowCountLimit)
}
queries[idx].SetLimit(limit)
}
return nil
}
// setExportResponseHeaders sets common HTTP headers for export responses.
func setExportResponseHeaders(rw http.ResponseWriter, format string) {
rw.Header().Set("Cache-Control", "no-cache")
rw.Header().Set("Vary", "Accept-Encoding")
rw.Header().Set("Access-Control-Expose-Headers", "Content-Disposition, X-Response-Complete")
rw.Header().Set("Trailer", "X-Response-Complete")
rw.Header().Set("Transfer-Encoding", "chunked")
filename := fmt.Sprintf("data_exported_%s.%s", time.Now().Format("2006-01-02_150405"), format)
rw.Header().Set("Content-Disposition", fmt.Sprintf("attachment; filename=\"%s\"", filename))
}
// executeExport streams data from rowChan to the response writer in the specified format.
func (handler *handler) executeExport(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, format string, rw http.ResponseWriter) (bool, error) {
switch format {
case "csv", "":
rw.Header().Set("Content-Type", "text/csv")
csvWriter := csv.NewWriter(rw)
isComplete, err := handler.exportRawDataCSV(rowChan, errChan, csvWriter)
isComplete, err = handler.exportLogsCSV(rowChan, errChan, csvWriter)
if err != nil {
return false, err
render.Error(rw, err)
return
}
csvWriter.Flush()
return isComplete, nil
case "jsonl":
rw.Header().Set("Content-Type", "application/x-ndjson")
return handler.exportRawDataJSONL(rowChan, errChan, rw)
isComplete, err = handler.exportLogsJSONL(rowChan, errChan, rw)
if err != nil {
render.Error(rw, err)
return
}
default:
return false, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl")
render.Error(rw, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl"))
return
}
rw.Header().Set("X-Response-Complete", strconv.FormatBool(isComplete))
}
// exportRawDataCSV is a generic CSV export function that works with any raw data (logs, traces, etc.)
func (handler *handler) exportRawDataCSV(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, csvWriter *csv.Writer) (bool, error) {
func (handler *handler) exportLogsCSV(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, csvWriter *csv.Writer) (bool, error) {
var header []string
headerToIndexMapping := make(map[string]int)
headerToIndexMapping := make(map[string]int, len(header))
totalBytes := uint64(0)
for {
@@ -191,8 +268,8 @@ func (handler *handler) exportRawDataCSV(rowChan <-chan *qbtypes.RawRow, errChan
}
}
// exportRawDataJSONL is a generic JSONL export function that works with any raw data (logs, traces, etc.)
func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, writer io.Writer) (bool, error) {
func (handler *handler) exportLogsJSONL(rowChan <-chan *qbtypes.RawRow, errChan <-chan error, writer io.Writer) (bool, error) {
totalBytes := uint64(0)
for {
select {
@@ -200,11 +277,9 @@ func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errCh
if !ok {
return true, nil
}
jsonBytes, err := json.Marshal(row.Data)
if err != nil {
return false, errors.NewUnexpectedf(errors.CodeInternal, "error marshaling JSON: %s", err)
}
totalBytes += uint64(len(jsonBytes)) + 1
// Handle JSON format (JSONL - one object per line)
jsonBytes, _ := json.Marshal(row.Data)
totalBytes += uint64(len(jsonBytes)) + 1 // +1 for newline
if _, err := writer.Write(jsonBytes); err != nil {
return false, errors.NewUnexpectedf(errors.CodeInternal, "error writing JSON: %s", err)
@@ -224,33 +299,74 @@ func (handler *handler) exportRawDataJSONL(rowChan <-chan *qbtypes.RawRow, errCh
}
}
// priorityColumns defines the columns that should appear first in the CSV output, in order.
var priorityColumns = []string{"timestamp", "id"}
func getExportQuerySource(queryParams url.Values) (string, error) {
switch queryParams.Get("source") {
case "logs", "":
return "logs", nil
case "metrics":
return "metrics", errors.NewInvalidInputf(errors.CodeInvalidInput, "metrics export not yet supported")
case "traces":
return "traces", errors.NewInvalidInputf(errors.CodeInvalidInput, "traces export not yet supported")
default:
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid source: must be logs, metrics or traces")
}
}
func getExportQueryFormat(queryParams url.Values) (string, error) {
switch queryParams.Get("format") {
case "csv", "":
return "csv", nil
case "jsonl":
return "jsonl", nil
default:
return "", errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid format: must be csv or jsonl")
}
}
func getExportQueryLimit(queryParams url.Values) (int, error) {
limitStr := queryParams.Get("limit")
if limitStr == "" {
return DefaultExportRowCountLimit, nil
} else {
limit, err := strconv.Atoi(limitStr)
if err != nil {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid limit format: %s", err.Error())
}
if limit <= 0 {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be positive")
}
if limit > MaxExportRowCountLimit {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "limit cannot be more than %d", MaxExportRowCountLimit)
}
return limit, nil
}
}
func getExportQueryTimeRange(queryParams url.Values) (uint64, uint64, error) {
startTimeStr := queryParams.Get("start")
endTimeStr := queryParams.Get("end")
if startTimeStr == "" || endTimeStr == "" {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "start and end time are required")
}
startTime, err := strconv.ParseUint(startTimeStr, 10, 64)
if err != nil {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid start time format: %s", err.Error())
}
endTime, err := strconv.ParseUint(endTimeStr, 10, 64)
if err != nil {
return 0, 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid end time format: %s", err.Error())
}
return startTime, endTime, nil
}
func constructCSVHeaderFromQueryResponse(data map[string]any) []string {
header := make([]string, 0, len(data))
for key := range data {
header = append(header, key)
}
// This is to ensure CSV output is consistent across multiple queries
slices.SortFunc(header, func(a, b string) int {
ai, bi := slices.Index(priorityColumns, a), slices.Index(priorityColumns, b)
switch {
case ai != -1 && bi != -1:
return ai - bi
case ai != -1:
return -1
case bi != -1:
return 1
default:
if a < b {
return -1
} else if a > b {
return 1
}
return 0
}
})
return header
}
@@ -311,12 +427,9 @@ func constructCSVRecordFromQueryResponse(data map[string]any, headerToIndexMappi
valueStr = v.String()
default:
jsonBytes, err := json.Marshal(v)
if err != nil {
valueStr = fmt.Sprintf("%v", v)
} else {
valueStr = string(jsonBytes)
}
// For all other complex types (maps, structs, etc.)
jsonBytes, _ := json.Marshal(v)
valueStr = string(jsonBytes)
}
record[index] = sanitizeForCSV(valueStr)
@@ -325,6 +438,26 @@ func constructCSVRecordFromQueryResponse(data map[string]any, headerToIndexMappi
return record
}
// getExportQueryColumns parses the "columns" query parameters and returns a slice of TelemetryFieldKey structs.
// Each column should be a valid telemetry field key in the format "context.field:type" or "context.field" or "field"
func getExportQueryColumns(queryParams url.Values) []telemetrytypes.TelemetryFieldKey {
columnParams := queryParams["columns"]
columns := make([]telemetrytypes.TelemetryFieldKey, 0, len(columnParams))
for _, columnStr := range columnParams {
// Skip empty strings
columnStr = strings.TrimSpace(columnStr)
if columnStr == "" {
continue
}
columns = append(columns, telemetrytypes.GetFieldKeyFromKeyText(columnStr))
}
return columns
}
func getsizeOfStringSlice(slice []string) uint64 {
var totalBytes uint64
for _, str := range slice {
@@ -332,3 +465,52 @@ func getsizeOfStringSlice(slice []string) uint64 {
}
return totalBytes
}
// getExportQueryOrderBy parses the "order_by" query parameters and returns a slice of OrderBy structs.
// Each "order_by" parameter should be in the format "column:direction"
// Each "column" should be a valid telemetry field key in the format "context.field:type" or "context.field" or "field"
func getExportQueryOrderBy(queryParams url.Values) ([]qbtypes.OrderBy, error) {
orderByParam := queryParams.Get("order_by")
orderByParam = strings.TrimSpace(orderByParam)
if orderByParam == "" {
return telemetrylogs.DefaultLogsV2SortingOrder, nil
}
parts := strings.Split(orderByParam, ":")
if len(parts) != 2 && len(parts) != 3 {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order_by format: %s, should be <column>:<direction>", orderByParam)
}
column := strings.Join(parts[:len(parts)-1], ":")
direction := parts[len(parts)-1]
orderDirection, ok := qbtypes.OrderDirectionMap[direction]
if !ok {
return nil, errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order_by direction: %s, should be one of %s, %s", direction, qbtypes.OrderDirectionAsc, qbtypes.OrderDirectionDesc)
}
orderByKey := telemetrytypes.GetFieldKeyFromKeyText(column)
orderBy := []qbtypes.OrderBy{
{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: orderByKey,
},
Direction: orderDirection,
},
}
// If we are ordering by the timestamp column, also order by the ID column
if orderByKey.Name == telemetrylogs.LogsV2TimestampColumn {
orderBy = append(orderBy, qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
Direction: orderDirection,
})
}
return orderBy, nil
}

View File

@@ -2,84 +2,58 @@ package implrawdataexport
import (
"net/url"
"strconv"
"testing"
"github.com/SigNoz/signoz/pkg/http/binding"
"github.com/SigNoz/signoz/pkg/types/exporttypes"
"github.com/SigNoz/signoz/pkg/telemetrylogs"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/stretchr/testify/assert"
)
func TestExportRawDataFormatQueryParam_BindingDefaults(t *testing.T) {
var params exporttypes.ExportRawDataFormatQueryParam
err := binding.Query.BindQuery(url.Values{}, &params)
assert.NoError(t, err)
assert.Equal(t, "csv", params.Format)
}
func logQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]{Limit: limit},
}
}
func traceQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{Limit: limit},
}
}
func traceOperatorQuery(limit int) qbtypes.QueryEnvelope {
return qbtypes.QueryEnvelope{
Type: qbtypes.QueryTypeTraceOperator,
Spec: qbtypes.QueryBuilderTraceOperator{Limit: limit, Expression: "A"},
}
}
func makeRequest(queries ...qbtypes.QueryEnvelope) qbtypes.QueryRangeRequest {
return qbtypes.QueryRangeRequest{
Start: 1000000000000,
End: 1000003600000,
RequestType: qbtypes.RequestTypeRaw,
CompositeQuery: qbtypes.CompositeQuery{Queries: queries},
}
}
func TestValidateSpecForExport(t *testing.T) {
func TestGetExportQuerySource(t *testing.T) {
tests := []struct {
name string
req qbtypes.QueryRangeRequest
expectedError bool
name string
queryParams url.Values
expectedSource string
expectedError bool
}{
{
name: "single log query",
req: makeRequest(logQuery(0)),
name: "default logs source",
queryParams: url.Values{},
expectedSource: "logs",
expectedError: false,
},
{
name: "single trace query",
req: makeRequest(traceQuery(0)),
name: "explicit logs source",
queryParams: url.Values{"source": {"logs"}},
expectedSource: "logs",
expectedError: false,
},
{
name: "trace operator alone",
req: makeRequest(traceOperatorQuery(0)),
name: "metrics source - not supported",
queryParams: url.Values{"source": {"metrics"}},
expectedSource: "metrics",
expectedError: true,
},
{
name: "multiple queries without trace operator",
req: makeRequest(logQuery(0), traceQuery(0)),
expectedError: true,
name: "traces source - not supported",
queryParams: url.Values{"source": {"traces"}},
expectedSource: "traces",
expectedError: true,
},
{
name: "unsupported query type",
req: makeRequest(qbtypes.QueryEnvelope{Type: qbtypes.QueryTypeBuilder, Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{}}),
expectedError: true,
name: "invalid source",
queryParams: url.Values{"source": {"invalid"}},
expectedSource: "",
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateSpecForExport(&tt.req)
source, err := getExportQuerySource(tt.queryParams)
assert.Equal(t, tt.expectedSource, source)
if tt.expectedError {
assert.Error(t, err)
} else {
@@ -89,69 +63,456 @@ func TestValidateSpecForExport(t *testing.T) {
}
}
func TestValidateAndApplyDefaultExportLimits(t *testing.T) {
func TestGetExportQueryFormat(t *testing.T) {
tests := []struct {
name string
queries []qbtypes.QueryEnvelope
expectedError bool
checkQueries func(t *testing.T, queries []qbtypes.QueryEnvelope)
name string
queryParams url.Values
expectedFormat string
expectedError bool
}{
{
name: "single log query, zero limit gets default",
queries: makeRequest(logQuery(0)).CompositeQuery.Queries,
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
},
name: "default csv format",
queryParams: url.Values{},
expectedFormat: "csv",
expectedError: false,
},
{
name: "single log query, valid limit kept",
queries: makeRequest(logQuery(1000)).CompositeQuery.Queries,
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, 1000, q[0].GetLimit())
},
name: "explicit csv format",
queryParams: url.Values{"format": {"csv"}},
expectedFormat: "csv",
expectedError: false,
},
{
name: "single log query, max limit kept",
queries: makeRequest(logQuery(MaxExportRowCountLimit)).CompositeQuery.Queries,
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, MaxExportRowCountLimit, q[0].GetLimit())
},
name: "jsonl format",
queryParams: url.Values{"format": {"jsonl"}},
expectedFormat: "jsonl",
expectedError: false,
},
{
name: "single log query, limit exceeds max",
queries: makeRequest(logQuery(MaxExportRowCountLimit + 1)).CompositeQuery.Queries,
name: "invalid format",
queryParams: url.Values{"format": {"xml"}},
expectedFormat: "",
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
format, err := getExportQueryFormat(tt.queryParams)
assert.Equal(t, tt.expectedFormat, format)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
func TestGetExportQueryLimit(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedLimit int
expectedError bool
}{
{
name: "default limit",
queryParams: url.Values{},
expectedLimit: DefaultExportRowCountLimit,
expectedError: false,
},
{
name: "valid limit",
queryParams: url.Values{"limit": {"5000"}},
expectedLimit: 5000,
expectedError: false,
},
{
name: "maximum limit",
queryParams: url.Values{"limit": {strconv.Itoa(MaxExportRowCountLimit)}},
expectedLimit: MaxExportRowCountLimit,
expectedError: false,
},
{
name: "limit exceeds maximum",
queryParams: url.Values{"limit": {"100000"}},
expectedLimit: 0,
expectedError: true,
},
{
name: "single log query, negative limit",
queries: makeRequest(logQuery(-1)).CompositeQuery.Queries,
name: "invalid limit format",
queryParams: url.Values{"limit": {"invalid"}},
expectedLimit: 0,
expectedError: true,
},
{
name: "single trace query, zero limit gets default",
queries: makeRequest(traceQuery(0)).CompositeQuery.Queries,
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
name: "negative limit",
queryParams: url.Values{"limit": {"-100"}},
expectedLimit: 0,
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
limit, err := getExportQueryLimit(tt.queryParams)
assert.Equal(t, tt.expectedLimit, limit)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
}
})
}
}
func TestGetExportQueryTimeRange(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedStartTime uint64
expectedEndTime uint64
expectedError bool
}{
{
name: "valid time range",
queryParams: url.Values{
"start": {"1640995200"},
"end": {"1641081600"},
},
expectedStartTime: 1640995200,
expectedEndTime: 1641081600,
expectedError: false,
},
{
name: "missing start time",
queryParams: url.Values{"end": {"1641081600"}},
expectedError: true,
},
{
name: "missing end time",
queryParams: url.Values{"start": {"1640995200"}},
expectedError: true,
},
{
name: "missing both times",
queryParams: url.Values{},
expectedError: true,
},
{
name: "invalid start time format",
queryParams: url.Values{
"start": {"invalid"},
"end": {"1641081600"},
},
expectedError: true,
},
{
name: "invalid end time format",
queryParams: url.Values{
"start": {"1640995200"},
"end": {"invalid"},
},
expectedError: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
startTime, endTime, err := getExportQueryTimeRange(tt.queryParams)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
assert.Equal(t, tt.expectedStartTime, startTime)
assert.Equal(t, tt.expectedEndTime, endTime)
}
})
}
}
func TestGetExportQueryColumns(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedColumns []telemetrytypes.TelemetryFieldKey
}{
{
name: "no columns specified",
queryParams: url.Values{},
expectedColumns: []telemetrytypes.TelemetryFieldKey{},
},
{
name: "single column",
queryParams: url.Values{
"columns": {"timestamp"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
},
},
{
name: "trace operator alone, zero limit gets default",
queries: makeRequest(traceOperatorQuery(0)).CompositeQuery.Queries,
checkQueries: func(t *testing.T, q []qbtypes.QueryEnvelope) {
assert.Equal(t, DefaultExportRowCountLimit, q[0].GetLimit())
name: "multiple columns",
queryParams: url.Values{
"columns": {"timestamp", "message", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "message"},
{Name: "level"},
},
},
{
name: "empty column name (should be skipped)",
queryParams: url.Values{
"columns": {"timestamp", "", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "level"},
},
},
{
name: "whitespace column name (should be skipped)",
queryParams: url.Values{
"columns": {"timestamp", " ", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "level"},
},
},
{
name: "valid column name with data type",
queryParams: url.Values{
"columns": {"timestamp", "attribute.user:string", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "user", FieldContext: telemetrytypes.FieldContextAttribute, FieldDataType: telemetrytypes.FieldDataTypeString},
{Name: "level"},
},
},
{
name: "valid column name with dot notation",
queryParams: url.Values{
"columns": {"timestamp", "attribute.user.string", "level"},
},
expectedColumns: []telemetrytypes.TelemetryFieldKey{
{Name: "timestamp"},
{Name: "user.string", FieldContext: telemetrytypes.FieldContextAttribute},
{Name: "level"},
},
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := validateAndApplyDefaultExportLimits(tt.queries)
columns := getExportQueryColumns(tt.queryParams)
assert.Equal(t, len(tt.expectedColumns), len(columns))
for i, expectedCol := range tt.expectedColumns {
assert.Equal(t, expectedCol, columns[i])
}
})
}
}
func TestGetExportQueryOrderBy(t *testing.T) {
tests := []struct {
name string
queryParams url.Values
expectedOrder []qbtypes.OrderBy
expectedError bool
}{
{
name: "no order specified",
queryParams: url.Values{},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "single order error, direction not specified",
queryParams: url.Values{
"order_by": {"timestamp"},
},
expectedOrder: nil,
expectedError: true,
},
{
name: "single order no error",
queryParams: url.Values{
"order_by": {"timestamp:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "multiple orders",
queryParams: url.Values{
"order_by": {"timestamp:asc", "body:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "empty order name (should be skipped)",
queryParams: url.Values{
"order_by": {"timestamp:asc", "", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "whitespace order name (should be skipped)",
queryParams: url.Values{
"order_by": {"timestamp:asc", " ", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2TimestampColumn,
},
},
},
{
Direction: qbtypes.OrderDirectionAsc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: telemetrylogs.LogsV2IDColumn,
},
},
},
},
expectedError: false,
},
{
name: "invalid order name (should error out)",
queryParams: url.Values{
"order_by": {"attributes.user:", "id:asc"},
},
expectedOrder: nil,
expectedError: true,
},
{
name: "valid order name (should be included)",
queryParams: url.Values{
"order_by": {"attribute.user:string:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "user",
FieldContext: telemetrytypes.FieldContextAttribute,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
},
},
},
expectedError: false,
},
{
name: "valid order name (should be included)",
queryParams: url.Values{
"order_by": {"attribute.user.string:desc", "id:asc"},
},
expectedOrder: []qbtypes.OrderBy{
{
Direction: qbtypes.OrderDirectionDesc,
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "user.string",
FieldContext: telemetrytypes.FieldContextAttribute,
},
},
},
},
expectedError: false,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
order, err := getExportQueryOrderBy(tt.queryParams)
if tt.expectedError {
assert.Error(t, err)
} else {
assert.NoError(t, err)
if tt.checkQueries != nil {
tt.checkQueries(t, tt.queries)
assert.Equal(t, len(tt.expectedOrder), len(order))
for i, expectedOrd := range tt.expectedOrder {
assert.Equal(t, expectedOrd, order[i])
}
}
})
@@ -168,8 +529,13 @@ func TestConstructCSVHeaderFromQueryResponse(t *testing.T) {
header := constructCSVHeaderFromQueryResponse(data)
// Priority columns come first in order, then the rest alphabetically.
assert.Equal(t, []string{"timestamp", "id", "level", "message"}, header)
// Since map iteration order is not guaranteed, check that all expected keys are present
expectedKeys := []string{"timestamp", "message", "level", "id"}
assert.Equal(t, len(expectedKeys), len(header))
for _, key := range expectedKeys {
assert.Contains(t, header, key)
}
}
func TestConstructCSVRecordFromQueryResponse(t *testing.T) {

View File

@@ -28,18 +28,8 @@ func (m *Module) ExportRawData(ctx context.Context, orgID valuer.UUID, rangeRequ
instrumentationtypes.CodeFunctionName: "ExportRawData",
})
traceOperatorQueryIndex := rangeRequest.TraceOperatorQueryIndex()
queries := rangeRequest.CompositeQuery.Queries
// If the trace operator query is present, mark the queries other than trace operator as disabled
if traceOperatorQueryIndex > -1 {
for idx := range len(queries) {
if idx != traceOperatorQueryIndex {
queries[idx].SetDisabled(true)
}
}
}
spec := rangeRequest.CompositeQuery.Queries[0].Spec.(qbtypes.QueryBuilderQuery[qbtypes.LogAggregation])
rowCountLimit := spec.Limit
rowChan := make(chan *qbtypes.RawRow, 1)
errChan := make(chan error, 1)
@@ -53,62 +43,52 @@ func (m *Module) ExportRawData(ctx context.Context, orgID valuer.UUID, rangeRequ
defer close(errChan)
defer close(rowChan)
if traceOperatorQueryIndex > -1 {
// If the trace operator query is present, we need to export the data for the trace operator query only
exportRawDataForSingleQuery(m.querier, contextWithTimeout, orgID, rangeRequest, rowChan, errChan, doneChan, traceOperatorQueryIndex)
} else {
// If the trace operator query is not present, we need to export the data for the first query only
exportRawDataForSingleQuery(m.querier, contextWithTimeout, orgID, rangeRequest, rowChan, errChan, doneChan, 0)
rowCount := 0
for rowCount < rowCountLimit {
spec.Limit = min(ChunkSize, rowCountLimit-rowCount)
spec.Offset = rowCount
rangeRequest.CompositeQuery.Queries[0].Spec = spec
response, err := m.querier.QueryRange(contextWithTimeout, orgID, rangeRequest)
if err != nil {
errChan <- err
return
}
newRowsCount := 0
for _, result := range response.Data.Results {
resultData, ok := result.(*qbtypes.RawData)
if !ok {
errChan <- errors.NewInternalf(errors.CodeInternal, "expected RawData, got %T", result)
return
}
newRowsCount += len(resultData.Rows)
for _, row := range resultData.Rows {
select {
case rowChan <- row:
case <-doneChan:
return
case <-ctx.Done():
errChan <- ctx.Err()
return
}
}
}
// Break if we did not receive any new rows
if newRowsCount == 0 {
return
}
rowCount += newRowsCount
}
}()
return rowChan, errChan
}
func exportRawDataForSingleQuery(querier querier.Querier, ctx context.Context, orgID valuer.UUID, rangeRequest *qbtypes.QueryRangeRequest, rowChan chan *qbtypes.RawRow, errChan chan error, doneChan chan any, queryIndex int) {
queries := rangeRequest.CompositeQuery.Queries
rowCountLimit := queries[queryIndex].GetLimit()
rowCount := 0
for rowCount < rowCountLimit {
chunkSize := min(ChunkSize, rowCountLimit-rowCount)
queries[queryIndex].SetLimit(chunkSize)
queries[queryIndex].SetOffset(rowCount)
response, err := querier.QueryRange(ctx, orgID, rangeRequest)
if err != nil {
errChan <- err
return
}
newRowsCount := 0
for _, result := range response.Data.Results {
resultData, ok := result.(*qbtypes.RawData)
if !ok {
errChan <- errors.NewInternalf(errors.CodeInternal, "expected RawData, got %T", result)
return
}
newRowsCount += len(resultData.Rows)
for _, row := range resultData.Rows {
select {
case rowChan <- row:
case <-doneChan:
return
case <-ctx.Done():
errChan <- ctx.Err()
return
}
}
}
rowCount += newRowsCount
// Stop if we received fewer rows than requested — no more data available
if newRowsCount < chunkSize {
return
}
}
}

View File

@@ -160,7 +160,7 @@ func (module *module) CreateCallbackAuthNSession(ctx context.Context, authNProvi
return "", errors.WithAdditionalf(err, "root user can only authenticate via password")
}
userRoles, err := module.userGetter.GetRolesByUserID(ctx, newUser.ID)
userRoles, err := module.userGetter.GetUserRoles(ctx, newUser.ID)
if err != nil {
return "", err
}

View File

@@ -37,7 +37,7 @@ func (module *getter) GetRootUserByOrgID(ctx context.Context, orgID valuer.UUID)
return rootUser, userRoles, nil
}
func (module *getter) ListDeprecatedUsersByOrgID(ctx context.Context, orgID valuer.UUID) ([]*types.DeprecatedUser, error) {
func (module *getter) ListByOrgID(ctx context.Context, orgID valuer.UUID) ([]*types.DeprecatedUser, error) {
users, err := module.store.ListUsersByOrgID(ctx, orgID)
if err != nil {
return nil, err
@@ -84,30 +84,13 @@ func (module *getter) ListDeprecatedUsersByOrgID(ctx context.Context, orgID valu
return deprecatedUsers, nil
}
func (module *getter) ListUsersByOrgID(ctx context.Context, orgID valuer.UUID) ([]*types.User, error) {
users, err := module.store.ListUsersByOrgID(ctx, orgID)
if err != nil {
return nil, err
}
// filter root users if feature flag `hide_root_users` is true
evalCtx := featuretypes.NewFlaggerEvaluationContext(orgID)
hideRootUsers := module.flagger.BooleanOrEmpty(ctx, flagger.FeatureHideRootUser, evalCtx)
if hideRootUsers {
users = slices.DeleteFunc(users, func(user *types.User) bool { return user.IsRoot })
}
return users, nil
}
func (module *getter) GetDeprecatedUserByOrgIDAndID(ctx context.Context, orgID valuer.UUID, id valuer.UUID) (*types.DeprecatedUser, error) {
user, err := module.store.GetByOrgIDAndID(ctx, orgID, id)
if err != nil {
return nil, err
}
userRoles, err := module.GetRolesByUserID(ctx, id)
userRoles, err := module.GetUserRoles(ctx, id)
if err != nil {
return nil, err
}
@@ -116,26 +99,18 @@ func (module *getter) GetDeprecatedUserByOrgIDAndID(ctx context.Context, orgID v
return nil, errors.New(errors.TypeUnexpected, authtypes.ErrCodeUserRolesNotFound, "no user roles entries found")
}
if userRoles[0].Role == nil {
return nil, errors.New(errors.TypeUnexpected, authtypes.ErrCodeRoleNotFound, "role not found for user role entry")
}
role := authtypes.SigNozManagedRoleToExistingLegacyRole[userRoles[0].Role.Name]
return types.NewDeprecatedUserFromUserAndRole(user, role), nil
}
func (module *getter) GetUserByOrgIDAndID(ctx context.Context, orgID valuer.UUID, userID valuer.UUID) (*types.User, error) {
return module.store.GetByOrgIDAndID(ctx, orgID, userID)
}
func (module *getter) Get(ctx context.Context, id valuer.UUID) (*types.DeprecatedUser, error) {
user, err := module.store.GetUser(ctx, id)
if err != nil {
return nil, err
}
userRoles, err := module.GetRolesByUserID(ctx, id)
userRoles, err := module.GetUserRoles(ctx, id)
if err != nil {
return nil, err
}
@@ -144,10 +119,6 @@ func (module *getter) Get(ctx context.Context, id valuer.UUID) (*types.Deprecate
return nil, errors.New(errors.TypeUnexpected, authtypes.ErrCodeUserRolesNotFound, "no user roles entries found")
}
if userRoles[0].Role == nil {
return nil, errors.New(errors.TypeUnexpected, authtypes.ErrCodeRoleNotFound, "role not found for user role entry")
}
role := authtypes.SigNozManagedRoleToExistingLegacyRole[userRoles[0].Role.Name]
return types.NewDeprecatedUserFromUserAndRole(user, role), nil
@@ -203,21 +174,11 @@ func (module *getter) GetNonDeletedUserByEmailAndOrgID(ctx context.Context, emai
}
func (module *getter) GetRolesByUserID(ctx context.Context, userID valuer.UUID) ([]*authtypes.UserRole, error) {
func (module *getter) GetUserRoles(ctx context.Context, userID valuer.UUID) ([]*authtypes.UserRole, error) {
userRoles, err := module.userRoleStore.GetUserRolesByUserID(ctx, userID)
if err != nil {
return nil, err
}
for _, ur := range userRoles {
if ur.Role == nil {
return nil, errors.New(errors.TypeUnexpected, authtypes.ErrCodeRoleNotFound, "role not found for user role entry")
}
}
return userRoles, nil
}
func (module *getter) GetUsersByOrgIDAndRoleID(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) ([]*types.User, error) {
return module.store.GetUsersByOrgIDAndRoleID(ctx, orgID, roleID)
}

View File

@@ -85,7 +85,7 @@ func (h *handler) CreateBulkInvite(rw http.ResponseWriter, r *http.Request) {
render.Success(rw, http.StatusCreated, nil)
}
func (h *handler) GetUserDeprecated(w http.ResponseWriter, r *http.Request) {
func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
@@ -106,39 +106,7 @@ func (h *handler) GetUserDeprecated(w http.ResponseWriter, r *http.Request) {
render.Success(w, http.StatusOK, user)
}
func (h *handler) GetUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
userID := mux.Vars(r)["id"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
user, err := h.getter.GetUserByOrgIDAndID(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(userID))
if err != nil {
render.Error(w, err)
return
}
userRoles, err := h.getter.GetRolesByUserID(ctx, user.ID)
if err != nil {
render.Error(w, err)
return
}
userWithRoles := &authtypes.UserWithRoles{
User: user,
UserRoles: userRoles,
}
render.Success(w, http.StatusOK, userWithRoles)
}
func (h *handler) GetMyUserDeprecated(w http.ResponseWriter, r *http.Request) {
func (h *handler) GetMyUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
@@ -157,80 +125,6 @@ func (h *handler) GetMyUserDeprecated(w http.ResponseWriter, r *http.Request) {
render.Success(w, http.StatusOK, user)
}
func (h *handler) GetMyUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
user, err := h.getter.GetUserByOrgIDAndID(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(claims.UserID))
if err != nil {
render.Error(w, err)
return
}
userRoles, err := h.getter.GetRolesByUserID(ctx, user.ID)
if err != nil {
render.Error(w, err)
return
}
userWithRoles := &authtypes.UserWithRoles{
User: user,
UserRoles: userRoles,
}
render.Success(w, http.StatusOK, userWithRoles)
}
func (h *handler) UpdateMyUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
updatableUser := new(types.UpdatableUser)
if err := json.NewDecoder(r.Body).Decode(&updatableUser); err != nil {
render.Error(w, err)
return
}
_, err = h.setter.UpdateUser(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(claims.UserID), updatableUser)
if err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusNoContent, nil)
}
func (h *handler) ListUsersDeprecated(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
users, err := h.getter.ListDeprecatedUsersByOrgID(ctx, valuer.MustNewUUID(claims.OrgID))
if err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusOK, users)
}
func (h *handler) ListUsers(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
@@ -241,7 +135,7 @@ func (h *handler) ListUsers(w http.ResponseWriter, r *http.Request) {
return
}
users, err := h.getter.ListUsersByOrgID(ctx, valuer.MustNewUUID(claims.OrgID))
users, err := h.getter.ListByOrgID(ctx, valuer.MustNewUUID(claims.OrgID))
if err != nil {
render.Error(w, err)
return
@@ -250,7 +144,7 @@ func (h *handler) ListUsers(w http.ResponseWriter, r *http.Request) {
render.Success(w, http.StatusOK, users)
}
func (h *handler) UpdateUserDeprecated(w http.ResponseWriter, r *http.Request) {
func (h *handler) UpdateUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
@@ -268,7 +162,7 @@ func (h *handler) UpdateUserDeprecated(w http.ResponseWriter, r *http.Request) {
return
}
updatedUser, err := h.setter.UpdateUserDeprecated(ctx, valuer.MustNewUUID(claims.OrgID), id, &user, claims.UserID)
updatedUser, err := h.setter.UpdateUser(ctx, valuer.MustNewUUID(claims.OrgID), id, &user, claims.UserID)
if err != nil {
render.Error(w, err)
return
@@ -277,38 +171,6 @@ func (h *handler) UpdateUserDeprecated(w http.ResponseWriter, r *http.Request) {
render.Success(w, http.StatusOK, updatedUser)
}
func (h *handler) UpdateUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
userID := mux.Vars(r)["id"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
if userID == claims.UserID {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "users cannot call this api on self"))
return
}
updatableUser := new(types.UpdatableUser)
if err := json.NewDecoder(r.Body).Decode(&updatableUser); err != nil {
render.Error(w, err)
return
}
_, err = h.setter.UpdateUser(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(userID), updatableUser)
if err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusNoContent, nil)
}
func (h *handler) DeleteUser(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
@@ -581,118 +443,3 @@ func (h *handler) RevokeAPIKey(w http.ResponseWriter, r *http.Request) {
render.Success(w, http.StatusNoContent, nil)
}
func (h *handler) GetRolesByUserID(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
userID := mux.Vars(r)["id"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
user, err := h.getter.GetUserByOrgIDAndID(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(userID))
if err != nil {
render.Error(w, err)
return
}
userRoles, err := h.getter.GetRolesByUserID(ctx, user.ID)
if err != nil {
render.Error(w, err)
return
}
roles := make([]*authtypes.Role, len(userRoles))
for idx, userRole := range userRoles {
roles[idx] = authtypes.NewRoleFromStorableRole(userRole.Role)
}
render.Success(w, http.StatusOK, roles)
}
func (h *handler) SetRoleByUserID(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
userID := mux.Vars(r)["id"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
if userID == claims.UserID {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "users cannot call this api on self"))
return
}
postableRole := new(types.PostableRole)
if err := json.NewDecoder(r.Body).Decode(postableRole); err != nil {
render.Error(w, err)
return
}
if postableRole.Name == "" {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "role name is required"))
return
}
if err := h.setter.AddUserRole(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(userID), postableRole.Name); err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusOK, nil)
}
func (h *handler) RemoveUserRoleByRoleID(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
userID := mux.Vars(r)["id"]
roleID := mux.Vars(r)["roleId"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
if userID == claims.UserID {
render.Error(w, errors.New(errors.TypeInvalidInput, errors.CodeInvalidInput, "users cannot call this api on self"))
return
}
if err := h.setter.RemoveUserRole(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(userID), valuer.MustNewUUID(roleID)); err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusNoContent, nil)
}
func (h *handler) GetUsersByRoleID(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
defer cancel()
roleID := mux.Vars(r)["id"]
claims, err := authtypes.ClaimsFromContext(ctx)
if err != nil {
render.Error(w, err)
return
}
users, err := h.getter.GetUsersByOrgIDAndRoleID(ctx, valuer.MustNewUUID(claims.OrgID), valuer.MustNewUUID(roleID))
if err != nil {
render.Error(w, err)
return
}
render.Success(w, http.StatusOK, users)
}

View File

@@ -23,7 +23,6 @@ type service struct {
authz authz.AuthZ
config user.RootConfig
stopC chan struct{}
healthyC chan struct{}
}
func NewService(
@@ -43,14 +42,12 @@ func NewService(
orgGetter: orgGetter,
authz: authz,
config: config,
stopC: make(chan struct{}),
healthyC: make(chan struct{}),
stopC: make(chan struct{}),
}
}
func (s *service) Start(ctx context.Context) error {
if !s.config.Enabled {
close(s.healthyC)
<-s.stopC
return nil
}
@@ -62,7 +59,6 @@ func (s *service) Start(ctx context.Context) error {
err := s.reconcile(ctx)
if err == nil {
s.settings.Logger().InfoContext(ctx, "root user reconciliation completed successfully")
close(s.healthyC)
<-s.stopC
return nil
}
@@ -78,10 +74,6 @@ func (s *service) Start(ctx context.Context) error {
}
}
func (s *service) Healthy() <-chan struct{} {
return s.healthyC
}
func (s *service) Stop(ctx context.Context) error {
close(s.stopC)
return nil
@@ -133,7 +125,7 @@ func (s *service) createOrPromoteRootUser(ctx context.Context, orgID valuer.UUID
}
if existingUser != nil {
userRoles, err := s.getter.GetRolesByUserID(ctx, existingUser.ID)
userRoles, err := s.getter.GetUserRoles(ctx, existingUser.ID)
if err != nil {
return err
}
@@ -156,7 +148,9 @@ func (s *service) createOrPromoteRootUser(ctx context.Context, orgID valuer.UUID
existingUser.PromoteToRoot()
err = s.store.RunInTx(ctx, func(ctx context.Context) error {
if err := s.setter.UpdateAnyUser(ctx, orgID, existingUser); err != nil {
// update users table
deprecatedUser := types.NewDeprecatedUserFromUserAndRole(existingUser, types.RoleAdmin)
if err := s.setter.UpdateAnyUser(ctx, orgID, deprecatedUser); err != nil {
return err
}
@@ -199,7 +193,8 @@ func (s *service) updateExistingRootUser(ctx context.Context, orgID valuer.UUID,
if existingRoot.Email != s.config.Email {
existingRoot.UpdateEmail(s.config.Email)
if err := s.setter.UpdateAnyUser(ctx, orgID, existingRoot); err != nil {
deprecatedUser := types.NewDeprecatedUserFromUserAndRole(existingRoot, types.RoleAdmin)
if err := s.setter.UpdateAnyUser(ctx, orgID, deprecatedUser); err != nil {
return err
}
}

View File

@@ -220,7 +220,7 @@ func (module *setter) CreateUser(ctx context.Context, user *types.User, opts ...
return nil
}
func (module *setter) UpdateUserDeprecated(ctx context.Context, orgID valuer.UUID, id string, user *types.DeprecatedUser, updatedBy string) (*types.DeprecatedUser, error) {
func (module *setter) UpdateUser(ctx context.Context, orgID valuer.UUID, id string, user *types.DeprecatedUser, updatedBy string) (*types.DeprecatedUser, error) {
existingUser, err := module.getter.GetDeprecatedUserByOrgIDAndID(ctx, orgID, valuer.MustNewUUID(id))
if err != nil {
return nil, err
@@ -265,7 +265,7 @@ func (module *setter) UpdateUserDeprecated(ctx context.Context, orgID valuer.UUI
existingUser.Update(user.DisplayName, user.Role)
// update the user - idempotent (this does analytics too so keeping it outside txn)
if err := module.UpdateAnyUserDeprecated(ctx, orgID, existingUser); err != nil {
if err := module.UpdateAnyUser(ctx, orgID, existingUser); err != nil {
return nil, err
}
@@ -291,46 +291,7 @@ func (module *setter) UpdateUserDeprecated(ctx context.Context, orgID valuer.UUI
return existingUser, nil
}
func (module *setter) UpdateUser(ctx context.Context, orgID valuer.UUID, userID valuer.UUID, updatable *types.UpdatableUser) (*types.User, error) {
existingUser, err := module.getter.GetUserByOrgIDAndID(ctx, orgID, userID)
if err != nil {
return nil, err
}
if err := existingUser.ErrIfRoot(); err != nil {
return nil, errors.WithAdditionalf(err, "cannot update root user")
}
if err := existingUser.ErrIfDeleted(); err != nil {
return nil, errors.WithAdditionalf(err, "cannot update deleted user")
}
existingUser.Update(updatable.DisplayName)
if err := module.UpdateAnyUser(ctx, orgID, existingUser); err != nil {
return nil, err
}
return existingUser, nil
}
func (module *setter) UpdateAnyUser(ctx context.Context, orgID valuer.UUID, user *types.User) error {
if err := module.store.UpdateUser(ctx, orgID, user); err != nil {
return err
}
if err := module.tokenizer.DeleteIdentity(ctx, user.ID); err != nil {
return err
}
// stats collector things
traits := types.NewTraitsFromUser(user)
module.analytics.IdentifyUser(ctx, user.OrgID.String(), user.ID.String(), traits)
module.analytics.TrackUser(ctx, user.OrgID.String(), user.ID.String(), "User Updated", traits)
return nil
}
func (module *setter) UpdateAnyUserDeprecated(ctx context.Context, orgID valuer.UUID, deprecateUser *types.DeprecatedUser) error {
func (module *setter) UpdateAnyUser(ctx context.Context, orgID valuer.UUID, deprecateUser *types.DeprecatedUser) error {
user := types.NewUserFromDeprecatedUser(deprecateUser)
if err := module.store.UpdateUser(ctx, orgID, user); err != nil {
return err
@@ -374,7 +335,7 @@ func (module *setter) DeleteUser(ctx context.Context, orgID valuer.UUID, id stri
return errors.New(errors.TypeForbidden, errors.CodeForbidden, "cannot self delete")
}
userRoles, err := module.getter.GetRolesByUserID(ctx, user.ID)
userRoles, err := module.getter.GetUserRoles(ctx, user.ID)
if err != nil {
return err
}
@@ -552,7 +513,7 @@ func (module *setter) UpdatePasswordByResetPasswordToken(ctx context.Context, to
return err
}
userRoles, err := module.getter.GetRolesByUserID(ctx, user.ID)
userRoles, err := module.getter.GetUserRoles(ctx, user.ID)
if err != nil {
return err
}
@@ -840,121 +801,16 @@ func (module *setter) activatePendingUser(ctx context.Context, user *types.User,
func (module *setter) UpdateUserRoles(ctx context.Context, orgID, userID valuer.UUID, finalRoleNames []string) error {
return module.store.RunInTx(ctx, func(ctx context.Context) error {
// delete old user_role entries
// delete old user_role entries and create new ones from SSO
if err := module.userRoleStore.DeleteUserRoles(ctx, userID); err != nil {
return err
}
// create fresh ones only if there are roles to assign
if len(finalRoleNames) > 0 {
return module.createUserRoleEntries(ctx, orgID, userID, finalRoleNames)
}
return nil
// create fresh ones
return module.createUserRoleEntries(ctx, orgID, userID, finalRoleNames)
})
}
func (module *setter) AddUserRole(ctx context.Context, orgID, userID valuer.UUID, roleName string) error {
existingUser, err := module.getter.GetUserByOrgIDAndID(ctx, orgID, userID)
if err != nil {
return err
}
if err := existingUser.ErrIfRoot(); err != nil {
return errors.WithAdditionalf(err, "cannot add role for root user")
}
if err := existingUser.ErrIfDeleted(); err != nil {
return errors.WithAdditionalf(err, "cannot add role for deleted user")
}
// validate that the role name exists
foundRoles, err := module.authz.ListByOrgIDAndNames(ctx, orgID, []string{roleName})
if err != nil {
return err
}
if len(foundRoles) != 1 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "role name not found: %s", roleName)
}
// check if user already has this role
existingUserRoles, err := module.getter.GetRolesByUserID(ctx, existingUser.ID)
if err != nil {
return err
}
for _, userRole := range existingUserRoles {
if userRole.Role != nil && userRole.Role.Name == roleName {
return nil // role already assigned no-op
}
}
// grant via authz (idempotent)
if err := module.authz.Grant(
ctx,
orgID,
[]string{roleName},
authtypes.MustNewSubject(authtypes.TypeableUser, existingUser.ID.StringValue(), existingUser.OrgID, nil),
); err != nil {
return err
}
// create user_role entry
userRoles := authtypes.NewUserRoles(userID, foundRoles)
if err := module.userRoleStore.CreateUserRoles(ctx, userRoles); err != nil {
return err
}
return module.tokenizer.DeleteIdentity(ctx, userID)
}
func (module *setter) RemoveUserRole(ctx context.Context, orgID, userID valuer.UUID, roleID valuer.UUID) error {
existingUser, err := module.getter.GetUserByOrgIDAndID(ctx, orgID, userID)
if err != nil {
return err
}
if err := existingUser.ErrIfRoot(); err != nil {
return errors.WithAdditionalf(err, "cannot remove role for root user")
}
if err := existingUser.ErrIfDeleted(); err != nil {
return errors.WithAdditionalf(err, "cannot remove role for deleted user")
}
// resolve role name for authz revoke
existingUserRoles, err := module.getter.GetRolesByUserID(ctx, existingUser.ID)
if err != nil {
return err
}
var roleName string
for _, ur := range existingUserRoles {
if ur.Role != nil && ur.RoleID == roleID {
roleName = ur.Role.Name
break
}
}
if roleName == "" {
return errors.Newf(errors.TypeNotFound, authtypes.ErrCodeUserRolesNotFound, "role %s not found for user %s", roleID, userID)
}
// revoke authz grant
if err := module.authz.Revoke(
ctx,
orgID,
[]string{roleName},
authtypes.MustNewSubject(authtypes.TypeableUser, existingUser.ID.StringValue(), existingUser.OrgID, nil),
); err != nil {
return err
}
if err := module.userRoleStore.DeleteUserRoleByUserIDAndRoleID(ctx, userID, roleID); err != nil {
return err
}
return module.tokenizer.DeleteIdentity(ctx, userID)
}
func roleNamesFromUserRoles(userRoles []*authtypes.UserRole) []string {
names := make([]string, 0, len(userRoles))
for _, ur := range userRoles {

View File

@@ -667,22 +667,3 @@ func (store *store) GetUsersByEmailsOrgIDAndStatuses(ctx context.Context, orgID
return users, nil
}
func (store *store) GetUsersByOrgIDAndRoleID(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) ([]*types.User, error) {
users := []*types.User{}
err := store.
sqlstore.
BunDBCtx(ctx).
NewSelect().
Model(&users).
Join(`JOIN user_role ON user_role.user_id = "users".id`).
Where(`"users".org_id = ?`, orgID).
Where("user_role.role_id = ?", roleID).
Scan(ctx)
if err != nil {
return nil, err
}
return users, nil
}

View File

@@ -65,21 +65,6 @@ func (store *userRoleStore) DeleteUserRoles(ctx context.Context, userID valuer.U
return nil
}
func (store *userRoleStore) DeleteUserRoleByUserIDAndRoleID(ctx context.Context, userID valuer.UUID, roleID valuer.UUID) error {
_, err := store.sqlstore.
BunDBCtx(ctx).
NewDelete().
Model(new(authtypes.UserRole)).
Where("user_id = ?", userID).
Where("role_id = ?", roleID).
Exec(ctx)
if err != nil {
return err
}
return nil
}
func (store *userRoleStore) GetUserRolesByUserID(ctx context.Context, userID valuer.UUID) ([]*authtypes.UserRole, error) {
userRoles := make([]*authtypes.UserRole, 0)

View File

@@ -3,5 +3,5 @@ package user
import "github.com/SigNoz/signoz/pkg/factory"
type Service interface {
factory.ServiceWithHealthy
factory.Service
}

View File

@@ -34,12 +34,10 @@ type Setter interface {
// Initiate forgot password flow for a user
ForgotPassword(ctx context.Context, orgID valuer.UUID, email valuer.Email, frontendBaseURL string) error
UpdateUserDeprecated(ctx context.Context, orgID valuer.UUID, id string, user *types.DeprecatedUser, updatedBy string) (*types.DeprecatedUser, error)
UpdateUser(ctx context.Context, orgID valuer.UUID, userID valuer.UUID, updatable *types.UpdatableUser) (*types.User, error)
UpdateUser(ctx context.Context, orgID valuer.UUID, id string, user *types.DeprecatedUser, updatedBy string) (*types.DeprecatedUser, error)
// UpdateAnyUser updates a user and persists the changes to the database along with the analytics and identity deletion.
UpdateAnyUserDeprecated(ctx context.Context, orgID valuer.UUID, deprecateUser *types.DeprecatedUser) error
UpdateAnyUser(ctx context.Context, orgID valuer.UUID, user *types.User) error
UpdateAnyUser(ctx context.Context, orgID valuer.UUID, user *types.DeprecatedUser) error
DeleteUser(ctx context.Context, orgID valuer.UUID, id string, deletedBy string) error
// invite
@@ -54,8 +52,6 @@ type Setter interface {
// Roles
UpdateUserRoles(ctx context.Context, orgID, userID valuer.UUID, finalRoleNames []string) error
AddUserRole(ctx context.Context, orgID, userID valuer.UUID, roleName string) error
RemoveUserRole(ctx context.Context, orgID, userID valuer.UUID, roleID valuer.UUID) error
statsreporter.StatsCollector
}
@@ -64,13 +60,11 @@ type Getter interface {
// Get root user by org id.
GetRootUserByOrgID(context.Context, valuer.UUID) (*types.User, []*authtypes.UserRole, error)
// Get gets the users based on the given org id
ListDeprecatedUsersByOrgID(context.Context, valuer.UUID) ([]*types.DeprecatedUser, error)
ListUsersByOrgID(ctx context.Context, orgID valuer.UUID) ([]*types.User, error)
// Get gets the users based on the given id
ListByOrgID(context.Context, valuer.UUID) ([]*types.DeprecatedUser, error)
// Get deprecated user object by orgID and id.
GetDeprecatedUserByOrgIDAndID(context.Context, valuer.UUID, valuer.UUID) (*types.DeprecatedUser, error)
GetUserByOrgIDAndID(ctx context.Context, orgID valuer.UUID, userID valuer.UUID) (*types.User, error)
// Get user by id.
Get(context.Context, valuer.UUID) (*types.DeprecatedUser, error)
@@ -91,10 +85,7 @@ type Getter interface {
GetNonDeletedUserByEmailAndOrgID(ctx context.Context, email valuer.Email, orgID valuer.UUID) (*types.User, error)
// Gets user_role with roles entries from db
GetRolesByUserID(ctx context.Context, userID valuer.UUID) ([]*authtypes.UserRole, error)
// Gets all the user with role using role id in an org id
GetUsersByOrgIDAndRoleID(ctx context.Context, orgID valuer.UUID, roleID valuer.UUID) ([]*types.User, error)
GetUserRoles(ctx context.Context, userID valuer.UUID) ([]*authtypes.UserRole, error)
}
type Handler interface {
@@ -102,21 +93,11 @@ type Handler interface {
CreateInvite(http.ResponseWriter, *http.Request)
CreateBulkInvite(http.ResponseWriter, *http.Request)
// users
ListUsersDeprecated(http.ResponseWriter, *http.Request)
ListUsers(http.ResponseWriter, *http.Request)
UpdateUserDeprecated(http.ResponseWriter, *http.Request)
UpdateUser(http.ResponseWriter, *http.Request)
DeleteUser(http.ResponseWriter, *http.Request)
GetUserDeprecated(http.ResponseWriter, *http.Request)
GetUser(http.ResponseWriter, *http.Request)
GetMyUserDeprecated(http.ResponseWriter, *http.Request)
GetMyUser(http.ResponseWriter, *http.Request)
UpdateMyUser(http.ResponseWriter, *http.Request)
GetRolesByUserID(http.ResponseWriter, *http.Request)
SetRoleByUserID(http.ResponseWriter, *http.Request)
RemoveUserRoleByRoleID(http.ResponseWriter, *http.Request)
GetUsersByRoleID(http.ResponseWriter, *http.Request)
// Reset Password
GetResetPasswordToken(http.ResponseWriter, *http.Request)

View File

@@ -3,7 +3,6 @@ package prometheus
import (
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/factory"
)
@@ -21,9 +20,6 @@ type Config struct {
//
// If not set, the prometheus default is used (currently 5m).
LookbackDelta time.Duration `mapstructure:"lookback_delta"`
// Timeout is the maximum time a query is allowed to run before being aborted.
Timeout time.Duration `mapstructure:"timeout"`
}
func NewConfigFactory() factory.ConfigFactory {
@@ -37,14 +33,10 @@ func newConfig() factory.Config {
Path: "",
MaxConcurrent: 20,
},
Timeout: 2 * time.Minute,
}
}
func (c Config) Validate() error {
if c.Timeout <= 0 {
return errors.Newf(errors.TypeInvalidInput, errors.CodeInvalidInput, "prometheus::timeout must be greater than 0")
}
return nil
}

View File

@@ -0,0 +1,3 @@
package prometheus
const FingerprintAsPromLabelName = "fingerprint"

View File

@@ -2,6 +2,7 @@ package prometheus
import (
"log/slog"
"time"
"github.com/prometheus/prometheus/promql"
)
@@ -20,7 +21,7 @@ func NewEngine(logger *slog.Logger, cfg Config) *Engine {
Logger: logger,
Reg: nil,
MaxSamples: 5_0000_000,
Timeout: cfg.Timeout,
Timeout: 2 * time.Minute,
ActiveQueryTracker: activeQueryTracker,
LookbackDelta: cfg.LookbackDelta,
})

View File

@@ -6,8 +6,6 @@ import (
"github.com/prometheus/prometheus/promql"
)
const FingerprintAsPromLabelName string = "fingerprint"
func RemoveExtraLabels(res *promql.Result, labelsToRemove ...string) error {
if len(labelsToRemove) == 0 || res == nil {
return nil

View File

@@ -4,7 +4,6 @@ import (
"context"
"encoding/base64"
"fmt"
"log/slog"
"strconv"
"strings"
"time"
@@ -12,7 +11,6 @@ import (
"github.com/ClickHouse/clickhouse-go/v2"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/telemetrystore"
"github.com/SigNoz/signoz/pkg/telemetrytraces"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
@@ -20,7 +18,6 @@ import (
)
type builderQuery[T any] struct {
logger *slog.Logger
telemetryStore telemetrystore.TelemetryStore
stmtBuilder qbtypes.StatementBuilder[T]
spec qbtypes.QueryBuilderQuery[T]
@@ -34,7 +31,6 @@ type builderQuery[T any] struct {
var _ qbtypes.Query = (*builderQuery[any])(nil)
func newBuilderQuery[T any](
logger *slog.Logger,
telemetryStore telemetrystore.TelemetryStore,
stmtBuilder qbtypes.StatementBuilder[T],
spec qbtypes.QueryBuilderQuery[T],
@@ -43,7 +39,6 @@ func newBuilderQuery[T any](
variables map[string]qbtypes.VariableItem,
) *builderQuery[T] {
return &builderQuery[T]{
logger: logger,
telemetryStore: telemetryStore,
stmtBuilder: stmtBuilder,
spec: spec,
@@ -310,45 +305,6 @@ func (q *builderQuery[T]) executeWindowList(ctx context.Context) (*qbtypes.Resul
totalBytes := uint64(0)
start := time.Now()
// Check if filter contains trace_id(s) and optimize time range if needed
if q.spec.Signal == telemetrytypes.SignalTraces &&
q.spec.Filter != nil && q.spec.Filter.Expression != "" {
traceIDs, found := telemetrytraces.ExtractTraceIDsFromFilter(q.spec.Filter.Expression)
if found && len(traceIDs) > 0 {
finder := telemetrytraces.NewTraceTimeRangeFinder(q.telemetryStore)
traceStart, traceEnd, ok := finder.GetTraceTimeRangeMulti(ctx, traceIDs)
traceStartMS := uint64(traceStart) / 1_000_000
traceEndMS := uint64(traceEnd) / 1_000_000
if !ok {
q.logger.DebugContext(ctx, "failed to get trace time range", slog.Any("trace_ids", traceIDs))
} else if traceStartMS > 0 && traceEndMS > 0 {
// no overlap — nothing to return
if uint64(traceStartMS) > toMS || uint64(traceEndMS) < fromMS {
return &qbtypes.Result{
Type: qbtypes.RequestTypeRaw,
Value: &qbtypes.RawData{
QueryName: q.spec.Name,
},
Stats: qbtypes.ExecStats{
DurationMS: uint64(time.Since(start).Milliseconds()),
},
}, nil
}
// clamp window to trace time range before bucketing
if uint64(traceStartMS) > fromMS {
fromMS = uint64(traceStartMS)
}
if uint64(traceEndMS) < toMS {
toMS = uint64(traceEndMS)
}
q.logger.DebugContext(ctx, "optimized time range for traces", slog.Any("trace_ids", traceIDs), slog.Uint64("start", fromMS), slog.Uint64("end", toMS))
}
}
}
// Get buckets and reverse them for ascending order
buckets := makeBuckets(fromMS, toMS)
if isAsc {

View File

@@ -36,28 +36,6 @@ var unquotedDottedNamePattern = regexp.MustCompile(`(?:^|[{,(\s])([a-zA-Z_][a-zA
// This is a common mistake when migrating to UTF-8 syntax.
var quotedMetricOutsideBracesPattern = regexp.MustCompile(`"([^"]+)"\s*\{`)
// tryEnhancePromQLExecError attempts to convert a PromQL execution error into
// a properly typed error. Returns nil if the error is not a recognized execution error.
func tryEnhancePromQLExecError(execErr error) error {
var eqc promql.ErrQueryCanceled
var eqt promql.ErrQueryTimeout
var es promql.ErrStorage
switch {
case errors.As(execErr, &eqc):
return errors.Newf(errors.TypeCanceled, errors.CodeCanceled, "query canceled").WithAdditional(eqc.Error())
case errors.As(execErr, &eqt):
return errors.Newf(errors.TypeTimeout, errors.CodeTimeout, "query timed out").WithAdditional(eqt.Error())
case errors.Is(execErr, context.DeadlineExceeded):
return errors.Newf(errors.TypeTimeout, errors.CodeTimeout, "query timed out")
case errors.Is(execErr, context.Canceled):
return errors.Newf(errors.TypeCanceled, errors.CodeCanceled, "query canceled")
case errors.As(execErr, &es):
return errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", execErr)
default:
return nil
}
}
// enhancePromQLError adds helpful context to PromQL parse errors,
// particularly for UTF-8 syntax migration issues where metric and label
// names containing dots need to be quoted.
@@ -235,20 +213,27 @@ func (q *promqlQuery) Execute(ctx context.Context) (*qbv5.Result, error) {
time.Unix(0, end),
q.query.Step.Duration,
)
if err != nil {
// NewRangeQuery can fail with execution errors (e.g. context deadline exceeded)
// during the query queue/scheduling stage, not just parse errors.
if err := tryEnhancePromQLExecError(err); err != nil {
return nil, err
}
if err != nil {
return nil, enhancePromQLError(query, err)
}
res := qry.Exec(ctx)
if res.Err != nil {
if err := tryEnhancePromQLExecError(res.Err); err != nil {
return nil, err
var eqc promql.ErrQueryCanceled
var eqt promql.ErrQueryTimeout
var es promql.ErrStorage
switch {
case errors.As(res.Err, &eqc):
return nil, errors.Newf(errors.TypeCanceled, errors.CodeCanceled, "query canceled")
case errors.As(res.Err, &eqt):
return nil, errors.Newf(errors.TypeTimeout, errors.CodeTimeout, "query timeout")
case errors.As(res.Err, &es):
return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", res.Err)
}
if errors.Is(res.Err, context.Canceled) {
return nil, errors.Newf(errors.TypeCanceled, errors.CodeCanceled, "query canceled")
}
return nil, errors.Newf(errors.TypeInternal, errors.CodeInternal, "query execution error: %v", res.Err)

View File

@@ -353,13 +353,13 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
case qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]:
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
bq := newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
bq := newBuilderQuery(q.telemetryStore, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
bq := newBuilderQuery(q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
queries[spec.Name] = bq
steps[spec.Name] = spec.StepInterval
case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
@@ -397,9 +397,9 @@ func (q *querier) QueryRange(ctx context.Context, orgID valuer.UUID, req *qbtype
if spec.Source == telemetrytypes.SourceMeter {
event.Source = telemetrytypes.SourceMeter.StringValue()
bq = newBuilderQuery(q.logger, q.telemetryStore, q.meterStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
bq = newBuilderQuery(q.telemetryStore, q.meterStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
} else {
bq = newBuilderQuery(q.logger, q.telemetryStore, q.metricStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
bq = newBuilderQuery(q.telemetryStore, q.metricStmtBuilder, spec, timeRange, req.RequestType, tmplVars)
}
queries[spec.Name] = bq
@@ -509,7 +509,7 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
case <-tick:
// timestamp end is not specified here
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: tsStart}, req.RequestType)
bq := newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
bq := newBuilderQuery(q.telemetryStore, q.logStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
"id": {
Value: updatedLogID,
},
@@ -801,22 +801,22 @@ func (q *querier) createRangedQuery(originalQuery qbtypes.Query, timeRange qbtyp
specCopy := qt.spec.Copy()
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
return newBuilderQuery(q.logger, q.telemetryStore, q.traceStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
return newBuilderQuery(q.telemetryStore, q.traceStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
case *builderQuery[qbtypes.LogAggregation]:
specCopy := qt.spec.Copy()
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
return newBuilderQuery(q.logger, q.telemetryStore, q.logStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
return newBuilderQuery(q.telemetryStore, q.logStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
case *builderQuery[qbtypes.MetricAggregation]:
specCopy := qt.spec.Copy()
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
if qt.spec.Source == telemetrytypes.SourceMeter {
return newBuilderQuery(q.logger, q.telemetryStore, q.meterStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
return newBuilderQuery(q.telemetryStore, q.meterStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
}
return newBuilderQuery(q.logger, q.telemetryStore, q.metricStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
return newBuilderQuery(q.telemetryStore, q.metricStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables)
case *traceOperatorQuery:
specCopy := qt.spec.Copy()
return &traceOperatorQuery{

View File

@@ -576,6 +576,9 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
aH.LicensingAPI.Activate(rw, req)
})).Methods(http.MethodGet)
// Export
router.HandleFunc("/api/v1/export_raw_data", am.ViewAccess(aH.Signoz.Handlers.RawDataExport.ExportRawData)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/span_percentile", am.ViewAccess(aH.Signoz.Handlers.SpanPercentile.GetSpanPercentileDetails)).Methods(http.MethodPost)
// Query Filter Analyzer api used to extract metric names and grouping columns from a query

View File

@@ -4,7 +4,6 @@ import (
"context"
"encoding/json"
"fmt"
"maps"
"slices"
"strings"
@@ -13,7 +12,6 @@ import (
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/query-service/agentConf"
"github.com/SigNoz/signoz/pkg/query-service/constants"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/utils"
@@ -36,101 +34,19 @@ type LogParsingPipelineController struct {
Repo
GetIntegrationPipelines func(context.Context, string) ([]pipelinetypes.GettablePipeline, error)
// TODO(Piyush): remove with qbv5 migration
reader interfaces.Reader
}
func NewLogParsingPipelinesController(
sqlStore sqlstore.SQLStore,
getIntegrationPipelines func(context.Context, string) ([]pipelinetypes.GettablePipeline, error),
reader interfaces.Reader,
) (*LogParsingPipelineController, error) {
repo := NewRepo(sqlStore)
return &LogParsingPipelineController{
Repo: repo,
GetIntegrationPipelines: getIntegrationPipelines,
reader: reader,
}, nil
}
// enrichPipelinesFilters resolves the type (tag vs resource) for filter keys that are
// missing type info, by looking them up in the store.
//
// TODO(Piyush): remove with qbv5 migration
func (pc *LogParsingPipelineController) enrichPipelinesFilters(
ctx context.Context, pipelines []pipelinetypes.GettablePipeline,
) ([]pipelinetypes.GettablePipeline, error) {
// Collect names of non-static keys that are missing type info.
// Static fields (body, trace_id, etc.) are intentionally Unspecified and map
// to top-level OTEL fields — they do not need enrichment.
unspecifiedNames := map[string]struct{}{}
for _, p := range pipelines {
if p.Filter != nil {
for _, item := range p.Filter.Items {
if item.Key.Type == v3.AttributeKeyTypeUnspecified {
// Skip static fields
if _, isStatic := constants.StaticFieldsLogsV3[item.Key.Key]; isStatic {
continue
}
// Skip enrich body.* fields
if strings.HasPrefix(item.Key.Key, "body.") {
continue
}
unspecifiedNames[item.Key.Key] = struct{}{}
}
}
}
}
if len(unspecifiedNames) == 0 {
return pipelines, nil
}
logFields, apiErr := pc.reader.GetLogFieldsFromNames(ctx, slices.Collect(maps.Keys(unspecifiedNames)))
if apiErr != nil {
slog.ErrorContext(ctx, "failed to fetch log fields for pipeline filter enrichment", "error", apiErr)
return pipelines, apiErr
}
// Build a simple name → AttributeKeyType map from the response.
fieldTypes := map[string]v3.AttributeKeyType{}
for _, f := range append(logFields.Selected, logFields.Interesting...) {
switch f.Type {
case constants.Resources:
fieldTypes[f.Name] = v3.AttributeKeyTypeResource
case constants.Attributes:
fieldTypes[f.Name] = v3.AttributeKeyTypeTag
}
}
// Set the resolved type on each untyped filter key in-place.
for i := range pipelines {
if pipelines[i].Filter != nil {
for j := range pipelines[i].Filter.Items {
key := &pipelines[i].Filter.Items[j].Key
if key.Type == v3.AttributeKeyTypeUnspecified {
// Skip static fields
if _, isStatic := constants.StaticFieldsLogsV3[key.Key]; isStatic {
continue
}
// Skip enrich body.* fields
if strings.HasPrefix(key.Key, "body.") {
continue
}
if t, ok := fieldTypes[key.Key]; ok {
key.Type = t
} else {
// default to attribute
key.Type = v3.AttributeKeyTypeTag
}
}
}
}
}
return pipelines, nil
}
// PipelinesResponse is used to prepare http response for pipelines config related requests
type PipelinesResponse struct {
*opamptypes.AgentConfigVersion
@@ -340,12 +256,7 @@ func (ic *LogParsingPipelineController) PreviewLogsPipelines(
ctx context.Context,
request *PipelinesPreviewRequest,
) (*PipelinesPreviewResponse, error) {
pipelines, err := ic.enrichPipelinesFilters(ctx, request.Pipelines)
if err != nil {
return nil, err
}
result, collectorLogs, err := SimulatePipelinesProcessing(ctx, pipelines, request.Logs)
result, collectorLogs, err := SimulatePipelinesProcessing(ctx, request.Pipelines, request.Logs)
if err != nil {
return nil, err
}
@@ -382,8 +293,10 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
if configVersion != nil {
pipelinesVersion = configVersion.Version
}
ctx := context.Background()
pipelinesResp, err := pc.GetPipelinesByVersion(ctx, orgId, pipelinesVersion)
pipelinesResp, err := pc.GetPipelinesByVersion(
context.Background(), orgId, pipelinesVersion,
)
if err != nil {
return nil, "", err
}
@@ -393,17 +306,12 @@ func (pc *LogParsingPipelineController) RecommendAgentConfig(
return nil, "", errors.WrapInternalf(err, CodeRawPipelinesMarshalFailed, "could not serialize pipelines to JSON")
}
enrichedPipelines, err := pc.enrichPipelinesFilters(ctx, pipelinesResp.Pipelines)
if err != nil {
return nil, "", err
}
if querybuilder.BodyJSONQueryEnabled {
// add default normalize pipeline at the beginning, only for sending to collector
enrichedPipelines = append([]pipelinetypes.GettablePipeline{pc.getNormalizePipeline()}, enrichedPipelines...)
pipelinesResp.Pipelines = append([]pipelinetypes.GettablePipeline{pc.getNormalizePipeline()}, pipelinesResp.Pipelines...)
}
updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, enrichedPipelines)
updatedConf, err := GenerateCollectorConfigWithPipelines(currentConfYaml, pipelinesResp.Pipelines)
if err != nil {
return nil, "", err
}

View File

@@ -1407,7 +1407,7 @@ func Test_querier_Traces_runWindowBasedListQueryDesc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -1633,7 +1633,7 @@ func Test_querier_Traces_runWindowBasedListQueryAsc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -1934,7 +1934,7 @@ func Test_querier_Logs_runWindowBasedListQueryDesc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -2162,7 +2162,7 @@ func Test_querier_Logs_runWindowBasedListQueryAsc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,

View File

@@ -1459,7 +1459,7 @@ func Test_querier_Traces_runWindowBasedListQueryDesc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -1685,7 +1685,7 @@ func Test_querier_Traces_runWindowBasedListQueryAsc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -1985,7 +1985,7 @@ func Test_querier_Logs_runWindowBasedListQueryDesc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,
@@ -2213,7 +2213,7 @@ func Test_querier_Logs_runWindowBasedListQueryAsc(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore),
"",
time.Duration(time.Second),
nil,

View File

@@ -121,7 +121,6 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
logParsingPipelineController, err := logparsingpipeline.NewLogParsingPipelinesController(
signoz.SQLStore,
integrationsController.GetPipelinesForInstalledIntegrations,
reader,
)
if err != nil {
return nil, err

View File

@@ -67,7 +67,7 @@ func getPathFromRootToSelectedSpanId(node *model.Span, selectedSpanId string, un
spansFromRootToNode := []string{}
if node.SpanID == selectedSpanId {
if isSelectedSpanIDUnCollapsed && !slices.Contains(uncollapsedSpans, node.SpanID) {
if isSelectedSpanIDUnCollapsed {
spansFromRootToNode = append(spansFromRootToNode, node.SpanID)
}
return true, spansFromRootToNode
@@ -88,15 +88,7 @@ func getPathFromRootToSelectedSpanId(node *model.Span, selectedSpanId string, un
return isPresentInSubtreeForTheNode, spansFromRootToNode
}
// traverseOpts holds the traversal configuration that remains constant
// throughout the recursion. Per-call state (level, isPartOfPreOrder, etc.)
// is passed as direct arguments.
type traverseOpts struct {
uncollapsedSpans []string
selectedSpanID string
}
func traverseTrace(span *model.Span, opts traverseOpts, level uint64, isPartOfPreOrder bool, hasSibling bool) []*model.Span {
func traverseTrace(span *model.Span, uncollapsedSpans []string, level uint64, isPartOfPreOrder bool, hasSibling bool, selectedSpanId string) []*model.Span {
preOrderTraversal := []*model.Span{}
// sort the children to maintain the order across requests
@@ -134,9 +126,8 @@ func traverseTrace(span *model.Span, opts traverseOpts, level uint64, isPartOfPr
preOrderTraversal = append(preOrderTraversal, &nodeWithoutChildren)
}
isAlreadyUncollapsed := slices.Contains(opts.uncollapsedSpans, span.SpanID)
for index, child := range span.Children {
_childTraversal := traverseTrace(child, opts, level+1, isPartOfPreOrder && isAlreadyUncollapsed, index != (len(span.Children)-1))
_childTraversal := traverseTrace(child, uncollapsedSpans, level+1, isPartOfPreOrder && slices.Contains(uncollapsedSpans, span.SpanID), index != (len(span.Children)-1), selectedSpanId)
preOrderTraversal = append(preOrderTraversal, _childTraversal...)
nodeWithoutChildren.SubTreeNodeCount += child.SubTreeNodeCount + 1
span.SubTreeNodeCount += child.SubTreeNodeCount + 1
@@ -177,11 +168,7 @@ func GetSelectedSpans(uncollapsedSpans []string, selectedSpanID string, traceRoo
_, spansFromRootToNode := getPathFromRootToSelectedSpanId(rootNode, selectedSpanID, updatedUncollapsedSpans, isSelectedSpanIDUnCollapsed)
updatedUncollapsedSpans = append(updatedUncollapsedSpans, spansFromRootToNode...)
opts := traverseOpts{
uncollapsedSpans: updatedUncollapsedSpans,
selectedSpanID: selectedSpanID,
}
_preOrderTraversal := traverseTrace(rootNode, opts, 0, true, false)
_preOrderTraversal := traverseTrace(rootNode, updatedUncollapsedSpans, 0, true, false, selectedSpanID)
_selectedSpanIndex := findIndexForSelectedSpanFromPreOrder(_preOrderTraversal, selectedSpanID)
if _selectedSpanIndex != -1 {

View File

@@ -1,355 +0,0 @@
// Package tracedetail tests — waterfall
//
// # Background
//
// The waterfall view renders a trace as a scrollable list of spans in
// pre-order (parent before children, siblings left-to-right). Because a trace
// can have thousands of spans, only a window of ~500 is returned per request.
// The window is centred on the selected span.
//
// # Key concepts
//
// uncollapsedSpans
//
// The set of span IDs the user has manually expanded in the UI.
// Only the direct children of an uncollapsed span are included in the
// output; grandchildren stay hidden until their parent is also uncollapsed.
// When multiple spans are uncollapsed their children are all visible at once.
//
// selectedSpanID
//
// The span currently focused — set when the user clicks a span in the
// waterfall or selects one from the flamegraph. The output window is always
// centred on this span. The path from the trace root down to the selected
// span is automatically uncollapsed so ancestors are visible even if they are
// not in uncollapsedSpans.
//
// isSelectedSpanIDUnCollapsed
//
// Controls whether the selected span's own children are shown:
// true — user expanded the span (click-to-open in waterfall or flamegraph);
// direct children of the selected span are included.
// false — user selected without expanding;
// the span is visible but its children remain hidden.
//
// traceRoots
//
// Root spans of the trace — spans with no parent in the current dataset.
// Normally one, but multiple roots are common when upstream services are
// not instrumented or their spans were not sampled/exported.
package tracedetail
import (
"fmt"
"testing"
"github.com/SigNoz/signoz/pkg/query-service/model"
"github.com/stretchr/testify/assert"
)
// Pre-order traversal is preserved: parent before children, siblings left-to-right.
func TestGetSelectedSpans_PreOrderTraversal(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("child1", "svc", mkSpan("grandchild", "svc")),
mkSpan("child2", "svc"),
)
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{"root", "child1"}, "root", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root", "child1", "grandchild", "child2"}, spanIDs(spans))
}
// Multiple roots: both trees are flattened into a single pre-order list with
// root1's subtree before root2's. Service/entry-point come from the first root.
//
// root1 svc-a ← selected
// └─ child1
// root2 svc-b
// └─ child2
//
// Expected output order: root1 → child1 → root2 → child2
func TestGetSelectedSpans_MultipleRoots(t *testing.T) {
root1 := mkSpan("root1", "svc-a", mkSpan("child1", "svc-a"))
root2 := mkSpan("root2", "svc-b", mkSpan("child2", "svc-b"))
spanMap := buildSpanMap(root1, root2)
spans, _, svcName, entryPoint := GetSelectedSpans([]string{"root1", "root2"}, "root1", []*model.Span{root1, root2}, spanMap, false)
assert.Equal(t, []string{"root1", "child1", "root2", "child2"}, spanIDs(spans), "root1 subtree must precede root2 subtree")
assert.Equal(t, "svc-a", svcName, "metadata comes from first root")
assert.Equal(t, "root1-op", entryPoint, "metadata comes from first root")
}
// isSelectedSpanIDUnCollapsed=true opens only the selected span's direct children,
// not deeper descendants.
//
// root → selected (expanded)
// ├─ child1 ✓
// │ └─ grandchild ✗ (only one level opened)
// └─ child2 ✓
func TestGetSelectedSpans_ExpandedSelectedSpan(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("selected", "svc",
mkSpan("child1", "svc", mkSpan("grandchild", "svc")),
mkSpan("child2", "svc"),
),
)
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{}, "selected", []*model.Span{root}, spanMap, true)
// root and selected are on the auto-uncollapsed path; child1/child2 are direct
// children of the expanded selected span; grandchild stays hidden.
assert.Equal(t, []string{"root", "selected", "child1", "child2"}, spanIDs(spans))
}
// Multiple spans uncollapsed simultaneously: children of all uncollapsed spans
// are visible at once.
//
// root
// ├─ childA (uncollapsed) → grandchildA ✓
// └─ childB (uncollapsed) → grandchildB ✓
func TestGetSelectedSpans_MultipleUncollapsed(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("childA", "svc", mkSpan("grandchildA", "svc")),
mkSpan("childB", "svc", mkSpan("grandchildB", "svc")),
)
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{"root", "childA", "childB"}, "root", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root", "childA", "grandchildA", "childB", "grandchildB"}, spanIDs(spans))
}
// Collapsing a span with other uncollapsed spans
//
// root
// ├─ childA (previously expanded — in uncollapsedSpans)
// │ ├─ grandchild1 ✓
// │ │ └─ greatGrandchild ✗ (grandchild1 not in uncollapsedSpans)
// │ └─ grandchild2 ✓
// └─ childB ← selected (not expanded)
func TestGetSelectedSpans_ManualUncollapse(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("childA", "svc",
mkSpan("grandchild1", "svc", mkSpan("greatGrandchild", "svc")),
mkSpan("grandchild2", "svc"),
),
mkSpan("childB", "svc"),
)
spanMap := buildSpanMap(root)
// childA was expanded in a previous interaction; childB is now selected without expanding
spans, _, _, _ := GetSelectedSpans([]string{"childA"}, "childB", []*model.Span{root}, spanMap, false)
// path to childB auto-uncollpases root → childA and childB appear; childA is in
// uncollapsedSpans so its children appear; greatGrandchild stays hidden.
assert.Equal(t, []string{"root", "childA", "grandchild1", "grandchild2", "childB"}, spanIDs(spans))
}
// A collapsed span hides all children.
func TestGetSelectedSpans_CollapsedSpan(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("child1", "svc"),
mkSpan("child2", "svc"),
)
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{}, "root", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root"}, spanIDs(spans))
}
// Selecting a span auto-uncollpases the path from root to that span so it is visible.
//
// root → parent → selected
func TestGetSelectedSpans_PathToSelectedIsUncollapsed(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("parent", "svc",
mkSpan("selected", "svc"),
),
)
spanMap := buildSpanMap(root)
// no manually uncollapsed spans — path should still be opened
spans, _, _, _ := GetSelectedSpans([]string{}, "selected", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root", "parent", "selected"}, spanIDs(spans))
}
// The path-to-selected spans are returned in updatedUncollapsedSpans.
func TestGetSelectedSpans_PathReturnedInUncollapsed(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("parent", "svc",
mkSpan("selected", "svc"),
),
)
spanMap := buildSpanMap(root)
spans, uncollapsed, _, _ := GetSelectedSpans([]string{}, "selected", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root", "parent"}, uncollapsed)
assert.Equal(t, []string{"root", "parent", "selected"}, spanIDs(spans))
}
// Siblings of ancestors are rendered as collapsed nodes but their subtrees
// must NOT be expanded.
//
// root
// ├─ unrelated → unrelated-child (✗)
// └─ parent → selected
func TestGetSelectedSpans_SiblingsNotExpanded(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("unrelated", "svc", mkSpan("unrelated-child", "svc")),
mkSpan("parent", "svc",
mkSpan("selected", "svc"),
),
)
spanMap := buildSpanMap(root)
spans, uncollapsed, _, _ := GetSelectedSpans([]string{}, "selected", []*model.Span{root}, spanMap, false)
// children of root sort alphabetically: parent < unrelated; unrelated-child stays hidden
assert.Equal(t, []string{"root", "parent", "selected", "unrelated"}, spanIDs(spans))
// only the path nodes are tracked as uncollapsed — unrelated is not
assert.Equal(t, []string{"root", "parent"}, uncollapsed)
}
// An unknown selectedSpanID must not panic; returns a window from index 0.
func TestGetSelectedSpans_UnknownSelectedSpan(t *testing.T) {
root := mkSpan("root", "svc", mkSpan("child", "svc"))
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{}, "nonexistent", []*model.Span{root}, spanMap, false)
assert.Equal(t, []string{"root"}, spanIDs(spans))
}
// Test to check if Level, HasChildren, HasSiblings, and SubTreeNodeCount are populated correctly.
//
// root level=0, hasChildren=true, hasSiblings=false, subTree=4
// child1 level=1, hasChildren=true, hasSiblings=true, subTree=2
// grandchild level=2, hasChildren=false, hasSiblings=false, subTree=1
// child2 level=1, hasChildren=false, hasSiblings=false, subTree=1
func TestGetSelectedSpans_SpanMetadata(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("child1", "svc", mkSpan("grandchild", "svc")),
mkSpan("child2", "svc"),
)
spanMap := buildSpanMap(root)
spans, _, _, _ := GetSelectedSpans([]string{"root", "child1"}, "root", []*model.Span{root}, spanMap, false)
byID := map[string]*model.Span{}
for _, s := range spans {
byID[s.SpanID] = s
}
assert.Equal(t, uint64(0), byID["root"].Level)
assert.Equal(t, uint64(1), byID["child1"].Level)
assert.Equal(t, uint64(1), byID["child2"].Level)
assert.Equal(t, uint64(2), byID["grandchild"].Level)
assert.True(t, byID["root"].HasChildren)
assert.True(t, byID["child1"].HasChildren)
assert.False(t, byID["child2"].HasChildren)
assert.False(t, byID["grandchild"].HasChildren)
assert.False(t, byID["root"].HasSiblings, "root has no siblings")
assert.True(t, byID["child1"].HasSiblings, "child1 has sibling child2")
assert.False(t, byID["child2"].HasSiblings, "child2 is the last child")
assert.False(t, byID["grandchild"].HasSiblings, "grandchild has no siblings")
assert.Equal(t, uint64(4), byID["root"].SubTreeNodeCount)
assert.Equal(t, uint64(2), byID["child1"].SubTreeNodeCount)
assert.Equal(t, uint64(1), byID["grandchild"].SubTreeNodeCount)
assert.Equal(t, uint64(1), byID["child2"].SubTreeNodeCount)
}
// If the selected span is already in uncollapsedSpans AND isSelectedSpanIDUnCollapsed=true,
func TestGetSelectedSpans_DuplicateInUncollapsed(t *testing.T) {
root := mkSpan("root", "svc",
mkSpan("selected", "svc", mkSpan("child", "svc")),
)
spanMap := buildSpanMap(root)
_, uncollapsed, _, _ := GetSelectedSpans(
[]string{"selected"}, // already present
"selected",
[]*model.Span{root}, spanMap,
true,
)
count := 0
for _, id := range uncollapsed {
if id == "selected" {
count++
}
}
assert.Equal(t, 1, count, "should appear once")
}
// makeChain builds a linear trace: span0 → span1 → … → span(n-1).
// All span IDs are "span0", "span1", … so the caller can reference them by index.
func makeChain(n int) (*model.Span, map[string]*model.Span, []string) {
spans := make([]*model.Span, n)
for i := n - 1; i >= 0; i-- {
if i == n-1 {
spans[i] = mkSpan(fmt.Sprintf("span%d", i), "svc")
} else {
spans[i] = mkSpan(fmt.Sprintf("span%d", i), "svc", spans[i+1])
}
}
uncollapsed := make([]string, n)
for i := range spans {
uncollapsed[i] = fmt.Sprintf("span%d", i)
}
return spans[0], buildSpanMap(spans[0]), uncollapsed
}
// The selected span is centred: 200 spans before it, 300 after (0.4 / 0.6 split).
func TestGetSelectedSpans_WindowCentredOnSelected(t *testing.T) {
root, spanMap, uncollapsed := makeChain(600)
spans, _, _, _ := GetSelectedSpans(uncollapsed, "span300", []*model.Span{root}, spanMap, false)
assert.Equal(t, 500, len(spans), "window should be 500 spans")
// window is [100, 600): span300 lands at position 200 (300 - 100)
assert.Equal(t, "span100", spans[0].SpanID, "window starts 200 before selected")
assert.Equal(t, "span300", spans[200].SpanID, "selected span at position 200 in window")
assert.Equal(t, "span599", spans[499].SpanID, "window ends 300 after selected")
}
// When the selected span is near the start, the window shifts right so no
// negative index is used — the result is still 500 spans.
func TestGetSelectedSpans_WindowShiftsAtStart(t *testing.T) {
root, spanMap, uncollapsed := makeChain(600)
spans, _, _, _ := GetSelectedSpans(uncollapsed, "span10", []*model.Span{root}, spanMap, false)
assert.Equal(t, 500, len(spans))
assert.Equal(t, "span0", spans[0].SpanID, "window clamped to start of trace")
assert.Equal(t, "span10", spans[10].SpanID, "selected span still in window")
}
func mkSpan(id, service string, children ...*model.Span) *model.Span {
return &model.Span{
SpanID: id,
ServiceName: service,
Name: id + "-op",
Children: children,
}
}
// spanIDs returns SpanIDs in order.
func spanIDs(spans []*model.Span) []string {
ids := make([]string, len(spans))
for i, s := range spans {
ids[i] = s.SpanID
}
return ids
}
// buildSpanMap indexes every span in a set of trees by SpanID.
func buildSpanMap(roots ...*model.Span) map[string]*model.Span {
m := map[string]*model.Span{}
var walk func(*model.Span)
walk = func(s *model.Span) {
m[s.SpanID] = s
for _, c := range s.Children {
walk(c)
}
}
for _, r := range roots {
walk(r)
}
return m
}

View File

@@ -2,7 +2,6 @@ package rules
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"sync"
@@ -191,24 +190,6 @@ func NewBaseRule(id string, orgID valuer.UUID, p *ruletypes.PostableRule, reader
return baseRule, nil
}
func (r *BaseRule) String() string {
ar := ruletypes.PostableRule{
AlertName: r.name,
RuleCondition: r.ruleCondition,
EvalWindow: r.evalWindow,
Labels: r.labels.Map(),
Annotations: r.annotations.Map(),
PreferredChannels: r.preferredChannels,
}
byt, err := json.Marshal(ar)
if err != nil {
return fmt.Sprintf("error marshaling alerting rule: %s", err.Error())
}
return string(byt)
}
func (r *BaseRule) matchType() ruletypes.MatchType {
if r.ruleCondition == nil {
return ruletypes.AtleastOnce

View File

@@ -698,7 +698,7 @@ func TestBaseRule_FilterNewSeries(t *testing.T) {
slog.Default(),
nil,
telemetryStore,
prometheustest.New(context.Background(), settings, prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore),
prometheustest.New(context.Background(), settings, prometheus.Config{}, telemetryStore),
"",
time.Second,
nil,

View File

@@ -1,239 +0,0 @@
package rules
import (
"context"
"encoding/json"
"fmt"
"time"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/query-service/model"
"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
"github.com/SigNoz/signoz/pkg/query-service/utils/times"
"github.com/SigNoz/signoz/pkg/query-service/utils/timestamp"
"github.com/SigNoz/signoz/pkg/types/ruletypes"
"github.com/SigNoz/signoz/pkg/units"
)
type EvalVectorOptions struct {
DeleteLabels []string
ExtraAnnotations func(ctx context.Context, ts time.Time, metric labels.Labels) []labels.Label
}
func (r *BaseRule) EvalVector(ctx context.Context, ts time.Time, res ruletypes.Vector, opts EvalVectorOptions) (int, error) {
prevState := r.State()
valueFormatter := units.FormatterFromUnit(r.Unit())
r.mtx.Lock()
defer r.mtx.Unlock()
resultFPs := map[uint64]struct{}{}
alerts := make(map[uint64]*ruletypes.Alert, len(res))
ruleReceivers := r.Threshold.GetRuleReceivers()
ruleReceiverMap := make(map[string][]string)
for _, value := range ruleReceivers {
ruleReceiverMap[value.Name] = value.Channels
}
for _, smpl := range res {
l := make(map[string]string, len(smpl.Metric))
for _, lbl := range smpl.Metric {
l[lbl.Name] = lbl.Value
}
r.logger.DebugContext(ctx, "alerting for series", "rule_name", r.Name(), "series", smpl)
value := valueFormatter.Format(smpl.V, r.Unit())
threshold := valueFormatter.Format(smpl.Target, smpl.TargetUnit)
r.logger.DebugContext(ctx, "Alert template data for rule", "rule_name", r.Name(), "formatter", valueFormatter.Name(), "value", value, "threshold", threshold)
tmplData := ruletypes.AlertTemplateData(l, value, threshold)
// Inject some convenience variables that are easier to remember for users
// who are not used to Go's templating system.
defs := "{{$labels := .Labels}}{{$value := .Value}}{{$threshold := .Threshold}}"
// utility function to apply go template on labels and annotations
expand := func(text string) string {
tmpl := ruletypes.NewTemplateExpander(
ctx,
defs+text,
"__alert_"+r.Name(),
tmplData,
times.Time(timestamp.FromTime(ts)),
nil,
)
result, err := tmpl.Expand()
if err != nil {
result = fmt.Sprintf("<error expanding template: %s>", err)
r.logger.ErrorContext(ctx, "Expanding alert template failed", "rule_name", r.Name(), errors.Attr(err), "data", tmplData)
}
return result
}
lb := labels.NewBuilder(smpl.Metric).Del(opts.DeleteLabels...).Del()
resultLabels := labels.NewBuilder(smpl.Metric).Del(opts.DeleteLabels...).Labels()
for name, value := range r.labels.Map() {
lb.Set(name, expand(value))
}
lb.Set(labels.AlertNameLabel, r.Name())
lb.Set(labels.AlertRuleIdLabel, r.ID())
lb.Set(labels.RuleSourceLabel, r.GeneratorURL())
annotations := make(labels.Labels, 0, len(r.annotations.Map()))
for name, value := range r.annotations.Map() {
annotations = append(annotations, labels.Label{Name: name, Value: expand(value)})
}
if smpl.IsMissing {
lb.Set(labels.AlertNameLabel, "[No data] "+r.Name())
lb.Set(labels.NoDataLabel, "true")
}
if opts.ExtraAnnotations != nil {
extra := opts.ExtraAnnotations(ctx, ts, smpl.Metric)
annotations = append(annotations, extra...)
}
lbs := lb.Labels()
h := lbs.Hash()
resultFPs[h] = struct{}{}
if _, ok := alerts[h]; ok {
r.logger.ErrorContext(ctx, "the alert query returns duplicate records", "rule_id", r.ID(), "alert", alerts[h])
err := fmt.Errorf("duplicate alert found, vector contains metrics with the same labelset after applying alert labels")
// We have already acquired the lock above hence using SetHealth and SetLastError will deadlock.
r.health = ruletypes.HealthBad
r.lastError = err
return 0, err
}
alerts[h] = &ruletypes.Alert{
Labels: lbs,
QueryResultLables: resultLabels,
Annotations: annotations,
ActiveAt: ts,
State: model.StatePending,
Value: smpl.V,
GeneratorURL: r.GeneratorURL(),
Receivers: ruleReceiverMap[lbs.Map()[ruletypes.LabelThresholdName]],
Missing: smpl.IsMissing,
IsRecovering: smpl.IsRecovering,
}
}
r.logger.InfoContext(ctx, "number of alerts found", "rule_name", r.Name(), "alerts_count", len(alerts))
// alerts[h] is ready, add or update active list now
for h, a := range alerts {
// Check whether we already have alerting state for the identifying label set.
// Update the last value and annotations if so, create a new alert entry otherwise.
if alert, ok := r.Active[h]; ok && alert.State != model.StateInactive {
alert.Value = a.Value
alert.Annotations = a.Annotations
// Update the recovering and missing state of existing alert
alert.IsRecovering = a.IsRecovering
alert.Missing = a.Missing
if v, ok := alert.Labels.Map()[ruletypes.LabelThresholdName]; ok {
alert.Receivers = ruleReceiverMap[v]
}
continue
}
r.Active[h] = a
}
itemsToAdd := []model.RuleStateHistory{}
// Check if any pending alerts should be removed or fire now. Write out alert timeseries.
for fp, a := range r.Active {
labelsJSON, err := json.Marshal(a.QueryResultLables)
if err != nil {
r.logger.ErrorContext(ctx, "error marshaling labels", errors.Attr(err), "rule_name", r.Name(), "labels", a.Labels)
}
if _, ok := resultFPs[fp]; !ok {
// If the alert was previously firing, keep it around for a given
// retention time so it is reported as resolved to the AlertManager.
if a.State == model.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
delete(r.Active, fp)
}
if a.State != model.StateInactive {
r.logger.DebugContext(ctx, "converting firing alert to inActive", "name", r.Name())
a.State = model.StateInactive
a.ResolvedAt = ts
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: model.StateInactive,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
continue
}
if a.State == model.StatePending && ts.Sub(a.ActiveAt) >= r.holdDuration.Duration() {
r.logger.DebugContext(ctx, "converting pending alert to firing", "name", r.Name())
a.State = model.StateFiring
a.FiredAt = ts
state := model.StateFiring
if a.Missing {
state = model.StateNoData
}
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
// We need to change firing alert to recovering if the returned sample meets recovery threshold
changeFiringToRecovering := a.State == model.StateFiring && a.IsRecovering
// We need to change recovering alerts to firing if the returned sample meets target threshold
changeRecoveringToFiring := a.State == model.StateRecovering && !a.IsRecovering && !a.Missing
// in any of the above case we need to update the status of alert
if changeFiringToRecovering || changeRecoveringToFiring {
state := model.StateRecovering
if changeRecoveringToFiring {
state = model.StateFiring
}
a.State = state
r.logger.DebugContext(ctx, "converting alert state", "name", r.Name(), "state", state)
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
}
currentState := r.State()
overallStateChanged := currentState != prevState
for idx, item := range itemsToAdd {
item.OverallStateChanged = overallStateChanged
item.OverallState = currentState
itemsToAdd[idx] = item
}
r.RecordRuleStateHistory(ctx, prevState, currentState, itemsToAdd)
r.health = ruletypes.HealthGood
r.lastError = nil
return len(r.Active), nil
}

View File

@@ -253,7 +253,7 @@ func TestManager_TestNotification_SendUnmatched_PromRule(t *testing.T) {
WillReturnRows(samplesRows)
// Create Prometheus provider for this test
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, store)
promProvider = prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, store)
},
ManagerOptionsHook: func(opts *ManagerOptions) {
// Set Prometheus provider for PromQL queries

View File

@@ -99,7 +99,7 @@ func NewTestManager(t *testing.T, testOpts *TestManagerOptions) *Manager {
options := clickhouseReader.NewOptions("", "", "archiveNamespace")
providerSettings := instrumentationtest.New().ToProviderSettings()
prometheus := prometheustest.New(context.Background(), providerSettings, prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
prometheus := prometheustest.New(context.Background(), providerSettings, prometheus.Config{}, telemetryStore)
reader := clickhouseReader.NewReader(
instrumentationtest.New().Logger(),
nil,

View File

@@ -2,20 +2,25 @@ package rules
import (
"context"
"encoding/json"
"fmt"
"log/slog"
"time"
plabels "github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/promql"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/prometheus"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
qslabels "github.com/SigNoz/signoz/pkg/query-service/utils/labels"
"github.com/SigNoz/signoz/pkg/query-service/utils/times"
"github.com/SigNoz/signoz/pkg/query-service/utils/timestamp"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/ruletypes"
"github.com/SigNoz/signoz/pkg/units"
"github.com/SigNoz/signoz/pkg/valuer"
)
@@ -181,16 +186,232 @@ func (r *PromRule) buildAndRunQuery(ctx context.Context, ts time.Time) (ruletype
}
func (r *PromRule) Eval(ctx context.Context, ts time.Time) (int, error) {
prevState := r.State()
valueFormatter := units.FormatterFromUnit(r.Unit())
// prepare query, run query get data and filter the data based on the threshold
res, err := r.buildAndRunQuery(ctx, ts)
results, err := r.buildAndRunQuery(ctx, ts)
if err != nil {
return 0, err
}
opts := EvalVectorOptions{
DeleteLabels: []string{labels.MetricNameLabel},
r.mtx.Lock()
defer r.mtx.Unlock()
resultFPs := map[uint64]struct{}{}
alerts := make(map[uint64]*ruletypes.Alert, len(results))
ruleReceivers := r.Threshold.GetRuleReceivers()
ruleReceiverMap := make(map[string][]string)
for _, value := range ruleReceivers {
ruleReceiverMap[value.Name] = value.Channels
}
return r.EvalVector(ctx, ts, res, opts)
for _, result := range results {
l := make(map[string]string, len(result.Metric))
for _, lbl := range result.Metric {
l[lbl.Name] = lbl.Value
}
r.logger.DebugContext(ctx, "alerting for series", "rule_name", r.Name(), "series", result)
threshold := valueFormatter.Format(result.Target, result.TargetUnit)
tmplData := ruletypes.AlertTemplateData(l, valueFormatter.Format(result.V, r.Unit()), threshold)
// Inject some convenience variables that are easier to remember for users
// who are not used to Go's templating system.
defs := "{{$labels := .Labels}}{{$value := .Value}}{{$threshold := .Threshold}}"
expand := func(text string) string {
tmpl := ruletypes.NewTemplateExpander(
ctx,
defs+text,
"__alert_"+r.Name(),
tmplData,
times.Time(timestamp.FromTime(ts)),
nil,
)
result, err := tmpl.Expand()
if err != nil {
result = fmt.Sprintf("<error expanding template: %s>", err)
r.logger.WarnContext(ctx, "Expanding alert template failed", "rule_name", r.Name(), errors.Attr(err), "data", tmplData)
}
return result
}
lb := qslabels.NewBuilder(result.Metric).Del(qslabels.MetricNameLabel)
resultLabels := qslabels.NewBuilder(result.Metric).Del(qslabels.MetricNameLabel).Labels()
for name, value := range r.labels.Map() {
lb.Set(name, expand(value))
}
lb.Set(qslabels.AlertNameLabel, r.Name())
lb.Set(qslabels.AlertRuleIdLabel, r.ID())
lb.Set(qslabels.RuleSourceLabel, r.GeneratorURL())
annotations := make(qslabels.Labels, 0, len(r.annotations.Map()))
for name, value := range r.annotations.Map() {
annotations = append(annotations, qslabels.Label{Name: name, Value: expand(value)})
}
if result.IsMissing {
lb.Set(qslabels.AlertNameLabel, "[No data] "+r.Name())
lb.Set(qslabels.NoDataLabel, "true")
}
lbs := lb.Labels()
h := lbs.Hash()
resultFPs[h] = struct{}{}
if _, ok := alerts[h]; ok {
err = fmt.Errorf("vector contains metrics with the same labelset after applying alert labels")
// We have already acquired the lock above hence using SetHealth and
// SetLastError will deadlock.
r.health = ruletypes.HealthBad
r.lastError = err
return 0, err
}
alerts[h] = &ruletypes.Alert{
Labels: lbs,
QueryResultLables: resultLabels,
Annotations: annotations,
ActiveAt: ts,
State: model.StatePending,
Value: result.V,
GeneratorURL: r.GeneratorURL(),
Receivers: ruleReceiverMap[lbs.Map()[ruletypes.LabelThresholdName]],
Missing: result.IsMissing,
IsRecovering: result.IsRecovering,
}
}
r.logger.InfoContext(ctx, "number of alerts found", "rule_name", r.Name(), "alerts_count", len(alerts))
// alerts[h] is ready, add or update active list now
for h, a := range alerts {
// Check whether we already have alerting state for the identifying label set.
// Update the last value and annotations if so, create a new alert entry otherwise.
if alert, ok := r.Active[h]; ok && alert.State != model.StateInactive {
alert.Value = a.Value
alert.Annotations = a.Annotations
// Update the recovering and missing state of existing alert
alert.IsRecovering = a.IsRecovering
alert.Missing = a.Missing
if v, ok := alert.Labels.Map()[ruletypes.LabelThresholdName]; ok {
alert.Receivers = ruleReceiverMap[v]
}
continue
}
r.Active[h] = a
}
itemsToAdd := []model.RuleStateHistory{}
// Check if any pending alerts should be removed or fire now. Write out alert timeseries.
for fp, a := range r.Active {
labelsJSON, err := json.Marshal(a.QueryResultLables)
if err != nil {
r.logger.ErrorContext(ctx, "error marshaling labels", errors.Attr(err), "rule_name", r.Name())
}
if _, ok := resultFPs[fp]; !ok {
// If the alert was previously firing, keep it around for a given
// retention time so it is reported as resolved to the AlertManager.
if a.State == model.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
delete(r.Active, fp)
}
if a.State != model.StateInactive {
a.State = model.StateInactive
a.ResolvedAt = ts
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: model.StateInactive,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
})
}
continue
}
if a.State == model.StatePending && ts.Sub(a.ActiveAt) >= r.holdDuration.Duration() {
a.State = model.StateFiring
a.FiredAt = ts
state := model.StateFiring
if a.Missing {
state = model.StateNoData
}
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
// We need to change firing alert to recovering if the returned sample meets recovery threshold
changeAlertingToRecovering := a.State == model.StateFiring && a.IsRecovering
// We need to change recovering alerts to firing if the returned sample meets target threshold
changeRecoveringToFiring := a.State == model.StateRecovering && !a.IsRecovering && !a.Missing
// in any of the above case we need to update the status of alert
if changeAlertingToRecovering || changeRecoveringToFiring {
state := model.StateRecovering
if changeRecoveringToFiring {
state = model.StateFiring
}
a.State = state
r.logger.DebugContext(ctx, "converting alert state", "name", r.Name(), "state", state)
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
}
r.health = ruletypes.HealthGood
r.lastError = err
currentState := r.State()
overallStateChanged := currentState != prevState
for idx, item := range itemsToAdd {
item.OverallStateChanged = overallStateChanged
item.OverallState = currentState
itemsToAdd[idx] = item
}
r.RecordRuleStateHistory(ctx, prevState, currentState, itemsToAdd)
return len(r.Active), nil
}
func (r *PromRule) String() string {
ar := ruletypes.PostableRule{
AlertName: r.name,
RuleCondition: r.ruleCondition,
EvalWindow: r.evalWindow,
Labels: r.labels.Map(),
Annotations: r.annotations.Map(),
PreferredChannels: r.preferredChannels,
}
byt, err := json.Marshal(ar)
if err != nil {
return fmt.Sprintf("error marshaling alerting rule: %s", err.Error())
}
return string(byt)
}
func (r *PromRule) RunAlertQuery(ctx context.Context, qs string, start, end time.Time, interval time.Duration) (promql.Matrix, error) {
@@ -242,7 +463,7 @@ func toCommonSeries(series promql.Series) v3.Series {
Points: make([]v3.Point, 0),
}
series.Metric.Range(func(lbl plabels.Label) {
series.Metric.Range(func(lbl labels.Label) {
commonSeries.Labels[lbl.Name] = lbl.Value
commonSeries.LabelsArray = append(commonSeries.LabelsArray, map[string]string{
lbl.Name: lbl.Value,

View File

@@ -940,7 +940,7 @@ func TestPromRuleUnitCombinations(t *testing.T) {
).
WillReturnRows(samplesRows)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore)
postableRule.RuleCondition.CompareOp = ruletypes.CompareOp(c.compareOp)
postableRule.RuleCondition.MatchType = ruletypes.MatchType(c.matchType)
@@ -1061,7 +1061,7 @@ func _Enable_this_after_9146_issue_fix_is_merged_TestPromRuleNoData(t *testing.T
WithArgs("test_metric", "__name__", "test_metric").
WillReturnRows(fingerprintRows)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore)
var target float64 = 0
postableRule.RuleCondition.Thresholds = &ruletypes.RuleThresholdData{
@@ -1281,7 +1281,7 @@ func TestMultipleThresholdPromRule(t *testing.T) {
).
WillReturnRows(samplesRows)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
promProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore)
postableRule.RuleCondition.CompareOp = ruletypes.CompareOp(c.compareOp)
postableRule.RuleCondition.MatchType = ruletypes.MatchType(c.matchType)
@@ -1441,7 +1441,7 @@ func TestPromRule_NoData(t *testing.T) {
promProvider := prometheustest.New(
context.Background(),
instrumentationtest.New().ToProviderSettings(),
prometheus.Config{Timeout: 2 * time.Minute},
prometheus.Config{},
telemetryStore,
)
defer func() {
@@ -1590,7 +1590,7 @@ func TestPromRule_NoData_AbsentFor(t *testing.T) {
promProvider := prometheustest.New(
context.Background(),
instrumentationtest.New().ToProviderSettings(),
prometheus.Config{Timeout: 2 * time.Minute},
prometheus.Config{},
telemetryStore,
)
defer func() {
@@ -1748,7 +1748,7 @@ func TestPromRuleEval_RequireMinPoints(t *testing.T) {
promProvider := prometheustest.New(
context.Background(),
instrumentationtest.New().ToProviderSettings(),
prometheus.Config{Timeout: 2 * time.Minute, LookbackDelta: lookBackDelta},
prometheus.Config{LookbackDelta: lookBackDelta},
telemetryStore,
)
defer func() {

View File

@@ -3,6 +3,7 @@ package rules
import (
"bytes"
"context"
"encoding/json"
"fmt"
"log/slog"
"math"
@@ -23,16 +24,21 @@ import (
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
querierV5 "github.com/SigNoz/signoz/pkg/querier"
logsv3 "github.com/SigNoz/signoz/pkg/query-service/app/logs/v3"
"github.com/SigNoz/signoz/pkg/query-service/app/querier"
querierV2 "github.com/SigNoz/signoz/pkg/query-service/app/querier/v2"
"github.com/SigNoz/signoz/pkg/query-service/app/queryBuilder"
tracesV4 "github.com/SigNoz/signoz/pkg/query-service/app/traces/v4"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/utils/labels"
querytemplate "github.com/SigNoz/signoz/pkg/query-service/utils/queryTemplate"
"github.com/SigNoz/signoz/pkg/query-service/utils/times"
"github.com/SigNoz/signoz/pkg/query-service/utils/timestamp"
logsv3 "github.com/SigNoz/signoz/pkg/query-service/app/logs/v3"
tracesV4 "github.com/SigNoz/signoz/pkg/query-service/app/traces/v4"
"github.com/SigNoz/signoz/pkg/units"
querierV5 "github.com/SigNoz/signoz/pkg/querier"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
)
@@ -277,7 +283,7 @@ func (r *ThresholdRule) prepareLinksToTraces(ctx context.Context, ts time.Time,
return contextlinks.PrepareLinksToTraces(start, end, filterItems)
}
func (r *ThresholdRule) prepareQueryRangeV5(ctx context.Context, ts time.Time) *qbtypes.QueryRangeRequest {
func (r *ThresholdRule) prepareQueryRangeV5(ctx context.Context, ts time.Time) (*qbtypes.QueryRangeRequest, error) {
r.logger.InfoContext(
ctx, "prepare query range request v5", "ts", ts.UnixMilli(), "eval_window", r.evalWindow.Milliseconds(), "eval_delay", r.evalDelay.Milliseconds(),
)
@@ -296,13 +302,16 @@ func (r *ThresholdRule) prepareQueryRangeV5(ctx context.Context, ts time.Time) *
}
req.CompositeQuery.Queries = make([]qbtypes.QueryEnvelope, len(r.Condition().CompositeQuery.Queries))
copy(req.CompositeQuery.Queries, r.Condition().CompositeQuery.Queries)
return req
return req, nil
}
func (r *ThresholdRule) prepareLinksToLogsV5(ctx context.Context, ts time.Time, lbls labels.Labels) string {
selectedQuery := r.GetSelectedQuery()
qr := r.prepareQueryRangeV5(ctx, ts)
qr, err := r.prepareQueryRangeV5(ctx, ts)
if err != nil {
return ""
}
start := time.UnixMilli(int64(qr.Start))
end := time.UnixMilli(int64(qr.End))
@@ -339,7 +348,10 @@ func (r *ThresholdRule) prepareLinksToLogsV5(ctx context.Context, ts time.Time,
func (r *ThresholdRule) prepareLinksToTracesV5(ctx context.Context, ts time.Time, lbls labels.Labels) string {
selectedQuery := r.GetSelectedQuery()
qr := r.prepareQueryRangeV5(ctx, ts)
qr, err := r.prepareQueryRangeV5(ctx, ts)
if err != nil {
return ""
}
start := time.UnixMilli(int64(qr.Start))
end := time.UnixMilli(int64(qr.End))
@@ -488,7 +500,10 @@ func (r *ThresholdRule) buildAndRunQuery(ctx context.Context, orgID valuer.UUID,
}
func (r *ThresholdRule) buildAndRunQueryV5(ctx context.Context, orgID valuer.UUID, ts time.Time) (ruletypes.Vector, error) {
params := r.prepareQueryRangeV5(ctx, ts)
params, err := r.prepareQueryRangeV5(ctx, ts)
if err != nil {
return nil, err
}
var results []*v3.Result
@@ -565,8 +580,13 @@ func (r *ThresholdRule) buildAndRunQueryV5(ctx context.Context, orgID valuer.UUI
}
func (r *ThresholdRule) Eval(ctx context.Context, ts time.Time) (int, error) {
prevState := r.State()
valueFormatter := units.FormatterFromUnit(r.Unit())
var res ruletypes.Vector
var err error
if r.version == "v5" {
r.logger.InfoContext(ctx, "running v5 query")
res, err = r.buildAndRunQueryV5(ctx, r.orgID, ts)
@@ -574,35 +594,247 @@ func (r *ThresholdRule) Eval(ctx context.Context, ts time.Time) (int, error) {
r.logger.InfoContext(ctx, "running v4 query")
res, err = r.buildAndRunQuery(ctx, r.orgID, ts)
}
if err != nil {
return 0, err
}
opts := EvalVectorOptions{
DeleteLabels: []string{labels.MetricNameLabel, labels.TemporalityLabel},
ExtraAnnotations: func(ctx context.Context, ts time.Time, smpl labels.Labels) []labels.Label {
host := r.hostFromSource()
if host == "" {
return nil
}
r.mtx.Lock()
defer r.mtx.Unlock()
//// Links with timestamps should go in annotations since labels
//// is used alert grouping, and we want to group alerts with the same
//// label set, but different timestamps, together.
switch r.typ { // DIFF
case ruletypes.AlertTypeTraces:
if link := r.prepareLinksToTraces(ctx, ts, smpl); link != "" {
r.logger.InfoContext(ctx, "adding traces link to annotations", "link", fmt.Sprintf("%s/traces-explorer?%s", host, link))
return []labels.Label{{Name: "related_traces", Value: fmt.Sprintf("%s/traces-explorer?%s", host, link)}}
}
case ruletypes.AlertTypeLogs:
if link := r.prepareLinksToLogs(ctx, ts, smpl); link != "" {
r.logger.InfoContext(ctx, "adding logs link to annotations", "link", fmt.Sprintf("%s/logs/logs-explorer?%s", host, link))
return []labels.Label{{Name: "related_logs", Value: fmt.Sprintf("%s/logs/logs-explorer?%s", host, link)}}
}
}
return nil
},
resultFPs := map[uint64]struct{}{}
alerts := make(map[uint64]*ruletypes.Alert, len(res))
ruleReceivers := r.Threshold.GetRuleReceivers()
ruleReceiverMap := make(map[string][]string)
for _, value := range ruleReceivers {
ruleReceiverMap[value.Name] = value.Channels
}
return r.EvalVector(ctx, ts, res, opts)
for _, smpl := range res {
l := make(map[string]string, len(smpl.Metric))
for _, lbl := range smpl.Metric {
l[lbl.Name] = lbl.Value
}
value := valueFormatter.Format(smpl.V, r.Unit())
// todo(aniket): handle different threshold
threshold := valueFormatter.Format(smpl.Target, smpl.TargetUnit)
r.logger.DebugContext(ctx, "Alert template data for rule", "rule_name", r.Name(), "formatter", valueFormatter.Name(), "value", value, "threshold", threshold)
tmplData := ruletypes.AlertTemplateData(l, value, threshold)
// Inject some convenience variables that are easier to remember for users
// who are not used to Go's templating system.
defs := "{{$labels := .Labels}}{{$value := .Value}}{{$threshold := .Threshold}}"
// utility function to apply go template on labels and annotations
expand := func(text string) string {
tmpl := ruletypes.NewTemplateExpander(
ctx,
defs+text,
"__alert_"+r.Name(),
tmplData,
times.Time(timestamp.FromTime(ts)),
nil,
)
result, err := tmpl.Expand()
if err != nil {
result = fmt.Sprintf("<error expanding template: %s>", err)
r.logger.ErrorContext(ctx, "Expanding alert template failed", errors.Attr(err), "data", tmplData)
}
return result
}
lb := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel)
resultLabels := labels.NewBuilder(smpl.Metric).Del(labels.MetricNameLabel).Del(labels.TemporalityLabel).Labels()
for name, value := range r.labels.Map() {
lb.Set(name, expand(value))
}
lb.Set(labels.AlertNameLabel, r.Name())
lb.Set(labels.AlertRuleIdLabel, r.ID())
lb.Set(labels.RuleSourceLabel, r.GeneratorURL())
annotations := make(labels.Labels, 0, len(r.annotations.Map()))
for name, value := range r.annotations.Map() {
annotations = append(annotations, labels.Label{Name: name, Value: expand(value)})
}
if smpl.IsMissing {
lb.Set(labels.AlertNameLabel, "[No data] "+r.Name())
lb.Set(labels.NoDataLabel, "true")
}
// Links with timestamps should go in annotations since labels
// is used alert grouping, and we want to group alerts with the same
// label set, but different timestamps, together.
switch r.typ {
case ruletypes.AlertTypeTraces:
link := r.prepareLinksToTraces(ctx, ts, smpl.Metric)
if link != "" && r.hostFromSource() != "" {
r.logger.InfoContext(ctx, "adding traces link to annotations", "link", fmt.Sprintf("%s/traces-explorer?%s", r.hostFromSource(), link))
annotations = append(annotations, labels.Label{Name: "related_traces", Value: fmt.Sprintf("%s/traces-explorer?%s", r.hostFromSource(), link)})
}
case ruletypes.AlertTypeLogs:
link := r.prepareLinksToLogs(ctx, ts, smpl.Metric)
if link != "" && r.hostFromSource() != "" {
r.logger.InfoContext(ctx, "adding logs link to annotations", "link", fmt.Sprintf("%s/logs/logs-explorer?%s", r.hostFromSource(), link))
annotations = append(annotations, labels.Label{Name: "related_logs", Value: fmt.Sprintf("%s/logs/logs-explorer?%s", r.hostFromSource(), link)})
}
}
lbs := lb.Labels()
h := lbs.Hash()
resultFPs[h] = struct{}{}
if _, ok := alerts[h]; ok {
return 0, fmt.Errorf("duplicate alert found, vector contains metrics with the same labelset after applying alert labels")
}
alerts[h] = &ruletypes.Alert{
Labels: lbs,
QueryResultLables: resultLabels,
Annotations: annotations,
ActiveAt: ts,
State: model.StatePending,
Value: smpl.V,
GeneratorURL: r.GeneratorURL(),
Receivers: ruleReceiverMap[lbs.Map()[ruletypes.LabelThresholdName]],
Missing: smpl.IsMissing,
IsRecovering: smpl.IsRecovering,
}
}
r.logger.InfoContext(ctx, "number of alerts found", "rule_name", r.Name(), "alerts_count", len(alerts))
// alerts[h] is ready, add or update active list now
for h, a := range alerts {
// Check whether we already have alerting state for the identifying label set.
// Update the last value and annotations if so, create a new alert entry otherwise.
if alert, ok := r.Active[h]; ok && alert.State != model.StateInactive {
alert.Value = a.Value
alert.Annotations = a.Annotations
// Update the recovering and missing state of existing alert
alert.IsRecovering = a.IsRecovering
alert.Missing = a.Missing
if v, ok := alert.Labels.Map()[ruletypes.LabelThresholdName]; ok {
alert.Receivers = ruleReceiverMap[v]
}
continue
}
r.Active[h] = a
}
itemsToAdd := []model.RuleStateHistory{}
// Check if any pending alerts should be removed or fire now. Write out alert timeseries.
for fp, a := range r.Active {
labelsJSON, err := json.Marshal(a.QueryResultLables)
if err != nil {
r.logger.ErrorContext(ctx, "error marshaling labels", errors.Attr(err), "labels", a.Labels)
}
if _, ok := resultFPs[fp]; !ok {
// If the alert was previously firing, keep it around for a given
// retention time so it is reported as resolved to the AlertManager.
if a.State == model.StatePending || (!a.ResolvedAt.IsZero() && ts.Sub(a.ResolvedAt) > ruletypes.ResolvedRetention) {
delete(r.Active, fp)
}
if a.State != model.StateInactive {
r.logger.DebugContext(ctx, "converting firing alert to inActive", "name", r.Name())
a.State = model.StateInactive
a.ResolvedAt = ts
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: model.StateInactive,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
continue
}
if a.State == model.StatePending && ts.Sub(a.ActiveAt) >= r.holdDuration.Duration() {
r.logger.DebugContext(ctx, "converting pending alert to firing", "name", r.Name())
a.State = model.StateFiring
a.FiredAt = ts
state := model.StateFiring
if a.Missing {
state = model.StateNoData
}
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
// We need to change firing alert to recovering if the returned sample meets recovery threshold
changeAlertingToRecovering := a.State == model.StateFiring && a.IsRecovering
// We need to change recovering alerts to firing if the returned sample meets target threshold
changeRecoveringToFiring := a.State == model.StateRecovering && !a.IsRecovering && !a.Missing
// in any of the above case we need to update the status of alert
if changeAlertingToRecovering || changeRecoveringToFiring {
state := model.StateRecovering
if changeRecoveringToFiring {
state = model.StateFiring
}
a.State = state
r.logger.DebugContext(ctx, "converting alert state", "name", r.Name(), "state", state)
itemsToAdd = append(itemsToAdd, model.RuleStateHistory{
RuleID: r.ID(),
RuleName: r.Name(),
State: state,
StateChanged: true,
UnixMilli: ts.UnixMilli(),
Labels: model.LabelsString(labelsJSON),
Fingerprint: a.QueryResultLables.Hash(),
Value: a.Value,
})
}
}
currentState := r.State()
overallStateChanged := currentState != prevState
for idx, item := range itemsToAdd {
item.OverallStateChanged = overallStateChanged
item.OverallState = currentState
itemsToAdd[idx] = item
}
r.RecordRuleStateHistory(ctx, prevState, currentState, itemsToAdd)
r.health = ruletypes.HealthGood
r.lastError = err
return len(r.Active), nil
}
func (r *ThresholdRule) String() string {
ar := ruletypes.PostableRule{
AlertName: r.name,
RuleCondition: r.ruleCondition,
EvalWindow: r.evalWindow,
Labels: r.labels.Map(),
Annotations: r.annotations.Map(),
PreferredChannels: r.preferredChannels,
}
byt, err := json.Marshal(ar)
if err != nil {
return fmt.Sprintf("error marshaling alerting rule: %s", err.Error())
}
return string(byt)
}

View File

@@ -779,7 +779,7 @@ func TestThresholdRuleUnitCombinations(t *testing.T) {
},
)
require.NoError(t, err)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore), "", time.Duration(time.Second), nil, readerCache, options)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore), "", time.Duration(time.Second), nil, readerCache, options)
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, reader, nil, logger)
rule.TemporalityMap = map[string]map[v3.Temporality]bool{
"signoz_calls_total": {
@@ -894,7 +894,7 @@ func TestThresholdRuleNoData(t *testing.T) {
)
assert.NoError(t, err)
options := clickhouseReader.NewOptions("", "", "archiveNamespace")
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore), "", time.Duration(time.Second), nil, readerCache, options)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore), "", time.Duration(time.Second), nil, readerCache, options)
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, reader, nil, logger)
rule.TemporalityMap = map[string]map[v3.Temporality]bool{
@@ -1014,7 +1014,7 @@ func TestThresholdRuleTracesLink(t *testing.T) {
}
options := clickhouseReader.NewOptions("", "", "archiveNamespace")
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore), "", time.Duration(time.Second), nil, nil, options)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore), "", time.Duration(time.Second), nil, nil, options)
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, reader, nil, logger)
rule.TemporalityMap = map[string]map[v3.Temporality]bool{
@@ -1151,7 +1151,7 @@ func TestThresholdRuleLogsLink(t *testing.T) {
}
options := clickhouseReader.NewOptions("", "", "archiveNamespace")
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore), "", time.Duration(time.Second), nil, nil, options)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore), "", time.Duration(time.Second), nil, nil, options)
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, reader, nil, logger)
rule.TemporalityMap = map[string]map[v3.Temporality]bool{
@@ -1418,7 +1418,7 @@ func TestMultipleThresholdRule(t *testing.T) {
},
)
require.NoError(t, err)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore), "", time.Second, nil, readerCache, options)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore), "", time.Second, nil, readerCache, options)
rule, err := NewThresholdRule("69", valuer.GenerateUUID(), &postableRule, reader, nil, logger)
rule.TemporalityMap = map[string]map[v3.Temporality]bool{
"signoz_calls_total": {
@@ -2220,7 +2220,7 @@ func TestThresholdEval_RequireMinPoints(t *testing.T) {
)
require.NoError(t, err)
prometheusProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{Timeout: 2 * time.Minute}, telemetryStore)
prometheusProvider := prometheustest.New(context.Background(), instrumentationtest.New().ToProviderSettings(), prometheus.Config{}, telemetryStore)
reader := clickhouseReader.NewReader(slog.Default(), nil, telemetryStore, prometheusProvider, "", time.Second, nil, readerCache, options)
rule, err := NewThresholdRule("some-id", valuer.GenerateUUID(), &postableRule, reader, nil, logger)

View File

@@ -12,8 +12,6 @@ import (
"github.com/SigNoz/signoz/pkg/licensing"
"github.com/SigNoz/signoz/pkg/modules/apdex"
"github.com/SigNoz/signoz/pkg/modules/apdex/implapdex"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration/implcloudintegration"
"github.com/SigNoz/signoz/pkg/modules/dashboard"
"github.com/SigNoz/signoz/pkg/modules/dashboard/impldashboard"
"github.com/SigNoz/signoz/pkg/modules/fields"
@@ -40,25 +38,23 @@ import (
)
type Handlers struct {
SavedView savedview.Handler
Apdex apdex.Handler
Dashboard dashboard.Handler
QuickFilter quickfilter.Handler
TraceFunnel tracefunnel.Handler
RawDataExport rawdataexport.Handler
SpanPercentile spanpercentile.Handler
Services services.Handler
MetricsExplorer metricsexplorer.Handler
Global global.Handler
FlaggerHandler flagger.Handler
GatewayHandler gateway.Handler
Fields fields.Handler
AuthzHandler authz.Handler
ZeusHandler zeus.Handler
QuerierHandler querier.Handler
ServiceAccountHandler serviceaccount.Handler
RegistryHandler factory.Handler
CloudIntegrationHandler cloudintegration.Handler
SavedView savedview.Handler
Apdex apdex.Handler
Dashboard dashboard.Handler
QuickFilter quickfilter.Handler
TraceFunnel tracefunnel.Handler
RawDataExport rawdataexport.Handler
SpanPercentile spanpercentile.Handler
Services services.Handler
MetricsExplorer metricsexplorer.Handler
Global global.Handler
FlaggerHandler flagger.Handler
GatewayHandler gateway.Handler
Fields fields.Handler
AuthzHandler authz.Handler
ZeusHandler zeus.Handler
QuerierHandler querier.Handler
ServiceAccountHandler serviceaccount.Handler
}
func NewHandlers(
@@ -73,27 +69,24 @@ func NewHandlers(
telemetryMetadataStore telemetrytypes.MetadataStore,
authz authz.AuthZ,
zeusService zeus.Zeus,
registryHandler factory.Handler,
) Handlers {
return Handlers{
SavedView: implsavedview.NewHandler(modules.SavedView),
Apdex: implapdex.NewHandler(modules.Apdex),
Dashboard: impldashboard.NewHandler(modules.Dashboard, providerSettings),
QuickFilter: implquickfilter.NewHandler(modules.QuickFilter),
TraceFunnel: impltracefunnel.NewHandler(modules.TraceFunnel),
RawDataExport: implrawdataexport.NewHandler(modules.RawDataExport),
Services: implservices.NewHandler(modules.Services),
MetricsExplorer: implmetricsexplorer.NewHandler(modules.MetricsExplorer),
SpanPercentile: implspanpercentile.NewHandler(modules.SpanPercentile),
Global: signozglobal.NewHandler(global),
FlaggerHandler: flagger.NewHandler(flaggerService),
GatewayHandler: gateway.NewHandler(gatewayService),
Fields: implfields.NewHandler(providerSettings, telemetryMetadataStore),
AuthzHandler: signozauthzapi.NewHandler(authz),
ZeusHandler: zeus.NewHandler(zeusService, licensing),
QuerierHandler: querierHandler,
ServiceAccountHandler: implserviceaccount.NewHandler(modules.ServiceAccount),
RegistryHandler: registryHandler,
CloudIntegrationHandler: implcloudintegration.NewHandler(),
SavedView: implsavedview.NewHandler(modules.SavedView),
Apdex: implapdex.NewHandler(modules.Apdex),
Dashboard: impldashboard.NewHandler(modules.Dashboard, providerSettings),
QuickFilter: implquickfilter.NewHandler(modules.QuickFilter),
TraceFunnel: impltracefunnel.NewHandler(modules.TraceFunnel),
RawDataExport: implrawdataexport.NewHandler(modules.RawDataExport),
Services: implservices.NewHandler(modules.Services),
MetricsExplorer: implmetricsexplorer.NewHandler(modules.MetricsExplorer),
SpanPercentile: implspanpercentile.NewHandler(modules.SpanPercentile),
Global: signozglobal.NewHandler(global),
FlaggerHandler: flagger.NewHandler(flaggerService),
GatewayHandler: gateway.NewHandler(gatewayService),
Fields: implfields.NewHandler(providerSettings, telemetryMetadataStore),
AuthzHandler: signozauthzapi.NewHandler(authz),
ZeusHandler: zeus.NewHandler(zeusService, licensing),
QuerierHandler: querierHandler,
ServiceAccountHandler: implserviceaccount.NewHandler(modules.ServiceAccount),
}
}

View File

@@ -10,7 +10,6 @@ import (
"github.com/SigNoz/signoz/pkg/alertmanager/nfmanager/nfmanagertest"
"github.com/SigNoz/signoz/pkg/alertmanager/signozalertmanager"
"github.com/SigNoz/signoz/pkg/emailing/emailingtest"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/factory/factorytest"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/instrumentation/instrumentationtest"
@@ -56,8 +55,7 @@ func TestNewHandlers(t *testing.T) {
modules := NewModules(sqlstore, tokenizer, emailing, providerSettings, orgGetter, alertmanager, nil, nil, nil, nil, nil, nil, nil, queryParser, Config{}, dashboardModule, userGetter, userRoleStore)
querierHandler := querier.NewHandler(providerSettings, nil, nil)
registryHandler := factory.NewHandler(nil)
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, registryHandler)
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil)
reflectVal := reflect.ValueOf(handlers)
for i := 0; i < reflectVal.NumField(); i++ {
f := reflectVal.Field(i)

View File

@@ -10,21 +10,18 @@ import (
"github.com/SigNoz/signoz/pkg/apiserver"
"github.com/SigNoz/signoz/pkg/apiserver/signozapiserver"
"github.com/SigNoz/signoz/pkg/authz"
"github.com/SigNoz/signoz/pkg/factory"
"github.com/SigNoz/signoz/pkg/flagger"
"github.com/SigNoz/signoz/pkg/gateway"
"github.com/SigNoz/signoz/pkg/global"
"github.com/SigNoz/signoz/pkg/http/handler"
"github.com/SigNoz/signoz/pkg/instrumentation"
"github.com/SigNoz/signoz/pkg/modules/authdomain"
"github.com/SigNoz/signoz/pkg/modules/cloudintegration"
"github.com/SigNoz/signoz/pkg/modules/dashboard"
"github.com/SigNoz/signoz/pkg/modules/fields"
"github.com/SigNoz/signoz/pkg/modules/metricsexplorer"
"github.com/SigNoz/signoz/pkg/modules/organization"
"github.com/SigNoz/signoz/pkg/modules/preference"
"github.com/SigNoz/signoz/pkg/modules/promote"
"github.com/SigNoz/signoz/pkg/modules/rawdataexport"
"github.com/SigNoz/signoz/pkg/modules/serviceaccount"
"github.com/SigNoz/signoz/pkg/modules/session"
"github.com/SigNoz/signoz/pkg/modules/user"
@@ -61,12 +58,9 @@ func NewOpenAPI(ctx context.Context, instrumentation instrumentation.Instrumenta
struct{ gateway.Handler }{},
struct{ fields.Handler }{},
struct{ authz.Handler }{},
struct{ rawdataexport.Handler }{},
struct{ zeus.Handler }{},
struct{ querier.Handler }{},
struct{ serviceaccount.Handler }{},
struct{ factory.Handler }{},
struct{ cloudintegration.Handler }{},
).New(ctx, instrumentation.ToProviderSettings(), apiserver.Config{})
if err != nil {
return nil, err

View File

@@ -274,12 +274,9 @@ func NewAPIServerProviderFactories(orgGetter organization.Getter, authz authz.Au
handlers.GatewayHandler,
handlers.Fields,
handlers.AuthzHandler,
handlers.RawDataExport,
handlers.ZeusHandler,
handlers.QuerierHandler,
handlers.ServiceAccountHandler,
handlers.RegistryHandler,
handlers.CloudIntegrationHandler,
),
)
}

View File

@@ -422,6 +422,21 @@ func New(
// Initialize the querier handler via callback (allows EE to decorate with anomaly detection)
querierHandler := querierHandlerCallback(providerSettings, querier, analytics)
// Initialize all handlers for the modules
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus)
// Initialize the API server
apiserver, err := factory.NewProviderFromNamedMap(
ctx,
providerSettings,
config.APIServer,
NewAPIServerProviderFactories(orgGetter, authz, modules, handlers),
"signoz",
)
if err != nil {
return nil, err
}
// Create a list of all stats collectors
statsCollectors := []statsreporter.StatsCollector{
alertmanager,
@@ -448,7 +463,6 @@ func New(
}
registry, err := factory.NewRegistry(
ctx,
instrumentation.Logger(),
factory.NewNamedService(factory.MustNewName("instrumentation"), instrumentation),
factory.NewNamedService(factory.MustNewName("pprof"), pprofService),
@@ -458,23 +472,7 @@ func New(
factory.NewNamedService(factory.MustNewName("statsreporter"), statsReporter),
factory.NewNamedService(factory.MustNewName("tokenizer"), tokenizer),
factory.NewNamedService(factory.MustNewName("authz"), authz),
factory.NewNamedService(factory.MustNewName("user"), userService, factory.MustNewName("authz")),
)
if err != nil {
return nil, err
}
// Initialize all handlers for the modules
registryHandler := factory.NewHandler(registry)
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus, registryHandler)
// Initialize the API server (after registry so it can access service health)
apiserverInstance, err := factory.NewProviderFromNamedMap(
ctx,
providerSettings,
config.APIServer,
NewAPIServerProviderFactories(orgGetter, authz, modules, handlers),
"signoz",
factory.NewNamedService(factory.MustNewName("user"), userService),
)
if err != nil {
return nil, err
@@ -492,7 +490,7 @@ func New(
Prometheus: prometheus,
Alertmanager: alertmanager,
Querier: querier,
APIServer: apiserverInstance,
APIServer: apiserver,
Zeus: zeus,
Licensing: licensing,
Emailing: emailing,

View File

@@ -165,7 +165,7 @@ func (provider *provider) Report(ctx context.Context) error {
continue
}
users, err := provider.userGetter.ListUsersByOrgID(ctx, org.ID)
users, err := provider.userGetter.ListByOrgID(ctx, org.ID)
if err != nil {
provider.settings.Logger().WarnContext(ctx, "failed to list users", errors.Attr(err), slog.Any("org_id", org.ID))
continue
@@ -178,7 +178,7 @@ func (provider *provider) Report(ctx context.Context) error {
}
for _, user := range users {
traits := types.NewTraitsFromUser(user)
traits := types.NewTraitsFromDeprecatedUser(user)
if maxLastObservedAt, ok := maxLastObservedAtPerUserID[user.ID]; ok {
traits["auth_token.last_observed_at.max.time"] = maxLastObservedAt.UTC()
traits["auth_token.last_observed_at.max.time_unix"] = maxLastObservedAt.Unix()

View File

@@ -111,6 +111,23 @@ func (b *traceQueryStatementBuilder) Build(
query = b.adjustKeys(ctx, keys, query, requestType)
// Check if filter contains trace_id(s) and optimize time range if needed
if query.Filter != nil && query.Filter.Expression != "" && b.telemetryStore != nil {
traceIDs, found := ExtractTraceIDsFromFilter(query.Filter.Expression)
if found && len(traceIDs) > 0 {
finder := NewTraceTimeRangeFinder(b.telemetryStore)
traceStart, traceEnd, ok := finder.GetTraceTimeRangeMulti(ctx, traceIDs)
if !ok {
b.logger.DebugContext(ctx, "failed to get trace time range", slog.Any("trace_ids", traceIDs))
} else if traceStart > 0 && traceEnd > 0 {
start = uint64(traceStart)
end = uint64(traceEnd)
b.logger.DebugContext(ctx, "optimized time range for traces", slog.Any("trace_ids", traceIDs), slog.Uint64("start", start), slog.Uint64("end", end))
}
}
}
// Create SQL builder
q := sqlbuilder.NewSelectBuilder()

View File

@@ -65,10 +65,10 @@ type StorableRole struct {
types.Identifiable
types.TimeAuditable
Name string `bun:"name,type:string" json:"name"`
Description string `bun:"description,type:string" json:"description"`
Type string `bun:"type,type:string" json:"type"`
OrgID string `bun:"org_id,type:string" json:"orgId"`
Name string `bun:"name,type:string"`
Description string `bun:"description,type:string"`
Type string `bun:"type,type:string"`
OrgID string `bun:"org_id,type:string"`
}
type Role struct {

Some files were not shown because too many files have changed in this diff Show More