/**
Replays an input sample file from the top of the head and adheres to the specified format, line rate and duration.
*/
Replays an input sample file from the top of the head and adheres to the specified format, line rate and duration.
/ ** Replays an input sample file from the top of the head and adheres to the specified format line rate and duration . * /
func GenerateLoadFromInputFile(promRegistry *prometheus.Registry, props *LoadGenProperties) {
	fmt.Println("Generating logs by replaying input file...")
	const MultiLineLimit = 1000
	var log = logrus.New()
	var totalLineCount int64
	var fileArray = make([]*os.File, props.FileCount, props.FileCount)
	var logHandlers = make([]*logrus.Logger, props.FileCount)
	f, err := os.Open(props.InputSourceFile)
	if err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
	if props.Rotate == true {
		fmt.Println("log rotation enabled")
		logHandlers = props.fetchLogHandlers(false)
		props.setupLogRotation(props.FilePath, logHandlers)
	} else {
		fmt.Println("log rotation is disabled")
		fileArray, _ = props.fileRef()
	}
	rateLimit := ratelimit.New(int(props.Lps))
	timer := time.After(props.durationInSeconds())
	numberOfFiles := int(props.FileCount)
	var multiLineString strings.Builder
	eventBreakers := eventbreaker.NewEventBreakers()
	fileCountIndex := 0
	var totalBytesProcessed int64
	counter := metrics.NewCounter()
	goGenMetricsRegistry := metrics.NewRegistry()
	goGenMetricsRegistry.Register("total-events-processed", counter)
	metricsLogger := logrus.New()
	metricsLogger.Out = os.Stdout
	metricsLogger.Formatter = utility.GetFormatter(false)
	go metrics.Log(goGenMetricsRegistry, 1*time.Second, metricsLogger)
	var promCounter, promTotalBytesProcessedCounter prometheus.Counter
	if props.EnableMetrics {
		promCounter = metricsUtility.GetEventsProcessedCounter()
		promTotalBytesProcessedCounter = metricsUtility.GetTotalBytesProcessedCounter()
		promRegistry.MustRegister(promCounter)
		promRegistry.MustRegister(promTotalBytesProcessedCounter)
	}
	metrics.NewRegisteredFunctionalGauge("total-bytes-processed", goGenMetricsRegistry, func() int64 {
		if props.EnableMetrics {
			promTotalBytesProcessedCounter.Add(float64(totalBytesProcessed))
		}
		return totalBytesProcessed
	})
	var fileLoopCount int64
	for {
		if props.ReplayCount != 0 && props.ReplayCount == fileLoopCount {
			wrapUpTask(props, totalLineCount)
			return
		}
		fileLoopCount++
		f.Seek(0, io.SeekStart)
		reader := bufio.NewReader(f)
		for {
			now := rateLimit.Take()
			if props.ReplayCount > 0 {
				select {
				case _ = <-timer:
					wrapUpTask(props, totalLineCount)
					return
				default:
				}
			}
			counter.Inc(1)
			if props.EnableMetrics {
				promCounter.Add(1)
			}
			prev := time.Now()
			if !props.Rotate {
				log.Out = fileArray[fileCountIndex]
				log.SetFormatter(utility.GetFormatter(false))
			} else {
				log = logHandlers[fileCountIndex]
			}
			currentLine, err := reader.ReadString('\n')
			if err != nil {
				if err == io.EOF {
					break
				}
				log.Fatalf("read file line error: %v", err)
				return
			}
			nextLine, err := reader.Peek(40)
			isNotMultiLine := eventBreakers.HasKnownEventBreakerString(currentLine) && eventBreakers.HasKnownEventBreakerBytes(nextLine)
			if isNotMultiLine {
				log.Info(currentLine)
				totalLineCount++
				totalBytesProcessed += int64(len(currentLine))
			} else {
				currentMultiLineCount := 0
				for {
					multiLineString.WriteString(currentLine)
					currentMultiLineCount++
					totalLineCount++
					nextLineBytes, _ := reader.Peek(40)
					if eventBreakers.HasKnownEventBreakerBytes(nextLineBytes) || currentMultiLineCount >= MultiLineLimit {
						break
					}
					currentLine, _ = reader.ReadString('\n')
				}
				log.Info(multiLineString.String())
				totalBytesProcessed += int64(multiLineString.Len())
				multiLineString.Reset()
			}
			if fileCountIndex == numberOfFiles-1 {
				fileCountIndex = 0
			} else {
				fileCountIndex++
			}
			now.Sub(prev)
			prev = now
		}
	}
	f.Close()
}
func GenerateLoadFromInputFile ( promRegistry * prometheus . Registry , props * LoadGenProperties ) { fmt . Println ( " " ) const MultiLineLimit = 1000 var log = logrus . New ( ) var totalLineCount int64 var fileArray = make ( [ ] * os . File , props . FileCount , props . FileCount ) var logHandlers = make ( [ ] * logrus . Logger , props . FileCount ) f , err := os . Open ( props . InputSourceFile ) if err != nil { fmt . Println ( err ) os . Exit ( 1 ) } if props . Rotate == true { fmt . Println ( " " ) logHandlers = props . fetchLogHandlers ( false ) props . setupLogRotation ( props . FilePath , logHandlers ) } else { fmt . Println ( " " ) fileArray , _ = props . fileRef ( ) } rateLimit := ratelimit . New ( int ( props . Lps ) ) timer := time . After ( props . durationInSeconds ( ) ) numberOfFiles := int ( props . FileCount ) var multiLineString strings . Builder eventBreakers := eventbreaker . NewEventBreakers ( ) fileCountIndex := 0 var totalBytesProcessed int64 counter := metrics . NewCounter ( ) goGenMetricsRegistry := metrics . NewRegistry ( ) goGenMetricsRegistry . Register ( " " , counter ) metricsLogger := logrus . New ( ) metricsLogger . Out = os . Stdout metricsLogger . Formatter = utility . GetFormatter ( false ) go metrics . Log ( goGenMetricsRegistry , 1 * time . Second , metricsLogger ) var promCounter , promTotalBytesProcessedCounter prometheus . Counter if props . EnableMetrics { promCounter = metricsUtility . GetEventsProcessedCounter ( ) promTotalBytesProcessedCounter = metricsUtility . GetTotalBytesProcessedCounter ( ) promRegistry . MustRegister ( promCounter ) promRegistry . MustRegister ( promTotalBytesProcessedCounter ) } metrics . NewRegisteredFunctionalGauge ( " " , goGenMetricsRegistry , func ( ) int64 { if props . EnableMetrics { promTotalBytesProcessedCounter . Add ( float64 ( totalBytesProcessed ) ) } return totalBytesProcessed } ) var fileLoopCount int64 for { if props . ReplayCount != 0 && props . ReplayCount == fileLoopCount { wrapUpTask ( props , totalLineCount ) return } fileLoopCount ++ f . Seek ( 0 , io . SeekStart ) reader := bufio . NewReader ( f ) for { now := rateLimit . Take ( ) if props . ReplayCount > 0 { select { case _ = <- timer : wrapUpTask ( props , totalLineCount ) return default : } } counter . Inc ( 1 ) if props . EnableMetrics { promCounter . Add ( 1 ) } prev := time . Now ( ) if ! props . Rotate { log . Out = fileArray [ fileCountIndex ] log . SetFormatter ( utility . GetFormatter ( false ) ) } else { log = logHandlers [ fileCountIndex ] } currentLine , err := reader . ReadString ( '\n' ) if err != nil { if err == io . EOF { break } log . Fatalf ( " " , err ) return } nextLine , err := reader . Peek ( 40 ) isNotMultiLine := eventBreakers . HasKnownEventBreakerString ( currentLine ) && eventBreakers . HasKnownEventBreakerBytes ( nextLine ) if isNotMultiLine { log . Info ( currentLine ) totalLineCount ++ totalBytesProcessed += int64 ( len ( currentLine ) ) } else { currentMultiLineCount := 0 for { multiLineString . WriteString ( currentLine ) currentMultiLineCount ++ totalLineCount ++ nextLineBytes , _ := reader . Peek ( 40 ) if eventBreakers . HasKnownEventBreakerBytes ( nextLineBytes ) || currentMultiLineCount >= MultiLineLimit { break } currentLine , _ = reader . ReadString ( '\n' ) } log . Info ( multiLineString . String ( ) ) totalBytesProcessed += int64 ( multiLineString . Len ( ) ) multiLineString . Reset ( ) } if fileCountIndex == numberOfFiles - 1 { fileCountIndex = 0 } else { fileCountIndex ++ } now . Sub ( prev ) prev = now } } f . Close ( ) }
Replays an input sample file from the top of the head and adheres to the specified format, line rate and duration.
Replays an input sample file from the top of the head and adheres to the specified format line rate and duration .
//prometheus stuff