Skip to content

Commit 4b09f5d

Browse files
realdatadrivenrealdatadriven
committed
add timezone support
1 parent d9436c6 commit 4b09f5d

12 files changed

Lines changed: 300 additions & 299 deletions

‎cmd/main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,7 @@ func main() {
3434
flag.Parse()
3535
config := make(map[string]any)
3636
// Parse the file content
37-
etlxlib := &etlx.ETLX{Config: config, Params: map[string]any{}}
37+
etlxlib := &etlx.ETLX{Config: config, Params: map[string]any{}, TimeZone: time.Local}
3838
etlxlib.MetadataOrder = true
3939
err := etlxlib.ConfigFromFile(*filePath)
4040
if err != nil {

‎internal/etlx/data_quality.go‎

Lines changed: 38 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ func (etlx *ETLX) DataQualityCheck(dbConn db.DBInterface, query any, item map[st
8080
if err != nil {
8181
_log2["success"] = false
8282
_log2["msg"] = fmt.Sprintf("%s", err)
83-
_log2["end_at"] = time.Now()
83+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
8484
_log2["nrows"] = 0
8585
return _log2
8686
}
@@ -94,16 +94,16 @@ func (etlx *ETLX) DataQualityCheck(dbConn db.DBInterface, query any, item map[st
9494
}
9595
if okConf && nRows != nil {
9696
_log2["success"] = true
97-
_log2["end_at"] = time.Now()
97+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
9898
_log2["nrows"] = nRows
9999
} else {
100100
_log2["success"] = false
101101
_log2["msg"] = fmt.Sprintf("failed to get md conf string query: %s column %s", query, column)
102-
_log2["end_at"] = time.Now()
102+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
103103
}
104104
} else {
105105
_log2["success"] = true
106-
_log2["end_at"] = time.Now()
106+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
107107
_log2["nrows"] = 0
108108
}
109109
return _log2
@@ -115,11 +115,11 @@ func (etlx *ETLX) DataQualityFix(dbConn db.DBInterface, query any, item map[stri
115115
if err != nil {
116116
_log2["success"] = false
117117
_log2["msg_fix"] = fmt.Sprintf("failed: %s", err)
118-
_log2["end_at"] = time.Now()
118+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
119119
} else {
120120
_log2["success"] = true
121121
_log2["nrows_fixed"] = rowsAffected
122-
_log2["end_at"] = time.Now()
122+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
123123
}
124124
return _log2
125125
}
@@ -132,7 +132,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
132132
}
133133
//fmt.Println(key, dateRef)
134134
var processLogs []map[string]any
135-
start := time.Now()
135+
start := time.Now().In(etlx.TimeZone)
136136
mem_alloc, mem_total_alloc, mem_sys, num_gc := etlx.RuntimeMemStats()
137137
processLogs = append(processLogs, map[string]any{
138138
"process": process,
@@ -155,8 +155,8 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
155155
"process": process,
156156
"name": fmt.Sprintf("KEY %s", key),
157157
"description": metadata["description"].(string),
158-
"key": key, "item_key": itemKey, "start_at": time.Now(),
159-
"end_at": time.Now(),
158+
"key": key, "item_key": itemKey, "start_at": time.Now().In(etlx.TimeZone),
159+
"end_at": time.Now().In(etlx.TimeZone),
160160
"success": true,
161161
"msg": "Deactivated",
162162
})
@@ -171,8 +171,8 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
171171
"process": process,
172172
"name": fmt.Sprintf("%s->%s", key, itemKey),
173173
"description": itemMetadata["description"].(string),
174-
"key": key, "item_key": itemKey, "start_at": time.Now(),
175-
"end_at": time.Now(),
174+
"key": key, "item_key": itemKey, "start_at": time.Now().In(etlx.TimeZone),
175+
"end_at": time.Now().In(etlx.TimeZone),
176176
"success": true,
177177
"msg": "Missing metadata in item",
178178
})
@@ -185,8 +185,8 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
185185
"process": process,
186186
"name": fmt.Sprintf("%s->%s", key, itemKey),
187187
"description": itemMetadata["description"].(string),
188-
"key": key, "item_key": itemKey, "start_at": time.Now(),
189-
"end_at": time.Now(),
188+
"key": key, "item_key": itemKey, "start_at": time.Now().In(etlx.TimeZone),
189+
"end_at": time.Now().In(etlx.TimeZone),
190190
"success": true,
191191
"msg": "Deactivated",
192192
})
@@ -202,8 +202,8 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
202202
"process": process,
203203
"name": fmt.Sprintf("%s->%s", key, itemKey),
204204
"description": itemMetadata["description"].(string),
205-
"key": key, "item_key": itemKey, "start_at": time.Now(),
206-
"end_at": time.Now(),
205+
"key": key, "item_key": itemKey, "start_at": time.Now().In(etlx.TimeZone),
206+
"end_at": time.Now().In(etlx.TimeZone),
207207
"success": true,
208208
"msg": "Excluded from the process",
209209
})
@@ -218,8 +218,8 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
218218
"process": process,
219219
"name": fmt.Sprintf("%s->%s", key, itemKey),
220220
"description": itemMetadata["description"].(string),
221-
"key": key, "item_key": itemKey, "start_at": time.Now(),
222-
"end_at": time.Now(),
221+
"key": key, "item_key": itemKey, "start_at": time.Now().In(etlx.TimeZone),
222+
"end_at": time.Now().In(etlx.TimeZone),
223223
"success": true,
224224
"msg": "Excluded from the process",
225225
})
@@ -252,7 +252,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
252252
if processLogs[0]["ref"] == nil {
253253
processLogs[0]["ref"] = dtRef
254254
}
255-
start3 := time.Now()
255+
start3 := time.Now().In(etlx.TimeZone)
256256
mem_alloc, mem_total_alloc, mem_sys, num_gc := etlx.RuntimeMemStats()
257257
_log2 := map[string]any{
258258
"process": process,
@@ -277,7 +277,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
277277
if err != nil {
278278
_log2["success"] = false
279279
_log2["msg"] = fmt.Sprintf("%s -> %s ERR: connecting to %s in : %s", key, itemKey, conn, err)
280-
_log2["end_at"] = time.Now()
280+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
281281
_log2["duration"] = time.Since(start3).Seconds()
282282
_log2["mem_alloc_end"] = mem_alloc
283283
_log2["mem_total_alloc_end"] = mem_total_alloc
@@ -289,7 +289,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
289289
defer dbConn.Close()
290290
_log2["success"] = true
291291
_log2["msg"] = fmt.Sprintf("%s -> %s CONN: Connectinon to %s successfull", key, itemKey, conn)
292-
_log2["end_at"] = time.Now()
292+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
293293
_log2["duration"] = time.Since(start3).Seconds()
294294
_log2["mem_alloc_end"] = mem_alloc
295295
_log2["mem_total_alloc_end"] = mem_total_alloc
@@ -298,7 +298,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
298298
processLogs = append(processLogs, _log2)
299299
// QUERIES TO RUN AT BEGINING
300300
if okBefore {
301-
start3 := time.Now()
301+
start3 := time.Now().In(etlx.TimeZone)
302302
mem_alloc, mem_total_alloc, mem_sys, num_gc = etlx.RuntimeMemStats()
303303
_log2 := map[string]any{
304304
"process": process,
@@ -316,12 +316,12 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
316316
if err != nil {
317317
_log2["success"] = false
318318
_log2["msg"] = fmt.Sprintf("%s -> %s Before error: %s", key, itemKey, err)
319-
_log2["end_at"] = time.Now()
319+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
320320
_log2["duration"] = time.Since(start3).Seconds()
321321
} else {
322322
_log2["success"] = true
323323
_log2["msg"] = fmt.Sprintf("%s -> %s Before ", key, itemKey)
324-
_log2["end_at"] = time.Now()
324+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
325325
_log2["duration"] = time.Since(start3).Seconds()
326326
}
327327
_log2["mem_alloc_end"] = mem_alloc
@@ -340,7 +340,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
340340
if err != nil {
341341
_log2["success"] = false
342342
_log2["msg"] = fmt.Sprintf("%s -> %s COND: failed %s", key, itemKey, err)
343-
_log2["end_at"] = time.Now()
343+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
344344
_log2["duration"] = time.Since(start3).Seconds()
345345
_log2["mem_alloc_end"] = mem_alloc
346346
_log2["mem_total_alloc_end"] = mem_total_alloc
@@ -351,7 +351,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
351351
} else if !cond {
352352
_log2["success"] = false
353353
_log2["msg"] = fmt.Sprintf("%s -> %s COND: failed the condition %s was not met!", key, itemKey, condition)
354-
_log2["end_at"] = time.Now()
354+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
355355
_log2["duration"] = time.Since(start3).Seconds()
356356
_log2["mem_alloc_end"] = mem_alloc
357357
_log2["mem_total_alloc_end"] = mem_total_alloc
@@ -366,7 +366,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
366366
}
367367
// MAIN QUERY
368368
mem_alloc, mem_total_alloc, mem_sys, num_gc = etlx.RuntimeMemStats()
369-
_log2["start_at"] = time.Now()
369+
_log2["start_at"] = time.Now().In(etlx.TimeZone)
370370
_log2["mem_alloc_start"] = mem_alloc
371371
_log2["mem_total_alloc_start"] = mem_total_alloc
372372
_log2["mem_sys_start"] = mem_sys
@@ -378,13 +378,13 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
378378
if !res["success"].(bool) {
379379
_log2["success"] = res["success"]
380380
_log2["msg"] = res["msg"]
381-
_log2["end_at"] = time.Now()
381+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
382382
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
383383
} else {
384384
_log2["success"] = res["success"]
385385
_log2["msg"] = fmt.Sprintf("%s -> %s CHECK: successfull", key, itemKey)
386386
_log2["nrows"] = res["nrows"]
387-
_log2["end_at"] = time.Now()
387+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
388388
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
389389
}
390390
_log2["mem_alloc_end"] = mem_alloc
@@ -399,13 +399,13 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
399399
if !res["success"].(bool) {
400400
_log2["success"] = res["success"]
401401
_log2["msg"] = res["msg_fix"]
402-
_log2["end_at"] = time.Now()
402+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
403403
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
404404
} else {
405405
_log2["success"] = res["success"]
406406
_log2["msg"] = fmt.Sprintf("%s -> %s FIX: successfull", key, itemKey)
407407
_log2["nrows_fixed"] = res["nrows_fixed"]
408-
_log2["end_at"] = time.Now()
408+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
409409
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
410410
}
411411
_log2["mem_alloc_end"] = mem_alloc
@@ -420,13 +420,13 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
420420
if !res["success"].(bool) {
421421
_log2["success"] = res["success"]
422422
_log2["msg"] = res["msg"]
423-
_log2["end_at"] = time.Now()
423+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
424424
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
425425
} else {
426426
_log2["success"] = res["success"]
427427
_log2["msg"] = fmt.Sprintf("%s -> %s successfull", key, itemKey)
428428
_log2["nrows"] = res["nrows"]
429-
_log2["end_at"] = time.Now()
429+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
430430
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
431431
_nrows, okNrows := res["nrows"].(int64)
432432
//fmt.Println("RES NROWS:", res["nrows"], "PROC NROWS:", _nrows)
@@ -436,13 +436,13 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
436436
if !res["success"].(bool) {
437437
_log2["success_fix"] = res["success"]
438438
_log2["msg_fix"] = res["msg_fix"]
439-
_log2["end_at"] = time.Now()
439+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
440440
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
441441
} else {
442442
_log2["success_fix"] = res["success"]
443443
_log2["msg_fix"] = res["msg_fix"]
444444
_log2["nrows_fixed"] = res["nrows_fixed"]
445-
_log2["end_at"] = time.Now()
445+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
446446
_log2["duration"] = time.Since(_log2["start_at"].(time.Time)).Seconds()
447447
}
448448
}
@@ -456,7 +456,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
456456
//fmt.Println(_log2)
457457
// QUERIES TO RUN AT THE END
458458
if okAfter {
459-
start3 := time.Now()
459+
start3 := time.Now().In(etlx.TimeZone)
460460
mem_alloc, mem_total_alloc, mem_sys, num_gc = etlx.RuntimeMemStats()
461461
_log2 := map[string]any{
462462
"process": process,
@@ -474,12 +474,12 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
474474
if err != nil {
475475
_log2["success"] = false
476476
_log2["msg"] = fmt.Sprintf("%s -> %s After error: %s", key, itemKey, err)
477-
_log2["end_at"] = time.Now()
477+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
478478
_log2["duration"] = time.Since(start3).Seconds()
479479
} else {
480480
_log2["success"] = true
481481
_log2["msg"] = fmt.Sprintf("%s -> %s After ", key, itemKey)
482-
_log2["end_at"] = time.Now()
482+
_log2["end_at"] = time.Now().In(etlx.TimeZone)
483483
_log2["duration"] = time.Since(start3).Seconds()
484484
}
485485
_log2["mem_alloc_end"] = mem_alloc
@@ -511,7 +511,7 @@ func (etlx *ETLX) RunDATA_QUALITY(dateRef []time.Time, conf map[string]any, extr
511511
"mem_total_alloc_start": processLogs[0]["mem_total_alloc_start"],
512512
"mem_sys_start": processLogs[0]["mem_sys_start"],
513513
"num_gc_start": processLogs[0]["num_gc_start"],
514-
"end_at": time.Now(),
514+
"end_at": time.Now().In(etlx.TimeZone),
515515
"duration": time.Since(start).Seconds(),
516516
"mem_alloc_end": mem_alloc,
517517
"mem_total_alloc_end": mem_total_alloc,

‎internal/etlx/etlxlib.go‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ type ETLX struct {
2727
MD string
2828
autoLogsDisabled bool
2929
MetadataOrder bool
30+
TimeZone *time.Location
3031
}
3132

3233
func addAutoLoggs(md string) string {
@@ -622,7 +623,7 @@ func (etlx *ETLX) ReplaceQueryStringDate(query string, dateRef any) string {
622623
frmtFinal = etlx.GetGODateFormat(frmtFinal)
623624
var procc string
624625
if regexp.MustCompile(`\b(?:STAMP|TSTAMP|TS)\b`).MatchString(m) {
625-
procc = regexp.MustCompile(patt.String()).ReplaceAllString(m, time.Now().Format(frmtFinal))
626+
procc = regexp.MustCompile(patt.String()).ReplaceAllString(m, time.Now().In(etlx.TimeZone).Format(frmtFinal))
626627
//fmt.Println("TIMESTAMP FORMAT", m, frmtFinal, procc)
627628
} else {
628629
if dates, ok := dateRef.([]time.Time); ok {
@@ -651,7 +652,7 @@ func (etlx *ETLX) ReplaceQueryStringDate(query string, dateRef any) string {
651652
frmtFinal = strings.ReplaceAll(frmtFinal, "}", "")
652653
frmtFinal = etlx.GetGODateFormat(frmtFinal)
653654
//fmt.Println(m, frmtFinal)
654-
now := time.Now()
655+
now := time.Now().In(etlx.TimeZone)
655656
var procc string
656657
procc = regexp.MustCompile(patt.String()).ReplaceAllString(m, now.Format(frmtFinal))
657658
patt = regexp.MustCompile(regexp.QuoteMeta(m))
@@ -772,7 +773,7 @@ func (etlx *ETLX) ProcessETL(config map[string]any, runner RunnerFunc) error {
772773
mainConn := metadata["connection"].(string)
773774
description := metadata["description"].(string)
774775
fmt.Printf("Starting ETL process: %s\n", description)
775-
start := time.Now()
776+
start := time.Now().In(etlx.TimeZone)
776777
for key, value := range etl {
777778
if key == "metadata" {
778779
continue
@@ -887,7 +888,7 @@ func (etlx *ETLX) ProcessMDKey(key string, config map[string]any, runner RunnerF
887888
}
888889
// description := metadata["description"].(string)
889890
// fmt.Printf("Starting %s process: %s\n", key, description)
890-
// start := time.Now()
891+
// start := time.Now().In(etlx.TimeZone)
891892
order, okOrder := data["__order"].([]any)
892893
if okOrder {
893894
for _, key2 := range order {

0 commit comments

Comments
 (0)