2022-11-29 22:53:28 +01:00
|
|
|
package main
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"fmt"
|
|
|
|
"log"
|
|
|
|
"sync"
|
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/barpool"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/remoteread"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/stepper"
|
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/app/vmctl/vm"
|
|
|
|
)
|
|
|
|
|
|
|
|
type remoteReadProcessor struct {
|
|
|
|
filter remoteReadFilter
|
|
|
|
|
|
|
|
dst *vm.Importer
|
|
|
|
src *remoteread.Client
|
|
|
|
|
2023-06-02 14:57:08 +02:00
|
|
|
cc int
|
|
|
|
isVerbose bool
|
2022-11-29 22:53:28 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
type remoteReadFilter struct {
|
2023-12-12 13:05:44 +01:00
|
|
|
timeStart *time.Time
|
|
|
|
timeEnd *time.Time
|
|
|
|
chunk string
|
|
|
|
timeReverse bool
|
2022-11-29 22:53:28 +01:00
|
|
|
}
|
|
|
|
|
2023-06-02 14:57:08 +02:00
|
|
|
func (rrp *remoteReadProcessor) run(ctx context.Context) error {
|
2022-11-29 22:53:28 +01:00
|
|
|
rrp.dst.ResetStats()
|
|
|
|
if rrp.filter.timeEnd == nil {
|
|
|
|
t := time.Now().In(rrp.filter.timeStart.Location())
|
|
|
|
rrp.filter.timeEnd = &t
|
|
|
|
}
|
|
|
|
if rrp.cc < 1 {
|
|
|
|
rrp.cc = 1
|
|
|
|
}
|
|
|
|
|
2023-12-12 13:05:44 +01:00
|
|
|
ranges, err := stepper.SplitDateRange(*rrp.filter.timeStart, *rrp.filter.timeEnd, rrp.filter.chunk, rrp.filter.timeReverse)
|
2022-11-29 22:53:28 +01:00
|
|
|
if err != nil {
|
|
|
|
return fmt.Errorf("failed to create date ranges for the given time filters: %v", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
question := fmt.Sprintf("Selected time range %q - %q will be split into %d ranges according to %q step. Continue?",
|
|
|
|
rrp.filter.timeStart.String(), rrp.filter.timeEnd.String(), len(ranges), rrp.filter.chunk)
|
2024-06-10 12:20:52 +02:00
|
|
|
if !prompt(question) {
|
2022-11-29 22:53:28 +01:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2024-06-10 12:20:52 +02:00
|
|
|
bar := barpool.AddWithTemplate(fmt.Sprintf(barTpl, "Processing ranges"), len(ranges))
|
|
|
|
if err := barpool.Start(); err != nil {
|
|
|
|
return err
|
2022-11-29 22:53:28 +01:00
|
|
|
}
|
2024-06-10 12:20:52 +02:00
|
|
|
|
2022-11-29 22:53:28 +01:00
|
|
|
defer func() {
|
2024-06-10 12:20:52 +02:00
|
|
|
barpool.Stop()
|
2022-11-29 22:53:28 +01:00
|
|
|
log.Println("Import finished!")
|
|
|
|
log.Print(rrp.dst.Stats())
|
|
|
|
}()
|
|
|
|
|
|
|
|
rangeC := make(chan *remoteread.Filter)
|
|
|
|
errCh := make(chan error)
|
|
|
|
|
|
|
|
var wg sync.WaitGroup
|
|
|
|
wg.Add(rrp.cc)
|
|
|
|
for i := 0; i < rrp.cc; i++ {
|
|
|
|
go func() {
|
|
|
|
defer wg.Done()
|
|
|
|
for r := range rangeC {
|
|
|
|
if err := rrp.do(ctx, r); err != nil {
|
|
|
|
errCh <- fmt.Errorf("request failed for: %s", err)
|
|
|
|
return
|
|
|
|
}
|
2024-06-10 12:20:52 +02:00
|
|
|
bar.Increment()
|
2022-11-29 22:53:28 +01:00
|
|
|
}
|
|
|
|
}()
|
|
|
|
}
|
|
|
|
|
|
|
|
for _, r := range ranges {
|
|
|
|
select {
|
|
|
|
case infErr := <-errCh:
|
|
|
|
return fmt.Errorf("remote read error: %s", infErr)
|
|
|
|
case vmErr := <-rrp.dst.Errors():
|
2023-06-02 14:57:08 +02:00
|
|
|
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
|
2022-11-29 22:53:28 +01:00
|
|
|
case rangeC <- &remoteread.Filter{
|
|
|
|
StartTimestampMs: r[0].UnixMilli(),
|
|
|
|
EndTimestampMs: r[1].UnixMilli(),
|
|
|
|
}:
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
close(rangeC)
|
|
|
|
wg.Wait()
|
|
|
|
rrp.dst.Close()
|
|
|
|
close(errCh)
|
|
|
|
// drain import errors channel
|
|
|
|
for vmErr := range rrp.dst.Errors() {
|
|
|
|
if vmErr.Err != nil {
|
2023-06-02 14:57:08 +02:00
|
|
|
return fmt.Errorf("import process failed: %s", wrapErr(vmErr, rrp.isVerbose))
|
2022-11-29 22:53:28 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
for err := range errCh {
|
|
|
|
return fmt.Errorf("import process failed: %s", err)
|
|
|
|
}
|
|
|
|
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (rrp *remoteReadProcessor) do(ctx context.Context, filter *remoteread.Filter) error {
|
|
|
|
return rrp.src.Read(ctx, filter, func(series *vm.TimeSeries) error {
|
|
|
|
if err := rrp.dst.Input(series); err != nil {
|
|
|
|
return fmt.Errorf(
|
|
|
|
"failed to read data for time range start: %d, end: %d, %s",
|
|
|
|
filter.StartTimestampMs, filter.EndTimestampMs, err)
|
|
|
|
}
|
|
|
|
return nil
|
|
|
|
})
|
|
|
|
}
|