stream: Add record/replay functionality.

For debugging purposes it is useful to be able to record all the
incoming transactions and commands and replay them locally under
debugger or with additional logging enabled.  This patch introduces
ability to record all the incoming stream data and replay it via new
stream provider named 'stream-replay'.  During the record phase all
the incoming stream data written to special replay_* files in the
application rundir.  On replay phase instead of opening real streams
application will open replay_* files and read all the incoming data
directly from them.

If enabled for ovsdb-server, for example, this allows to record all
the connections and transactions from the big setup and replay them
locally afterwards to debug the behaviour or test performance.

To start application in recording mode there is a --record cmdline
option. --replay is to replay previously recorded streams.

Current version doesn't work well with time-based stream events like
inactivity probes or any other events generated internally.  This is
a point for further improvement.

Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
Acked-by: Dumitru Ceara <dceara@redhat.com>

lib/automake.mk

@@ -312,6 +312,7 @@ lib_libopenvswitch_la_SOURCES = \
 	lib/stream-fd.c \
 	lib/stream-fd.h \
 	lib/stream-provider.h \
+	lib/stream-replay.c \
 	lib/stream-ssl.h \
 	lib/stream-tcp.c \
 	lib/stream.c \

lib/stream-provider.h

@@ -18,6 +18,7 @@
 #define STREAM_PROVIDER_H 1
 
 #include <sys/types.h>
+#include "ovs-replay.h"
 #include "stream.h"
 
 /* Active stream connection. */
@@ -29,6 +30,7 @@ struct stream {
     const struct stream_class *class;
     int state;
     int error;
+    replay_file_t replay_wfd;
     char *name;
     char *peer_id;
 };
@@ -133,6 +135,7 @@ struct pstream {
     const struct pstream_class *class;
     char *name;
     ovs_be16 bound_port;
+    replay_file_t replay_wfd;
 };
 
 void pstream_init(struct pstream *, const struct pstream_class *, char *name);
@@ -200,5 +203,7 @@ extern const struct pstream_class pwindows_pstream_class;
 extern const struct stream_class ssl_stream_class;
 extern const struct pstream_class pssl_pstream_class;
 #endif
+extern const struct stream_class replay_stream_class;
+extern const struct pstream_class preplay_pstream_class;
 
 #endif /* stream-provider.h */

lib/stream-replay.c

