Fix flaky capture service start (#20024)

#### 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.

Author: zbud-msft

src/sonic-eventd/src/eventd.cpp

@@ -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);
@@ -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 {
         /*
@@ -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));

src/sonic-eventd/tests/eventd_ut.cpp

@@ -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;
@@ -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;
@@ -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");
@@ -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));
@@ -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 */