internal/command: evaluate log lines as templates with --template

The log command streamed each line verbatim, so fixtures with absolute timestamps eventually age out of time-windowed consumers (for example a transform that filters on now-90d). Re-dating the fixtures by hand only resets a ticking clock. Add a --template flag that evaluates each line as a Go text/template before sending, using the same function set as the http-server config (env, hostname, sum, file, glob, minify_json, now). A streamed line can then keep its timestamps current, for example: "eventTime":"{{ (now "-720h").Format "2006-01-02T15:04:05Z07:00" }}" Each line is rendered independently, and a template error aborts the run reporting the offending line number. To keep one implementation, the template function map and helpers move from internal/httpserver into a new internal/templates package that both the http-server and the log command use. Updates elastic/integrations#20792 Co-authored-by: Opus 4.8 High
kcreddy · Aug 20, 2026 · Aug 19, 2026 · Aug 19, 2026 · 20c4e4e2b28bb8da22d103dab7ec22940cce9127
commit 20c4e4e2b28bb8da22d103dab7ec22940cce9127
diff --git a/.changelog/213.txt b/.changelog/213.txt
@@ -0,0 +1,3 @@
+```release-note:enhancement
+log: Added a `--template` flag that evaluates each log line as a Go template before sending, using the same functions as the http-server config (including `now`). This lets streamed fixtures keep timestamps current relative to wall-clock time.
+```
diff --git a/README.md b/README.md
@@ -18,7 +18,9 @@ stream is a test utility for streaming data via:
 
 Input data can be read from:
 
