follower_flush_mgr.go

// Copyright (c) 2017 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package aggregator

import (
	"sync"
	"time"

	schema "github.com/m3db/m3/src/aggregator/generated/proto/flush"
	"github.com/m3db/m3/src/x/clock"
	xsync "github.com/m3db/m3/src/x/sync"
	"github.com/m3db/m3/src/x/watch"

	"github.com/uber-go/tally"
	"go.uber.org/zap"
)

type standardFollowerFlusherMetrics struct {
	shardNotFound        tally.Counter
	resolutionNotFound   tally.Counter
	kvUpdates            tally.Counter
	flushWindowsNotEnded tally.Counter
}

func newStandardFlusherMetrics(scope tally.Scope) standardFollowerFlusherMetrics {
	return standardFollowerFlusherMetrics{
		shardNotFound:        scope.Counter("shard-not-found"),
		resolutionNotFound:   scope.Counter("resolution-not-found"),
		kvUpdates:            scope.Counter("kv-updates"),
		flushWindowsNotEnded: scope.Counter("flush-windows-not-ended"),
	}
}

type forwardedFollowerFlusherMetrics struct {
	shardNotFound             tally.Counter
	resolutionNotFound        tally.Counter
	nilForwardedTimes         tally.Counter
	numForwardedTimesNotFound tally.Counter
	kvUpdates                 tally.Counter
	flushWindowsNotEnded      tally.Counter
}

func newForwardedFlusherMetrics(scope tally.Scope) forwardedFollowerFlusherMetrics {
	return forwardedFollowerFlusherMetrics{
		shardNotFound:             scope.Counter("shard-not-found"),
		resolutionNotFound:        scope.Counter("resolution-not-found"),
		nilForwardedTimes:         scope.Counter("nil-forwarded-times"),
		numForwardedTimesNotFound: scope.Counter("num-forwarded-times-not-found"),
		kvUpdates:                 scope.Counter("kv-updates"),
		flushWindowsNotEnded:      scope.Counter("flush-windows-not-ended"),
	}
}

type followerFlushManagerMetrics struct {
	watchCreateErrors tally.Counter
	kvUpdateFlush     tally.Counter
	forcedFlush       tally.Counter
	notCampaigning    tally.Counter
	standard          standardFollowerFlusherMetrics
	forwarded         forwardedFollowerFlusherMetrics
	timed             standardFollowerFlusherMetrics
}

func newFollowerFlushManagerMetrics(scope tally.Scope) followerFlushManagerMetrics {
	standardScope := scope.Tagged(map[string]string{"flusher-type": "standard"})
	forwardedScope := scope.Tagged(map[string]string{"flusher-type": "forwarded"})
	timedScope := scope.Tagged(map[string]string{"flusher-type": "timed"})
	return followerFlushManagerMetrics{
		watchCreateErrors: scope.Counter("watch-create-errors"),
		kvUpdateFlush:     scope.Counter("kv-update-flush"),
		forcedFlush:       scope.Counter("forced-flush"),
		notCampaigning:    scope.Counter("not-campaigning"),
		standard:          newStandardFlusherMetrics(standardScope),
		forwarded:         newForwardedFlusherMetrics(forwardedScope),
		timed:             newStandardFlusherMetrics(timedScope),
	}
}

type followerFlushManager struct {
	sync.RWMutex
	sync.WaitGroup

	nowFn                 clock.NowFn
	checkEvery            time.Duration
	workers               xsync.WorkerPool
	placementManager      PlacementManager
	electionManager       ElectionManager
	flushTimesManager     FlushTimesManager
	maxBufferSize         time.Duration
	forcedFlushWindowSize time.Duration
	logger                *zap.Logger
	scope                 tally.Scope

	doneCh          <-chan struct{}
	received        *schema.ShardSetFlushTimes
	processed       *schema.ShardSetFlushTimes
	flushTimesState flushTimesState
	flushMode       followerFlushMode
	lastFlushed     time.Time
	openedAt        time.Time
	flushTask       *followerFlushTask
	sleepFn         sleepFn
	metrics         followerFlushManagerMetrics
}

func newFollowerFlushManager(
	doneCh <-chan struct{},
	opts FlushManagerOptions,
) roleBasedFlushManager {
	nowFn := opts.ClockOptions().NowFn()
	instrumentOpts := opts.InstrumentOptions()
	scope := instrumentOpts.MetricsScope()
	mgr := &followerFlushManager{
		nowFn:                 nowFn,
		checkEvery:            opts.CheckEvery(),
		workers:               opts.WorkerPool(),
		placementManager:      opts.PlacementManager(),
		electionManager:       opts.ElectionManager(),
		flushTimesManager:     opts.FlushTimesManager(),
		maxBufferSize:         opts.MaxBufferSize(),
		forcedFlushWindowSize: opts.ForcedFlushWindowSize(),
		logger:                instrumentOpts.Logger(),
		scope:                 scope,
		doneCh:                doneCh,
		flushTimesState:       flushTimesUninitialized,
		flushMode:             unknownFollowerFlush,
		lastFlushed:           nowFn(),
		sleepFn:               time.Sleep,
		metrics:               newFollowerFlushManagerMetrics(scope),
	}
	mgr.flushTask = &followerFlushTask{mgr: mgr}
	return mgr
}

