-
Notifications
You must be signed in to change notification settings - Fork 52
/
multi_target.go
133 lines (116 loc) · 3.44 KB
/
multi_target.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
package data
import (
"context"
"fmt"
"net/http"
"sync"
"time"
v3pb "github.com/go-graphite/protocol/carbonapi_v3_pb"
"github.com/lomik/graphite-clickhouse/config"
"github.com/lomik/graphite-clickhouse/helper/errs"
"github.com/lomik/graphite-clickhouse/pkg/alias"
"github.com/lomik/graphite-clickhouse/pkg/scope"
"go.uber.org/zap"
)
// TimeFrame contains information about fetch request time conditions
type TimeFrame struct {
From int64
Until int64
MaxDataPoints int64
}
// MultiTarget is a map of TimeFrame keys and targets slice of strings values
type MultiTarget map[TimeFrame]*Targets
func MFRToMultiTarget(v3Request *v3pb.MultiFetchRequest) MultiTarget {
multiTarget := make(MultiTarget)
if len(v3Request.Metrics) > 0 {
for _, m := range v3Request.Metrics {
tf := TimeFrame{
From: m.StartTime,
Until: m.StopTime,
MaxDataPoints: m.MaxDataPoints,
}
if _, ok := multiTarget[tf]; ok {
target := multiTarget[tf]
target.List = append(multiTarget[tf].List, m.PathExpression)
} else {
multiTarget[tf] = &Targets{List: []string{m.PathExpression}, AM: alias.New()}
}
}
}
return multiTarget
}
func (m *MultiTarget) checkMetricsLimitExceeded(num int) error {
if num <= 0 {
// zero or negative means unlimited
return nil
}
for _, t := range *m {
if num < t.AM.Len() {
return errs.NewErrorWithCode(fmt.Sprintf("metrics limit exceeded: %d < %d", num, t.AM.Len()), http.StatusForbidden)
}
}
return nil
}
func getDataTimeout(cfg *config.Config, m *MultiTarget) time.Duration {
dataTimeout := cfg.ClickHouse.DataTimeout
if len(cfg.ClickHouse.QueryParams) > 1 {
var maxDuration time.Duration
for tf := range *m {
duration := time.Second * time.Duration(tf.Until-tf.From)
if duration >= maxDuration {
maxDuration = duration
}
}
n := config.GetQueryParam(cfg.ClickHouse.QueryParams, maxDuration)
return cfg.ClickHouse.QueryParams[n].DataTimeout
}
return dataTimeout
}
// Fetch fetches the parsed ClickHouse data returns CHResponses
func (m *MultiTarget) Fetch(ctx context.Context, cfg *config.Config, chContext string) (CHResponses, error) {
var lock sync.RWMutex
var wg sync.WaitGroup
logger := scope.Logger(ctx)
setCarbonlinkClient(&cfg.Carbonlink)
err := m.checkMetricsLimitExceeded(cfg.Common.MaxMetricsPerTarget)
if err != nil {
logger.Error("data fetch", zap.Error(err))
return nil, err
}
dataTimeout := getDataTimeout(cfg, m)
ctxTimeout, cancel := context.WithTimeout(ctx, dataTimeout)
defer cancel()
errors := make([]error, 0, len(*m))
query := newQuery(cfg, len(*m))
for tf, targets := range *m {
tf, targets := tf, targets
cond := &conditions{TimeFrame: &tf, Targets: targets, aggregated: cfg.ClickHouse.InternalAggregation}
if cond.MaxDataPoints <= 0 || int64(cfg.ClickHouse.MaxDataPoints) < cond.MaxDataPoints {
cond.MaxDataPoints = int64(cfg.ClickHouse.MaxDataPoints)
}
err := cond.selectDataTable(cfg, cond.TimeFrame, chContext)
if err != nil {
lock.Lock()
errors = append(errors, err)
lock.Unlock()
logger.Error("data tables is not specified", zap.Error(err))
return EmptyResponse(), err
}
wg.Add(1)
go func(cond *conditions) {
defer wg.Done()
err := query.getDataPoints(ctxTimeout, cond)
if err != nil {
lock.Lock()
errors = append(errors, err)
lock.Unlock()
return
}
}(cond)
}
wg.Wait()
for len(errors) != 0 {
return EmptyResponse(), errors[0]
}
return query.CHResponses, nil
}