Skip to content

Commit

Permalink
Fix flaky capture service start (#20024)
Browse files Browse the repository at this point in the history
#### Why I did it

Dropping control character (message sent when XSUB connects to XPUB as part of ZMQ Proxy setup to notify that subscription has been made) in do capture has been flaky since control character is not guaranteed to be the first message sent if there are events (like event-down-ctr) being published to XSUB.

Scenarios

1) Control character is sent and is first message when starting capture service

`eventd#eventd#eventd: :- heartbeat_ctrl: Set heartbeat_ctrl pause=1`
`eventd#eventd#eventd: :- do_capture: Received subscription message when XSUB connects to XPUB`

2) Events like event-down ctr is sent before control character

`eventd#eventd#eventd: :- run: Dropping Message: 22 serialization::archive 18 17 sonic-events-host`
`eventd#eventd#eventd: :- run: Dropping Message: 22 serialization::archive 18 0 0 4 0 0 0 1 d 103 {"sonic-events-host:event-stopped-ctr":{"ctr_name":"EVENTD","timestamp":"2024-08-27T00:02:51.407518Z"}} 1 r 36 3357542f-bae1-458f-a804-660e620d21f5 1 s 1 9 1 t 19 1724716971407591080`
`heartbeat_ctrl: Set heartbeat_ctrl pause=1`
`do_capture: Received subscription message when XSUB connects to XPUB`

3) Control character is not sent at all

`eventd#eventd#eventd: :- heartbeat_ctrl: Set heartbeat_ctrl pause=1`

4) Control character is delayed and not caught when starting capture service, but is then caught after causing deserialize error.

`do_capture: Receiving event from source: 22 serialization::archive 18 17 sonic-events-host, will read second part of event`
`deserialize: deserialize Failed: input stream errorstr[0:64]:(#1) data type: std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >&`
`zmq_read_part: Failed to deserialize part rc=-2`
`zmq_read_part: last:errno=11`
`zmq_message_read: Failure to read part1 rc=-2`
`zmq_message_read: last:errno=11`

We can cover these scenarios by just dropping the control character inside zmq_message_read as part of events_common in swsscommon (different PR). In this PR we will remove such handling logic and make sure that empty events that will be sent by control character are ignored.

##### Work item tracking
- Microsoft ADO **(number only)**:28728116

#### How I did it

Remove logic for handling control character

#### How to verify it

UT and sonic-mgmt test cases.
  • Loading branch information
zbud-msft authored and mssonicbld committed Sep 4, 2024
1 parent 0c95bfa commit a27674c
Show file tree
Hide file tree
Showing 2 changed files with 9 additions and 38 deletions.
24 changes: 3 additions & 21 deletions src/sonic-eventd/src/eventd.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -302,7 +302,9 @@ static bool
validate_event(const internal_event_t &event, runtime_id_t &rid, sequence_t &seq)
{
bool ret = false;

if(event.empty()) {
return ret;
}
internal_event_t::const_iterator itc_r, itc_s, itc_e;
itc_r = event.find(EVENT_RUNTIME_ID);
itc_s = event.find(EVENT_SEQUENCE);
Expand Down Expand Up @@ -357,7 +359,6 @@ capture_service::do_capture()
int init_cnt;
void *cap_sub_sock = NULL;
counters_t total_overflow = 0;
static bool init_done = false;

typedef enum {
/*
Expand Down Expand Up @@ -394,25 +395,6 @@ capture_service::do_capture()

m_cap_run = true;

if(!init_done) {
zmq_msg_t msg;
zmq_msg_init(&msg);
int rc = zmq_msg_recv(&msg, cap_sub_sock, 0);
RET_ON_ERR(rc == 1, "Failed to read subscription message when XSUB connects to XPUB");
/*
* When XSUB socket connects to XPUB, a subscription message is sent as a single byte 1.
* When capture service begins to read, the very first message that it will read is this
* control character.
*
* We will handle by reading this message and dropping it before we begin reading for
* cached events.
*
* This behavior will only happen once when XSUB connects to XPUB not everytime cache is started.
*
*/
init_done = true;
}

while (m_ctrl != START_CAPTURE) {
/* Wait for capture start */
this_thread::sleep_for(chrono::milliseconds(10));
Expand Down
23 changes: 6 additions & 17 deletions src/sonic-eventd/tests/eventd_ut.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -150,9 +150,8 @@ static const test_data_t ldata[] = {
},
};


void run_cap(void *zctx, bool &term, string &read_source,
int &cnt, bool &should_read_control)
int &cnt)
{
void *mock_cap = zmq_socket (zctx, ZMQ_SUB);
string source;
Expand All @@ -165,11 +164,10 @@ void run_cap(void *zctx, bool &term, string &read_source,
EXPECT_EQ(0, zmq_setsockopt(mock_cap, ZMQ_SUBSCRIBE, "", 0));
EXPECT_EQ(0, zmq_setsockopt(mock_cap, ZMQ_RCVTIMEO, &block_ms, sizeof (block_ms)));

if(should_read_control) {
zmq_msg_t msg;
zmq_msg_init(&msg);
EXPECT_NE(1, zmq_msg_recv(&msg, mock_cap, 0)); // Subscription message should be read by do_capture
}
zmq_msg_t msg;
zmq_msg_init(&msg);
int rc = zmq_msg_recv(&msg, mock_cap, 0);
EXPECT_EQ(1, rc); // read control character

while(!term) {
string source;
Expand Down Expand Up @@ -228,7 +226,6 @@ void run_pub(void *mock_pub, const string wr_source, internal_events_lst_t &lst)
TEST(eventd, proxy)
{
printf("Proxy TEST started\n");
bool should_read_control = false;
bool term_sub = false;
bool term_cap = false;
string rd_csource, rd_source, wr_source("hello");
Expand All @@ -246,7 +243,7 @@ TEST(eventd, proxy)
EXPECT_EQ(0, pxy->init());

/* capture in a thread */
thread thrc(&run_cap, zctx, ref(term_cap), ref(rd_csource), ref(rd_cevts_sz), ref(should_read_control));
thread thrc(&run_cap, zctx, ref(term_cap), ref(rd_csource), ref(rd_cevts_sz));

/* subscriber in a thread */
thread thr(&run_sub, zctx, ref(term_sub), ref(rd_source), ref(rd_evts), ref(rd_evts_sz));
Expand Down Expand Up @@ -283,17 +280,9 @@ TEST(eventd, proxy)

zmq_close(mock_pub);

/* Do control test */

should_read_control = true;

/* capture in a thread */
thread thrcc(&run_cap, zctx, ref(term_cap), ref(rd_csource), ref(rd_cevts_sz), ref(should_read_control));

delete pxy;
pxy = NULL;

thrcc.join();
zmq_ctx_term(zctx);

/* Provide time for async proxy removal to complete */
Expand Down

0 comments on commit a27674c

Please sign in to comment.