Add VSL rate limiting

This adds rate limiting to varnishncsa and varnishlog.

Rate limiting is done on a per-transaction basis, respective to the
grouping mode selected. I.e. for -g request the limit will apply on a
per-request basis, -g raw on a per-record basis, etc.

Rate limit is specified as -R <limit>[/duration]. Default period if not
specified is seconds ('s').

bin/varnishlog/varnishlog_options.h

@@ -70,6 +70,7 @@ VUT_OPT_n
 VUT_GLOBAL_OPT_P
 VUT_OPT_q
 VUT_OPT_r
+VSL_OPT_R
 VUT_OPT_t
 VSL_OPT_T
 VSL_OPT_v

bin/varnishncsa/varnishncsa_options.h

@@ -88,6 +88,7 @@ VUT_OPT_n
 VUT_GLOBAL_OPT_P
 VUT_OPT_q
 VUT_OPT_r
+VSL_OPT_R
 VUT_OPT_t
 VUT_GLOBAL_OPT_V
 NCSA_OPT_w

include/vapi/vapi_options.h

@@ -77,6 +77,22 @@
 	    " running queries. Defaults to 1000 transactions."		\
 	)
 
+#define VSL_OPT_R							\
+	VOPT("R:", "[-R <limit[/duration]>]", "Output rate limit",	\
+	    "Restrict the output to the specified limit."		\
+	    " Transactions exceeding the limit will be suppressed."	\
+	    " The limit is specified as the maximum number of"		\
+	    " transactions (with respect to the chosen grouping"	\
+	    " method) and an optional time period. If no duration"	\
+	    " is specified, a default of ``s`` is used. The duration"	\
+	    " field can be formatted as in VCL (e.g. ``-R 10/2m``) or"	\
+	    " as a simple time period without the prefix (e.g."		\
+	    " ``-R 5/m``)."						\
+	    " When in ``-g raw`` grouping mode, this setting can"	\
+	    " not be used alongside ``-i``, ``-I``, ``-x`` or "		\
+	    "``-X``, and we advise using ``-q`` instead."		\
+	)
+
 #define VSL_OPT_T							\
 	VOPT("T:", "[-T <seconds>]", "Transaction end timeout",		\
 	    "Sets the transaction timeout in seconds. This defines the"	\

lib/libvarnishapi/vsl_api.h

@@ -84,6 +84,8 @@ struct VSL_data {
 	int				c_opt;
 	int				C_opt;
 	int				L_opt;
+	int				R_opt_l;
+	vtim_dur			R_opt_p;
 	double				T_opt;
 	int				v_opt;
 };

lib/libvarnishapi/vsl_arg.c

@@ -300,6 +300,41 @@ vsl_IX_arg(struct VSL_data *vsl, int opt, const char *arg)
 	return (1);
 }
 
