dima/diploma
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1st

Browse Source
This commit is contained in:
dima
2025-06-21 21:05:15 +00:00
parent 5e49c66e15
commit 84ed171fdf
16 changed files with 1644 additions and 2848 deletions
Download Patch File Download Diff File Expand all files Collapse all files

325
atree/aggregate.go
View File

@@ -1,325 +0,0 @@
package atree
import (
"fmt"
"time"
"gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/timeutil"
)
// AGGREGATE
type InstantAggregator struct {
firstHourOfDay int
lastDayOfMonth int
time2period func(uint32) uint32
currentPeriod uint32
since uint32
until uint32
min float64
max float64
total float64
entries int
}
type InstantAggregatorOptions struct {
GroupBy diploma.GroupBy
FirstHourOfDay int
LastDayOfMonth int
}
func NewInstantAggregator(opt InstantAggregatorOptions) (*InstantAggregator, error) {
s := &InstantAggregator{
firstHourOfDay: opt.FirstHourOfDay,
lastDayOfMonth: opt.LastDayOfMonth,
}
switch opt.GroupBy {
case diploma.GroupByHour:
s.time2period = groupByHour
case diploma.GroupByDay:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByDayUsingFHD
} else {
s.time2period = groupByDay
}
case diploma.GroupByMonth:
if s.firstHourOfDay > 0 {
if s.lastDayOfMonth > 0 {
s.time2period = s.groupByMonthUsingFHDAndLDM
} else {
s.time2period = s.groupByMonthUsingFHD
}
} else {
if s.lastDayOfMonth > 0 {
s.time2period = s.groupByMonthUsingLDM
} else {
s.time2period = groupByMonth
}
}
default:
return nil, fmt.Errorf("unknown groupBy %d option", opt.GroupBy)
}
return s, nil
}
// Приходят данные от свежих к старым, тоесть сперва получаю Until.
// return period complete flag
func (s *InstantAggregator) Feed(timestamp uint32, value float64, p *InstantPeriod) bool {
period := s.time2period(timestamp)
//fmt.Printf("feed: %s %v, period: %s\n", time.Unix(int64(timestamp), 0), value, time.Unix(int64(period), 0))
if s.entries == 0 {
s.currentPeriod = period
s.since = timestamp
s.until = timestamp
s.min = value
s.max = value
s.total = value
s.entries = 1
return false
}
if period != s.currentPeriod {
// готовый период
s.FillPeriod(timestamp, p)
s.currentPeriod = period
s.since = timestamp
s.until = timestamp
s.min = value
s.max = value
s.total = value
s.entries = 1
return true
}
if value < s.min {
s.min = value
} else if value > s.max {
s.max = value
}
// для подсчета AVG
s.total += value
s.entries++
// начало периода
s.since = timestamp
return false
}
func (s *InstantAggregator) FillPeriod(prevTimestamp uint32, p *InstantPeriod) bool {
if s.entries == 0 {
return false
}
//fmt.Printf("FillPeriod: %s, prevTimestamp: %s\n", time.Unix(int64(s.currentPeriod), 0), time.Unix(int64(prevTimestamp), 0))
p.Period = s.currentPeriod
if prevTimestamp > 0 {
p.Since = prevTimestamp
} else {
p.Since = s.since
}
p.Until = s.until
p.Min = s.min
p.Max = s.max
p.Avg = s.total / float64(s.entries)
return true
}
func (s *InstantAggregator) groupByDayUsingFHD(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return uint32(tm.Unix())
}
func (s *InstantAggregator) groupByMonthUsingFHD(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return uint32(tm.Unix())
}
func (s *InstantAggregator) groupByMonthUsingLDM(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Day() > s.lastDayOfMonth {
tm = tm.AddDate(0, 1, 0)
}
return uint32(tm.Unix())
}
func (s *InstantAggregator) groupByMonthUsingFHDAndLDM(timestamp uint32) uint32 {
// ВАЖНО!
// Сперва проверяю время.
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
if tm.Day() > s.lastDayOfMonth {
tm = tm.AddDate(0, 1, 0)
}
return uint32(tm.Unix())
}
// CUMULATIVE
type CumulativeAggregator struct {
firstHourOfDay int
lastDayOfMonth int
time2period func(uint32) uint32
currentPeriod uint32
since uint32
until uint32
sinceValue float64
untilValue float64
entries int
}
type CumulativeAggregatorOptions struct {
GroupBy diploma.GroupBy
FirstHourOfDay int
LastDayOfMonth int
}
func NewCumulativeAggregator(opt CumulativeAggregatorOptions) (*CumulativeAggregator, error) {
s := &CumulativeAggregator{
firstHourOfDay: opt.FirstHourOfDay,
lastDayOfMonth: opt.LastDayOfMonth,
}
switch opt.GroupBy {
case diploma.GroupByHour:
s.time2period = groupByHour
case diploma.GroupByDay:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByDayUsingFHD
} else {
s.time2period = groupByDay
}
case diploma.GroupByMonth:
if s.firstHourOfDay > 0 {
if s.lastDayOfMonth > 0 {
s.time2period = s.groupByMonthUsingFHDAndLDM
} else {
s.time2period = s.groupByMonthUsingFHD
}
} else {
if s.lastDayOfMonth > 0 {
s.time2period = s.groupByMonthUsingLDM
} else {
s.time2period = groupByMonth
}
}
default:
return nil, fmt.Errorf("unknown groupBy %d option", opt.GroupBy)
}
return s, nil
}
// return period complete flag
func (s *CumulativeAggregator) Feed(timestamp uint32, value float64, p *CumulativePeriod) bool {
period := s.time2period(timestamp)
if s.entries == 0 {
s.currentPeriod = period
s.since = timestamp
s.until = timestamp
s.sinceValue = value
s.untilValue = value
s.entries = 1
return false
}
if period != s.currentPeriod {
// готовый период
s.FillPeriod(timestamp, value, p)
s.currentPeriod = period
s.since = timestamp
s.until = timestamp
s.sinceValue = value
s.untilValue = value
s.entries = 1
return true
}
// начало периода
s.since = timestamp
s.sinceValue = value
s.entries++
return false
}
func (s *CumulativeAggregator) FillPeriod(prevTimestamp uint32, value float64, p *CumulativePeriod) bool {
if s.entries == 0 {
return false
}
p.Period = s.currentPeriod
if prevTimestamp > 0 {
p.Since = prevTimestamp
p.Total = s.untilValue - value
} else {
p.Since = s.since
p.Total = s.untilValue - s.sinceValue
}
p.Until = s.until
p.EndValue = s.untilValue
return true
}
func (s *CumulativeAggregator) groupByDayUsingFHD(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return uint32(tm.Unix())
}
func (s *CumulativeAggregator) groupByMonthUsingFHD(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return uint32(tm.Unix())
}
func (s *CumulativeAggregator) groupByMonthUsingLDM(timestamp uint32) uint32 {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Day() > s.lastDayOfMonth {
tm = tm.AddDate(0, 1, 0)
}
return uint32(tm.Unix())
}
func (s *CumulativeAggregator) groupByMonthUsingFHDAndLDM(timestamp uint32) uint32 {
// ВАЖНО!
// Сперва проверяю время.
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
if tm.Day() > s.lastDayOfMonth {
tm = tm.AddDate(0, 1, 0)
}
return uint32(tm.Unix())
}
func groupByHour(timestamp uint32) uint32 {
return uint32(timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "h").Unix())
}
func groupByDay(timestamp uint32) uint32 {
return uint32(timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d").Unix())
}
func groupByMonth(timestamp uint32) uint32 {
return uint32(timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m").Unix())
}

16
atree/io.go
View File

@@ -8,7 +8,7 @@ import (
"math" "math"
"os" "os"
octopus "gordenko.dev/dima/diploma" diploma "gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/atree/redo" "gordenko.dev/dima/diploma/atree/redo"
"gordenko.dev/dima/diploma/bin" "gordenko.dev/dima/diploma/bin"
) )
@@ -74,8 +74,8 @@ func (s *Atree) releaseIndexPage(pageNo uint32) {
p.ReferenceCount-- p.ReferenceCount--
return return
} else { } else {
octopus.Abort( diploma.Abort(
octopus.ReferenceCountBug, diploma.ReferenceCountBug,
fmt.Errorf("call releaseIndexPage on page %d with reference count = %d", fmt.Errorf("call releaseIndexPage on page %d with reference count = %d",
pageNo, p.ReferenceCount), pageNo, p.ReferenceCount),
) )
@@ -98,7 +98,7 @@ func (s *Atree) allocIndexPage() AllocatedPage {
} else { } else {
s.mutex.Lock() s.mutex.Lock()
if s.allocatedIndexPagesQty == math.MaxUint32 { if s.allocatedIndexPagesQty == math.MaxUint32 {
octopus.Abort(octopus.MaxAtreeSizeExceeded, diploma.Abort(diploma.MaxAtreeSizeExceeded,
errors.New("no space in Atree index")) errors.New("no space in Atree index"))
} }
s.allocatedIndexPagesQty++ s.allocatedIndexPagesQty++
@@ -163,8 +163,8 @@ func (s *Atree) releaseDataPage(pageNo uint32) {
p.ReferenceCount-- p.ReferenceCount--
return return
} else { } else {
octopus.Abort( diploma.Abort(
octopus.ReferenceCountBug, diploma.ReferenceCountBug,
fmt.Errorf("call releaseDataPage on page %d with reference count = %d", fmt.Errorf("call releaseDataPage on page %d with reference count = %d",
pageNo, p.ReferenceCount), pageNo, p.ReferenceCount),
) )
@@ -186,7 +186,7 @@ func (s *Atree) allocDataPage() AllocatedPage {
} else { } else {
s.mutex.Lock() s.mutex.Lock()
if s.allocatedDataPagesQty == math.MaxUint32 { if s.allocatedDataPagesQty == math.MaxUint32 {
octopus.Abort(octopus.MaxAtreeSizeExceeded, diploma.Abort(diploma.MaxAtreeSizeExceeded,
errors.New("no space in Atree index")) errors.New("no space in Atree index"))
} }
s.allocatedDataPagesQty++ s.allocatedDataPagesQty++
@@ -303,7 +303,7 @@ func (s *Atree) pageWriter() {
case <-s.writeSignalCh: case <-s.writeSignalCh:
err := s.writeTasks() err := s.writeTasks()
if err != nil { if err != nil {
octopus.Abort(octopus.WriteToAtreeFailed, err) diploma.Abort(diploma.WriteToAtreeFailed, err)
} }
} }
} }

530
atree/select.go
View File

@@ -3,81 +3,16 @@ package atree
import ( import (
"fmt" "fmt"
octopus "gordenko.dev/dima/diploma" "gordenko.dev/dima/diploma"
) )
type IterateAllCumulativeByTreeCursorReq struct { type ContinueFullScanReq struct {
FracDigits byte FracDigits byte
PageNo uint32 ResponseWriter AtreeMeasureConsumer
EndTimestamp uint32
EndValue float64
ResponseWriter *CumulativeMeasureWriter
}
func (s *Atree) IterateAllCumulativeByTreeCursor(req IterateAllCumulativeByTreeCursorReq) error {
buf, err := s.fetchDataPage(req.PageNo)
if err != nil {
return err
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: req.PageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Cumulative,
})
if err != nil {
return err
}
defer treeCursor.Close()
var (
endTimestamp = req.EndTimestamp
endValue = req.EndValue
)
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done {
err := req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue,
})
if err != nil {
return err
}
return nil
}
err = req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue - value,
})
if err != nil {
return err
}
endTimestamp = timestamp
endValue = value
}
}
type ContinueIterateCumulativeByTreeCursorReq struct {
FracDigits byte
Since uint32
Until uint32
LastPageNo uint32 LastPageNo uint32
EndTimestamp uint32
EndValue float64
ResponseWriter *CumulativeMeasureWriter
} }
func (s *Atree) ContinueIterateCumulativeByTreeCursor(req ContinueIterateCumulativeByTreeCursorReq) error { func (s *Atree) ContinueFullScan(req ContinueFullScanReq) error {
buf, err := s.fetchDataPage(req.LastPageNo) buf, err := s.fetchDataPage(req.LastPageNo)
if err != nil { if err != nil {
return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err) return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err)
@@ -88,147 +23,7 @@ func (s *Atree) ContinueIterateCumulativeByTreeCursor(req ContinueIterateCumulat
PageData: buf, PageData: buf,
Atree: s, Atree: s,
FracDigits: req.FracDigits, FracDigits: req.FracDigits,
MetricType: octopus.Cumulative, MetricType: diploma.Instant,
})
if err != nil {
return err
}
defer treeCursor.Close()
var (
endTimestamp = req.EndTimestamp
endValue = req.EndValue
)
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done {
err := req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue,
})
if err != nil {
return err
}
return nil
}
if timestamp <= req.Until {
err := req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue - value,
})
if err != nil {
return err
}
if timestamp < req.Since {
return nil
}
} else {
// bug panic
panic("continue cumulative but timestamp > req.Until")
}
}
}
type FindAndIterateCumulativeByTreeCursorReq struct {
FracDigits byte
Since uint32
Until uint32
RootPageNo uint32
ResponseWriter *CumulativeMeasureWriter
}
func (s *Atree) FindAndIterateCumulativeByTreeCursor(req FindAndIterateCumulativeByTreeCursorReq) error {
pageNo, buf, err := s.findDataPage(req.RootPageNo, req.Until)
if err != nil {
return err
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: pageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Cumulative,
})
if err != nil {
return err
}
defer treeCursor.Close()
var (
endTimestamp uint32
endValue float64
)
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done {
if endTimestamp > 0 {
err := req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue,
})
if err != nil {
return err
}
}
return nil
}
if timestamp > req.Until {
continue
}
if endTimestamp > 0 {
err := req.ResponseWriter.WriteMeasure(CumulativeMeasure{
Timestamp: endTimestamp,
Value: endValue,
Total: endValue - value,
})
if err != nil {
return err
}
}
endTimestamp = timestamp
endValue = value
if timestamp < req.Since {
return nil
}
}
}
type IterateAllInstantByTreeCursorReq struct {
FracDigits byte
PageNo uint32
ResponseWriter *InstantMeasureWriter
}
func (s *Atree) IterateAllInstantByTreeCursor(req IterateAllInstantByTreeCursorReq) error {
buf, err := s.fetchDataPage(req.PageNo)
if err != nil {
return err
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: req.PageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Instant,
}) })
if err != nil { if err != nil {
return err return err
@@ -240,30 +35,21 @@ func (s *Atree) IterateAllInstantByTreeCursor(req IterateAllInstantByTreeCursorR
if err != nil { if err != nil {
return err return err
} }
if done { if done {
return nil return nil
} }
req.ResponseWriter.Feed(timestamp, value)
err = req.ResponseWriter.WriteMeasure(InstantMeasure{
Timestamp: timestamp,
Value: value,
})
if err != nil {
return err
}
} }
} }
type ContinueIterateInstantByTreeCursorReq struct { type ContinueRangeScanReq struct {
FracDigits byte FracDigits byte
Since uint32 ResponseWriter AtreeMeasureConsumer
Until uint32
LastPageNo uint32 LastPageNo uint32
ResponseWriter *InstantMeasureWriter Since uint32
} }
func (s *Atree) ContinueIterateInstantByTreeCursor(req ContinueIterateInstantByTreeCursorReq) error { func (s *Atree) ContinueRangeScan(req ContinueRangeScanReq) error {
buf, err := s.fetchDataPage(req.LastPageNo) buf, err := s.fetchDataPage(req.LastPageNo)
if err != nil { if err != nil {
return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err) return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err)
@@ -274,7 +60,7 @@ func (s *Atree) ContinueIterateInstantByTreeCursor(req ContinueIterateInstantByT
PageData: buf, PageData: buf,
Atree: s, Atree: s,
FracDigits: req.FracDigits, FracDigits: req.FracDigits,
MetricType: octopus.Instant, MetricType: diploma.Instant,
}) })
if err != nil { if err != nil {
return err return err
@@ -286,334 +72,58 @@ func (s *Atree) ContinueIterateInstantByTreeCursor(req ContinueIterateInstantByT
if err != nil { if err != nil {
return err return err
} }
if done {
// - записи закончились;
return nil
}
if timestamp > req.Until {
panic("continue instant timestamp > req.Until")
}
if timestamp < req.Since {
return nil
}
err = req.ResponseWriter.WriteMeasure(InstantMeasure{
Timestamp: timestamp,
Value: value,
})
if err != nil {
return err
}
}
}
type FindAndIterateInstantByTreeCursorReq struct {
FracDigits byte
Since uint32
Until uint32
RootPageNo uint32
ResponseWriter *InstantMeasureWriter
}
func (s *Atree) FindAndIterateInstantByTreeCursor(req FindAndIterateInstantByTreeCursorReq) error {
pageNo, buf, err := s.findDataPage(req.RootPageNo, req.Until)
if err != nil {
return err
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: pageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Instant,
})
if err != nil {
return err
}
defer treeCursor.Close()
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done { if done {
return nil return nil
} }
req.ResponseWriter.Feed(timestamp, value)
if timestamp > req.Until {
continue
}
if timestamp < req.Since { if timestamp < req.Since {
return nil return nil
} }
err = req.ResponseWriter.WriteMeasure(InstantMeasure{
Timestamp: timestamp,
Value: value,
})
if err != nil {
return err
}
} }
} }
type ContinueCollectInstantPeriodsReq struct { type RangeScanReq struct {
FracDigits byte FracDigits byte
Aggregator *InstantAggregator ResponseWriter AtreeMeasureConsumer
ResponseWriter *InstantPeriodsWriter
LastPageNo uint32
Since uint32
Until uint32
}
func (s *Atree) ContinueCollectInstantPeriods(req ContinueCollectInstantPeriodsReq) error {
buf, err := s.fetchDataPage(req.LastPageNo)
if err != nil {
return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err)
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: req.LastPageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Instant,
})
if err != nil {
return err
}
defer treeCursor.Close()
var period InstantPeriod
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done || timestamp < req.Since {
isCompleted := req.Aggregator.FillPeriod(timestamp, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
return nil
}
if timestamp <= req.Until {
isCompleted := req.Aggregator.Feed(timestamp, value, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
}
}
}
type FindInstantPeriodsReq struct {
FracDigits byte
ResponseWriter *InstantPeriodsWriter
RootPageNo uint32 RootPageNo uint32
Since uint32 Since uint32
Until uint32 Until uint32
GroupBy octopus.GroupBy
FirstHourOfDay int
LastDayOfMonth int
} }
func (s *Atree) FindInstantPeriods(req FindInstantPeriodsReq) error { func (s *Atree) RangeScan(req RangeScanReq) error {
pageNo, buf, err := s.findDataPage(req.RootPageNo, req.Until) pageNo, buf, err := s.findDataPage(req.RootPageNo, req.Until)
if err != nil { if err != nil {
return err return err
} }
aggregator, err := NewInstantAggregator(InstantAggregatorOptions{
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
})
if err != nil {
return err
}
cursor, err := NewBackwardCursor(BackwardCursorOptions{ cursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: pageNo, PageNo: pageNo,
PageData: buf, PageData: buf,
Atree: s, Atree: s,
FracDigits: req.FracDigits, FracDigits: req.FracDigits,
MetricType: octopus.Instant, MetricType: diploma.Instant,
}) })
if err != nil { if err != nil {
return err return err
} }
defer cursor.Close() defer cursor.Close()
var period InstantPeriod
for { for {
timestamp, value, done, err := cursor.Prev() timestamp, value, done, err := cursor.Prev()
if err != nil { if err != nil {
return err return err
} }
if done {
if done || timestamp < req.Since {
isCompleted := aggregator.FillPeriod(timestamp, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
return nil return nil
} }
if timestamp <= req.Until { if timestamp <= req.Until {
isCompleted := aggregator.Feed(timestamp, value, &period) req.ResponseWriter.Feed(timestamp, value)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
}
}
}
type FindCumulativePeriodsReq struct { if timestamp < req.Since {
FracDigits byte // - записи, удовлетворяющие временным рамкам, закончились.
ResponseWriter *CumulativePeriodsWriter
RootPageNo uint32
Since uint32
Until uint32
GroupBy octopus.GroupBy
FirstHourOfDay int
LastDayOfMonth int
}
func (s *Atree) FindCumulativePeriods(req FindCumulativePeriodsReq) error {
pageNo, buf, err := s.findDataPage(req.RootPageNo, req.Until)
if err != nil {
return err
}
aggregator, err := NewCumulativeAggregator(CumulativeAggregatorOptions{
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
})
if err != nil {
return err
}
cursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: pageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Cumulative,
})
if err != nil {
return err
}
defer cursor.Close()
var period CumulativePeriod
for {
timestamp, value, done, err := cursor.Prev()
if err != nil {
return err
}
if done || timestamp < req.Since {
isCompleted := aggregator.FillPeriod(timestamp, value, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
return nil return nil
} }
if timestamp <= req.Until {
isCompleted := aggregator.Feed(timestamp, value, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
}
}
}
type ContinueCollectCumulativePeriodsReq struct {
FracDigits byte
Aggregator *CumulativeAggregator
ResponseWriter *CumulativePeriodsWriter
LastPageNo uint32
Since uint32
Until uint32
}
func (s *Atree) ContinueCollectCumulativePeriods(req ContinueCollectCumulativePeriodsReq) error {
buf, err := s.fetchDataPage(req.LastPageNo)
if err != nil {
return fmt.Errorf("fetchDataPage(%d): %s", req.LastPageNo, err)
}
treeCursor, err := NewBackwardCursor(BackwardCursorOptions{
PageNo: req.LastPageNo,
PageData: buf,
Atree: s,
FracDigits: req.FracDigits,
MetricType: octopus.Cumulative,
})
if err != nil {
return err
}
defer treeCursor.Close()
var period CumulativePeriod
for {
timestamp, value, done, err := treeCursor.Prev()
if err != nil {
return err
}
if done || timestamp < req.Since {
isCompleted := req.Aggregator.FillPeriod(timestamp, value, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
return nil
}
if timestamp <= req.Until {
isCompleted := req.Aggregator.Feed(timestamp, value, &period)
if isCompleted {
err := req.ResponseWriter.WritePeriod(period)
if err != nil {
return err
}
}
} }
} }
} }

