7350837242

Committed 28 Dec 2023 07:59PM UTC coverage: 98.458% (+0.09%) from 98.364%

Build # 7350837242

Build Type

Pull #683

github

Committed by

gjasny

Commit Message

feat: don't create temporary objects for serialization

Fix: #646

Pull Request Pull Request #683: Avoid temporary objects

Run Details

116 of 118 new or added lines in 6 files covered. (98.31%)

1 existing line in 1 file now uncovered.

894 of 908 relevant lines covered (98.46%)

96521.87 hits per line

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

94.64

/pull/src/handler.cc

#include "handler.h"

#include <algorithm>
#include <cassert>
#include <chrono>
#include <cstring>
#include <iterator>
#include <numeric>
#include <string>

#ifdef HAVE_ZLIB
#include <zconf.h>
#include <zlib.h>
#endif

#include "civetweb.h"
#include "metrics_collector.h"
#include "prometheus/counter.h"
#include "prometheus/iovector.h"
#include "prometheus/metric_family.h"
#include "prometheus/summary.h"
#include "prometheus/text_serializer.h"

#if CIVETWEB_VERSION_MAJOR < 1 || \
    (CIVETWEB_VERSION_MAJOR == 1 && CIVETWEB_VERSION_MINOR < 14)
// https://github.com/civetweb/civetweb/issues/954
#error "Civetweb version 1.14 or higher required"
#endif

namespace prometheus {
namespace detail {

MetricsHandler::MetricsHandler(Registry& registry)
    : bytes_transferred_family_(
          BuildCounter()
              .Name("exposer_transferred_bytes_total")
              .Help("Transferred bytes to metrics services")
              .Register(registry)),
      bytes_transferred_(bytes_transferred_family_.Add({})),
      num_scrapes_family_(BuildCounter()
                              .Name("exposer_scrapes_total")
                              .Help("Number of times metrics were scraped")
                              .Register(registry)),
      num_scrapes_(num_scrapes_family_.Add({})),
      request_latencies_family_(
          BuildSummary()
              .Name("exposer_request_latencies")
              .Help("Latencies of serving scrape requests, in microseconds")
              .Register(registry)),
      request_latencies_(request_latencies_family_.Add(
          {}, Summary::Quantiles{{0.5, 0.05}, {0.9, 0.01}, {0.99, 0.001}})) {}

#ifdef HAVE_ZLIB
static bool IsEncodingAccepted(struct mg_connection* conn,
                               const char* encoding) {
  auto accept_encoding = mg_get_header(conn, "Accept-Encoding");
  if (!accept_encoding) {
    return false;
  }
  return std::strstr(accept_encoding, encoding) != nullptr;
}

static IOVector GZipCompress(const IOVector& input) {
  auto zs = z_stream{};
  auto windowSize = 16 + MAX_WBITS;
  auto memoryLevel = 9;

  if (deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, windowSize,
                   memoryLevel, Z_DEFAULT_STRATEGY) != Z_OK) {
    return {};
  }

  auto s = input.size();

  int ret;

  IOVector output;

  for (std::size_t i = 0; i < input.data.size(); ++i) {
    bool last = i == input.data.size() - 1U;
    auto chunk = input.data[i];

    zs.next_in =
        const_cast<Bytef*>(reinterpret_cast<const Bytef*>(chunk.data()));
    zs.avail_in = chunk.size();

    do {
      static constexpr std::size_t maximumChunkSize = 1 * 1024 * 1024;
      if (output.data.empty() ||
          output.data.back().size() >= maximumChunkSize) {
        output.data.emplace_back();
        output.data.back().reserve(maximumChunkSize);
      }

      auto&& chunk = output.data.back();

      const auto previouslyUsed = chunk.size();
      const auto remainingChunkSize = maximumChunkSize - previouslyUsed;

      zs.avail_out = remainingChunkSize;
      chunk.resize(chunk.size() + remainingChunkSize);
      zs.next_out = reinterpret_cast<Bytef*>(chunk.data() + previouslyUsed);

      ret = deflate(&zs, last ? Z_FINISH : Z_NO_FLUSH);
      assert(ret != Z_STREAM_ERROR);

      chunk.resize(maximumChunkSize - zs.avail_out);
    } while (zs.avail_out == 0U);
    assert(zs.avail_in == 0);
  }
  assert(ret == Z_STREAM_END);
  assert(zs.total_out == output.size());

