diff --git a/pkg/query-service/app/logparsingpipeline/pipelineBuilder.go b/pkg/query-service/app/logparsingpipeline/pipelineBuilder.go index d20430e0a6..14ac63d3cc 100644 --- a/pkg/query-service/app/logparsingpipeline/pipelineBuilder.go +++ b/pkg/query-service/app/logparsingpipeline/pipelineBuilder.go @@ -138,7 +138,6 @@ func getOperators(ops []PipelineOperator) ([]PipelineOperator, error) { } // TODO(Raj): Maybe add support for gotime too eventually - operator.OnError = "send" } filteredOp = append(filteredOp, operator) diff --git a/pkg/query-service/app/logparsingpipeline/postablePipeline.go b/pkg/query-service/app/logparsingpipeline/postablePipeline.go index 743e777f1a..307f39ee9f 100644 --- a/pkg/query-service/app/logparsingpipeline/postablePipeline.go +++ b/pkg/query-service/app/logparsingpipeline/postablePipeline.go @@ -171,6 +171,7 @@ func isValidOperator(op PipelineOperator) error { return fmt.Errorf(fmt.Sprintf("parse from of time parsing processor %s cannot be empty", op.ID)) } if op.LayoutType != "epoch" && op.LayoutType != "strptime" { + // TODO(Raj): Maybe add support for gotime format return fmt.Errorf(fmt.Sprintf( "invalid format type '%s' of time parsing processor %s", op.LayoutType, op.ID, )) @@ -186,6 +187,9 @@ func isValidOperator(op PipelineOperator) error { )) } + // TODO(Raj): Add validation for strptime layouts via + // collector simulator maybe. + default: return fmt.Errorf(fmt.Sprintf("operator type %s not supported for %s, use one of (grok_parser, regex_parser, copy, move, add, remove, trace_parser, retain)", op.Type, op.ID)) } diff --git a/pkg/query-service/app/logparsingpipeline/postablePipeline_test.go b/pkg/query-service/app/logparsingpipeline/postablePipeline_test.go index ce6d4bfc55..5140458f0f 100644 --- a/pkg/query-service/app/logparsingpipeline/postablePipeline_test.go +++ b/pkg/query-service/app/logparsingpipeline/postablePipeline_test.go @@ -309,10 +309,9 @@ var operatorTest = []struct { }, { Name: "Timestamp Parser - invalid epoch layout", Operator: PipelineOperator{ - ID: "time", - Type: "time_parser", - ParseFrom: "attributes.test_timestamp", - // TODO(Raj): Maybe add support for gotime format + ID: "time", + Type: "time_parser", + ParseFrom: "attributes.test_timestamp", LayoutType: "epoch", Layout: "%Y-%m-%d", }, diff --git a/pkg/query-service/app/logparsingpipeline/time_parser.go b/pkg/query-service/app/logparsingpipeline/time_parser.go index 58d1fa3fab..a0ec384867 100644 --- a/pkg/query-service/app/logparsingpipeline/time_parser.go +++ b/pkg/query-service/app/logparsingpipeline/time_parser.go @@ -8,7 +8,12 @@ import ( ) // Regex for strptime format placeholders supported by the time parser. +// Used for defining if conditions on time parsing operators so they do not +// spam collector logs when encountering values that can't be parsed. +// // Based on ctimeSubstitutes defined in https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/internal/coreinternal/timeutils/internal/ctimefmt/ctimefmt.go#L22 +// +// TODO(Raj): Maybe make the expressions tighter. var ctimeRegex = map[string]string{ // %Y - Year, zero-padded (0001, 0002, ..., 2019, 2020, ..., 9999) "%Y": "[0-9]{4}", @@ -105,8 +110,8 @@ func RegexForStrptimeLayout(layout string) (string, error) { return "" } - ctimeRegexp := regexp.MustCompile(`%.`) - layoutRegex = ctimeRegexp.ReplaceAllStringFunc(layoutRegex, replaceStrptimeDirectiveWithRegex) + strptimeDirectiveRegexp := regexp.MustCompile(`%.`) + layoutRegex = strptimeDirectiveRegexp.ReplaceAllStringFunc(layoutRegex, replaceStrptimeDirectiveWithRegex) if len(errs) != 0 { return "", fmt.Errorf("couldn't generate regex for ctime format: %v", errs) } diff --git a/pkg/query-service/app/logparsingpipeline/time_parser_test.go b/pkg/query-service/app/logparsingpipeline/time_parser_test.go index a214f26b6f..15c41bf73e 100644 --- a/pkg/query-service/app/logparsingpipeline/time_parser_test.go +++ b/pkg/query-service/app/logparsingpipeline/time_parser_test.go @@ -91,8 +91,6 @@ func TestTimestampParsingProcessor(t *testing.T) { }, } - // Start with JSON serialized timestamp parser to validate deserialization too - // TODO(Raj): Is this needed? var timestampParserOp PipelineOperator err := json.Unmarshal([]byte(` {