<commit_before>package mdata

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/gocql/gocql"
	"github.com/hailocab/go-hostpool"
	"github.com/raintank/metrictank/cassandra"
	"github.com/raintank/metrictank/mdata/chunk"
	"github.com/raintank/metrictank/stats"
	"github.com/raintank/worldping-api/pkg/log"
)

// write aggregated data to cassandra.

const Month_sec = 60 * 60 * 24 * 28

const keyspace_schema = `CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}  AND durable_writes = true`
const table_schema = `CREATE TABLE IF NOT EXISTS %s.%s (
    key ascii,
    ts int,
    data blob,
    PRIMARY KEY (key, ts)
) WITH CLUSTERING ORDER BY (ts DESC)
    AND compaction = { 'class': 'TimeWindowCompactionStrategy', 'compaction_window_unit': 'HOURS', 'compaction_window_size': '%d', 'tombstone_threshold': '0.2', 'tombstone_compaction_interval': '86400'}
    AND compression = { 'class': 'LZ4Compressor' }`
const Table_name_format = `metric_%d`

var (
	errChunkTooSmall  = errors.New("unpossibly small chunk in cassandra")
	errStartBeforeEnd = errors.New("start must be before end.")
	errTableNotFound  = errors.New("table for given TTL not found")

	// metric store.cassandra.get.exec is the duration of getting from cassandra store
	cassGetExecDuration = stats.NewLatencyHistogram15s32("store.cassandra.get.exec")
	// metric store.cassandra.get.wait is the duration of the get spent in the queue
	cassGetWaitDuration = stats.NewLatencyHistogram12h32("store.cassandra.get.wait")
	// metric store.cassandra.put.exec is the duration of putting in cassandra store
	cassPutExecDuration = stats.NewLatencyHistogram15s32("store.cassandra.put.exec")
	// metric store.cassandra.put.wait is the duration of a put in the wait queue
	cassPutWaitDuration = stats.NewLatencyHistogram12h32("store.cassandra.put.wait")

	// metric store.cassandra.chunks_per_row is how many chunks are retrieved per row in get queries
	cassChunksPerRow = stats.NewMeter32("store.cassandra.chunks_per_row", false)
	// metric store.cassandra.rows_per_response is how many rows come per get response
	cassRowsPerResponse = stats.NewMeter32("store.cassandra.rows_per_response", false)
	// metric store.cassandra.get_chunks is the duration of how long it takes to get chunks
	cassGetChunksDuration = stats.NewLatencyHistogram15s32("store.cassandra.get_chunks")
	// metric store.cassandra.to_iter is the duration of converting chunks to iterators
	cassToIterDuration = stats.NewLatencyHistogram15s32("store.cassandra.to_iter")

	// metric store.cassandra.chunk_operations.save_ok is counter of successful saves
	chunkSaveOk = stats.NewCounter32("store.cassandra.chunk_operations.save_ok")
	// metric store.cassandra.chunk_operations.save_fail is counter of failed saves
	chunkSaveFail = stats.NewCounter32("store.cassandra.chunk_operations.save_fail")
	// metric store.cassandra.chunk_size.at_save is the sizes of chunks seen when saving them
	chunkSizeAtSave = stats.NewMeter32("store.cassandra.chunk_size.at_save", true)
	// metric store.cassandra.chunk_size.at_load is the sizes of chunks seen when loading them
	chunkSizeAtLoad = stats.NewMeter32("store.cassandra.chunk_size.at_load", true)

	errmetrics = cassandra.NewErrMetrics("store.cassandra")
)

/*
https://godoc.org/github.com/gocql/gocql#Session
Session is the interface used by users to interact with the database.
It's safe for concurrent use by multiple goroutines and a typical usage scenario is to have one global session
object to interact with the whole Cassandra cluster.
*/

type TTLTables map[uint32]ttlTable
type ttlTable struct {
	Table      string
	WindowSize uint32
}

type CassandraStore struct {
	Session     *gocql.Session
	writeQueues []chan *ChunkWriteRequest
	readQueue   chan *ChunkReadRequest
	ttlTables   TTLTables
}

