audit.got

package store

import (
    "errors"
    "io"
    "path"
    "strconv"
    "strings"
    "sync"
    "sync/atomic"
    "time"

    "go.uber.org/zap"
    "golang.org/x/net/context"
)

type {{.t}}AuditState struct {
    interval        int
    ageThreshold    int64

    startupShutdownLock sync.Mutex
    notifyChan          chan *bgNotification
}

func (store *default{{.T}}Store) auditConfig(cfg *{{.T}}StoreConfig) {
    store.auditState.interval = cfg.AuditInterval
    store.auditState.ageThreshold = int64(cfg.AuditAgeThreshold) * int64(time.Second)
}

func (store *default{{.T}}Store) auditStartup() {
    store.auditState.startupShutdownLock.Lock()
    if store.auditState.notifyChan == nil {
        store.auditState.notifyChan = make(chan *bgNotification, 1)
        go store.auditLauncher(store.auditState.notifyChan)
    }
    store.auditState.startupShutdownLock.Unlock()
}

func (store *default{{.T}}Store) auditShutdown() {
    store.auditState.startupShutdownLock.Lock()
    if store.auditState.notifyChan != nil {
        c := make(chan struct{}, 1)
        store.auditState.notifyChan <- &bgNotification{
            action:     _BG_DISABLE,
            doneChan:   c,
        }
        <-c
        store.auditState.notifyChan = nil
    }
    store.auditState.startupShutdownLock.Unlock()
}

func (store *default{{.T}}Store) AuditPass(ctx context.Context) error {
    store.auditState.startupShutdownLock.Lock()
    if store.auditState.notifyChan == nil {
        store.auditPass(true, make(chan *bgNotification))
    } else {
        c := make(chan struct{}, 1)
        store.auditState.notifyChan <- &bgNotification{
            action:     _BG_PASS,
            doneChan:   c,
        }
        <-c
    }
    store.auditState.startupShutdownLock.Unlock()
    return nil
}

func (store *default{{.T}}Store) auditLauncher(notifyChan chan *bgNotification) {
    interval := float64(store.auditState.interval) * float64(time.Second)
    store.randMutex.Lock()
    nextRun := time.Now().Add(time.Duration(interval + interval*store.rand.NormFloat64()*0.1))
    store.randMutex.Unlock()
    var notification *bgNotification
    running := true
    for running {
        if notification == nil {
            sleep := nextRun.Sub(time.Now())
            if sleep > 0 {
                select {
                case notification = <-notifyChan:
                case <-time.After(sleep):
                }
            } else {
                select {
                case notification = <-notifyChan:
                default:
                }
            }
        }
        store.randMutex.Lock()
        nextRun = time.Now().Add(time.Duration(interval + interval*store.rand.NormFloat64()*0.1))
        store.randMutex.Unlock()
        if notification != nil {
            var nextNotification *bgNotification
            switch notification.action {
            case _BG_PASS:
                nextNotification = store.auditPass(true, notifyChan)
            case _BG_DISABLE:
                running = false
            default:
                // Critical because there was a coding error that needs to be
                // fixed by a person.
                store.logger.Error("invalid action requested", zap.String("name", store.loggerPrefix + "audit"), zap.Int("action", int(notification.action)))
            }
            notification.doneChan <- struct{}{}
            notification = nextNotification
        } else {
            notification = store.auditPass(false, notifyChan)
        }
    }
}

