release: 0.5.0: Pre-1.0 security review, domain-admin role, doc consolidation
A bounce could be recorded as a successful delivery. The log-tailer's Fixed: - A bounce could be recorded as a successful delivery. The log-tailer's - Log-tailer resumes where it stopped instead of jumping to end-of-file on - Level-2 rate limit no longer overshoots under concurrency: messages that Changed: - The project has a single public home: `github.com/mixeme/selfpost`. Codeberg - Code comments no longer cite the archived specification. References like - `docs/code-review.md` is gone. Its plan is finished — phases 0 (bar the - `docs/architecture.md` gained a *Code layers* section: a diagram of - Phase 1 doc/code hygiene (`docs/code-review.md`): removed ~30 stale … (1 more in CHANGELOG) Security: - Pre-release security review (plan § D, model Fable, 2026-08-06): full pass - `saslpasswd2` argv: the application login is now passed after a `--` Added: - Optional password encryption for the two secret-bearing downloads (plan - Domain import accepts an encrypted export: the envelope is detected by its - `selfpost-backup` writes encrypted archives and reads them back: - `TestParseDelivery` covers the exotic mail.log shapes the review asked for - docs: README *Encrypting a backup or export*; `docs/security.md` § … (14 more in CHANGELOG) Removed: - docs: `security.md` — accepted risk «restore old backup revives session rows» Changed: - docs: `implementation-plan.md` trimmed to the sole open v1.x gate — pre-release - docs: `architecture.md` — sessions (SQLite, idle renew, password change), - docs: `development.md` — expanded e2e stack and `release.yml` matrix workflow. - docs: `security.md` — accepted risk for send-log rows stuck at `queued` after - docs: `roadmap.md` — optional send-log / `mail.log` follow-ups under v1.x tail. … (10 more in CHANGELOG) Full detail: CHANGELOG.md (0.5.0).
This commit is contained in:
+50
-21
@@ -1,9 +1,10 @@
|
||||
// Package logtail follows Postfix's mail.log and reconciles the send-log
|
||||
// delivery statuses the journal-milter could not know at receive time (spec
|
||||
// 7.3). A milter row starts life as "queued"; Postfix only decides sent /
|
||||
// deferred / bounced later, per recipient, and reports it in mail.log. This
|
||||
// package parses those lines by queue-id + recipient and advances the matching
|
||||
// rows, and prunes rows past the retention window.
|
||||
// delivery statuses the journal-milter could not know at receive time
|
||||
// (architecture.md § Persistence). A milter row starts life as "queued";
|
||||
// Postfix only decides sent / deferred / bounced later, per recipient, and
|
||||
// reports it in mail.log. This package parses those lines by queue-id +
|
||||
// recipient and advances the matching rows, and prunes rows past the retention
|
||||
// window.
|
||||
package logtail
|
||||
|
||||
import (
|
||||
@@ -17,14 +18,17 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"codeberg.org/mix/selfpost/internal/store"
|
||||
"github.com/mixeme/selfpost/internal/store"
|
||||
)
|
||||
|
||||
// StatusStore is the slice of the store the log-tailer needs. *store.Store
|
||||
// satisfies it.
|
||||
// StatusStore is the slice of the store the log-tailer needs: advancing
|
||||
// delivery statuses, pruning the retention window, and remembering how far into
|
||||
// mail.log it has read. *store.Store satisfies it.
|
||||
type StatusStore interface {
|
||||
UpdateStatus(queueID, recipient, status string) (int64, error)
|
||||
DeleteSendLogBefore(cutoff time.Time) (int64, error)
|
||||
LogtailState(path string) (store.LogtailState, bool, error)
|
||||
SaveLogtailState(path string, st store.LogtailState) error
|
||||
}
|
||||
|
||||
// pollInterval is how often the tail loop checks for new bytes / rotation. It
|
||||
@@ -36,7 +40,7 @@ const (
|
||||
// startup). The window itself is configurable; the cadence need not be.
|
||||
retentionInterval = 6 * time.Hour
|
||||
// defaultRetentionDays applies when the configured value is unset/invalid
|
||||
// (spec 7.3).
|
||||
// (README § Environment variables: SEND_LOG_RETENTION_DAYS).
|
||||
defaultRetentionDays = 90
|
||||
)
|
||||
|
||||
@@ -47,7 +51,13 @@ const (
|
||||
//
|
||||
// The "<queue-id>: to=<addr>, …, status=<word>" shape is specific to the
|
||||
// delivery agents; qmgr/smtpd/cleanup lines do not match.
|
||||
var deliveryRe = regexp.MustCompile(`\b([0-9A-Za-z]+): to=<([^>]*)>,.*\bstatus=(\w+)`)
|
||||
//
|
||||
// The run before status= is lazy on purpose. Postfix appends the remote
|
||||
// server's reply verbatim, so a greedy match would take the *last* status= on
|
||||
// the line — and that one can come from the reply text, which the far end
|
||||
// controls. A bounce whose reply quoted "status=sent" would then be filed as a
|
||||
// success. The real field is always the first one after to=<…>.
|
||||
var deliveryRe = regexp.MustCompile(`\b([0-9A-Za-z]+): to=<([^>]*)>,.*?\bstatus=(\w+)`)
|
||||
|
||||
// parseDelivery extracts (queue-id, recipient, status) from a mail.log line.
|
||||
// ok is false for lines that are not recognised delivery results.
|
||||
@@ -73,12 +83,13 @@ func parseDelivery(line string) (queueID, recipient, status string, ok bool) {
|
||||
}
|
||||
|
||||
// Run follows path and updates send-log statuses until ctx is cancelled, while
|
||||
// a background sweep prunes rows older than retentionDays. It returns nil on a
|
||||
// clean shutdown.
|
||||
// a background sweep prunes rows older than retentionDays. Reading resumes at
|
||||
// the offset the previous run persisted, so a restart parses the delivery lines
|
||||
// written while the panel was down. It returns nil on a clean shutdown.
|
||||
func Run(ctx context.Context, path string, st StatusStore, retentionDays int) error {
|
||||
go retentionLoop(ctx, st, retentionDays)
|
||||
|
||||
return follow(ctx, path, func(line string) {
|
||||
return follow(ctx, path, &tracker{st: st, path: path}, func(line string) {
|
||||
queueID, recipient, status, ok := parseDelivery(line)
|
||||
if !ok {
|
||||
return
|
||||
@@ -120,10 +131,10 @@ func retentionLoop(ctx context.Context, st StatusStore, retentionDays int) {
|
||||
}
|
||||
|
||||
// TailLines returns up to n of the most recent lines from path, for the
|
||||
// panel's mail.log monitoring view (spec 7.2.13). It is a one-shot,
|
||||
// point-in-time read on request — unrelated to the background follow loop
|
||||
// above — that reads backwards in chunks so it stays cheap against a
|
||||
// multi-megabyte log rather than reading the whole file every poll.
|
||||
// panel's mail.log monitoring view (architecture.md § Panel HTTP surface). It
|
||||
// is a one-shot, point-in-time read on request — unrelated to the background
|
||||
// follow loop above — that reads backwards in chunks so it stays cheap against
|
||||
// a multi-megabyte log rather than reading the whole file every poll.
|
||||
func TailLines(path string, n int) ([]string, error) {
|
||||
f, err := os.Open(path)
|
||||
if err != nil {
|
||||
@@ -166,10 +177,11 @@ func TailLines(path string, n int) ([]string, error) {
|
||||
}
|
||||
|
||||
// follow tails path line by line, calling handle for each complete line, until
|
||||
// ctx is cancelled. It starts at end-of-file (so a restart does not reprocess
|
||||
// history) and reopens the file when it is rotated (inode change from
|
||||
// ctx is cancelled. Where it starts is tr's decision (a persisted offset, the
|
||||
// start of a file that changed while the panel was down, or end-of-file on a
|
||||
// first ever run); it reopens the file when it is rotated (inode change from
|
||||
// logrotate's create, or truncation from copytruncate) so nothing is missed.
|
||||
func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
func follow(ctx context.Context, path string, tr *tracker, handle func(string)) error {
|
||||
var (
|
||||
f *os.File
|
||||
r *bufio.Reader
|
||||
@@ -199,7 +211,7 @@ func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
|
||||
// The container may start before Postfix has created mail.log; wait for it.
|
||||
for {
|
||||
if err := openAt(0, io.SeekEnd); err == nil {
|
||||
if err := openAt(0, io.SeekStart); err == nil {
|
||||
break
|
||||
}
|
||||
select {
|
||||
@@ -208,6 +220,10 @@ func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
case <-time.After(pollInterval):
|
||||
}
|
||||
}
|
||||
if _, err := f.Seek(tr.resume(f), io.SeekStart); err != nil {
|
||||
log.Printf("log-tailer: seek %s: %v", path, err)
|
||||
}
|
||||
r.Reset(f) // the reader buffered from the pre-seek position
|
||||
defer func() {
|
||||
if f != nil {
|
||||
f.Close()
|
||||
@@ -231,11 +247,20 @@ func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
}
|
||||
}
|
||||
|
||||
// read returns how many bytes of the open file have actually been consumed:
|
||||
// the descriptor position less the partial line bufio handed back at EOF,
|
||||
// which is re-read (and completed) on the next drain or the next start.
|
||||
read := func() int64 {
|
||||
pos, _ := f.Seek(0, io.SeekCurrent)
|
||||
return pos - int64(len(pending))
|
||||
}
|
||||
|
||||
ticker := time.NewTicker(pollInterval)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
tr.record(f, read(), true) // shutdown: the next start resumes here
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
drain()
|
||||
@@ -252,8 +277,12 @@ func follow(ctx context.Context, path string, handle func(string)) error {
|
||||
drain()
|
||||
if err := openAt(0, io.SeekStart); err != nil {
|
||||
log.Printf("log-tailer: reopen %s: %v", path, err)
|
||||
continue
|
||||
}
|
||||
tr.adopt(f)
|
||||
continue
|
||||
}
|
||||
tr.record(f, read(), false)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,11 +4,12 @@ import (
|
||||
"context"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"codeberg.org/mix/selfpost/internal/store"
|
||||
"github.com/mixeme/selfpost/internal/store"
|
||||
)
|
||||
|
||||
func TestParseDelivery(t *testing.T) {
|
||||
@@ -60,6 +61,59 @@ func TestParseDelivery(t *testing.T) {
|
||||
line: "host postfix/smtpd[10]: 41E862C00D9E: client=unknown[203.0.113.7]",
|
||||
wantOK: false,
|
||||
},
|
||||
{
|
||||
// The remote server's reply is quoted verbatim at the end of the
|
||||
// line and is entirely attacker-influenced text. A "status=" that
|
||||
// appears in there must not win over the real field, or a bounce
|
||||
// would be filed as a success.
|
||||
name: "status= quoted in the remote reply does not win",
|
||||
line: "host postfix/smtp[26]: 9F1A2C00D9E: to=<a@example.net>, relay=mx.example.net[203.0.113.9]:25, dsn=5.1.1, status=bounced (host mx.example.net said: 550 5.1.1 unknown status=sent (in reply to RCPT TO command))",
|
||||
wantOK: true,
|
||||
queueID: "9F1A2C00D9E",
|
||||
recipient: "a@example.net",
|
||||
status: store.StatusBounced,
|
||||
},
|
||||
{
|
||||
// Postfix logs the null sender's own delivery (double bounce) with
|
||||
// an empty recipient. It parses, and the empty recipient simply
|
||||
// matches no send-log row — the panel only ever records mail it
|
||||
// accepted from an authenticated client.
|
||||
name: "null recipient parses with an empty address",
|
||||
line: "host postfix/smtp[26]: A1B2C3: to=<>, relay=none, delay=0.1, dsn=2.0.0, status=sent (250 OK)",
|
||||
wantOK: true,
|
||||
queueID: "A1B2C3",
|
||||
recipient: "",
|
||||
status: store.StatusSent,
|
||||
},
|
||||
{
|
||||
// An alias/virtual expansion carries orig_to= as well; the address
|
||||
// the message was actually delivered to is the one in to=.
|
||||
name: "orig_to is ignored in favour of to",
|
||||
line: "host postfix/lmtp[26]: 4Xk9tS1abcz: to=<real@example.net>, orig_to=<alias@example.net>, relay=x, dsn=2.0.0, status=sent (ok)",
|
||||
wantOK: true,
|
||||
queueID: "4Xk9tS1abcz",
|
||||
recipient: "real@example.net",
|
||||
status: store.StatusSent,
|
||||
},
|
||||
{
|
||||
// Postfix's own delivery agents write these two, but neither is a
|
||||
// final result we model: "deliverable" comes from address
|
||||
// verification probes, and anything unrecognised is dropped rather
|
||||
// than guessed at, leaving the row in its previous state.
|
||||
name: "unknown status word is not a delivery result",
|
||||
line: "host postfix/smtp[26]: BEEF01: to=<a@example.net>, relay=x, status=deliverable (ok)",
|
||||
wantOK: false,
|
||||
},
|
||||
{
|
||||
name: "status matching is case-sensitive, as Postfix writes it",
|
||||
line: "host postfix/smtp[26]: BEEF02: to=<a@example.net>, relay=x, dsn=4.0.0, status=Deferred (connect timed out)",
|
||||
wantOK: false,
|
||||
},
|
||||
{
|
||||
name: "cleanup message-id line ignored",
|
||||
line: "host postfix/cleanup[12]: BEEF03: message-id=<x@example.com>",
|
||||
wantOK: false,
|
||||
},
|
||||
}
|
||||
for _, c := range cases {
|
||||
t.Run(c.name, func(t *testing.T) {
|
||||
@@ -77,10 +131,16 @@ func TestParseDelivery(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// captureStore records UpdateStatus calls for the follow integration test.
|
||||
// captureStore records UpdateStatus calls for the follow integration test and
|
||||
// keeps the persisted read offset in memory, so a "restart" in a test is a
|
||||
// second Run against the same captureStore.
|
||||
type captureStore struct {
|
||||
mu sync.Mutex
|
||||
calls []string
|
||||
|
||||
state store.LogtailState
|
||||
haveState bool
|
||||
stateErr error
|
||||
}
|
||||
|
||||
func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, error) {
|
||||
@@ -92,12 +152,37 @@ func (c *captureStore) UpdateStatus(queueID, recipient, status string) (int64, e
|
||||
|
||||
func (c *captureStore) DeleteSendLogBefore(time.Time) (int64, error) { return 0, nil }
|
||||
|
||||
func (c *captureStore) LogtailState(string) (store.LogtailState, bool, error) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.stateErr != nil {
|
||||
return store.LogtailState{}, false, c.stateErr
|
||||
}
|
||||
return c.state, c.haveState, nil
|
||||
}
|
||||
|
||||
func (c *captureStore) SaveLogtailState(_ string, st store.LogtailState) error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.stateErr != nil {
|
||||
return c.stateErr
|
||||
}
|
||||
c.state, c.haveState = st, true
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *captureStore) snapshot() []string {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return append([]string(nil), c.calls...)
|
||||
}
|
||||
|
||||
func (c *captureStore) reset() {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.calls = nil
|
||||
}
|
||||
|
||||
// TestFollowTailsAndRotates writes delivery lines to a log file, then rotates
|
||||
// it (rename + fresh create, as logrotate does) and writes more, asserting the
|
||||
// tailer picks up lines from both the original and rotated file.
|
||||
@@ -142,6 +227,80 @@ func TestFollowTailsAndRotates(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestFollowResumesAfterRestart covers the persisted read offset: a restart
|
||||
// must parse the delivery lines written while the tailer was down (rows that
|
||||
// would otherwise stay "queued" forever), without re-parsing what it already
|
||||
// read, and must fall back to reading the whole file when the log was rotated
|
||||
// or recreated in the meantime.
|
||||
func TestFollowResumesAfterRestart(t *testing.T) {
|
||||
old := pollInterval
|
||||
pollInterval = 10 * time.Millisecond
|
||||
t.Cleanup(func() { pollInterval = old })
|
||||
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "mail.log")
|
||||
// A head longer than fingerprintSize, so the file stays identifiable across
|
||||
// the restart; the lines themselves predate the first start and are ignored.
|
||||
seed := strings.Repeat("host postfix/qmgr[1]: seed line, not a delivery\n", 20)
|
||||
if err := os.WriteFile(path, []byte(seed), 0o644); err != nil {
|
||||
t.Fatalf("seed log: %v", err)
|
||||
}
|
||||
|
||||
cs := &captureStore{}
|
||||
stop := startRun(t, path, cs)
|
||||
appendLine(t, path, "host postfix/smtp[1]: Q1: to=<a@example.net>, dsn=2.0.0, status=sent (ok)")
|
||||
waitFor(t, func() bool { return contains(cs.snapshot(), "Q1|a@example.net|sent") })
|
||||
stop() // persists the offset past Q1
|
||||
|
||||
// Down: Postfix keeps delivering.
|
||||
appendLine(t, path, "host postfix/smtp[1]: Q2: to=<b@example.net>, dsn=2.0.0, status=sent (ok)")
|
||||
|
||||
cs.reset()
|
||||
stop = startRun(t, path, cs)
|
||||
waitFor(t, func() bool { return contains(cs.snapshot(), "Q2|b@example.net|sent") })
|
||||
if contains(cs.snapshot(), "Q1|a@example.net|sent") {
|
||||
t.Fatal("resumed run re-parsed Q1: offset was not honoured")
|
||||
}
|
||||
stop()
|
||||
|
||||
// Down again, and this time the log is replaced (logrotate + fresh create).
|
||||
// The stored offset belongs to a file that no longer exists, so the new one
|
||||
// must be read from the start.
|
||||
if err := os.WriteFile(path, []byte(strings.Repeat("host postfix/qmgr[1]: fresh log after rotation\n", 20)+
|
||||
"host postfix/smtp[1]: Q3: to=<c@example.net>, dsn=5.1.1, status=bounced (nope)\n"), 0o644); err != nil {
|
||||
t.Fatalf("recreate log: %v", err)
|
||||
}
|
||||
|
||||
cs.reset()
|
||||
stop = startRun(t, path, cs)
|
||||
waitFor(t, func() bool { return contains(cs.snapshot(), "Q3|c@example.net|bounced") })
|
||||
stop()
|
||||
}
|
||||
|
||||
// startRun launches the tailer and returns a function that cancels it and waits
|
||||
// for a clean return, the way a panel restart bookends a run.
|
||||
func startRun(t *testing.T, path string, cs *captureStore) func() {
|
||||
t.Helper()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- Run(ctx, path, cs, 90) }()
|
||||
// follow() opens and seeks on start; give it a moment before the caller
|
||||
// appends, so the append is not raced by the initial open.
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
return func() {
|
||||
t.Helper()
|
||||
cancel()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("Run: %v", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("Run did not return after cancel")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func appendLine(t *testing.T, path, line string) {
|
||||
t.Helper()
|
||||
f, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
|
||||
|
||||
@@ -0,0 +1,136 @@
|
||||
package logtail
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
"github.com/mixeme/selfpost/internal/store"
|
||||
)
|
||||
|
||||
const (
|
||||
// fingerprintSize is how many bytes from the head of the log identify it.
|
||||
// Postfix writes a timestamped line per event, so the first 512 bytes are
|
||||
// effectively unique per log generation — enough to tell "the file we were
|
||||
// reading" from "a fresh one created by logrotate while we were down",
|
||||
// which os.SameFile cannot answer across a restart.
|
||||
fingerprintSize = 512
|
||||
// persistInterval throttles the offset write. Losing up to this much
|
||||
// progress on a crash only means re-parsing a few lines (UpdateStatus is
|
||||
// idempotent), which is much cheaper than a database write per poll tick.
|
||||
persistInterval = 5 * time.Second
|
||||
)
|
||||
|
||||
// tracker persists the tailer's read position so a restart resumes where the
|
||||
// previous run stopped instead of jumping to end-of-file — the "send-log rows
|
||||
// stay queued forever" gap (architecture.md § Log tailer). It is used from the
|
||||
// follow loop only, so it needs no locking.
|
||||
type tracker struct {
|
||||
st StatusStore
|
||||
path string
|
||||
|
||||
fp string // fingerprint of the file currently open ("" if too short)
|
||||
saved int64 // last offset written to the store
|
||||
lastSave time.Time
|
||||
}
|
||||
|
||||
// resume returns the byte offset the tailer should start reading f at, having
|
||||
// recorded f's fingerprint for later saves.
|
||||
//
|
||||
// The rules, in order: no stored state at all (first ever start) means start at
|
||||
// the end, so installing the panel does not replay a pre-existing log; a stored
|
||||
// state whose fingerprint still matches means continue from it, parsing the
|
||||
// tail written while the panel was down; anything else means the file is not
|
||||
// the one the offset referred to (rotated, recreated or truncated in the
|
||||
// meantime), so read it from the start. Re-parsing lines already seen is
|
||||
// harmless: UpdateStatus writes the same status onto the same row.
|
||||
func (t *tracker) resume(f *os.File) int64 {
|
||||
size, err := fileSize(f)
|
||||
if err != nil {
|
||||
log.Printf("log-tailer: stat %s: %v (reading from the start)", t.path, err)
|
||||
return 0
|
||||
}
|
||||
t.fp = fingerprintOf(f)
|
||||
|
||||
prev, ok, err := t.st.LogtailState(t.path)
|
||||
if err != nil {
|
||||
log.Printf("log-tailer: read stored offset: %v (starting at end)", err)
|
||||
return size
|
||||
}
|
||||
switch {
|
||||
case !ok:
|
||||
t.saved = size
|
||||
return size
|
||||
case prev.Fingerprint != "" && prev.Fingerprint == t.fp && prev.Offset <= size:
|
||||
t.saved = prev.Offset
|
||||
if prev.Offset < size {
|
||||
log.Printf("log-tailer: resuming %s at offset %d (%d bytes to catch up)",
|
||||
t.path, prev.Offset, size-prev.Offset)
|
||||
}
|
||||
return prev.Offset
|
||||
default:
|
||||
log.Printf("log-tailer: %s changed while the panel was down; reading from the start", t.path)
|
||||
t.saved = 0
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// adopt re-fingerprints after the follow loop switched to a rotated-in file and
|
||||
// persists the fresh start immediately, so a restart right after a rotation
|
||||
// does not resume at the old file's offset.
|
||||
func (t *tracker) adopt(f *os.File) {
|
||||
t.fp = fingerprintOf(f)
|
||||
t.saved = -1 // force the write below even if the old offset happened to be 0
|
||||
t.record(f, 0, true)
|
||||
}
|
||||
|
||||
// record persists offset, at most once per persistInterval unless force is set
|
||||
// (rotation and shutdown, where the write must not be skipped).
|
||||
func (t *tracker) record(f *os.File, offset int64, force bool) {
|
||||
if offset == t.saved {
|
||||
return
|
||||
}
|
||||
if !force && time.Since(t.lastSave) < persistInterval {
|
||||
return
|
||||
}
|
||||
if t.fp == "" {
|
||||
// The log was shorter than a fingerprint when we opened it; now that it
|
||||
// has grown, an identifiable one may be available.
|
||||
t.fp = fingerprintOf(f)
|
||||
}
|
||||
if err := t.st.SaveLogtailState(t.path, store.LogtailState{Fingerprint: t.fp, Offset: offset}); err != nil {
|
||||
log.Printf("log-tailer: save offset: %v", err)
|
||||
return
|
||||
}
|
||||
t.saved = offset
|
||||
t.lastSave = time.Now()
|
||||
}
|
||||
|
||||
// fingerprintOf hashes the head of the file. It returns "" for a file too short
|
||||
// to identify — the head would still change as Postfix appends, so such a
|
||||
// fingerprint could not be compared meaningfully on the next start.
|
||||
func fingerprintOf(f *os.File) string {
|
||||
buf := make([]byte, fingerprintSize)
|
||||
n, err := f.ReadAt(buf, 0)
|
||||
if err != nil && !errors.Is(err, io.EOF) {
|
||||
log.Printf("log-tailer: fingerprint read: %v", err)
|
||||
return ""
|
||||
}
|
||||
if n < fingerprintSize {
|
||||
return ""
|
||||
}
|
||||
sum := sha256.Sum256(buf)
|
||||
return hex.EncodeToString(sum[:])
|
||||
}
|
||||
|
||||
func fileSize(f *os.File) (int64, error) {
|
||||
info, err := f.Stat()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return info.Size(), nil
|
||||
}
|
||||
Reference in New Issue
Block a user