307
atree/writers.go
View File

@@ -1,306 +1,15 @@
package atree package atree
import ( type PeriodsWriter interface {
"bytes" Feed(uint32, float64)
"fmt" FeedNoSend(uint32, float64)
"io" Close() error
octopus "gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/bin"
"gordenko.dev/dima/diploma/proto"
)
// CURRENT VALUE WRITER
type CurrentValue struct {
MetricID uint32
Timestamp uint32
Value float64
} }
type CurrentValueWriter struct { type WorkerMeasureConsumer interface {
arr []byte FeedNoSend(uint32, float64)
responder *ChunkedResponder
} }
func NewCurrentValueWriter(dst io.Writer) *CurrentValueWriter { type AtreeMeasureConsumer interface {
return &CurrentValueWriter{ Feed(uint32, float64)
arr: make([]byte, 16),
responder: NewChunkedResponder(dst),
}
}
func (s *CurrentValueWriter) BufferValue(m CurrentValue) {
bin.PutUint32(s.arr[0:], m.MetricID)
bin.PutUint32(s.arr[4:], m.Timestamp)
bin.PutFloat64(s.arr[8:], m.Value)
s.responder.BufferRecord(s.arr)
}
func (s *CurrentValueWriter) Close() error {
return s.responder.Flush()
}
// INSTANT MEASURE WRITER
type InstantMeasure struct {
Timestamp uint32
Value float64
}
type InstantMeasureWriter struct {
arr []byte
responder *ChunkedResponder
}
func NewInstantMeasureWriter(dst io.Writer) *InstantMeasureWriter {
return &InstantMeasureWriter{
arr: make([]byte, 12),
responder: NewChunkedResponder(dst),
}
}
func (s *InstantMeasureWriter) BufferMeasure(m InstantMeasure) {
bin.PutUint32(s.arr[0:], m.Timestamp)
bin.PutFloat64(s.arr[4:], m.Value)
s.responder.BufferRecord(s.arr)
}
func (s *InstantMeasureWriter) WriteMeasure(m InstantMeasure) error {
bin.PutUint32(s.arr[0:], m.Timestamp)
bin.PutFloat64(s.arr[4:], m.Value)
return s.responder.AppendRecord(s.arr)
}
func (s *InstantMeasureWriter) Close() error {
return s.responder.Flush()
}
// CUMULATIVE MEASURE WRITER
type CumulativeMeasure struct {
Timestamp uint32
Value float64
Total float64
}
type CumulativeMeasureWriter struct {
arr []byte
responder *ChunkedResponder
}
func NewCumulativeMeasureWriter(dst io.Writer) *CumulativeMeasureWriter {
return &CumulativeMeasureWriter{
arr: make([]byte, 20),
responder: NewChunkedResponder(dst),
}
}
func (s *CumulativeMeasureWriter) BufferMeasure(m CumulativeMeasure) {
bin.PutUint32(s.arr[0:], m.Timestamp)
bin.PutFloat64(s.arr[4:], m.Value)
bin.PutFloat64(s.arr[12:], m.Total)
s.responder.BufferRecord(s.arr)
}
func (s *CumulativeMeasureWriter) WriteMeasure(m CumulativeMeasure) error {
bin.PutUint32(s.arr[0:], m.Timestamp)
bin.PutFloat64(s.arr[4:], m.Value)
bin.PutFloat64(s.arr[12:], m.Total)
return s.responder.AppendRecord(s.arr)
}
func (s *CumulativeMeasureWriter) Close() error {
return s.responder.Flush()
}
// INSTANT AGGREGATE WRITER
type InstantPeriodsWriter struct {
aggregateFuncs byte
arr []byte
responder *ChunkedResponder
}
func NewInstantPeriodsWriter(dst io.Writer, aggregateFuncs byte) *InstantPeriodsWriter {
var q int
if (aggregateFuncs & octopus.AggregateMin) == octopus.AggregateMin {
q++
}
if (aggregateFuncs & octopus.AggregateMax) == octopus.AggregateMax {
q++
}
if (aggregateFuncs & octopus.AggregateAvg) == octopus.AggregateAvg {
q++
}
return &InstantPeriodsWriter{
aggregateFuncs: aggregateFuncs,
arr: make([]byte, 12+q*8),
responder: NewChunkedResponder(dst),
}
}
type InstantPeriod struct {
Period uint32
Since uint32
Until uint32
Min float64
Max float64
Avg float64
}
func (s *InstantPeriodsWriter) BufferMeasure(p InstantPeriod) {
s.pack(p)
s.responder.BufferRecord(s.arr)
}
func (s *InstantPeriodsWriter) WritePeriod(p InstantPeriod) error {
s.pack(p)
return s.responder.AppendRecord(s.arr)
}
func (s *InstantPeriodsWriter) Close() error {
return s.responder.Flush()
}
func (s *InstantPeriodsWriter) pack(p InstantPeriod) {
bin.PutUint32(s.arr[0:], p.Period)
bin.PutUint32(s.arr[4:], p.Since)
bin.PutUint32(s.arr[8:], p.Until)
pos := 12
if (s.aggregateFuncs & octopus.AggregateMin) == octopus.AggregateMin {
bin.PutFloat64(s.arr[pos:], p.Min)
pos += 8
}
if (s.aggregateFuncs & octopus.AggregateMax) == octopus.AggregateMax {
bin.PutFloat64(s.arr[pos:], p.Max)
pos += 8
}
if (s.aggregateFuncs & octopus.AggregateAvg) == octopus.AggregateAvg {
bin.PutFloat64(s.arr[pos:], p.Avg)
}
}
// CUMULATIVE AGGREGATE WRITER
type CumulativePeriodsWriter struct {
arr []byte
responder *ChunkedResponder
}
func NewCumulativePeriodsWriter(dst io.Writer) *CumulativePeriodsWriter {
return &CumulativePeriodsWriter{
arr: make([]byte, 28),
responder: NewChunkedResponder(dst),
}
}
type CumulativePeriod struct {
Period uint32
Since uint32
Until uint32
EndValue float64
Total float64
}
func (s *CumulativePeriodsWriter) BufferMeasure(p CumulativePeriod) {
s.pack(p)
s.responder.BufferRecord(s.arr)
}
func (s *CumulativePeriodsWriter) WritePeriod(p CumulativePeriod) error {
s.pack(p)
return s.responder.AppendRecord(s.arr)
}
func (s *CumulativePeriodsWriter) Close() error {
return s.responder.Flush()
}
func (s *CumulativePeriodsWriter) pack(p CumulativePeriod) {
bin.PutUint32(s.arr[0:], p.Period)
bin.PutUint32(s.arr[4:], p.Since)
bin.PutUint32(s.arr[8:], p.Until)
bin.PutFloat64(s.arr[12:], p.EndValue)
bin.PutFloat64(s.arr[20:], p.Total)
}
// CHUNKED RESPONDER
//const headerSize = 3
var endMsg = []byte{
proto.RespEndOfValue, // end of stream
}
type ChunkedResponder struct {
recordsQty int
buf *bytes.Buffer
dst io.Writer
}
func NewChunkedResponder(dst io.Writer) *ChunkedResponder {
s := &ChunkedResponder{
recordsQty: 0,
buf: bytes.NewBuffer(nil),
dst: dst,
}
s.buf.Write([]byte{
proto.RespPartOfValue, // message type
0, 0, 0, 0, // records qty
})
return s
}
func (s *ChunkedResponder) BufferRecord(rec []byte) {
s.buf.Write(rec)
s.recordsQty++
}
func (s *ChunkedResponder) AppendRecord(rec []byte) error {
s.buf.Write(rec)
s.recordsQty++
if s.buf.Len() < 1500 {
return nil
}
if err := s.sendBuffered(); err != nil {
return err
}
s.buf.Write([]byte{
proto.RespPartOfValue, // message type
0, 0, 0, 0, // records qty
})
s.recordsQty = 0
return nil
}
func (s *ChunkedResponder) Flush() error {
if s.recordsQty > 0 {
if err := s.sendBuffered(); err != nil {
return err
}
}
if _, err := s.dst.Write(endMsg); err != nil {
return err
}
return nil
}
func (s *ChunkedResponder) sendBuffered() (err error) {
msg := s.buf.Bytes()
bin.PutUint32(msg[1:], uint32(s.recordsQty))
n, err := s.dst.Write(msg)
if err != nil {
return
}
if n != len(msg) {
return fmt.Errorf("incomplete write %d bytes instead of %d", n, len(msg))
}
s.buf.Reset()
return
} }

