Unverified Commit fad0b0d2 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Fix error stats and failed command stats for blocked clients (#10309)

This is a followup work for #10278, and a discussion about #10279

The changes:
- fix failed_calls in command stats for blocked clients that got error.
  including CLIENT UNBLOCK, and module replying an error from a thread.
- fix latency stats for XREADGROUP that filed with -NOGROUP

Theory behind which errors should be counted:
- error stats represents errors returned to the user, so an error handled by a
  module should not be counted.
- total error counter should be the same.
- command stats represents execution of commands (even with RM_Call, and if
  they fail or get rejected it counts these calls in commandstats, so it should
  also count failed_calls)

Some thoughts about Scripts:
for scripts it could be different since they're part of user code, not the infra (not an extension to redis)
we certainly want commandstats to contain all calls and errors
a simple script is like mult-exec transaction so an error inside it should be counted in error stats
a script that replies with an error to the user (using redis.error_reply) should also be counted in error stats
but then the problem is that a plain `return redis.call("SET")` should not be counted twice (once for the SET
and once for EVAL)
so that's something left to be resolved in #10279
parent b59bb9b4
...@@ -108,9 +108,11 @@ void blockClient(client *c, int btype) { ...@@ -108,9 +108,11 @@ void blockClient(client *c, int btype) {
/* This function is called after a client has finished a blocking operation /* This function is called after a client has finished a blocking operation
* in order to update the total command duration, log the command into * in order to update the total command duration, log the command into
* the Slow log if needed, and log the reply duration event if needed. */ * the Slow log if needed, and log the reply duration event if needed. */
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us){ void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors){
const ustime_t total_cmd_duration = c->duration + blocked_us + reply_us; const ustime_t total_cmd_duration = c->duration + blocked_us + reply_us;
c->lastcmd->microseconds += total_cmd_duration; c->lastcmd->microseconds += total_cmd_duration;
if (had_errors)
c->lastcmd->failed_calls++;
if (server.latency_tracking_enabled) if (server.latency_tracking_enabled)
updateCommandLatencyHistogram(&(c->lastcmd->latency_histogram), total_cmd_duration*1000); updateCommandLatencyHistogram(&(c->lastcmd->latency_histogram), total_cmd_duration*1000);
/* Log the command into the Slow log if needed. */ /* Log the command into the Slow log if needed. */
...@@ -314,6 +316,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) { ...@@ -314,6 +316,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
* call. */ * call. */
if (dstkey) incrRefCount(dstkey); if (dstkey) incrRefCount(dstkey);
long long prev_error_replies = server.stat_total_error_replies;
client *old_client = server.current_client; client *old_client = server.current_client;
server.current_client = receiver; server.current_client = receiver;
monotime replyTimer; monotime replyTimer;
...@@ -322,7 +325,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) { ...@@ -322,7 +325,7 @@ void serveClientsBlockedOnListKey(robj *o, readyList *rl) {
rl->key, dstkey, rl->db, rl->key, dstkey, rl->db,
wherefrom, whereto, wherefrom, whereto,
&deleted); &deleted);
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer)); updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer), server.stat_total_error_replies != prev_error_replies);
unblockClient(receiver); unblockClient(receiver);
afterCommand(receiver); afterCommand(receiver);
server.current_client = old_client; server.current_client = old_client;
...@@ -366,6 +369,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) { ...@@ -366,6 +369,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
? 1 : 0; ? 1 : 0;
int reply_nil_when_empty = use_nested_array; int reply_nil_when_empty = use_nested_array;
long long prev_error_replies = server.stat_total_error_replies;
client *old_client = server.current_client; client *old_client = server.current_client;
server.current_client = receiver; server.current_client = receiver;
monotime replyTimer; monotime replyTimer;
...@@ -388,7 +392,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) { ...@@ -388,7 +392,7 @@ void serveClientsBlockedOnSortedSetKey(robj *o, readyList *rl) {
decrRefCount(argv[1]); decrRefCount(argv[1]);
if (count != -1) decrRefCount(argv[2]); if (count != -1) decrRefCount(argv[2]);
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer)); updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer), server.stat_total_error_replies != prev_error_replies);
unblockClient(receiver); unblockClient(receiver);
afterCommand(receiver); afterCommand(receiver);
server.current_client = old_client; server.current_client = old_client;
...@@ -421,6 +425,12 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -421,6 +425,12 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
bkinfo *bki = dictFetchValue(receiver->bpop.keys,rl->key); bkinfo *bki = dictFetchValue(receiver->bpop.keys,rl->key);
streamID *gt = &bki->stream_id; streamID *gt = &bki->stream_id;
long long prev_error_replies = server.stat_total_error_replies;
client *old_client = server.current_client;
server.current_client = receiver;
monotime replyTimer;
elapsedStart(&replyTimer);
/* If we blocked in the context of a consumer /* If we blocked in the context of a consumer
* group, we need to resolve the group and update the * group, we need to resolve the group and update the
* last ID the client is blocked for: this is needed * last ID the client is blocked for: this is needed
...@@ -440,8 +450,7 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -440,8 +450,7 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
addReplyError(receiver, addReplyError(receiver,
"-NOGROUP the consumer group this client " "-NOGROUP the consumer group this client "
"was blocked on no longer exists"); "was blocked on no longer exists");
unblockClient(receiver); goto unblock_receiver;
continue;
} else { } else {
*gt = group->last_id; *gt = group->last_id;
} }
...@@ -470,10 +479,6 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -470,10 +479,6 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
} }
} }
client *old_client = server.current_client;
server.current_client = receiver;
monotime replyTimer;
elapsedStart(&replyTimer);
/* Emit the two elements sub-array consisting of /* Emit the two elements sub-array consisting of
* the name of the stream and the data we * the name of the stream and the data we
* extracted from it. Wrapped in a single-item * extracted from it. Wrapped in a single-item
...@@ -493,11 +498,13 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) { ...@@ -493,11 +498,13 @@ void serveClientsBlockedOnStreamKey(robj *o, readyList *rl) {
streamReplyWithRange(receiver,s,&start,NULL, streamReplyWithRange(receiver,s,&start,NULL,
receiver->bpop.xread_count, receiver->bpop.xread_count,
0, group, consumer, noack, &pi); 0, group, consumer, noack, &pi);
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer));
/* Note that after we unblock the client, 'gt' /* Note that after we unblock the client, 'gt'
* and other receiver->bpop stuff are no longer * and other receiver->bpop stuff are no longer
* valid, so we must do the setup above before * valid, so we must do the setup above before
* this call. */ * the unblockClient call. */
unblock_receiver:
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer), server.stat_total_error_replies != prev_error_replies);
unblockClient(receiver); unblockClient(receiver);
afterCommand(receiver); afterCommand(receiver);
server.current_client = old_client; server.current_client = old_client;
...@@ -545,12 +552,13 @@ void serveClientsBlockedOnKeyByModule(readyList *rl) { ...@@ -545,12 +552,13 @@ void serveClientsBlockedOnKeyByModule(readyList *rl) {
* different modules with different triggers to consider if a key * different modules with different triggers to consider if a key
* is ready or not. This means we can't exit the loop but need * is ready or not. This means we can't exit the loop but need
* to continue after the first failure. */ * to continue after the first failure. */
long long prev_error_replies = server.stat_total_error_replies;
client *old_client = server.current_client; client *old_client = server.current_client;
server.current_client = receiver; server.current_client = receiver;
monotime replyTimer; monotime replyTimer;
elapsedStart(&replyTimer); elapsedStart(&replyTimer);
if (!moduleTryServeClientBlockedOnKey(receiver, rl->key)) continue; if (!moduleTryServeClientBlockedOnKey(receiver, rl->key)) continue;
updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer)); updateStatsOnUnblock(receiver, 0, elapsedUs(replyTimer), server.stat_total_error_replies != prev_error_replies);
moduleUnblockClient(receiver); moduleUnblockClient(receiver);
afterCommand(receiver); afterCommand(receiver);
server.current_client = old_client; server.current_client = old_client;
......
...@@ -7101,6 +7101,7 @@ void moduleHandleBlockedClients(void) { ...@@ -7101,6 +7101,7 @@ void moduleHandleBlockedClients(void) {
* was blocked on keys (RM_BlockClientOnKeys()), because we already * was blocked on keys (RM_BlockClientOnKeys()), because we already
* called such callback in moduleTryServeClientBlockedOnKey() when * called such callback in moduleTryServeClientBlockedOnKey() when
* the key was signaled as ready. */ * the key was signaled as ready. */
long long prev_error_replies = server.stat_total_error_replies;
uint64_t reply_us = 0; uint64_t reply_us = 0;
if (c && !bc->blocked_on_keys && bc->reply_callback) { if (c && !bc->blocked_on_keys && bc->reply_callback) {
RedisModuleCtx ctx; RedisModuleCtx ctx;
...@@ -7115,13 +7116,6 @@ void moduleHandleBlockedClients(void) { ...@@ -7115,13 +7116,6 @@ void moduleHandleBlockedClients(void) {
reply_us = elapsedUs(replyTimer); reply_us = elapsedUs(replyTimer);
moduleFreeContext(&ctx); moduleFreeContext(&ctx);
} }
/* Update stats now that we've finished the blocking operation.
* This needs to be out of the reply callback above given that a
* module might not define any callback and still do blocking ops.
*/
if (c && !bc->blocked_on_keys) {
updateStatsOnUnblock(c, bc->background_duration, reply_us);
}
/* Free privdata if any. */ /* Free privdata if any. */
if (bc->privdata && bc->free_privdata) { if (bc->privdata && bc->free_privdata) {
...@@ -7142,6 +7136,14 @@ void moduleHandleBlockedClients(void) { ...@@ -7142,6 +7136,14 @@ void moduleHandleBlockedClients(void) {
moduleReleaseTempClient(bc->reply_client); moduleReleaseTempClient(bc->reply_client);
moduleReleaseTempClient(bc->thread_safe_ctx_client); moduleReleaseTempClient(bc->thread_safe_ctx_client);
/* Update stats now that we've finished the blocking operation.
* This needs to be out of the reply callback above given that a
* module might not define any callback and still do blocking ops.
*/
if (c && !bc->blocked_on_keys) {
updateStatsOnUnblock(c, bc->background_duration, reply_us, server.stat_total_error_replies != prev_error_replies);
}
if (c != NULL) { if (c != NULL) {
/* Before unblocking the client, set the disconnect callback /* Before unblocking the client, set the disconnect callback
* to NULL, because if we reached this point, the client was * to NULL, because if we reached this point, the client was
...@@ -7199,10 +7201,11 @@ void moduleBlockedClientTimedOut(client *c) { ...@@ -7199,10 +7201,11 @@ void moduleBlockedClientTimedOut(client *c) {
ctx.client = bc->client; ctx.client = bc->client;
ctx.blocked_client = bc; ctx.blocked_client = bc;
ctx.blocked_privdata = bc->privdata; ctx.blocked_privdata = bc->privdata;
long long prev_error_replies = server.stat_total_error_replies;
bc->timeout_callback(&ctx,(void**)c->argv,c->argc); bc->timeout_callback(&ctx,(void**)c->argv,c->argc);
moduleFreeContext(&ctx); moduleFreeContext(&ctx);
if (!bc->blocked_on_keys) { if (!bc->blocked_on_keys) {
updateStatsOnUnblock(c, bc->background_duration, 0); updateStatsOnUnblock(c, bc->background_duration, 0, server.stat_total_error_replies != prev_error_replies);
} }
/* For timeout events, we do not want to call the disconnect callback, /* For timeout events, we do not want to call the disconnect callback,
* because the blocked client will be automatically disconnected in * because the blocked client will be automatically disconnected in
......
...@@ -2956,6 +2956,7 @@ NULL ...@@ -2956,6 +2956,7 @@ NULL
else else
replyToBlockedClientTimedOut(target); replyToBlockedClientTimedOut(target);
unblockClient(target); unblockClient(target);
updateStatsOnUnblock(target, 0, 0, 1);
addReply(c,shared.cone); addReply(c,shared.cone);
} else { } else {
addReply(c,shared.czero); addReply(c,shared.czero);
......
...@@ -3167,6 +3167,12 @@ void call(client *c, int flags) { ...@@ -3167,6 +3167,12 @@ void call(client *c, int flags) {
* the same error twice. */ * the same error twice. */
if ((server.stat_total_error_replies - prev_err_count) > 0) { if ((server.stat_total_error_replies - prev_err_count) > 0) {
real_cmd->failed_calls++; real_cmd->failed_calls++;
} else if (c->deferred_reply_errors) {
/* When call is used from a module client, error stats, and total_error_replies
* isn't updated since these errors, if handled by the module, are internal,
* and not reflected to users. however, the commandstats does show these calls
* (made by RM_Call), so it should log if they failed or succeeded. */
real_cmd->failed_calls++;
} }
/* After executing command, we will close the client after writing entire /* After executing command, we will close the client after writing entire
......
...@@ -3090,7 +3090,7 @@ void disconnectAllBlockedClients(void); ...@@ -3090,7 +3090,7 @@ void disconnectAllBlockedClients(void);
void handleClientsBlockedOnKeys(void); void handleClientsBlockedOnKeys(void);
void signalKeyAsReady(redisDb *db, robj *key, int type); void signalKeyAsReady(redisDb *db, robj *key, int type);
void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct blockPos *blockpos, streamID *ids); void blockForKeys(client *c, int btype, robj **keys, int numkeys, long count, mstime_t timeout, robj *target, struct blockPos *blockpos, streamID *ids);
void updateStatsOnUnblock(client *c, long blocked_us, long reply_us); void updateStatsOnUnblock(client *c, long blocked_us, long reply_us, int had_errors);
/* timeout.c -- Blocked clients timeout and connections timeout. */ /* timeout.c -- Blocked clients timeout and connections timeout. */
void addClientToTimeoutTable(client *c); void addClientToTimeoutTable(client *c);
......
...@@ -238,6 +238,7 @@ start_server {tags {"info" "external:skip"}} { ...@@ -238,6 +238,7 @@ start_server {tags {"info" "external:skip"}} {
assert_equal [s total_error_replies] 1 assert_equal [s total_error_replies] 1
r config resetstat r config resetstat
assert_match {} [errorstat OOM] assert_match {} [errorstat OOM]
r config set maxmemory 0
} }
test {errorstats: rejected call by authorization error} { test {errorstats: rejected call by authorization error} {
...@@ -253,6 +254,25 @@ start_server {tags {"info" "external:skip"}} { ...@@ -253,6 +254,25 @@ start_server {tags {"info" "external:skip"}} {
assert_equal [s total_error_replies] 1 assert_equal [s total_error_replies] 1
r config resetstat r config resetstat
assert_match {} [errorstat NOPERM] assert_match {} [errorstat NOPERM]
r auth default ""
} }
test {errorstats: blocking commands} {
r config resetstat
set rd [redis_deferring_client]
$rd client id
set rd_id [$rd read]
r del list1{t}
$rd blpop list1{t} 0
wait_for_blocked_client
r client unblock $rd_id error
assert_error {UNBLOCKED*} {$rd read}
assert_match {*count=1*} [errorstat UNBLOCKED]
assert_match {*calls=1,*,rejected_calls=0,failed_calls=1} [cmdstat blpop]
assert_equal [s total_error_replies] 1
$rd close
}
} }
} }
...@@ -183,21 +183,32 @@ start_server {tags {"modules"}} { ...@@ -183,21 +183,32 @@ start_server {tags {"modules"}} {
test {module client error stats} { test {module client error stats} {
r config resetstat r config resetstat
# simple module command that replies with string error
assert_error "NULL reply returned" {r do_rm_call hgetalllll} assert_error "NULL reply returned" {r do_rm_call hgetalllll}
assert_equal [errorrstat NULL r] {count=1} assert_equal [errorrstat NULL r] {count=1}
# module command that replies with string error from bg thread
assert_error "NULL reply returned" {r do_bg_rm_call hgetalllll} assert_error "NULL reply returned" {r do_bg_rm_call hgetalllll}
assert_equal [errorrstat NULL r] {count=2} assert_equal [errorrstat NULL r] {count=2}
# module command that returns an arity error
r do_rm_call set x x r do_rm_call set x x
assert_error "ERR wrong number of arguments for 'do_rm_call' command" {r do_rm_call} assert_error "ERR wrong number of arguments for 'do_rm_call' command" {r do_rm_call}
assert_equal [errorrstat ERR r] {count=1} assert_equal [errorrstat ERR r] {count=1}
# RM_Call that propagates an error
assert_error "WRONGTYPE*" {r do_rm_call hgetall x} assert_error "WRONGTYPE*" {r do_rm_call hgetall x}
assert_equal [errorrstat WRONGTYPE r] {count=1} assert_equal [errorrstat WRONGTYPE r] {count=1}
assert_match {*calls=1,*,rejected_calls=0,failed_calls=1} [cmdrstat hgetall r]
# RM_Call from bg thread that propagates an error
assert_error "WRONGTYPE*" {r do_bg_rm_call hgetall x} assert_error "WRONGTYPE*" {r do_bg_rm_call hgetall x}
assert_equal [errorrstat WRONGTYPE r] {count=2} assert_equal [errorrstat WRONGTYPE r] {count=2}
assert_match {*calls=2,*,rejected_calls=0,failed_calls=2} [cmdrstat hgetall r]
assert_equal [s total_error_replies] 5
assert_match {*calls=4,*,rejected_calls=0,failed_calls=3} [cmdrstat do_rm_call r]
assert_match {*calls=2,*,rejected_calls=0,failed_calls=2} [cmdrstat do_bg_rm_call r]
} }
test "Unload the module - blockedclient" { test "Unload the module - blockedclient" {
......
...@@ -281,14 +281,20 @@ start_server { ...@@ -281,14 +281,20 @@ start_server {
} }
test {XGROUP DESTROY should unblock XREADGROUP with -NOGROUP} { test {XGROUP DESTROY should unblock XREADGROUP with -NOGROUP} {
r config resetstat
r del mystream r del mystream
r XGROUP CREATE mystream mygroup $ MKSTREAM r XGROUP CREATE mystream mygroup $ MKSTREAM
set rd [redis_deferring_client] set rd [redis_deferring_client]
$rd XREADGROUP GROUP mygroup Alice BLOCK 0 STREAMS mystream ">" $rd XREADGROUP GROUP mygroup Alice BLOCK 0 STREAMS mystream ">"
wait_for_blocked_clients_count 1 wait_for_blocked_clients_count 1
r XGROUP DESTROY mystream mygroup r XGROUP DESTROY mystream mygroup
assert_error "*NOGROUP*" {$rd read} assert_error "NOGROUP*" {$rd read}
$rd close $rd close
# verify command stats, error stats and error counter work on failed blocked command
assert_match {*count=1*} [errorrstat NOGROUP r]
assert_match {*calls=1,*,rejected_calls=0,failed_calls=1} [cmdrstat xreadgroup r]
assert_equal [s total_error_replies] 1
} }
test {RENAME can unblock XREADGROUP with data} { test {RENAME can unblock XREADGROUP with data} {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment