Repository: ozontech/file.d (go)

Issue:
Feature: Allow aliases of formats in do_if timestamp parser
Now we have only custom format.
See how it works in throttle plugin.



Pull request:
Allow aliases of formats in do_if timestamp parser
Fixes #794

Patch:
diff --git a/fd/util.go b/fd/util.go
index 2aa127b..e57444c 100644
--- a/fd/util.go
+++ b/fd/util.go
@@ -360,7 +360,7 @@ const (
 
 const (
 	defaultTsCmpValUpdateInterval = 10 * time.Second
-	defaultTsFormat               = time.RFC3339Nano
+	defaultTsFormat               = "rfc3339nano"
 )
 
 func extractTsCmpOpNode(_ string, jsonNode *simplejson.Json) (doif.Node, error) {
diff --git a/pipeline/doif/README.md b/pipeline/doif/README.md
index 9153420..7595453 100755
--- a/pipeline/doif/README.md
+++ b/pipeline/doif/README.md
@@ -367,7 +367,10 @@ It contains operation that compares timestamps with certain value.
 Params:
   - `op` - must be `ts_cmp`. Required.
   - `field` - name of the field to apply operation. Required. Field will be parsed with `time.Parse` function.
-  - `format` - format for timestamps representation. Optional; default = `time.RFC3339Nano`.
+  - `format` - format for timestamps representation.
+Can be specified as a datetime layout in Go [time.Parse](https://pkg.go.dev/time#Parse) format or by alias.
+List of available datetime format aliases can be found [here](/pipeline/README.md#datetime-parse-formats).
+Optional; default = `rfc3339nano`.
   - `cmp_op` - comparison operation name (same as for length comparison operations). Required.
   - `value` - timestamp value to compare field timestamps with. It must have `RFC3339Nano` format. Required.
 Also, it may be `now` or `file_d_start`. If it is `now` then value to compare timestamps with is periodically updated current time.
diff --git a/pipeline/doif/ts_cmp_op.go b/pipeline/doif/ts_cmp_op.go
index 402bd99..f24753c 100644
--- a/pipeline/doif/ts_cmp_op.go
+++ b/pipeline/doif/ts_cmp_op.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	"github.com/ozontech/file.d/cfg"
+	"github.com/ozontech/file.d/xtime"
 	insaneJSON "github.com/ozontech/insane-json"
 )
 
@@ -18,7 +19,10 @@ It contains operation that compares timestamps with certain value.
 Params:
   - `op` - must be `ts_cmp`. Required.
   - `field` - name of the field to apply operation. Required. Field will be parsed with `time.Parse` function.
-  - `format` - format for timestamps representation. Optional; default = `time.RFC3339Nano`.
+  - `format` - format for timestamps representation.
+Can be specified as a datetime layout in Go [time.Parse](https://pkg.go.dev/time#Parse) format or by alias.
+List of available datetime format aliases can be found [here](/pipeline/README.md#datetime-parse-formats).
+Optional; default = `rfc3339nano`.
   - `cmp_op` - comparison operation name (same as for length comparison operations). Required.
   - `value` - timestamp value to compare field timestamps with. It must have `RFC3339Nano` format. Required.
 Also, it may be `now` or `file_d_start`. If it is `now` then value to compare timestamps with is periodically updated current time.
@@ -98,9 +102,14 @@ func NewTsCmpOpNode(field string, format string, cmpOp string, cmpValChangeMode
 		return nil, fmt.Errorf("unknown ts cmp mode: %s", cmpValChangeMode)
 	}
 
+	parsedFormat, err := xtime.ParseFormatName(format)
+	if err != nil {
+		parsedFormat = format
+	}
+
 	result := &tsCmpOpNode{
 		fieldPath:        fieldPath,
-		format:           format,
+		format:           parsedFormat,
 		cmpOp:            typedCmpOp,
 		cmpValChangeMode: resCmpValChangeMode,
 		constCmpValue:    cmpValue.UnixNano(),
@@ -138,7 +147,7 @@ func (n *tsCmpOpNode) Check(eventRoot *insaneJSON.Root) bool {
 		return false
 	}
 
-	timeVal, err := time.Parse(n.format, node.AsString())
+	timeVal, err := xtime.ParseTime(n.format, node.AsString())
 	if err != nil {
 		return false
 	}
diff --git a/pipeline/util.go b/pipeline/util.go
index 112a70b..8e53632 100644
--- a/pipeline/util.go
+++ b/pipeline/util.go
@@ -1,12 +1,8 @@
 package pipeline
 
 import (
-	"fmt"
-	"math"
 	"reflect"
-	"strconv"
 	"strings"
-	"time"
 	"unsafe"
 
 	insaneJSON "github.com/ozontech/insane-json"
@@ -56,127 +52,6 @@ func StringToByteUnsafe(s string) (b []byte) {
 }
 */
 
-const (
-	formats       = "ansic|unixdate|rubydate|rfc822|rfc822z|rfc850|rfc1123|rfc1123z|rfc3339|rfc3339nano|kitchen|stamp|stampmilli|stampmicro|stampnano|unixtime|unixtimemilli|unixtimemicro|unixtimenano|nginx_errorlog"
-	UnixTime      = "unixtime"
-	UnixTimeMilli = "unixtimemilli"
-	UnixTimeMicro = "unixtimemicro"
-	UnixTimeNano  = "unixtimenano"
-	nginxDateFmt  = "2006/01/02 15:04:05"
-)
-
-func ParseFormatName(formatName string) (string, error) {
-	formatNameProcessed := strings.ToLower(strings.TrimSpace(formatName))
-	switch formatNameProcessed {
-	case "ansic":
-		return time.ANSIC, nil
-	case "unixdate":
-		return time.UnixDate, nil
-	case "rubydate":
-		return time.RubyDate, nil
-	case "rfc822":
-		return time.RFC822, nil
-	case "rfc822z":
-		return time.RFC822Z, nil
-	case "rfc850":
-		return time.RFC850, nil
-	case "rfc1123":
-		return time.RFC1123, nil
-	case "rfc1123z":
-		return time.RFC1123Z, nil
-	case "rfc3339":
-		return time.RFC3339, nil
-	case "rfc3339nano":
-		return time.RFC3339Nano, nil
-	case "kitchen":
-		return time.Kitchen, nil
-	case "stamp":
-		return time.Stamp, nil
-	case "stampmilli":
-		return time.StampMilli, nil
-	case "stampmicro":
-		return time.StampMicro, nil
-	case "stampnano":
-		return time.StampNano, nil
-	case "nginx_errorlog":
-		return nginxDateFmt, nil
-	case UnixTime, UnixTimeMilli, UnixTimeMicro, UnixTimeNano:
-		return formatNameProcessed, nil
-	default:
-		return "", fmt.Errorf("unknown format name %q, should be one of %s", formatName, formats)
-	}
-}
-
-type unixTimeFormat int
-
-const (
-	unixTimeSec unixTimeFormat = iota
-	unixTimeMilli
-	unixTimeMicro
-	unixTimeNano
-)
-
-func ParseTime(format, value string) (time.Time, error) {
-	switch format {
-	case UnixTime:
-		return parseUnixTime(value, unixTimeSec)
-	case UnixTimeMilli:
-		return parseUnixTime(value, unixTimeMilli)
-	case UnixTimeMicro:
-		return parseUnixTime(value, unixTimeMicro)
-	case UnixTimeNano:
-		return parseUnixTime(value, unixTimeNano)
-	default:
-		return time.Parse(format, value)
-	}
-}
-
-func parseUnixTime(value string, format unixTimeFormat) (time.Time, error) {
-	numbers := strings.Split(value, ".")
-	var sec, nsec, val int64
-	var err error
-	switch len(numbers) {
-	case 1:
-		val, err = strconv.ParseInt(numbers[0], 10, 64)
-		if err != nil {
-			return time.Time{}, err
-		}
-		switch format {
-		case unixTimeSec:
-			sec = val
-		case unixTimeMilli:
-			sec = val / 1e3
-			nsec = (val % 1e3) * 1e6
-		case unixTimeMicro:
-			sec = val / 1e6
-			nsec = (val % 1e6) * 1e3
-		case unixTimeNano:
-			sec = val / 1e9
-			nsec = val % 1e9
-		}
-	case 2:
-		// when timestamp is presented as a float number its whole part is always considered as seconds
-		// and the fractional part is fractions of a second
-		sec, err = strconv.ParseInt(numbers[0], 10, 64)
-		if err != nil {
-			return time.Time{}, err
-		}
-		nsec, err = strconv.ParseInt(numbers[1], 10, 64)
-		if err != nil {
-			return time.Time{}, err
-		}
-		// if there are less than 9 digits to the right of the decimal point
-		// it must be multiplied by 10^(9 - digits) to get nsec value
-		digits := len(numbers[1])
-		if digits < 9 {
-			nsec *= int64(math.Pow10(9 - digits))
-		}
-	default:
-		return time.Time{}, fmt.Errorf("unexpected time format")
-	}
-	return time.Unix(sec, nsec), nil
-}
-
 type LogLevel int
 
 const (
diff --git a/plugin/action/convert_date/convert_date.go b/plugin/action/convert_date/convert_date.go
index e1cbbc5..2ee6754 100644
--- a/plugin/action/convert_date/convert_date.go
+++ b/plugin/action/convert_date/convert_date.go
@@ -4,6 +4,7 @@ import (
 	"github.com/ozontech/file.d/cfg"
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/pipeline"
+	"github.com/ozontech/file.d/xtime"
 )
 
 /*{ introduction
@@ -58,14 +59,14 @@ func (p *Plugin) Start(config pipeline.AnyConfig, _ *pipeline.ActionPluginParams
 	p.config = config.(*Config)
 
 	for _, formatName := range p.config.SourceFormats {
-		format, err := pipeline.ParseFormatName(formatName)
+		format, err := xtime.ParseFormatName(formatName)
 		if err != nil {
 			format = formatName
 		}
 		p.config.SourceFormats_ = append(p.config.SourceFormats_, format)
 	}
 
-	format, err := pipeline.ParseFormatName(p.config.TargetFormat)
+	format, err := xtime.ParseFormatName(p.config.TargetFormat)
 	if err != nil {
 		format = p.config.TargetFormat
 	}
@@ -85,16 +86,16 @@ func (p *Plugin) Do(event *pipeline.Event) pipeline.ActionResult {
 	if isValidType {
 		date := dateNode.AsString()
 		for _, format := range p.config.SourceFormats_ {
-			t, err := pipeline.ParseTime(format, date)
+			t, err := xtime.ParseTime(format, date)
 			if err == nil {
 				switch p.config.TargetFormat_ {
-				case pipeline.UnixTime:
+				case xtime.UnixTime:
 					dateNode.MutateToInt(int(t.Unix()))
-				case pipeline.UnixTimeMilli:
+				case xtime.UnixTimeMilli:
 					dateNode.MutateToInt(int(t.UnixMilli()))
-				case pipeline.UnixTimeMicro:
+				case xtime.UnixTimeMicro:
 					dateNode.MutateToInt(int(t.UnixMicro()))
-				case pipeline.UnixTimeNano:
+				case xtime.UnixTimeNano:
 					dateNode.MutateToInt(int(t.UnixNano()))
 				default:
 					dateNode.MutateToString(t.Format(p.config.TargetFormat_))
diff --git a/plugin/action/set_time/set_time.go b/plugin/action/set_time/set_time.go
index 1b3d425..864ed86 100644
--- a/plugin/action/set_time/set_time.go
+++ b/plugin/action/set_time/set_time.go
@@ -5,6 +5,7 @@ import (
 
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/pipeline"
+	"github.com/ozontech/file.d/xtime"
 )
 
 /*{ introduction
@@ -50,7 +51,7 @@ func factory() (pipeline.AnyPlugin, pipeline.AnyConfig) {
 func (p *Plugin) Start(config pipeline.AnyConfig, _ *pipeline.ActionPluginParams) {
 	p.config = config.(*Config)
 
-	format, err := pipeline.ParseFormatName(p.config.Format)
+	format, err := xtime.ParseFormatName(p.config.Format)
 	if err != nil {
 		// to support custom formats
 		format = p.config.Format
@@ -76,13 +77,13 @@ func (p *Plugin) do(event *pipeline.Event, t time.Time) pipeline.ActionResult {
 	}
 
 	switch p.config.Format_ {
-	case pipeline.UnixTime:
+	case xtime.UnixTime:
 		dateNode.MutateToInt64(t.Unix())
-	case pipeline.UnixTimeMilli, "timestampmilli": // timestamp(milli|micro|nano) are left for backward compatibility
+	case xtime.UnixTimeMilli, "timestampmilli": // timestamp(milli|micro|nano) are left for backward compatibility
 		dateNode.MutateToInt64(t.UnixMilli())
-	case pipeline.UnixTimeMicro, "timestampmicro":
+	case xtime.UnixTimeMicro, "timestampmicro":
 		dateNode.MutateToInt64(t.UnixMicro())
-	case pipeline.UnixTimeNano, "timestampnano":
+	case xtime.UnixTimeNano, "timestampnano":
 		dateNode.MutateToInt64(t.UnixNano())
 	default:
 		dateNode.MutateToString(t.Format(p.config.Format_))
diff --git a/plugin/action/throttle/throttle.go b/plugin/action/throttle/throttle.go
index 342e141..0e98737 100644
--- a/plugin/action/throttle/throttle.go
+++ b/plugin/action/throttle/throttle.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/go-redis/redis"
+	"github.com/ozontech/file.d/xtime"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 
@@ -307,7 +308,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.ActionPluginP
 	p.ctx = ctx
 	p.cancel = cancel
 
-	format, err := pipeline.ParseFormatName(p.config.TimeFieldFormat)
+	format, err := xtime.ParseFormatName(p.config.TimeFieldFormat)
 	if err != nil {
 		format = p.config.TimeFieldFormat
 	}
@@ -425,7 +426,7 @@ func (p *Plugin) isAllowed(event *pipeline.Event) bool {
 
 	if len(p.config.TimeField_) != 0 {
 		tsValue := event.Root.Dig(p.config.TimeField_...).AsString()
-		t, err := pipeline.ParseTime(p.format, tsValue)
+		t, err := xtime.ParseTime(p.format, tsValue)
 		if err != nil || t.IsZero() {
 			p.logger.Warnf("can't parse field %q using format %s: %s", p.config.TimeField, p.config.TimeFieldFormat, tsValue)
 			ts = time.Now()
diff --git a/plugin/output/gelf/gelf.go b/plugin/output/gelf/gelf.go
index 3fbfb0a..e21a88d 100644
--- a/plugin/output/gelf/gelf.go
+++ b/plugin/output/gelf/gelf.go
@@ -9,6 +9,7 @@ import (
 	"github.com/ozontech/file.d/fd"
 	"github.com/ozontech/file.d/metric"
 	"github.com/ozontech/file.d/pipeline"
+	"github.com/ozontech/file.d/xtime"
 	insaneJSON "github.com/ozontech/insane-json"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
@@ -208,7 +209,7 @@ func (p *Plugin) Start(config pipeline.AnyConfig, params *pipeline.OutputPluginP
 	p.config.defaultShortMessageValue = strings.TrimSpace(p.config.DefaultShortMessageValue)
 	p.config.fullMessageField = pipeline.ByteToStringUnsafe(p.formatExtraField(nil, p.config.FullMessageField))
 	p.config.timestampField = pipeline.ByteToStringUnsafe(p.formatExtraField(nil, p.config.TimestampField))
-	format, err := pipeline.ParseFormatName(p.config.TimestampFieldFormat)
+	format, err := xtime.ParseFormatName(p.config.TimestampFieldFormat)
 	if err != nil {
 		// TODO: refactor plugin code to unify with datetime parsing in other plugins
 		params.Logger.Errorf("unknown time format: %s", err.Error())
@@ -395,7 +396,7 @@ func (p *Plugin) makeTimestampField(root *insaneJSON.Root, timestampField string
 			ts /= 1000
 		}
 	} else if node.IsString() {
-		t, err := pipeline.ParseTime(timestampFieldFormat, node.AsString())
+		t, err := xtime.ParseTime(timestampFieldFormat, node.AsString())
 		if err == nil {
 			ts = float64(t.UnixNano()) / float64(time.Second)
 		}
diff --git a/xtime/parse_format_name.go b/xtime/parse_format_name.go
new file mode 100644
index 0000000..d63785e
--- /dev/null
+++ b/xtime/parse_format_name.go
@@ -0,0 +1,58 @@
+package xtime
+
+import (
+	"fmt"
+	"strings"
+	"time"
+)
+
+const (
+	formats       = "ansic|unixdate|rubydate|rfc822|rfc822z|rfc850|rfc1123|rfc1123z|rfc3339|rfc3339nano|kitchen|stamp|stampmilli|stampmicro|stampnano|unixtime|unixtimemilli|unixtimemicro|unixtimenano|nginx_errorlog"
+	UnixTime      = "unixtime"
+	UnixTimeMilli = "unixtimemilli"
+	UnixTimeMicro = "unixtimemicro"
+	UnixTimeNano  = "unixtimenano"
+	nginxDateFmt  = "2006/01/02 15:04:05"
+)
+
+func ParseFormatName(formatName string) (string, error) {
+	formatNameProcessed := strings.ToLower(strings.TrimSpace(formatName))
+	switch formatNameProcessed {
+	case "ansic":
+		return time.ANSIC, nil
+	case "unixdate":
+		return time.UnixDate, nil
+	case "rubydate":
+		return time.RubyDate, nil
+	case "rfc822":
+		return time.RFC822, nil
+	case "rfc822z":
+		return time.RFC822Z, nil
+	case "rfc850":
+		return time.RFC850, nil
+	case "rfc1123":
+		return time.RFC1123, nil
+	case "rfc1123z":
+		return time.RFC1123Z, nil
+	case "rfc3339":
+		return time.RFC3339, nil
+	case "rfc3339nano":
+		return time.RFC3339Nano, nil
+	case "kitchen":
+		return time.Kitchen, nil
+	case "stamp":
+		return time.Stamp, nil
+	case "stampmilli":
+		return time.StampMilli, nil
+	case "stampmicro":
+		return time.StampMicro, nil
+	case "stampnano":
+		return time.StampNano, nil
+	case "nginx_errorlog":
+		return nginxDateFmt, nil
+	case UnixTime, UnixTimeMilli, UnixTimeMicro, UnixTimeNano:
+		return formatNameProcessed, nil
+	default:
+		return "", fmt.Errorf("unknown format name %q, should be one of %s", formatName, formats)
+	}
+}
diff --git a/xtime/parse_time.go b/xtime/parse_time.go
new file mode 100644
index 0000000..7703068
--- /dev/null
+++ b/xtime/parse_time.go
@@ -0,0 +1,79 @@
+package xtime
+
+import (
+	"fmt"
+	"math"
+	"strconv"
+	"strings"
+	"time"
+)
+
+type unixTimeFormat int
+
+const (
+	unixTimeSec unixTimeFormat = iota
+	unixTimeMilli
+	unixTimeMicro
+	unixTimeNano
+)
+
+func ParseTime(format, value string) (time.Time, error) {
+	switch format {
+	case UnixTime:
+		return parseUnixTime(value, unixTimeSec)
+	case UnixTimeMilli:
+		return parseUnixTime(value, unixTimeMilli)
+	case UnixTimeMicro:
+		return parseUnixTime(value, unixTimeMicro)
+	case UnixTimeNano:
+		return parseUnixTime(value, unixTimeNano)
+	default:
+		return time.Parse(format, value)
+	}
+}
+
+func parseUnixTime(value string, format unixTimeFormat) (time.Time, error) {
+	numbers := strings.Split(value, ".")
+	var sec, nsec, val int64
+	var err error
+	switch len(numbers) {
+	case 1:
+		val, err = strconv.ParseInt(numbers[0], 10, 64)
+		if err != nil {
+			return time.Time{}, err
+		}
+		switch format {
+		case unixTimeSec:
+			sec = val
+		case unixTimeMilli:
+			sec = val / 1e3
+			nsec = (val % 1e3) * 1e6
+		case unixTimeMicro:
+			sec = val / 1e6
+			nsec = (val % 1e6) * 1e3
+		case unixTimeNano:
+			sec = val / 1e9
+			nsec = val % 1e9
+		}
+	case 2:
+		// when timestamp is presented as a float number its whole part is always considered as seconds
+		// and the fractional part is fractions of a second
+		sec, err = strconv.ParseInt(numbers[0], 10, 64)
+		if err != nil {
+			return time.Time{}, err
+		}
+		nsec, err = strconv.ParseInt(numbers[1], 10, 64)
+		if err != nil {
+			return time.Time{}, err
+		}
+		// if there are less than 9 digits to the right of the decimal point
+		// it must be multiplied by 10^(9 - digits) to get nsec value
+		digits := len(numbers[1])
+		if digits < 9 {
+			nsec *= int64(math.Pow10(9 - digits))
+		}
+	default:
+		return time.Time{}, fmt.Errorf("unexpected time format")
+	}
+	return time.Unix(sec, nsec), nil
+}


Test patch:
diff --git a/fd/util_test.go b/fd/util_test.go
index c69f29f..deeac04 100644
--- a/fd/util_test.go
+++ b/fd/util_test.go
@@ -328,6 +328,27 @@ func Test_extractDoIfChecker(t *testing.T) {
 				tsUpdateInterval:   defaultTsCmpValUpdateInterval,
 			},
 		},
+		{
+			name: "ok_ts_cmp_op_format_alias",
+			args: args{
+				cfgStr: `{
+					"op": "ts_cmp",
+					"field": "timestamp",
+					"cmp_op": "lt",
+					"format": "rfc3339",
+					"value": "now"}`,
+			},
+			want: &doIfTreeNode{
+				tsCmpOp:            true,
+				cmpOp:              "lt",
+				fieldName:          "timestamp",
+				tsCmpValChangeMode: tsCmpModeNowTag,
+				tsFormat:           time.RFC3339,
+				tsCmpValueShift:    0,
+				tsUpdateInterval:   defaultTsCmpValUpdateInterval,
+			},
+		},
+
 		{
 			name: "ok_check_type",
 			args: args{
@@ -491,59 +512,45 @@ func Test_extractDoIfChecker(t *testing.T) {
 			},
 			wantErr: true,
 		},
-		{
-			name: "error_ts_cmp_op_no_format",
-			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp"}`,
-			},
-			wantErr: true,
-		},
-		{
-			name: "error_ts_cmp_op_format_is_not_string",
-			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":123}`,
-			},
-			wantErr: true,
-		},
 		{
 			name: "error_ts_cmp_op_no_cmp_op",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00"}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp"}`,
 			},
 			wantErr: true,
 		},
 		{
 			name: "error_ts_cmp_op_cmp_op_is_not_string",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00","cmp_op":123}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp","cmp_op":123}`,
 			},
 			wantErr: true,
 		},
 		{
 			name: "error_ts_cmp_op_no_cmp_value",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00","cmp_op":"lt"}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp","cmp_op":"lt"}`,
 			},
 			wantErr: true,
 		},
 		{
 			name: "error_ts_cmp_op_cmp_value_is_not_string",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00","cmp_op":"lt","value":123}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp","cmp_op":"lt","value":123}`,
 			},
 			wantErr: true,
 		},
 		{
 			name: "error_ts_cmp_op_invalid_cmp_value",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00","cmp_op":"lt","value":"qwe"}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp","cmp_op":"lt","value":"qwe"}`,
 			},
 			wantErr: true,
 		},
 		{
 			name: "error_ts_cmp_op_invalid_cmp_op",
 			args: args{
-				cfgStr: `{"op":"ts_cmp","field":"timestamp","format":"2006-01-02T15:04:05.999999999Z07:00","cmp_op":"qwe","value":"2009-11-10T23:00:00Z"}`,
+				cfgStr: `{"op":"ts_cmp","field":"timestamp","cmp_op":"qwe","value":"2009-11-10T23:00:00Z"}`,
 			},
 			wantErr: true,
 		},
@@ -555,7 +562,6 @@ func Test_extractDoIfChecker(t *testing.T) {
 					"field": "timestamp",
 					"cmp_op": "lt",
 					"value": "2009-11-10T23:00:00Z",
-					"format": "2006-01-02T15:04:05.999999999Z07:00",
 					"update_interval": "qwe"}`,
 			},
 			wantErr: true,
diff --git a/pipeline/doif/do_if_test.go b/pipeline/doif/do_if_test.go
index 31f4339..0d42964 100644
--- a/pipeline/doif/do_if_test.go
+++ b/pipeline/doif/do_if_test.go
@@ -1038,6 +1038,22 @@ func TestCheck(t *testing.T) {
 				{getTsLog(timestamp.Add(1 * time.Second)), false},
 			},
 		},
+		{
+			name: "ts_cmp_eq_unixtime",
+			tree: treeNode{
+				tsCmpOp:            true,
+				cmpOp:              "eq",
+				fieldName:          "ts",
+				tsFormat:           "unixtime",
+				tsCmpValChangeMode: "const",
+				tsCmpValue:         time.Unix(1735678800, 0),
+			},
+			data: []argsResp{
+				{`{"ts":"1735678799"}`, false},
+				{`{"ts":"1735678800"}`, true},
+				{`{"ts":"1735678801"}`, false},
+			},
+		},
 		{
 			name: "ts_cmp_ne",
 			tree: treeNode{
diff --git a/pipeline/util_test.go b/pipeline/util_test.go
index 70cd261..394e071 100644
--- a/pipeline/util_test.go
+++ b/pipeline/util_test.go
@@ -1,9 +1,7 @@
 package pipeline
 
 import (
-	"strconv"
 	"testing"
-	"time"
 
 	insaneJSON "github.com/ozontech/insane-json"
 	"github.com/stretchr/testify/require"
@@ -107,46 +105,3 @@ func TestLevelParsing(t *testing.T) {
 		require.Equal(t, got, expected)
 	}
 }
-
-func TestParseTime(t *testing.T) {
-	testTime := time.Now()
-	tests := []struct {
-		name   string
-		format string
-		value  string
-		want   int64
-	}{
-		{
-			name:   "unixtime_ok",
-			format: "unixtime",
-			value:  strconv.FormatInt(testTime.Unix(), 10),
-			want:   testTime.Unix() * 1e9,
-		},
-		{
-			name:   "unixtimemilli_ok",
-			format: "unixtimemilli",
-			value:  strconv.FormatInt(testTime.UnixMilli(), 10),
-			want:   testTime.UnixMilli() * 1e6,
-		},
-		{
-			name:   "unixtimemicro_ok",
-			format: "unixtimemicro",
-			value:  strconv.FormatInt(testTime.UnixMicro(), 10),
-			want:   testTime.UnixMicro() * 1e3,
-		},
-		{
-			name:   "unixtimenano_ok",
-			format: "unixtimenano",
-			value:  strconv.FormatInt(testTime.UnixNano(), 10),
-			want:   testTime.UnixNano(),
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			got, err := ParseTime(tt.format, tt.value)
-			require.NoError(t, err, "must be no error")
-			require.Equal(t, tt.want, got.UnixNano())
-		})
-	}
-}
diff --git a/plugin/action/convert_date/convert_date_test.go b/plugin/action/convert_date/convert_date_test.go
index 3005987..89d9198 100644
--- a/plugin/action/convert_date/convert_date_test.go
+++ b/plugin/action/convert_date/convert_date_test.go
@@ -6,11 +6,12 @@ import (
 
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/test"
+	"github.com/ozontech/file.d/xtime"
 	"github.com/stretchr/testify/assert"
 )
 
 func TestConvert(t *testing.T) {
-	config := &Config{SourceFormats: []string{"rfc3339nano", "rfc3339", "ansic", pipeline.UnixTime, "nginx_errorlog"}}
+	config := &Config{SourceFormats: []string{"rfc3339nano", "rfc3339", "ansic", xtime.UnixTime, "nginx_errorlog"}}
 	test.NewConfig(config, nil)
 
 	p, input, output := test.NewPipelineMock(test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, nil, false))
diff --git a/plugin/action/set_time/set_time_test.go b/plugin/action/set_time/set_time_test.go
index 18a46a4..1821312 100644
--- a/plugin/action/set_time/set_time_test.go
+++ b/plugin/action/set_time/set_time_test.go
@@ -8,6 +8,7 @@ import (
 	"github.com/ozontech/file.d/cfg"
 	"github.com/ozontech/file.d/pipeline"
 	"github.com/ozontech/file.d/test"
+	"github.com/ozontech/file.d/xtime"
 	insaneJSON "github.com/ozontech/insane-json"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/atomic"
@@ -30,7 +31,7 @@ func TestPlugin_Do(t *testing.T) {
 		{
 			Name: "unix",
 			Config: &Config{
-				Format: pipeline.UnixTime,
+				Format: xtime.UnixTime,
 				Field:  "time",
 			},
 			Root: `{}`,
@@ -132,7 +133,7 @@ func TestPlugin_Do(t *testing.T) {
 }
 
 func TestE2E_Plugin(t *testing.T) {
-	config := test.NewConfig(&Config{Format: pipeline.UnixTime, Field: "timestamp"}, nil)
+	config := test.NewConfig(&Config{Format: xtime.UnixTime, Field: "timestamp"}, nil)
 	p, input, output := test.NewPipelineMock(test.NewActionPluginStaticInfo(factory, config, pipeline.MatchModeAnd, nil, false))
 
 	counter := atomic.Int32{}
diff --git a/xtime/parse_time_test.go b/xtime/parse_time_test.go
new file mode 100644
index 0000000..ba0905a
--- /dev/null
+++ b/xtime/parse_time_test.go
@@ -0,0 +1,52 @@
+package xtime
+
+import (
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestParseTime(t *testing.T) {
+	testTime := time.Now()
+	tests := []struct {
+		name   string
+		format string
+		value  string
+		want   int64
+	}{
+		{
+			name:   "unixtime_ok",
+			format: "unixtime",
+			value:  strconv.FormatInt(testTime.Unix(), 10),
+			want:   testTime.Unix() * 1e9,
+		},
+		{
+			name:   "unixtimemilli_ok",
+			format: "unixtimemilli",
+			value:  strconv.FormatInt(testTime.UnixMilli(), 10),
+			want:   testTime.UnixMilli() * 1e6,
+		},
+		{
+			name:   "unixtimemicro_ok",
+			format: "unixtimemicro",
+			value:  strconv.FormatInt(testTime.UnixMicro(), 10),
+			want:   testTime.UnixMicro() * 1e3,
+		},
+		{
+			name:   "unixtimenano_ok",
+			format: "unixtimenano",
+			value:  strconv.FormatInt(testTime.UnixNano(), 10),
+			want:   testTime.UnixNano(),
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := ParseTime(tt.format, tt.value)
+			require.NoError(t, err, "must be no error")
+			require.Equal(t, tt.want, got.UnixNano())
+		})
+	}
+}