<commit_before>package cron

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/Cepave/open-falcon-backend/common/model"
	"github.com/Cepave/open-falcon-backend/modules/alarm/g"
	eventmodel "github.com/Cepave/open-falcon-backend/modules/alarm/model/event"
	"github.com/garyburd/redigo/redis"
	log "github.com/sirupsen/logrus"
)

func ReadHighEvent() {
	queues := g.Config().Redis.HighQueues
	if len(queues) == 0 {
		return
	}

	for {
		event, err := popEvent(queues)
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		consume(event, true)
	}
}

func ReadLowEvent() {
	queues := g.Config().Redis.LowQueues
	if len(queues) == 0 {
		return
	}

	for {
		event, err := popEvent(queues)
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		consume(event, false)
	}
}

func popEvent(queues []string) (*model.Event, error) {
	count := len(queues)

	params := make([]interface{}, count+1)
	for i := 0; i < count; i++ {
		params[i] = queues[i]
	}
	// set timeout 0
	params[count] = 0

	rc := g.RedisConnPool.Get()
	defer rc.Close()

	reply, err := redis.Strings(rc.Do("BRPOP", params...))
	if err != nil {
		log.Error(fmt.Sprintf("get alarm event from redis fail: %v", err))
		return nil, err
	}

	var event model.Event
	err = json.Unmarshal([]byte(reply[1]), &event)
	if err != nil {
		log.Error(fmt.Sprintf("parse alarm event fail: %v", err))
		return nil, err
	}

	log.Debug(event.String())
	//insert event into database
	err = eventmodel.InsertEvent(&event, "owl")
	if err != nil {
		log.Error(err.Error())
	}
	// save in memory. display in dashboard
	g.Events.Put(&event)

	return &event, nil
}

func ReadExternalEvent() {
	queues := g.Config().Redis.ExternalQueues.Queues
	if len(queues) == 0 {
		return
	}

	for {
		err := popExternalEvent(queues)
		if err != nil {
			log.Error(err.Error())
			time.Sleep(time.Second)
			continue
		}
	}
}

func popExternalEvent(queues []string) error {
	count := len(queues)

	params := make([]interface{}, count+1)
	for i := 0; i < count; i++ {
		params[i] = queues[i]
	}
	// set timeout 0
	params[count] = 0

	rc := g.RedisConnPool.Get()
	defer rc.Close()

	reply, err := redis.Strings(rc.Do("BRPOP", params...))
	if err != nil {
		log.Error(fmt.Sprintf("get alarm event from redis fail: %v", err))
		return err
	}

	event := eventmodel.ExternalEvent{}

	err = json.Unmarshal([]byte(reply[1]), &event)
	if err != nil {
		log.Error(fmt.Sprintf("parse alarm event fail: %v", err))
		return err
	}
	if err := event.CheckFormating(); err != nil {
		errMsg := fmt.Sprintf("check alarm formating got error: %v, event: %v", err, event)
		if g.Config().Redis.ErrorQueue.Enable {
			params := []interface{}{g.Config().Redis.ErrorQueue.Queue, errMsg}
			_, err := rc.Do("LPUSH", params...)
			if err != nil {
				log.Error(err.Error())
			}
		}
		return err
	}

	event = event.ForceFixStepWhenStatusOk()
	//insert event into database
	err = eventmodel.InsertExternalEvent(event)
	if err != nil {
		log.Error(err.Error())
		errMsg = fmt.Sprintf("insert event got error: %v, event: %v", err.Error(), event)
		if g.Config().Redis.ErrorQueue.Enable {
			params := []interface{}{g.Config().Redis.ErrorQueue.Queue, errMsg}
			_, err := rc.Do("LPUSH", params...)
			if err != nil {
				log.Error(err.Error())
			}
		}
	}

	return nil
}<commit_msg>fix error of undefined: errMsg<commit_after>package cron