@@ -0,0 +1,456 @@
+/*
+ * Copyright (c) 2021, Red Hat, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at:
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include <config.h>
+#include <ctype.h>
+#include <errno.h>
+#include <poll.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+#include "ovs-atomic.h"
+#include "ovs-replay.h"
+#include "util.h"
+#include "stream-provider.h"
+#include "stream.h"
+#include "openvswitch/poll-loop.h"
+#include "openvswitch/vlog.h"
+
+VLOG_DEFINE_THIS_MODULE(stream_replay);
+
+static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(10, 25);
+
+/* Active replay stream. */
+
+struct stream_replay {
+    struct stream stream;
+    replay_file_t f;
+    int seqno;
+};
+
+const struct stream_class replay_stream_class;
+
+/* Creates a new stream named 'name' that will emulate sending and receiving
+ * data using replay file and stores a pointer to the stream in '*streamp'.
+ *
+ * Returns 0 if successful, otherwise a positive errno value. */
+static int
+new_replay_stream(const char *name, struct stream **streamp)
+{
+    struct stream_replay *s;
+    int seqno = 0, error = 0, open_result;
+    replay_file_t f;
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, &seqno);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to open stream.", name);
+        goto unlock;
+    }
+
+    error = ovs_replay_read(f, NULL, 0, &open_result, &seqno, true);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to read 'open' record.", name);
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    if (open_result) {
+        error = -open_result;
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    s = xmalloc(sizeof *s);
+    stream_init(&s->stream, &replay_stream_class, 0, xstrdup(name));
+    s->f = f;
+    s->seqno = seqno;
+    *streamp = &s->stream;
+unlock:
+    ovs_replay_unlock();
+    return error;
+}
+
+static struct stream_replay *
+stream_replay_cast(struct stream *stream)
+{
+    stream_assert_class(stream, &replay_stream_class);
+    return CONTAINER_OF(stream, struct stream_replay, stream);
+}
+
+void
+stream_replay_open_wfd(struct stream *s, int open_result, const char *name)
+{
+    int state = ovs_replay_get_state();
+    int error = 0;
+    replay_file_t f;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, NULL);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to open replay file for stream.", name);
+        ovs_replay_unlock();
+        return;
+    }
+    ovs_replay_unlock();
+
+    if (ovs_replay_write(f, NULL, -open_result, true)) {
+        VLOG_ERR_RL(&rl, "%s: failed to write 'open' failure: %d",
+                    s->name, open_result);
+    }
+    if (open_result) {
+        /* We recorded failure to open the stream. */
+        ovs_replay_file_close(f);
+    } else {
+        s->replay_wfd = f;
+    }
+}
+
+void
+stream_replay_write(struct stream *s, const void *buffer, int n, bool is_read)
+{
+    int state = ovs_replay_get_state();
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    if (ovs_replay_write(s->replay_wfd, buffer, n, is_read)) {
+        VLOG_ERR_RL(&rl, "%s: failed to write buffer.", s->name);
+    }
+}
+
+void
+stream_replay_close_wfd(struct stream *s)
+{
+    if (s->replay_wfd) {
+        ovs_replay_file_close(s->replay_wfd);
+    }
+}
+
+static int
+stream_replay_open(const char *name, char *suffix OVS_UNUSED,
+                   struct stream **streamp, uint8_t dscp OVS_UNUSED)
+{
+    return new_replay_stream(name, streamp);
+}
+
+static void
+stream_replay_close(struct stream *stream)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    ovs_replay_file_close(s->f);
+    free(s);
+}
+
+static ssize_t
+stream_replay_recv(struct stream *stream, void *buffer, size_t n)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
+    int error, len;
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || !ovs_replay_seqno_is_read(s->seqno)) {
+        error = EAGAIN;
+        goto unlock;
+    }
+
+    error = ovs_replay_read(s->f, buffer, n, &len, &s->seqno, true);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to read from replay file.", stream->name);
+        goto unlock;
+    }
+
+unlock:
+    ovs_replay_unlock();
+    return error ? -error : len;
+}
+
+static ssize_t
+stream_replay_send(struct stream *stream OVS_UNUSED,
+                   const void *buffer OVS_UNUSED, size_t n)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    int norm_seqno = ovs_replay_normalized_seqno(s->seqno);
+    int error, len;
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || ovs_replay_seqno_is_read(s->seqno)) {
+        error = EAGAIN;
+        goto unlock;
+    }
+
+    error = ovs_replay_read(s->f, NULL, 0, &len, &s->seqno, false);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to read from replay file.", stream->name);
+        goto unlock;
+    }
+    ovs_assert(len < 0 || len <= n);
+
+unlock:
+    ovs_replay_unlock();
+    return error ? -error : len;
+}
+
+static void
+stream_replay_wait(struct stream *stream, enum stream_wait_type wait)
+{
+    struct stream_replay *s = stream_replay_cast(stream);
+    switch (wait) {
+    case STREAM_CONNECT:
+        /* Connect does nothing and always available. */
+        poll_immediate_wake();
+        break;
+
+    case STREAM_SEND:
+        if (s->seqno != INT_MAX && !ovs_replay_seqno_is_read(s->seqno)) {
+            /* Stream waits for write. */
+            poll_immediate_wake();
+        }
+        break;
+
+    case STREAM_RECV:
+        if (s->seqno != INT_MAX && ovs_replay_seqno_is_read(s->seqno)) {
+            /* We still have something to read. */
+            poll_immediate_wake();
+        }
+        break;
+
+    default:
+        OVS_NOT_REACHED();
+    }
+}
+
+const struct stream_class replay_stream_class = {
+    "replay",                   /* name */
+    false,                      /* needs_probes */
+    stream_replay_open,         /* open */
+    stream_replay_close,        /* close */
+    NULL,                       /* connect */
+    stream_replay_recv,         /* recv */
+    stream_replay_send,         /* send */
+    NULL,                       /* run */
+    NULL,                       /* run_wait */
+    stream_replay_wait,         /* wait */
+};
+
+/* Passive replay stream. */
+
+struct replay_pstream
+{
+    struct pstream pstream;
+    replay_file_t f;
+    int seqno;
+};
+
+const struct pstream_class preplay_pstream_class;
+
+static struct replay_pstream *
+replay_pstream_cast(struct pstream *pstream)
+{
+    pstream_assert_class(pstream, &preplay_pstream_class);
+    return CONTAINER_OF(pstream, struct replay_pstream, pstream);
+}
+
+/* Creates a new pstream named 'name' that will accept new replay connections
+ * reading them from the replay file and stores a pointer to the stream in
+ * '*pstreamp'.
+ *
+ * Returns 0 if successful, otherwise a positive errno value. */
+static int
+pstream_replay_listen(const char *name, char *suffix OVS_UNUSED,
+                      struct pstream **pstreamp, uint8_t dscp OVS_UNUSED)
+{
+    int seqno = 0, error = 0, listen_result;
+    replay_file_t f;
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, &seqno);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to open pstream.", name);
+        goto unlock;
+    }
+
+    error = ovs_replay_read(f, NULL, 0, &listen_result, &seqno, true);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to read 'listen' record.", name);
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    if (listen_result) {
+        error = -listen_result;
+        ovs_replay_file_close(f);
+        goto unlock;
+    }
+
+    struct replay_pstream *ps = xmalloc(sizeof *ps);
+    pstream_init(&ps->pstream, &preplay_pstream_class, xstrdup(name));
+    ps->f = f;
+    ps->seqno = seqno;
+    *pstreamp = &ps->pstream;
+unlock:
+    ovs_replay_unlock();
+    return error;
+}
+
+void
+pstream_replay_open_wfd(struct pstream *ps, int listen_result,
+                        const char *name)
+{
+    int state = ovs_replay_get_state();
+    int error = 0;
+    replay_file_t f;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    ovs_replay_lock();
+    error = ovs_replay_file_open(name, &f, NULL);
+    if (error) {
+        VLOG_ERR_RL(&rl, "%s: failed to open replay file for pstream.", name);
+        ovs_replay_unlock();
+        return;
+    }
+    ovs_replay_unlock();
+
+    if (ovs_replay_write(f, NULL, -listen_result, true)) {
+        VLOG_ERR_RL(&rl, "%s: failed to write 'listen' result: %d",
+                    ps->name, listen_result);
+    }
+
+    if (listen_result) {
+        /* We recorded failure to open the stream. */
+        ovs_replay_file_close(f);
+    } else {
+        ps->replay_wfd = f;
+    }
+}
+
+void
+pstream_replay_write_accept(struct pstream *ps, const struct stream *s,
+                            int accept_result)
+{
+    int state = ovs_replay_get_state();
+    int len;
+
+    if (OVS_LIKELY(state != OVS_REPLAY_WRITE)) {
+        return;
+    }
+
+    if (!accept_result) {
+        len = strlen(s->name);
+        if (ovs_replay_write(ps->replay_wfd, s->name, len, true)) {
+            VLOG_ERR_RL(&rl, "%s: failed to write accept name: %s",
+                        ps->name, s->name);
+        }
+    } else if (ovs_replay_write(ps->replay_wfd, NULL, -accept_result, true)) {
+        VLOG_ERR_RL(&rl, "%s: failed to write 'accept' failure: %d",
+                    ps->name, accept_result);
+    }
+}
+
+void
+pstream_replay_close_wfd(struct pstream *ps)
+{
+    if (ps->replay_wfd) {
+        ovs_replay_file_close(ps->replay_wfd);
+    }
+}
+
+static void
+pstream_replay_close(struct pstream *pstream)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+
+    ovs_replay_file_close(ps->f);
+    free(ps);
+}
+
+#define MAX_NAME_LEN 65536
+
+static int
+pstream_replay_accept(struct pstream *pstream, struct stream **new_streamp)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+    int norm_seqno = ovs_replay_normalized_seqno(ps->seqno);
+    int retval, len;
+    char name[MAX_NAME_LEN];
+
+    ovs_replay_lock();
+    ovs_assert(norm_seqno >= ovs_replay_seqno());
+
+    if (norm_seqno != ovs_replay_seqno()
+        || !ovs_replay_seqno_is_read(ps->seqno)) {
+        retval = EAGAIN;
+        ovs_replay_unlock();
+        goto exit;
+    }
+
+    retval = ovs_replay_read(ps->f, name, MAX_NAME_LEN - 1,
+                             &len, &ps->seqno, true);
+    if (retval) {
+        VLOG_ERR_RL(&rl, "%s: failed to read from replay file.",
+                    pstream->name);
+        ovs_replay_unlock();
+        goto exit;
+    }
+
+    ovs_replay_unlock();
+
+    if (len > 0) {
+        name[len] = 0;
+        retval = new_replay_stream(name, new_streamp);
+    } else {
+        retval = -len;
+    }
+exit:
+    return retval;
+}
+
+static void
+pstream_replay_wait(struct pstream *pstream)
+{
+    struct replay_pstream *ps = replay_pstream_cast(pstream);
+
+    if (ps->seqno != INT_MAX) {
+        /* Replay always has something to say. */
+        poll_immediate_wake();
+    }
+}
+
+const struct pstream_class preplay_pstream_class = {
+    "preplay",
+    false,
+    pstream_replay_listen,
+    pstream_replay_close,
+    pstream_replay_accept,
+    pstream_replay_wait,
+};

