<commit_before>package main

import (
	"encoding/binary"
	"strings"

	influxdb "github.com/influxdb/influxdb/client"
	"github.com/yosisa/fluxion/buffer"
	"github.com/yosisa/fluxion/event"
	"github.com/yosisa/fluxion/plugin"
)

type Series influxdb.Series

func (s *Series) Size() int64 {
	return int64(binary.Size(s))
}

type Config struct {
	Server   string `codec:"server"`
	User     string `codec:"user"`
	Password string `codec:"password"`
	Database string `codec:"database"`
	UseUDP   bool   `codec:"use_udp"`
	StripTag int    `codec:"strip_tag"`
}

type InfluxdbOutput struct {
	env    *plugin.Env
	conf   *Config
	client *influxdb.Client
}

func (p *InfluxdbOutput) Name() string {
	return "out-influxdb"
}

func (p *InfluxdbOutput) Init(env *plugin.Env) (err error) {
	p.env = env
	p.conf = &Config{}
	if err = env.ReadConfig(p.conf); err != nil {
		return
	}
	return
}

func (p *InfluxdbOutput) Start() (err error) {
	p.client, err = influxdb.New(&influxdb.ClientConfig{
		Host:     p.conf.Server,
		Username: p.conf.User,
		Password: <PASSWORD>,
		Database: p.conf.Database,
		IsUDP:    p.conf.UseUDP,
	})
	return
}

func (p *InfluxdbOutput) Encode(r *event.Record) (buffer.Sizer, error) {
	p.env.Log.Info(r.Time)
	tag := r.Tag
	if p.conf.StripTag > 0 {
		parts := strings.SplitN(tag, ".", p.conf.StripTag+1)
		tag = parts[len(parts)-1]
	}

	s := &Series{
		Name:    tag,
		Columns: []string{"time"},
	}
	point := []interface{}{int64(float64(r.Time.UnixNano()) * 1e-6)}
	for k, v := range r.Value {
		s.Columns = append(s.Columns, k)
		point = append(point, v)
	}
	s.Points = append(s.Points, point)
	return s, nil
}

func (p *InfluxdbOutput) Write(l []buffer.Sizer) (int, error) {
	series := make([]*influxdb.Series, len(l))
	for i, v := range l {
		series[i] = (*influxdb.Series)(v.(*Series))
	}

	var err error
	if p.conf.UseUDP {
		err = p.client.WriteSeriesOverUDP(series)
	} else {
		err = p.client.WriteSeries(series)
	}
	if err != nil {
		return 0, err
	}
	return len(l), nil
}

func main() {
	plugin.New(func() plugin.Plugin {
		return &InfluxdbOutput{}
	}).Run()
}<commit_msg>Follow changes on fluxion<commit_after>package main

import (
	"encoding/binary"
	"strings"

	influxdb "github.com/influxdb/influxdb/client"
	"github.com/yosisa/fluxion/buffer"
	"github.com/yosisa/fluxion/message"
	"github.com/yosisa/fluxion/plugin"
)

type Series influxdb.Series

func (s *Series) Size() int64 {
	return int64(binary.Size(s))
}

type Config struct {
	Server   string `toml:"server"`
	User     string `toml:"user"`
	Password string `toml:"password"`
	Database string `toml:"database"`
	UseUDP   bool   `toml:"use_udp"`
	StripTag int    `toml:"strip_tag"`
}

type InfluxdbOutput struct {
	env    *plugin.Env
	conf   *Config
	client *influxdb.Client
}

func (p *InfluxdbOutput) Init(env *plugin.Env) (err error) {
	p.env = env
	p.conf = &Config{}
	if err = env.ReadConfig(p.conf); err != nil {
		return
	}
	return
}

func (p *InfluxdbOutput) Start() (err error) {
	p.client, err = influxdb.New(&influxdb.ClientConfig{
		Host:     p.conf.Server,
		Username: p.conf.User,
		Password: p.<PASSWORD>,
		Database: p.conf.Database,
		IsUDP:    p.conf.UseUDP,
	})
	return
}

func (p *InfluxdbOutput) Encode(ev *message.Event) (buffer.Sizer, error) {
	tag := ev.Tag
	if p.conf.StripTag > 0 {
		parts := strings.SplitN(tag, ".", p.conf.StripTag+1)
		tag = parts[len(parts)-1]
	}

	s := &Series{
		Name:    tag,
		Columns: []string{"time"},
	}
	point := []interface{}{int64(float64(ev.Time.UnixNano()) * 1e-6)}
	for k, v := range ev.Record {
		s.Columns = append(s.Columns, k)
		point = append(point, v)
	}
	s.Points = append(s.Points, point)
	return s, nil
}

func (p *InfluxdbOutput) Write(l []buffer.Sizer) (int, error) {
	series := make([]*influxdb.Series, len(l))
	for i, v := range l {
		series[i] = (*influxdb.Series)(v.(*Series))
	}

	var err error
	if p.conf.UseUDP {
		err = p.client.WriteSeriesOverUDP(series)
	} else {
		err = p.client.WriteSeries(series)
	}
	if err != nil {
		return 0, err
	}
	return len(l), nil
}

func (p *InfluxdbOutput) Close() error {
	return nil
}

func main() {
	plugin.New("out-influxdb", func() plugin.Plugin {
		return &InfluxdbOutput{}
	}).Run()
}
{'fraction_non_alphanumeric': 0.16567199809023633, 'fraction_numerical': 0.0066841728336118405, 'mean_word_length': 3.7831050228310503, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '4747114', 'n_tokens_mistral': 1752, 'n_tokens_neox': 1678, 'n_words': 367}