Loki: add __line__ and __timestamp__ to label_format (grafana#6983)

<!--  Thanks for sending a pull request!  Before submitting:

1. Read our CONTRIBUTING.md guide
2. Name your PR as `<Feature Area>: Describe your change`.
  a. Do not end the title with punctuation. It will be added in the changelog.
  b. Start with an imperative verb. Example: Fix the latency between System A and System B.
  c. Use sentence case, not title case.
  d. Use a complete phrase or sentence. The PR title will appear in a changelog, so help other people understand what your change will be.
3. Rebase your PR if it gets out of sync with main
-->

**What this PR does / why we need it**:

We have the `__line__` and `__timestamp__` functions currently available in the `line_format` stage but not in the `label_format` stage. This PR fixes that.

Specifically `__timestamp__` can be valuable in a label because you can do things like

```
label_format hour_of_day=`{{__timestamp__ | date "15"}}`
```

To be able to conditionaly select log lines based on the hour of the day they were created, this can be useful for conditional alerting as an example.



**Which issue(s) this PR fixes**:
Fixes grafana#6402

**Special notes for your reviewer**:

<!--
Note about CHANGELOG entries, if a change adds:
* an important feature
* fixes an issue present in a previous release, 
* causes a change in operation that would be useful for an operator of Loki to know
then please add a CHANGELOG entry.

For documentation changes, build changes, simple fixes etc please skip this step. We are attempting to curate a changelog of the most relevant and important changes to be easier to ingest by end users of Loki.

Note about the upgrade guide, if this changes:
* default configuration values
* metric names or label names
* changes existing log lines such as the metrics.go query output line
* configuration parameters 
* anything to do with any API
* any other change that would require special attention or extra steps to upgrade
Please document clearly what changed AND what needs to be done in the upgrade guide.
-->
**Checklist**
- [x] Documentation added
- [ ] Tests updated
- [x] Is this an important fix or new feature? Add an entry in the `CHANGELOG.md`.
- [ ] Changes that require user attention or interaction to upgrade are documented in `docs/sources/upgrading/_index.md`

Signed-off-by: Edward Welch <[email protected]>

Author: slim-bean

CHANGELOG.md

@@ -5,6 +5,7 @@
 #### Loki
 
 ##### Enhancements
+* [6983](https://github.com/grafana/loki/pull/6983) **slim-bean**: `__timestamp__` and `__line__` are now available in the logql `label_format` query stage.
 * [6821](https://github.com/grafana/loki/pull/6821) **kavirajk**: Introduce new cache type `embedded-cache` which is an in-process cache system that runs loki without the need for an external cache (like memcached, redis, etc). It can be run in two modes `distributed: false` (default, and same as old `fifocache`) and `distributed: true` which runs cache in distributed fashion sharding keys across peers if Loki is run in microservices or SSD mode.
 * [6691](https://github.com/grafana/loki/pull/6691) **dannykopping**: Update production-ready Loki cluster in docker-compose
 * [6317](https://github.com/grafana/loki/pull/6317) **dannykoping**: General: add cache usage statistics

docs/sources/logql/template_functions.md

@@ -13,7 +13,7 @@ All labels are added as variables in the template engine. They can be referenced
 {{ .path }}
 ```
 
-Additionally you can also access the log line using the [`__line__`](#__line__) function.
+Additionally you can also access the log line using the [`__line__`](#__line__) function and the timestamp using the [`__timestamp__`](#__timestamp__) function.
 
 You can take advantage of [pipeline](https://golang.org/pkg/text/template/#hdr-Pipelines) to join together multiple functions.
 In a chained pipeline, the result of each command is passed as the last argument of the following command.

pkg/logql/log/fmt.go

@@ -90,6 +90,20 @@ var (
 	}
 )
 
+func addLineAndTimestampFunctions(currLine func() string, currTimestamp func() int64) map[string]interface{} {
+	functions := make(map[string]interface{}, len(functionMap)+2)
+	for k, v := range functionMap {
+		functions[k] = v
+	}
+	functions[functionLineName] = func() string {
+		return currLine()
+	}
+	functions[functionTimestampName] = func() time.Time {
+		return time.Unix(0, currTimestamp())
+	}
+	return functions
+}
+
 func init() {
 	sprigFuncMap := sprig.GenericFuncMap()
 	for _, v := range templateFunctions {
@@ -112,16 +126,13 @@ func NewFormatter(tmpl string) (*LineFormatter, error) {
 	lf := &LineFormatter{
 		buf: bytes.NewBuffer(make([]byte, 4096)),
 	}
-	functions := make(map[string]interface{}, len(functionMap)+1)
-	for k, v := range functionMap {
-		functions[k] = v
-	}
-	functions[functionLineName] = func() string {
+
+	functions := addLineAndTimestampFunctions(func() string {
 		return unsafeGetString(lf.currentLine)
-	}
-	functions[functionTimestampName] = func() time.Time {
-		return time.Unix(0, lf.currentTs)
-	}
+	}, func() int64 {
+		return lf.currentTs
+	})
+
 	t, err := template.New("line").Option("missingkey=zero").Funcs(functions).Parse(tmpl)
 	if err != nil {
 		return nil, fmt.Errorf("invalid line template: %w", err)
@@ -235,6 +246,9 @@ type labelFormatter struct {
 type LabelsFormatter struct {
 	formats []labelFormatter
 	buf     *bytes.Buffer
+
+	currentLine []byte
+	currentTs   int64
 }
 
 // NewLabelsFormatter creates a new formatter that can format multiple labels at once.
@@ -246,21 +260,29 @@ func NewLabelsFormatter(fmts []LabelFmt) (*LabelsFormatter, error) {
 	}
 	formats := make([]labelFormatter, 0, len(fmts))
 
+	lf := &LabelsFormatter{
+		buf: bytes.NewBuffer(make([]byte, 1024)),
+	}
+
+	functions := addLineAndTimestampFunctions(func() string {
+		return unsafeGetString(lf.currentLine)
+	}, func() int64 {
+		return lf.currentTs
+	})
+
 	for _, fm := range fmts {
 		toAdd := labelFormatter{LabelFmt: fm}
 		if !fm.Rename {
-			t, err := template.New("label").Option("missingkey=zero").Funcs(functionMap).Parse(fm.Value)
+			t, err := template.New("label").Option("missingkey=zero").Funcs(functions).Parse(fm.Value)
 			if err != nil {
 				return nil, fmt.Errorf("invalid template for label '%s': %s", fm.Name, err)
 			}
 			toAdd.tmpl = t
 		}
 		formats = append(formats, toAdd)
 	}
-	return &LabelsFormatter{
-		formats: formats,
-		buf:     bytes.NewBuffer(make([]byte, 1024)),
-	}, nil
+	lf.formats = formats
+	return lf, nil
 }
 
 func validate(fmts []LabelFmt) error {
@@ -279,7 +301,10 @@ func validate(fmts []LabelFmt) error {
 	return nil
 }
 
-func (lf *LabelsFormatter) Process(_ int64, l []byte, lbs *LabelsBuilder) ([]byte, bool) {
+func (lf *LabelsFormatter) Process(ts int64, l []byte, lbs *LabelsBuilder) ([]byte, bool) {
+	lf.currentLine = l
+	lf.currentTs = ts
+
 	var data interface{}
 	for _, f := range lf.formats {
 		if f.Rename {

pkg/logql/log/fmt_test.go

@@ -420,13 +420,43 @@ func Test_labelsFormatter_Format(t *testing.T) {
 				{Name: "__error_details__", Value: "template: label:1:2: executing \"label\" at <replace>: wrong number of args for replace: want 3 got 2"},
 			},
 		},
+		{
+			"line",
+			mustNewLabelsFormatter([]LabelFmt{NewTemplateLabelFmt("line", "{{ __line__ }}")}),
+			labels.Labels{{Name: "foo", Value: "blip"}, {Name: "bar", Value: "blop"}},
+			labels.Labels{
+				{Name: "foo", Value: "blip"},
+				{Name: "bar", Value: "blop"},
+				{Name: "line", Value: "test line"},
+			},
+		},
+		{
+			"timestamp",
+			mustNewLabelsFormatter([]LabelFmt{NewTemplateLabelFmt("ts", "{{ __timestamp__ | date \"2006-01-02\" }}")}),
+			labels.Labels{{Name: "foo", Value: "blip"}, {Name: "bar", Value: "blop"}},
+			labels.Labels{
+				{Name: "foo", Value: "blip"},
+				{Name: "bar", Value: "blop"},
+				{Name: "ts", Value: "2022-08-26"},
+			},
+		},
+		{
+			"timestamp_unix",
+			mustNewLabelsFormatter([]LabelFmt{NewTemplateLabelFmt("ts", "{{ __timestamp__ | unixEpoch }}")}),
+			labels.Labels{{Name: "foo", Value: "blip"}, {Name: "bar", Value: "blop"}},
+			labels.Labels{
+				{Name: "foo", Value: "blip"},
+				{Name: "bar", Value: "blop"},
+				{Name: "ts", Value: "1661518453"},
+			},
+		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			builder := NewBaseLabelsBuilder().ForLabels(tt.in, tt.in.Hash())
 			builder.Reset()
-			_, _ = tt.fmter.Process(0, nil, builder)
+			_, _ = tt.fmter.Process(1661518453244672570, []byte("test line"), builder)
 			sort.Sort(tt.want)
 			require.Equal(t, tt.want, builder.LabelsResult().Labels())
 		})