<commit_before>package persister

import (
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"time"

	"devroom.ru/lomik/carbon/points"

	"github.com/Sirupsen/logrus"
	"github.com/lomik/go-whisper"
)

// Whisper write data to *.wsp files
type Whisper struct {
	in          chan *points.Points
	exit        chan bool
	schemas     *WhisperSchemas
	rootPath    string
	graphPrefix string
	commited    uint64 // (updateOperations << 32) + commitedPoints
	created     uint32 // counter
}

// NewWhisper create instance of Whisper
func NewWhisper(rootPath string, schemas *WhisperSchemas, in chan *points.Points) *Whisper {
	return &Whisper{
		in:       in,
		exit:     make(chan bool),
		schemas:  schemas,
		rootPath: rootPath,
	}
}

// SetGraphPrefix for internal cache metrics
func (p *Whisper) SetGraphPrefix(prefix string) {
	p.graphPrefix = prefix
}

// Stat sends internal statistics to cache
func (p *Whisper) Stat(metric string, value float64) {
	p.in <- points.OnePoint(
		fmt.Sprintf("%spersister.%s", p.graphPrefix, metric),
		value,
		time.Now().Unix(),
	)
}

func (p *Whisper) store(values *points.Points) {
	// @TODO: lock or shard by hash(metricName), no thread safe
	path := filepath.Join(p.rootPath, strings.Replace(values.Metric, ".", "/", -1)+".wsp")

	w, err := whisper.Open(path)
	if err != nil {
		schema := p.schemas.match(values.Metric)
		if schema == nil {
			logrus.Errorf("No storage schema defined for %s", values.Metric)
			return
		}

		logrus.Infof("Creating %s: %s, retention %#v (section %#v)",
			values.Metric, path, schema.retentionStr, schema.name)

		if err = os.MkdirAll(filepath.Dir(path), os.ModeDir|os.ModePerm); err != nil {
			logrus.Error(err)
			return
		}

		w, err = whisper.Create(path, schema.retentions, whisper.Average, 0.5)
		if err != nil {
			logrus.Warningf("Failed to create new whisper file %s: %s", path, err.Error())
			return
		}

		atomic.AddUint32(&p.created, 1)
	}

	points := make([]*whisper.TimeSeriesPoint, len(values.Data))
	for i, r := range values.Data {
		points[i] = &whisper.TimeSeriesPoint{Time: int(r.Timestamp), Value: r.Value}
	}

	atomic.AddUint64(&p.commited, (uint64(1)<<32)+uint64(len(values.Data)))
	defer w.Close()

	defer func() {
		if r := recover(); r != nil {
			logrus.Warningf("UpdateMany %s recovered: %s", path, r)
		}
	}()
	w.UpdateMany(points)
}

func (p *Whisper) worker(in chan *points.Points) {
	for {
		select {
		case <-p.exit:
			break
		case values := <-in:
			p.store(values)
		}
	}
}

func (p *Whisper) statWorker() {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-p.exit:
			break
		case <-ticker.C:<commit_msg>new line<commit_after>package persister

import (
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync/atomic"
	"time"

	"devroom.ru/lomik/carbon/points"

	"github.com/Sirupsen/logrus"
	"github.com/lomik/go-whisper"
)

// Whisper write data to *.wsp files
type Whisper struct {
	in          chan *points.Points
	exit        chan bool
	schemas     *WhisperSchemas
	rootPath    string
	graphPrefix string
	commited    uint64 // (updateOperations << 32) + commitedPoints
	created     uint32 // counter
}

// NewWhisper create instance of Whisper
func NewWhisper(rootPath string, schemas *WhisperSchemas, in chan *points.Points) *Whisper {
	return &Whisper{
		in:       in,
		exit:     make(chan bool),
		schemas:  schemas,
		rootPath: rootPath,
	}
}

// SetGraphPrefix for internal cache metrics
func (p *Whisper) SetGraphPrefix(prefix string) {
	p.graphPrefix = prefix
}

// Stat sends internal statistics to cache
func (p *Whisper) Stat(metric string, value float64) {
	p.in <- points.OnePoint(
		fmt.Sprintf("%spersister.%s", p.graphPrefix, metric),
		value,
		time.Now().Unix(),
	)
}

func (p *Whisper) store(values *points.Points) {
	// @TODO: lock or shard by hash(metricName), no thread safe
	path := filepath.Join(p.rootPath, strings.Replace(values.Metric, ".", "/", -1)+".wsp")

	w, err := whisper.Open(path)
	if err != nil {
		schema := p.schemas.match(values.Metric)
		if schema == nil {
			logrus.Errorf("No storage schema defined for %s", values.Metric)
			return
		}

		logrus.Infof("Creating %s: %s, retention %#v (section %#v)",
			values.Metric, path, schema.retentionStr, schema.name)

		if err = os.MkdirAll(filepath.Dir(path), os.ModeDir|os.ModePerm); err != nil {
			logrus.Error(err)
			return
		}

		w, err = whisper.Create(path, schema.retentions, whisper.Average, 0.5)
		if err != nil {
			logrus.Warningf("Failed to create new whisper file %s: %s", path, err.Error())
			return
		}

		atomic.AddUint32(&p.created, 1)
	}

	points := make([]*whisper.TimeSeriesPoint, len(values.Data))
	for i, r := range values.Data {
		points[i] = &whisper.TimeSeriesPoint{Time: int(r.Timestamp), Value: r.Value}
	}

	atomic.AddUint64(&p.commited, (uint64(1)<<32)+uint64(len(values.Data)))

	defer w.Close()

	defer func() {
		if r := recover(); r != nil {
			logrus.Warningf("UpdateMany %s recovered: %s", path, r)
		}
	}()
	w.UpdateMany(points)
}

func (p *Whisper) worker(in chan *points.Points) {
	for {
		select {
		case <-p.exit:
			break
		case values := <-in:
			p.store(values)
		}
	}
}

func (p *Whisper) statWorker() {
	ticker := time.NewTicker(time.Minute)
	defer ticker.Stop()

	for {
		select {
		case <-p.exit:
			break
		case <-ticker.C:
{'fraction_non_alphanumeric': 0.15141242937853108, 'fraction_numerical': 0.008662900188323917, 'mean_word_length': 3.5431993156544053, 'pattern_counts': {'":': 0, '<': 21, '<?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': '4471496', 'n_tokens_mistral': 2155, 'n_tokens_neox': 1973, 'n_words': 474}