// NOTE: For now, there is no difference between speed=true and speed=false;
// eventually the background audits will try to slow themselves down to finish
// in approximately the store.auditState.interval.
func (store *default{{.T}}Store) auditPass(speed bool, notifyChan chan *bgNotification) *bgNotification {
    begin := time.Now()
    defer func() {
        elapsed := time.Now().Sub(begin)
        store.logger.Debug("pass completed", zap.String("name", store.loggerPrefix + "audit"), zap.Duration("elapsed", elapsed))
        atomic.StoreInt64(&store.auditNanoseconds, elapsed.Nanoseconds())
    }()
    names, err := store.readdirnames(store.pathtoc)
    if err != nil {
        store.logger.Warn("error with readdirnames", zap.String("name", store.loggerPrefix + "audit"), zap.String("path", store.pathtoc), zap.Error(err))
        return nil
    }
    shuffledNames := make([]string, len(names))
    store.randMutex.Lock()
    for x, y := range store.rand.Perm(len(names)) {
        shuffledNames[x] = names[y]
    }
    store.randMutex.Unlock()
    names = shuffledNames
    for i := 0; i < len(names); i++ {
        select {
        case notification := <-notifyChan:
            return notification
        default:
        }
        if !strings.HasSuffix(names[i], ".{{.t}}toc") {
            continue
        }
        namets := int64(0)
        if namets, err = strconv.ParseInt(names[i][:len(names[i])-len(".{{.t}}toc")], 10, 64); err != nil {
            store.logger.Warn("bad timestamp in name", zap.String("name", store.loggerPrefix + "audit"), zap.String("name", names[i]))
            continue
        }
        if namets == 0 {
            store.logger.Warn("bad timestamp in name", zap.String("name", store.loggerPrefix + "audit"), zap.String("name", names[i]))
            continue
        }
        if namets == int64(atomic.LoadUint64(&store.activeTOCA)) || namets == int64(atomic.LoadUint64(&store.activeTOCB)) {
            store.logger.Debug("skipping current", zap.String("name", store.loggerPrefix + "audit"), zap.String("name", names[i]))
            continue
        }
        if namets >= time.Now().UnixNano()-store.auditState.ageThreshold {
            store.logger.Debug("skipping young", zap.String("name", store.loggerPrefix + "audit"), zap.String("name", names[i]))
            continue
        }
        store.logger.Debug("checking", zap.String("name", store.loggerPrefix + "audit"), zap.String("name", names[i]))
        failedAudit := uint32(0)
        canceledAudit := uint32(0)
        dataName := names[i][:len(names[i])-3]
        fpr, err := store.openReadSeeker(path.Join(store.path, dataName))
        if err != nil {
            atomic.AddUint32(&failedAudit, 1)
            if store.isNotExist(err) {
                store.logger.Debug("error opening", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", dataName), zap.Error(err))
            } else {
                store.logger.Warn("error opening", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", dataName), zap.Error(err))
            }
        } else {
            corruptions, errs := {{.t}}ChecksumVerify(fpr)
            closeIfCloser(fpr)
            for _, err := range errs {
                if err != io.EOF && err != io.ErrUnexpectedEOF {
                    store.logger.Warn("ChecksumVerify error", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", dataName), zap.Error(err))
                }
            }
            workers := uint64(1)
            pendingBatchChans := make([]chan []{{.t}}TOCEntry, workers)
            freeBatchChans := make([]chan []{{.t}}TOCEntry, len(pendingBatchChans))
            for i := 0; i < len(pendingBatchChans); i++ {
                pendingBatchChans[i] = make(chan []{{.t}}TOCEntry, 3)
                freeBatchChans[i] = make(chan []{{.t}}TOCEntry, cap(pendingBatchChans[i]))
                for j := 0; j < cap(freeBatchChans[i]); j++ {
                    freeBatchChans[i] <- make([]{{.t}}TOCEntry, store.recoveryBatchSize)
                }
            }
            nextNotificationChan := make(chan *bgNotification, 1)
            controlChan := make(chan struct{})
            go func() {
                select {
                case n := <-notifyChan:
                    if atomic.AddUint32(&canceledAudit, 1) == 0 {
                        close(controlChan)
                    }
                    nextNotificationChan <- n
                case <-controlChan:
                    nextNotificationChan <- nil
                }
            }()
            wg := &sync.WaitGroup{}
            wg.Add(len(pendingBatchChans))
            for i := 0; i < len(pendingBatchChans); i++ {
                go func(pendingBatchChan chan []{{.t}}TOCEntry, freeBatchChan chan []{{.t}}TOCEntry) {
                    for {
                        batch := <-pendingBatchChan
                        if batch == nil {
                            break
                        }
                        if atomic.LoadUint32(&failedAudit) == 0 {
                            for j := 0; j < len(batch); j++ {
                                wr := &batch[j]
                                if wr.TimestampBits & _TSB_DELETION != 0 {
                                    continue
                                }
                                if {{.t}}InCorruptRange(wr.Offset, wr.Length, corruptions) {
                                    if atomic.AddUint32(&failedAudit, 1) == 0 {
                                        close(controlChan)
                                    }
                                    break
                                }
                            }
                        }
                        freeBatchChan <- batch
                    }
                    wg.Done()
                }(pendingBatchChans[i], freeBatchChans[i])
            }
            fpr, err = store.openReadSeeker(path.Join(store.pathtoc, names[i]))
            if err != nil {
                atomic.AddUint32(&failedAudit, 1)
                if !store.isNotExist(err) {
                    store.logger.Warn("error opening", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", names[i]), zap.Error(err))
                }
            } else {
                // NOTE: The block ID is unimportant in this context, so it's
                // just set 1 and ignored elsewhere.
                _, errs := {{.t}}ReadTOCEntriesBatched(fpr, 1, freeBatchChans, pendingBatchChans, controlChan)
                closeIfCloser(fpr)
                if len(errs) > 0 {
                    atomic.AddUint32(&failedAudit, 1)
                    for _, err := range errs {
                        store.logger.Warn("ReadTOCEntriesBatched error", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", names[i]), zap.Error(err))
                    }
                }
            }
            for i := 0; i < len(pendingBatchChans); i++ {
                pendingBatchChans[i] <- nil
            }
            wg.Wait()
            close(controlChan)
            if n := <-nextNotificationChan; n != nil {
                return n
            }
        }
        if atomic.LoadUint32(&canceledAudit) != 0 {
            store.logger.Debug("canceled during", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", names[i]))
        } else if atomic.LoadUint32(&failedAudit) == 0 {
            store.logger.Debug("passed", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", names[i]))
        } else {
            store.logger.Warn("failed", zap.String("name", store.loggerPrefix + "audit"), zap.String("filename", names[i]))
            nextNotificationChan := make(chan *bgNotification, 1)
            controlChan := make(chan struct{})
            controlChan2 := make(chan struct{})
            go func() {
                select {
                case n := <-notifyChan:
                    close(controlChan)
                    nextNotificationChan <- n
                case <-controlChan2:
                    nextNotificationChan <- nil
                }
            }()
            store.compactFile(names[i], store.locBlockIDFromTimestampnano(namets), controlChan, "auditPass")
            close(controlChan2)
            if n := <-nextNotificationChan; n != nil {
                return n
            }
            go func() {
                store.logger.Warn("all audit actions require store restarts at this time", zap.String("name", store.loggerPrefix + "audit"))
                store.Shutdown(context.Background())
                store.restartChan <- errors.New("audit failure occurred requiring a restart")
            }()
            return &bgNotification{
                action:     _BG_DISABLE,
                doneChan:   make(chan struct{}, 1),
            }
        }
    }
    return nil
}