18620024447

Committed 18 Oct 2025 07:29PM UTC coverage: 96.907%. Remained the same

Build # 18620024447

Build Type

Pull #80

github

Committed by

web-flow

Commit Message

Merge de0f5583e into 7661f1305

Pull Request Pull Request #80: bump up OTel version

Run Details

19 of 19 new or added lines in 6 files covered. (100.0%)

9 existing lines in 2 files now uncovered.

940 of 970 relevant lines covered (96.91%)

167.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

86.67

/src/http2_client/client_impl.cpp

#include "client_impl.hpp"

#include <nghttp2/asio_http2_client.h>
#include <syslog.h>

#include <atomic>
#include <boost/asio.hpp>
#include <boost/bind/bind.hpp>
#include <boost/system/error_code.hpp>
#include <chrono>
#include <iostream>
#include <map>
#include <mutex>
#include <optional>
#include <shared_mutex>
#include <utility>

#include "connection.hpp"
#include "opentelemetry/semconv/url_attributes.h"
#include "opentelemetry/semconv/incubating/http_attributes.h"
#include "script.hpp"
#include "script_queue.hpp"
#include "stats.hpp"
#include "tracer.hpp"

namespace ng = nghttp2::asio_http2;
using namespace std::chrono;
namespace ot_trace = opentelemetry::trace;
namespace ot_conv = opentelemetry::semconv;


namespace http2_client
{
client_impl::client_impl(std::shared_ptr<stats::stats_if> st, boost::asio::io_context& io_ctx,
                         std::unique_ptr<traffic::script_queue_if> q, const std::string& h,
                         const std::string& p, const bool secure_session)
    : stats(std::move(st)),
      io_ctx(io_ctx),
      queue(std::move(q)),
      host(h),
      port(p),
      secure_session(secure_session),
      conn(std::make_unique<connection>(h, p, secure_session))
{
    if (!conn->wait_to_be_connected())
    {
        std::cerr << "Fatal error. Could not connect to: " << host << ":" << port << std::endl;
    }
}

void client_impl::handle_timeout(const std::shared_ptr<race_control>& control,
                                 const std::string& msg_name) const
{
    std::scoped_lock guard(control->mtx);
    if (control->answered)
    {
        return;
    }
    control->timed_out = true;
    stats->add_timeout(msg_name);
    queue->cancel_script();
}
// TODO: Add timeout handling in spans
void client_impl::handle_timeout_cancelled(const std::shared_ptr<race_control>& control,
                                           const std::string& msg_name) const
{
    if (control->mtx.try_lock())
    {
        if (!control->answered)
        {
            control->timed_out = true;
            stats->add_error(msg_name, 469);
            queue->cancel_script();
        }
        control->mtx.unlock();
    }
}

void client_impl::on_timeout(const boost::system::error_code& e,
                             std::shared_ptr<race_control> control,
                             const std::string& msg_name) const
{
    if (e.value() == 0)
    {
        handle_timeout(control, msg_name);
    }
    else
    {
        handle_timeout_cancelled(control, msg_name);
    }
}

void client_impl::open_new_connection()
{
    if (!mtx.try_lock())
    {
        return;
    }
    conn.reset();

    if (auto new_conn = std::make_unique<connection>(host, port, secure_session);
        new_conn->wait_to_be_connected())
    {
        conn = std::move(new_conn);
    }
    else
    {
        new_conn.reset();
    }
    mtx.unlock();
}

void client_impl::send()
{
    auto script_opt = queue->get_next_script();
    if (!script_opt.has_value())
    {
        return;
    }
    const auto& script = *script_opt;
    request req = get_next_request(host, port, script);

    if (!is_connected())
    {
        stats->add_client_error(req.name, 466);
        queue->cancel_script();
        open_new_connection();
        return;
    }

    if (!mtx.try_lock_shared())
    {
        stats->add_client_error(req.name, 467);
        queue->cancel_script();
        return;
    }

    const auto& session = conn->get_session();
    session.io_service().post(
        [this, script, &session, req]
        {
            boost::system::error_code ec;
            auto init_time = std::make_shared<time_point<steady_clock>>(steady_clock::now());

            auto span = o11y::create_child_span(req.name, script.get_span());
            span->SetAttribute(ot_conv::url::kUrlFull, req.url);
            span->SetAttribute(ot_conv::http::kHttpRequestMethod, req.method);

            auto nghttp_req = session.submit(ec, req.method, req.url, req.body, req.headers);
            if (!nghttp_req)
            {
                std::cerr << "Error submitting. Closing connection:" << ec.message() << std::endl;
                conn->close();
                stats->add_client_error(req.name, 468);
                queue->cancel_script();
                return;
            }

            stats->increase_sent(req.name);
            span->AddEvent("Request sent");

            auto ctrl = std::make_shared<race_control>();
            auto timer = std::make_shared<boost::asio::steady_timer>(io_ctx);
            timer->expires_after(milliseconds(script.get_timeout_ms()));
            timer->async_wait(boost::bind(&client_impl::on_timeout, this,
                                          boost::asio::placeholders::error, ctrl, req.name));

            nghttp_req->on_response(
                [this, timer, init_time, script, ctrl, req, span](const ng::client::response& res)
                {
                    auto elapsed_time =
                        duration_cast<microseconds>(steady_clock::now() - (*init_time)).count();

                    std::lock_guard guard(ctrl->mtx);
                    if (ctrl->timed_out)
                    {
                        return;
                    }
                    ctrl->answered = true;
                    timer->cancel();

                    span->AddEvent("Response received");
                    auto answer = std::make_shared<std::string>();
                    res.on_data(
                        [this, &res, script, answer, elapsed_time, req, span](const uint8_t* data,
                                                                              std::size_t len)
                        {
                            if (len > 0)
                            {
                                std::string json(reinterpret_cast<const char*>(data), len);
                                *answer += json;
                            }
                            else
                            {
                                span->AddEvent("Body received");
                                traffic::answer_type ans = {res.status_code(), *answer,
                                                            res.header()};
                                span->SetAttribute(
                                    ot_conv::http::kHttpResponseStatusCode,
                                    res.status_code());

                                bool valid_answer = script.validate_answer(ans);
                                if (valid_answer)
                                {
                                    stats->add_measurement(req.name, elapsed_time,
                                                           res.status_code());
                                    span->SetStatus(ot_trace::StatusCode::kOk);
                                    span->End();
                                    queue->enqueue_script(script, ans);
                                }
                                else
                                {
                                    stats->add_error(req.name, res.status_code());
                                    span->SetStatus(opentelemetry::trace::StatusCode::kError);
                                    span->End();
                                    queue->cancel_script();
                                }
                            }
                        });
                });

            nghttp_req->on_close(
                []([[maybe_unused]] uint32_t error_code)
                {
                    // on_close is registered here for the sake of completion and
                    // because it helps debugging sometimes, but no implementation needed.
                });
        });
    mtx.unlock_shared();
}

}  // namespace http2_client

