package metrics import ( "encoding/binary" "os" "github.com/michaelquigley/df/dd" "github.com/michaelquigley/df/dl" "github.com/nxadm/tail" "github.com/pkg/errors" ) const FileSourceType = "fileSource" type FileSourceConfig struct { Path string PointerPath string } func LoadFileSource(v map[string]any) (dd.Dynamic, error) { cfg, err := dd.New[FileSourceConfig](v) if err != nil { return nil, err } return &fileSource{cfg: cfg}, nil } type fileSource struct { cfg *FileSourceConfig ptrF *os.File t *tail.Tail } func (s *fileSource) Type() string { return FileSourceType } func (s *fileSource) ToMap() (map[string]any, error) { return nil, nil } func (s *fileSource) Start(events chan ZitiEventMsg) (join chan struct{}, err error) { f, err := os.Open(s.cfg.Path) if err != nil { return nil, errors.Wrapf(err, "error opening '%v'", s.cfg.Path) } _ = f.Close() s.ptrF, err = os.OpenFile(s.pointerPath(), os.O_CREATE|os.O_RDWR, os.ModePerm) if err != nil { return nil, errors.Wrapf(err, "error opening pointer '%v'", s.pointerPath()) } ptr, err := s.readPtr() if err != nil { dl.Errorf("error reading pointer: %v", err) } dl.Infof("retrieved stored position pointer at '%d'", ptr) join = make(chan struct{}) go func() { s.tail(ptr, events) close(join) }() return join, nil } func (s *fileSource) Stop() { if err := s.t.Stop(); err != nil { dl.Error(err) } } func (s *fileSource) tail(ptr int64, events chan ZitiEventMsg) { dl.Info("started") defer dl.Info("stopped") var err error s.t, err = tail.TailFile(s.cfg.Path, tail.Config{ ReOpen: true, Follow: true, Location: &tail.SeekInfo{Offset: ptr}, }) if err != nil { dl.Errorf("error starting tail: %v", err) return } for event := range s.t.Lines { events <- &ZitiEventJsonMsg{ data: ZitiEventJson(event.Text), } if err := s.writePtr(event.SeekInfo.Offset); err != nil { dl.Error(err) } } } func (s *fileSource) pointerPath() string { if s.cfg.PointerPath == "" { return s.cfg.Path + ".ptr" } else { return s.cfg.PointerPath } } func (s *fileSource) readPtr() (int64, error) { ptr := int64(0) buf := make([]byte, 8) if n, err := s.ptrF.Seek(0, 0); err == nil && n == 0 { if n, err := s.ptrF.Read(buf); err == nil && n == 8 { ptr = int64(binary.LittleEndian.Uint64(buf)) return ptr, nil } else { return 0, errors.Wrapf(err, "error reading pointer (%d): %v", n, err) } } else { return 0, errors.Wrapf(err, "error seeking pointer (%d): %v", n, err) } } func (s *fileSource) writePtr(ptr int64) error { buf := make([]byte, 8) binary.LittleEndian.PutUint64(buf, uint64(ptr)) if n, err := s.ptrF.Seek(0, 0); err == nil && n == 0 { if n, err := s.ptrF.Write(buf); err != nil || n != 8 { return errors.Wrapf(err, "error writing pointer (%d): %v", n, err) } } else { return errors.Wrapf(err, "error seeking pointer (%d): %v", n, err) } return nil }