lotus/journal/fs.go

128 lines
2.4 KiB
Go
Raw Normal View History

2020-07-17 13:22:37 +00:00
package journal
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"sync"
"time"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/repo"
)
// fsJournal is a basic journal backed by files on a filesystem.
type fsJournal struct {
2020-08-10 14:02:15 +00:00
EventTypeFactory
2020-07-17 13:22:37 +00:00
dir string
sizeLimit int64
lk sync.Mutex
fi *os.File
fSize int64
incoming chan *Event
2020-07-17 13:22:37 +00:00
closing chan struct{}
2020-08-11 12:48:32 +00:00
closed chan struct{}
2020-07-17 13:22:37 +00:00
}
// OpenFSJournal constructs a rolling filesystem journal, with a default
// per-file size limit of 1GiB.
2020-08-11 12:48:32 +00:00
func OpenFSJournal(lr repo.LockedRepo, disabled DisabledEvents) (Journal, error) {
2020-07-17 13:22:37 +00:00
dir := filepath.Join(lr.Path(), "journal")
if err := os.MkdirAll(dir, 0755); err != nil {
return nil, fmt.Errorf("failed to mk directory %s for file journal: %w", dir, err)
}
f := &fsJournal{
2020-08-10 14:02:15 +00:00
EventTypeFactory: NewEventTypeFactory(disabled),
2020-07-17 17:35:15 +00:00
dir: dir,
sizeLimit: 1 << 30,
incoming: make(chan *Event, 32),
2020-07-17 17:35:15 +00:00
closing: make(chan struct{}),
2020-08-11 12:48:32 +00:00
closed: make(chan struct{}),
2020-07-17 13:22:37 +00:00
}
if err := f.rollJournalFile(); err != nil {
return nil, err
}
go f.runLoop()
return f, nil
}
func (f *fsJournal) RecordEvent(evtType EventType, obj interface{}) {
je := &Event{
2020-07-17 13:22:37 +00:00
EventType: evtType,
Timestamp: build.Clock.Now(),
Data: obj,
}
select {
case f.incoming <- je:
case <-f.closing:
log.Warnw("journal closed but tried to log event", "event", je)
2020-07-17 13:22:37 +00:00
}
}
func (f *fsJournal) Close() error {
close(f.closing)
2020-08-11 12:48:32 +00:00
<-f.closed
2020-07-17 13:22:37 +00:00
return nil
}
func (f *fsJournal) putEvent(evt *Event) error {
b, err := json.Marshal(evt)
2020-07-17 13:22:37 +00:00
if err != nil {
return err
}
n, err := f.fi.Write(append(b, '\n'))
if err != nil {
return err
}
f.fSize += int64(n)
if f.fSize >= f.sizeLimit {
_ = f.rollJournalFile()
}
return nil
}
func (f *fsJournal) rollJournalFile() error {
if f.fi != nil {
_ = f.fi.Close()
}
nfi, err := os.Create(filepath.Join(f.dir, fmt.Sprintf("lotus-journal-%s.ndjson", build.Clock.Now().Format(time.RFC3339))))
if err != nil {
return xerrors.Errorf("failed to open journal file: %w", err)
}
f.fi = nfi
f.fSize = 0
return nil
}
func (f *fsJournal) runLoop() {
2020-08-11 12:48:32 +00:00
defer close(f.closed)
2020-07-17 13:22:37 +00:00
for {
select {
case je := <-f.incoming:
if err := f.putEvent(je); err != nil {
log.Errorw("failed to write out journal event", "event", je, "err", err)
2020-07-17 13:22:37 +00:00
}
case <-f.closing:
_ = f.fi.Close()
return
}
}
}