497 lines
12 KiB
Go
497 lines
12 KiB
Go
package actor
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"path/filepath"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"google.golang.org/protobuf/proto"
|
|
)
|
|
|
|
const defaultWriterIdleTTL = 5 * time.Minute
|
|
|
|
type QueueEvent struct {
|
|
TimeStamp time.Time
|
|
Message proto.Message
|
|
}
|
|
|
|
// diskLogWriter holds one append-only log file kept open between writes.
|
|
type diskLogWriter struct {
|
|
mu sync.Mutex
|
|
file *os.File
|
|
lastUsed time.Time
|
|
purged bool // set by purgeIdleWriters before closing; getWriter re-checks after acquiring mu
|
|
}
|
|
|
|
type DiskStorage[V any] struct {
|
|
*StateStorage
|
|
path string
|
|
done chan struct{}
|
|
queue *sync.Map // map[uint64][]QueueEvent
|
|
dirOnce sync.Once
|
|
dirErr error
|
|
writersMu sync.Mutex
|
|
writers map[uint64]*diskLogWriter
|
|
writerIdleTTL time.Duration
|
|
}
|
|
|
|
type LogStorage[V any] interface {
|
|
LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error
|
|
LoadEventsFunc(ctx context.Context, id uint64, grain Grain[V], condition func(msg proto.Message, index int, timeStamp time.Time) bool) error
|
|
AppendMutations(id uint64, msg ...proto.Message) error
|
|
}
|
|
|
|
func NewDiskStorage[V any](path string, registry MutationRegistry) *DiskStorage[V] {
|
|
s := &DiskStorage[V]{
|
|
StateStorage: NewState(registry),
|
|
path: path,
|
|
done: make(chan struct{}),
|
|
writers: make(map[uint64]*diskLogWriter),
|
|
writerIdleTTL: defaultWriterIdleTTL,
|
|
}
|
|
go s.purgeLoop()
|
|
go s.filesExistingLoop()
|
|
return s
|
|
}
|
|
|
|
// filesExistingLoop refreshes the EventLogFilesExisting gauge every
|
|
// 30 seconds. The gauge represents the count of .events.log files on
|
|
// disk, which diverges from the open-writer count whenever the idle
|
|
// purge closes a writer but the file is still kept (the .log file is
|
|
// not deleted by the idle purge). Periodic re-listing is cheap (one
|
|
// os.ReadDir per tick) and is the only way the gauge can pick up files
|
|
// created or removed out of band (e.g. GDPR purge, or the init-time
|
|
// rebuild of a grain that already has a log).
|
|
func (s *DiskStorage[V]) filesExistingLoop() {
|
|
ticker := time.NewTicker(30 * time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-s.done:
|
|
return
|
|
case <-ticker.C:
|
|
s.refreshFilesExisting()
|
|
}
|
|
}
|
|
}
|
|
|
|
// refreshFilesExisting counts .events.log files in the storage dir
|
|
// and writes the value to the EventLogFilesExisting gauge. Safe to
|
|
// call concurrently — os.ReadDir is atomic for our purposes (the dir
|
|
// is owned by this pod, mutations go through the writersMu lock).
|
|
// No-op when the dir does not exist yet (cold start) or when metrics
|
|
// are disabled (s.StateStorage.metrics == nil).
|
|
func (s *DiskStorage[V]) refreshFilesExisting() {
|
|
if s.StateStorage.metrics == nil {
|
|
return
|
|
}
|
|
entries, err := os.ReadDir(s.path)
|
|
if err != nil {
|
|
// Dir may not exist yet on a cold start, or briefly during a
|
|
// retention-driven directory recreation. Leave the gauge
|
|
// untouched in that case — the next tick will catch up.
|
|
return
|
|
}
|
|
count := 0
|
|
for _, e := range entries {
|
|
if !e.IsDir() && strings.HasSuffix(e.Name(), ".events.log") {
|
|
count++
|
|
}
|
|
}
|
|
s.StateStorage.metrics.EventLogFilesExisting.Set(float64(count))
|
|
}
|
|
|
|
func (s *DiskStorage[V]) ensureDir() error {
|
|
s.dirOnce.Do(func() {
|
|
s.dirErr = os.MkdirAll(s.path, 0o755)
|
|
})
|
|
return s.dirErr
|
|
}
|
|
|
|
// getWriter returns an open append handle for id. The caller must unlock w.mu
|
|
// when finished writing.
|
|
|
|
|
|
// SetMetrics attaches a Metrics instance to the embedded StateStorage
|
|
// so every append / replay / unknown-type path has access to the
|
|
// same event-log counters. The metrics struct is shared by reference;
|
|
// do not mutate its fields from outside. Pass nil to disable
|
|
// instrumentation (used by tests that do not care about Prometheus).
|
|
func (s *DiskStorage[V]) SetMetrics(m *Metrics) {
|
|
s.StateStorage.metrics = m
|
|
}
|
|
|
|
func (s *DiskStorage[V]) recordAppend(n int) {
|
|
m := s.StateStorage.metrics
|
|
if m == nil {
|
|
return
|
|
}
|
|
m.EventLogAppends.Inc()
|
|
if n > 0 {
|
|
m.EventLogBytesWritten.Add(float64(n))
|
|
}
|
|
m.EventLogLastAppendUnix.Set(float64(time.Now().Unix()))
|
|
}
|
|
|
|
func (s *DiskStorage[V]) writeEvents(id uint64, events []QueueEvent) error {
|
|
if len(events) == 0 {
|
|
return nil
|
|
}
|
|
w, err := s.getWriter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer w.mu.Unlock()
|
|
|
|
for _, evt := range events {
|
|
n, err := s.Append(w.file, evt.Message, evt.TimeStamp)
|
|
if err != nil {
|
|
// A partial write still counts the bytes that DID land on
|
|
// disk, so the throughput gauge is honest about a mid-write
|
|
// failure.
|
|
s.recordAppend(n)
|
|
return err
|
|
}
|
|
s.recordAppend(n)
|
|
}
|
|
w.lastUsed = time.Now()
|
|
return nil
|
|
}
|
|
|
|
func (s *DiskStorage[V]) writeMutations(id uint64, msgs ...proto.Message) error {
|
|
if len(msgs) == 0 {
|
|
return nil
|
|
}
|
|
w, err := s.getWriter(id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer w.mu.Unlock()
|
|
|
|
now := time.Now()
|
|
for _, m := range msgs {
|
|
n, err := s.Append(w.file, m, now)
|
|
if err != nil {
|
|
s.recordAppend(n)
|
|
return err
|
|
}
|
|
s.recordAppend(n)
|
|
}
|
|
w.lastUsed = time.Now()
|
|
return nil
|
|
}
|
|
|
|
func (s *DiskStorage[V]) purgeLoop() {
|
|
ticker := time.NewTicker(time.Minute)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-s.done:
|
|
return
|
|
case <-ticker.C:
|
|
s.purgeIdleWriters()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) purgeIdleWriters() {
|
|
cutoff := time.Now().Add(-s.writerIdleTTL)
|
|
|
|
s.writersMu.Lock()
|
|
defer s.writersMu.Unlock()
|
|
|
|
for id, w := range s.writers {
|
|
w.mu.Lock()
|
|
if w.lastUsed.Before(cutoff) {
|
|
w.purged = true
|
|
delete(s.writers, id)
|
|
if err := w.file.Close(); err != nil {
|
|
log.Printf("failed to close idle event log %d: %v", id, err)
|
|
}
|
|
}
|
|
w.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) getWriter(id uint64) (*diskLogWriter, error) {
|
|
if err := s.ensureDir(); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
for {
|
|
s.writersMu.Lock()
|
|
w, ok := s.writers[id]
|
|
if !ok {
|
|
fh, err := os.OpenFile(s.logPath(id), os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
|
|
if err != nil {
|
|
s.writersMu.Unlock()
|
|
return nil, err
|
|
}
|
|
w = &diskLogWriter{file: fh, lastUsed: time.Now()}
|
|
s.writers[id] = w
|
|
}
|
|
s.writersMu.Unlock()
|
|
|
|
w.mu.Lock()
|
|
if !w.purged {
|
|
w.lastUsed = time.Now()
|
|
return w, nil
|
|
}
|
|
// Writer was purged by purgeIdleWriters between our map lookup and acquiring w.mu.
|
|
// Release it and retry — the map no longer contains this entry, so the next
|
|
// iteration will create a fresh writer.
|
|
w.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) closeWriters() {
|
|
s.writersMu.Lock()
|
|
defer s.writersMu.Unlock()
|
|
for id, w := range s.writers {
|
|
w.mu.Lock()
|
|
w.purged = true
|
|
if err := w.file.Close(); err != nil {
|
|
log.Printf("failed to close event log %d: %v", id, err)
|
|
}
|
|
delete(s.writers, id)
|
|
w.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) SaveLoop(duration time.Duration) {
|
|
s.queue = &sync.Map{}
|
|
ticker := time.NewTicker(duration)
|
|
defer ticker.Stop()
|
|
for {
|
|
select {
|
|
case <-s.done:
|
|
s.save()
|
|
return
|
|
case <-ticker.C:
|
|
s.save()
|
|
}
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) save() {
|
|
if s.queue == nil {
|
|
return
|
|
}
|
|
carts := 0
|
|
lines := 0
|
|
s.queue.Range(func(key, value any) bool {
|
|
id := key.(uint64)
|
|
qe, ok := value.([]QueueEvent)
|
|
if !ok {
|
|
s.queue.Delete(id)
|
|
return true
|
|
}
|
|
if err := s.writeEvents(id, qe); err != nil {
|
|
log.Printf("failed to append events for grain %d: %v", id, err)
|
|
return true
|
|
}
|
|
lines += len(qe)
|
|
carts++
|
|
s.queue.Delete(id)
|
|
return true
|
|
})
|
|
if lines > 0 {
|
|
log.Printf("Appended %d carts and %d lines to disk", carts, lines)
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) logPath(id uint64) string {
|
|
return filepath.Join(s.path, fmt.Sprintf("%d.events.log", id))
|
|
}
|
|
|
|
// loadReplayMetrics times the replay, records duration, and tracks
|
|
// handler errors. It is shared by LoadEvents and LoadEventsFunc. The
|
|
// returned `onMessage` closure bumps EventLogMutationErrors on any
|
|
// non-nil error from registry.Apply, and the deferred function bumps
|
|
// EventLogReplayTotal / Failures / Duration on the overall return.
|
|
func (s *DiskStorage[V]) loadReplayMetrics() (onMessage func(err error), done func(err error)) {
|
|
m := s.StateStorage.metrics
|
|
if m == nil {
|
|
noop := func(error) {}
|
|
return noop, noop
|
|
}
|
|
start := time.Now()
|
|
onMessage = func(err error) {
|
|
if err != nil {
|
|
m.EventLogMutationErrors.Inc()
|
|
}
|
|
}
|
|
done = func(err error) {
|
|
if err != nil {
|
|
m.EventLogReplayFailures.Inc()
|
|
return
|
|
}
|
|
m.EventLogReplayTotal.Inc()
|
|
m.EventLogReplayDuration.Observe(time.Since(start).Seconds())
|
|
}
|
|
return onMessage, done
|
|
}
|
|
|
|
// recordReplayFailure bumps EventLogReplayFailures when the open
|
|
// step fails before the Load loop runs (and therefore before the
|
|
// loadReplayMetrics closure is in scope). Nil-safe.
|
|
func (s *DiskStorage[V]) recordReplayFailure() {
|
|
if m := s.StateStorage.metrics; m != nil {
|
|
m.EventLogReplayFailures.Inc()
|
|
}
|
|
}
|
|
|
|
func (s *DiskStorage[V]) LoadEventsFunc(ctx context.Context, id uint64, grain Grain[V], condition func(msg proto.Message, index int, timeStamp time.Time) bool) error {
|
|
path := s.logPath(id)
|
|
if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
|
|
// No log -> nothing to replay
|
|
return nil
|
|
}
|
|
|
|
fh, err := os.Open(path)
|
|
if err != nil {
|
|
s.recordReplayFailure()
|
|
return fmt.Errorf("open replay file: %w", err)
|
|
}
|
|
defer fh.Close()
|
|
trackApply, done := s.loadReplayMetrics()
|
|
index := 0
|
|
loadErr := s.Load(fh, func(msg proto.Message, when time.Time) {
|
|
if condition(msg, index, when) {
|
|
results, _ := s.registry.Apply(ctx, grain, msg)
|
|
for _, r := range results {
|
|
if r.Error != nil {
|
|
trackApply(r.Error)
|
|
}
|
|
}
|
|
}
|
|
index++
|
|
})
|
|
done(loadErr)
|
|
return loadErr
|
|
}
|
|
|
|
func (s *DiskStorage[V]) LoadEvents(ctx context.Context, id uint64, grain Grain[V]) error {
|
|
path := s.logPath(id)
|
|
if _, err := os.Stat(path); errors.Is(err, os.ErrNotExist) {
|
|
// No log -> nothing to replay
|
|
return nil
|
|
}
|
|
|
|
fh, err := os.Open(path)
|
|
if err != nil {
|
|
s.recordReplayFailure()
|
|
return fmt.Errorf("open replay file: %w", err)
|
|
}
|
|
defer fh.Close()
|
|
trackApply, done := s.loadReplayMetrics()
|
|
loadErr := s.Load(fh, func(msg proto.Message, _ time.Time) {
|
|
results, _ := s.registry.Apply(ctx, grain, msg)
|
|
for _, r := range results {
|
|
if r.Error != nil {
|
|
trackApply(r.Error)
|
|
}
|
|
}
|
|
})
|
|
done(loadErr)
|
|
return loadErr
|
|
}
|
|
|
|
func (s *DiskStorage[V]) Close() {
|
|
if s.queue != nil {
|
|
s.save()
|
|
}
|
|
s.closeWriters()
|
|
close(s.done)
|
|
}
|
|
|
|
func (s *DiskStorage[V]) AppendMutations(id uint64, msg ...proto.Message) error {
|
|
if s.queue != nil {
|
|
queue := make([]QueueEvent, 0)
|
|
data, found := s.queue.Load(id)
|
|
if found {
|
|
queue = data.([]QueueEvent)
|
|
}
|
|
for _, m := range msg {
|
|
queue = append(queue, QueueEvent{Message: m, TimeStamp: time.Now()})
|
|
}
|
|
s.queue.Store(id, queue)
|
|
return nil
|
|
}
|
|
return s.writeMutations(id, msg...)
|
|
}
|
|
|
|
func (s *DiskStorage[V]) openWriterCount() int {
|
|
s.writersMu.Lock()
|
|
defer s.writersMu.Unlock()
|
|
return len(s.writers)
|
|
}
|
|
|
|
func (s *DiskStorage[V]) PurgeOlderThan(ctx context.Context, retention time.Duration, isGrainActive func(id uint64) bool) (int, error) {
|
|
if err := s.ensureDir(); err != nil {
|
|
return 0, err
|
|
}
|
|
|
|
files, err := os.ReadDir(s.path)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("read storage dir: %w", err)
|
|
}
|
|
|
|
cutoff := time.Now().Add(-retention)
|
|
purgedCount := 0
|
|
|
|
for _, entry := range files {
|
|
if entry.IsDir() {
|
|
continue
|
|
}
|
|
name := entry.Name()
|
|
if !strings.HasSuffix(name, ".events.log") {
|
|
continue
|
|
}
|
|
|
|
var id uint64
|
|
_, err := fmt.Sscanf(name, "%d.events.log", &id)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
|
|
info, err := entry.Info()
|
|
if err != nil {
|
|
log.Printf("failed to get file info for %s: %v", name, err)
|
|
continue
|
|
}
|
|
|
|
if info.ModTime().Before(cutoff) {
|
|
// Check if grain is active in memory
|
|
if isGrainActive != nil && isGrainActive(id) {
|
|
continue
|
|
}
|
|
|
|
// Check if we have an active writer open
|
|
s.writersMu.Lock()
|
|
_, hasWriter := s.writers[id]
|
|
s.writersMu.Unlock()
|
|
if hasWriter {
|
|
continue
|
|
}
|
|
|
|
// Delete the file
|
|
fullPath := filepath.Join(s.path, name)
|
|
if err := os.Remove(fullPath); err != nil {
|
|
log.Printf("failed to delete expired cart log %s: %v", fullPath, err)
|
|
} else {
|
|
purgedCount++
|
|
}
|
|
}
|
|
}
|
|
|
|
return purgedCount, nil
|
|
}
|
|
|