waku-org
/
nwaku
mirror of https://github.com/waku-org/nwaku.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(store): add query and insertion time metrics

This commit is contained in:
Lorenzo Delgado 2022-09-13 16:48:33 +02:00 committed by Lorenzo Delgado
parent 728e298dd7
commit a4860d92a0
1 changed files with 47 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
waku/v2/protocol/waku_store/protocol.nim
View File

@ -31,6 +31,8 @@ declarePublicGauge waku_store_messages, "number of historical messages", ["type"
declarePublicGauge waku_store_peers, "number of store peers" declarePublicGauge waku_store_peers, "number of store peers"
declarePublicGauge waku_store_errors, "number of store protocol errors", ["type"] declarePublicGauge waku_store_errors, "number of store protocol errors", ["type"]
declarePublicGauge waku_store_queries, "number of store queries received" declarePublicGauge waku_store_queries, "number of store queries received"
declarePublicHistogram waku_store_insert_time, "time spent storing a message (ms)"
declarePublicHistogram waku_store_query_time, "time spent processing a history query (ms)"
logScope: logScope:
topics = "wakustore" topics = "wakustore"
@ -91,8 +93,12 @@ proc findMessages(w: WakuStore, query: HistoryQuery): HistoryResponse {.gcsafe.}
else: none(Timestamp) else: none(Timestamp)
qMaxPageSize = query.pagingInfo.pageSize qMaxPageSize = query.pagingInfo.pageSize
qAscendingOrder = query.pagingInfo.direction == PagingDirection.FORWARD qAscendingOrder = query.pagingInfo.direction == PagingDirection.FORWARD
let queryStartTime = getTime().toUnixFloat()
let queryRes = block: let queryRes = block:
# TODO: Move this logic, together with the insert message logic and load messages on boot
# into a "dual-store" message store implementation.
if w.isSqliteOnly: if w.isSqliteOnly:
w.store.getMessagesByHistoryQuery( w.store.getMessagesByHistoryQuery(
contentTopic = qContentTopics, contentTopic = qContentTopics,
@ -113,7 +119,11 @@ proc findMessages(w: WakuStore, query: HistoryQuery): HistoryResponse {.gcsafe.}
maxPageSize = qMaxPageSize, maxPageSize = qMaxPageSize,
ascendingOrder = qAscendingOrder ascendingOrder = qAscendingOrder
) )
let queryTime = getTime().toUnixFloat() - queryStartTime
waku_store_query_time.observe(getMillisecondTime(queryTime))
# Build response # Build response
# TODO: Improve error reporting # TODO: Improve error reporting
if queryRes.isErr(): if queryRes.isErr():
@ -215,43 +225,51 @@ proc handleMessage*(w: WakuStore, pubsubTopic: string, msg: WakuMessage) {.async
if msg.ephemeral: if msg.ephemeral:
# The message is ephemeral, should not be stored # The message is ephemeral, should not be stored
return return
let insertStartTime = getTime().toUnixFloat()
let now = getNanosecondTime(getTime().toUnixFloat()) let now = getNanosecondTime(getTime().toUnixFloat())
let index = Index.compute(msg, receivedTime=now, pubsubTopic=pubsubTopic) let index = Index.compute(msg, receivedTime=now, pubsubTopic=pubsubTopic)
trace "handling message", topic=pubsubTopic, index=index trace "handling message", topic=pubsubTopic, index=index
if w.isSqliteOnly: block:
# Add messages to persistent store, if present if w.isSqliteOnly:
if w.store.isNil(): # Add messages to persistent store, if present
return if w.store.isNil():
return
let resPutStore = w.store.put(index, msg, pubsubTopic) let resPutStore = w.store.put(index, msg, pubsubTopic)
if resPutStore.isErr(): if resPutStore.isErr():
debug "failed to insert message to persistent store", index=index, err=resPutStore.error() debug "failed to insert message to persistent store", index=index, err=resPutStore.error()
waku_store_errors.inc(labelValues = [insertFailure]) waku_store_errors.inc(labelValues = [insertFailure])
waku_store_messages.set(w.store.getMessagesCount(), labelValues = ["stored"])
# TODO: Move this logic, together with the load from persistent store on init waku_store_messages.set(w.store.getMessagesCount(), labelValues = ["stored"])
# into a "dual-store" message store implementation.
else:
# Add message to in-memory store
let resPutInmemory = w.messages.put(index, msg, pubsubTopic)
if resPutInmemory.isErr():
debug "failed to insert message to in-memory store", index=index, err=resPutInmemory.error()
waku_store_errors.inc(labelValues = [insertFailure])
return
waku_store_messages.set(w.messages.getMessagesCount(), labelValues = ["stored"])
# Add messages to persistent store, if present # TODO: Move this logic, together with the load from persistent store on init
if w.store.isNil(): # into a "dual-store" message store implementation.
return else:
# Add message to in-memory store
let resPutStore = w.store.put(index, msg, pubsubTopic) let resPutInmemory = w.messages.put(index, msg, pubsubTopic)
if resPutStore.isErr(): if resPutInmemory.isErr():
debug "failed to insert message to persistent store", index=index, err=resPutStore.error() debug "failed to insert message to in-memory store", index=index, err=resPutInmemory.error()
waku_store_errors.inc(labelValues = [insertFailure]) waku_store_errors.inc(labelValues = [insertFailure])
return
waku_store_messages.set(w.messages.getMessagesCount(), labelValues = ["stored"])
# Add messages to persistent store, if present
if w.store.isNil():
return
let resPutStore = w.store.put(index, msg, pubsubTopic)
if resPutStore.isErr():
debug "failed to insert message to persistent store", index=index, err=resPutStore.error()
waku_store_errors.inc(labelValues = [insertFailure])
return
let insertTime = getTime().toUnixFloat() - insertStartTime
waku_store_insert_time.observe(getMillisecondTime(insertTime))
# TODO: Remove after converting the query method into a non-callback method # TODO: Remove after converting the query method into a non-callback method