617 lines
14 KiB
Go
617 lines
14 KiB
Go
package main
|
||
|
||
import (
|
||
"bufio"
|
||
"database/sql"
|
||
"fmt"
|
||
"log"
|
||
"os"
|
||
"sync"
|
||
"time"
|
||
|
||
_ "github.com/go-sql-driver/mysql"
|
||
bin "gordenko.dev/dima/bin/little"
|
||
"gordenko.dev/dima/qx"
|
||
)
|
||
|
||
type MetricType int
|
||
type MetricCategory int
|
||
|
||
const (
|
||
MetricTypeCumulative MetricType = 1
|
||
MetricTypeInstant MetricType = 2
|
||
// Категории метрик
|
||
CategoryVolume MetricCategory = 1
|
||
CategoryVolumeFlow MetricCategory = 2
|
||
CategoryEnergy MetricCategory = 3
|
||
CategoryMass MetricCategory = 5
|
||
CategoryMassFlow MetricCategory = 6
|
||
CategoryPower MetricCategory = 7
|
||
CategoryTemperature MetricCategory = 8
|
||
CategoryTempDiff MetricCategory = 9
|
||
CategoryLevel MetricCategory = 10
|
||
CategoryPressure MetricCategory = 11
|
||
CategoryFrequency MetricCategory = 12
|
||
CategoryRunningFrequency MetricCategory = 13
|
||
CategoryCurrent MetricCategory = 14
|
||
CategoryVoltage MetricCategory = 15
|
||
CategoryBusVoltage MetricCategory = 16
|
||
CategorySpeed MetricCategory = 17
|
||
CategoryTorque MetricCategory = 18
|
||
CategoryDuration MetricCategory = 19 // час роботи, роботи з помилкою, роботи без помилок
|
||
CategoryPressureSetting MetricCategory = 20 // уставка тиску
|
||
CategoryBatteryPower MetricCategory = 21
|
||
CategorySignalQuality MetricCategory = 22
|
||
CategoryError MetricCategory = 23
|
||
CategoryPercent MetricCategory = 24
|
||
CategoryResistance MetricCategory = 25
|
||
|
||
DSN = "root:zkx75fcy@/rrc"
|
||
)
|
||
|
||
var (
|
||
MetricCategoryToMetricType = map[MetricCategory]MetricType{
|
||
CategoryVolume: MetricTypeCumulative,
|
||
CategoryVolumeFlow: MetricTypeInstant,
|
||
CategoryEnergy: MetricTypeCumulative,
|
||
CategoryMass: MetricTypeCumulative,
|
||
CategoryMassFlow: MetricTypeInstant,
|
||
CategoryPower: MetricTypeInstant,
|
||
CategoryTemperature: MetricTypeInstant,
|
||
CategoryTempDiff: MetricTypeInstant,
|
||
CategoryLevel: MetricTypeInstant,
|
||
CategoryPressure: MetricTypeInstant,
|
||
CategoryPressureSetting: MetricTypeInstant,
|
||
CategoryFrequency: MetricTypeInstant,
|
||
CategoryRunningFrequency: MetricTypeInstant,
|
||
CategoryCurrent: MetricTypeInstant,
|
||
CategoryVoltage: MetricTypeInstant,
|
||
CategoryBusVoltage: MetricTypeInstant,
|
||
CategorySpeed: MetricTypeInstant,
|
||
CategoryTorque: MetricTypeInstant,
|
||
CategoryDuration: MetricTypeCumulative,
|
||
CategoryBatteryPower: MetricTypeInstant,
|
||
CategorySignalQuality: MetricTypeInstant,
|
||
CategoryError: MetricTypeInstant,
|
||
CategoryPercent: MetricTypeInstant,
|
||
CategoryResistance: MetricTypeInstant,
|
||
}
|
||
)
|
||
|
||
func CategoryToMetricType(category MetricCategory) MetricType {
|
||
switch category {
|
||
case CategoryVolume, CategoryMass, CategoryEnergy, CategoryDuration:
|
||
return MetricTypeCumulative
|
||
}
|
||
return MetricTypeInstant
|
||
}
|
||
|
||
/*
|
||
Список метрик:
|
||
ID, тип метрики, fracDigits, начало диапазона, конец диапазона.
|
||
|
||
Есть смысл записать показания в файлы. Итерация (запись в субд) пойдет быстрее.
|
||
|
||
Получить список метрик.
|
||
В цикле получить показания и записатьв файл?
|
||
*/
|
||
|
||
type Metric struct {
|
||
MetricID int64 `json:"metricID"`
|
||
MetricType MetricType `json:"metricType"`
|
||
FracDigits int `json:"fracDigits"`
|
||
}
|
||
|
||
func listMetrics(db *qx.Db) (list []Metric, err error) {
|
||
var tmp []struct {
|
||
MetricID int64
|
||
FracDigits int
|
||
ConfigMetricID int64
|
||
CustomCategory MetricCategory
|
||
}
|
||
|
||
err = db.ListQuery(&tmp, `
|
||
SELECT m.metricID, mp.fracDigits, mp.configMetricID, mp.customCategory
|
||
FROM metrics m
|
||
INNER JOIN metric_profiles mp ON mp.profileID = m.metricProfileID
|
||
ORDER BY m.metricID ASC`)
|
||
|
||
for _, x := range tmp {
|
||
var (
|
||
category MetricCategory
|
||
metricType MetricType
|
||
)
|
||
if x.CustomCategory > 0 {
|
||
metricType = CategoryToMetricType(x.CustomCategory)
|
||
} else {
|
||
if x.ConfigMetricID > 0 {
|
||
found, err := db.OneQuery(&category, `SELECT category FROM config_metrics WHERE metricID=?`, x.ConfigMetricID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if !found {
|
||
return nil, fmt.Errorf("not found configMetric %d", x.ConfigMetricID)
|
||
}
|
||
|
||
metricType = CategoryToMetricType(category)
|
||
} else {
|
||
return nil, fmt.Errorf("customCategory not set and configMetricID = 0\n")
|
||
}
|
||
}
|
||
|
||
list = append(list, Metric{
|
||
MetricID: x.MetricID,
|
||
FracDigits: x.FracDigits,
|
||
MetricType: metricType,
|
||
})
|
||
}
|
||
return
|
||
}
|
||
|
||
type Measure struct {
|
||
Time int64 `json:"tm" `
|
||
Value float64 `json:"value"`
|
||
}
|
||
|
||
func listMeasures(db *qx.Db, metricID int64) (list []Measure, err error) {
|
||
err = db.ListQuery(&list, "SELECT tm, value FROM f64 WHERE metricID=? ORDER BY tm ASC", metricID)
|
||
return
|
||
}
|
||
|
||
type AppendTask struct {
|
||
MetricID uint32
|
||
MetricType MetricType
|
||
FracDigits byte
|
||
}
|
||
|
||
func copyMeasures(db *qx.Db) (err error) {
|
||
metrics, err := listMetrics(db)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
fmt.Printf("found %d metrics\n", len(metrics))
|
||
|
||
wg := new(sync.WaitGroup)
|
||
stopCh := make(chan struct{})
|
||
taskCh := make(chan AppendTask)
|
||
|
||
for range 10 {
|
||
go writer(db, taskCh, wg, stopCh)
|
||
}
|
||
|
||
for _, metric := range metrics {
|
||
taskCh <- AppendTask{
|
||
MetricID: uint32(metric.MetricID),
|
||
MetricType: metric.MetricType,
|
||
FracDigits: byte(metric.FracDigits),
|
||
}
|
||
}
|
||
|
||
close(stopCh)
|
||
|
||
wg.Wait()
|
||
return nil
|
||
}
|
||
|
||
func writer(db *qx.Db, taskCh chan AppendTask, wg *sync.WaitGroup, stopCh chan struct{}) {
|
||
wg.Add(1)
|
||
|
||
for {
|
||
select {
|
||
case task := <-taskCh:
|
||
err := writeMetricIntoFile(db, task)
|
||
if err != nil {
|
||
log.Println(err)
|
||
}
|
||
|
||
case <-stopCh:
|
||
wg.Done()
|
||
return
|
||
}
|
||
}
|
||
|
||
}
|
||
|
||
// const maxQtyInPack = 65535
|
||
|
||
// func writer(taskCh chan AppendTask, wg *sync.WaitGroup, stopCh chan struct{}) {
|
||
// wg.Add(1)
|
||
|
||
// c, err := client.Connect(":12345")
|
||
// if err != nil {
|
||
// log.Fatalln(err)
|
||
// }
|
||
|
||
// for {
|
||
// select {
|
||
// case task := <-taskCh:
|
||
// err = writeMetricIntoOctopus(c, task)
|
||
// if err != nil {
|
||
// log.Println(err)
|
||
// }
|
||
|
||
// case <-stopCh:
|
||
// wg.Done()
|
||
// return
|
||
// }
|
||
// }
|
||
|
||
// }
|
||
|
||
func correctToMonotonic(measures []Measure) []Measure {
|
||
if len(measures) == 0 {
|
||
return nil
|
||
}
|
||
|
||
if measures[0].Value < 0 {
|
||
var (
|
||
idx int
|
||
m Measure
|
||
)
|
||
for idx, m = range measures {
|
||
if m.Value >= 0 {
|
||
break
|
||
}
|
||
}
|
||
measures = measures[idx:]
|
||
}
|
||
|
||
if len(measures) == 0 {
|
||
return nil
|
||
}
|
||
|
||
var (
|
||
baseValue float64
|
||
prevValue = measures[0].Value // как прислал прибор
|
||
)
|
||
for idx := 1; idx < len(measures); idx++ {
|
||
measure := measures[idx]
|
||
|
||
if measure.Value < 0 {
|
||
continue
|
||
}
|
||
|
||
if measure.Value < prevValue {
|
||
// произошел сброс
|
||
baseValue += prevValue
|
||
}
|
||
prevValue = measure.Value
|
||
|
||
if baseValue > 0 {
|
||
measure.Value += baseValue
|
||
measures[idx] = measure
|
||
}
|
||
}
|
||
return measures
|
||
}
|
||
|
||
func writeMetricIntoFile(db *qx.Db, task AppendTask) (err error) {
|
||
measures, err := listMeasures(db, int64(task.MetricID))
|
||
if err != nil {
|
||
return fmt.Errorf("listMeasures(%d): %s", task.MetricID, err)
|
||
}
|
||
//fmt.Printf("metric %d has %d measures\n", metric.MetricID, len(measures))
|
||
if len(measures) < 1000 {
|
||
return
|
||
}
|
||
|
||
filename := fmt.Sprintf("%d.%d", task.MetricID, task.FracDigits)
|
||
|
||
if task.MetricType == MetricTypeCumulative {
|
||
measures = correctToMonotonic(measures)
|
||
|
||
if len(measures) < 1000 {
|
||
return
|
||
}
|
||
|
||
filename = "cumulative/" + filename
|
||
} else {
|
||
filename = "instant/" + filename
|
||
}
|
||
|
||
file, err := os.OpenFile(filename, os.O_CREATE|os.O_RDWR, 0666)
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
dst := bufio.NewWriter(file)
|
||
|
||
for _, measure := range measures {
|
||
err = bin.WriteUint32(dst, uint32(measure.Time))
|
||
if err != nil {
|
||
return
|
||
}
|
||
err = bin.WriteFloat64(dst, measure.Value)
|
||
if err != nil {
|
||
return
|
||
}
|
||
}
|
||
|
||
err = dst.Flush()
|
||
if err != nil {
|
||
return
|
||
}
|
||
|
||
err = file.Close()
|
||
if err != nil {
|
||
return
|
||
}
|
||
fmt.Printf("%d: %d measures written\n", task.MetricID, len(measures))
|
||
return
|
||
}
|
||
|
||
// func writeMetricIntoOctopus(c *client.Connection, task AppendTask) error {
|
||
// err := c.AddMetric(proto.AddMetricReq{
|
||
// MetricID: task.MetricID,
|
||
// MetricType: octopus.MetricType(task.MetricType),
|
||
// FracDigits: int(task.FracDigits),
|
||
// })
|
||
// if err != nil {
|
||
// log.Fatal(err)
|
||
// } else {
|
||
// fmt.Println("metric added")
|
||
// }
|
||
|
||
// t1 := time.Now()
|
||
// var (
|
||
// prevTime int64
|
||
// pack []proto.Measure
|
||
// )
|
||
|
||
// for _, measure := range task.Measures {
|
||
// if measure.Time <= prevTime {
|
||
// continue
|
||
// } else {
|
||
// prevTime = measure.Time
|
||
// }
|
||
|
||
// pack = append(pack, proto.Measure{
|
||
// Timestamp: uint32(measure.Time),
|
||
// Value: measure.Value,
|
||
// })
|
||
|
||
// if len(pack) == maxQtyInPack {
|
||
// err = c.AppendMeasures(proto.AppendMeasuresReq{
|
||
// MetricID: task.MetricID,
|
||
// Measures: pack,
|
||
// })
|
||
// if err != nil {
|
||
// return fmt.Errorf("append measures (%d): %s", task.MetricID, err)
|
||
// } else {
|
||
// //fmt.Printf("written %d measures of %d\n", len(pack), len(task.Measures))
|
||
// pack = nil
|
||
// }
|
||
// }
|
||
// }
|
||
|
||
// if len(pack) > 0 {
|
||
// err = c.AppendMeasures(proto.AppendMeasuresReq{
|
||
// MetricID: task.MetricID,
|
||
// Measures: pack,
|
||
// })
|
||
// if err != nil {
|
||
// return fmt.Errorf("append measures (%d): %s", task.MetricID, err)
|
||
// } else {
|
||
// //fmt.Printf("written %d measures of %d\n", len(pack), len(task.Measures))
|
||
// }
|
||
// }
|
||
//fmt.Printf("written %d in %.2f seconds\n", len(task.Measures), time.Since(t1).Seconds())
|
||
//return nil
|
||
//}
|
||
|
||
func main() {
|
||
db, err := qx.Open("mysql", DSN)
|
||
if err != nil {
|
||
log.Fatalln(err)
|
||
}
|
||
|
||
copyAllMetrics(db)
|
||
//copyAllMetricsByOne()
|
||
//copyOneMetric(db, 134)
|
||
// metrics, err := listMetrics(db)
|
||
// if err != nil {
|
||
// log.Fatalln(err)
|
||
// }
|
||
|
||
// pretty.Println(metrics)
|
||
//checkAllMetrics(db)
|
||
}
|
||
|
||
func copyAllMetrics(db *qx.Db) {
|
||
t1 := time.Now()
|
||
err := copyMeasures(db)
|
||
if err != nil {
|
||
log.Fatalln(err)
|
||
}
|
||
|
||
fmt.Printf("total time: %.1f seconds\n", time.Since(t1).Seconds())
|
||
}
|
||
|
||
// func copyOneMetric(db *qx.Db, metricID int64) {
|
||
// measures, err := listMeasures(db, metricID)
|
||
// if err != nil {
|
||
// log.Fatalln(err)
|
||
// }
|
||
|
||
// c, err := client.Connect(":12345")
|
||
// if err != nil {
|
||
// log.Fatalln(err)
|
||
// }
|
||
|
||
// writeMetricIntoOctopus(c, AppendTask{
|
||
// MetricID: uint32(metricID),
|
||
// MetricType: MetricTypeInstant,
|
||
// FracDigits: 3,
|
||
// Measures: measures,
|
||
// })
|
||
|
||
// // measures = measures[:2300]
|
||
|
||
// // timestampsBuf := conbuf.New(nil)
|
||
// // timestamps := chunkenc.NewReverseTimeDeltaOfDeltaCompressor(timestampsBuf, 0)
|
||
|
||
// // valuesBuf := conbuf.New(nil)
|
||
// // values := chunkenc.NewReverseInstantDeltaCompressor(valuesBuf, 0, 3)
|
||
|
||
// // for _, measure := range measures[:10] {
|
||
// // // if measure.Time < 1686671940 {
|
||
// // // fmt.Printf("idx: %d\n", idx)
|
||
// // // continue
|
||
// // // }
|
||
|
||
// // //fmt.Printf("ts: %d, v: %.2f\n", measure.Time, measure.Value)
|
||
|
||
// // timestamps.Append(uint32(measure.Time))
|
||
// // values.Append(measure.Value)
|
||
// // }
|
||
|
||
// // timestampDecompressor := chunkenc.NewReverseTimeDeltaOfDeltaDecompressor(
|
||
// // timestampsBuf,
|
||
// // timestamps.Size(),
|
||
// // )
|
||
|
||
// // valueDecompressor := chunkenc.NewReverseInstantDeltaDecompressor(
|
||
// // valuesBuf,
|
||
// // values.Size(),
|
||
// // 3,
|
||
// // )
|
||
|
||
// // fmt.Println("----------------------------------------------------")
|
||
|
||
// // var (
|
||
// // value float64
|
||
// // timestamp uint32
|
||
// // done bool
|
||
// // )
|
||
|
||
// // for {
|
||
// // timestamp, done = timestampDecompressor.NextValue()
|
||
// // if done {
|
||
// // break
|
||
// // }
|
||
|
||
// // value, done = valueDecompressor.NextValue()
|
||
// // if done {
|
||
// // fmt.Printf("ts before crash: %d\n", timestamp)
|
||
// // panic("FUCK")
|
||
// // }
|
||
|
||
// // fmt.Printf("ts: %d, v: %.2f\n", timestamp, value)
|
||
// // }
|
||
// }
|
||
|
||
// func checkAllMetrics(db *qx.Db) {
|
||
// metrics, err := listMetrics(db)
|
||
// if err != nil {
|
||
// return
|
||
// }
|
||
|
||
// //metrics = metrics[:100]
|
||
|
||
// c, err := client.Connect(":12345")
|
||
// if err != nil {
|
||
// log.Fatalln(err)
|
||
// }
|
||
|
||
// var total int
|
||
|
||
// for _, metric := range metrics {
|
||
// if metric.MetricType == MetricTypeCumulative {
|
||
// list, err := c.ListAllCumulativeMeasures(uint32(metric.MetricID))
|
||
// if err != nil {
|
||
// fmt.Printf("ListAllCumulativeMeasures(%d): %s\n", metric.MetricID, err)
|
||
// } else {
|
||
// total += len(list)
|
||
// }
|
||
// } else {
|
||
// list, err := c.ListAllInstantMeasures(uint32(metric.MetricID))
|
||
// if err != nil {
|
||
// fmt.Printf("ListAllInstantMeasures(%d): %s\n", metric.MetricID, err)
|
||
// } else {
|
||
// total += len(list)
|
||
// }
|
||
// }
|
||
// //fmt.Printf("#%d: %d\n", idx, metric.MetricID)
|
||
|
||
// }
|
||
// fmt.Printf("total: %d\n", total)
|
||
// }
|
||
|
||
// копирование равномерное
|
||
type Reading struct {
|
||
MetricID int
|
||
ReadAt time.Time
|
||
Value float64
|
||
}
|
||
|
||
func copyAllMetricsByOne() {
|
||
mydb, err := qx.Open("mysql", DSN)
|
||
if err != nil {
|
||
log.Fatalln(err)
|
||
}
|
||
|
||
metrics, err := listMetrics(mydb)
|
||
if err != nil {
|
||
return
|
||
}
|
||
mydb.Close()
|
||
|
||
//metrics = metrics[:100]
|
||
|
||
fmt.Printf("found %d metrics\n", len(metrics))
|
||
|
||
// Підключення до MySQL
|
||
db, err := sql.Open("mysql", DSN)
|
||
if err != nil {
|
||
log.Fatal(err)
|
||
}
|
||
defer db.Close()
|
||
|
||
// Завантажити всі показники, відсортовані по часу
|
||
var (
|
||
ts = time.Now().Unix()
|
||
total int
|
||
)
|
||
|
||
for {
|
||
// Рівномірна вставка: поки є хоча б один показник
|
||
tx, err := db.Begin()
|
||
if err != nil {
|
||
log.Fatalln(err)
|
||
}
|
||
|
||
localTs := ts
|
||
|
||
for range 100 {
|
||
localTs += 60
|
||
for _, metric := range metrics {
|
||
_, err = tx.Exec("INSERT INTO readings (metricID, tm, value) VALUES (?, ?, ?)",
|
||
metric.MetricID, localTs, 1232232)
|
||
if err != nil {
|
||
log.Fatalf("Insert error: %v", err)
|
||
}
|
||
|
||
total++
|
||
if total > 841000000 {
|
||
break
|
||
}
|
||
}
|
||
|
||
}
|
||
|
||
err = tx.Commit()
|
||
if err != nil {
|
||
log.Fatalln(err)
|
||
}
|
||
|
||
fmt.Printf("inserted: %d\n", total)
|
||
|
||
if total > 841000000 {
|
||
break
|
||
}
|
||
|
||
ts += 660
|
||
}
|
||
}
|