14
14
package main
15
15
16
16
import (
17
+ "flag"
17
18
"fmt"
18
19
"io"
19
20
"log"
@@ -98,10 +99,8 @@ type BucketConfig struct {
98
99
MaxTime int64 `yaml:"maxTime"`
99
100
}
100
101
101
- type configState struct {
102
+ type bucketState struct {
102
103
bucketConfig * BucketConfig
103
- Err error
104
- absoluteTime int64
105
104
}
106
105
107
106
func NewQuerier (groupID int , target , prNumber string , qg QueryGroup ) * Querier {
@@ -133,14 +132,16 @@ func NewQuerier(groupID int, target, prNumber string, qg QueryGroup) *Querier {
133
132
}
134
133
135
134
// Function to load `minTime` and `maxTime` from bucket-config.yml
136
- func loadKeyConfig () (* BucketConfig , error ) {
137
- filePath := "/config/bucket-config.yml"
138
- _ , err := os .Stat (filePath )
135
+ func loadBucketConfig () (* BucketConfig , error ) {
136
+ filePath := flag .String ("bucketconfig-file" , "/config/bucket-config.yml" , "Path to the bucket configuration file" )
137
+ flag .Parse ()
138
+
139
+ _ , err := os .Stat (* filePath )
139
140
if os .IsNotExist (err ) {
140
- return nil , fmt .Errorf ("file not found: %s" , filePath )
141
+ return nil , fmt .Errorf ("file not found: %s" , * filePath )
141
142
}
142
143
143
- data , err := os .ReadFile (filePath )
144
+ data , err := os .ReadFile (* filePath )
144
145
if err != nil {
145
146
return nil , fmt .Errorf ("error reading file: %w" , err )
146
147
}
@@ -154,35 +155,37 @@ func loadKeyConfig() (*BucketConfig, error) {
154
155
return & bucketConfig , nil
155
156
}
156
157
157
- func configstate (v * BucketConfig , err error ) * configState {
158
- var absolutetime int64
159
- if v != nil {
160
- absolutetime = v . MaxTime
158
+ func setconfig (v * BucketConfig , err error ) * bucketState {
159
+ // If there is an error in reading bucket-config.yml file then just return nil.
160
+ if err != nil {
161
+ return nil
161
162
}
162
- return & configState {
163
+ return & bucketState {
163
164
bucketConfig : v ,
164
- Err : err ,
165
- absoluteTime : absolutetime ,
166
165
}
167
166
}
168
167
169
- func (q * Querier ) run (wg * sync.WaitGroup , timeBound * configState ) {
168
+ func (q * Querier ) run (wg * sync.WaitGroup , timeBound * bucketState ) {
170
169
defer wg .Done ()
171
170
fmt .Printf ("Running querier %s %s for %s\n " , q .target , q .name , q .url )
172
171
time .Sleep (20 * time .Second )
173
172
174
173
for {
175
174
start := time .Now ()
175
+ // If timeBound is not nil, both the "absolute" and "current" blocks will run;
176
+ // otherwise, only the "current" block will execute. This execution pattern is used
177
+ // because if Downloaded block data is present, both the head block and downloaded block
178
+ // need to be processed.
176
179
runBlockMode := "current"
177
180
for _ , query := range q .queries {
178
181
if runBlockMode == "current" {
179
182
q .query (query .Expr , "current" , nil )
180
- } else if timeBound . Err == nil {
183
+ } else if runBlockMode == "absolute" {
181
184
q .query (query .Expr , "absolute" , timeBound )
182
185
}
183
- if runBlockMode == "current" && timeBound . Err = = nil {
186
+ if runBlockMode == "current" && timeBound ! = nil {
184
187
runBlockMode = "absolute"
185
- } else if timeBound . Err = = nil {
188
+ } else if timeBound ! = nil {
186
189
runBlockMode = "current"
187
190
}
188
191
}
@@ -194,7 +197,7 @@ func (q *Querier) run(wg *sync.WaitGroup, timeBound *configState) {
194
197
}
195
198
}
196
199
197
- func (q * Querier ) query (expr string , timeMode string , timeBound * configState ) {
200
+ func (q * Querier ) query (expr string , timeMode string , timeBound * bucketState ) {
198
201
queryCount .WithLabelValues (q .target , q .name , expr , q .qtype ).Inc ()
199
202
start := time .Now ()
200
203
@@ -219,7 +222,7 @@ func (q *Querier) query(expr string, timeMode string, timeBound *configState) {
219
222
qParams .Set ("step" , q .step )
220
223
}
221
224
} else if timeMode == "absolute" {
222
- blockinstime := time .Unix (0 , timeBound .absoluteTime * int64 (time .Millisecond ))
225
+ blockinstime := time .Unix (0 , timeBound .bucketConfig . MaxTime * int64 (time .Millisecond ))
223
226
qParams .Set ("time" , fmt .Sprintf ("%d" , int64 (blockinstime .Unix ())))
224
227
}
225
228
req .URL .RawQuery = qParams .Encode ()
@@ -268,9 +271,13 @@ func main() {
268
271
}
269
272
prNumber := os .Args [2 ]
270
273
271
- configFile , err := os .ReadFile ("/etc/loadgen/config.yaml" )
274
+ configPath := flag .String ("config-file" , "/etc/loadgen/config.yaml" , "Path to the configuration file" )
275
+ flag .Parse ()
276
+
277
+ configFile , err := os .ReadFile (* configPath )
272
278
if err != nil {
273
- log .Fatalf ("Failed to load config: %v" , err )
279
+ fmt .Printf ("Error reading config file: %v\n " , err )
280
+ return
274
281
}
275
282
276
283
var config struct {
@@ -286,11 +293,8 @@ func main() {
286
293
287
294
var wg sync.WaitGroup
288
295
289
- bucketConfig , err := loadKeyConfig ()
290
- if err != nil {
291
- fmt .Printf ("bucket-config.yml file is not present: %v\n " , err )
292
- }
293
- timeBound := configstate (bucketConfig , err )
296
+ bucketConfig , err := loadBucketConfig ()
297
+ timeBound := setconfig (bucketConfig , err )
294
298
295
299
for i , group := range config .Querier .Groups {
296
300
wg .Add (1 )
0 commit comments