func (mgr *followerFlushManager) Open() {
	mgr.Lock()
	defer mgr.Unlock()

	mgr.openedAt = mgr.nowFn()
	mgr.Add(1)
	go mgr.watchFlushTimes()
}

// NB(xichen): no actions needed for initializing the follower flush manager.
func (mgr *followerFlushManager) Init([]*flushBucket) {}

func (mgr *followerFlushManager) Prepare(buckets []*flushBucket) (flushTask, time.Duration) {
	// NB(xichen): a flush is triggered in the following scenarios:
	// * The flush times persisted in kv have been updated since last flush, or
	// * Sufficient time (a.k.a. maxBufferSize) has elapsed since last flush.
	mgr.Lock()
	defer mgr.Unlock()

	var (
		now                = mgr.nowFn()
		needsFlush         bool
		flushersByInterval []flushersGroup
	)
	if mgr.flushTimesState == flushTimesUpdated {
		mgr.flushTimesState = flushTimesProcessed
		mgr.processed = mgr.received
		mgr.flushMode = kvUpdateFollowerFlush
		flushersByInterval = mgr.flushersFromKVUpdateWithLock(buckets)
		needsFlush = true
		mgr.metrics.kvUpdateFlush.Inc(1)
	} else {
		durationSinceLastFlush := now.Sub(mgr.lastFlushed)
		// If the follower has accumulated enough data since last flush without receiving a KV
		// update, we enter the forced flush mode.
		if mgr.flushMode != forcedFollowerFlush && durationSinceLastFlush >= mgr.maxBufferSize {
			mgr.flushMode = forcedFollowerFlush
		}
		// Each time the follower flushes the olded data accumulated for a given window size.
		if mgr.flushMode == forcedFollowerFlush && durationSinceLastFlush >= mgr.forcedFlushWindowSize {
			flushBeforeNanos := now.Add(-mgr.maxBufferSize).UnixNano()
			flushersByInterval = mgr.flushersFromForcedFlush(buckets, flushBeforeNanos)
			needsFlush = true
			mgr.metrics.forcedFlush.Inc(1)
		}
	}

	if !needsFlush {
		return nil, mgr.checkEvery
	}
	mgr.lastFlushed = now
	mgr.flushTask.flushersByInterval = flushersByInterval
	return mgr.flushTask, 0
}

// NB(xichen): The follower flush manager flushes data based on the flush times
// stored in kv and does not need to take extra actions when a new bucket is added.
func (mgr *followerFlushManager) OnBucketAdded(int, *flushBucket) {}

// NB(xichen): The follower flush manager flushes data based on the flush times
// stored in kv and does not need to take extra actions when a new flusher is added.
func (mgr *followerFlushManager) OnFlusherAdded(
	bucketIdx int,
	bucket *flushBucket,
	flusher flushingMetricList,
) {
}

// The follower flush manager may only lead if and only if all the following conditions
// are met:
// * The instance is campaigning.
// * All the aggregation windows since the flush manager is opened have ended.
func (mgr *followerFlushManager) CanLead() bool {
	mgr.RLock()
	defer mgr.RUnlock()

	if !mgr.electionManager.IsCampaigning() {
		mgr.metrics.notCampaigning.Inc(1)
		return false
	}
	if mgr.processed == nil {
		return false
	}

	for _, shardFlushTimes := range mgr.processed.ByShard {
		// If the shard is tombstoned, there is no need to examine its flush times.
		if shardFlushTimes.Tombstoned {
			continue
		}

		// Check that for standard metrics, all the open windows containing the process
		// start time are closed, meaning the standard metrics that didn't make to the
		// process have been flushed successfully downstream.
		if !mgr.canLead(shardFlushTimes.StandardByResolution, mgr.metrics.standard) {
			return false
		}
		if !mgr.canLead(shardFlushTimes.TimedByResolution, mgr.metrics.timed) {
			return false
		}

		// Check that the forwarded metrics have been flushed past the process start
		// time, meaning the forwarded metrics that didn't make to the process have been
		// flushed successfully downstream.
		for windowNanos, fbr := range shardFlushTimes.ForwardedByResolution {
			if fbr == nil {
				mgr.metrics.forwarded.nilForwardedTimes.Inc(1)
				return false
			}
			// Since the timestamps of the forwarded metrics are aligned to the resolution
			// boundaries, we simply need to make sure that all forwarded metrics in or before
			// the window containing the process start time have been flushed to assert that
			// the process can safely take over leadership.
			windowSize := time.Duration(windowNanos)
			waitTillFlushedTime := mgr.openedAt.Truncate(windowSize)
			if waitTillFlushedTime.Equal(mgr.openedAt) {
				waitTillFlushedTime = waitTillFlushedTime.Add(-windowSize)
			}
			for _, lastFlushedNanos := range fbr.ByNumForwardedTimes {
				if lastFlushedNanos <= waitTillFlushedTime.UnixNano() {
					mgr.metrics.forwarded.flushWindowsNotEnded.Inc(1)
					return false
				}
			}
		}
	}

	return true
}

