Cluster compatibility check (#13846)

### Background
The program runs normally in standalone mode, but migrating to cluster
mode may cause errors, this is because some cross slot commands can not
run in cluster mode. We should provide an approach to detect this issue
when running in standalone mode, and need to expose a metric which
indicates the usage of no incompatible commands.

### Solution
To avoid perf impact, we introduce a new config
`cluster-compatibility-sample-ratio` which define the sampling ratio
(0-100) for checking command compatibility in cluster mode. When a
command is executed, it is sampled at the specified ratio to determine
if it complies with Redis cluster constraints, such as cross-slot
restrictions.

A new metric is exposed: `cluster_incompatible_ops` in `info stats`
output.

The following operations will be considered incompatible operations.

- cross-slot command
   If a command has multiple cross slot keys, it is incompatible
- `swap, copy, move, select` command
These commands involve multi databases in some cases, we don't allow
multiple DB in cluster mode, so there are not compatible
- Module command with `no-cluster` flag
If a module command has `no-cluster` flag, we will encounter an error
when loading module, leading to fail to load module if cluster is
enabled, so this is incompatible.
- Script/function with `no-cluster` flag
Similar with module command, if we declare `no-cluster` in shebang of
script/function, we also can not run it in cluster mode
- `sort` command by/get pattern
When `sort` command has `by/get` pattern option, we must ask that the
pattern slot is equal with the slot of keys, otherwise it is
incompatible in cluster mode.

- The script/function command accesses the keys and declared keys have
different slots
For the script/function command, we not only check the slot of declared
keys, but only check the slot the accessing keys, if they are different,
we think it is incompatible.

**Besides**, commands like `keys, scan, flushall, script/function
flush`, that in standalone mode iterate over all data to perform the
operation, are only valid for the server that executes the command in
cluster mode and are not broadcasted. However, this does not lead to
errors, so we do not consider them as incompatible commands.

### Performance impact test
**cross slot test**
Below are the test commands and results. When using MSET with 8 keys,
performance drops by approximately 3%.

**single key test**
It may be due to the overhead of the sampling function, and single-key
commands could cause a 1-2% performance drop.

redis.conf

@@ -1783,6 +1783,21 @@ aof-timestamp-enabled no
 #
 # cluster-preferred-endpoint-type ip
 
+# This configuration defines the sampling ratio (0-100) for checking command
+# compatibility in cluster mode. When a command is executed, it is sampled at
+# the specified ratio to determine if it complies with Redis cluster constraints,
+# such as cross-slot restrictions.
+#
+# - A value of 0 means no commands are sampled for compatibility checks.
+# - A value of 100 means all commands are checked.
+# - Intermediate values (e.g., 10) mean that approximately 10% of the commands
+#   are randomly selected for compatibility verification.
+#
+# Higher sampling ratios may introduce additional performance overhead, especially
+# under high QPS. The default value is 0 (no sampling).
+#
+# cluster-compatibility-sample-ratio 0
+
 # In order to setup your cluster make sure to read the documentation
 # available at https://redis.io web site.
 

src/config.c

@@ -3197,6 +3197,7 @@ standardConfig static_configs[] = {
     createIntConfig("watchdog-period", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.watchdog_period, 0, INTEGER_CONFIG, NULL, updateWatchdogPeriod),
     createIntConfig("shutdown-timeout", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.shutdown_timeout, 10, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("repl-diskless-sync-max-replicas", NULL, MODIFIABLE_CONFIG, 0, INT_MAX, server.repl_diskless_sync_max_replicas, 0, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("cluster-compatibility-sample-ratio", NULL, MODIFIABLE_CONFIG, 0, 100, server.cluster_compatibility_sample_ratio, 0, INTEGER_CONFIG, NULL, NULL),
 
     /* Unsigned int configs */
     createUIntConfig("maxclients", NULL, MODIFIABLE_CONFIG, 1, UINT_MAX, server.maxclients, 10000, INTEGER_CONFIG, NULL, updateMaxclients),

src/db.c

@@ -986,6 +986,11 @@ void selectCommand(client *c) {
         addReplyError(c,"SELECT is not allowed in cluster mode");
         return;
     }
+
+    if (id != 0) {
+        server.stat_cluster_incompatible_ops++;
+    }
+
     if (selectDb(c,id) == C_ERR) {
         addReplyError(c,"DB index is out of range");
     } else {
@@ -1698,6 +1703,9 @@ void moveCommand(client *c) {
         return;
     }
 
+    /* Record incompatible operations in cluster mode */
+    server.stat_cluster_incompatible_ops++;
+
     /* Check if the element exists and get a reference */
     o = lookupKeyWrite(c->db,c->argv[1]);
     if (!o) {
@@ -1791,6 +1799,10 @@ void copyCommand(client *c) {
         return;
     }
 
+    if (srcid != 0 || dbid != 0) {
+        server.stat_cluster_incompatible_ops++;
+    }
+
     /* Check if the element exists and get a reference */
     o = lookupKeyRead(c->db, key);
     if (!o) {
@@ -2029,6 +2041,7 @@ void swapdbCommand(client *c) {
         RedisModuleSwapDbInfo si = {REDISMODULE_SWAPDBINFO_VERSION,id1,id2};
         moduleFireServerEvent(REDISMODULE_EVENT_SWAPDB,0,&si);
         server.dirty++;
+        server.stat_cluster_incompatible_ops++;
         addReply(c,shared.ok);
     }
 }

src/module.c

@@ -1273,6 +1273,10 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
     if ((flags & CMD_MODULE_NO_CLUSTER) && server.cluster_enabled)
         return REDISMODULE_ERR;
 
+    /* We will encounter an error as above if cluster is enable */
+    if (flags & CMD_MODULE_NO_CLUSTER)
+        server.stat_cluster_incompatible_ops++;
+
     /* Check if the command name is valid. */
     if (!isCommandNameValid(name))
         return REDISMODULE_ERR;
@@ -1400,6 +1404,10 @@ int RM_CreateSubcommand(RedisModuleCommand *parent, const char *name, RedisModul
     if ((flags & CMD_MODULE_NO_CLUSTER) && server.cluster_enabled)
         return REDISMODULE_ERR;
 
+    /* We will encounter an error as above if cluster is enable */
+    if (flags & CMD_MODULE_NO_CLUSTER)
+        server.stat_cluster_incompatible_ops++;
+
     struct redisCommand *parent_cmd = parent->rediscmd;
 
     if (parent_cmd->parent)

src/networking.c

@@ -172,6 +172,7 @@ client *createClient(connection *conn) {
     c->io_flags = CLIENT_IO_READ_ENABLED | CLIENT_IO_WRITE_ENABLED;
     c->read_error = 0;
     c->slot = -1;
+    c->cluster_compatibility_check_slot = -2;
     c->ctime = c->lastinteraction = server.unixtime;
     c->duration = 0;
     clientSetDefaultAuth(c);
@@ -2238,6 +2239,7 @@ static inline void resetClientInternal(client *c, int free_argv) {
     c->multibulklen = 0;
     c->bulklen = -1;
     c->slot = -1;
+    c->cluster_compatibility_check_slot = -2;
     c->flags &= ~CLIENT_EXECUTING_COMMAND;
 
     /* Make sure the duration has been recorded to some command. */

src/script.c

@@ -182,6 +182,11 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
             return C_ERR;
         }
 
+        /* Can't run script with 'non-cluster' flag as above when cluster is enabled. */
+        if (script_flags & SCRIPT_FLAG_NO_CLUSTER) {
+            server.stat_cluster_incompatible_ops++;
+        }
+
         if (running_stale && !(script_flags & SCRIPT_FLAG_ALLOW_STALE)) {
             addReplyError(caller, "-MASTERDOWN Link with MASTER is down, "
                              "replica-serve-stale-data is set to 'no' "
@@ -249,6 +254,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
     run_ctx->original_client = caller;
     run_ctx->funcname = funcname;
     run_ctx->slot = caller->slot;
+    run_ctx->cluster_compatibility_check_slot = caller->cluster_compatibility_check_slot;
 
     client *script_client = run_ctx->c;
     client *curr_client = run_ctx->original_client;
@@ -303,6 +309,7 @@ void scriptResetRun(scriptRunCtx *run_ctx) {
     }
 
     run_ctx->slot = -1;
+    run_ctx->cluster_compatibility_check_slot = -2;
 
     preventCommandPropagation(run_ctx->original_client);
 
@@ -521,6 +528,33 @@ static int scriptVerifyClusterState(scriptRunCtx *run_ctx, client *c, client *or
     return C_OK;
 }
 
+static void scriptCheckClusterCompatibility(scriptRunCtx *run_ctx, client *c) {
+    int hashslot = -1;
+
+    /* If we don't need to detect for this script or slot violation already
+     * detected and reported for this script, exit */
+    if (run_ctx->cluster_compatibility_check_slot == -2)  return;
+
+    if (!areCommandKeysInSameSlot(c, &hashslot)) {
+        server.stat_cluster_incompatible_ops++;
+        /* Already found cross slot usage, skip the check for the rest of the script */
+        run_ctx->cluster_compatibility_check_slot = -2;
+    } else {
+        /* Check whether the declared keys and the accessed keys belong to the same slot.
+         * If having SCRIPT_ALLOW_CROSS_SLOT flag, skip this check since it's allowed
+         * in cluster mode, but it may fail when the slot doesn't belong to the node. */
+        if (hashslot != -1 && !(run_ctx->flags & SCRIPT_ALLOW_CROSS_SLOT)) {
+            if (run_ctx->cluster_compatibility_check_slot == -1) {
+                run_ctx->cluster_compatibility_check_slot = hashslot;
+            } else if (run_ctx->cluster_compatibility_check_slot != hashslot) {
+                server.stat_cluster_incompatible_ops++;
+                /* Already found cross slot usage, skip the check for the rest of the script */
+                run_ctx->cluster_compatibility_check_slot = -2;
+            }
+        }
+    }
+}
+
 /* set RESP for a given run_ctx */
 int scriptSetResp(scriptRunCtx *run_ctx, int resp) {
     if (resp != 2 && resp != 3) {
@@ -618,6 +652,8 @@ void scriptCall(scriptRunCtx *run_ctx, sds *err) {
         goto error;
     }
 
+    scriptCheckClusterCompatibility(run_ctx, c);
+
     int call_flags = CMD_CALL_NONE;
     if (run_ctx->repl_flags & PROPAGATE_AOF) {
         call_flags |= CMD_CALL_PROPAGATE_AOF;

src/script.h

@@ -54,6 +54,7 @@ struct scriptRunCtx {
     int repl_flags;
     monotime start_time;
     int slot;
+    int cluster_compatibility_check_slot;
 };
 
 /* Scripts flags */

src/server.c

@@ -2660,6 +2660,7 @@ void resetServerStats(void) {
     server.aof_delayed_fsync = 0;
     server.stat_reply_buffer_shrinks = 0;
     server.stat_reply_buffer_expands = 0;
+    server.stat_cluster_incompatible_ops = 0;
     memset(server.duration_stats, 0, sizeof(durationStats) * EL_DURATION_TYPE_NUM);
     server.el_cmd_cnt_max = 0;
     lazyfreeResetStats();
@@ -4124,6 +4125,21 @@ int processCommand(client *c) {
         }
     }
 
+    /* Check if the command keys are all in the same slot for cluster compatibility */
+    if (server.cluster_compatibility_sample_ratio && !server.cluster_enabled &&
+        !(!(c->cmd->flags&CMD_MOVABLE_KEYS) && c->cmd->key_specs_num == 0 &&
+          c->cmd->proc != execCommand) && SHOULD_CLUSTER_COMPATIBILITY_SAMPLE())
+    {
+        c->cluster_compatibility_check_slot = -1;
+        if (!areCommandKeysInSameSlot(c, &c->cluster_compatibility_check_slot)) {
+            server.stat_cluster_incompatible_ops++;
+            /* If we find cross slot keys, reset slot to -2 to indicate we won't
+             * check this command again. That is useful for script, since we need
+             * this variable to decide if we continue checking accessing keys. */
+            c->cluster_compatibility_check_slot = -2;
+        }
+    }
+
     /* Disconnect some clients if total clients memory is too high. We do this
      * before key eviction, after the last command was executed and consumed
      * some client output buffer memory. */
@@ -4316,6 +4332,46 @@ int processCommand(client *c) {
     return C_OK;
 }
 
+/* Checks if all keys in a command (or a MULTI-EXEC) belong to the same hash slot.
+ * If yes, return 1, otherwise 0. If hashslot is not NULL, it will be set to the
+ * slot of the keys. */
+int areCommandKeysInSameSlot(client *c, int *hashslot) {
+    int slot = -1;
+    multiState *ms = NULL;
+
+    if (c->cmd->proc == execCommand) {
+        if (!(c->flags & CLIENT_MULTI)) return 1;
+        else ms = &c->mstate;
+    }
+
+    /* If client is in multi-exec, we need to check the slot of all keys
+     * in the transaction. */
+    for (int i = 0; i < (ms ? ms->count : 1); i++) {
+        struct redisCommand *cmd = ms ? ms->commands[i].cmd : c->cmd;
+        robj **argv = ms ? ms->commands[i].argv : c->argv;
+        int argc = ms ? ms->commands[i].argc : c->argc;
+
+        getKeysResult result = GETKEYS_RESULT_INIT;
+        int numkeys = getKeysFromCommand(cmd, argv, argc, &result);
+        keyReference *keyindex = result.keys;
+
+        /* Check if all keys have the same slots, increment the metric if not */
+        for (int j = 0; j < numkeys; j++) {
+            robj *thiskey = argv[keyindex[j].pos];
+            int thisslot = keyHashSlot((char*)thiskey->ptr, sdslen(thiskey->ptr));
+            if (slot == -1) {
+                slot = thisslot;
+            } else if (slot != thisslot) {
+                getKeysFreeResult(&result);
+                return 0;
+            }
+        }
+        getKeysFreeResult(&result);
+    }
+    if (hashslot) *hashslot = slot;
+    return 1;
+}
+
 /* ====================== Error lookup and execution ===================== */
 
 /* Users who abuse lua error_reply will generate a new error object on each
@@ -6083,6 +6139,9 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
             "instantaneous_eventloop_cycles_per_sec:%llu\r\n", getInstantaneousMetric(STATS_METRIC_EL_CYCLE),
             "instantaneous_eventloop_duration_usec:%llu\r\n", getInstantaneousMetric(STATS_METRIC_EL_DURATION)));
         info = genRedisInfoStringACLStats(info);
+        if (!server.cluster_enabled && server.cluster_compatibility_sample_ratio) {
+            sdscatprintf(info, "cluster_incompatible_ops:%lld\r\n", server.stat_cluster_incompatible_ops);
+        }
     }
 
     /* Replication */

src/server.h

@@ -1239,6 +1239,10 @@ typedef struct {
     size_t mem_usage_sum;
 } clientMemUsageBucket;
 
+#define SHOULD_CLUSTER_COMPATIBILITY_SAMPLE() \
+            (server.cluster_compatibility_sample_ratio == 100 || \
+             (double)rand()/RAND_MAX * 100 < server.cluster_compatibility_sample_ratio)
+
 #ifdef LOG_REQ_RES
 /* Structure used to log client's requests and their
  * responses (see logreqres.c) */
@@ -1305,6 +1309,11 @@ typedef struct client {
     time_t ctime;           /* Client creation time. */
     long duration;          /* Current command duration. Used for measuring latency of blocking/non-blocking cmds */
     int slot;               /* The slot the client is executing against. Set to -1 if no slot is being used */
+    int cluster_compatibility_check_slot; /* The slot the client is executing against for cluster compatibility check.
+                                           * -2 means we don't need to check slot violation, or we already found
+                                           * a violation, reported it and don't need to continue checking.
+                                           * -1 means we're looking for the slot number and didn't find it yet.
+                                           * any positive number means we found a slot and no violation yet. */
     dictEntry *cur_script;  /* Cached pointer to the dictEntry of the script being executed. */
     time_t lastinteraction; /* Time of the last interaction, used for timeout */
     time_t obuf_soft_limit_reached_time;
@@ -1848,6 +1857,7 @@ struct redisServer {
     redisAtomic long long stat_io_writes_processed[IO_THREADS_MAX_NUM]; /* Number of write events processed by IO / Main threads */
     redisAtomic long long stat_client_qbuf_limit_disconnections;  /* Total number of clients reached query buf length limit */
     long long stat_client_outbuf_limit_disconnections;  /* Total number of clients reached output buf length limit */
+    long long stat_cluster_incompatible_ops; /* Number of operations that are incompatible with cluster mode */
     /* The following two are used to track instantaneous metrics, like
      * number of operations per second, network traffic. */
     struct {
@@ -1903,6 +1913,8 @@ struct redisServer {
     int latency_tracking_info_percentiles_len;
     unsigned int max_new_tls_conns_per_cycle; /* The maximum number of tls connections that will be accepted during each invocation of the event loop. */
     unsigned int max_new_conns_per_cycle; /* The maximum number of tcp connections that will be accepted during each invocation of the event loop. */
+    int cluster_compatibility_sample_ratio; /* Sampling ratio for cluster mode incompatible commands. */
+
     /* AOF persistence */
     int aof_enabled;                /* AOF configuration */
     int aof_state;                  /* AOF_(ON|OFF|WAIT_REWRITE) */
@@ -3232,6 +3244,7 @@ int processCommand(client *c);
 void commandProcessed(client *c);
 int processPendingCommandAndInputBuffer(client *c);
 int processCommandAndResetClient(client *c);
+int areCommandKeysInSameSlot(client *c, int *hashslot);
 void setupSignalHandlers(void);
 int createSocketAcceptHandler(connListener *sfd, aeFileProc *accept_handler);
 connListener *listenerByType(const char *typename);

src/sort.c

@@ -230,6 +230,17 @@ void sortCommandGeneric(client *c, int readonly) {
                     syntax_error++;
                     break;
                 }
+
+                /* If the BY pattern slot is not equal with the slot of keys, we will record
+                 * an incompatible behavior as above comments. */
+                if (server.cluster_compatibility_sample_ratio && !server.cluster_enabled &&
+                    SHOULD_CLUSTER_COMPATIBILITY_SAMPLE())
+                {
+                    if (patternHashSlot(sortby->ptr, sdslen(sortby->ptr)) !=
+                        (int)keyHashSlot(c->argv[1]->ptr, sdslen(c->argv[1]->ptr)))
+                        server.stat_cluster_incompatible_ops++;
+                }
+
                 /* If BY is specified with a real pattern, we can't accept
                  * it if no full ACL key access is applied for this command. */
                 if (!user_has_full_key_access) {
@@ -253,6 +264,18 @@ void sortCommandGeneric(client *c, int readonly) {
                 syntax_error++;
                 break;
             }
+
+            /* If the GET pattern slot is not equal with the slot of keys, we will record
+             * an incompatible behavior as above comments. */
+            if (server.cluster_compatibility_sample_ratio && !server.cluster_enabled &&
+                strcmp(c->argv[j+1]->ptr, "#") &&
+                SHOULD_CLUSTER_COMPATIBILITY_SAMPLE())
+            {
+                if (patternHashSlot(c->argv[j+1]->ptr, sdslen(c->argv[j+1]->ptr)) !=
+                    (int)keyHashSlot(c->argv[1]->ptr, sdslen(c->argv[1]->ptr)))
+                    server.stat_cluster_incompatible_ops++;
+            }
+
             if (!user_has_full_key_access) {
                 addReplyError(c,"GET option of SORT denied due to insufficient ACL permissions.");
                 syntax_error++;

tests/modules/misc.c

@@ -540,6 +540,14 @@ int test_keyslot(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     return RedisModule_ReplyWithLongLong(ctx, slot);
 }
 
+int only_reply_ok(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    RedisModule_ReplyWithSimpleString(ctx, "OK");
+    return REDISMODULE_OK;
+}
+
 int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     REDISMODULE_NOT_USED(argv);
     REDISMODULE_NOT_USED(argc);
@@ -607,6 +615,13 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
         return REDISMODULE_ERR;
     if (RedisModule_CreateCommand(ctx, "test.keyslot", test_keyslot, "", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
+    if (RedisModule_CreateCommand(ctx, "test.incompatible_cluster_cmd", only_reply_ok, "", 1, -1, 2) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+    if (RedisModule_CreateCommand(ctx, "test.no_cluster_cmd", NULL, "no-cluster", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+    RedisModuleCommand *parent = RedisModule_GetCommand(ctx, "test.no_cluster_cmd");
+    if (RedisModule_CreateSubcommand(parent, "set", only_reply_ok, "no-cluster", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
 
     return REDISMODULE_OK;
 }

tests/unit/moduleapi/misc.tcl

@@ -561,3 +561,27 @@ if {[string match {*jemalloc*} [s mem_allocator]]} {
         assert_equal {OK} [r module unload misc]
     }
 }
+
+start_server {tags {"modules"}} {
+    test {Detect incompatible operations in cluster mode for module} {
+        r config set cluster-compatibility-sample-ratio 100
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # since test.no_cluster_cmd and its subcommand have 'no-cluster' flag,
+        # they should not be counted as incompatible ops, increment the counter by 2
+        r module load $testmodule
+        assert_equal [expr $incompatible_ops+2] [s cluster_incompatible_ops]
+
+        # incompatible_cluster_cmd is similar with MSET, check if it is counted as
+        # incompatible ops with different number of keys
+        # only 1 key, should not increment the counter
+        r test.incompatible_cluster_cmd foo bar
+        assert_equal [expr $incompatible_ops+2] [s cluster_incompatible_ops]
+        # 2 cross slot keys, should increment the counter
+        r test.incompatible_cluster_cmd foo bar bar foo
+        assert_equal [expr $incompatible_ops+3] [s cluster_incompatible_ops]
+        # 2 non cross slot keys, should not increment the counter
+        r test.incompatible_cluster_cmd foo bar bar{foo} bar
+        assert_equal [expr $incompatible_ops+3] [s cluster_incompatible_ops]
+    }
+}

tests/unit/other.tcl

@@ -530,3 +530,204 @@ start_server {tags {"other external:skip"}} {
         }
     }
 }
+
+start_server {tags {"other external:skip"} overrides {cluster-compatibility-sample-ratio 100}} {
+    test {Cross DB command is incompatible with cluster mode} {
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # SELECT with 0 is compatible command in cluster mode
+        assert_equal {OK} [r select 0]
+        assert_equal $incompatible_ops [s cluster_incompatible_ops]
+
+        # SELECT with nonzero is incompatible command in cluster mode
+        assert_equal {OK} [r select 1]
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # SWAPDB is incompatible command in cluster mode
+        assert_equal {OK} [r swapdb 0 1]
+        assert_equal [expr $incompatible_ops + 2] [s cluster_incompatible_ops]
+
+
+        # If destination db in COPY command is equal to source db, it is compatible
+        # with cluster mode, otherwise it is incompatible.
+        r select 0
+        r set key1 value1
+        set incompatible_ops [s cluster_incompatible_ops]
+        assert_equal {1} [r copy key1 key2{key1}] ;# destination db is equal to source db
+        assert_equal $incompatible_ops [s cluster_incompatible_ops]
+        assert_equal {1} [r copy key2{key1} key1 db 1] ;# destination db is not equal to source db
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # If destination db in MOVE command is not equal to source db, it is incompatible
+        # with cluster mode.
+        r set key3 value3
+        assert_equal {1} [r move key3 1]
+        assert_equal [expr $incompatible_ops + 2] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Function no-cluster flag is incompatible with cluster mode} {
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # no-cluster flag is incompatible with cluster mode
+        r function load {#!lua name=test
+            redis.register_function{function_name='f1', callback=function() return 'hello' end, flags={'no-cluster'}}
+        }
+        r fcall f1 0
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # It is compatible without no-cluster flag, should not increase the cluster_incompatible_ops
+        r function load {#!lua name=test2
+            redis.register_function{function_name='f2', callback=function() return 'hello' end}
+        }
+        r fcall f2 0
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Script no-cluster flag is incompatible with cluster mode} {
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # no-cluster flag is incompatible with cluster mode
+        r eval {#!lua flags=no-cluster
+                return 1
+            } 0
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # It is compatible without no-cluster flag, should not increase the cluster_incompatible_ops
+        r eval {#!lua
+                return 1
+            } 0
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {SORT command incompatible operations with cluster mode} {
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # If the BY pattern slot is not equal with the slot of keys, we consider
+        # an incompatible behavior, otherwise it is compatible, should not increase
+        # the cluster_incompatible_ops
+        r lpush mylist 1 2 3
+        for {set i 1} {$i < 4} {incr i} {
+            r set weight_$i [expr 4 - $i]
+        }
+        assert_equal {3 2 1} [r sort mylist BY weight_*]
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+        # weight{mylist}_* and mylist have the same slot
+        for {set i 1} {$i < 4} {incr i} {
+            r set weight{mylist}_$i [expr 4 - $i]
+        }
+        assert_equal {3 2 1} [r sort mylist BY weight{mylist}_*]
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # If the GET pattern slot is not equal with the slot of keys, we consider
+        # an incompatible behavior, otherwise it is compatible, should not increase
+        # the cluster_incompatible_ops
+        for {set i 1} {$i < 4} {incr i} {
+            r set object_$i o_$i
+        }
+        assert_equal {o_3 o_2 o_1} [r sort mylist BY weight{mylist}_* GET object_*]
+        assert_equal [expr $incompatible_ops + 2] [s cluster_incompatible_ops]
+        # object{mylist}_*, weight{mylist}_* and mylist have the same slot
+        for {set i 1} {$i < 4} {incr i} {
+            r set object{mylist}_$i o_$i
+        }
+        assert_equal {o_3 o_2 o_1} [r sort mylist BY weight{mylist}_* GET object{mylist}_*]
+        assert_equal [expr $incompatible_ops + 2] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Normal cross slot commands are incompatible with cluster mode} {
+        # Normal cross slot command
+        set incompatible_ops [s cluster_incompatible_ops]
+        r mset foo bar bar foo
+        r del foo bar
+        assert_equal [expr $incompatible_ops + 2] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Transaction is incompatible with cluster mode} {
+        set incompatible_ops [s cluster_incompatible_ops]
+
+        # Incomplete transaction
+        catch {r EXEC}
+        r multi
+        r exec
+        assert_equal $incompatible_ops [s cluster_incompatible_ops]
+
+        # Transaction, SET and DEL have keys with different slots
+        r multi
+        r set foo bar
+        r del bar
+        r exec
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Lua scripts are incompatible with cluster mode} {
+        # Lua script, declared keys have different slots, it is not a compatible operation
+        set incompatible_ops [s cluster_incompatible_ops]
+        r eval {#!lua
+            redis.call('mset', KEYS[1], 0, KEYS[2], 0)
+        } 2 foo bar
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # Lua script, no declared keys, but accessing keys have different slots,
+        # it is not a compatible operation
+        set incompatible_ops [s cluster_incompatible_ops]
+        r eval {#!lua
+            redis.call('mset', 'foo', 0, 'bar', 0)
+        } 0
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # Lua script, declared keys have the same slot, but accessing keys
+        # have different slots in one command, even with flag 'allow-cross-slot-keys',
+        # it still is not a compatible operation
+        set incompatible_ops [s cluster_incompatible_ops]
+        r eval {#!lua flags=allow-cross-slot-keys
+            redis.call('mset', 'foo', 0, 'bar', 0)
+            redis.call('mset', KEYS[1], 0, KEYS[2], 0)
+        } 2 foo bar{foo}
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+
+        # Lua script, declared keys have the same slot, but accessing keys have different slots
+        # in multiple commands, and with flag 'allow-cross-slot-keys', it is a compatible operation
+        set incompatible_ops [s cluster_incompatible_ops]
+        r eval {#!lua flags=allow-cross-slot-keys
+            redis.call('set', 'foo', 0)
+            redis.call('set', 'bar', 0)
+            redis.call('mset', KEYS[1], 0, KEYS[2], 0)
+        } 2 foo bar{foo}
+        assert_equal $incompatible_ops [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {Shard subscribe commands are incompatible with cluster mode} {
+        set rd1 [redis_deferring_client]
+        set incompatible_ops [s cluster_incompatible_ops]
+        assert_equal {1 2} [ssubscribe $rd1 {foo bar}]
+        assert_equal [expr $incompatible_ops + 1] [s cluster_incompatible_ops]
+    } {} {cluster:skip}
+
+    test {cluster-compatibility-sample-ratio configuration can work} {
+        # Disable cluster compatibility sampling, no increase in cluster_incompatible_ops
+        set incompatible_ops [s cluster_incompatible_ops]
+        r config set cluster-compatibility-sample-ratio 0
+        for {set i 0} {$i < 100} {incr i} {
+            r mset foo bar$i bar foo$i
+        }
+        # Enable cluster compatibility sampling again to show the metric
+        r config set cluster-compatibility-sample-ratio 1
+        assert_equal $incompatible_ops [s cluster_incompatible_ops]
+
+        # 100% sample ratio, all operations should increase cluster_incompatible_ops
+        set incompatible_ops [s cluster_incompatible_ops]
+        r config set cluster-compatibility-sample-ratio 100
+        for {set i 0} {$i < 100} {incr i} {
+            r mset foo bar$i bar foo$i
+        }
+        assert_equal [expr $incompatible_ops + 100] [s cluster_incompatible_ops]
+
+        # 30% sample ratio, cluster_incompatible_ops should increase between 20% and 40%
+        set incompatible_ops [s cluster_incompatible_ops]
+        r config set cluster-compatibility-sample-ratio 30
+        for {set i 0} {$i < 1000} {incr i} {
+            r mset foo bar$i bar foo$i
+        }
+        assert_range [s cluster_incompatible_ops] [expr $incompatible_ops + 200] [expr $incompatible_ops + 400]
+    } {} {cluster:skip}
+}