mirror of
https://github.com/SigNoz/signoz.git
synced 2026-08-18 10:40:46 +01:00
Compare commits
1 Commits
main
...
issue-2988
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
56027c4f34 |
@@ -2,17 +2,14 @@ import { useTranslation } from 'react-i18next';
|
||||
import { Input } from '@signozhq/ui/input';
|
||||
import { Switch } from '@signozhq/ui/switch';
|
||||
import { Form, Select, Space } from 'antd';
|
||||
import { FeatureKeys } from 'constants/features';
|
||||
import { ModalFooterTitle } from 'container/PipelinePage/styles';
|
||||
import { useAppContext } from 'providers/App/App';
|
||||
import { ProcessorData } from 'types/api/pipeline/def';
|
||||
|
||||
import { formValidationRules } from '../config';
|
||||
import { ProcessorFormField } from './config';
|
||||
import { processorFields, ProcessorFormField } from './config';
|
||||
import CSVInput from './FormFields/CSVInput';
|
||||
import JsonFlattening from './FormFields/JsonFlattening';
|
||||
import { FormWrapper, PipelineIndexIcon, StyledSelect } from './styles';
|
||||
import { resolveProcessorFields } from './utils';
|
||||
|
||||
import './styles.scss';
|
||||
|
||||
@@ -136,23 +133,16 @@ function ProcessorForm({
|
||||
selectedProcessorData,
|
||||
isAdd,
|
||||
}: ProcessorFormProps): JSX.Element {
|
||||
const { featureFlags } = useAppContext();
|
||||
const isBodyJsonEnabled =
|
||||
featureFlags?.find((flag) => flag.name === FeatureKeys.USE_JSON_BODY)
|
||||
?.active || false;
|
||||
|
||||
return (
|
||||
<div className="processor-form-container">
|
||||
{resolveProcessorFields(processorType, isBodyJsonEnabled).map(
|
||||
(fieldData: ProcessorFormField) => (
|
||||
<ProcessorFieldInput
|
||||
key={fieldData.name + String(fieldData.initialValue)}
|
||||
fieldData={fieldData}
|
||||
selectedProcessorData={selectedProcessorData}
|
||||
isAdd={isAdd}
|
||||
/>
|
||||
),
|
||||
)}
|
||||
{processorFields[processorType]?.map((fieldData: ProcessorFormField) => (
|
||||
<ProcessorFieldInput
|
||||
key={fieldData.name + String(fieldData.initialValue)}
|
||||
fieldData={fieldData}
|
||||
selectedProcessorData={selectedProcessorData}
|
||||
isAdd={isAdd}
|
||||
/>
|
||||
))}
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
@@ -1,24 +0,0 @@
|
||||
import { processorFields, ProcessorFormField } from './config';
|
||||
|
||||
const BODY_PARSE_FROM = 'body';
|
||||
const JSON_BODY_PARSE_FROM = 'body.message';
|
||||
|
||||
// With use_json_body the collector normalizes every body into a map before user
|
||||
// operators run, so a parser pointed at `body` gets a map it cannot read and
|
||||
// silently extracts nothing. The log text lives at body.message.
|
||||
export function resolveProcessorFields(
|
||||
processorType: string,
|
||||
isBodyJsonEnabled: boolean,
|
||||
): Array<ProcessorFormField> {
|
||||
const fields = processorFields[processorType] ?? [];
|
||||
|
||||
if (!isBodyJsonEnabled) {
|
||||
return fields;
|
||||
}
|
||||
|
||||
return fields.map((field) =>
|
||||
field.name === 'parse_from' && field.initialValue === BODY_PARSE_FROM
|
||||
? { ...field, initialValue: JSON_BODY_PARSE_FROM }
|
||||
: field,
|
||||
);
|
||||
}
|
||||
@@ -1,45 +0,0 @@
|
||||
import { processorFields } from '../PipelineListsView/AddNewProcessor/config';
|
||||
import { resolveProcessorFields } from '../PipelineListsView/AddNewProcessor/utils';
|
||||
|
||||
const parseFromDefault = (
|
||||
fields: ReturnType<typeof resolveProcessorFields>,
|
||||
): unknown => fields.find((field) => field.name === 'parse_from')?.initialValue;
|
||||
|
||||
describe('resolveProcessorFields', () => {
|
||||
it.each(['grok_parser', 'regex_parser', 'json_parser'])(
|
||||
'defaults %s parse_from to body.message when use_json_body is on',
|
||||
(processorType) => {
|
||||
expect(parseFromDefault(resolveProcessorFields(processorType, true))).toBe(
|
||||
'body.message',
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it.each(['grok_parser', 'regex_parser', 'json_parser'])(
|
||||
'keeps %s parse_from as body when use_json_body is off',
|
||||
(processorType) => {
|
||||
expect(parseFromDefault(resolveProcessorFields(processorType, false))).toBe(
|
||||
'body',
|
||||
);
|
||||
},
|
||||
);
|
||||
|
||||
it('leaves parse_from defaults that do not point at the body alone', () => {
|
||||
expect(parseFromDefault(resolveProcessorFields('time_parser', true))).toBe(
|
||||
'attributes.timestamp',
|
||||
);
|
||||
expect(
|
||||
parseFromDefault(resolveProcessorFields('severity_parser', true)),
|
||||
).toBe('attributes.logLevel');
|
||||
});
|
||||
|
||||
it('does not mutate the shared config', () => {
|
||||
resolveProcessorFields('grok_parser', true);
|
||||
|
||||
expect(parseFromDefault(processorFields.grok_parser)).toBe('body');
|
||||
});
|
||||
|
||||
it('returns an empty list for an unknown processor type', () => {
|
||||
expect(resolveProcessorFields('does_not_exist', true)).toStrictEqual([]);
|
||||
});
|
||||
});
|
||||
@@ -51,6 +51,28 @@
|
||||
},
|
||||
"name": "Region"
|
||||
}
|
||||
},
|
||||
{
|
||||
"kind": "ListVariable",
|
||||
"spec": {
|
||||
"display": {
|
||||
"name": "FunctionName",
|
||||
"description": "Name of the Lambda function"
|
||||
},
|
||||
"allowAllValue": true,
|
||||
"allowMultiple": true,
|
||||
"customAllValue": "",
|
||||
"capturingRegexp": "",
|
||||
"sort": "none",
|
||||
"plugin": {
|
||||
"kind": "signoz/DynamicVariable",
|
||||
"spec": {
|
||||
"name": "FunctionName",
|
||||
"signal": "metrics"
|
||||
}
|
||||
},
|
||||
"name": "FunctionName"
|
||||
}
|
||||
}
|
||||
],
|
||||
"panels": {
|
||||
@@ -118,7 +140,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -218,7 +240,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -318,7 +340,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -418,7 +440,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -518,7 +540,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -618,7 +640,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -718,7 +740,7 @@
|
||||
],
|
||||
"disabled": false,
|
||||
"filter": {
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS)"
|
||||
"expression": "(cloud.account.id = $Account AND cloud.region = $Region AND FunctionName EXISTS AND Resource NOT EXISTS) AND FunctionName IN $FunctionName"
|
||||
},
|
||||
"groupBy": [
|
||||
{
|
||||
@@ -831,4 +853,4 @@
|
||||
"refreshInterval": "",
|
||||
"links": []
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,7 +29,6 @@ type builderQuery[T any] struct {
|
||||
telemetryStore telemetrystore.TelemetryStore
|
||||
orgID valuer.UUID
|
||||
stmtBuilder qbtypes.StatementBuilder[T]
|
||||
queryType qbtypes.QueryType
|
||||
spec qbtypes.QueryBuilderQuery[T]
|
||||
variables map[string]qbtypes.VariableItem
|
||||
|
||||
@@ -52,7 +51,6 @@ func newBuilderQuery[T any](
|
||||
telemetryStore telemetrystore.TelemetryStore,
|
||||
orgID valuer.UUID,
|
||||
stmtBuilder qbtypes.StatementBuilder[T],
|
||||
queryType qbtypes.QueryType,
|
||||
spec qbtypes.QueryBuilderQuery[T],
|
||||
tr qbtypes.TimeRange,
|
||||
kind qbtypes.RequestType,
|
||||
@@ -64,7 +62,6 @@ func newBuilderQuery[T any](
|
||||
telemetryStore: telemetryStore,
|
||||
orgID: orgID,
|
||||
stmtBuilder: stmtBuilder,
|
||||
queryType: queryType,
|
||||
spec: spec,
|
||||
variables: variables,
|
||||
fromMS: tr.From,
|
||||
@@ -84,7 +81,7 @@ func (q *builderQuery[T]) Fingerprint() string {
|
||||
|
||||
// Create a deterministic fingerprint for builder queries
|
||||
// This needs to include all fields that affect the query results
|
||||
parts := []string{q.queryType.StringValue()}
|
||||
parts := []string{"builder"}
|
||||
|
||||
// Add signal type
|
||||
parts = append(parts, fmt.Sprintf("signal=%s", q.spec.Signal.StringValue()))
|
||||
|
||||
@@ -3,7 +3,6 @@ package querier
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/querybuilder"
|
||||
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
|
||||
@@ -21,8 +20,7 @@ func TestBuilderQueryFingerprint(t *testing.T) {
|
||||
{
|
||||
name: "fingerprint includes shiftby when ShiftBy field is set",
|
||||
query: &builderQuery[qbtypes.MetricAggregation]{
|
||||
queryType: qbtypes.QueryTypeBuilder,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
ShiftBy: 3600,
|
||||
@@ -42,8 +40,7 @@ func TestBuilderQueryFingerprint(t *testing.T) {
|
||||
{
|
||||
name: "fingerprint includes shiftby but not other functions",
|
||||
query: &builderQuery[qbtypes.MetricAggregation]{
|
||||
queryType: qbtypes.QueryTypeBuilder,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
ShiftBy: 3600,
|
||||
@@ -66,8 +63,7 @@ func TestBuilderQueryFingerprint(t *testing.T) {
|
||||
{
|
||||
name: "no shiftby in fingerprint when ShiftBy is zero",
|
||||
query: &builderQuery[qbtypes.MetricAggregation]{
|
||||
queryType: qbtypes.QueryTypeBuilder,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
|
||||
Signal: telemetrytypes.SignalMetrics,
|
||||
ShiftBy: 0,
|
||||
@@ -98,29 +94,6 @@ func TestBuilderQueryFingerprint(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestBuilderQueryFingerprintQueryType(t *testing.T) {
|
||||
spec := qbtypes.QueryBuilderQuery[qbtypes.TraceAggregation]{
|
||||
Signal: telemetrytypes.SignalTraces,
|
||||
StepInterval: qbtypes.Step{Duration: 60 * time.Second},
|
||||
Aggregations: []qbtypes.TraceAggregation{{Expression: "count()"}},
|
||||
Filter: &qbtypes.Filter{Expression: "gen_ai.request.model EXISTS"},
|
||||
}
|
||||
regular := &builderQuery[qbtypes.TraceAggregation]{
|
||||
queryType: qbtypes.QueryTypeBuilder,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
spec: spec,
|
||||
}
|
||||
ai := &builderQuery[qbtypes.TraceAggregation]{
|
||||
queryType: qbtypes.QueryTypeBuilderAI,
|
||||
kind: qbtypes.RequestTypeTimeSeries,
|
||||
spec: spec,
|
||||
}
|
||||
|
||||
assert.True(t, strings.HasPrefix(regular.Fingerprint(), qbtypes.QueryTypeBuilder.StringValue()+"&"))
|
||||
assert.True(t, strings.HasPrefix(ai.Fingerprint(), qbtypes.QueryTypeBuilderAI.StringValue()+"&"))
|
||||
assert.NotEqual(t, regular.Fingerprint(), ai.Fingerprint())
|
||||
}
|
||||
|
||||
func TestMakeBucketsOrder(t *testing.T) {
|
||||
// Test that makeBuckets returns buckets in reverse chronological order by default
|
||||
// Using milliseconds as input - need > 1 hour range to get multiple buckets
|
||||
|
||||
@@ -305,7 +305,7 @@ func (q *querier) buildQueries(
|
||||
}
|
||||
spec.ShiftBy = extractShiftFromBuilderQuery(spec)
|
||||
timeRange := adjustTimeRangeForShift(spec, qbtypes.TimeRange{From: req.Start, To: req.End}, req.RequestType)
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.aiTraceStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.aiTraceStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
queries[spec.Name] = bq
|
||||
steps[spec.Name] = spec.StepInterval
|
||||
case qbtypes.QueryTypeBuilder:
|
||||
@@ -313,7 +313,7 @@ func (q *querier) buildQueries(
|
||||
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, orgID, q.traceStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, q.traceStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
queries[spec.Name] = bq
|
||||
steps[spec.Name] = spec.StepInterval
|
||||
case qbtypes.QueryBuilderQuery[qbtypes.LogAggregation]:
|
||||
@@ -323,7 +323,7 @@ func (q *querier) buildQueries(
|
||||
if spec.Source == telemetrytypes.SourceAudit {
|
||||
stmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, q.builderConfig)
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, stmtBuilder, spec, timeRange, req.RequestType, tmplVars, q.builderConfig)
|
||||
queries[spec.Name] = bq
|
||||
steps[spec.Name] = spec.StepInterval
|
||||
case qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]:
|
||||
@@ -340,9 +340,9 @@ func (q *querier) buildQueries(
|
||||
|
||||
if spec.Source == telemetrytypes.SourceMeter {
|
||||
event.Source = telemetrytypes.SourceMeter.StringValue()
|
||||
bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.meterStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.meterStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
} else {
|
||||
bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.metricStmtBuilder, query.Type, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
bq = newBuilderQuery(q.logger, q.telemetryStore, orgID, q.metricStmtBuilder, spec, timeRange, req.RequestType, tmplVars, builderConfig{})
|
||||
}
|
||||
|
||||
queries[spec.Name] = bq
|
||||
@@ -618,7 +618,7 @@ func (q *querier) QueryRawStream(ctx context.Context, orgID valuer.UUID, req *qb
|
||||
if spec.Source == telemetrytypes.SourceAudit {
|
||||
liveTailStmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, liveTailStmtBuilder, query.Type, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
|
||||
bq := newBuilderQuery(q.logger, q.telemetryStore, orgID, liveTailStmtBuilder, spec, timeRange, req.RequestType, map[string]qbtypes.VariableItem{
|
||||
"id": {
|
||||
Value: updatedLogID,
|
||||
},
|
||||
@@ -941,9 +941,8 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
|
||||
specCopy := qt.spec.Copy()
|
||||
specCopy.ShiftBy = extractShiftFromBuilderQuery(specCopy)
|
||||
adjustedTimeRange := adjustTimeRangeForShift(specCopy, timeRange, qt.kind)
|
||||
// reuse the original query's statement builder and type so an AI query
|
||||
// keeps its AI builder and cache key
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, qt.stmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, qt.builderConfig)
|
||||
// reuse the original query's statement builder so an AI query keeps its AI builder
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, qt.stmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
|
||||
|
||||
case *builderQuery[qbtypes.LogAggregation]:
|
||||
specCopy := qt.spec.Copy()
|
||||
@@ -953,16 +952,16 @@ func (q *querier) createRangedQuery(_ valuer.UUID, originalQuery qbtypes.Query,
|
||||
if qt.spec.Source == telemetrytypes.SourceAudit {
|
||||
shiftStmtBuilder = q.auditStmtBuilder
|
||||
}
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, shiftStmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, q.builderConfig)
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, shiftStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, q.builderConfig)
|
||||
|
||||
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, qt.orgID, q.meterStmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, q.meterStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
|
||||
}
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, q.metricStmtBuilder, qt.queryType, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
|
||||
return newBuilderQuery(q.logger, q.telemetryStore, qt.orgID, q.metricStmtBuilder, specCopy, adjustedTimeRange, qt.kind, qt.variables, builderConfig{})
|
||||
case *traceOperatorQuery:
|
||||
specCopy := qt.spec.Copy()
|
||||
return &traceOperatorQuery{
|
||||
|
||||
Reference in New Issue
Block a user