mir/ovs
2
0
Fork 0
mirror of https://github.com/openvswitch/ovs synced 2025-09-01 06:45:17 +00:00
Code Issues Releases Wiki Activity

stream: Add record/replay functionality.

Browse Source 
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>
This commit is contained in:
Ilya Maximets
2021-05-27 15:29:00 +02:00
parent 610ac1e82c
commit fae1ae0434
5 changed files with 507 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
lib/automake.mk
View File

@@ -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 \

5
lib/stream-provider.h
View File

@@ -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 */

456
lib/stream-replay.c Normal file
View File

@@ -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,
};

41
lib/stream.c
View File

@@ -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)) {
*classp = class;
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
: n == 0 ? 0
: (stream->class->recv)(stream, buffer, n));
retval = retval ? -retval
: n == 0 ? 0
: (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
: n == 0 ? 0
: (stream->class->send)(stream, buffer, n));
retval = retval ? -retval
: n == 0 ? 0
: (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)) {
*classp = class;
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;
}

12
lib/stream.h
View File

@@ -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 */