WIP: kvs: support configuration of max transaction count

doc/man5/flux-config-kvs.rst

@@ -29,6 +29,20 @@ gc-threshold
    point. (Default: garbage collection must be manually requested with
    `flux-shutdown --gc`).
 
+transaction-max-ops
+   (optional) Sets the maximum number of transactions that can be
+   performed in a single KVS commit.  This configuration is to prevent
+   a single transaction from taking up too much of the KVS's time
+   (i.e. to prevent a denial-of-service from a large transaction).  By
+   default the maximum is 65536.
+
+fence-max-ops
+   (optional) Sets the maximum number of operations that can be
+   performed in total by a KVS fence.  This configuration is to
+   prevent a single fence from taking up too much of the KVS's time
+   (i.e. to prevent a denial-of-service from a large transaction).  By
+   default the maximum is 1048576.
+
 
 EXAMPLE
 =======

doc/test/spell.en.pws

@@ -942,3 +942,4 @@ SATTR
 myprogram
 unref
 sigprocmask
+KVS's

src/modules/kvs/kvs.c

@@ -64,6 +64,10 @@
  */
 const double max_namespace_age = 3600.;
 
+/* Transaction max ops
+ */
+const uint64_t kvs_transaction_max_ops = 65536;
+
 struct kvs_ctx {
     struct cache *cache;    /* blobref => cache_entry */
     kvsroot_mgr_t *krm;
@@ -74,6 +78,7 @@
     flux_watcher_t *idle_w;
     flux_watcher_t *check_w;
     int transaction_merge;
+    uint64_t transaction_max_ops;
     bool events_init;            /* flag */
     char *hash_name;
     unsigned int seq;           /* for commit transactions */
@@ -103,6 +108,9 @@
 static void work_queue_check_append (struct kvs_ctx *ctx,
                                      struct kvsroot *root);
 static void kvstxn_apply (kvstxn_t *kt);
+static int max_ops_parse (struct kvs_ctx *ctx,
+                          const flux_conf_t *conf,
+                          flux_error_t *errp);
 
 /*
  * kvs_ctx functions
@@ -182,6 +190,7 @@
             goto error;
     }
     ctx->transaction_merge = 1;
+    ctx->transaction_max_ops = kvs_transaction_max_ops;
     if (!(ctx->requests = msg_hash_create (MSG_HASH_TYPE_UUID_MATCHTAG)))
         goto error;
     list_head_init (&ctx->work_queue);
@@ -1772,6 +1781,11 @@
         goto error;
     }
 
+    if (json_array_size (ops) > ctx->transaction_max_ops) {
+        errno = E2BIG;
+        goto error;
+    }
+
     if (!(root = getroot (ctx, ns, mh, msg, NULL, &stall))) {
         if (stall) {
             request_tracking_add (ctx, msg);
@@ -2700,6 +2714,10 @@
 
     if (flux_conf_reload_decode (msg, &conf) < 0)
         goto error;
+    if (max_ops_parse (ctx, conf, &error) < 0) {
+        errstr = error.text;
+        goto error;
+    }
     if (kvs_checkpoint_reload (ctx->kcp, conf, &error) < 0) {
         errstr = error.text;
         goto error;
@@ -2844,11 +2862,39 @@
     FLUX_MSGHANDLER_TABLE_END,
 };
 
+static int max_ops_parse (struct kvs_ctx *ctx,
+                          const flux_conf_t *conf,
+                          flux_error_t *errp)
+{
+    uint64_t t_max_ops = kvs_transaction_max_ops;
+    flux_error_t error;
+    if (flux_conf_unpack (conf,
+                          &error,
+                          "{s?{s?I}}",
+                          "kvs",
+                          "transaction-max-ops", &t_max_ops) < 0) {
+        errprintf (errp, "error reading config for kvs: %s", error.text);
+        return -1;
+    }
+    if (t_max_ops <= 0) {
+        errprintf (errp, "kvs transaction-max-ops invalid");
+        return -1;
+    }
+    ctx->transaction_max_ops = t_max_ops;
+    return 0;
+}
+
 static int process_config (struct kvs_ctx *ctx)
 {
     flux_error_t error;
+    const flux_conf_t *conf = flux_get_conf (ctx->h);
+
+    if (max_ops_parse (ctx, conf, &error) < 0) {
+        flux_log (ctx->h, LOG_ERR, "%s", error.text);
+        return -1;
+    }
     if (kvs_checkpoint_config_parse (ctx->kcp,
-                                     flux_get_conf (ctx->h),
+                                     conf,
                                      &error) < 0) {
         flux_log (ctx->h, LOG_ERR, "%s", error.text);
         return -1;

t/t1005-kvs-security.t

@@ -412,4 +412,54 @@ test_expect_success 'kvs: no pending requests at end of tests' '
 	test $pendingcount -eq 0
 '
 
+#
+# test transaction-max-ops
+#
+
+test_expect_success 'configure illegal transaction-max-ops' '
+	test_must_fail flux config load <<-EOF
+	[kvs]
+	transaction-max-ops = "foobar"
+	EOF
+'
+
+test_expect_success 'configure bad transaction-max-ops' '
+	test_must_fail flux config load <<-EOF
+	[kvs]
+	transaction-max-ops = 0
+	EOF
+'
+
+test_expect_success 'configure small transaction-max-ops' '
+	flux exec flux config load <<-EOF
+	[kvs]
+	transaction-max-ops = 3
+	EOF
+'
+
+# N.B. flux kvs put will place each key=val on command line into 1
+# transaction
+
+test_expect_success 'kvs: txns of small size work' '
+	flux kvs put test.a=1 &&
+	flux kvs put test.b=1 test.c=1 &&
+	flux kvs put test.d=1 test.e=1 test.f=1
+'
+
+test_expect_success 'kvs: txns of small size work (not rank 0)' '
+	flux exec -r 1 flux kvs put test.a=1 &&
+	flux exec -r 1 flux kvs put test.b=1 test.c=1 &&
+	flux exec -r 1 flux kvs put test.d=1 test.e=1 test.f=1
+'
+
+test_expect_success 'kvs: txns above limit fail' '
+	test_must_fail flux kvs put test.a=2 test.b=2 test.c=2 test.d=2 2> fence1.err &&
+	grep "Argument list too long" fence1.err
+'
+
+test_expect_success 'kvs: txns above limit fail (not rank 0)' '
+	test_must_fail flux exec -r 1 flux kvs put test.a=3 test.b=3 test.c=3 test.d=3 2> fence2.err &&
+	grep "Argument list too long" fence2.err
+'
+
 test_done