LogQL: Add unwrap bytes() conversion function (#2876)

* Unwrap bytes: Added bytes() op to parser & lexer

* Unwrap bytes: Added bytes() conversion function

docs/sources/logql/_index.md

@@ -492,7 +492,9 @@ The unwrap expression is noted `| unwrap label_identifier` where the label ident
 Since label values are string, by default a conversion into a float (64bits) will be attempted, in case of failure the `__error__` label is added to the sample.
 Optionally the label identifier can be wrapped by a conversion function `| unwrap <function>(label_identifier)`, which will attempt to convert the label value from a specific format.
 
-We currently support only the function `duration_seconds` (or its short equivalent `duration`) which will convert the label value in seconds from the [go duration format](https://golang.org/pkg/time/#ParseDuration) (e.g `5m`, `24s30ms`).
+We currently support the functions: 
+- `duration_seconds(label_identifier)` (or its short equivalent `duration`) which will convert the label value in seconds from the [go duration format](https://golang.org/pkg/time/#ParseDuration) (e.g `5m`, `24s30ms`).
+- `bytes(label_identifier)` which will convert the label value to raw bytes applying the bytes unit  (e.g. `5 MiB`, `3k`, `1G`).
 
 Supported function for operating over unwrapped ranges are:
 

pkg/logql/ast.go

@@ -528,6 +528,7 @@ const (
 	OpUnwrap = "unwrap"
 
 	// conversion Op
+	OpConvBytes           = "bytes"
 	OpConvDuration        = "duration"
 	OpConvDurationSeconds = "duration_seconds"
 )

pkg/logql/ast_test.go

@@ -85,6 +85,11 @@ func Test_SampleExpr_String(t *testing.T) {
 		/
 			count_over_time({namespace="tns"} | logfmt | label_format foo=bar[5m])
 		)`,
+		`sum by (job) (
+			sum_over_time({namespace="tns"} |= "level=error" | json | foo=5 and bar<25ms | unwrap bytes(latency)[5m])
+		/
+			count_over_time({namespace="tns"} | logfmt | label_format foo=bar[5m])
+		)`,
 		`sum by (job) (
 			sum_over_time(
 				{namespace="tns"} |= "level=error" | json | avg=5 and bar<25ms | unwrap duration(latency) [5m]

pkg/logql/expr.y

@@ -89,7 +89,7 @@ import (
 %token <val>      MATCHERS LABELS EQ RE NRE OPEN_BRACE CLOSE_BRACE OPEN_BRACKET CLOSE_BRACKET COMMA DOT PIPE_MATCH PIPE_EXACT
                   OPEN_PARENTHESIS CLOSE_PARENTHESIS BY WITHOUT COUNT_OVER_TIME RATE SUM AVG MAX MIN COUNT STDDEV STDVAR BOTTOMK TOPK
                   BYTES_OVER_TIME BYTES_RATE BOOL JSON REGEXP LOGFMT PIPE LINE_FMT LABEL_FMT UNWRAP AVG_OVER_TIME SUM_OVER_TIME MIN_OVER_TIME
-                  MAX_OVER_TIME STDVAR_OVER_TIME STDDEV_OVER_TIME QUANTILE_OVER_TIME DURATION_CONV DURATION_SECONDS_CONV
+                  MAX_OVER_TIME STDVAR_OVER_TIME STDDEV_OVER_TIME QUANTILE_OVER_TIME BYTES_CONV DURATION_CONV DURATION_SECONDS_CONV
 
 // Operators are listed with increasing precedence.
 %left <binOp> OR
@@ -146,7 +146,8 @@ unwrapExpr:
   ;
 
 convOp:
-    DURATION_CONV           { $$ = OpConvDuration }
+    BYTES_CONV              { $$ = OpConvBytes }
+  | DURATION_CONV           { $$ = OpConvDuration }
   | DURATION_SECONDS_CONV   { $$ = OpConvDurationSeconds }
   ;
 

pkg/logql/functions.go

@@ -50,6 +50,8 @@ func (r rangeAggregationExpr) extractor(gr *grouping, all bool) (log.SampleExtra
 	if r.left.unwrap != nil {
 		var convOp string
 		switch r.left.unwrap.operation {
+		case OpConvBytes:
+			convOp = log.ConvertBytes
 		case OpConvDuration, OpConvDurationSeconds:
 			convOp = log.ConvertDuration
 		default:

pkg/logql/lex.go

@@ -86,6 +86,7 @@ var functionTokens = map[string]int{
 	OpTypeTopK:    TOPK,
 
 	// conversion Op
+	OpConvBytes:           BYTES_CONV,
 	OpConvDuration:        DURATION_CONV,
 	OpConvDurationSeconds: DURATION_SECONDS_CONV,
 }

pkg/logql/log/metrics_extraction.go

@@ -7,9 +7,12 @@ import (
 
 	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/pkg/labels"
+
+	"github.com/dustin/go-humanize"
 )
 
 const (
+	ConvertBytes    = "bytes"
 	ConvertDuration = "duration"
 	ConvertFloat    = "float"
 )
@@ -117,6 +120,8 @@ func LabelExtractorWithStages(
 ) (SampleExtractor, error) {
 	var convFn convertionFn
 	switch conversion {
+	case ConvertBytes:
+		convFn = convertBytes
 	case ConvertDuration:
 		convFn = convertDuration
 	case ConvertFloat:
@@ -195,3 +200,11 @@ func convertDuration(v string) (float64, error) {
 	}
 	return d.Seconds(), nil
 }
+
+func convertBytes(v string) (float64, error) {
+	b, err := humanize.ParseBytes(v)
+	if err != nil {
+		return 0, err
+	}
+	return float64(b), nil
+}

pkg/logql/log/metrics_extraction_test.go

@@ -90,6 +90,24 @@ func Test_labelSampleExtractor_Extract(t *testing.T) {
 			},
 			true,
 		},
+		{
+			"convert bytes",
+			mustSampleExtractor(LabelExtractorWithStages(
+				"foo", ConvertBytes, []string{"bar", "buzz"}, false, false, nil, NoopStage,
+			)),
+			labels.Labels{
+				{Name: "foo", Value: "13 MiB"},
+				{Name: "bar", Value: "foo"},
+				{Name: "buzz", Value: "blip"},
+				{Name: "namespace", Value: "dev"},
+			},
+			13 * 1024 * 1024,
+			labels.Labels{
+				{Name: "bar", Value: "foo"},
+				{Name: "buzz", Value: "blip"},
+			},
+			true,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

pkg/logql/parser_test.go

@@ -1228,6 +1228,27 @@ func TestParse(t *testing.T) {
 				OpRangeTypeStdvar, nil, nil,
 			),
 		},
+		{
+			in: `sum_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms| unwrap bytes(foo) [5m])`,
+			exp: newRangeAggregationExpr(
+				newLogRange(&pipelineExpr{
+					left: newMatcherExpr([]*labels.Matcher{{Type: labels.MatchEqual, Name: "namespace", Value: "tns"}}),
+					pipeline: MultiStageExpr{
+						newLineFilterExpr(nil, labels.MatchEqual, "level=error"),
+						newLabelParserExpr(OpParserTypeJSON, ""),
+						&labelFilterExpr{
+							LabelFilterer: log.NewAndLabelFilter(
+								log.NewNumericLabelFilter(log.LabelFilterGreaterThanOrEqual, "foo", 5),
+								log.NewDurationLabelFilter(log.LabelFilterLesserThan, "bar", 25*time.Millisecond),
+							),
+						},
+					},
+				},
+					5*time.Minute,
+					newUnwrapExpr("foo", OpConvBytes)),
+				OpRangeTypeSum, nil, nil,
+			),
+		},
 		{
 			in: `sum_over_time({namespace="tns"} |= "level=error" | json |foo>=5,bar<25ms| unwrap latency [5m])`,
 			exp: newRangeAggregationExpr(