func ttlUnits(ttl uint32) float64 {
	// convert ttl to hours
	return float64(ttl) / (60 * 60)
}

func PrepareChunkData(span uint32, data []byte) []byte {
	chunkSizeAtSave.Value(len(data))
	version := chunk.FormatStandardGoTszWithSpan
	buf := new(bytes.Buffer)
	binary.Write(buf, binary.LittleEndian, version)

	spanCode, ok := chunk.RevChunkSpans[span]
	if !ok {
		// it's probably better to panic than to persist the chunk with a wrong length
		panic(fmt.Sprintf("Chunk span invalid: %d", span))
	}
	binary.Write(buf, binary.LittleEndian, spanCode)
	buf.Write(data)
	return buf.Bytes()
}

func GetTTLTables(ttls []uint32, windowFactor int, nameFormat string) TTLTables {
	tables := make(TTLTables)
	for _, ttl := range ttls {
		tables[ttl] = GetTTLTable(ttl, windowFactor, nameFormat)
	}
	return tables
}

func GetTTLTable(ttl uint32, windowFactor int, nameFormat string) ttlTable {
	/*
	 * the purpose of this is to bucket metrics of similar TTLs.
	 * we first calculate the largest power of 2 that's smaller than the TTL and then divide the result by
	 * the window factor. for example with a window factor of 20 we want to group the metrics like this:
	 *
	 * generated with: https://gist.github.com/replay/69ad7cfd523edfa552cd12851fa74c58
	 *
	 * +------------------------+---------------+---------------------+----------+
	 * |              TTL hours |    table_name | window_size (hours) | sstables |
	 * +------------------------+---------------+---------------------+----------+
	 * |         0 <= hours < 1 |     metrics_0 |                   1 |    0 - 2 |
	 * |         1 <= hours < 2 |     metrics_1 |                   1 |    1 - 3 |
	 * |         2 <= hours < 4 |     metrics_2 |                   1 |    2 - 5 |
	 * |         4 <= hours < 8 |     metrics_4 |                   1 |    4 - 9 |
	 * |        8 <= hours < 16 |     metrics_8 |                   1 |   8 - 17 |
	 * |       16 <= hours < 32 |    metrics_16 |                   1 |  16 - 33 |
	 * |       32 <= hours < 64 |    metrics_32 |                   2 |  16 - 33 |
	 * |      64 <= hours < 128 |    metrics_64 |                   4 |  16 - 33 |
	 * |     128 <= hours < 256 |   metrics_128 |                   7 |  19 - 38 |
	 * |     256 <= hours < 512 |   metrics_256 |                  13 |  20 - 41 |
	 * |    512 <= hours < 1024 |   metrics_512 |                  26 |  20 - 41 |
	 * |   1024 <= hours < 2048 |  metrics_1024 |                  52 |  20 - 41 |
	 * |   2048 <= hours < 4096 |  metrics_2048 |                 103 |  20 - 41 |
	 * |   4096 <= hours < 8192 |  metrics_4096 |                 205 |  20 - 41 |
	 * |  8192 <= hours < 16384 |  metrics_8192 |                 410 |  20 - 41 |
	 * | 16384 <= hours < 32768 | metrics_16384 |                 820 |  20 - 41 |
	 * | 32768 <= hours < 65536 | metrics_32768 |                1639 |  20 - 41 |
	 * +------------------------+---------------+---------------------+----------+
	 */

	// calculate the pre factor window by finding the largest power of 2 that's smaller than ttl
	preFactorWindow := uint32(math.Exp2(math.Floor(math.Log2(ttlUnits(ttl)))))
	tableName := fmt.Sprintf(nameFormat, preFactorWindow)
	return ttlTable{
		Table:      tableName,
		WindowSize: preFactorWindow/uint32(windowFactor) + 1,
	}
}