  deflateEnd(&zs);

  if (ret != Z_STREAM_END) {
    return {};
  }

  return output;
}
#endif

static std::size_t WriteResponse(struct mg_connection* conn,
                                 const IOVector& body) {
  mg_printf(conn,
            "HTTP/1.1 200 OK\r\n"
            "Content-Type: text/plain; charset=utf-8\r\n");

#ifdef HAVE_ZLIB
  auto acceptsGzip = IsEncodingAccepted(conn, "gzip");

  if (acceptsGzip) {
    auto compressed = GZipCompress(body);
    if (!compressed.empty()) {
      const std::size_t contentSize = compressed.size();
      mg_printf(conn,
                "Content-Encoding: gzip\r\n"
                "Content-Length: %s\r\n\r\n",
                std::to_string(contentSize).c_str());
      for (auto&& chunk : compressed.data) {
        mg_write(conn, chunk.data(), chunk.size());
      }
      return contentSize;
    }
  }
#endif

  std::size_t contentSize = body.size();

  mg_printf(conn, "Content-Length: %s\r\n\r\n",
            std::to_string(contentSize).c_str());
  for (auto&& chunk : body.data) {
    mg_write(conn, chunk.data(), chunk.size());
  }
  return contentSize;
}

void MetricsHandler::RegisterCollectable(
    const std::weak_ptr<Collectable>& collectable) {
  std::lock_guard<std::mutex> lock{collectables_mutex_};
  CleanupStalePointers(collectables_);
  collectables_.push_back(collectable);
}

void MetricsHandler::RemoveCollectable(
    const std::weak_ptr<Collectable>& collectable) {
  std::lock_guard<std::mutex> lock{collectables_mutex_};

  auto locked = collectable.lock();
  auto same_pointer = [&locked](const std::weak_ptr<Collectable>& candidate) {
    return locked == candidate.lock();
  };

  collectables_.erase(std::remove_if(std::begin(collectables_),
                                     std::end(collectables_), same_pointer),
                      std::end(collectables_));
}

bool MetricsHandler::handleGet(CivetServer*, struct mg_connection* conn) {
  auto start_time_of_request = std::chrono::steady_clock::now();

  IOVector ioVector;
  const auto serializer = TextSerializer{ioVector};

  {
    std::lock_guard<std::mutex> lock{collectables_mutex_};
    CollectMetrics(serializer, collectables_);
  }

  auto bodySize = WriteResponse(conn, ioVector);

  auto stop_time_of_request = std::chrono::steady_clock::now();
  auto duration = std::chrono::duration_cast<std::chrono::microseconds>(
      stop_time_of_request - start_time_of_request);
  request_latencies_.Observe(duration.count());

  bytes_transferred_.Increment(bodySize);
  num_scrapes_.Increment();
  return true;
}

void MetricsHandler::CleanupStalePointers(
    std::vector<std::weak_ptr<Collectable>>& collectables) {
  collectables.erase(
      std::remove_if(std::begin(collectables), std::end(collectables),
                     [](const std::weak_ptr<Collectable>& candidate) {
                       return candidate.expired();
                     }),
      std::end(collectables));
}
}  // namespace detail
}  // namespace prometheus