func (mgr *followerFlushManager) canLead(
	flushTimes map[int64]int64,
	metrics standardFollowerFlusherMetrics,
) bool {
	for windowNanos, lastFlushedNanos := range flushTimes {
		windowSize := time.Duration(windowNanos)
		windowEndAt := mgr.openedAt.Truncate(windowSize)
		if windowEndAt.Before(mgr.openedAt) {
			windowEndAt = windowEndAt.Add(windowSize)
		}
		if lastFlushedNanos < windowEndAt.UnixNano() {
			metrics.flushWindowsNotEnded.Inc(1)
			return false
		}
	}
	return true
}

func (mgr *followerFlushManager) Close() { mgr.Wait() }

func (mgr *followerFlushManager) flushersFromKVUpdateWithLock(buckets []*flushBucket) []flushersGroup {
	flushersByInterval := make([]flushersGroup, len(buckets))
	for i, bucket := range buckets {
		bucketID := bucket.bucketID
		flushersByInterval[i].interval = bucket.interval
		flushersByInterval[i].duration = bucket.duration
		switch bucketID.listType {
		case standardMetricListType:
			flushersByInterval[i].flushers = mgr.standardFlushersFromKVUpdateWithLock(
				bucketID.standard.resolution,
				bucket.flushers,
				getStandardFlushTimesByResolutionFn,
				mgr.metrics.standard,
				mgr.logger.With(zap.String("flusherType", "standard")),
			)
		case forwardedMetricListType:
			flushersByInterval[i].flushers = mgr.forwardedFlushersFromKVUpdateWithLock(bucketID.forwarded, bucket.flushers)
		case timedMetricListType:
			flushersByInterval[i].flushers = mgr.standardFlushersFromKVUpdateWithLock(
				bucketID.timed.resolution,
				bucket.flushers,
				getTimedFlushTimesByResolutionFn,
				mgr.metrics.timed,
				mgr.logger.With(zap.String("flusherType", "timed")),
			)
		default:
			panic("should never get here")
		}
	}
	return flushersByInterval
}

func (mgr *followerFlushManager) standardFlushersFromKVUpdateWithLock(
	resolution time.Duration,
	flushers []flushingMetricList,
	getFlushTimesByResolutionFn getFlushTimesByResolutionFn,
	metrics standardFollowerFlusherMetrics,
	logger *zap.Logger,
) []flusherWithTime {
	var (
		flushersWithTime = make([]flusherWithTime, 0, defaultInitialFlushCapacity)
	)
	for _, flusher := range flushers {
		shard := flusher.Shard()
		shardFlushTimes, exists := mgr.received.ByShard[shard]
		if !exists {
			metrics.shardNotFound.Inc(1)
			logger.Warn("shard not found in flush times",
				zap.Uint32("shard", shard),
			)
			continue
		}
		flushTimes := getFlushTimesByResolutionFn(shardFlushTimes)
		lastFlushedAtNanos, exists := flushTimes[int64(resolution)]
		if !exists {
			metrics.resolutionNotFound.Inc(1)
			logger.Warn("resolution not found in flush times",
				zap.Uint32("shard", shard),
				zap.Stringer("resolution", resolution),
			)
			continue
		}
		newFlushTarget := flusherWithTime{
			flusher:          flusher,
			flushBeforeNanos: lastFlushedAtNanos,
		}
		flushersWithTime = append(flushersWithTime, newFlushTarget)
		metrics.kvUpdates.Inc(1)
	}
	return flushersWithTime
}

