<commit_before>package main

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"log"
	"time"

	"github.com/bitly/go-nsq"
	"github.com/raintank/raintank-metric/metricdef"
	"github.com/raintank/raintank-metric/metricstore"
)

type KairosGateway struct {
	kairos     *metricstore.Kairosdb
	inHighPrio chan Job
	inLowPrio  chan Job
}

func NewKairosGateway() (*KairosGateway, error) {
	kairos, err := metricstore.NewKairosdb("http://kairosdb:8080")
	if err != nil {
		return nil, err
	}
	kg := &KairosGateway{
		kairos:     kairos,
		inHighPrio: make(chan Job, 4), // always useful to have some packets queue up to enforce the prio
		inLowPrio:  make(chan Job),
	}
	go kg.Run()
	return kg, nil
}

func (kg *KairosGateway) Run() {
	for {
		select {
		case job := <-kg.inHighPrio:
			job.done <- kg.process("high-prio", job.msg)
		default:
			select {
			case job := <-kg.inHighPrio:
				job.done <- kg.process("high-prio", job.msg)
			case job := <-kg.inLowPrio:
				job.done <- kg.process("low--prio", job.msg)
			}
		}
	}
}
func (kg *KairosGateway) ProcessHighPrio(msg *nsq.Message) error {
	job := NewJob(msg)
	kg.inHighPrio <- job
	return <-job.done
}
func (kg *KairosGateway) ProcessLowPrio(msg *nsq.Message) error {
	job := NewJob(msg)
	kg.inLowPrio <- job
	return <-job.done
}

func (kg *KairosGateway) process(qualifier string, msg *nsq.Message) error {
	format := "unknown"
	if msg.Body[0] == '\x00' {
		format = "msgFormatMetricDefinitionArrayJson"
	}
	log.Printf("DEBUG: processing %s msg %s. timestamp: %s. format: %s. attempts: %d\n", qualifier, msg.ID, time.Unix(0, msg.Timestamp), format, msg.Attempts)
	buf := bytes.NewReader(msg.Body[1:9])
	var id int64
	binary.Read(buf, binary.BigEndian, &id)
	log.Printf("DEBUG: DIETER-PROCESS-%s %d\n", qualifier, id)
	metrics := make([]*metricdef.IndvMetric, 0)
	if err := json.Unmarshal(msg.Body[9:], &metrics); err != nil {
		log.Printf("ERROR: failure to unmarshal message body: %s. skipping message", err)
		return nil
	}
	err := kg.kairos.SendMetricPointers(metrics)
	if err != nil {
		log.Printf("WARNING: can't send to kairosdb: %s. retrying later", err)
	}
	log.Printf("DEBUG: DIETER-FINISHED-%s %d  - %d metrics sent\n", qualifier, id, len(metrics))
	return err
}
<commit_msg>process metric batches in goroutine<commit_after>package main

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"log"
	"time"

	"github.com/bitly/go-nsq"
	"github.com/raintank/raintank-metric/metricdef"
	"github.com/raintank/raintank-metric/metricstore"
)

type KairosGateway struct {
	kairos     *metricstore.Kairosdb
	inHighPrio chan Job
	inLowPrio  chan Job
}

func NewKairosGateway() (*KairosGateway, error) {
	kairos, err := metricstore.NewKairosdb("http://kairosdb:8080")
	if err != nil {
		return nil, err
	}
	kg := &KairosGateway{
		kairos:     kairos,
		inHighPrio: make(chan Job, 4), // always useful to have some packets queue up to enforce the prio
		inLowPrio:  make(chan Job),
	}
	go kg.Run()
	return kg, nil
}

func (kg *KairosGateway) Run() {
	for {
		select {
		case job := <-kg.inHighPrio:
			job.done <- kg.process("high-prio", job.msg)
		default:
			select {
			case job := <-kg.inHighPrio:
				go func() {
					job.done <- kg.process("high-prio", job.msg)
				}()
			case job := <-kg.inLowPrio:
				job.done <- kg.process("low--prio", job.msg)
			}
		}
	}
}
func (kg *KairosGateway) ProcessHighPrio(msg *nsq.Message) error {
	job := NewJob(msg)
	kg.inHighPrio <- job
	return <-job.done
}
func (kg *KairosGateway) ProcessLowPrio(msg *nsq.Message) error {
	job := NewJob(msg)
	kg.inLowPrio <- job
	return <-job.done
}

func (kg *KairosGateway) process(qualifier string, msg *nsq.Message) error {
	format := "unknown"
	if msg.Body[0] == '\x00' {
		format = "msgFormatMetricDefinitionArrayJson"
	}
	log.Printf("DEBUG: processing %s msg %s. timestamp: %s. format: %s. attempts: %d\n", qualifier, msg.ID, time.Unix(0, msg.Timestamp), format, msg.Attempts)
	buf := bytes.NewReader(msg.Body[1:9])
	var id int64
	binary.Read(buf, binary.BigEndian, &id)
	log.Printf("DEBUG: DIETER-PROCESS-%s %d\n", qualifier, id)
	metrics := make([]*metricdef.IndvMetric, 0)
	if err := json.Unmarshal(msg.Body[9:], &metrics); err != nil {
		log.Printf("ERROR: failure to unmarshal message body: %s. skipping message", err)
		return nil
	}
	err := kg.kairos.SendMetricPointers(metrics)
	if err != nil {
		log.Printf("WARNING: can't send to kairosdb: %s. retrying later", err)
	}
	log.Printf("DEBUG: DIETER-FINISHED-%s %d  - %d metrics sent\n", qualifier, id, len(metrics))
	return err
}

{'fraction_non_alphanumeric': 0.15683675723437154, 'fraction_numerical': 0.0066269052352551355, 'mean_word_length': 3.9110629067245117, 'pattern_counts': {'":': 0, '<': 23, '<?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': '100177', 'n_tokens_mistral': 1883, 'n_tokens_neox': 1780, 'n_words': 414}