featurebase/dax/writelogger/writelogger.go

// Package writelogger provides the writelogger structs.
package writelogger

import (
	"fmt"
	"io"
	"io/fs"
	"os"
	"path"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/featurebasedb/featurebase/v3/dax"
	"github.com/featurebasedb/featurebase/v3/dax/computer"
	"github.com/featurebasedb/featurebase/v3/errors"
	"github.com/featurebasedb/featurebase/v3/logger"
)

type Writelogger struct {
	dataDir string

	mu        sync.RWMutex
	logFiles  map[string]*os.File
	lockFiles map[string]*os.File

	logger logger.Logger
}

func New(dir string, log logger.Logger) *Writelogger {
	return &Writelogger{
		dataDir:   dir,
		logFiles:  make(map[string]*os.File),
		lockFiles: make(map[string]*os.File),
		logger:    log,
	}
}

// SetLogger sets the logger used for logging messages. Note, this is not the
// same "logger" that the Writelogger represents, which logs data writes.
func (w *Writelogger) SetLogger(l logger.Logger) {
	w.logger = l
}

func (w *Writelogger) AppendMessage(bucket string, key string, version int, message []byte) error {
	fKey := fullKey(bucket, key, version)
	logFile, err := w.logFileByKey(fKey)
	if err != nil {
		return errors.Wrapf(err, "getting log file by key: %s", fKey)
	}

	_, err = logFile.Write(append(message, "\n"...))
	if err != nil {
		return errors.Wrapf(err, "writing to log file %s", logFile.Name())
	}
	err = logFile.Sync()
	return errors.Wrapf(err, "syncing log file %s", logFile.Name())
}

func (w *Writelogger) List(bucket, key string) ([]computer.WriteLogInfo, error) {
	dirpath := path.Join(w.dataDir, bucket, key)

	entries, err := os.ReadDir(dirpath)
	if err != nil {
		if pe, ok := err.(*os.PathError); ok && pe.Err == syscall.ENOENT {
			return nil, nil
		}
		return nil, errors.Wrap(err, "reading directory")
	}

	wLogs := make([]computer.WriteLogInfo, len(entries))
	for i, entry := range entries {
		version, err := strconv.ParseInt(entry.Name(), 10, 64)
		if err != nil {
			return nil, errors.Wrapf(err, "writelog filename '%s' could not be parsed to version number", entry.Name())
		}
		wLogs[i] = computer.WriteLogInfo{
			Version: int(version),
		}
	}
	return wLogs, nil
}

func (w *Writelogger) LogReader(bucket, key string, version int) (io.ReadCloser, error) {
	return w.LogReaderFrom(bucket, key, version, 0)
}

func (w *Writelogger) LogReaderFrom(bucket string, key string, version int, offset int) (io.ReadCloser, error) {
	_, filePath := w.paths(fullKey(bucket, key, version))

	f, err := os.Open(filePath)
	if err != nil {
		if e, ok := err.(*fs.PathError); ok {
			return nil, e
		}
		return nil, err
	}
	if offset > 0 {
		f.Seek(int64(offset), io.SeekStart)
	}
	w.logger.Debugf("Writelogger LogReader file: %s", f.Name())

	return f, nil
}

func (w *Writelogger) DeleteLog(bucket string, key string, version int) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	fullKey := fullKey(bucket, key, version)

	f, ok := w.logFiles[fullKey]
	if !ok {
		return nil
	}

	// Close the log file.
	if err := f.Close(); err != nil {
		return errors.Wrap(err, "closing log file")
	}

	delete(w.logFiles, fullKey)

	// Remove the log file.
	return os.Remove(f.Name())
}

func (w *Writelogger) lockFile(bucket, key string) (string, string) {
	lockFile := path.Join(w.dataDir, bucket, fmt.Sprintf("_lock_%s", key))
	return path.Dir(lockFile), lockFile
}

