lotus/journal/fs.go

137 lines
2.6 KiB
Go
Raw Normal View History

2020-07-17 13:22:37 +00:00
package journal
import (
"encoding/json"
"fmt"
"os"
"path/filepath"
"golang.org/x/xerrors"
"github.com/filecoin-project/lotus/build"
"github.com/filecoin-project/lotus/node/repo"
)
const RFC3339nocolon = "2006-01-02T150405Z0700"
2020-07-17 13:22:37 +00:00
// fsJournal is a basic journal backed by files on a filesystem.
type fsJournal struct {
2020-08-26 15:09:37 +00:00
EventTypeRegistry
2020-07-17 13:22:37 +00:00
dir string
sizeLimit int64
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-26 15:09:37 +00:00
EventTypeRegistry: NewEventTypeRegistry(disabled),
dir: dir,
sizeLimit: 1 << 30,
incoming: make(chan *Event, 32),
closing: make(chan struct{}),
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
}
2020-08-26 15:09:37 +00:00
func (f *fsJournal) RecordEvent(evtType EventType, supplier func() interface{}) {
defer func() {
if r := recover(); r != nil {
log.Warnf("recovered from panic while recording journal event; type=%s, err=%v", evtType, r)
}
}()
if !evtType.Enabled() {
return
}
je := &Event{
2020-07-17 13:22:37 +00:00
EventType: evtType,
Timestamp: build.Clock.Now(),
2020-08-26 15:09:37 +00:00
Data: supplier(),
2020-07-17 13:22:37 +00:00
}
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(RFC3339nocolon))))
2020-07-17 13:22:37 +00:00
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
}
}
}