270
client/client.go
View File

@@ -66,13 +66,7 @@ func (s *Connection) mustSuccess(reader *bufreader.BufferedReader) (err error) {
} }
} }
type Metric struct { func (s *Connection) AddMetric(req proto.AddMetricReq) error {
MetricID uint32
MetricType diploma.MetricType
FracDigits byte
}
func (s *Connection) AddMetric(req Metric) error {
arr := []byte{ arr := []byte{
proto.TypeAddMetric, proto.TypeAddMetric,
0, 0, 0, 0, // 0, 0, 0, 0, //
@@ -87,7 +81,22 @@ func (s *Connection) AddMetric(req Metric) error {
return s.mustSuccess(s.src) return s.mustSuccess(s.src)
} }
func (s *Connection) GetMetric(metricID uint32) (*Metric, error) { //
// func (s *Connection) UpdateMetric(req Metric) error {
// arr := []byte{
// proto.TypeUpdateMetric,
// 0, 0, 0, 0, //
// byte(req.FracDigits),
// }
// bin.PutUint32(arr[1:], req.MetricID)
// if _, err := s.conn.Write(arr); err != nil {
// return err
// }
// return s.mustSuccess(s.src)
// }
func (s *Connection) GetMetric(metricID uint32) (*proto.Metric, error) {
arr := []byte{ arr := []byte{
proto.TypeGetMetric, proto.TypeGetMetric,
0, 0, 0, 0, 0, 0, 0, 0,
@@ -110,10 +119,10 @@ func (s *Connection) GetMetric(metricID uint32) (*Metric, error) {
return nil, fmt.Errorf("read body: %s", err) return nil, fmt.Errorf("read body: %s", err)
} }
return &Metric{ return &proto.Metric{
MetricID: bin.GetUint32(arr), MetricID: bin.GetUint32(arr),
MetricType: diploma.MetricType(arr[4]), MetricType: diploma.MetricType(arr[4]),
FracDigits: arr[5], FracDigits: int(arr[5]),
}, nil }, nil
case proto.RespError: case proto.RespError:
@@ -137,13 +146,7 @@ func (s *Connection) DeleteMetric(metricID uint32) error {
return s.mustSuccess(s.src) return s.mustSuccess(s.src)
} }
type AppendMeasureReq struct { func (s *Connection) AppendMeasure(req proto.AppendMeasureReq) (err error) {
MetricID uint32
Timestamp uint32
Value float64
}
func (s *Connection) AppendMeasure(req AppendMeasureReq) (err error) {
arr := []byte{ arr := []byte{
proto.TypeAppendMeasure, proto.TypeAppendMeasure,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
@@ -160,17 +163,7 @@ func (s *Connection) AppendMeasure(req AppendMeasureReq) (err error) {
return s.mustSuccess(s.src) return s.mustSuccess(s.src)
} }
type AppendMeasuresReq struct { func (s *Connection) AppendMeasures(req proto.AppendMeasuresReq) (err error) {
MetricID uint32
Measures []Measure
}
type Measure struct {
Timestamp uint32
Value float64
}
func (s *Connection) AppendMeasures(req AppendMeasuresReq) (err error) {
if len(req.Measures) > 65535 { if len(req.Measures) > 65535 {
return fmt.Errorf("wrong measures qty: %d", len(req.Measures)) return fmt.Errorf("wrong measures qty: %d", len(req.Measures))
} }
@@ -191,15 +184,74 @@ func (s *Connection) AppendMeasures(req AppendMeasuresReq) (err error) {
if _, err := s.conn.Write(arr); err != nil { if _, err := s.conn.Write(arr); err != nil {
return err return err
} }
//fmt.Printf("encode measures: %d\n", len(req.Measures))
return s.mustSuccess(s.src) return s.mustSuccess(s.src)
} }
type InstantMeasure struct { // type AppendMeasurePerMetricReq struct {
Timestamp uint32 // MetricID uint32
Value float64 // Measures []Measure
// }
func (s *Connection) AppendMeasurePerMetric(list []proto.MetricMeasure) (_ []proto.AppendError, err error) {
if len(list) > 65535 {
return nil, fmt.Errorf("wrong measures qty: %d", len(list))
}
var (
// 3 bytes: 1b message type + 2b records qty
fixedSize = 3
recordSize = 16
arr = make([]byte, fixedSize+len(list)*recordSize)
)
arr[0] = proto.TypeAppendMeasures
bin.PutUint16(arr[1:], uint16(len(list)))
pos := fixedSize
for _, item := range list {
bin.PutUint32(arr[pos:], item.MetricID)
bin.PutUint32(arr[pos+4:], item.Timestamp)
bin.PutFloat64(arr[pos+8:], item.Value)
pos += recordSize
}
if _, err := s.conn.Write(arr); err != nil {
return nil, err
} }
func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]InstantMeasure, error) { code, err := s.src.ReadByte()
if err != nil {
return nil, fmt.Errorf("read response code: %s", err)
}
switch code {
case proto.RespValue:
var (
qty uint16
appendErrors []proto.AppendError
)
qty, err = bin.ReadUint16(s.src)
if err != nil {
return
}
for range qty {
var ae proto.AppendError
ae.MetricID, err = bin.ReadUint32(s.src)
if err != nil {
return
}
ae.ErrorCode, err = bin.ReadUint16(s.src)
if err != nil {
return
}
appendErrors = append(appendErrors, ae)
}
return appendErrors, nil
default:
return nil, fmt.Errorf("unknown reponse code %d", code)
}
}
func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]proto.InstantMeasure, error) {
arr := []byte{ arr := []byte{
proto.TypeListAllInstantMeasures, proto.TypeListAllInstantMeasures,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
@@ -211,7 +263,7 @@ func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]InstantMeasure,
} }
var ( var (
result []InstantMeasure result []proto.InstantMeasure
tmp = make([]byte, 12) tmp = make([]byte, 12)
) )
@@ -221,6 +273,8 @@ func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]InstantMeasure,
return nil, fmt.Errorf("read response code: %s", err) return nil, fmt.Errorf("read response code: %s", err)
} }
//fmt.Printf("code: %d\n", code)
switch code { switch code {
case proto.RespPartOfValue: case proto.RespPartOfValue:
q, err := bin.ReadUint32(s.src) q, err := bin.ReadUint32(s.src)
@@ -228,13 +282,15 @@ func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]InstantMeasure,
return nil, fmt.Errorf("read records qty: %s", err) return nil, fmt.Errorf("read records qty: %s", err)
} }
//fmt.Printf("q: %d\n", q)
for i := range int(q) { for i := range int(q) {
err = bin.ReadNInto(s.src, tmp) err = bin.ReadNInto(s.src, tmp)
if err != nil { if err != nil {
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, InstantMeasure{ result = append(result, proto.InstantMeasure{
Timestamp: bin.GetUint32(tmp), Timestamp: bin.GetUint32(tmp),
Value: bin.GetFloat64(tmp[4:]), Value: bin.GetFloat64(tmp[4:]),
}) })
@@ -252,13 +308,12 @@ func (s *Connection) ListAllInstantMeasures(metricID uint32) ([]InstantMeasure,
} }
} }
func (s *Connection) ListInstantMeasures(req proto.ListInstantMeasuresReq) ([]InstantMeasure, error) { func (s *Connection) ListInstantMeasures(req proto.ListInstantMeasuresReq) ([]proto.InstantMeasure, error) {
arr := []byte{ arr := []byte{
proto.TypeListInstantMeasures, proto.TypeListInstantMeasures,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
0, 0, 0, 0, // since 0, 0, 0, 0, // since
0, 0, 0, 0, // until 0, 0, 0, 0, // until
byte(req.FirstHourOfDay),
} }
bin.PutUint32(arr[1:], req.MetricID) bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], req.Since) bin.PutUint32(arr[5:], req.Since)
@@ -269,7 +324,7 @@ func (s *Connection) ListInstantMeasures(req proto.ListInstantMeasuresReq) ([]In
} }
var ( var (
result []InstantMeasure result []proto.InstantMeasure
tmp = make([]byte, 12) tmp = make([]byte, 12)
) )
@@ -292,7 +347,7 @@ func (s *Connection) ListInstantMeasures(req proto.ListInstantMeasuresReq) ([]In
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, InstantMeasure{ result = append(result, proto.InstantMeasure{
Timestamp: bin.GetUint32(tmp), Timestamp: bin.GetUint32(tmp),
Value: bin.GetFloat64(tmp[4:]), Value: bin.GetFloat64(tmp[4:]),
}) })
@@ -310,13 +365,7 @@ func (s *Connection) ListInstantMeasures(req proto.ListInstantMeasuresReq) ([]In
} }
} }
type CumulativeMeasure struct { func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]proto.CumulativeMeasure, error) {
Timestamp uint32
Value float64
Total float64
}
func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]CumulativeMeasure, error) {
arr := []byte{ arr := []byte{
proto.TypeListAllCumulativeMeasures, proto.TypeListAllCumulativeMeasures,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
@@ -328,7 +377,7 @@ func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]CumulativeMea
} }
var ( var (
result []CumulativeMeasure result []proto.CumulativeMeasure
tmp = make([]byte, 20) tmp = make([]byte, 20)
) )
@@ -338,6 +387,8 @@ func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]CumulativeMea
return nil, fmt.Errorf("read response code: %s", err) return nil, fmt.Errorf("read response code: %s", err)
} }
//fmt.Printf("code: %d\n", code)
switch code { switch code {
case proto.RespPartOfValue: case proto.RespPartOfValue:
q, err := bin.ReadUint32(s.src) q, err := bin.ReadUint32(s.src)
@@ -351,11 +402,19 @@ func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]CumulativeMea
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, CumulativeMeasure{ //fmt.Printf("tmp: %d\n", tmp)
result = append(result, proto.CumulativeMeasure{
Timestamp: bin.GetUint32(tmp), Timestamp: bin.GetUint32(tmp),
Value: bin.GetFloat64(tmp[4:]), Value: bin.GetFloat64(tmp[4:]),
Total: bin.GetFloat64(tmp[12:]), Total: bin.GetFloat64(tmp[12:]),
}) })
// pretty.PPrintln("measure", CumulativeMeasure{
// Timestamp: bin.GetUint32(tmp),
// Value: bin.GetFloat64(tmp[4:]),
// Total: bin.GetFloat64(tmp[12:]),
// })
} }
case proto.RespEndOfValue: case proto.RespEndOfValue:
@@ -370,13 +429,12 @@ func (s *Connection) ListAllCumulativeMeasures(metricID uint32) ([]CumulativeMea
} }
} }
func (s *Connection) ListCumulativeMeasures(req proto.ListCumulativeMeasuresReq) ([]CumulativeMeasure, error) { func (s *Connection) ListCumulativeMeasures(req proto.ListCumulativeMeasuresReq) ([]proto.CumulativeMeasure, error) {
arr := []byte{ arr := []byte{
proto.TypeListCumulativeMeasures, proto.TypeListCumulativeMeasures,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
0, 0, 0, 0, // since 0, 0, 0, 0, // since
0, 0, 0, 0, // until 0, 0, 0, 0, // until
byte(req.FirstHourOfDay),
} }
bin.PutUint32(arr[1:], req.MetricID) bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], req.Since) bin.PutUint32(arr[5:], req.Since)
@@ -387,7 +445,7 @@ func (s *Connection) ListCumulativeMeasures(req proto.ListCumulativeMeasuresReq)
} }
var ( var (
result []CumulativeMeasure result []proto.CumulativeMeasure
tmp = make([]byte, 20) tmp = make([]byte, 20)
) )
@@ -410,7 +468,7 @@ func (s *Connection) ListCumulativeMeasures(req proto.ListCumulativeMeasuresReq)
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, CumulativeMeasure{ result = append(result, proto.CumulativeMeasure{
Timestamp: bin.GetUint32(tmp), Timestamp: bin.GetUint32(tmp),
Value: bin.GetFloat64(tmp[4:]), Value: bin.GetFloat64(tmp[4:]),
Total: bin.GetFloat64(tmp[12:]), Total: bin.GetFloat64(tmp[12:]),
@@ -429,16 +487,7 @@ func (s *Connection) ListCumulativeMeasures(req proto.ListCumulativeMeasuresReq)
} }
} }
type InstantPeriod struct { func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]proto.InstantPeriod, error) {
Period uint32
Since uint32
Until uint32
Min float64
Max float64
Avg float64
}
func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]InstantPeriod, error) {
arr := []byte{ arr := []byte{
proto.TypeListInstantPeriods, proto.TypeListInstantPeriods,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
@@ -447,11 +496,14 @@ func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]Inst
byte(req.GroupBy), byte(req.GroupBy),
req.AggregateFuncs, req.AggregateFuncs,
byte(req.FirstHourOfDay), byte(req.FirstHourOfDay),
byte(req.LastDayOfMonth),
} }
bin.PutUint32(arr[1:], req.MetricID) bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], req.Since) bin.PutUint16(arr[5:], uint16(req.Since.Year))
bin.PutUint32(arr[9:], req.Until) arr[7] = byte(req.Since.Month)
arr[8] = byte(req.Since.Day)
bin.PutUint16(arr[9:], uint16(req.Until.Year))
arr[11] = byte(req.Until.Month)
arr[12] = byte(req.Until.Day)
if _, err := s.conn.Write(arr); err != nil { if _, err := s.conn.Write(arr); err != nil {
return nil, err return nil, err
@@ -469,7 +521,7 @@ func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]Inst
} }
var ( var (
result []InstantPeriod result []proto.InstantPeriod
// 12 bytes - period, since, until // 12 bytes - period, since, until
// q * 8 bytes - min, max, avg // q * 8 bytes - min, max, avg
tmp = make([]byte, 12+q*8) tmp = make([]byte, 12+q*8)
@@ -481,6 +533,8 @@ func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]Inst
return nil, fmt.Errorf("read response code: %s", err) return nil, fmt.Errorf("read response code: %s", err)
} }
//fmt.Printf("code: %d\n", code)
switch code { switch code {
case proto.RespPartOfValue: case proto.RespPartOfValue:
q, err := bin.ReadUint32(s.src) q, err := bin.ReadUint32(s.src)
@@ -495,7 +549,7 @@ func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]Inst
} }
var ( var (
p = InstantPeriod{ p = proto.InstantPeriod{
Period: bin.GetUint32(tmp[0:]), Period: bin.GetUint32(tmp[0:]),
Since: bin.GetUint32(tmp[4:]), Since: bin.GetUint32(tmp[4:]),
Until: bin.GetUint32(tmp[8:]), Until: bin.GetUint32(tmp[8:]),
@@ -530,15 +584,7 @@ func (s *Connection) ListInstantPeriods(req proto.ListInstantPeriodsReq) ([]Inst
} }
} }
type CumulativePeriod struct { func (s *Connection) ListCumulativePeriods(req proto.ListCumulativePeriodsReq) ([]proto.CumulativePeriod, error) {
Period uint32
Since uint32
Until uint32
EndValue float64
Total float64
}
func (s *Connection) ListCumulativePeriods(req proto.ListCumulativePeriodsReq) ([]CumulativePeriod, error) {
arr := []byte{ arr := []byte{
proto.TypeListCumulativePeriods, proto.TypeListCumulativePeriods,
0, 0, 0, 0, // metricID 0, 0, 0, 0, // metricID
@@ -546,18 +592,21 @@ func (s *Connection) ListCumulativePeriods(req proto.ListCumulativePeriodsReq) (
0, 0, 0, 0, // until 0, 0, 0, 0, // until
byte(req.GroupBy), byte(req.GroupBy),
byte(req.FirstHourOfDay), byte(req.FirstHourOfDay),
byte(req.LastDayOfMonth),
} }
bin.PutUint32(arr[1:], req.MetricID) bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], req.Since) bin.PutUint16(arr[5:], uint16(req.Since.Year))
bin.PutUint32(arr[9:], req.Until) arr[7] = byte(req.Since.Month)
arr[8] = byte(req.Since.Day)
bin.PutUint16(arr[9:], uint16(req.Until.Year))
arr[11] = byte(req.Until.Month)
arr[12] = byte(req.Until.Day)
if _, err := s.conn.Write(arr); err != nil { if _, err := s.conn.Write(arr); err != nil {
return nil, err return nil, err
} }
var ( var (
result []CumulativePeriod result []proto.CumulativePeriod
tmp = make([]byte, 28) tmp = make([]byte, 28)
) )
@@ -579,7 +628,7 @@ func (s *Connection) ListCumulativePeriods(req proto.ListCumulativePeriodsReq) (
if err != nil { if err != nil {
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, CumulativePeriod{ result = append(result, proto.CumulativePeriod{
Period: bin.GetUint32(tmp[0:]), Period: bin.GetUint32(tmp[0:]),
Since: bin.GetUint32(tmp[4:]), Since: bin.GetUint32(tmp[4:]),
Until: bin.GetUint32(tmp[8:]), Until: bin.GetUint32(tmp[8:]),
@@ -600,13 +649,7 @@ func (s *Connection) ListCumulativePeriods(req proto.ListCumulativePeriodsReq) (
} }
} }
type CurrentValue struct { func (s *Connection) ListCurrentValues(metricIDs []uint32) ([]proto.CurrentValue, error) {
MetricID uint32
Timestamp uint32
Value float64
}
func (s *Connection) ListCurrentValues(metricIDs []uint32) ([]CurrentValue, error) {
arr := make([]byte, 3+metricKeySize*len(metricIDs)) arr := make([]byte, 3+metricKeySize*len(metricIDs))
arr[0] = proto.TypeListCurrentValues arr[0] = proto.TypeListCurrentValues
@@ -623,7 +666,7 @@ func (s *Connection) ListCurrentValues(metricIDs []uint32) ([]CurrentValue, erro
} }
var ( var (
result []CurrentValue result []proto.CurrentValue
tmp = make([]byte, 16) tmp = make([]byte, 16)
) )
@@ -646,7 +689,7 @@ func (s *Connection) ListCurrentValues(metricIDs []uint32) ([]CurrentValue, erro
return nil, fmt.Errorf("read record #%d: %s", i, err) return nil, fmt.Errorf("read record #%d: %s", i, err)
} }
result = append(result, CurrentValue{ result = append(result, proto.CurrentValue{
MetricID: bin.GetUint32(tmp), MetricID: bin.GetUint32(tmp),
Timestamp: bin.GetUint32(tmp[4:]), Timestamp: bin.GetUint32(tmp[4:]),
Value: bin.GetFloat64(tmp[8:]), Value: bin.GetFloat64(tmp[8:]),
@@ -680,55 +723,6 @@ func (s *Connection) DeleteMeasures(req proto.DeleteMeasuresReq) (err error) {
return s.mustSuccess(s.src) return s.mustSuccess(s.src)
} }
type RangeTotalResp struct {
Since uint32
SinceValue float64
Until uint32
UntilValue float64
}
func (s *Connection) RangeTotal(req proto.RangeTotalReq) (*RangeTotalResp, error) {
arr := []byte{
proto.TypeGetMetric,
0, 0, 0, 0,
0, 0, 0, 0,
0, 0, 0, 0,
}
bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], req.Since)
bin.PutUint32(arr[9:], req.MetricID)
if _, err := s.conn.Write(arr); err != nil {
return nil, err
}
code, err := s.src.ReadByte()
if err != nil {
return nil, fmt.Errorf("read response code: %s", err)
}
switch code {
case proto.RespValue:
arr, err := s.src.ReadN(24)
if err != nil {
return nil, fmt.Errorf("read body: %s", err)
}
return &RangeTotalResp{
Since: bin.GetUint32(arr),
SinceValue: bin.GetFloat64(arr[4:]),
Until: bin.GetUint32(arr[12:]),
UntilValue: bin.GetFloat64(arr[16:]),
}, nil
case proto.RespError:
return nil, s.onError()
default:
return nil, fmt.Errorf("unknown reponse code %d", code)
}
}
func (s *Connection) onError() error { func (s *Connection) onError() error {
errorCode, err := bin.ReadUint16(s.src) errorCode, err := bin.ReadUint16(s.src)
if err != nil { if err != nil {

453
database/api.go
View File

@@ -13,6 +13,7 @@ import (
"gordenko.dev/dima/diploma/chunkenc" "gordenko.dev/dima/diploma/chunkenc"
"gordenko.dev/dima/diploma/conbuf" "gordenko.dev/dima/diploma/conbuf"
"gordenko.dev/dima/diploma/proto" "gordenko.dev/dima/diploma/proto"
"gordenko.dev/dima/diploma/transform"
"gordenko.dev/dima/diploma/txlog" "gordenko.dev/dima/diploma/txlog"
) )
@@ -326,6 +327,9 @@ type FilledPage struct {
} }
type tryAppendMeasureResult struct { type tryAppendMeasureResult struct {
MetricID uint32
Timestamp uint32
Value float64
FilledPage *FilledPage FilledPage *FilledPage
ResultCode byte ResultCode byte
} }
@@ -441,7 +445,6 @@ func (s *Database) AppendMeasures(req proto.AppendMeasuresReq) uint16 {
) )
for idx, measure := range req.Measures { for idx, measure := range req.Measures {
//fmt.Printf("%d %v\n", measure.Timestamp, measure.Value)
if since == 0 { if since == 0 {
since = measure.Timestamp since = measure.Timestamp
} else { } else {
@@ -470,7 +473,6 @@ func (s *Database) AppendMeasures(req proto.AppendMeasuresReq) uint16 {
) )
<-waitCh <-waitCh
} }
//fmt.Printf("m.Value: %v < untilValue: %v\n", measure.Value, untilValue)
return proto.ErrNonMonotonicValue return proto.ErrNonMonotonicValue
} }
} }
@@ -593,23 +595,27 @@ func (s *Database) DeleteMeasures(req proto.DeleteMeasuresReq) uint16 {
result := <-resultCh result := <-resultCh
switch result.ResultCode { switch result.ResultCode {
case Succeed: case NoMeasuresToDelete:
var ( // ok
freeDataPages []uint32
freeIndexPages []uint32 case DeleteFromAtreeNotNeeded:
) // регистрирую удаление в TransactionLog
if result.RootPageNo > 0 { waitCh := s.txlog.WriteDeletedMeasures(txlog.DeletedMeasures{
pageLists, err := s.atree.GetAllPages(result.RootPageNo) MetricID: req.MetricID,
})
<-waitCh
case DeleteFromAtreeRequired:
// собираю номера всех data и index страниц метрики (типа запись REDO лога).
pageLists, err := s.atree.GetAllPages(req.MetricID)
if err != nil { if err != nil {
diploma.Abort(diploma.FailedAtreeRequest, err) diploma.Abort(diploma.FailedAtreeRequest, err)
} }
freeDataPages = pageLists.DataPages // регистрирую удаление в TransactionLog
freeIndexPages = pageLists.IndexPages
}
waitCh := s.txlog.WriteDeletedMeasures(txlog.DeletedMeasures{ waitCh := s.txlog.WriteDeletedMeasures(txlog.DeletedMeasures{
MetricID: req.MetricID, MetricID: req.MetricID,
FreeDataPages: freeDataPages, FreeDataPages: pageLists.DataPages,
FreeIndexPages: freeIndexPages, FreeIndexPages: pageLists.IndexPages,
}) })
<-waitCh <-waitCh
@@ -624,98 +630,32 @@ func (s *Database) DeleteMeasures(req proto.DeleteMeasuresReq) uint16 {
// SELECT // SELECT
type instantMeasuresResult struct { type fullScanResult struct {
ResultCode byte ResultCode byte
FracDigits byte FracDigits byte
PageNo uint32 LastPageNo uint32
} }
func (s *Database) ListAllInstantMeasures(conn net.Conn, req proto.ListAllInstantMetricMeasuresReq) error { func (s *Database) ListAllInstantMeasures(conn net.Conn, req proto.ListAllInstantMetricMeasuresReq) error {
resultCh := make(chan instantMeasuresResult, 1) responseWriter := transform.NewInstantMeasureWriter(conn, 0)
responseWriter := atree.NewInstantMeasureWriter(conn) return s.fullScan(fullScanReq{
s.appendJobToWorkerQueue(tryListAllInstantMeasuresReq{
MetricID: req.MetricID, MetricID: req.MetricID,
ResponseWriter: responseWriter, MetricType: diploma.Instant,
ResultCh: resultCh, Conn: conn,
})
result := <-resultCh
switch result.ResultCode {
case QueryDone:
responseWriter.Close()
case UntilFound:
err := s.atree.IterateAllInstantByTreeCursor(atree.IterateAllInstantByTreeCursorReq{
FracDigits: result.FracDigits,
PageNo: result.PageNo,
ResponseWriter: responseWriter, ResponseWriter: responseWriter,
}) })
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case NoMetric:
reply(conn, proto.ErrNoMetric)
case WrongMetricType:
reply(conn, proto.ErrWrongMetricType)
default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
}
return nil
} }
func (s *Database) ListAllCumulativeMeasures(conn io.Writer, req proto.ListAllCumulativeMeasuresReq) error { func (s *Database) ListAllCumulativeMeasures(conn io.Writer, req proto.ListAllCumulativeMeasuresReq) error {
resultCh := make(chan cumulativeMeasuresResult, 1) responseWriter := transform.NewCumulativeMeasureWriter(conn, 0)
responseWriter := atree.NewCumulativeMeasureWriter(conn) return s.fullScan(fullScanReq{
s.appendJobToWorkerQueue(tryListAllCumulativeMeasuresReq{
MetricID: req.MetricID, MetricID: req.MetricID,
ResponseWriter: responseWriter, MetricType: diploma.Cumulative,
ResultCh: resultCh, Conn: conn,
})
result := <-resultCh
switch result.ResultCode {
case QueryDone:
responseWriter.Close()
case UntilFound:
err := s.atree.IterateAllCumulativeByTreeCursor(atree.IterateAllCumulativeByTreeCursorReq{
FracDigits: result.FracDigits,
PageNo: result.PageNo,
EndTimestamp: result.EndTimestamp,
EndValue: result.EndValue,
ResponseWriter: responseWriter, ResponseWriter: responseWriter,
}) })
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case NoMetric:
reply(conn, proto.ErrNoMetric)
case WrongMetricType:
reply(conn, proto.ErrWrongMetricType)
default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
}
return nil
} }
func (s *Database) ListInstantMeasures(conn net.Conn, req proto.ListInstantMeasuresReq) error { func (s *Database) ListInstantMeasures(conn net.Conn, req proto.ListInstantMeasuresReq) error {
@@ -724,182 +664,116 @@ func (s *Database) ListInstantMeasures(conn net.Conn, req proto.ListInstantMeasu
return nil return nil
} }
var since, until uint32 responseWriter := transform.NewInstantMeasureWriter(conn, req.Since)
if req.FirstHourOfDay > 0 {
since, until = correctToFHD(req.Since, req.Until, req.FirstHourOfDay)
} else {
since = req.Since
until = req.Until
}
resultCh := make(chan instantMeasuresResult, 1) return s.rangeScan(rangeScanReq{
responseWriter := atree.NewInstantMeasureWriter(conn)
s.appendJobToWorkerQueue(tryListInstantMeasuresReq{
MetricID: req.MetricID, MetricID: req.MetricID,
Since: since, MetricType: diploma.Instant,
Until: until, Since: req.Since,
ResponseWriter: responseWriter, Until: req.Until - 1,
ResultCh: resultCh, Conn: conn,
})
result := <-resultCh
switch result.ResultCode {
case QueryDone:
responseWriter.Close()
case UntilFound:
err := s.atree.ContinueIterateInstantByTreeCursor(atree.ContinueIterateInstantByTreeCursorReq{
FracDigits: result.FracDigits,
Since: since,
Until: until,
LastPageNo: result.PageNo,
ResponseWriter: responseWriter, ResponseWriter: responseWriter,
}) })
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case UntilNotFound:
err := s.atree.FindAndIterateInstantByTreeCursor(atree.FindAndIterateInstantByTreeCursorReq{
FracDigits: result.FracDigits,
Since: since,
Until: until,
RootPageNo: result.PageNo,
ResponseWriter: responseWriter,
})
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case NoMetric:
reply(conn, proto.ErrNoMetric)
case WrongMetricType:
reply(conn, proto.ErrWrongMetricType)
default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
}
return nil
}
type cumulativeMeasuresResult struct {
ResultCode byte
FracDigits byte
PageNo uint32
EndTimestamp uint32
EndValue float64
} }
func (s *Database) ListCumulativeMeasures(conn net.Conn, req proto.ListCumulativeMeasuresReq) error { func (s *Database) ListCumulativeMeasures(conn net.Conn, req proto.ListCumulativeMeasuresReq) error {
resultCh := make(chan cumulativeMeasuresResult, 1) if req.Since > req.Until {
responseWriter := atree.NewCumulativeMeasureWriter(conn) reply(conn, proto.ErrInvalidRange)
s.appendJobToWorkerQueue(tryListCumulativeMeasuresReq{
MetricID: req.MetricID,
Since: req.Since,
Until: req.Until,
ResponseWriter: responseWriter,
ResultCh: resultCh,
})
result := <-resultCh
switch result.ResultCode {
case QueryDone:
responseWriter.Close()
case UntilFound:
err := s.atree.ContinueIterateCumulativeByTreeCursor(atree.ContinueIterateCumulativeByTreeCursorReq{
FracDigits: result.FracDigits,
Since: req.Since,
Until: req.Until,
LastPageNo: result.PageNo,
EndTimestamp: result.EndTimestamp,
EndValue: result.EndValue,
ResponseWriter: responseWriter,
})
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case UntilNotFound:
err := s.atree.FindAndIterateCumulativeByTreeCursor(atree.FindAndIterateCumulativeByTreeCursorReq{
FracDigits: result.FracDigits,
Since: req.Since,
Until: req.Until,
RootPageNo: result.PageNo,
ResponseWriter: responseWriter,
})
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
}
case NoMetric:
reply(conn, proto.ErrNoMetric)
case WrongMetricType:
reply(conn, proto.ErrWrongMetricType)
default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
}
return nil return nil
} }
type instantPeriodsResult struct { responseWriter := transform.NewCumulativeMeasureWriter(conn, req.Since)
return s.rangeScan(rangeScanReq{
MetricID: req.MetricID,
MetricType: diploma.Cumulative,
Since: req.Since,
Until: req.Until - 1,
Conn: conn,
ResponseWriter: responseWriter,
})
}
type rangeScanResult struct {
ResultCode byte ResultCode byte
FracDigits byte FracDigits byte
PageNo uint32 RootPageNo uint32
LastPageNo uint32
} }
func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriodsReq) error { func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriodsReq) error {
var ( since, until := timeBoundsOfAggregation(req.Since, req.Until, req.GroupBy, req.FirstHourOfDay)
since = req.Since if since.After(until) {
until = req.Until reply(conn, proto.ErrInvalidRange)
) return nil
if req.FirstHourOfDay > 0 {
since, until = correctToFHD(since, until, req.FirstHourOfDay)
} }
if req.LastDayOfMonth > 0 { responseWriter, err := transform.NewInstantPeriodsWriter(transform.InstantPeriodsWriterOptions{
// fix Dst: conn,
}
resultCh := make(chan instantPeriodsResult, 1)
aggregator, err := atree.NewInstantAggregator(atree.InstantAggregatorOptions{
GroupBy: req.GroupBy, GroupBy: req.GroupBy,
AggregateFuncs: req.AggregateFuncs,
FirstHourOfDay: req.FirstHourOfDay, FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
}) })
if err != nil { if err != nil {
reply(conn, proto.ErrUnexpected) reply(conn, proto.ErrUnexpected)
return nil return nil
} }
responseWriter := atree.NewInstantPeriodsWriter(conn, req.AggregateFuncs) return s.rangeScan(rangeScanReq{
MetricID: req.MetricID,
MetricType: diploma.Instant,
Since: uint32(since.Unix()),
Until: uint32(until.Unix()),
Conn: conn,
ResponseWriter: responseWriter,
})
}
s.appendJobToWorkerQueue(tryListInstantPeriodsReq{ func (s *Database) ListCumulativePeriods(conn net.Conn, req proto.ListCumulativePeriodsReq) error {
since, until := timeBoundsOfAggregation(req.Since, req.Until, req.GroupBy, req.FirstHourOfDay)
if since.After(until) {
reply(conn, proto.ErrInvalidRange)
return nil
}
responseWriter, err := transform.NewCumulativePeriodsWriter(transform.CumulativePeriodsWriterOptions{
Dst: conn,
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
})
if err != nil {
reply(conn, proto.ErrUnexpected)
return nil
}
return s.rangeScan(rangeScanReq{
MetricID: req.MetricID,
MetricType: diploma.Cumulative,
Since: uint32(since.Unix()),
Until: uint32(until.Unix()),
Conn: conn,
ResponseWriter: responseWriter,
})
}
type rangeScanReq struct {
MetricID uint32
MetricType diploma.MetricType
Since uint32
Until uint32
Conn io.Writer
ResponseWriter atree.PeriodsWriter
}
func (s *Database) rangeScan(req rangeScanReq) error {
resultCh := make(chan rangeScanResult, 1)
s.appendJobToWorkerQueue(tryRangeScanReq{
MetricID: req.MetricID, MetricID: req.MetricID,
Since: req.Since, Since: req.Since,
Until: req.Until, Until: req.Until,
Aggregator: aggregator, MetricType: req.MetricType,
ResponseWriter: responseWriter, ResponseWriter: req.ResponseWriter,
ResultCh: resultCh, ResultCh: resultCh,
}) })
@@ -907,49 +781,44 @@ func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriod
switch result.ResultCode { switch result.ResultCode {
case QueryDone: case QueryDone:
responseWriter.Close() req.ResponseWriter.Close()
case UntilFound: case UntilFound:
err := s.atree.ContinueCollectInstantPeriods(atree.ContinueCollectInstantPeriodsReq{ err := s.atree.ContinueRangeScan(atree.ContinueRangeScanReq{
FracDigits: result.FracDigits, FracDigits: result.FracDigits,
Aggregator: aggregator, ResponseWriter: req.ResponseWriter,
ResponseWriter: responseWriter, LastPageNo: result.LastPageNo,
LastPageNo: result.PageNo,
Since: req.Since, Since: req.Since,
Until: req.Until,
}) })
s.metricRUnlock(req.MetricID) s.metricRUnlock(req.MetricID)
if err != nil { if err != nil {
reply(conn, proto.ErrUnexpected) reply(req.Conn, proto.ErrUnexpected)
} else { } else {
responseWriter.Close() req.ResponseWriter.Close()
} }
case UntilNotFound: case UntilNotFound:
err := s.atree.FindInstantPeriods(atree.FindInstantPeriodsReq{ err := s.atree.RangeScan(atree.RangeScanReq{
FracDigits: result.FracDigits, FracDigits: result.FracDigits,
ResponseWriter: responseWriter, ResponseWriter: req.ResponseWriter,
RootPageNo: result.PageNo, RootPageNo: result.RootPageNo,
Since: req.Since, Since: req.Since,
Until: req.Until, Until: req.Until,
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
}) })
s.metricRUnlock(req.MetricID) s.metricRUnlock(req.MetricID)
if err != nil { if err != nil {
reply(conn, proto.ErrUnexpected) reply(req.Conn, proto.ErrUnexpected)
} else { } else {
responseWriter.Close() req.ResponseWriter.Close()
} }
case NoMetric: case NoMetric:
reply(conn, proto.ErrNoMetric) reply(req.Conn, proto.ErrNoMetric)
case WrongMetricType: case WrongMetricType:
reply(conn, proto.ErrWrongMetricType) reply(req.Conn, proto.ErrWrongMetricType)
default: default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug) diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
@@ -957,33 +826,20 @@ func (s *Database) ListInstantPeriods(conn net.Conn, req proto.ListInstantPeriod
return nil return nil
} }
type cumulativePeriodsResult struct { type fullScanReq struct {
ResultCode byte MetricID uint32
FracDigits byte MetricType diploma.MetricType
PageNo uint32 Conn io.Writer
ResponseWriter atree.PeriodsWriter
} }
func (s *Database) ListCumulativePeriods(conn net.Conn, req proto.ListCumulativePeriodsReq) error { func (s *Database) fullScan(req fullScanReq) error {
resultCh := make(chan cumulativePeriodsResult, 1) resultCh := make(chan fullScanResult, 1)
aggregator, err := atree.NewCumulativeAggregator(atree.CumulativeAggregatorOptions{ s.appendJobToWorkerQueue(tryFullScanReq{
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
})
if err != nil {
reply(conn, proto.ErrUnexpected)
return nil
}
responseWriter := atree.NewCumulativePeriodsWriter(conn)
s.appendJobToWorkerQueue(tryListCumulativePeriodsReq{
MetricID: req.MetricID, MetricID: req.MetricID,
Since: req.Since, MetricType: req.MetricType,
Until: req.Until, ResponseWriter: req.ResponseWriter,
Aggregator: aggregator,
ResponseWriter: responseWriter,
ResultCh: resultCh, ResultCh: resultCh,
}) })
@@ -991,49 +847,26 @@ func (s *Database) ListCumulativePeriods(conn net.Conn, req proto.ListCumulative
switch result.ResultCode { switch result.ResultCode {
case QueryDone: case QueryDone:
responseWriter.Close() req.ResponseWriter.Close()
case UntilFound: case UntilFound:
err := s.atree.ContinueCollectCumulativePeriods(atree.ContinueCollectCumulativePeriodsReq{ err := s.atree.ContinueFullScan(atree.ContinueFullScanReq{
FracDigits: result.FracDigits, FracDigits: result.FracDigits,
Aggregator: aggregator, ResponseWriter: req.ResponseWriter,
ResponseWriter: responseWriter, LastPageNo: result.LastPageNo,
LastPageNo: result.PageNo,
Since: req.Since,
Until: req.Until,
}) })
s.metricRUnlock(req.MetricID) s.metricRUnlock(req.MetricID)
if err != nil { if err != nil {
reply(conn, proto.ErrUnexpected) reply(req.Conn, proto.ErrUnexpected)
} else { } else {
responseWriter.Close() req.ResponseWriter.Close()
}
case UntilNotFound:
err := s.atree.FindCumulativePeriods(atree.FindCumulativePeriodsReq{
FracDigits: result.FracDigits,
ResponseWriter: responseWriter,
RootPageNo: result.PageNo,
Since: req.Since,
Until: req.Until,
GroupBy: req.GroupBy,
FirstHourOfDay: req.FirstHourOfDay,
LastDayOfMonth: req.LastDayOfMonth,
})
s.metricRUnlock(req.MetricID)
if err != nil {
reply(conn, proto.ErrUnexpected)
} else {
responseWriter.Close()
} }
case NoMetric: case NoMetric:
reply(conn, proto.ErrNoMetric) reply(req.Conn, proto.ErrNoMetric)
case WrongMetricType: case WrongMetricType:
reply(conn, proto.ErrWrongMetricType) reply(req.Conn, proto.ErrWrongMetricType)
default: default:
diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug) diploma.Abort(diploma.WrongResultCodeBug, ErrWrongResultCodeBug)
@@ -1042,7 +875,7 @@ func (s *Database) ListCumulativePeriods(conn net.Conn, req proto.ListCumulative
} }
func (s *Database) ListCurrentValues(conn net.Conn, req proto.ListCurrentValuesReq) error { func (s *Database) ListCurrentValues(conn net.Conn, req proto.ListCurrentValuesReq) error {
responseWriter := atree.NewCurrentValueWriter(conn) responseWriter := transform.NewCurrentValueWriter(conn)
defer responseWriter.Close() defer responseWriter.Close()
resultCh := make(chan struct{}) resultCh := make(chan struct{})

24
database/helpers.go
View File

@@ -5,8 +5,32 @@ import (
"io/fs" "io/fs"
"os" "os"
"time" "time"
"gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/proto"
) )
func timeBoundsOfAggregation(since, until proto.TimeBound, groupBy diploma.GroupBy, firstHourOfDay int) (s time.Time, u time.Time) {
switch groupBy {
case diploma.GroupByHour, diploma.GroupByDay:
s = time.Date(since.Year, since.Month, since.Day, 0, 0, 0, 0, time.Local)
u = time.Date(until.Year, until.Month, until.Day, 0, 0, 0, 0, time.Local)
case diploma.GroupByMonth:
s = time.Date(since.Year, since.Month, 1, 0, 0, 0, 0, time.Local)
u = time.Date(until.Year, until.Month, 1, 0, 0, 0, 0, time.Local)
}
if firstHourOfDay > 0 {
duration := time.Duration(firstHourOfDay) * time.Hour
s = s.Add(duration)
u = u.Add(duration)
}
u = u.Add(-1 * time.Second)
return
}
func isFileExist(fileName string) (bool, error) { func isFileExist(fileName string) (bool, error) {
_, err := os.Stat(fileName) _, err := os.Stat(fileName)
if err != nil { if err != nil {

961
database/proc.go
View File

File diff suppressed because it is too large Load Diff

30
examples/loadtest/loadtest.go
View File

@@ -333,11 +333,22 @@ func execQuery(conn *client.Connection, queryGenerator *RandomQueryGenerator, st
} }
case listInstantPeriods: case listInstantPeriods:
since := time.Unix(int64(recipe.Since), 0)
until := time.Unix(int64(recipe.Until), 0)
//
t1 := time.Now() t1 := time.Now()
_, err := conn.ListInstantPeriods(proto.ListInstantPeriodsReq{ _, err := conn.ListInstantPeriods(proto.ListInstantPeriodsReq{
MetricID: recipe.MetricID, MetricID: recipe.MetricID,
Since: recipe.Since, Since: proto.TimeBound{
Until: recipe.Until, Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: recipe.GroupBy, GroupBy: recipe.GroupBy,
AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg, AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg,
}) })
@@ -354,11 +365,22 @@ func execQuery(conn *client.Connection, queryGenerator *RandomQueryGenerator, st
} }
case listCumulativePeriods: case listCumulativePeriods:
since := time.Unix(int64(recipe.Since), 0)
until := time.Unix(int64(recipe.Until), 0)
//
t1 := time.Now() t1 := time.Now()
_, err := conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{ _, err := conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{
MetricID: recipe.MetricID, MetricID: recipe.MetricID,
Since: recipe.Since, Since: proto.TimeBound{
Until: recipe.Until, Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: recipe.GroupBy, GroupBy: recipe.GroupBy,
}) })
elapsedTime = time.Since(t1) elapsedTime = time.Since(t1)

14
examples/requests/generate.go
View File

@@ -4,12 +4,12 @@ import (
"math/rand" "math/rand"
"time" "time"
"gordenko.dev/dima/diploma/client" "gordenko.dev/dima/diploma/proto"
) )
func GenerateCumulativeMeasures(days int) []client.Measure { func GenerateCumulativeMeasures(days int) []proto.Measure {
var ( var (
measures []client.Measure measures []proto.Measure
minutes = []int{14, 29, 44, 59} minutes = []int{14, 29, 44, 59}
hoursPerDay = 24 hoursPerDay = 24
totalHours = days * hoursPerDay totalHours = days * hoursPerDay
@@ -31,7 +31,7 @@ func GenerateCumulativeMeasures(days int) []client.Measure {
time.Local, time.Local,
) )
measure := client.Measure{ measure := proto.Measure{
Timestamp: uint32(measureTime.Unix()), Timestamp: uint32(measureTime.Unix()),
Value: totalValue, Value: totalValue,
} }
@@ -43,9 +43,9 @@ func GenerateCumulativeMeasures(days int) []client.Measure {
return measures return measures
} }
func GenerateInstantMeasures(days int, baseValue float64) []client.Measure { func GenerateInstantMeasures(days int, baseValue float64) []proto.Measure {
var ( var (
measures []client.Measure measures []proto.Measure
minutes = []int{14, 29, 44, 59} minutes = []int{14, 29, 44, 59}
hoursPerDay = 24 hoursPerDay = 24
totalHours = days * hoursPerDay totalHours = days * hoursPerDay
@@ -70,7 +70,7 @@ func GenerateInstantMeasures(days int, baseValue float64) []client.Measure {
fluctuation := baseValue * 0.1 fluctuation := baseValue * 0.1
value := baseValue + (rand.Float64()*2-1)*fluctuation value := baseValue + (rand.Float64()*2-1)*fluctuation
measure := client.Measure{ measure := proto.Measure{
Timestamp: uint32(measureTime.Unix()), Timestamp: uint32(measureTime.Unix()),
Value: value, Value: value,
} }

82
examples/requests/requests.go
View File

@@ -14,7 +14,7 @@ func sendRequests(conn *client.Connection) {
var ( var (
instantMetricID uint32 = 10000 instantMetricID uint32 = 10000
cumulativeMetricID uint32 = 10001 cumulativeMetricID uint32 = 10001
fracDigits byte = 2 fracDigits int = 2
err error err error
) )
@@ -23,7 +23,7 @@ func sendRequests(conn *client.Connection) {
// ADD INSTANT METRIC // ADD INSTANT METRIC
err = conn.AddMetric(client.Metric{ err = conn.AddMetric(proto.AddMetricReq{
MetricID: instantMetricID, MetricID: instantMetricID,
MetricType: diploma.Instant, MetricType: diploma.Instant,
FracDigits: fracDigits, FracDigits: fracDigits,
@@ -53,7 +53,7 @@ GetMetric:
instantMeasures := GenerateInstantMeasures(62, 220) instantMeasures := GenerateInstantMeasures(62, 220)
err = conn.AppendMeasures(client.AppendMeasuresReq{ err = conn.AppendMeasures(proto.AppendMeasuresReq{
MetricID: instantMetricID, MetricID: instantMetricID,
Measures: instantMeasures, Measures: instantMeasures,
}) })
@@ -104,8 +104,16 @@ GetMetric:
instantPeriods, err := conn.ListInstantPeriods(proto.ListInstantPeriodsReq{ instantPeriods, err := conn.ListInstantPeriods(proto.ListInstantPeriodsReq{
MetricID: instantMetricID, MetricID: instantMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByHour, GroupBy: diploma.GroupByHour,
AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg, AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg,
}) })
@@ -125,8 +133,16 @@ GetMetric:
instantPeriods, err = conn.ListInstantPeriods(proto.ListInstantPeriodsReq{ instantPeriods, err = conn.ListInstantPeriods(proto.ListInstantPeriodsReq{
MetricID: instantMetricID, MetricID: instantMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByDay, GroupBy: diploma.GroupByDay,
AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg, AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg,
}) })
@@ -146,8 +162,16 @@ GetMetric:
instantPeriods, err = conn.ListInstantPeriods(proto.ListInstantPeriodsReq{ instantPeriods, err = conn.ListInstantPeriods(proto.ListInstantPeriodsReq{
MetricID: instantMetricID, MetricID: instantMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByMonth, GroupBy: diploma.GroupByMonth,
AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg, AggregateFuncs: diploma.AggregateMin | diploma.AggregateMax | diploma.AggregateAvg,
}) })
@@ -182,7 +206,7 @@ GetMetric:
// ADD CUMULATIVE METRIC // ADD CUMULATIVE METRIC
err = conn.AddMetric(client.Metric{ err = conn.AddMetric(proto.AddMetricReq{
MetricID: cumulativeMetricID, MetricID: cumulativeMetricID,
MetricType: diploma.Cumulative, MetricType: diploma.Cumulative,
FracDigits: fracDigits, FracDigits: fracDigits,
@@ -212,7 +236,7 @@ GetMetric:
cumulativeMeasures := GenerateCumulativeMeasures(62) cumulativeMeasures := GenerateCumulativeMeasures(62)
err = conn.AppendMeasures(client.AppendMeasuresReq{ err = conn.AppendMeasures(proto.AppendMeasuresReq{
MetricID: cumulativeMetricID, MetricID: cumulativeMetricID,
Measures: cumulativeMeasures, Measures: cumulativeMeasures,
}) })
@@ -264,8 +288,16 @@ GetMetric:
cumulativePeriods, err := conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{ cumulativePeriods, err := conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{
MetricID: cumulativeMetricID, MetricID: cumulativeMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByHour, GroupBy: diploma.GroupByHour,
}) })
if err != nil { if err != nil {
@@ -284,8 +316,16 @@ GetMetric:
cumulativePeriods, err = conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{ cumulativePeriods, err = conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{
MetricID: cumulativeMetricID, MetricID: cumulativeMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByDay, GroupBy: diploma.GroupByDay,
}) })
if err != nil { if err != nil {
@@ -304,8 +344,16 @@ GetMetric:
cumulativePeriods, err = conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{ cumulativePeriods, err = conn.ListCumulativePeriods(proto.ListCumulativePeriodsReq{
MetricID: cumulativeMetricID, MetricID: cumulativeMetricID,
Since: uint32(since.Unix()), Since: proto.TimeBound{
Until: uint32(until.Unix()), Year: since.Year(),
Month: since.Month(),
Day: since.Day(),
},
Until: proto.TimeBound{
Year: until.Year(),
Month: until.Month(),
Day: until.Day(),
},
GroupBy: diploma.GroupByMonth, GroupBy: diploma.GroupByMonth,
}) })
if err != nil { if err != nil {

609
proto/proto.go
View File

@@ -2,8 +2,9 @@ package proto
import ( import (
"fmt" "fmt"
"time"
octopus "gordenko.dev/dima/diploma" "gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/bin" "gordenko.dev/dima/diploma/bin"
"gordenko.dev/dima/diploma/bufreader" "gordenko.dev/dima/diploma/bufreader"
) )
@@ -13,13 +14,14 @@ const (
TypeListCurrentValues byte = 2 TypeListCurrentValues byte = 2
TypeListInstantMeasures byte = 3 TypeListInstantMeasures byte = 3
TypeListCumulativeMeasures byte = 33 TypeListCumulativeMeasures byte = 33
TypeListInstantPeriods byte = 4 TypeListInstantPeriods byte = 4
TypeListCumulativePeriods byte = 44 TypeListCumulativePeriods byte = 44
TypeGetMetric byte = 5 TypeGetMetric byte = 5
TypeAddMetric byte = 6 TypeAddMetric byte = 6
TypeListAllInstantMeasures byte = 8 TypeListAllInstantMeasures byte = 8
TypeListAllCumulativeMeasures byte = 88 TypeListAllCumulativeMeasures byte = 88
TypeRangeTotal byte = 9
TypeAppendMeasure byte = 10 TypeAppendMeasure byte = 10
TypeAppendMeasures byte = 11 TypeAppendMeasures byte = 11
TypeDeleteMetric byte = 12 TypeDeleteMetric byte = 12
@@ -66,218 +68,65 @@ func ErrorCodeToText(code uint16) string {
} }
} }
type GetMetricReq struct { // common
MetricID uint32
}
type ListCurrentValuesReq struct { type Metric struct {
MetricIDs []uint32
}
type AddMetricReq struct {
MetricID uint32 MetricID uint32
MetricType octopus.MetricType MetricType diploma.MetricType
FracDigits int FracDigits int
} }
type UpdateMetricReq struct { type AppendError struct {
MetricID uint32 MetricID uint32
MetricType octopus.MetricType ErrorCode uint16
FracDigits int
} }
type DeleteMetricReq struct { type TimeBound struct {
MetricID uint32 Year int
Month time.Month
Day int
} }
type DeleteMeasuresReq struct { type CumulativeMeasure struct {
MetricID uint32 Timestamp uint32
Since uint32 // timestamp (optional) Value float64
Total float64
} }
type AppendMeasureReq struct { type CumulativePeriod struct {
Period uint32
Since uint32
Until uint32
EndValue float64
Total float64
}
type InstantMeasure struct {
Timestamp uint32
Value float64
}
type InstantPeriod struct {
Period uint32
Since uint32
Until uint32
Min float64
Max float64
Avg float64
}
type CurrentValue struct {
MetricID uint32 MetricID uint32
Timestamp uint32 Timestamp uint32
Value float64 Value float64
} }
type ListAllInstantMetricMeasuresReq struct { // API reqs
type GetMetricReq struct {
MetricID uint32 MetricID uint32
} }
type ListAllCumulativeMeasuresReq struct {
MetricID uint32
}
type ListInstantMeasuresReq struct {
MetricID uint32
Since uint32
Until uint32
FirstHourOfDay int
}
type ListCumulativeMeasuresReq struct {
MetricID uint32
Since uint32
Until uint32
FirstHourOfDay int
}
type ListInstantPeriodsReq struct {
MetricID uint32
Since uint32
Until uint32
GroupBy octopus.GroupBy
AggregateFuncs byte
FirstHourOfDay int
LastDayOfMonth int
}
type ListCumulativePeriodsReq struct {
MetricID uint32
Since uint32
Until uint32
GroupBy octopus.GroupBy
FirstHourOfDay int
LastDayOfMonth int
}
type Metric struct {
MetricID uint32
MetricType octopus.MetricType
FracDigits int
}
type RangeTotalReq struct {
MetricID uint32
Since uint32
Until uint32
}
func PackAddMetricReq(req AddMetricReq) []byte {
arr := []byte{
TypeAddMetric,
0, 0, 0, 0, //
byte(req.MetricType),
byte(req.FracDigits),
}
bin.PutUint32(arr[1:], req.MetricID)
return arr
}
func PackDeleteMetricReq(req DeleteMetricReq) []byte {
arr := []byte{
TypeDeleteMetric,
0, 0, 0, 0, // metricID
}
bin.PutUint32(arr[1:], req.MetricID)
return arr
}
func PackAppendMeasure(req AppendMeasureReq) []byte {
arr := []byte{
TypeAppendMeasure,
0, 0, 0, 0, // metricID
0, 0, 0, 0, // timestamp
0, 0, 0, 0, 0, 0, 0, 0, // value
}
bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], uint32(req.Timestamp))
bin.PutFloat64(arr[9:], req.Value)
return arr
}
func PackDeleteMeasuresReq(req DeleteMeasuresReq) []byte {
arr := []byte{
TypeDeleteMeasures,
0, 0, 0, 0, // metricID
0, 0, 0, 0, // since
}
bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint32(arr[5:], uint32(req.Since))
return arr
}
// UNPACK reqs
func UnpackAddMetricReq(arr []byte) (m AddMetricReq) {
m.MetricID = bin.GetUint32(arr)
m.MetricType = octopus.MetricType(arr[4])
m.FracDigits = int(arr[5])
return
}
func UnpackUpdateMetricReq(arr []byte) (m UpdateMetricReq) {
m.MetricID = bin.GetUint32(arr)
m.MetricType = octopus.MetricType(arr[4])
m.FracDigits = int(arr[5])
return
}
func UnpackDeleteMetricReq(arr []byte) (m DeleteMetricReq) {
m.MetricID = bin.GetUint32(arr)
return
}
func UnpackAppendMeasureReq(arr []byte) (m AppendMeasureReq) {
m.MetricID = bin.GetUint32(arr)
m.Timestamp = bin.GetUint32(arr[4:])
m.Value = bin.GetFloat64(arr[8:])
return
}
func UnpackDeleteMeasuresReq(arr []byte) (m DeleteMeasuresReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
return
}
func UnpackListInstantMeasuresReq(arr []byte) (m ListInstantMeasuresReq) {
m.MetricID = bin.GetUint32(arr[0:])
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
m.FirstHourOfDay = int(arr[12])
return
}
func UnpackListCumulativeMeasuresReq(arr []byte) (m ListCumulativeMeasuresReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
m.FirstHourOfDay = int(arr[12])
return
}
func UnpackListInstantPeriodsReq(arr []byte) (m ListInstantPeriodsReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
m.GroupBy = octopus.GroupBy(arr[12])
m.AggregateFuncs = arr[13]
m.FirstHourOfDay = int(arr[14])
m.LastDayOfMonth = int(arr[15])
return
}
func UnpackListCumulativePeriodsReq(arr []byte) (m ListCumulativePeriodsReq) {
m.MetricID = bin.GetUint32(arr[0:])
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
m.GroupBy = octopus.GroupBy(arr[12])
m.FirstHourOfDay = int(arr[13])
m.LastDayOfMonth = int(arr[14])
return
}
func UnpackRangeTotalReq(arr []byte) (m RangeTotalReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
return
}
// READ reqs
func ReadGetMetricReq(r *bufreader.BufferedReader) (m GetMetricReq, err error) { func ReadGetMetricReq(r *bufreader.BufferedReader) (m GetMetricReq, err error) {
m.MetricID, err = bin.ReadUint32(r) m.MetricID, err = bin.ReadUint32(r)
if err != nil { if err != nil {
@@ -287,112 +136,8 @@ func ReadGetMetricReq(r *bufreader.BufferedReader) (m GetMetricReq, err error) {
return return
} }
func ReadAddMetricReq(r *bufreader.BufferedReader) (m AddMetricReq, err error) { type ListCurrentValuesReq struct {
arr, err := r.ReadN(6) MetricIDs []uint32
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackAddMetricReq(arr), nil
}
func ReadUpdateMetricReq(r *bufreader.BufferedReader) (m UpdateMetricReq, err error) {
arr, err := r.ReadN(6)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackUpdateMetricReq(arr), nil
}
func ReadDeleteMetricReq(r *bufreader.BufferedReader) (m DeleteMetricReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
func ReadAppendMeasureReq(r *bufreader.BufferedReader) (m AppendMeasureReq, err error) {
arr, err := r.ReadN(16)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackAppendMeasureReq(arr), nil
}
func ReadDeleteMeasuresReq(r *bufreader.BufferedReader) (m DeleteMeasuresReq, err error) {
arr, err := r.ReadN(8)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackDeleteMeasuresReq(arr), nil
}
func ReadListAllInstantMeasuresReq(r *bufreader.BufferedReader) (m ListAllInstantMetricMeasuresReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
func ReadListAllCumulativeMeasuresReq(r *bufreader.BufferedReader) (m ListAllCumulativeMeasuresReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
func ReadListInstantMeasuresReq(r *bufreader.BufferedReader) (m ListInstantMeasuresReq, err error) {
arr, err := r.ReadN(13)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListInstantMeasuresReq(arr), nil
}
func ReadListCumulativeMeasuresReq(r *bufreader.BufferedReader) (m ListCumulativeMeasuresReq, err error) {
arr, err := r.ReadN(13)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListCumulativeMeasuresReq(arr), nil
}
func ReadListInstantPeriodsReq(r *bufreader.BufferedReader) (m ListInstantPeriodsReq, err error) {
arr, err := r.ReadN(16)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListInstantPeriodsReq(arr), nil
}
func ReadListCumulativePeriodsReq(r *bufreader.BufferedReader) (m ListCumulativePeriodsReq, err error) {
arr, err := r.ReadN(15)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListCumulativePeriodsReq(arr), nil
}
func ReadRangeTotalReq(r *bufreader.BufferedReader) (m RangeTotalReq, err error) {
arr, err := r.ReadN(12)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackRangeTotalReq(arr), nil
} }
func ReadListCurrentValuesReq(r *bufreader.BufferedReader) (m ListCurrentValuesReq, err error) { func ReadListCurrentValuesReq(r *bufreader.BufferedReader) (m ListCurrentValuesReq, err error) {
@@ -414,6 +159,83 @@ func ReadListCurrentValuesReq(r *bufreader.BufferedReader) (m ListCurrentValuesR
return return
} }
type AddMetricReq struct {
MetricID uint32
MetricType diploma.MetricType
FracDigits int
}
func ReadAddMetricReq(r *bufreader.BufferedReader) (m AddMetricReq, err error) {
arr, err := r.ReadN(6)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackAddMetricReq(arr), nil
}
func UnpackAddMetricReq(arr []byte) (m AddMetricReq) {
m.MetricID = bin.GetUint32(arr)
m.MetricType = diploma.MetricType(arr[4])
m.FracDigits = int(arr[5])
return
}
type DeleteMetricReq struct {
MetricID uint32
}
func ReadDeleteMetricReq(r *bufreader.BufferedReader) (m DeleteMetricReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
type DeleteMeasuresReq struct {
MetricID uint32
Since uint32 // timestamp (optional)
}
func ReadDeleteMeasuresReq(r *bufreader.BufferedReader) (m DeleteMeasuresReq, err error) {
arr, err := r.ReadN(8)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackDeleteMeasuresReq(arr), nil
}
func UnpackDeleteMeasuresReq(arr []byte) (m DeleteMeasuresReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
return
}
type AppendMeasureReq struct {
MetricID uint32
Timestamp uint32
Value float64
}
func ReadAppendMeasureReq(r *bufreader.BufferedReader) (m AppendMeasureReq, err error) {
arr, err := r.ReadN(16)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackAppendMeasureReq(arr), nil
}
func UnpackAppendMeasureReq(arr []byte) (m AppendMeasureReq) {
m.MetricID = bin.GetUint32(arr)
m.Timestamp = bin.GetUint32(arr[4:])
m.Value = bin.GetFloat64(arr[8:])
return
}
type AppendMeasuresReq struct { type AppendMeasuresReq struct {
MetricID uint32 MetricID uint32
Measures []Measure Measures []Measure
@@ -424,27 +246,6 @@ type Measure struct {
Value float64 Value float64
} }
func PackAppendMeasures(req AppendMeasuresReq) []byte {
if len(req.Measures) > 65535 {
panic(fmt.Errorf("wrong measures qty: %d", len(req.Measures)))
}
var (
prefixSize = 7
recordSize = 12
arr = make([]byte, prefixSize+len(req.Measures)*recordSize)
)
arr[0] = TypeAppendMeasures
bin.PutUint32(arr[1:], req.MetricID)
bin.PutUint16(arr[5:], uint16(len(req.Measures)))
pos := prefixSize
for _, measure := range req.Measures {
bin.PutUint32(arr[pos:], measure.Timestamp)
bin.PutFloat64(arr[pos+4:], measure.Value)
pos += recordSize
}
return arr
}
func ReadAppendMeasuresReq(r *bufreader.BufferedReader) (m AppendMeasuresReq, err error) { func ReadAppendMeasuresReq(r *bufreader.BufferedReader) (m AppendMeasuresReq, err error) {
prefix, err := bin.ReadN(r, 6) // metricID + measures qty prefix, err := bin.ReadN(r, 6) // metricID + measures qty
if err != nil { if err != nil {
@@ -471,3 +272,169 @@ func ReadAppendMeasuresReq(r *bufreader.BufferedReader) (m AppendMeasuresReq, er
} }
return return
} }
type MetricMeasure struct {
MetricID uint32
Timestamp uint32
Value float64
}
func ReadAppendMeasurePerMetricReq(r *bufreader.BufferedReader) (measures []MetricMeasure, err error) {
qty, err := bin.ReadUint16(r)
if err != nil {
return
}
var tmp = make([]byte, 16)
for range int(qty) {
err = bin.ReadNInto(r, tmp)
if err != nil {
return
}
measures = append(measures, MetricMeasure{
MetricID: bin.GetUint32(tmp[0:]),
Timestamp: bin.GetUint32(tmp[4:]),
Value: bin.GetFloat64(tmp[8:]),
})
}
return
}
type ListAllInstantMetricMeasuresReq struct {
MetricID uint32
}
func ReadListAllInstantMeasuresReq(r *bufreader.BufferedReader) (m ListAllInstantMetricMeasuresReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
type ListAllCumulativeMeasuresReq struct {
MetricID uint32
}
func ReadListAllCumulativeMeasuresReq(r *bufreader.BufferedReader) (m ListAllCumulativeMeasuresReq, err error) {
m.MetricID, err = bin.ReadUint32(r)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return
}
type ListInstantMeasuresReq struct {
MetricID uint32
Since uint32
Until uint32
}
func ReadListInstantMeasuresReq(r *bufreader.BufferedReader) (m ListInstantMeasuresReq, err error) {
arr, err := r.ReadN(12)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListInstantMeasuresReq(arr), nil
}
func UnpackListInstantMeasuresReq(arr []byte) (m ListInstantMeasuresReq) {
m.MetricID = bin.GetUint32(arr[0:])
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
return
}
type ListCumulativeMeasuresReq struct {
MetricID uint32
Since uint32
Until uint32
}
func ReadListCumulativeMeasuresReq(r *bufreader.BufferedReader) (m ListCumulativeMeasuresReq, err error) {
arr, err := r.ReadN(12)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListCumulativeMeasuresReq(arr), nil
}
func UnpackListCumulativeMeasuresReq(arr []byte) (m ListCumulativeMeasuresReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = bin.GetUint32(arr[4:])
m.Until = bin.GetUint32(arr[8:])
return
}
type ListInstantPeriodsReq struct {
MetricID uint32
Since TimeBound
Until TimeBound
GroupBy diploma.GroupBy
AggregateFuncs byte
FirstHourOfDay int
}
func ReadListInstantPeriodsReq(r *bufreader.BufferedReader) (m ListInstantPeriodsReq, err error) {
arr, err := r.ReadN(15)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListInstantPeriodsReq(arr), nil
}
func UnpackListInstantPeriodsReq(arr []byte) (m ListInstantPeriodsReq) {
m.MetricID = bin.GetUint32(arr)
m.Since = TimeBound{
Year: int(bin.GetUint16(arr[4:])),
Month: time.Month(arr[6]),
Day: int(arr[7]),
}
m.Until = TimeBound{
Year: int(bin.GetUint16(arr[8:])),
Month: time.Month(arr[10]),
Day: int(arr[11]),
}
m.GroupBy = diploma.GroupBy(arr[12])
m.AggregateFuncs = arr[13]
m.FirstHourOfDay = int(arr[14])
return
}
type ListCumulativePeriodsReq struct {
MetricID uint32
Since TimeBound
Until TimeBound
GroupBy diploma.GroupBy
FirstHourOfDay int
}
func ReadListCumulativePeriodsReq(r *bufreader.BufferedReader) (m ListCumulativePeriodsReq, err error) {
arr, err := r.ReadN(14)
if err != nil {
err = fmt.Errorf("read req: %s", err)
return
}
return UnpackListCumulativePeriodsReq(arr), nil
}
func UnpackListCumulativePeriodsReq(arr []byte) (m ListCumulativePeriodsReq) {
m.MetricID = bin.GetUint32(arr[0:])
m.Since = TimeBound{
Year: int(bin.GetUint16(arr[4:])),
Month: time.Month(arr[6]),
Day: int(arr[7]),
}
m.Until = TimeBound{
Year: int(bin.GetUint16(arr[8:])),
Month: time.Month(arr[10]),
Day: int(arr[11]),
}
m.GroupBy = diploma.GroupBy(arr[12])
m.FirstHourOfDay = int(arr[13])
return
}

408
transform/aggregate.go Normal file
View File

@@ -0,0 +1,408 @@
package transform
import (
"errors"
"fmt"
"io"
"time"
"gordenko.dev/dima/diploma"
"gordenko.dev/dima/diploma/bin"
"gordenko.dev/dima/diploma/timeutil"
)
// INSTANT
type InstantPeriodsWriterOptions struct {
Dst io.Writer
GroupBy diploma.GroupBy
AggregateFuncs byte
FirstHourOfDay int
}
type InstantPeriodsWriter struct {
aggregateFuncs byte
arr []byte
responder *ChunkedResponder
groupBy diploma.GroupBy
firstHourOfDay int
time2period func(uint32) time.Time
currentPeriod time.Time
lastTimestamp uint32
endTimestamp uint32 // время показания на конец периода
min float64
max float64
total float64
entries int
}
func NewInstantPeriodsWriter(opt InstantPeriodsWriterOptions) (*InstantPeriodsWriter, error) {
if opt.Dst == nil {
return nil, errors.New("Dst option is required")
}
if opt.FirstHourOfDay < 0 || opt.FirstHourOfDay > 23 {
return nil, fmt.Errorf("wrong FirstHourOfDay option: %d", opt.FirstHourOfDay)
}
// Считаю q, чтобы заранее выделить массив для упаковки периодов
var q int
if (opt.AggregateFuncs & diploma.AggregateMin) == diploma.AggregateMin {
q++
}
if (opt.AggregateFuncs & diploma.AggregateMax) == diploma.AggregateMax {
q++
}
if (opt.AggregateFuncs & diploma.AggregateAvg) == diploma.AggregateAvg {
q++
}
if q == 0 {
return nil, errors.New("AggregateFuncs option is required")
}
// 12 - это period, since, until
// 8 - это размер float64
s := &InstantPeriodsWriter{
aggregateFuncs: opt.AggregateFuncs,
arr: make([]byte, 12+q*8),
responder: NewChunkedResponder(opt.Dst),
groupBy: opt.GroupBy,
firstHourOfDay: opt.FirstHourOfDay,
}
switch opt.GroupBy {
case diploma.GroupByHour:
s.time2period = groupByHour
case diploma.GroupByDay:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByDayUsingFHD
} else {
s.time2period = groupByDay
}
case diploma.GroupByMonth:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByMonthUsingFHD
} else {
s.time2period = groupByMonth
}
default:
return nil, fmt.Errorf("unknown groupBy %d option", opt.GroupBy)
}
return s, nil
}
func (s *InstantPeriodsWriter) groupByDayUsingFHD(timestamp uint32) time.Time {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return tm
}
func (s *InstantPeriodsWriter) groupByMonthUsingFHD(timestamp uint32) time.Time {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return tm
}
func (s *InstantPeriodsWriter) Feed(timestamp uint32, value float64) {
s.feed(timestamp, value, false)
}
func (s *InstantPeriodsWriter) FeedNoSend(timestamp uint32, value float64) {
s.feed(timestamp, value, true)
}
func (s *InstantPeriodsWriter) feed(timestamp uint32, value float64, isBuffer bool) {
if s.entries > 0 {
period := s.time2period(timestamp)
if period != s.currentPeriod {
// закрываю период
// готовый период
s.packPeriod(timestamp)
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
// затем
s.decrementPeriod()
//fmt.Println(" period: ", period.Format("2006-01-02 15:04:05"))
//fmt.Println("current period: ", s.currentPeriod.Format("2006-01-02 15:04:05"))
for period.Before(s.currentPeriod) {
// вставляю пустышку
s.packBlankPeriod()
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
s.decrementPeriod()
//fmt.Println(" period: ", period.Format("2006-01-02 15:04:05"))
//fmt.Println("current period: ", s.currentPeriod.Format("2006-01-02 15:04:05"))
//return
}
s.endTimestamp = timestamp
s.min = value
s.max = value
s.total = value
s.entries = 1
} else {
if value < s.min {
s.min = value
} else if value > s.max {
s.max = value
}
// для подсчета AVG
s.total += value
s.entries++
}
} else {
s.endTimestamp = timestamp
s.min = value
s.max = value
s.total = value
s.entries = 1
s.currentPeriod = s.time2period(timestamp)
}
s.lastTimestamp = timestamp
}
func (s *InstantPeriodsWriter) decrementPeriod() {
switch s.groupBy {
case diploma.GroupByHour:
s.currentPeriod = s.currentPeriod.Add(-1 * time.Hour)
//fmt.Println("decrement")
case diploma.GroupByDay:
s.currentPeriod = s.currentPeriod.AddDate(0, 0, -1)
case diploma.GroupByMonth:
s.currentPeriod = s.currentPeriod.AddDate(0, -1, 0)
}
}
func (s *InstantPeriodsWriter) packBlankPeriod() {
//period := s.currentPeriod.Format("2006-01-02 15:04:05")
//since := "0"
//until := "0"
//fmt.Printf("%s: %s - %s, %.0f - %.0f\n", period, since, until, 0.0, 0.0)
// until - это endTimestamp всегда
bin.PutUint32(s.arr[0:], uint32(s.currentPeriod.Unix()))
for i := 4; i < len(s.arr); i++ {
s.arr[i] = 0
}
}
func (s *InstantPeriodsWriter) Close() (err error) {
if s.entries > 0 {
s.packPeriod(s.lastTimestamp)
s.responder.AppendRecord(s.arr)
}
return s.responder.Flush()
}
func (s *InstantPeriodsWriter) packPeriod(timestamp uint32) {
bin.PutUint32(s.arr[0:], uint32(s.currentPeriod.Unix()))
bin.PutUint32(s.arr[4:], timestamp)
bin.PutUint32(s.arr[8:], s.endTimestamp)
pos := 12
if (s.aggregateFuncs & diploma.AggregateMin) == diploma.AggregateMin {
bin.PutFloat64(s.arr[pos:], s.min)
pos += 8
}
if (s.aggregateFuncs & diploma.AggregateMax) == diploma.AggregateMax {
bin.PutFloat64(s.arr[pos:], s.max)
pos += 8
}
if (s.aggregateFuncs & diploma.AggregateAvg) == diploma.AggregateAvg {
bin.PutFloat64(s.arr[pos:], s.total/float64(s.entries))
}
}
/*
Идея с разбивкой на периоды:
Для каждого периода нахожу одно последнее значение.
Начало периода - это конец предыдущего. Если предыдущий не строго предыдущий,
а с пропусками - на место пропусков вставляю пустышки.
Плюс такого решения - я всегда показываю реальное значение на конец периода.
*/
type CumulativePeriodsWriter struct {
arr []byte
responder *ChunkedResponder
firstHourOfDay int
currentPeriod time.Time
groupBy diploma.GroupBy
time2period func(uint32) time.Time
endTimestamp uint32
endValue float64
lastTimestamp uint32
lastValue float64
}
type CumulativePeriodsWriterOptions struct {
Dst io.Writer
GroupBy diploma.GroupBy
FirstHourOfDay int
}
func NewCumulativePeriodsWriter(opt CumulativePeriodsWriterOptions) (*CumulativePeriodsWriter, error) {
if opt.Dst == nil {
return nil, errors.New("Dst option is required")
}
// Считаю q, чтобы заранее выделить массив для упаковки периодов
if opt.FirstHourOfDay < 0 || opt.FirstHourOfDay > 23 {
return nil, fmt.Errorf("wrong firstHourOfDay option: %d", opt.FirstHourOfDay)
}
s := &CumulativePeriodsWriter{
arr: make([]byte, 28),
responder: NewChunkedResponder(opt.Dst),
firstHourOfDay: opt.FirstHourOfDay,
groupBy: opt.GroupBy,
}
s.time2period = func(timestamp uint32) time.Time {
return timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "h")
}
switch opt.GroupBy {
case diploma.GroupByHour:
s.time2period = groupByHour
case diploma.GroupByDay:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByDayUsingFHD
} else {
s.time2period = groupByDay
}
case diploma.GroupByMonth:
if s.firstHourOfDay > 0 {
s.time2period = s.groupByMonthUsingFHD
} else {
s.time2period = groupByMonth
}
default:
return nil, fmt.Errorf("unknown groupBy %d option", opt.GroupBy)
}
return s, nil
}
func (s *CumulativePeriodsWriter) groupByDayUsingFHD(timestamp uint32) time.Time {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return tm
}
func (s *CumulativePeriodsWriter) groupByMonthUsingFHD(timestamp uint32) time.Time {
tm := timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
if tm.Hour() < s.firstHourOfDay {
tm = tm.AddDate(0, 0, -1)
}
return tm
}
func (s *CumulativePeriodsWriter) Feed(timestamp uint32, value float64) {
s.feed(timestamp, value, false)
}
func (s *CumulativePeriodsWriter) FeedNoSend(timestamp uint32, value float64) {
s.feed(timestamp, value, true)
}
func (s *CumulativePeriodsWriter) feed(timestamp uint32, value float64, isBuffer bool) {
if s.endTimestamp > 0 {
period := s.time2period(timestamp)
if period != s.currentPeriod {
// закрываю период
s.packPeriod(timestamp, value)
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
// затем
s.decrementPeriod()
//fmt.Println(" period: ", period.Format("2006-01-02 15:04:05"))
//fmt.Println("current period: ", s.currentPeriod.Format("2006-01-02 15:04:05"))
for period.Before(s.currentPeriod) {
// вставляю пустышку
s.packBlankPeriod()
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
s.decrementPeriod()
//fmt.Println(" period: ", period.Format("2006-01-02 15:04:05"))
//fmt.Println("current period: ", s.currentPeriod.Format("2006-01-02 15:04:05"))
//return
}
s.endTimestamp = timestamp
s.endValue = value
}
} else {
s.endTimestamp = timestamp
s.endValue = value
s.currentPeriod = s.time2period(timestamp)
}
s.lastTimestamp = timestamp
s.lastValue = value
}
func (s *CumulativePeriodsWriter) decrementPeriod() {
switch s.groupBy {
case diploma.GroupByHour:
s.currentPeriod = s.currentPeriod.Add(-1 * time.Hour)
//fmt.Println("decrement")
case diploma.GroupByDay:
s.currentPeriod = s.currentPeriod.AddDate(0, 0, -1)
case diploma.GroupByMonth:
s.currentPeriod = s.currentPeriod.AddDate(0, -1, 0)
}
}
func (s *CumulativePeriodsWriter) packBlankPeriod() {
//period := s.currentPeriod.Format("2006-01-02 15:04:05")
//since := "0"
//until := "0"
//fmt.Printf("%s: %s - %s, %.0f - %.0f\n", period, since, until, 0.0, 0.0)
// until - это endTimestamp всегда
bin.PutUint32(s.arr[0:], uint32(s.currentPeriod.Unix()))
for i := 4; i < len(s.arr); i++ {
s.arr[i] = 0
}
}
func (s *CumulativePeriodsWriter) packPeriod(start uint32, startValue float64) {
//period := s.currentPeriod.Format("2006-01-02 15:04:05")
//since := time.Unix(int64(start), 0).Format("2006-01-02 15:04:05")
//until := time.Unix(int64(s.endTimestamp), 0).Format("2006-01-02 15:04:05")
//fmt.Printf("%s: %s - %s, %.0f - %.0f\n", period, since, until, startValue, s.endValue)
// until - это endTimestamp всегда
bin.PutUint32(s.arr[0:], uint32(s.currentPeriod.Unix()))
bin.PutUint32(s.arr[4:], start)
bin.PutUint32(s.arr[8:], s.endTimestamp)
bin.PutFloat64(s.arr[12:], startValue)
bin.PutFloat64(s.arr[20:], s.endValue)
}
func (s *CumulativePeriodsWriter) Close() error {
if s.endTimestamp > 0 {
if s.lastTimestamp != s.endTimestamp {
s.packPeriod(s.lastTimestamp, s.lastValue)
} else {
s.packPeriod(s.endTimestamp, s.endValue)
}
s.responder.AppendRecord(s.arr)
}
return s.responder.Flush()
}

147
transform/raw.go Normal file
View File

@@ -0,0 +1,147 @@
package transform
import (
"io"
"gordenko.dev/dima/diploma/bin"
)
// CURRENT VALUE WRITER
type CurrentValue struct {
MetricID uint32
Timestamp uint32
Value float64
}
type CurrentValueWriter struct {
arr []byte
responder *ChunkedResponder
}
func NewCurrentValueWriter(dst io.Writer) *CurrentValueWriter {
return &CurrentValueWriter{
arr: make([]byte, 16),
responder: NewChunkedResponder(dst),
}
}
func (s *CurrentValueWriter) BufferValue(m CurrentValue) {
bin.PutUint32(s.arr[0:], m.MetricID)
bin.PutUint32(s.arr[4:], m.Timestamp)
bin.PutFloat64(s.arr[8:], m.Value)
s.responder.BufferRecord(s.arr)
}
func (s *CurrentValueWriter) Close() error {
return s.responder.Flush()
}
// INSTANT MEASURE WRITER
type InstantMeasure struct {
Timestamp uint32
Value float64
}
type InstantMeasureWriter struct {
arr []byte
responder *ChunkedResponder
since uint32
}
func NewInstantMeasureWriter(dst io.Writer, since uint32) *InstantMeasureWriter {
// 12 - это timestamp, value
return &InstantMeasureWriter{
arr: make([]byte, 12),
responder: NewChunkedResponder(dst),
since: since,
}
}
func (s *InstantMeasureWriter) Feed(timestamp uint32, value float64) {
s.feed(timestamp, value, false)
}
func (s *InstantMeasureWriter) FeedNoSend(timestamp uint32, value float64) {
s.feed(timestamp, value, true)
}
func (s *InstantMeasureWriter) feed(timestamp uint32, value float64, isBuffer bool) {
if timestamp < s.since {
return
}
bin.PutUint32(s.arr[0:], timestamp)
bin.PutFloat64(s.arr[4:], value)
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
}
func (s *InstantMeasureWriter) Close() error {
return s.responder.Flush()
}
// CUMULATIVE MEASURE WRITER
type CumulativeMeasure struct {
Timestamp uint32
Value float64
Total float64
}
type CumulativeMeasureWriter struct {
arr []byte
responder *ChunkedResponder
since uint32
endTimestamp uint32
endValue float64
}
func NewCumulativeMeasureWriter(dst io.Writer, since uint32) *CumulativeMeasureWriter {
// 20 - это timestamp, value, total
return &CumulativeMeasureWriter{
arr: make([]byte, 20),
responder: NewChunkedResponder(dst),
since: since,
}
}
func (s *CumulativeMeasureWriter) Feed(timestamp uint32, value float64) {
s.feed(timestamp, value, false)
}
func (s *CumulativeMeasureWriter) FeedNoSend(timestamp uint32, value float64) {
s.feed(timestamp, value, true)
}
func (s *CumulativeMeasureWriter) feed(timestamp uint32, value float64, isBuffer bool) {
if s.endTimestamp > 0 {
s.pack(s.endValue - value)
if isBuffer {
s.responder.BufferRecord(s.arr)
} else {
s.responder.AppendRecord(s.arr)
}
}
s.endTimestamp = timestamp
s.endValue = value
}
func (s *CumulativeMeasureWriter) pack(total float64) {
bin.PutUint32(s.arr[0:], s.endTimestamp)
bin.PutFloat64(s.arr[4:], s.endValue)
bin.PutFloat64(s.arr[12:], total)
}
func (s *CumulativeMeasureWriter) Close() error {
if s.endTimestamp >= s.since {
// endTimestamp внутри заданного периода. Других показаний нет,
// поэтому время добавляю, но накопленную сумму ставлю 0.
s.pack(0)
// Если < since - ничего делать не нужно, ибо накопленная сумма уже добавлена
}
return s.responder.Flush()
}

105
transform/responder.go Normal file
View File

@@ -0,0 +1,105 @@
package transform
import (
"bytes"
"fmt"
"io"
"gordenko.dev/dima/diploma/bin"
"gordenko.dev/dima/diploma/proto"
)
// CHUNKED RESPONDER
var endMsg = []byte{
proto.RespEndOfValue, // end of stream
}
type ChunkedResponder struct {
recordsQty int
buf *bytes.Buffer
dst io.Writer
}
func NewChunkedResponder(dst io.Writer) *ChunkedResponder {
s := &ChunkedResponder{
recordsQty: 0,
buf: bytes.NewBuffer(nil),
dst: dst,
}
s.buf.Write([]byte{
proto.RespPartOfValue, // message type
0, 0, 0, 0, // records qty
})
return s
}
func (s *ChunkedResponder) BufferRecord(rec []byte) {
s.buf.Write(rec)
s.recordsQty++
}
func (s *ChunkedResponder) AppendRecord(rec []byte) error {
s.buf.Write(rec)
s.recordsQty++
if s.buf.Len() < 1500 {
return nil
}
if err := s.sendBuffered(); err != nil {
return err
}
s.buf.Write([]byte{
proto.RespPartOfValue, // message type
0, 0, 0, 0, // records qty
})
s.recordsQty = 0
return nil
}
func (s *ChunkedResponder) Flush() error {
if s.recordsQty > 0 {
if err := s.sendBuffered(); err != nil {
return err
}
}
if _, err := s.dst.Write(endMsg); err != nil {
return err
}
//fmt.Printf("sent endMsg %d\n", endMsg)
return nil
}
func (s *ChunkedResponder) sendBuffered() (err error) {
msg := s.buf.Bytes()
bin.PutUint32(msg[1:], uint32(s.recordsQty))
//fmt.Printf("put uint16: %d\n", msg[:3])
//fmt.Printf("send %d records\n", s.recordsQty)
//fmt.Printf("send buffered: %d, qty: %d\n", msg, s.recordsQty)
n, err := s.dst.Write(msg)
if err != nil {
return
}
if n != len(msg) {
return fmt.Errorf("incomplete write %d bytes instead of %d", n, len(msg))
}
s.buf.Reset()
return
}
// Для Aggregation пишем функцию определения периода и пуляем фактические периоды
//
// By default net/http.Server uses 4KB buffers, which are flushed to client with chunked responses.
// These buffers may result in visible overhead for responses exceeding a few megabytes.
// So allocate 64Kb buffers.
// bw: bufio.NewWriterSize(w, 64*1024),

19
transform/transform.go Normal file
View File

@@ -0,0 +1,19 @@
package transform
import (
"time"
"gordenko.dev/dima/diploma/timeutil"
)
func groupByHour(timestamp uint32) time.Time {
return timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "h")
}
func groupByDay(timestamp uint32) time.Time {
return timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "d")
}
func groupByMonth(timestamp uint32) time.Time {
return timeutil.FirstSecondInPeriod(time.Unix(int64(timestamp), 0), "m")
}