lib/stream.c

@@ -33,6 +33,7 @@
 #include "openvswitch/ofp-print.h"
 #include "openvswitch/ofpbuf.h"
 #include "openvswitch/vlog.h"
+#include "ovs-replay.h"
 #include "ovs-thread.h"
 #include "packets.h"
 #include "openvswitch/poll-loop.h"
@@ -184,7 +185,11 @@ stream_lookup_class(const char *name, const struct stream_class **classp)
         const struct stream_class *class = stream_classes[i];
         if (strlen(class->name) == prefix_len
             && !memcmp(class->name, name, prefix_len)) {
+            if (ovs_replay_get_state() == OVS_REPLAY_READ) {
+                *classp = &replay_stream_class;
+            } else {
                 *classp = class;
+            }
             return 0;
         }
     }
@@ -227,6 +232,8 @@ stream_open(const char *name, struct stream **streamp, uint8_t dscp)
     suffix_copy = xstrdup(strchr(name, ':') + 1);
     error = class->open(name, suffix_copy, &stream, dscp);
     free(suffix_copy);
+
+    stream_replay_open_wfd(stream, error, name);
     if (error) {
         goto error;
     }
@@ -295,6 +302,7 @@ stream_close(struct stream *stream)
     if (stream != NULL) {
         char *name = stream->name;
         char *peer_id = stream->peer_id;
+        stream_replay_close_wfd(stream);
         (stream->class->close)(stream);
         free(name);
         free(peer_id);
@@ -367,9 +375,13 @@ int
 stream_recv(struct stream *stream, void *buffer, size_t n)
 {
     int retval = stream_connect(stream);
-    return (retval ? -retval
+
+    retval = retval ? -retval
              : n == 0 ? 0
-            : (stream->class->recv)(stream, buffer, n));
+             : (stream->class->recv)(stream, buffer, n);
+
+    stream_replay_write(stream, buffer, retval, true);
+    return retval;
 }
 
 /* Tries to send up to 'n' bytes of 'buffer' on 'stream', and returns:
@@ -385,9 +397,12 @@ int
 stream_send(struct stream *stream, const void *buffer, size_t n)
 {
     int retval = stream_connect(stream);
-    return (retval ? -retval
+    retval = retval ? -retval
              : n == 0 ? 0
-            : (stream->class->send)(stream, buffer, n));
+             : (stream->class->send)(stream, buffer, n);
+
+    stream_replay_write(stream, buffer, retval, false);
+    return retval;
 }
 
 /* Allows 'stream' to perform maintenance activities, such as flushing
@@ -482,7 +497,11 @@ pstream_lookup_class(const char *name, const struct pstream_class **classp)
         const struct pstream_class *class = pstream_classes[i];
         if (strlen(class->name) == prefix_len
             && !memcmp(class->name, name, prefix_len)) {
+            if (ovs_replay_get_state() == OVS_REPLAY_READ) {
+                *classp = &preplay_pstream_class;
+            } else {
                 *classp = class;
+            }
             return 0;
         }
     }
@@ -544,6 +563,8 @@ pstream_open(const char *name, struct pstream **pstreamp, uint8_t dscp)
     suffix_copy = xstrdup(strchr(name, ':') + 1);
     error = class->listen(name, suffix_copy, &pstream, dscp);
     free(suffix_copy);
+
+    pstream_replay_open_wfd(pstream, error, name);
     if (error) {
         goto error;
     }
@@ -571,6 +592,7 @@ pstream_close(struct pstream *pstream)
 {
     if (pstream != NULL) {
         char *name = pstream->name;
+        pstream_replay_close_wfd(pstream);
         (pstream->class->close)(pstream);
         free(name);
     }
@@ -588,9 +610,12 @@ pstream_accept(struct pstream *pstream, struct stream **new_stream)
     int retval = (pstream->class->accept)(pstream, new_stream);
     if (retval) {
         *new_stream = NULL;
+        pstream_replay_write_accept(pstream, NULL, retval);
     } else {
         ovs_assert((*new_stream)->state != SCS_CONNECTING
                    || (*new_stream)->class->connect);
+        pstream_replay_write_accept(pstream, *new_stream, 0);
+        stream_replay_open_wfd(*new_stream, 0, (*new_stream)->name);
     }
     return retval;
 }

lib/stream.h

@@ -94,4 +94,16 @@ enum stream_content_type {
 void stream_report_content(const void *, ssize_t, enum stream_content_type,
                            struct vlog_module *, const char *stream_name);
 
+
+/* Stream replay helpers. */
+void stream_replay_open_wfd(struct stream *, int open_result,
+                            const char *name);
+void pstream_replay_open_wfd(struct pstream *, int listen_result,
+                             const char *name);
+void stream_replay_close_wfd(struct stream *);
+void pstream_replay_close_wfd(struct pstream *);
+void stream_replay_write(struct stream *, const void *, int, bool is_read);
+void pstream_replay_write_accept(struct pstream *, const struct stream *,
+                                 int accept_result);
+
 #endif /* stream.h */