fix (dot/telemetry): refactor telemetry to reduce CPU usage #1597
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
|
|
@@ -17,38 +17,35 @@ | |||||
| package telemetry | ||||||
|
|
||||||
| import ( | ||||||
| "encoding/json" | ||||||
| "fmt" | ||||||
| "sync" | ||||||
| "time" | ||||||
|
|
||||||
| "github.com/ChainSafe/gossamer/lib/genesis" | ||||||
| "github.com/gorilla/websocket" | ||||||
| ) | ||||||
|
|
||||||
| type telemetryConnection struct { | ||||||
| wsconn *websocket.Conn | ||||||
| verbosity int | ||||||
| } | ||||||
|
|
||||||
| // Message struct to hold telemetry message data | ||||||
| type Message struct { | ||||||
| values map[string]interface{} | ||||||
| } | ||||||
|
|
||||||
| // Handler struct for holding telemetry related things | ||||||
| type Handler struct { | ||||||
| msg chan Message | ||||||
| connections []telemetryConnection | ||||||
| } | ||||||
|
|
||||||
| // KeyValue object to hold key value pairs used in telemetry messages | ||||||
| type KeyValue struct { | ||||||
| key string | ||||||
| value interface{} | ||||||
| } | ||||||
|
|
||||||
| var ( | ||||||
|
|
@@ -57,126 +54,81 @@ var ( | |||||
| ) | ||||||
|
|
||||||
| // GetInstance singleton pattern to for accessing TelemetryHandler | ||||||
| func GetInstance() *Handler { //nolint | ||||||
| if handlerInstance == nil { | ||||||
| once.Do( | ||||||
| func() { | ||||||
| handlerInstance = &Handler{ | ||||||
| msg: make(chan Message, 10), | ||||||
|
Contributor
There was a problem hiding this comment. Increase the buffer size to
Contributor
Author
There was a problem hiding this comment. updateded to 256. |
||||||
| } | ||||||
| go handlerInstance.startListening() | ||||||
| }) | ||||||
| } | ||||||
| return handlerInstance | ||||||
| } | ||||||
|
|
||||||
| // NewTelemetryMessage builds a telemetry message | ||||||
| func NewTelemetryMessage(values ...KeyValue) *Message { //nolint | ||||||
| mvals := make(map[string]interface{}) | ||||||
| for _, v := range values { | ||||||
| mvals[v.key] = v.value | ||||||
| } | ||||||
| return &Message{ | ||||||
| values: mvals, | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| // NewKeyValue builds a key value pair for telemetry messages | ||||||
| func NewKeyValue(key string, value interface{}) KeyValue { //nolint | ||||||
|
Contributor
There was a problem hiding this comment. Make this function return pointer.
Contributor
Author
There was a problem hiding this comment. Made function return pointer. |
||||||
| return KeyValue{ | ||||||
| key: key, | ||||||
| value: value, | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| // AddConnections adds the given telemetry endpoint as listeners that will receive telemetry data | ||||||
| func (t *Handler) AddConnections(conns []*genesis.TelemetryEndpoint) { | ||||||
| for _, v := range conns { | ||||||
| c, _, err := websocket.DefaultDialer.Dial(v.Endpoint, nil) | ||||||
| if err != nil { | ||||||
| // todo (ed) try reconnecting if there is an error connecting | ||||||
| fmt.Printf("Error %v\n", err) | ||||||
| continue | ||||||
| } | ||||||
| tConn := telemetryConnection{ | ||||||
| wsconn: c, | ||||||
| verbosity: v.Verbosity, | ||||||
| } | ||||||
| t.connections = append(t.connections, tConn) | ||||||
| } | ||||||
| } | ||||||
|
|
||||||
| // SendMessage sends Message to connected telemetry listeners | ||||||
| func (t *Handler) SendMessage(msg *Message) { | ||||||
|
Contributor
There was a problem hiding this comment.
Suggested change
|
||||||
| t.msg <- *msg | ||||||
|
Contributor
There was a problem hiding this comment. improvement: add a timeout and error return for this method. I see that that it's a buffered channel of length 256, so this shouldn't happen, but it's a good practice so that this function call isn't blocking.
Contributor
Author
There was a problem hiding this comment. Good suggestion, I've added timeout and return error. |
||||||
| } | ||||||
|
|
||||||
| func (t *Handler) startListening() { | ||||||
| for { | ||||||
| msg := <-t.msg | ||||||
|
Member
There was a problem hiding this comment. In this case, is possible to have a context on
Contributor
There was a problem hiding this comment. agree,
Contributor
Author
There was a problem hiding this comment. A added context to the Handler struct. |
||||||
| for _, v := range t.connections { | ||||||
| err := v.wsconn.WriteMessage(websocket.TextMessage, msgToBytes(msg)) | ||||||
| if err != nil { | ||||||
| // TODO (ed) determine how to handle this error | ||||||
| fmt.Printf("ERROR connecting to telemetry %v\n", err) | ||||||
| } | ||||||
|
Contributor
There was a problem hiding this comment. can you ignore this error for now - I sometimes see a lot of logs of this |
||||||
| } | ||||||
|
Contributor
There was a problem hiding this comment. maybe put this in a goroutine so that reading/writing to/from
Contributor
Author
There was a problem hiding this comment. I tried putting this in a go routine, however then I was getting
Contributor
Author
There was a problem hiding this comment. Disregard above comment, I've moved mutex to Handler struct, where it makes more sense. |
||||||
| } | ||||||
| } | ||||||
|
|
||||||
| func msgToBytes(message Message) []byte { | ||||||
| res := make(map[string]interface{}) | ||||||
|
Contributor
There was a problem hiding this comment. maybe create a helper struct to construct the json response.
Contributor
Author
There was a problem hiding this comment. Good idea, created. |
||||||
| res["id"] = 1 // todo (ed) determine how this is used | ||||||
| res["payload"] = message.values | ||||||
| res["ts"] = time.Now() | ||||||
| resB, err := json.Marshal(res) | ||||||
| if err != nil { | ||||||
| return nil | ||||||
| } | ||||||
| return resB | ||||||
| } | ||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
the block here is the best hash, not the best finalized hash
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@noot I'm confused, in there a difference between finalized hash and best finalized hash? How do I obtain finalized hash?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This gives the finalized header.
I think
finalizedalready gives the latest finalized block.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Also, have a single
NewKeyValueper line.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
yes, nevermind, I read this wrong, it's actually fine