1	#include "handler.h"
2
3	#include <algorithm>
4	#include <cassert>
5	#include <chrono>
6	#include <cstring>
7	#include <iterator>
8	#include <numeric>
9	#include <string>
10
11	#ifdef HAVE_ZLIB
12	#include <zconf.h>
13	#include <zlib.h>
14	#endif
15
16	#include "civetweb.h"
17	#include "metrics_collector.h"
18	#include "prometheus/counter.h"
19	#include "prometheus/iovector.h"
20	#include "prometheus/metric_family.h"
21	#include "prometheus/summary.h"
22	#include "prometheus/text_serializer.h"
23
24	#if CIVETWEB_VERSION_MAJOR < 1 \|\| \
25	(CIVETWEB_VERSION_MAJOR == 1 && CIVETWEB_VERSION_MINOR < 14)
26	// https://github.com/civetweb/civetweb/issues/954
27	#error "Civetweb version 1.14 or higher required"
28	#endif
29
30	namespace prometheus {
31	namespace detail {
32
33	MetricsHandler::MetricsHandler(Registry& registry)	28✔
34	: bytes_transferred_family_(
35	BuildCounter()	28✔
36	.Name("exposer_transferred_bytes_total")	56✔
37	.Help("Transferred bytes to metrics services")	56✔
38	.Register(registry)),	28✔
39	bytes_transferred_(bytes_transferred_family_.Add({})),	28✔
40	num_scrapes_family_(BuildCounter()	28✔
41	.Name("exposer_scrapes_total")	56✔
42	.Help("Number of times metrics were scraped")	56✔
43	.Register(registry)),	28✔
44	num_scrapes_(num_scrapes_family_.Add({})),	28✔
45	request_latencies_family_(
46	BuildSummary()	28✔
47	.Name("exposer_request_latencies")	56✔
48	.Help("Latencies of serving scrape requests, in microseconds")	56✔
49	.Register(registry)),	28✔
50	request_latencies_(request_latencies_family_.Add(	28✔
51	{}, Summary::Quantiles{{0.5, 0.05}, {0.9, 0.01}, {0.99, 0.001}})) {}	196✔
52
53	#ifdef HAVE_ZLIB
54	static bool IsEncodingAccepted(struct mg_connection* conn,	16✔
55	const char* encoding) {
56	auto accept_encoding = mg_get_header(conn, "Accept-Encoding");	16✔
57	if (!accept_encoding) {	16✔
58	return false;	14✔
59	}
60	return std::strstr(accept_encoding, encoding) != nullptr;	2✔
61	}
62
63	static IOVector GZipCompress(const IOVector& input) {	2✔
64	auto zs = z_stream{};	2✔
65	auto windowSize = 16 + MAX_WBITS;	2✔
66	auto memoryLevel = 9;	2✔
67
68	if (deflateInit2(&zs, Z_DEFAULT_COMPRESSION, Z_DEFLATED, windowSize,	2✔
69	memoryLevel, Z_DEFAULT_STRATEGY) != Z_OK) {	2✔
70	return {};	×
71	}
72
73	auto s = input.size();	2✔
74
75	int ret;
76
77	IOVector output;	4✔
78
79	for (std::size_t i = 0; i < input.data.size(); ++i) {	4✔
80	bool last = i == input.data.size() - 1U;	2✔
81	auto chunk = input.data[i];	4✔
82
83	zs.next_in =	2✔
84	const_cast<Bytef>(reinterpret_cast<const Bytef>(chunk.data()));	2✔
85	zs.avail_in = chunk.size();	2✔
86
NEW 87	do {	×
88	static constexpr std::size_t maximumChunkSize = 1 * 1024 * 1024;
89	if (output.data.empty() \|\|	2✔
NEW 90	output.data.back().size() >= maximumChunkSize) {	×
91	output.data.emplace_back();	2✔
92	output.data.back().reserve(maximumChunkSize);	2✔
93	}
94
95	auto&& chunk = output.data.back();	2✔
96
97	const auto previouslyUsed = chunk.size();	2✔
98	const auto remainingChunkSize = maximumChunkSize - previouslyUsed;	2✔
99
100	zs.avail_out = remainingChunkSize;	2✔
101	chunk.resize(chunk.size() + remainingChunkSize);	2✔
102	zs.next_out = reinterpret_cast<Bytef*>(chunk.data() + previouslyUsed);	2✔
103
104	ret = deflate(&zs, last ? Z_FINISH : Z_NO_FLUSH);	2✔
105	assert(ret != Z_STREAM_ERROR);	2✔
106
107	chunk.resize(maximumChunkSize - zs.avail_out);	2✔
108	} while (zs.avail_out == 0U);	2✔
109	assert(zs.avail_in == 0);	2✔
110	}
111	assert(ret == Z_STREAM_END);	2✔
112	assert(zs.total_out == output.size());	2✔
113
114	deflateEnd(&zs);	2✔
115
116	if (ret != Z_STREAM_END) {	2✔
117	return {};	×
118	}
119
120	return output;	2✔
121	}
122	#endif
123
124	static std::size_t WriteResponse(struct mg_connection* conn,	16✔
125	const IOVector& body) {
126	mg_printf(conn,	16✔
127	"HTTP/1.1 200 OK\r\n"
128	"Content-Type: text/plain; charset=utf-8\r\n");
129
130	#ifdef HAVE_ZLIB
131	auto acceptsGzip = IsEncodingAccepted(conn, "gzip");	16✔
132
133	if (acceptsGzip) {	16✔
134	auto compressed = GZipCompress(body);	2✔
135	if (!compressed.empty()) {	2✔
136	const std::size_t contentSize = compressed.size();	2✔
137	mg_printf(conn,	2✔
138	"Content-Encoding: gzip\r\n"
139	"Content-Length: %s\r\n\r\n",
140	std::to_string(contentSize).c_str());	4✔
141	for (auto&& chunk : compressed.data) {	4✔
142	mg_write(conn, chunk.data(), chunk.size());	2✔
143	}
144	return contentSize;	2✔
145	}
146	}
147	#endif
148
149	std::size_t contentSize = body.size();	14✔
150
151	mg_printf(conn, "Content-Length: %s\r\n\r\n",	14✔
152	std::to_string(contentSize).c_str());	28✔
153	for (auto&& chunk : body.data) {	28✔
154	mg_write(conn, chunk.data(), chunk.size());	14✔
155	}
156	return contentSize;	14✔
157	}
158
159	void MetricsHandler::RegisterCollectable(	58✔
160	const std::weak_ptr<Collectable>& collectable) {
161	std::lock_guard<std::mutex> lock{collectables_mutex_};	116✔
162	CleanupStalePointers(collectables_);	58✔
163	collectables_.push_back(collectable);	58✔
164	}	58✔
165
166	void MetricsHandler::RemoveCollectable(	2✔
167	const std::weak_ptr<Collectable>& collectable) {
168	std::lock_guard<std::mutex> lock{collectables_mutex_};	4✔
169
170	auto locked = collectable.lock();	2✔
171	auto same_pointer = [&locked](const std::weak_ptr<Collectable>& candidate) {	8✔
172	return locked == candidate.lock();	4✔
173	};	2✔
174
175	collectables_.erase(std::remove_if(std::begin(collectables_),	×
176	std::end(collectables_), same_pointer),	2✔
177	std::end(collectables_));	4✔
178	}	2✔
179
180	bool MetricsHandler::handleGet(CivetServer, struct mg_connection conn) {	16✔
181	auto start_time_of_request = std::chrono::steady_clock::now();	16✔
182
183	IOVector ioVector;	32✔
184	const auto serializer = TextSerializer{ioVector};	16✔
185
186	{
187	std::lock_guard<std::mutex> lock{collectables_mutex_};	32✔
188	CollectMetrics(serializer, collectables_);	16✔
189	}
190
191	auto bodySize = WriteResponse(conn, ioVector);	16✔
192
193	auto stop_time_of_request = std::chrono::steady_clock::now();	16✔
194	auto duration = std::chrono::duration_cast<std::chrono::microseconds>(
195	stop_time_of_request - start_time_of_request);	16✔
196	request_latencies_.Observe(duration.count());	16✔
197
198	bytes_transferred_.Increment(bodySize);	16✔
199	num_scrapes_.Increment();	16✔
200	return true;	32✔
201	}
202
203	void MetricsHandler::CleanupStalePointers(	58✔
204	std::vector<std::weak_ptr<Collectable>>& collectables) {
205	collectables.erase(
206	std::remove_if(std::begin(collectables), std::end(collectables),	×
207	[](const std::weak_ptr<Collectable>& candidate) {	32✔
208	return candidate.expired();	32✔
209	}),	58✔
210	std::end(collectables));	116✔
211	}	58✔
212	} // namespace detail
213	} // namespace prometheus

jupp0r / prometheus-cpp / 7350837242

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