Introduce fatlog to record big response packet.

src/commands.def

@@ -6637,6 +6637,115 @@ struct COMMAND_ARG FAILOVER_Args[] = {
 {MAKE_ARG("milliseconds",ARG_TYPE_INTEGER,-1,"TIMEOUT",NULL,NULL,CMD_ARG_OPTIONAL,0,NULL)},
 };
 
+/********** FATLOG GET ********************/
+
+#ifndef SKIP_CMD_HISTORY_TABLE
+/* FATLOG GET history */
+#define FATLOG_GET_History NULL
+#endif
+
+#ifndef SKIP_CMD_TIPS_TABLE
+/* FATLOG GET tips */
+const char *FATLOG_GET_Tips[] = {
+"request_policy:all_nodes",
+"nondeterministic_output",
+};
+#endif
+
+#ifndef SKIP_CMD_KEY_SPECS_TABLE
+/* FATLOG GET key specs */
+#define FATLOG_GET_Keyspecs NULL
+#endif
+
+/* FATLOG GET argument table */
+struct COMMAND_ARG FATLOG_GET_Args[] = {
+{MAKE_ARG("count",ARG_TYPE_INTEGER,-1,NULL,NULL,NULL,CMD_ARG_OPTIONAL,0,NULL)},
+};
+
+/********** FATLOG HELP ********************/
+
+#ifndef SKIP_CMD_HISTORY_TABLE
+/* FATLOG HELP history */
+#define FATLOG_HELP_History NULL
+#endif
+
+#ifndef SKIP_CMD_TIPS_TABLE
+/* FATLOG HELP tips */
+#define FATLOG_HELP_Tips NULL
+#endif
+
+#ifndef SKIP_CMD_KEY_SPECS_TABLE
+/* FATLOG HELP key specs */
+#define FATLOG_HELP_Keyspecs NULL
+#endif
+
+/********** FATLOG LEN ********************/
+
+#ifndef SKIP_CMD_HISTORY_TABLE
+/* FATLOG LEN history */
+#define FATLOG_LEN_History NULL
+#endif
+
+#ifndef SKIP_CMD_TIPS_TABLE
+/* FATLOG LEN tips */
+const char *FATLOG_LEN_Tips[] = {
+"request_policy:all_nodes",
+"response_policy:agg_sum",
+"nondeterministic_output",
+};
+#endif
+
+#ifndef SKIP_CMD_KEY_SPECS_TABLE
+/* FATLOG LEN key specs */
+#define FATLOG_LEN_Keyspecs NULL
+#endif
+
+/********** FATLOG RESET ********************/
+
+#ifndef SKIP_CMD_HISTORY_TABLE
+/* FATLOG RESET history */
+#define FATLOG_RESET_History NULL
+#endif
+
+#ifndef SKIP_CMD_TIPS_TABLE
+/* FATLOG RESET tips */
+const char *FATLOG_RESET_Tips[] = {
+"request_policy:all_nodes",
+"response_policy:all_succeeded",
+};
+#endif
+
+#ifndef SKIP_CMD_KEY_SPECS_TABLE
+/* FATLOG RESET key specs */
+#define FATLOG_RESET_Keyspecs NULL
+#endif
+
+/* FATLOG command table */
+struct COMMAND_STRUCT FATLOG_Subcommands[] = {
+{MAKE_CMD("get","Returns the fat log's entries.","O(N) where N is the number of entries returned","8.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FATLOG_GET_History,0,FATLOG_GET_Tips,2,slowlogCommand,-2,CMD_ADMIN|CMD_LOADING|CMD_STALE,0,FATLOG_GET_Keyspecs,0,NULL,1),.args=FATLOG_GET_Args},
+{MAKE_CMD("help","Show helpful text about the different subcommands","O(1)","8.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FATLOG_HELP_History,0,FATLOG_HELP_Tips,0,slowlogCommand,2,CMD_LOADING|CMD_STALE,0,FATLOG_HELP_Keyspecs,0,NULL,0)},
+{MAKE_CMD("len","Returns the number of entries in the fat log.","O(1)","8.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FATLOG_LEN_History,0,FATLOG_LEN_Tips,3,slowlogCommand,2,CMD_ADMIN|CMD_LOADING|CMD_STALE,0,FATLOG_LEN_Keyspecs,0,NULL,0)},
+{MAKE_CMD("reset","Clears all entries from the fat log.","O(N) where N is the number of entries in the fatlog","8.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FATLOG_RESET_History,0,FATLOG_RESET_Tips,2,slowlogCommand,2,CMD_ADMIN|CMD_LOADING|CMD_STALE,0,FATLOG_RESET_Keyspecs,0,NULL,0)},
+{0}
+};
+
+/********** FATLOG ********************/
+
+#ifndef SKIP_CMD_HISTORY_TABLE
+/* FATLOG history */
+#define FATLOG_History NULL
+#endif
+
+#ifndef SKIP_CMD_TIPS_TABLE
+/* FATLOG tips */
+#define FATLOG_Tips NULL
+#endif
+
+#ifndef SKIP_CMD_KEY_SPECS_TABLE
+/* FATLOG key specs */
+#define FATLOG_Keyspecs NULL
+#endif
+
 /********** FLUSHALL ********************/
 
 #ifndef SKIP_CMD_HISTORY_TABLE
@@ -10805,6 +10914,7 @@ struct COMMAND_STRUCT serverCommandTable[] = {
 {MAKE_CMD("dbsize","Returns the number of keys in the database.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,DBSIZE_History,0,DBSIZE_Tips,2,dbsizeCommand,1,CMD_READONLY|CMD_FAST,ACL_CATEGORY_KEYSPACE,DBSIZE_Keyspecs,0,NULL,0)},
 {MAKE_CMD("debug","A container for debugging commands.","Depends on subcommand.","1.0.0",CMD_DOC_SYSCMD,NULL,NULL,"server",COMMAND_GROUP_SERVER,DEBUG_History,0,DEBUG_Tips,0,debugCommand,-2,CMD_ADMIN|CMD_NOSCRIPT|CMD_LOADING|CMD_STALE|CMD_PROTECTED,0,DEBUG_Keyspecs,0,NULL,0)},
 {MAKE_CMD("failover","Starts a coordinated failover from a server to one of its replicas.","O(1)","6.2.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FAILOVER_History,0,FAILOVER_Tips,0,failoverCommand,-1,CMD_ADMIN|CMD_NOSCRIPT|CMD_STALE,0,FAILOVER_Keyspecs,0,NULL,3),.args=FAILOVER_Args},
+{MAKE_CMD("fatlog","A container for fat log commands.","Depends on subcommand.","8.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FATLOG_History,0,FATLOG_Tips,0,NULL,-2,0,0,FATLOG_Keyspecs,0,NULL,0),.subcommands=FATLOG_Subcommands},
 {MAKE_CMD("flushall","Removes all keys from all databases.","O(N) where N is the total number of keys in all databases","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FLUSHALL_History,2,FLUSHALL_Tips,2,flushallCommand,-1,CMD_WRITE,ACL_CATEGORY_KEYSPACE|ACL_CATEGORY_DANGEROUS,FLUSHALL_Keyspecs,0,NULL,1),.args=FLUSHALL_Args},
 {MAKE_CMD("flushdb","Remove all keys from the current database.","O(N) where N is the number of keys in the selected database","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,FLUSHDB_History,2,FLUSHDB_Tips,2,flushdbCommand,-1,CMD_WRITE,ACL_CATEGORY_KEYSPACE|ACL_CATEGORY_DANGEROUS,FLUSHDB_Keyspecs,0,NULL,1),.args=FLUSHDB_Args},
 {MAKE_CMD("info","Returns information and statistics about the server.","O(1)","1.0.0",CMD_DOC_NONE,NULL,NULL,"server",COMMAND_GROUP_SERVER,INFO_History,1,INFO_Tips,3,infoCommand,-1,CMD_LOADING|CMD_STALE|CMD_SENTINEL,ACL_CATEGORY_DANGEROUS,INFO_Keyspecs,0,NULL,1),.args=INFO_Args},

src/commands/fatlog-get.json

@@ -0,0 +1,69 @@
+{
+    "GET": {
+        "summary": "Returns the fat log's entries.",
+        "complexity": "O(N) where N is the number of entries returned",
+        "group": "server",
+        "since": "8.0.0",
+        "arity": -2,
+        "container": "FATLOG",
+        "function": "slowlogCommand",
+        "history": [],
+        "command_flags": [
+            "ADMIN",
+            "LOADING",
+            "STALE"
+        ],
+        "command_tips": [
+            "REQUEST_POLICY:ALL_NODES",
+            "NONDETERMINISTIC_OUTPUT"
+        ],
+        "reply_schema": {
+            "type": "array",
+            "description": "Entries from the slow log in chronological order.",
+            "uniqueItems": true,
+            "items": {
+                "type": "array",
+                "minItems": 6,
+                "maxItems": 6,
+                "items": [
+                    {
+                        "type": "integer",
+                        "description": "Fat log entry ID."
+                    },
+                    {
+                        "type": "integer",
+                        "description": "The unix timestamp at which the logged command was processed.",
+                        "minimum": 0
+                    },
+                    {
+                        "type": "integer",
+                        "description": "The size of the response to the query in bytes.",
+                        "minimum": 0
+                    },
+                    {
+                        "type": "array",
+                        "description": "The arguments of the command.",
+                        "items": {
+                            "type": "string"
+                        }
+                    },
+                    {
+                        "type": "string",
+                        "description": "Client IP address and port."
+                    },
+                    {
+                        "type": "string",
+                        "description": "Client name if set via the CLIENT SETNAME command."
+                    }
+                ]
+            }
+        },
+        "arguments": [
+            {
+                "name": "count",
+                "type": "integer",
+                "optional": true
+            }
+        ]
+    }
+}

src/commands/fatlog-help.json

@@ -0,0 +1,22 @@
+{
+    "HELP": {
+        "summary": "Show helpful text about the different subcommands",
+        "complexity": "O(1)",
+        "group": "server",
+        "since": "8.0.0",
+        "arity": 2,
+        "container": "FATLOG",
+        "function": "slowlogCommand",
+        "command_flags": [
+            "LOADING",
+            "STALE"
+        ],
+        "reply_schema": {
+            "type": "array",
+            "description": "Helpful text about subcommands.",
+            "items": {
+                "type": "string"
+            }
+        }
+    }
+}

src/commands/fatlog-len.json

@@ -0,0 +1,26 @@
+{
+    "LEN": {
+        "summary": "Returns the number of entries in the fat log.",
+        "complexity": "O(1)",
+        "group": "server",
+        "since": "8.0.0",
+        "arity": 2,
+        "container": "FATLOG",
+        "function": "slowlogCommand",
+        "command_flags": [
+            "ADMIN",
+            "LOADING",
+            "STALE"
+        ],
+        "command_tips": [
+            "REQUEST_POLICY:ALL_NODES",
+            "RESPONSE_POLICY:AGG_SUM",
+            "NONDETERMINISTIC_OUTPUT"
+        ],
+        "reply_schema": {
+            "type": "integer",
+            "description": "Number of entries in the fat log.",
+            "minimum": 0
+        }
+    }
+}

src/commands/fatlog-reset.json

@@ -0,0 +1,23 @@
+{
+    "RESET": {
+        "summary": "Clears all entries from the fat log.",
+        "complexity": "O(N) where N is the number of entries in the fatlog",
+        "group": "server",
+        "since": "8.0.0",
+        "arity": 2,
+        "container": "FATLOG",
+        "function": "slowlogCommand",
+        "command_flags": [
+            "ADMIN",
+            "LOADING",
+            "STALE"
+        ],
+        "command_tips": [
+            "REQUEST_POLICY:ALL_NODES",
+            "RESPONSE_POLICY:ALL_SUCCEEDED"
+        ],
+        "reply_schema": {
+            "const": "OK"
+        }
+    }
+}

src/commands/fatlog.json

@@ -0,0 +1,9 @@
+{
+    "FATLOG": {
+        "summary": "A container for fat log commands.",
+        "complexity": "Depends on subcommand.",
+        "group": "server",
+        "since": "8.0.0",
+        "arity": -2
+    }
+}

src/config.c

@@ -3157,6 +3157,7 @@ standardConfig static_configs[] = {
     createULongConfig("active-defrag-max-scan-fields", NULL, MODIFIABLE_CONFIG, 1, LONG_MAX, server.active_defrag_max_scan_fields, 1000, INTEGER_CONFIG, NULL, NULL), /* Default: keys with more than 1000 fields will be processed separately */
     createULongConfig("slowlog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.slowlog_max_len, 128, INTEGER_CONFIG, NULL, NULL),
     createULongConfig("acllog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.acllog_max_len, 128, INTEGER_CONFIG, NULL, NULL),
+    createULongConfig("fatlog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.fatlog_max_len, 128, INTEGER_CONFIG, NULL, NULL),
 
     /* Long Long configs */
     createLongLongConfig("busy-reply-threshold", "lua-time-limit", MODIFIABLE_CONFIG, 0, LONG_MAX, server.busy_reply_threshold, 5000, INTEGER_CONFIG, NULL, NULL), /* milliseconds */
@@ -3167,6 +3168,7 @@ standardConfig static_configs[] = {
     createLongLongConfig("proto-max-bulk-len", NULL, DEBUG_CONFIG | MODIFIABLE_CONFIG, 1024 * 1024, LONG_MAX, server.proto_max_bulk_len, 512ll * 1024 * 1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */
     createLongLongConfig("stream-node-max-entries", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.stream_node_max_entries, 100, INTEGER_CONFIG, NULL, NULL),
     createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, server.repl_backlog_size, 1024 * 1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
+    createLongLongConfig("fatlog-log-bigger-than", NULL, MODIFIABLE_CONFIG, -1, LLONG_MAX, server.fatlog_log_bigger_than, 1024 * 1024, MEMORY_CONFIG, NULL, NULL),
 
     /* Unsigned Long Long configs */
     createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, ULLONG_MAX, server.maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory),

src/networking.c

@@ -184,6 +184,7 @@ client *createClient(connection *conn) {
     c->reply = listCreate();
     c->deferred_reply_errors = NULL;
     c->reply_bytes = 0;
+    c->cmd_reply_length = 0;
     c->obuf_soft_limit_reached_time = 0;
     listSetFreeMethod(c->reply, freeClientReplyValue);
     listSetDupMethod(c->reply, dupClientReplyValue);
@@ -439,6 +440,9 @@ void _addReplyToBufferOrList(client *c, const char *s, size_t len) {
      * buffer offset (see function comment) */
     reqresSaveClientReplyOffset(c);
 
+    /* Record reply length. */
+    c->cmd_reply_length += len;
+
     /* If we're processing a push message into the current client (i.e. executing PUBLISH
      * to a channel which we are subscribed to, then we wanna postpone that message to be added
      * after the command's reply (specifically important during multi-exec). the exception is
@@ -791,6 +795,9 @@ void setDeferredReply(client *c, void *node, const char *s, size_t length) {
     if (node == NULL) return;
     serverAssert(!listNodeValue(ln));
 
+    /* Record reply length. */
+    c->cmd_reply_length += length;
+
     /* Normally we fill this dummy NULL node, added by addReplyDeferredLen(),
      * with a new buffer structure containing the protocol needed to specify
      * the length of the array following. However sometimes there might be room

src/server.c

@@ -3271,6 +3271,18 @@ void slowlogPushCurrentCommand(client *c, struct serverCommand *cmd, ustime_t du
     slowlogPushEntryIfNeeded(c, argv, argc, duration);
 }
 
+/* Log the last command a client executed into the fatlog. */
+void fatlogPushCurrentCommand(client *c, struct serverCommand *cmd, size_t size) {
+    /* Some commands may contain sensitive data that should not be available in the fatlog. */
+    if (cmd->flags & CMD_SKIP_SLOWLOG) return;
+
+    /* If command argument vector was rewritten, use the original
+     * arguments. */
+    robj **argv = c->original_argv ? c->original_argv : c->argv;
+    int argc = c->original_argv ? c->original_argc : c->argc;
+    fatlogPushEntryIfNeeded(c, argv, argc, size);
+}
+
 /* This function is called in order to update the total command histogram duration.
  * The latency unit is nano-seconds.
  * If needed it will allocate the histogram memory and trim the duration to the upper/lower tracking limits*/
@@ -3462,6 +3474,13 @@ void call(client *c, int flags) {
      * re-processed. */
     if (reprocessing_command) c->flags |= CLIENT_REPROCESSING_COMMAND;
 
+    if (update_command_stats) {
+        fatlogPushCurrentCommand(c, real_cmd, c->argv_len_sum);
+    }
+
+    /* To record how many reply bytes generated in this command. */
+    c->cmd_reply_length = 0;
+
     monotime monotonic_start = 0;
     if (monotonicGetType() == MONOTONIC_CLOCK_HW) monotonic_start = getMonotonicUs();
 
@@ -3520,7 +3539,10 @@ void call(client *c, int flags) {
 
     /* Log the command into the Slow log if needed.
      * If the client is blocked we will handle slowlog when it is unblocked. */
-    if (update_command_stats && !(c->flags & CLIENT_BLOCKED)) slowlogPushCurrentCommand(c, real_cmd, c->duration);
+    if (update_command_stats && !(c->flags & CLIENT_BLOCKED)) {
+        slowlogPushCurrentCommand(c, real_cmd, c->duration);
+        fatlogPushCurrentCommand(c, real_cmd, c->cmd_reply_length);
+    }
 
     /* Send the command to clients in MONITOR mode if applicable,
      * since some administrative commands are considered too dangerous to be shown.