1	#include "client_impl.hpp"
2
3	#include <nghttp2/asio_http2_client.h>
4	#include <syslog.h>
5
6	#include <atomic>
7	#include <boost/asio.hpp>
8	#include <boost/bind/bind.hpp>
9	#include <boost/system/error_code.hpp>
10	#include <chrono>
11	#include <iostream>
12	#include <map>
13	#include <mutex>
14	#include <optional>
15	#include <shared_mutex>
16	#include <utility>
17
18	#include "connection.hpp"
19	#include "opentelemetry/semconv/url_attributes.h"
20	#include "opentelemetry/semconv/incubating/http_attributes.h"
21	#include "script.hpp"
22	#include "script_queue.hpp"
23	#include "stats.hpp"
24	#include "tracer.hpp"
25
26	namespace ng = nghttp2::asio_http2;
27	using namespace std::chrono;
28	namespace ot_trace = opentelemetry::trace;
29	namespace ot_conv = opentelemetry::semconv;
30
31
32	namespace http2_client
33	{
34	client_impl::client_impl(std::shared_ptr<stats::stats_if> st, boost::asio::io_context& io_ctx,	36✔
35	std::unique_ptr<traffic::script_queue_if> q, const std::string& h,
36	const std::string& p, const bool secure_session)	36✔
37	: stats(std::move(st)),	36✔
38	io_ctx(io_ctx),	36✔
39	queue(std::move(q)),	36✔
40	host(h),	36✔
41	port(p),	36✔
42	secure_session(secure_session),	36✔
43	conn(std::make_unique<connection>(h, p, secure_session))	72✔
44	{
45	if (!conn->wait_to_be_connected())	36✔
46	{
47	std::cerr << "Fatal error. Could not connect to: " << host << ":" << port << std::endl;	4✔
48	}
49	}	36✔
50
51	void client_impl::handle_timeout(const std::shared_ptr<race_control>& control,	4✔
52	const std::string& msg_name) const
53	{
54	std::scoped_lock guard(control->mtx);	4✔
55	if (control->answered)	4✔
56	{
UNCOV 57	return;	×
58	}
59	control->timed_out = true;	4✔
60	stats->add_timeout(msg_name);	4✔
61	queue->cancel_script();	4✔
62	}	4✔
63	// TODO: Add timeout handling in spans
64	void client_impl::handle_timeout_cancelled(const std::shared_ptr<race_control>& control,	12✔
65	const std::string& msg_name) const
66	{
67	if (control->mtx.try_lock())	12✔
68	{
69	if (!control->answered)	12✔
70	{
71	control->timed_out = true;	×
72	stats->add_error(msg_name, 469);	×
UNCOV 73	queue->cancel_script();	×
74	}
75	control->mtx.unlock();	12✔
76	}
77	}	12✔
78
79	void client_impl::on_timeout(const boost::system::error_code& e,	16✔
80	std::shared_ptr<race_control> control,
81	const std::string& msg_name) const
82	{
83	if (e.value() == 0)	16✔
84	{
85	handle_timeout(control, msg_name);	4✔
86	}
87	else
88	{
89	handle_timeout_cancelled(control, msg_name);	12✔
90	}
91	}	16✔
92
93	void client_impl::open_new_connection()	8✔
94	{
95	if (!mtx.try_lock())	8✔
96	{
UNCOV 97	return;	×
98	}
99	conn.reset();	8✔
100
101	if (auto new_conn = std::make_unique<connection>(host, port, secure_session);	8✔
102	new_conn->wait_to_be_connected())	8✔
103	{
104	conn = std::move(new_conn);	4✔
105	}
106	else
107	{
108	new_conn.reset();	4✔
109	}	8✔
110	mtx.unlock();	8✔
111	}
112
113	void client_impl::send()	24✔
114	{
115	auto script_opt = queue->get_next_script();	24✔
116	if (!script_opt.has_value())	24✔
117	{
UNCOV 118	return;	×
119	}
120	const auto& script = *script_opt;	24✔
121	request req = get_next_request(host, port, script);	24✔
122
123	if (!is_connected())	24✔
124	{
125	stats->add_client_error(req.name, 466);	8✔
126	queue->cancel_script();	8✔
127	open_new_connection();	8✔
128	return;	8✔
129	}
130
131	if (!mtx.try_lock_shared())	16✔
132	{
133	stats->add_client_error(req.name, 467);	×
134	queue->cancel_script();	×
UNCOV 135	return;	×
136	}
137
138	const auto& session = conn->get_session();	16✔
139	session.io_service().post(	16✔
140	[this, script, &session, req]	16✔
141	{
142	boost::system::error_code ec;	16✔
143	auto init_time = std::make_shared<time_point<steady_clock>>(steady_clock::now());	16✔
144
145	auto span = o11y::create_child_span(req.name, script.get_span());	16✔
146	span->SetAttribute(ot_conv::url::kUrlFull, req.url);	16✔
147	span->SetAttribute(ot_conv::http::kHttpRequestMethod, req.method);	16✔
148
149	auto nghttp_req = session.submit(ec, req.method, req.url, req.body, req.headers);	16✔
150	if (!nghttp_req)	16✔
151	{
152	std::cerr << "Error submitting. Closing connection:" << ec.message() << std::endl;	×
153	conn->close();	×
154	stats->add_client_error(req.name, 468);	×
155	queue->cancel_script();	×
UNCOV 156	return;	×
157	}
158
159	stats->increase_sent(req.name);	16✔
160	span->AddEvent("Request sent");	16✔
161
162	auto ctrl = std::make_shared<race_control>();	16✔
163	auto timer = std::make_shared<boost::asio::steady_timer>(io_ctx);	16✔
164	timer->expires_after(milliseconds(script.get_timeout_ms()));	16✔
165	timer->async_wait(boost::bind(&client_impl::on_timeout, this,	32✔
166	boost::asio::placeholders::error, ctrl, req.name));	16✔
167
168	nghttp_req->on_response(	16✔
169	[this, timer, init_time, script, ctrl, req, span](const ng::client::response& res)	32✔
170	{
171	auto elapsed_time =
172	duration_cast<microseconds>(steady_clock::now() - (*init_time)).count();	12✔
173
174	std::lock_guard guard(ctrl->mtx);	12✔
175	if (ctrl->timed_out)	12✔
176	{
UNCOV 177	return;	×
178	}
179	ctrl->answered = true;	12✔
180	timer->cancel();	12✔
181
182	span->AddEvent("Response received");	12✔
183	auto answer = std::make_shared<std::string>();	12✔
184	res.on_data(	12✔
185	[this, &res, script, answer, elapsed_time, req, span](const uint8_t* data,	24✔
186	std::size_t len)
187	{
188	if (len > 0)	24✔
189	{
190	std::string json(reinterpret_cast<const char*>(data), len);	12✔
191	*answer += json;	12✔
192	}	12✔
193	else
194	{
195	span->AddEvent("Body received");	12✔
196	traffic::answer_type ans = {res.status_code(), *answer,	12✔
197	res.header()};	12✔
198	span->SetAttribute(	24✔
199	ot_conv::http::kHttpResponseStatusCode,
200	res.status_code());	12✔
201
202	bool valid_answer = script.validate_answer(ans);	12✔
203	if (valid_answer)	12✔
204	{
205	stats->add_measurement(req.name, elapsed_time,	8✔
206	res.status_code());
207	span->SetStatus(ot_trace::StatusCode::kOk);	8✔
208	span->End();	8✔
209	queue->enqueue_script(script, ans);	8✔
210	}
211	else
212	{
213	stats->add_error(req.name, res.status_code());	4✔
214	span->SetStatus(opentelemetry::trace::StatusCode::kError);	4✔
215	span->End();	4✔
216	queue->cancel_script();	4✔
217	}
218	}	12✔
219	});	24✔
220	});	12✔
221
222	nghttp_req->on_close(	16✔
UNCOV 223	[]([[maybe_unused]] uint32_t error_code)	×
224	{
225	// on_close is registered here for the sake of completion and
226	// because it helps debugging sometimes, but no implementation needed.
227	});	16✔
228	});	16✔
229	mtx.unlock_shared();	16✔
230	}	32✔
231
232	} // namespace http2_client

jgomezselles / hermes / 18620024447

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous