<commit_before>package influx

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/influxdata/influxdb/client/v2"
	"github.com/jcelliott/lumber"
	"github.com/spf13/viper"

	"github.com/nanopack/pulse/plexer"
)

var clientConn client.Client

func Query(sql string) (*client.Response, error) {
	c, err := influxClient()
	if err != nil {
		return nil, err
	}
	return c.Query(client.NewQuery(fmt.Sprint(sql), "statistics", "s"))
}

func Insert(messages plexer.MessageSet) error {
	lumber.Trace("[PULSE :: INFLUX] Insert: %v...", messages)
	tags := make(map[string]string, 0)
	for _, tag := range messages.Tags {
		elems := strings.SplitN(tag, ":", 2)
		if len(elems) < 2 {
			continue
		}

		tags[elems[0]] = elems[1]
	}

	fields := make(map[string]interface{}, len(messages.Messages))
	for _, message := range messages.Messages {
		value, err := strconv.ParseFloat(message.Data, 64)
		if err != nil {
			value = -1
		}
		fields[message.ID] = value
	}

	// create a point
	point, err := client.NewPoint("metrics", tags, fields, time.Now())
	if err != nil {
		return err
	}
	return writePoints("statistics", "2.days", *point)
}

func writePoints(database, retain string, point client.Point) error {
	// Create a new point batch
	batchPoint, _ := client.NewBatchPoints(client.BatchPointsConfig{
		Database:        database,
		RetentionPolicy: retain,
		Precision:       "s",
	})

	batchPoint.AddPoint(&point)
	c, err := influxClient()
	if err != nil {
		return err
	}
	return c.Write(batchPoint)
}

func influxClient() (client.Client, error) {
	var err error

	if clientConn != nil {
		return clientConn, nil
	}<commit_msg>make the message set parsing in influx account for the full gambit of options available in the messageset<commit_after>package influx

import (
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/influxdata/influxdb/client/v2"
	"github.com/jcelliott/lumber"
	"github.com/spf13/viper"

	"github.com/nanopack/pulse/plexer"
)

var clientConn client.Client

func Query(sql string) (*client.Response, error) {
	c, err := influxClient()
	if err != nil {
		return nil, err
	}
	return c.Query(client.NewQuery(fmt.Sprint(sql), "statistics", "s"))
}

func Insert(messageSet plexer.MessageSet) error {
	lumber.Trace("[PULSE :: INFLUX] Insert: %v...", messageSet)

	// create a set of points we will be inserting
	points := []*client.Point{}

	for _, message := range messageSet.Messages {
		// create a list of tags for each message
		tags := map[string]string{}

		// make sure to include the MessageSet's tags
		for _, tag := range append(messageSet.Tags, message.Tags...) {
			elems := strings.SplitN(tag, ":", 2)
			// only include tags with key:value format
			if len(elems) < 2 {
				continue
			}

			// insert the tag into my list of tags
			tags[elems[0]] = elems[1]
		}

		// if there
		value, err := strconv.ParseFloat(message.Data, 64)
		if err != nil {
			value = -1
		}

		// only one field per set of message tags.
		field := map[string]interface{}{message.ID: value}
		// create a point
		point, err := client.NewPoint("metrics", tags, field, time.Now())
		if err != nil {
			continue
		}
		points = append(points, point)
	}
	return writePoints("statistics", "2.days", points)
}

func writePoints(database, retain string, points []*client.Point) error {
	// Create a new point batch
	batchPoint, _ := client.NewBatchPoints(client.BatchPointsConfig{
		Database:        database,
		RetentionPolicy: retain,
		Precision:       "s",
	})
	for _, point := range points {
		batchPoint.AddPoint(point)
	}

	c, err := influxClient()
	if err != nil {
		return err
	}
	return c.Write(batchPoint)
}

func influxClient() (client.Client, error) {
	var err error

	if clientConn != nil {
		return clientConn, nil
	}
{'fraction_non_alphanumeric': 0.14132475194422098, 'fraction_numerical': 0.006167873424510592, 'mean_word_length': 3.6566791510611734, 'pattern_counts': {'":': 2, '<': 5, '<?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': '1305547', 'n_tokens_mistral': 1393, 'n_tokens_neox': 1285, 'n_words': 345}