func NewCassandraStore(addrs, keyspace, consistency, CaPath, Username, Password, hostSelectionPolicy string, timeout, readers, writers, readqsize, writeqsize, retries, protoVer, windowFactor int, ssl, auth, hostVerification bool, ttls []uint32) (*CassandraStore, error) {

	stats.NewGauge32("store.cassandra.write_queue.size").Set(writeqsize)
	stats.NewGauge32("store.cassandra.num_writers").Set(writers)

	cluster := gocql.NewCluster(strings.Split(addrs, ",")...)
	if ssl {
		cluster.SslOpts = &gocql.SslOptions{
			CaPath:                 CaPath,
			EnableHostVerification: hostVerification,
		}
	}
	if auth {
		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username: Username,
			Password: Password,
		}
	}
	cluster.Consistency = gocql.ParseConsistency(consistency)
	cluster.Timeout = time.Duration(timeout) * time.Millisecond
	cluster.NumConns = writers
	cluster.ProtoVersion = protoVer
	var err error
	tmpSession, err := cluster.CreateSession()
	if err != nil {
		return nil, err
	}
	// ensure the keyspace and table exist.
	err = tmpSession.Query(fmt.Sprintf(keyspace_schema, keyspace)).Exec()
	if err != nil {
		return nil, err
	}

	ttlTables := GetTTLTables(ttls, windowFactor, Table_name_format)
	for _, result := range ttlTables {
		err := tmpSession.Query(fmt.Sprintf(table_schema, keyspace, result.Table, result.WindowSize)).Exec()
		if err != nil {
			return nil, err
		}
	}

	if err != nil {
		return nil, err
	}
	tmpSession.Close()
	cluster.Keyspace = keyspace
	cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: retries}

	switch hostSelectionPolicy {<commit_msg>set gc_grace_seconds to the compaction_window_size<commit_after>package mdata

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"math"
	"sort"
	"strings"
	"time"

	"github.com/gocql/gocql"
	"github.com/hailocab/go-hostpool"
	"github.com/raintank/metrictank/cassandra"
	"github.com/raintank/metrictank/mdata/chunk"
	"github.com/raintank/metrictank/stats"
	"github.com/raintank/worldping-api/pkg/log"
)

// write aggregated data to cassandra.

const Month_sec = 60 * 60 * 24 * 28

const keyspace_schema = `CREATE KEYSPACE IF NOT EXISTS %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}  AND durable_writes = true`
const table_schema = `CREATE TABLE IF NOT EXISTS %s.%s (
    key ascii,
    ts int,
    data blob,
    PRIMARY KEY (key, ts)
) WITH CLUSTERING ORDER BY (ts DESC)
    AND compaction = { 'class': 'TimeWindowCompactionStrategy', 'compaction_window_unit': 'HOURS', 'compaction_window_size': '%d', 'tombstone_threshold': '0.2', 'tombstone_compaction_interval': '86400'}
    AND compression = { 'class': 'LZ4Compressor' }
    AND gc_grace_seconds = %d`
const Table_name_format = `metric_%d`

