innov
/
nightingale1
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

forward samples in sequence

This commit is contained in:
Ulric Qin 2022-06-10 14:20:18 +08:00
parent a8c60c9f2b
commit ea2249c30c
2 changed files with 20 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
etc/server.conf
View File

@ -180,8 +180,8 @@ BasicAuthUser = ""
# Basic auth password # Basic auth password
BasicAuthPass = "" BasicAuthPass = ""
# timeout settings, unit: ms # timeout settings, unit: ms
Timeout = 30000 Timeout = 10000
DialTimeout = 10000 DialTimeout = 3000
TLSHandshakeTimeout = 30000 TLSHandshakeTimeout = 30000
ExpectContinueTimeout = 1000 ExpectContinueTimeout = 1000
IdleConnTimeout = 90000 IdleConnTimeout = 90000

18
src/server/writer/writer.go
View File

@ -7,6 +7,7 @@ import (
"hash/crc32" "hash/crc32"
"net" "net"
"net/http" "net/http"
"sync"
"time" "time"
"github.com/didi/nightingale/v5/src/server/config" "github.com/didi/nightingale/v5/src/server/config"
@ -143,8 +144,23 @@ func (ws *WritersType) StartConsumer(index int, ch chan *prompb.TimeSeries) {
// @Author: quzhihao // @Author: quzhihao
func (ws *WritersType) post(index int, series []*prompb.TimeSeries) { func (ws *WritersType) post(index int, series []*prompb.TimeSeries) {
header := map[string]string{"hash": fmt.Sprintf("%s-%d", config.C.Heartbeat.Endpoint, index)} header := map[string]string{"hash": fmt.Sprintf("%s-%d", config.C.Heartbeat.Endpoint, index)}
if len(ws.backends) == 1 {
for key := range ws.backends { for key := range ws.backends {
go ws.backends[key].Write(series, header) ws.backends[key].Write(series, header)
}
return
}
if len(ws.backends) > 1 {
wg := new(sync.WaitGroup)
for key := range ws.backends {
wg.Add(1)
go func(wg *sync.WaitGroup, backend WriterType, items []*prompb.TimeSeries, header map[string]string) {
defer wg.Done()
backend.Write(series, header)
}(wg, ws.backends[key], series, header)
}
wg.Wait()
} }
} }