-- log file - Newline delimited files are streamed line by line.
+- log file - Newline delimited files are streamed line by line. Lines can
+  optionally be evaluated as Go templates before sending (see
+  [Log input reference](#log-input-reference)).
 - pcap file - Each packet's transport layer payload is streamed as a packet.
   Useful for replaying netflow and IPFIX captures. Both pcap and pcapng files are
   supported, including gzip compressed ones.
@@ -52,6 +54,40 @@ certificates. There is no shell inside it.
 `SIGINT` and `SIGTERM` are recognized, because those are the only signals the Go
 runtime emulates there.
 
+## Log input reference
+
+The `log` subcommand streams a newline delimited file line by line to the
+configured output:
+
+```bash
+stream log --addr=localhost:8080 -p=tcp ./events.log
+```
+
+### Templating
+
+Pass `--template` to evaluate each line as a [Go template](https://golang.org/pkg/text/template/)
+before it is sent. This is useful for values that must stay current relative to
+wall-clock time, such as timestamps behind a rolling retention window:
+
+```bash
+stream log --addr=localhost:8080 -p=tcp --template ./events.log
+```
+
+With `--template` the line:
+
+```json
+{"eventTime":"{{ (now "-720h").Format "2006-01-02T15:04:05Z07:00" }}"}
+```
+
+is rendered to a timestamp 30 days in the past each time it is streamed. Each
+line is rendered independently, and a template error aborts the run and reports
+the offending line number.
+
+The same functions available to the `http-server` config `body`/`headers` are
+available here: `env`, `hostname`, `sum`, `file`, `glob`, `minify_json`, and
+`now` (see [HTTP Server mock reference](#http-server-mock-reference) for their
+descriptions). Missing template keys render as their zero value.
+
 ## HTTP Server mock reference
 
 `stream` can also serve logs setting up a complete HTTP mock server.

diff --git a/internal/command/log.go b/internal/command/log.go
@@ -6,19 +6,22 @@ package command
 
 import (
 	"bufio"
+	"fmt"
 	"os"
 
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 
 	"github.com/elastic/stream/internal/cmdutil"
 	"github.com/elastic/stream/internal/output"
+	"github.com/elastic/stream/internal/templates"
 )
 
 type logRunner struct {
-	logger *zap.SugaredLogger
-	cmd    *cobra.Command
-	out    *output.Options
+	logger   *zap.SugaredLogger
+	cmd      *cobra.Command
+	out      *output.Options
+	template bool
 }
 
 func newLogRunner(options *output.Options, logger *zap.Logger) *cobra.Command {
@@ -31,6 +34,9 @@ func newLogRunner(options *output.Options, logger *zap.Logger) *cobra.Command {
 		},
 	}
 
+	r.cmd.PersistentFlags().BoolVar(&r.template, "template", false,
+		"evaluate each log line as a Go text/template before sending, using the same functions as the http-server config (e.g. now)")
+
 	r.cmd.RunE = func(_ *cobra.Command, args []string) error {
 		r.logger = logger.Sugar().With("address", options.Addr)
 		return r.Run(args)
@@ -80,7 +86,17 @@ func (r *logRunner) sendLog(path string, out output.Output) error {
 		}
 
 		logger.Debugw("Sending log line.", "line_number", totalLines+1)
-		n, err := out.Write(s.Bytes())
+
+		line := s.Bytes()
+		if r.template {
+			rendered, err := templates.Render(s.Text())
+			if err != nil {
+				return fmt.Errorf("failed to render template on line %d of %s: %w", totalLines+1, path, err)
+			}
+			line = rendered
+		}
+
+		n, err := out.Write(line)
 		if err != nil {
 			return err
 		}

diff --git a/internal/command/log_test.go b/internal/command/log_test.go
@@ -0,0 +1,86 @@
+// Licensed to Elasticsearch B.V. under one or more agreements.
+// Elasticsearch B.V. licenses this file to you under the Apache 2.0 License.
+// See the LICENSE file in the project root for more information.
+
+package command
+
+import (
+	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/spf13/cobra"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+
+	"github.com/elastic/stream/internal/output"
+)
+
+func newLogTestRunner(t *testing.T, tmpl bool) *logRunner {
+	t.Helper()
+
+	cmd := &cobra.Command{}
+	cmd.SetContext(context.Background())
+	return &logRunner{
+		logger:   zap.NewNop().Sugar(),
+		cmd:      cmd,
+		out:      &output.Options{MaxLogLineSize: 500 * 1024},
+		template: tmpl,
+	}
+}
+
+func writeLog(t *testing.T, lines string) string {
+	t.Helper()
+
+	path := filepath.Join(t.TempDir(), "test.log")
+	require.NoError(t, os.WriteFile(path, []byte(lines), 0o600))
+	return path
+}
+
+func TestSendLog(t *testing.T) {
+	t.Run("sends lines verbatim without --template", func(t *testing.T) {
+		// A brace expression must be left untouched when templating is off.
+		const line = `{"eventTime":"{{ now }}"}`
+		path := writeLog(t, line+"\n")
+
+		r := newLogTestRunner(t, false)
+		out := &memoryOutput{}
+		require.NoError(t, r.sendLog(path, out))
+
+		require.Equal(t, []string{line}, out.payloads())
+	})
+
+	t.Run("renders each line with --template", func(t *testing.T) {
+		const line = `{"eventTime":"{{ (now "-720h").Format "2006-01-02T15:04:05Z07:00" }}"}`
+		path := writeLog(t, line+"\n")
+		want := time.Now().UTC().Add(-720 * time.Hour)
+
+		r := newLogTestRunner(t, true)
+		out := &memoryOutput{}
+		require.NoError(t, r.sendLog(path, out))
+
+		payloads := out.payloads()
+		require.Len(t, payloads, 1)
+
+		var result struct {
+			EventTime string `json:"eventTime"`
+		}
+		require.NoError(t, json.Unmarshal([]byte(payloads[0]), &result))
+		ts, err := time.Parse(time.RFC3339, result.EventTime)
+		require.NoError(t, err)
+		assert.WithinDuration(t, want, ts, time.Minute)
+	})
+
+	t.Run("reports an error for an invalid template", func(t *testing.T) {
+		path := writeLog(t, `{{ now "bogus" }}`+"\n")
+
+		r := newLogTestRunner(t, true)
+		err := r.sendLog(path, &memoryOutput{})
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "line 1")
+	})
+}
diff --git a/internal/httpserver/config.go b/internal/httpserver/config.go
@@ -5,17 +5,13 @@
 package httpserver
 
 import (
-	"encoding/json"
 	"errors"
-	"fmt"
-	"os"
-	"path/filepath"
-	"strings"
 	"text/template"
-	"time"
 
 	ucfg "github.com/elastic/go-ucfg"
 	"github.com/elastic/go-ucfg/yaml"
+
+	"github.com/elastic/stream/internal/templates"
 )
 
 type config struct {
@@ -52,15 +48,7 @@ type tpl struct {
 func (t *tpl) Unpack(in string) error {
 	parsed, err := template.New("").
 		Option("missingkey=zero").
-		Funcs(template.FuncMap{
-			"env":         env,
-			"hostname":    hostname,
-			"sum":         sum,
-			"file":        file,
-			"glob":        filepath.Glob,
-			"minify_json": minify,
-			"now":         now,
-		}).
+		Funcs(templates.Funcs()).
 		Parse(in)
 	if err != nil {
 		return err
@@ -88,48 +76,3 @@ func newConfigFromFile(file string) (*config, error) {
 
 	return &config, nil
 }
-
-func env(key string) string {
-	return os.Getenv(key)
-}
-
-func hostname() string {
-	h, _ := os.Hostname()
-	return h
-}
-
-func sum(a, b int) int {
-	return a + b
-}
-
-func file(path string) (string, error) {
-	b, err := os.ReadFile(path)
-	if err != nil {
-		return "", err
-	}
-	return string(b), nil
-}
-
-func minify(body string) (string, error) {
-	var buf strings.Builder
-	enc := json.NewEncoder(&buf)
-	enc.SetEscapeHTML(false)
-	err := enc.Encode(json.RawMessage(body))
-	return strings.TrimSpace(buf.String()), err
-}
-
-// now returns the current UTC time. An optional Go duration string
-// offsets the result (e.g. "-720h" for 30 days ago). The returned
-// time.Time value exposes its methods to templates, so callers can
-// format it as needed: {{ (now).Format "2006-01-02" }}.
-func now(offset ...string) (time.Time, error) {
-	t := time.Now().UTC()
-	if len(offset) == 0 {
-		return t, nil
-	}
-	d, err := time.ParseDuration(offset[0])
-	if err != nil {
-		return time.Time{}, fmt.Errorf("invalid duration %q: %w", offset[0], err)
-	}
-	return t.Add(d), nil
-}
diff --git a/internal/httpserver/httpserver_test.go b/internal/httpserver/httpserver_test.go
@@ -297,48 +297,6 @@ func TestHTTPServer(t *testing.T) {
 	})
 }
 
-func TestNow(t *testing.T) {
-	t.Run("no offset", func(t *testing.T) {
-		before := time.Now().UTC()
-		got, err := now()
-		if err != nil {
-			t.Fatalf("now() error: %v", err)
-		}
-		if got.Before(before.Add(-time.Second)) || got.After(time.Now().UTC().Add(time.Second)) {
-			t.Errorf("now() = %s; want within 1s of current time", got)
-		}
-	})
-
-	t.Run("negative offset", func(t *testing.T) {
-		before := time.Now().UTC().Add(-24 * time.Hour)
-		got, err := now("-24h")
-		if err != nil {
-			t.Fatalf("now(%q) error: %v", "-24h", err)
-		}
-		if got.Before(before.Add(-time.Second)) || got.After(before.Add(time.Second)) {
-			t.Errorf("now(%q) = %s; want within 1s of %s", "-24h", got, before)
-		}
-	})
-
-	t.Run("positive offset", func(t *testing.T) {
-		expected := time.Now().UTC().Add(2 * time.Hour)
-		got, err := now("2h")
-		if err != nil {
-			t.Fatalf("now(%q) error: %v", "2h", err)
-		}
-		if got.Before(expected.Add(-time.Second)) || got.After(expected.Add(time.Second)) {
-			t.Errorf("now(%q) = %s; want within 1s of %s", "2h", got, expected)
-		}
-	})
-
-	t.Run("invalid offset", func(t *testing.T) {
-		_, err := now("bogus")
-		if err == nil {
-			t.Error("now(\"bogus\") error = nil; want error")
-		}
-	})
-}
-
 func TestRunAsSequence(t *testing.T) {
 	cfg := `---
   as_sequence: true