Commit d6a8e64e authored by Oran Agra's avatar Oran Agra
Browse files

Fix and improve module error reply statistics (#10278)

This PR handles several aspects
1. Calls to RM_ReplyWithError from thread safe contexts don't violate thread safety.
2. Errors returning from RM_Call to the module aren't counted in the statistics (they
  might be handled silently by the module)
3. When a module propagates a reply it got from RM_Call to it's client, then the error
  statistics are counted.

This is done by:
1. When appending an error reply to the output buffer, we avoid updating the global
  error statistics, instead we cache that error in a deferred list in the client struct.
2. When creating a RedisModuleCallReply object, the deferred error list is moved from
  the client into that object.
3. when a module calls RM_ReplyWithCallReply we copy the deferred replies to the dest
  client (if that's a real client, then that's when the error statistics are updated to the server)

Note about RM_ReplyWithCallReply: if the original reply had an array with errors, and the module
replied with just a portion of the original reply, and not the entire reply, the errors are currently not
propagated and the errors stats will not get propagated.

Fix #10180

(cherry picked from commit b099889a)
parent a06f10b0
...@@ -269,6 +269,7 @@ typedef struct RedisModuleCallReply { ...@@ -269,6 +269,7 @@ typedef struct RedisModuleCallReply {
long long ll; /* Reply value for integer reply. */ long long ll; /* Reply value for integer reply. */
struct RedisModuleCallReply *array; /* Array of sub-reply elements. */ struct RedisModuleCallReply *array; /* Array of sub-reply elements. */
} val; } val;
list *deferred_error_list; /* list of errors in sds form or NULL */
} RedisModuleCallReply; } RedisModuleCallReply;
/* Structure representing a blocked client. We get a pointer to such /* Structure representing a blocked client. We get a pointer to such
...@@ -1702,6 +1703,15 @@ int RM_ReplyWithCallReply(RedisModuleCtx *ctx, RedisModuleCallReply *reply) { ...@@ -1702,6 +1703,15 @@ int RM_ReplyWithCallReply(RedisModuleCtx *ctx, RedisModuleCallReply *reply) {
if (c == NULL) return REDISMODULE_OK; if (c == NULL) return REDISMODULE_OK;
sds proto = sdsnewlen(reply->proto, reply->protolen); sds proto = sdsnewlen(reply->proto, reply->protolen);
addReplySds(c,proto); addReplySds(c,proto);
/* Propagate the error list from that reply to the other client, to do some
* post error reply handling, like statistics.
* Note that if the original reply had an array with errors, and the module
* replied with just a portion of the original reply, and not the entire
* reply, the errors are currently not propagated and the errors stats
* will not get propagated. */
if (reply->deferred_error_list)
deferredAfterErrorReply(c, reply->deferred_error_list);
return REDISMODULE_OK; return REDISMODULE_OK;
} }
...@@ -3753,12 +3763,16 @@ long long RM_StreamTrimByID(RedisModuleKey *key, int flags, RedisModuleStreamID ...@@ -3753,12 +3763,16 @@ long long RM_StreamTrimByID(RedisModuleKey *key, int flags, RedisModuleStreamID
/* Create a new RedisModuleCallReply object. The processing of the reply /* Create a new RedisModuleCallReply object. The processing of the reply
* is lazy, the object is just populated with the raw protocol and later * is lazy, the object is just populated with the raw protocol and later
* is processed as needed. Initially we just make sure to set the right * is processed as needed. Initially we just make sure to set the right
* reply type, which is extremely cheap to do. */ * reply type, which is extremely cheap to do.
RedisModuleCallReply *moduleCreateCallReplyFromProto(RedisModuleCtx *ctx, sds proto) { * The deferred_error_list is an optional list of errors that are present
* in the reply blob, if given, this function will take ownership on it.
*/
RedisModuleCallReply *moduleCreateCallReplyFromProto(RedisModuleCtx *ctx, sds proto, list *deferred_error_list) {
RedisModuleCallReply *reply = zmalloc(sizeof(*reply)); RedisModuleCallReply *reply = zmalloc(sizeof(*reply));
reply->ctx = ctx; reply->ctx = ctx;
reply->proto = proto; reply->proto = proto;
reply->protolen = sdslen(proto); reply->protolen = sdslen(proto);
reply->deferred_error_list = deferred_error_list;
reply->flags = REDISMODULE_REPLYFLAG_TOPARSE; /* Lazy parsing. */ reply->flags = REDISMODULE_REPLYFLAG_TOPARSE; /* Lazy parsing. */
switch(proto[0]) { switch(proto[0]) {
case '$': case '$':
...@@ -3876,11 +3890,14 @@ void moduleFreeCallReplyRec(RedisModuleCallReply *reply, int freenested){ ...@@ -3876,11 +3890,14 @@ void moduleFreeCallReplyRec(RedisModuleCallReply *reply, int freenested){
} }
} }
/* For nested replies, we don't free reply->proto (which if not NULL /* For nested replies, we don't free reply->proto (which if not NULL
* references the parent reply->proto buffer), nor the structure * references the parent reply->proto buffer), nor the structure
* itself which is allocated as an array of structures, and is freed * itself which is allocated as an array of structures, and is freed
* when the array value is released. */ * when the array value is released. */
if (!(reply->flags & REDISMODULE_REPLYFLAG_NESTED)) { if (!(reply->flags & REDISMODULE_REPLYFLAG_NESTED)) {
if (reply->deferred_error_list)
listRelease(reply->deferred_error_list);
if (reply->proto) sdsfree(reply->proto); if (reply->proto) sdsfree(reply->proto);
zfree(reply); zfree(reply);
} }
...@@ -4202,7 +4219,8 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -4202,7 +4219,8 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
proto = sdscatlen(proto,o->buf,o->used); proto = sdscatlen(proto,o->buf,o->used);
listDelNode(c->reply,listFirst(c->reply)); listDelNode(c->reply,listFirst(c->reply));
} }
reply = moduleCreateCallReplyFromProto(ctx,proto); reply = moduleCreateCallReplyFromProto(ctx,proto,c->deferred_reply_errors);
c->deferred_reply_errors = NULL; /* now the responsibility of the reply object. */
autoMemoryAdd(ctx,REDISMODULE_AM_REPLY,reply); autoMemoryAdd(ctx,REDISMODULE_AM_REPLY,reply);
cleanup: cleanup:
......
...@@ -168,6 +168,7 @@ client *createClient(connection *conn) { ...@@ -168,6 +168,7 @@ client *createClient(connection *conn) {
c->slave_addr = NULL; c->slave_addr = NULL;
c->slave_capa = SLAVE_CAPA_NONE; c->slave_capa = SLAVE_CAPA_NONE;
c->reply = listCreate(); c->reply = listCreate();
c->deferred_reply_errors = NULL;
c->reply_bytes = 0; c->reply_bytes = 0;
c->obuf_soft_limit_reached_time = 0; c->obuf_soft_limit_reached_time = 0;
listSetFreeMethod(c->reply,freeClientReplyValue); listSetFreeMethod(c->reply,freeClientReplyValue);
...@@ -417,6 +418,18 @@ void addReplyErrorLength(client *c, const char *s, size_t len) { ...@@ -417,6 +418,18 @@ void addReplyErrorLength(client *c, const char *s, size_t len) {
/* Do some actions after an error reply was sent (Log if needed, updates stats, etc.) */ /* Do some actions after an error reply was sent (Log if needed, updates stats, etc.) */
void afterErrorReply(client *c, const char *s, size_t len) { void afterErrorReply(client *c, const char *s, size_t len) {
/* Module clients fall into two categories:
* Calls to RM_Call, in which case the error isn't being returned to a client, so should not be counted.
* Module thread safe context calls to RM_ReplyWithError, which will be added to a real client by the main thread later. */
if (c->flags & CLIENT_MODULE) {
if (!c->deferred_reply_errors) {
c->deferred_reply_errors = listCreate();
listSetFreeMethod(c->deferred_reply_errors, (void (*)(void*))sdsfree);
}
listAddNodeTail(c->deferred_reply_errors, sdsnewlen(s, len));
return;
}
/* Increment the global error counter */ /* Increment the global error counter */
server.stat_total_error_replies++; server.stat_total_error_replies++;
/* Increment the error stats /* Increment the error stats
...@@ -965,6 +978,12 @@ void AddReplyFromClient(client *dst, client *src) { ...@@ -965,6 +978,12 @@ void AddReplyFromClient(client *dst, client *src) {
src->reply_bytes = 0; src->reply_bytes = 0;
src->bufpos = 0; src->bufpos = 0;
if (src->deferred_reply_errors) {
deferredAfterErrorReply(dst, src->deferred_reply_errors);
listRelease(src->deferred_reply_errors);
src->deferred_reply_errors = NULL;
}
/* Check output buffer limits */ /* Check output buffer limits */
closeClientOnOutputBufferLimitReached(dst, 1); closeClientOnOutputBufferLimitReached(dst, 1);
} }
...@@ -981,6 +1000,18 @@ void copyClientOutputBuffer(client *dst, client *src) { ...@@ -981,6 +1000,18 @@ void copyClientOutputBuffer(client *dst, client *src) {
dst->reply_bytes = src->reply_bytes; dst->reply_bytes = src->reply_bytes;
} }
/* Append the listed errors to the server error statistics. the input
* list is not modified and remains the responsibility of the caller. */
void deferredAfterErrorReply(client *c, list *errors) {
listIter li;
listNode *ln;
listRewind(errors,&li);
while((ln = listNext(&li))) {
sds err = ln->value;
afterErrorReply(c, err, sdslen(err));
}
}
/* Return true if the specified client has pending reply buffers to write to /* Return true if the specified client has pending reply buffers to write to
* the socket. */ * the socket. */
int clientHasPendingReplies(client *c) { int clientHasPendingReplies(client *c) {
...@@ -1375,6 +1406,8 @@ void freeClient(client *c) { ...@@ -1375,6 +1406,8 @@ void freeClient(client *c) {
listRelease(c->reply); listRelease(c->reply);
freeClientArgv(c); freeClientArgv(c);
freeClientOriginalArgv(c); freeClientOriginalArgv(c);
if (c->deferred_reply_errors)
listRelease(c->deferred_reply_errors);
/* Unlink the client: this will close the socket, remove the I/O /* Unlink the client: this will close the socket, remove the I/O
* handlers, and remove references of the client from different * handlers, and remove references of the client from different
...@@ -1661,6 +1694,10 @@ void resetClient(client *c) { ...@@ -1661,6 +1694,10 @@ void resetClient(client *c) {
c->multibulklen = 0; c->multibulklen = 0;
c->bulklen = -1; c->bulklen = -1;
if (c->deferred_reply_errors)
listRelease(c->deferred_reply_errors);
c->deferred_reply_errors = NULL;
/* We clear the ASKING flag as well if we are not inside a MULTI, and /* We clear the ASKING flag as well if we are not inside a MULTI, and
* if what we just executed is not the ASKING command itself. */ * if what we just executed is not the ASKING command itself. */
if (!(c->flags & CLIENT_MULTI) && prevcmd != askingCommand) if (!(c->flags & CLIENT_MULTI) && prevcmd != askingCommand)
......
...@@ -882,6 +882,7 @@ typedef struct client { ...@@ -882,6 +882,7 @@ typedef struct client {
long bulklen; /* Length of bulk argument in multi bulk request. */ long bulklen; /* Length of bulk argument in multi bulk request. */
list *reply; /* List of reply objects to send to the client. */ list *reply; /* List of reply objects to send to the client. */
unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */ unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
list *deferred_reply_errors; /* Used for module thread safe contexts. */
size_t sentlen; /* Amount of bytes already sent in the current size_t sentlen; /* Amount of bytes already sent in the current
buffer or object being sent. */ buffer or object being sent. */
time_t ctime; /* Client creation time. */ time_t ctime; /* Client creation time. */
...@@ -1855,6 +1856,7 @@ void addReplyHelp(client *c, const char **help); ...@@ -1855,6 +1856,7 @@ void addReplyHelp(client *c, const char **help);
void addReplySubcommandSyntaxError(client *c); void addReplySubcommandSyntaxError(client *c);
void addReplyLoadedModules(client *c); void addReplyLoadedModules(client *c);
void copyClientOutputBuffer(client *dst, client *src); void copyClientOutputBuffer(client *dst, client *src);
void deferredAfterErrorReply(client *c, list *errors);
size_t sdsZmallocSize(sds s); size_t sdsZmallocSize(sds s);
size_t getStringObjectSdsUsedMemory(robj *o); size_t getStringObjectSdsUsedMemory(robj *o);
void freeClientReplyValue(void *o); void freeClientReplyValue(void *o);
......
...@@ -90,4 +90,28 @@ start_server {tags {"modules"}} { ...@@ -90,4 +90,28 @@ start_server {tags {"modules"}} {
set clients [r client list] set clients [r client list]
assert_no_match "*name=myclient*" $clients assert_no_match "*name=myclient*" $clients
} }
test {module client error stats} {
r config resetstat
assert_error "NULL reply returned" {r do_rm_call hgetalllll}
assert_equal [errorrstat NULL r] {count=1}
assert_error "NULL reply returned" {r do_bg_rm_call hgetalllll}
assert_equal [errorrstat NULL r] {count=2}
r do_rm_call set x x
assert_error "ERR wrong number of arguments for 'do_rm_call' command" {r do_rm_call}
assert_equal [errorrstat ERR r] {count=1}
assert_error "WRONGTYPE*" {r do_rm_call hgetall x}
assert_equal [errorrstat WRONGTYPE r] {count=1}
assert_error "WRONGTYPE*" {r do_bg_rm_call hgetall x}
assert_equal [errorrstat WRONGTYPE r] {count=2}
}
test "Unload the module - blockedclient" {
assert_equal {OK} [r module unload blockedclient]
}
} }
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