+static int
+vsl_R_arg(struct VSL_data *vsl, const char *arg)
+{
+	char buf[32] = "";
+	char *p;
+	long l;
+
+	AN(arg);
+	CHECK_OBJ_NOTNULL(vsl, VSL_MAGIC);
+
+	l = strtol(arg, &p, 0);
+	if (l <= 0 || l > INT_MAX)
+		return (vsl_diag(vsl, "-R: Range error"));
+	vsl->R_opt_l = l;
+	assert(p != arg);
+	AN(p);
+	if (*p == '\0') {
+		vsl->R_opt_p = 1.0;
+		return (1);
+	}
+	if (*p != '/' || p[1] == '\0')
+		return (vsl_diag(vsl, "-R: Syntax error"));
+	p++;
+	if (strlen(p) > sizeof(buf) - 2)
+		return (vsl_diag(vsl, "-R: Syntax error"));
+	if (!isdigit(*p))
+		strcat(buf, "1");
+	strcat(buf, p);
+	vsl->R_opt_p = VNUM_duration(buf);
+	if (isnan(vsl->R_opt_p) || vsl->R_opt_p <= 0.0)
+		return (vsl_diag(vsl,
+			"-R: Syntax error: Invalid duration"));
+	return (1);
+}
+
 int
 VSL_Arg(struct VSL_data *vsl, int opt, const char *arg)
 {
@@ -334,6 +369,8 @@ VSL_Arg(struct VSL_data *vsl, int opt, const char *arg)
 			return (vsl_diag(vsl, "-L: Range error"));
 		vsl->L_opt = (int)l;
 		return (1);
+	case 'R':
+		return (vsl_R_arg(vsl, arg));
 	case 'T':
 		AN(arg);
 		d = VNUM(arg);

lib/libvarnishapi/vsl_dispatch.c

@@ -194,6 +194,10 @@ struct VSLQ {
 	VTAILQ_HEAD(,vtx)	cache;
 	unsigned		n_cache;
 
+	/* Rate limiting */
+	double			credits;
+	vtim_mono		last_use;
+
 	/* Raw mode */
 	struct {
 		struct vslc_raw		c;
@@ -908,10 +912,33 @@ vtx_force(struct VSLQ *vslq, struct vtx *vtx, const char *reason)
 	AN(vtx->flags & VTX_F_COMPLETE);
 }
 
+static int
+vslq_ratelimit(struct VSLQ *vslq)
+{
+	vtim_mono now;
+	vtim_dur delta;
+
+	CHECK_OBJ_NOTNULL(vslq, VSLQ_MAGIC);
+	CHECK_OBJ_NOTNULL(vslq->vsl, VSL_MAGIC);
+
+	now = VTIM_mono();
+	delta = now - vslq->last_use;
+	vslq->credits += (delta / vslq->vsl->R_opt_p) * vslq->vsl->R_opt_l;
+	if (vslq->credits > vslq->vsl->R_opt_l)
+		vslq->credits = vslq->vsl->R_opt_l;
+	vslq->last_use = now;
+
+	if (vslq->credits < 1.0)
+		return (0);
+
+	vslq->credits -= 1.0;
+	return (1);
+}
+
 /* Build transaction array, do the query and callback. Returns 0 or the
    return value from func */
 static int
-vslq_callback(const struct VSLQ *vslq, struct vtx *vtx, VSLQ_dispatch_f *func,
+vslq_callback(struct VSLQ *vslq, struct vtx *vtx, VSLQ_dispatch_f *func,
     void *priv)
 {
 	unsigned n = vtx->n_descend + 1;
@@ -973,6 +1000,9 @@ vslq_callback(const struct VSLQ *vslq, struct vtx *vtx, VSLQ_dispatch_f *func,
 	if (vslq->query != NULL && !vslq_runquery(vslq->query, ptrans))
 		return (0);
 
+	if (vslq->vsl->R_opt_l != 0 && !vslq_ratelimit(vslq))
+		return (0);
+
 	/* Callback */
 	return ((func)(vslq->vsl, ptrans, priv));
 }
@@ -1078,6 +1108,10 @@ VSLQ_New(struct VSL_data *vsl, struct VSL_cursor **cp,
 	}
 	vslq->grouping = grouping;
 	vslq->query = query;
+	if (vslq->vsl->R_opt_l != 0) {
+		vslq->last_use = VTIM_mono();
+		vslq->credits = 1;
+	}
 
 	/* Setup normal mode */
 	VRBT_INIT(&vslq->tree);
@@ -1195,6 +1229,9 @@ vslq_raw(struct VSLQ *vslq, VSLQ_dispatch_f *func, void *priv)
 	    !vslq_runquery(vslq->query, vslq->raw.ptrans))
 		return (r);
 
+	if (vslq->vsl->R_opt_l != 0 && !vslq_ratelimit(vslq))
+		return (r);
+
 	i = (func)(vslq->vsl, vslq->raw.ptrans, priv);
 	if (i)
 		return (i);