var (
	errChunkTooSmall  = errors.New("unpossibly small chunk in cassandra")
	errStartBeforeEnd = errors.New("start must be before end.")
	errTableNotFound  = errors.New("table for given TTL not found")

	// metric store.cassandra.get.exec is the duration of getting from cassandra store
	cassGetExecDuration = stats.NewLatencyHistogram15s32("store.cassandra.get.exec")
	// metric store.cassandra.get.wait is the duration of the get spent in the queue
	cassGetWaitDuration = stats.NewLatencyHistogram12h32("store.cassandra.get.wait")
	// metric store.cassandra.put.exec is the duration of putting in cassandra store
	cassPutExecDuration = stats.NewLatencyHistogram15s32("store.cassandra.put.exec")
	// metric store.cassandra.put.wait is the duration of a put in the wait queue
	cassPutWaitDuration = stats.NewLatencyHistogram12h32("store.cassandra.put.wait")

	// metric store.cassandra.chunks_per_row is how many chunks are retrieved per row in get queries
	cassChunksPerRow = stats.NewMeter32("store.cassandra.chunks_per_row", false)
	// metric store.cassandra.rows_per_response is how many rows come per get response
	cassRowsPerResponse = stats.NewMeter32("store.cassandra.rows_per_response", false)
	// metric store.cassandra.get_chunks is the duration of how long it takes to get chunks
	cassGetChunksDuration = stats.NewLatencyHistogram15s32("store.cassandra.get_chunks")
	// metric store.cassandra.to_iter is the duration of converting chunks to iterators
	cassToIterDuration = stats.NewLatencyHistogram15s32("store.cassandra.to_iter")

	// metric store.cassandra.chunk_operations.save_ok is counter of successful saves
	chunkSaveOk = stats.NewCounter32("store.cassandra.chunk_operations.save_ok")
	// metric store.cassandra.chunk_operations.save_fail is counter of failed saves
	chunkSaveFail = stats.NewCounter32("store.cassandra.chunk_operations.save_fail")
	// metric store.cassandra.chunk_size.at_save is the sizes of chunks seen when saving them
	chunkSizeAtSave = stats.NewMeter32("store.cassandra.chunk_size.at_save", true)
	// metric store.cassandra.chunk_size.at_load is the sizes of chunks seen when loading them
	chunkSizeAtLoad = stats.NewMeter32("store.cassandra.chunk_size.at_load", true)

	errmetrics = cassandra.NewErrMetrics("store.cassandra")
)

/*
https://godoc.org/github.com/gocql/gocql#Session
Session is the interface used by users to interact with the database.
It's safe for concurrent use by multiple goroutines and a typical usage scenario is to have one global session
object to interact with the whole Cassandra cluster.
*/

type TTLTables map[uint32]ttlTable
type ttlTable struct {
	Table      string
	WindowSize uint32
}

type CassandraStore struct {
	Session     *gocql.Session
	writeQueues []chan *ChunkWriteRequest
	readQueue   chan *ChunkReadRequest
	ttlTables   TTLTables
}

func ttlUnits(ttl uint32) float64 {
	// convert ttl to hours
	return float64(ttl) / (60 * 60)
}

func PrepareChunkData(span uint32, data []byte) []byte {
	chunkSizeAtSave.Value(len(data))
	version := chunk.FormatStandardGoTszWithSpan
	buf := new(bytes.Buffer)
	binary.Write(buf, binary.LittleEndian, version)

	spanCode, ok := chunk.RevChunkSpans[span]
	if !ok {
		// it's probably better to panic than to persist the chunk with a wrong length
		panic(fmt.Sprintf("Chunk span invalid: %d", span))
	}
	binary.Write(buf, binary.LittleEndian, spanCode)
	buf.Write(data)
	return buf.Bytes()
}

func GetTTLTables(ttls []uint32, windowFactor int, nameFormat string) TTLTables {
	tables := make(TTLTables)
	for _, ttl := range ttls {
		tables[ttl] = GetTTLTable(ttl, windowFactor, nameFormat)
	}
	return tables
}

