<commit_before>package cloudstress_agent_consumer

import (
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/plugins/inputs"
	"github.com/influxdata/telegraf/proto/result"

	zmq "github.com/pebbe/zmq4"
)

type CloudStressAgent struct {
	sync.Mutex

	SubscriberPort int
	RecvTimeout    int

	subscriber *zmq.Socket

	deltaValues map[string]map[string]uint64
	deltaRates  map[string]map[string]map[string]int64

	msgs chan []string
	done chan struct{}

	// keep the accumulator internally:
	acc telegraf.Accumulator

	prevTime  time.Time
	prevValue int64
	currValue int64
}

var sampleConfig = `
  ## cloudstress subscriber port
  subscriberPort = 40004
`

func (c *CloudStressAgent) SampleConfig() string {
	return sampleConfig
}

func (c *CloudStressAgent) Description() string {
	return "Read metrics from cloudstress agents"
}

func (c *CloudStressAgent) Start(acc telegraf.Accumulator) error {
	c.Lock()
	defer c.Unlock()

	c.acc = acc
	//acc.SetPrecision(time.Second, 0)

	c.msgs = make(chan []string)
	c.done = make(chan struct{})

	c.deltaValues = make(map[string]map[string]uint64)
	c.deltaRates = make(map[string]map[string]map[string]int64)

	c.subscriber, _ = zmq.NewSocket(zmq.SUB)
	c.subscriber.Bind("tcp://*:" + strconv.Itoa(c.SubscriberPort))
	//for _, agentIp := range c.AgentList {
	//	c.subscriber.Connect("tcp://" + agentIp + ":" + strconv.Itoa(c.PublisherPort))
	//}
	c.subscriber.SetSubscribe("")

	// Start the zmq message subscriber
	go c.subscribe()

	log.Printf("I! Started the cloudstress agent consumer service. Subscribing on *:%d\n", c.SubscriberPort)

	return nil
}

func (c *CloudStressAgent) Stop() {
	c.Lock()
	defer c.Unlock()

	close(c.done)
	log.Printf("I! Stopping the cloudstress agent consumer service\n")
	if err := c.subscriber.Close(); err != nil {
		log.Printf("E! Error closing cloudstress agent consumer service: %s\n", err.Error())
	}
}

func (c *CloudStressAgent) Gather(acc telegraf.Accumulator) error {
	currTime := time.Now()
	diffTime := currTime.Sub(c.prevTime) / time.Second
	c.prevTime = currTime
	diffValue := c.currValue - c.prevValue
	c.prevValue = c.currValue

	if diffTime == 0 {
		return nil
	}

	rate := float64(diffValue) / float64(diffTime)
	log.Printf("I! Processed %f cloud-stress agent metrics per second\n", rate)
	return nil
}

// subscribe() reads all incoming messages from the cloudstress agents, and parses them into
// influxdb metric points.
func (c *CloudStressAgent) subscribe() {
	go c.processMessages()
	for {
		msg, err := c.subscriber.RecvMessage(0)
		if err != nil {
			break
		} else {
			c.msgs <- msg
		}
	}
}

