Method loop

storage/diskmetricstore.go:204–266

(persistenceInterval time.Duration)

}

func (dms *DiskMetricStore) loop(persistenceInterval time.Duration) {
	lastPersist := time.Now()
	persistScheduled := false
	lastWrite := time.Time{}
	persistDone := make(chan time.Time)
	var persistTimer *time.Timer

	checkPersist := func() {
		if dms.persistenceFile != "" && !persistScheduled && lastWrite.After(lastPersist) {
			persistTimer = time.AfterFunc(
				persistenceInterval-lastWrite.Sub(lastPersist),
				func() {
					persistStarted := time.Now()
					if err := dms.persist(); err != nil {
						dms.logger.Error("error persisting metrics", "err", err)
					} else {
						dms.logger.Info("metrics persisted", "file", dms.persistenceFile)
					}
					persistDone <- persistStarted
				},
			)
			persistScheduled = true
		}
	}

	for {
		select {
		case wr := <-dms.writeQueue:
			lastWrite = time.Now()
			if dms.checkWriteRequest(wr) {
				dms.processWriteRequest(wr)
			} else {
				dms.setPushFailedTimestamp(wr)
			}
			if wr.Done != nil {
				close(wr.Done)
			}
			checkPersist()
		case lastPersist = <-persistDone:
			persistScheduled = false
			checkPersist() // In case something has been written in the meantime.
		case <-dms.drain:
			// Prevent a scheduled persist from firing later.
			if persistTimer != nil {
				persistTimer.Stop()
			}
			// Now draining...
			for {
				select {
				case wr := <-dms.writeQueue:
					if dms.checkWriteRequest(wr) {
						dms.processWriteRequest(wr)
					} else {
						dms.setPushFailedTimestamp(wr)
					}
				default:
					dms.done <- dms.persist()
					return