wal,record: failover write path code

failover_manager.go contains the failoverManager (which implements
wal.Manager) for the write path, and helper classes. dirProber monitors
the primary dir when failed over to use the secondary, to decide when to
failback to the primary. failoverMonitor uses the latency and error seen
by the current *LogWriter, and probing state, to decide when to switch to
a different dir.

failover_writer.go contains the failoverWriter, that can switch across
a sequence of record.LogWriters.

record.LogWriter is changed to accommodate both standalone and failover
mode, without affecting the synchronization in standalone mode.
In failover mode there is some additional synchronization, in the
failoverWriter queue, which is not lock free, but is hopefully fast
enough given the fastpath will use read locks.

Informs #3230

record/log_writer_test.go

@@ -76,11 +76,12 @@ func TestSyncQueue(t *testing.T) {
 
 func TestFlusherCond(t *testing.T) {
 	var mu sync.Mutex
-	var q syncQueue
 	var c flusherCond
 	var closed bool
 
-	c.init(&mu, &q)
+	psq := &pendingSyncsWithSyncQueue{}
+	c.init(&mu, psq)
+	q := &psq.syncQueue
 
 	var flusherWG sync.WaitGroup
 	flusherWG.Add(1)
@@ -299,7 +300,8 @@ func TestMinSyncInterval(t *testing.T) {
 		}
 		// NB: we can't use syncQueue.load() here as that will return 0,0 while the
 		// syncQueue is blocked.
-		head, tail := w.flusher.syncQ.unpack(w.flusher.syncQ.headTail.Load())
+		syncQ := w.flusher.pendingSyncs.(*pendingSyncsWithSyncQueue)
+		head, tail := syncQ.unpack(syncQ.headTail.Load())
 		waiters := head - tail
 		if waiters != uint32(i+1) {
 			t.Fatalf("expected %d waiters, but found %d", i+1, waiters)

vfs/errorfs/errorfs.go

@@ -81,6 +81,8 @@ const (
 	OpFileStat
 	// OpFileSync describes a file sync operation.
 	OpFileSync
+	// OpFileSyncData describes a file sync operation.
+	OpFileSyncData
 	// OpFileFlush describes a file flush operation.
 	OpFileFlush
 )
@@ -90,7 +92,7 @@ func (o OpKind) ReadOrWrite() OpReadWrite {
 	switch o {
 	case OpOpen, OpOpenDir, OpList, OpStat, OpGetDiskUsage, OpFileRead, OpFileReadAt, OpFileStat:
 		return OpIsRead
-	case OpCreate, OpLink, OpRemove, OpRemoveAll, OpRename, OpReuseForWrite, OpMkdirAll, OpLock, OpFileClose, OpFileWrite, OpFileWriteAt, OpFileSync, OpFileFlush, OpFilePreallocate:
+	case OpCreate, OpLink, OpRemove, OpRemoveAll, OpRename, OpReuseForWrite, OpMkdirAll, OpLock, OpFileClose, OpFileWrite, OpFileWriteAt, OpFileSync, OpFileSyncData, OpFileFlush, OpFilePreallocate:
 		return OpIsWrite
 	default:
 		panic(fmt.Sprintf("unrecognized op %v\n", o))
@@ -528,7 +530,9 @@ func (f *errorFile) Sync() error {
 }
 
 func (f *errorFile) SyncData() error {
-	// TODO(jackson): Consider error injection.
+	if err := f.inj.MaybeError(Op{Kind: OpFileSyncData, Path: f.path}); err != nil {
+		return err
+	}
 	return f.file.SyncData()
 }