func (c *CloudStressAgent) processMessages() {
	for {
		select {
		case <-c.done:
			return
		case msg := <-c.msgs:
			updateMsg := &result.Update{}
			err := proto.Unmarshal([]byte(msg[1]), updateMsg)
			if err != nil {
				log.Fatal("E! unmarshaling error: ", err)
			}

			tags := make(map[string]string)
			tags["uuid"] = updateMsg.GetUuid()

			for _, d := range updateMsg.Tags {
				tags[*d.Key] = *d.Value
			}

			timestamp := int64(updateMsg.GetTimestamp())
			loadType := "cs_agent_" + strings.ToLower(updateMsg.GetType().String())

			switch updateMsg.GetType() {
			case result.Update_CPU:
				metrics := c.processCPUMessage(loadType, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currValue++
			case result.Update_MEMORY:
				metrics := c.processMemoryMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currValue++
			case result.Update_BLOCK:
				metrics := c.processBlockMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currValue++
			case result.Update_NETWORK:
				metrics := c.processNetworkMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currValue++
			default:
				log.Printf("E! Unknown Type\n")<commit_msg>Calculate and print total and average metrics messages processed per second as well as total and average number of metrics values processed per second<commit_after>package cloudstress_agent_consumer

import (
	"log"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/golang/protobuf/proto"
	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/plugins/inputs"
	"github.com/influxdata/telegraf/proto/result"

	zmq "github.com/pebbe/zmq4"
)

type CloudStressAgent struct {
	sync.Mutex

	SubscriberPort int
	RecvTimeout    int

	subscriber *zmq.Socket

	deltaValues map[string]map[string]uint64
	deltaRates  map[string]map[string]map[string]int64

	msgs chan []string
	done chan struct{}

	// keep the accumulator internally:
	acc telegraf.Accumulator

	running           bool
	startTime         time.Time
	totalTime         time.Duration
	prevTime          time.Time
	prevMsgsValue     int64
	currMsgsValue     int64
	totalMsgsValue    int64
	prevMetricsValue  int64
	currMetriscValue  int64
	totalMetricsValue int64
}

var sampleConfig = `
  ## cloudstress subscriber port
  subscriberPort = 40004
`

func (c *CloudStressAgent) SampleConfig() string {
	return sampleConfig
}

func (c *CloudStressAgent) Description() string {
	return "Read metrics from cloudstress agents"
}

func (c *CloudStressAgent) Start(acc telegraf.Accumulator) error {
	c.Lock()
	defer c.Unlock()

	c.acc = acc
	//acc.SetPrecision(time.Second, 0)

	c.msgs = make(chan []string)
	c.done = make(chan struct{})

	c.deltaValues = make(map[string]map[string]uint64)
	c.deltaRates = make(map[string]map[string]map[string]int64)

	c.subscriber, _ = zmq.NewSocket(zmq.SUB)
	c.subscriber.Bind("tcp://*:" + strconv.Itoa(c.SubscriberPort))
	//for _, agentIp := range c.AgentList {
	//	c.subscriber.Connect("tcp://" + agentIp + ":" + strconv.Itoa(c.PublisherPort))
	//}
	c.subscriber.SetSubscribe("")

	// Start the zmq message subscriber
	go c.subscribe()

	log.Printf("I! Started the cloudstress agent consumer service. Subscribing on *:%d\n", c.SubscriberPort)

	return nil
}

func (c *CloudStressAgent) Stop() {
	c.Lock()
	defer c.Unlock()

	close(c.done)
	log.Printf("I! Stopping the cloudstress agent consumer service\n")
	if err := c.subscriber.Close(); err != nil {
		log.Printf("E! Error closing cloudstress agent consumer service: %s\n", err.Error())
	}
}

func (c *CloudStressAgent) Gather(acc telegraf.Accumulator) error {
	currTime := time.Now()
	diffTime := currTime.Sub(c.prevTime) / time.Second
	c.prevTime = currTime
	diffMsgsValue := c.currMsgsValue - c.prevMsgsValue
	c.prevMsgsValue = c.currMsgsValue
	diffMetricsValue := c.currMetriscValue - c.prevMetricsValue
	c.prevMetricsValue = c.currMetriscValue

	if diffTime == 0 {
		return nil
	}

	if c.running && diffMsgsValue > 0 {
		c.totalTime = s.Sub(c.startTime) / time.Second
	} else {
		if diffMsgsValue == 0 {
			c.running = false
		}
	}

	rate := float64(diffMsgsValue) / float64(diffTime)
	log.Printf("I! Processed %f cloud-stress agent metric messages per second\n", rate)
	log.Printf("I! Processed %d total cloud-stress agent metric messages\n", c.totalMsgsValue)
	rate = float64(diffMetricsValue) / float64(diffTime)
	log.Printf("I! Processed %f cloud-stress agent metric values per second\n", rate)
	log.Printf("I! Processed %d total cloud-stress agent metric values\n", c.totalMetricsValue)
	rate = float64(c.totalMetricsValue) / float64(c.totalTime)
	log.Printf("I! Proccessed on average of %f agent metrics values per second", rate)
	return nil
}

// subscribe() reads all incoming messages from the cloudstress agents, and parses them into
// influxdb metric points.
func (c *CloudStressAgent) subscribe() {
	go c.processMessages()
	for {
		msg, err := c.subscriber.RecvMessage(0)
		if err != nil {
			break
		} else {
			if !c.running {
				c.startTime = time.Now()
				c.running = true
			}
			c.msgs <- msg
		}
	}
}

func (c *CloudStressAgent) processMessages() {
	for {
		select {
		case <-c.done:
			return
		case msg := <-c.msgs:
			updateMsg := &result.Update{}
			err := proto.Unmarshal([]byte(msg[1]), updateMsg)
			if err != nil {
				log.Fatal("E! unmarshaling error: ", err)
			}

			tags := make(map[string]string)
			tags["uuid"] = updateMsg.GetUuid()

			for _, d := range updateMsg.Tags {
				tags[*d.Key] = *d.Value
			}

			timestamp := int64(updateMsg.GetTimestamp())
			loadType := "cs_agent_" + strings.ToLower(updateMsg.GetType().String())

			switch updateMsg.GetType() {
			case result.Update_CPU:
				metrics := c.processCPUMessage(loadType, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currMsgsValue++
				c.totalMsgsValue++
				c.currMetriscValue += int64(len(metrics))
				c.totalMetricsValue += int64(len(metrics))
			case result.Update_MEMORY:
				metrics := c.processMemoryMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currMsgsValue++
				c.totalMsgsValue++
				c.currMetriscValue += int64(len(metrics))
				c.totalMetricsValue += int64(len(metrics))
			case result.Update_BLOCK:
				metrics := c.processBlockMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currMsgsValue++
				c.totalMsgsValue++
				c.currMetriscValue += int64(len(metrics))
				c.totalMetricsValue += int64(len(metrics))
			case result.Update_NETWORK:
				metrics := c.processNetworkMessage(loadType, timestamp, updateMsg)
				go c.acc.AddFields(loadType, metrics, tags, time.Now())
				c.currMsgsValue++
				c.totalMsgsValue++
				c.currMetriscValue += int64(len(metrics))
				c.totalMetricsValue += int64(len(metrics))
			default:
				log.Printf("E! Unknown Type\n")
{'fraction_non_alphanumeric': 0.130851722004537, 'fraction_numerical': 0.00928026397195298, 'mean_word_length': 4.022786121180736, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 4, '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': '1695641', 'n_tokens_mistral': 3749, 'n_tokens_neox': 3372, 'n_words': 761}