func (mgr *followerFlushManager) forwardedFlushersFromKVUpdateWithLock(
	listID forwardedMetricListID,
	flushers []flushingMetricList,
) []flusherWithTime {
	var (
		resolution        = listID.resolution
		numForwardedTimes = listID.numForwardedTimes
		flushersWithTime  = make([]flusherWithTime, 0, defaultInitialFlushCapacity)
	)
	for _, flusher := range flushers {
		shard := flusher.Shard()
		shardFlushTimes, exists := mgr.received.ByShard[shard]
		if !exists {
			mgr.metrics.forwarded.shardNotFound.Inc(1)
			mgr.logger.Warn("shard not found in flush times",
				zap.String("flusherType", "forwarded"),
				zap.Uint32("shard", shard),
			)
			continue
		}
		flushTimesForResolution, exists := shardFlushTimes.ForwardedByResolution[int64(resolution)]
		if !exists {
			mgr.metrics.forwarded.resolutionNotFound.Inc(1)
			mgr.logger.Warn("resolution not found in flush times",
				zap.String("flusherType", "forwarded"),
				zap.Uint32("shard", shard),
				zap.Stringer("resolution", resolution),
			)
			continue
		}
		if flushTimesForResolution == nil {
			mgr.metrics.forwarded.nilForwardedTimes.Inc(1)
			mgr.logger.Warn("nil flush times",
				zap.String("flusherType", "forwarded"),
				zap.Uint32("shard", shard),
				zap.Stringer("resolution", resolution),
			)
			continue
		}
		lastFlushedAtNanos, exists := flushTimesForResolution.ByNumForwardedTimes[int32(numForwardedTimes)]
		if !exists {
			mgr.metrics.forwarded.numForwardedTimesNotFound.Inc(1)
			mgr.logger.Warn("numForwardedTimes not found in flush times",
				zap.String("flusherType", "forwarded"),
				zap.Uint32("shard", shard),
				zap.Stringer("resolution", resolution),
				zap.Int("numForwardedTimes", numForwardedTimes),
			)
			continue
		}
		newFlushTarget := flusherWithTime{
			flusher:          flusher,
			flushBeforeNanos: lastFlushedAtNanos,
		}
		flushersWithTime = append(flushersWithTime, newFlushTarget)
		mgr.metrics.forwarded.kvUpdates.Inc(1)
	}
	return flushersWithTime
}

func (mgr *followerFlushManager) flushersFromForcedFlush(
	buckets []*flushBucket,
	flushBeforeNanos int64,
) []flushersGroup {
	flushersByInterval := make([]flushersGroup, len(buckets))
	for i, bucket := range buckets {
		flushersByInterval[i].interval = bucket.interval
		flushersByInterval[i].duration = bucket.duration
		flushersByInterval[i].flushers = make([]flusherWithTime, 0, defaultInitialFlushCapacity)
		for _, flusher := range bucket.flushers {
			newFlushTarget := flusherWithTime{
				flusher:          flusher,
				flushBeforeNanos: flushBeforeNanos,
			}
			flushersByInterval[i].flushers = append(flushersByInterval[i].flushers, newFlushTarget)
		}
	}
	return flushersByInterval
}

func (mgr *followerFlushManager) watchFlushTimes() {
	defer mgr.Done()

	var (
		throttlePeriod  = time.Second
		flushTimesWatch watch.Watch
		err             error
	)

	for {
		if flushTimesWatch == nil {
			flushTimesWatch, err = mgr.flushTimesManager.Watch()
			if err != nil {
				mgr.metrics.watchCreateErrors.Inc(1)
				mgr.sleepFn(throttlePeriod)
				continue
			}
		}

		select {
		case <-flushTimesWatch.C():
			mgr.Lock()
			mgr.received = flushTimesWatch.Get().(*schema.ShardSetFlushTimes)
			mgr.flushTimesState = flushTimesUpdated
			mgr.Unlock()
		case <-mgr.doneCh:
			return
		}
	}
}

type followerFlushTask struct {
	mgr                *followerFlushManager
	flushersByInterval []flushersGroup
}

func (t *followerFlushTask) Run() {
	var (
		mgr       = t.mgr
		wgWorkers sync.WaitGroup
	)
	for _, group := range t.flushersByInterval {
		start := mgr.nowFn()
		for _, flusherWithTime := range group.flushers {
			flusherWithTime := flusherWithTime
			wgWorkers.Add(1)
			mgr.workers.Go(func() {
				flusherWithTime.flusher.DiscardBefore(flusherWithTime.flushBeforeNanos)
				wgWorkers.Done()
			})
		}
		wgWorkers.Wait()
		group.duration.Record(mgr.nowFn().Sub(start))
	}
}

type flushTimesState int

const (
	flushTimesUninitialized flushTimesState = iota
	flushTimesUpdated
	flushTimesProcessed
)

type followerFlushMode int

const (
	unknownFollowerFlush followerFlushMode = iota
	kvUpdateFollowerFlush
	forcedFollowerFlush
)

type flusherWithTime struct {
	flusher          flushingMetricList
	flushBeforeNanos int64
}

type flushersGroup struct {
	interval time.Duration
	duration tally.Timer
	flushers []flusherWithTime
}