[Tarantool-patches] [RFC v3 3/3] relay: provide information about downstream lag

[Cyrill Gorcunov via Tarantool-patches <tarantool-patches@dev.tarantool.org>]

We already have `box.replication.upstream.lag` entry for monitoring
sake. Same time in synchronous replication timeouts are key properties
for quorum gathering procedure. Thus we would like to know how long
it took of a transaction to traverse `initiator WAL -> network ->
remote applier -> ACK` path.

In this patch we use new applier's functionality which returns us
the timestamp of first written xrow in a transaction such that we
can calculate the downstream lag.

Closes #5447

Signed-off-by: Cyrill Gorcunov <gorcunov@gmail.com>
---
 src/box/lua/info.c |  3 +++
 src/box/relay.cc   | 46 ++++++++++++++++++++++++++++++++++++++++++----
 src/box/relay.h    |  3 +++
 3 files changed, 48 insertions(+), 4 deletions(-)

diff --git a/src/box/lua/info.c b/src/box/lua/info.c
index 0eb48b823..b72dd9915 100644
--- a/src/box/lua/info.c
+++ b/src/box/lua/info.c
@@ -143,6 +143,9 @@ lbox_pushrelay(lua_State *L, struct relay *relay)
 		lua_pushnumber(L, ev_monotonic_now(loop()) -
 			       relay_last_row_time(relay));
 		lua_settable(L, -3);
+		lua_pushstring(L, "lag");
+		lua_pushnumber(L, relay_peer_wal_lag(relay));
+		lua_settable(L, -3);
 		break;
 	case RELAY_STOPPED:
 	{
diff --git a/src/box/relay.cc b/src/box/relay.cc
index ff43c2fc7..6d880932a 100644
--- a/src/box/relay.cc
+++ b/src/box/relay.cc
@@ -158,6 +158,12 @@ struct relay {
 	struct stailq pending_gc;
 	/** Time when last row was sent to peer. */
 	double last_row_time;
+	/**
+	 * A time difference between moment when we
+	 * wrote the row in a local WAL and peer wrote
+	 * it in own WAL.
+	 */
+	double peer_wal_lag;
 	/** Relay sync state. */
 	enum relay_state state;
 
@@ -217,6 +223,12 @@ relay_last_row_time(const struct relay *relay)
 	return relay->last_row_time;
 }
 
+double
+relay_peer_wal_lag(const struct relay *relay)
+{
+	return relay->peer_wal_lag;
+}
+
 static void
 relay_send(struct relay *relay, struct xrow_header *packet);
 static void
@@ -614,15 +626,41 @@ relay_reader_f(va_list ap)
 	coio_create(&io, relay->io.fd);
 	ibuf_create(&ibuf, &cord()->slabc, 1024);
 	try {
-		while (!fiber_is_cancelled()) {
-			struct xrow_header xrow;
+		struct xrow_header xrow;
+		double prev_tm;
+
+		/*
+		 * Make a first read as a separate action because
+		 * we need previous timestamp from the xrow to
+		 * calculate delta from (to eliminate branching
+		 * in next reads).
+		 */
+		if (!fiber_is_cancelled()) {
 			coio_read_xrow_timeout_xc(&io, &ibuf, &xrow,
-					replication_disconnect_timeout());
+				replication_disconnect_timeout());
+			prev_tm = xrow.tm;
+		}
+
+		do {
 			/* vclock is followed while decoding, zeroing it. */
 			vclock_create(&relay->recv_vclock);
 			xrow_decode_vclock_xc(&xrow, &relay->recv_vclock);
+			/*
+			 * Old instances do not report the timestamp.
+			 * Same time in case of idle cycles the xrow.tm
+			 * is the same so update lag only when new data
+			 * been acked.
+			 */
+			if (prev_tm != xrow.tm) {
+				double delta = ev_now(loop()) - xrow.tm;
+				relay->peer_wal_lag = delta;
+				prev_tm = xrow.tm;
+			}
 			fiber_cond_signal(&relay->reader_cond);
-		}
+
+			coio_read_xrow_timeout_xc(&io, &ibuf, &xrow,
+					replication_disconnect_timeout());
+		} while (!fiber_is_cancelled());
 	} catch (Exception *e) {
 		relay_set_error(relay, e);
 		fiber_cancel(relay_f);
diff --git a/src/box/relay.h b/src/box/relay.h
index b32e2ea2a..99e2179cb 100644
--- a/src/box/relay.h
+++ b/src/box/relay.h
@@ -93,6 +93,9 @@ relay_vclock(const struct relay *relay);
 double
 relay_last_row_time(const struct relay *relay);
 
+double
+relay_peer_wal_lag(const struct relay *relay);
+
 /**
  * Send a Raft update request to the relay channel. It is not
  * guaranteed that it will be delivered. The connection may break.
-- 
2.30.2