func (w *Writelogger) Lock(bucket, key string) error {
	lockDir, lockFile := w.lockFile(bucket, key)

	if err := os.MkdirAll(lockDir, 0777); err != nil {
		return errors.Wrapf(err, "lock dir %s", lockDir)
	}

	// We introduced this retry logic when we thought that a node which was
	// having its partition ownership removed (in place of this node) was still
	// holding the lock. While that still may be the case, it's more likely that
	// the problem we were seeing was that we weren't actually calling
	// `resource.Unlock()` in ApplyDirective for resources being removed. With
	// that said, it doesn't hurt to leave this retry logic here.
	var f *os.File
	var err error
	if err := w.retryUntil(10*time.Second, func() error {
		f, err = os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL|syscall.O_NONBLOCK, 0644)
		if err != nil {
			return errors.Wrapf(err, "opening lock file: %s", lockFile)
		}
		return nil
	}); err != nil {
		return errors.Wrapf(err, "opening lock file: %s", lockFile)
	}

	w.mu.Lock()
	defer w.mu.Unlock()
	w.lockFiles[lockFile] = f

	// fd, err = syscall.Open(lockFile, syscall.O_RDWR|syscall.O_CREAT, 0644)
	// if err != nil {
	// 	return 0, errors.Wrapf(err, "syscall opening %s", lockFile)
	// }
	// err = syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
	// 	Type: syscall.F_WRLCK,
	// })
	return nil
}

// retryUntil repeatedly executes fn until it returns nil or timeout occurs.
func (w *Writelogger) retryUntil(timeout time.Duration, fn func() error) (err error) {
	timer := time.NewTimer(timeout)
	defer timer.Stop()
	ticker := time.NewTicker(100 * time.Millisecond)
	defer ticker.Stop()

	var i int
	for {
		if err = fn(); err == nil {
			return nil
		}
		i++
		w.logger.Debugf("Writelogger retryUntil try: %d", i)

		select {
		case <-timer.C:
			return err
		case <-ticker.C:
		}
	}
}

func (w *Writelogger) Unlock(bucket, key string) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	// remove all local state associated with this bucket/key
	keyPrefix := path.Join(bucket, key)
	for logKey, logFile := range w.logFiles {
		if strings.HasPrefix(logKey, keyPrefix) {
			_ = logFile.Close()
			delete(w.logFiles, logKey)
		}
	}
	// TODO(jaffee) since the file isn't guaranteed to be removed if
	// the process is killed, we should actually use flock instead of
	// EXCL file creation. Problem with that is it makes testing
	// tricky because file handles from the same process are able to
	// acquire the flock simultaneously. Headache.
	_, lockFile := w.lockFile(bucket, key)

	if f, ok := w.lockFiles[lockFile]; ok {
		_ = f.Close()
	} else {
		w.logger.Warnf("unlocking %s not find cached file to unlock", lockFile)
	}
	err := os.Remove(lockFile)
	delete(w.lockFiles, lockFile)

	// defer func() {
	// 	err := syscall.Close(fd)
	// 	if err != nil {
	// 		w.logger.Printf("error closing lockfile %s", lockFile)
	// 	}
	// }()
	// err := syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
	// 	Type: syscall.F_UNLCK,
	// })
	return errors.Wrap(err, "removing lock file")
}

func (w *Writelogger) DeleteTable(qtid dax.QualifiedTableID) error {
	dir := path.Join(w.dataDir, string(qtid.Key()))
	err := os.RemoveAll(dir)
	if err != nil {
		return errors.Wrapf(err, "dropping %s from writelogger", dir)
	}
	return nil
}

// paths takes a key and returns the full file path (including the root data
// directory) as well as the full directory path (i.e. the file path without the
// file portion).
func (w *Writelogger) paths(key string) (string, string) {
	filePath := path.Join(w.dataDir, key)
	dirPath, _ := path.Split(filePath)
	return dirPath, filePath
}

// logFileByKey returns a pointer to the file specified by key. If the file does
// not exist, the file is created (along with any directories in which the file
// is nested).
func (w *Writelogger) logFileByKey(key string) (*os.File, error) {
	w.mu.Lock()
	defer w.mu.Unlock()

	if f, ok := w.logFiles[key]; ok {
		return f, nil
	}

	dirPath, filePath := w.paths(key)

	// make directories
	if err := os.MkdirAll(dirPath, 0777); err != nil {
		return nil, errors.Wrapf(err, "making direcory: %s", dirPath)
	}

	// open log file
	f, err := os.OpenFile(filePath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		return nil, errors.Wrapf(err, "opening file: %s", filePath)
	}

	w.logFiles[key] = f

	return f, nil
}

// fullKey returns the full file key including the bucket and version.
func fullKey(bucket string, key string, version int) string {
	return path.Join(bucket, key, fmt.Sprintf("%d", version))
}