func GetTTLTable(ttl uint32, windowFactor int, nameFormat string) ttlTable {
	/*
	 * the purpose of this is to bucket metrics of similar TTLs.
	 * we first calculate the largest power of 2 that's smaller than the TTL and then divide the result by
	 * the window factor. for example with a window factor of 20 we want to group the metrics like this:
	 *
	 * generated with: https://gist.github.com/replay/69ad7cfd523edfa552cd12851fa74c58
	 *
	 * +------------------------+---------------+---------------------+----------+
	 * |              TTL hours |    table_name | window_size (hours) | sstables |
	 * +------------------------+---------------+---------------------+----------+
	 * |         0 <= hours < 1 |     metrics_0 |                   1 |    0 - 2 |
	 * |         1 <= hours < 2 |     metrics_1 |                   1 |    1 - 3 |
	 * |         2 <= hours < 4 |     metrics_2 |                   1 |    2 - 5 |
	 * |         4 <= hours < 8 |     metrics_4 |                   1 |    4 - 9 |
	 * |        8 <= hours < 16 |     metrics_8 |                   1 |   8 - 17 |
	 * |       16 <= hours < 32 |    metrics_16 |                   1 |  16 - 33 |
	 * |       32 <= hours < 64 |    metrics_32 |                   2 |  16 - 33 |
	 * |      64 <= hours < 128 |    metrics_64 |                   4 |  16 - 33 |
	 * |     128 <= hours < 256 |   metrics_128 |                   7 |  19 - 38 |
	 * |     256 <= hours < 512 |   metrics_256 |                  13 |  20 - 41 |
	 * |    512 <= hours < 1024 |   metrics_512 |                  26 |  20 - 41 |
	 * |   1024 <= hours < 2048 |  metrics_1024 |                  52 |  20 - 41 |
	 * |   2048 <= hours < 4096 |  metrics_2048 |                 103 |  20 - 41 |
	 * |   4096 <= hours < 8192 |  metrics_4096 |                 205 |  20 - 41 |
	 * |  8192 <= hours < 16384 |  metrics_8192 |                 410 |  20 - 41 |
	 * | 16384 <= hours < 32768 | metrics_16384 |                 820 |  20 - 41 |
	 * | 32768 <= hours < 65536 | metrics_32768 |                1639 |  20 - 41 |
	 * +------------------------+---------------+---------------------+----------+
	 */

	// calculate the pre factor window by finding the largest power of 2 that's smaller than ttl
	preFactorWindow := uint32(math.Exp2(math.Floor(math.Log2(ttlUnits(ttl)))))
	tableName := fmt.Sprintf(nameFormat, preFactorWindow)
	return ttlTable{
		Table:      tableName,
		WindowSize: preFactorWindow/uint32(windowFactor) + 1,
	}
}

func NewCassandraStore(addrs, keyspace, consistency, CaPath, Username, Password, hostSelectionPolicy string, timeout, readers, writers, readqsize, writeqsize, retries, protoVer, windowFactor int, ssl, auth, hostVerification bool, ttls []uint32) (*CassandraStore, error) {

	stats.NewGauge32("store.cassandra.write_queue.size").Set(writeqsize)
	stats.NewGauge32("store.cassandra.num_writers").Set(writers)

	cluster := gocql.NewCluster(strings.Split(addrs, ",")...)
	if ssl {
		cluster.SslOpts = &gocql.SslOptions{
			CaPath:                 CaPath,
			EnableHostVerification: hostVerification,
		}
	}
	if auth {
		cluster.Authenticator = gocql.PasswordAuthenticator{
			Username: Username,
			Password: Password,
		}
	}
	cluster.Consistency = gocql.ParseConsistency(consistency)
	cluster.Timeout = time.Duration(timeout) * time.Millisecond
	cluster.NumConns = writers
	cluster.ProtoVersion = protoVer
	var err error
	tmpSession, err := cluster.CreateSession()
	if err != nil {
		return nil, err
	}
	// ensure the keyspace and table exist.
	err = tmpSession.Query(fmt.Sprintf(keyspace_schema, keyspace)).Exec()
	if err != nil {
		return nil, err
	}

	ttlTables := GetTTLTables(ttls, windowFactor, Table_name_format)
	for _, result := range ttlTables {
		err := tmpSession.Query(fmt.Sprintf(table_schema, keyspace, result.Table, result.WindowSize, result.WindowSize*60*60)).Exec()
		if err != nil {
			return nil, err
		}
	}

	if err != nil {
		return nil, err
	}
	tmpSession.Close()
	cluster.Keyspace = keyspace
	cluster.RetryPolicy = &gocql.SimpleRetryPolicy{NumRetries: retries}

	switch hostSelectionPolicy {
{'fraction_non_alphanumeric': 0.12726444115301355, 'fraction_numerical': 0.038965477953742736, 'mean_word_length': 2.9600721858786376, 'pattern_counts': {'":': 0, '<': 71, '<?xml version=': 0, '>': 3, 'https://': 4, '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': '7286254', 'n_tokens_mistral': 6204, 'n_tokens_neox': 5543, 'n_words': 1470}