import (
	"encoding/json"
	"fmt"
	"time"

	"github.com/Cepave/open-falcon-backend/common/model"
	"github.com/Cepave/open-falcon-backend/modules/alarm/g"
	eventmodel "github.com/Cepave/open-falcon-backend/modules/alarm/model/event"
	"github.com/garyburd/redigo/redis"
	log "github.com/sirupsen/logrus"
)

func ReadHighEvent() {
	queues := g.Config().Redis.HighQueues
	if len(queues) == 0 {
		return
	}

	for {
		event, err := popEvent(queues)
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		consume(event, true)
	}
}

func ReadLowEvent() {
	queues := g.Config().Redis.LowQueues
	if len(queues) == 0 {
		return
	}

	for {
		event, err := popEvent(queues)
		if err != nil {
			time.Sleep(time.Second)
			continue
		}
		consume(event, false)
	}
}

func popEvent(queues []string) (*model.Event, error) {
	count := len(queues)

	params := make([]interface{}, count+1)
	for i := 0; i < count; i++ {
		params[i] = queues[i]
	}
	// set timeout 0
	params[count] = 0

	rc := g.RedisConnPool.Get()
	defer rc.Close()

	reply, err := redis.Strings(rc.Do("BRPOP", params...))
	if err != nil {
		log.Error(fmt.Sprintf("get alarm event from redis fail: %v", err))
		return nil, err
	}

	var event model.Event
	err = json.Unmarshal([]byte(reply[1]), &event)
	if err != nil {
		log.Error(fmt.Sprintf("parse alarm event fail: %v", err))
		return nil, err
	}

	log.Debug(event.String())
	//insert event into database
	err = eventmodel.InsertEvent(&event, "owl")
	if err != nil {
		log.Error(err.Error())
	}
	// save in memory. display in dashboard
	g.Events.Put(&event)

	return &event, nil
}

func ReadExternalEvent() {
	queues := g.Config().Redis.ExternalQueues.Queues
	if len(queues) == 0 {
		return
	}

	for {
		err := popExternalEvent(queues)
		if err != nil {
			log.Error(err.Error())
			time.Sleep(time.Second)
			continue
		}
	}
}

func popExternalEvent(queues []string) error {
	count := len(queues)

	params := make([]interface{}, count+1)
	for i := 0; i < count; i++ {
		params[i] = queues[i]
	}
	// set timeout 0
	params[count] = 0

	rc := g.RedisConnPool.Get()
	defer rc.Close()

	reply, err := redis.Strings(rc.Do("BRPOP", params...))
	if err != nil {
		log.Error(fmt.Sprintf("get alarm event from redis fail: %v", err))
		return err
	}

	event := eventmodel.ExternalEvent{}

	err = json.Unmarshal([]byte(reply[1]), &event)
	if err != nil {
		log.Error(fmt.Sprintf("parse alarm event fail: %v", err))
		return err
	}
	if err := event.CheckFormating(); err != nil {
		errMsg := fmt.Sprintf("check alarm formating got error: %v, event: %v", err, event)
		if g.Config().Redis.ErrorQueue.Enable {
			params := []interface{}{g.Config().Redis.ErrorQueue.Queue, errMsg}
			_, err := rc.Do("LPUSH", params...)
			if err != nil {
				log.Error(err.Error())
			}
		}
		return err
	}

	event = event.ForceFixStepWhenStatusOk()
	//insert event into database
	err = eventmodel.InsertExternalEvent(event)
	if err != nil {
		log.Error(err.Error())
		errMsg := fmt.Sprintf("insert event got error: %v, event: %v", err.Error(), event)
		if g.Config().Redis.ErrorQueue.Enable {
			params := []interface{}{g.Config().Redis.ErrorQueue.Queue, errMsg}
			_, err := rc.Do("LPUSH", params...)
			if err != nil {
				log.Error(err.Error())
			}
		}
	}

	return nil
}
{'fraction_non_alphanumeric': 0.17308275234919673, 'fraction_numerical': 0.003940588056986966, 'mean_word_length': 3.4921715452688904, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6618323', 'n_tokens_mistral': 2685, 'n_tokens_neox': 2469, 'n_words': 560}