GitHub-Proxy/signoz
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/SigNoz/signoz synced 2025-08-14 14:35:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

tail function updated to use values instead of pointers

Browse Source
This commit is contained in:
nityanandagohain 2022-07-22 15:44:07 +05:30
parent bdb6901c74
commit 420d46ab01
1 changed files with 14 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
pkg/query-service/app/clickhouseReader/reader.go
View File

@ -2908,15 +2908,14 @@ func (r *ClickHouseReader) TailLogs(ctx context.Context, client *model.LogsTailC
"CAST((resources_string_key, resources_string_value), 'Map(String, String)') as resources_string "+ "CAST((resources_string_key, resources_string_value), 'Map(String, String)') as resources_string "+
"from %s.%s", r.logsDB, r.logsTable) "from %s.%s", r.logsDB, r.logsTable)
currentTime := uint64(time.Now().UnixNano() / int64(time.Millisecond)) tsStart := uint64(time.Now().UnixNano() / int64(time.Millisecond))
tsStart := &currentTime
if client.Filter.TimestampStart != nil { if client.Filter.TimestampStart != nil {
tsStart = client.Filter.TimestampStart tsStart = *client.Filter.TimestampStart
} }
var idStart *string var idStart string
if client.Filter.IdStart != nil { if client.Filter.IdStart != nil {
idStart = client.Filter.IdStart idStart = *client.Filter.IdStart
} }
for { for {
@ -2927,23 +2926,23 @@ func (r *ClickHouseReader) TailLogs(ctx context.Context, client *model.LogsTailC
zap.S().Debug("closing go routine : " + client.Name) zap.S().Debug("closing go routine : " + client.Name)
return return
default: default:
tmpQuery := fmt.Sprintf("%s where timestamp >='%d'", query, *tsStart) tmpQuery := fmt.Sprintf("%s where timestamp >='%d'", query, tsStart)
if filterSql != "" { if filterSql != "" {
tmpQuery += fmt.Sprintf(" and %s", filterSql) tmpQuery += fmt.Sprintf(" and %s", filterSql)
} }
if idStart != nil { if idStart != "" {
tmpQuery += fmt.Sprintf(" and id > '%s'", *idStart) tmpQuery += fmt.Sprintf(" and id > '%s'", idStart)
} }
tmpQuery = fmt.Sprintf("%s order by timestamp asc, id asc limit 1000", tmpQuery) tmpQuery = fmt.Sprintf("%s order by timestamp asc, id asc limit 1000", tmpQuery)
zap.S().Debug(tmpQuery) zap.S().Debug(tmpQuery)
response := &[]model.GetLogsResponse{} response := []model.GetLogsResponse{}
err := r.db.Select(ctx, response, tmpQuery) err := r.db.Select(ctx, &response, tmpQuery)
if err != nil { if err != nil {
zap.S().Debug(err) zap.S().Debug(err)
client.Error <- err client.Error <- err
return return
} }
len := len(*response) len := len(response)
for i := 0; i < len; i++ { for i := 0; i < len; i++ {
select { select {
case <-ctx.Done(): case <-ctx.Done():
@ -2952,16 +2951,15 @@ func (r *ClickHouseReader) TailLogs(ctx context.Context, client *model.LogsTailC
zap.S().Debug("closing go routine while sending logs : " + client.Name) zap.S().Debug("closing go routine while sending logs : " + client.Name)
return return
default: default:
client.Logs <- &(*response)[i] client.Logs <- &response[i]
if i == len-1 { if i == len-1 {
tsStart = &(*response)[i].Timestamp tsStart = response[i].Timestamp
idStart = &(*response)[i].ID idStart = response[i].ID
} }
} }
} }
if len == 0 { if len == 0 {
currentTime := uint64(time.Now().UnixNano() / int64(time.Millisecond)) tsStart = uint64(time.Now().UnixNano() / int64(time.Millisecond))
tsStart = &currentTime
} }
time.Sleep(2 * time.Second) time.Sleep(2 * time.Second)
} }