Rooba
/
odyssey
mirror of https://github.com/yandex/odyssey.git
1
1
Fork 0
Code Issues Packages Projects Releases Wiki Activity

odissey: rework server recv/sent stats; calculate avg

This commit is contained in:
Dmitry Simonenko 2017-08-14 17:19:54 +03:00
parent 610d5c09a6
commit 9635be2650
6 changed files with 65 additions and 31 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
sources/backend.c
View File

@ -405,7 +405,6 @@ od_backend_query(od_server_t *server, char *context, char *query, int len)
/* update server sync state and stats */
od_server_sync_request(server);
od_server_stat_request(server);
od_server_stat_sent(server, shapito_stream_used(stream));
rc = od_backend_ready_wait(server, context, UINT32_MAX);
if (rc == -1)

10
sources/console.c
View File

@ -142,13 +142,13 @@ od_console_show_stats_add(shapito_stream_t *stream,
return -1;
/* total_received */
data_len = snprintf(data, sizeof(data), "%" PRIu64, total->bytes_recv);
data_len = snprintf(data, sizeof(data), "%" PRIu64, total->recv_client);
rc = shapito_be_write_data_row_add(stream, offset, data, data_len);
if (rc == -1)
return -1;
/* total_sent */
data_len = snprintf(data, sizeof(data), "%" PRIu64, total->bytes_sent);
data_len = snprintf(data, sizeof(data), "%" PRIu64, total->recv_server);
rc = shapito_be_write_data_row_add(stream, offset, data, data_len);
if (rc == -1)
return -1;
@ -166,12 +166,14 @@ od_console_show_stats_add(shapito_stream_t *stream,
return -1;
/* avg_recv */
rc = shapito_be_write_data_row_add(stream, offset, "0", 1);
data_len = snprintf(data, sizeof(data), "%" PRIu64, avg->recv_client);
rc = shapito_be_write_data_row_add(stream, offset, data, data_len);
if (rc == -1)
return -1;
/* avg_sent */
rc = shapito_be_write_data_row_add(stream, offset, "0", 1);
data_len = snprintf(data, sizeof(data), "%" PRIu64, avg->recv_server);
rc = shapito_be_write_data_row_add(stream, offset, data, data_len);
if (rc == -1)
return -1;

13
sources/frontend.c
View File

@ -266,7 +266,7 @@ od_frontend_copy_in(od_client_t *client)
if (rc == -1)
return OD_RS_ECLIENT_READ;
od_server_stat_recv(server, shapito_stream_used(stream));
od_server_stat_recv_client(server, shapito_stream_used(stream));
type = *stream->start;
od_debug_client(&instance->logger, &client->id, "copy",
@ -275,8 +275,6 @@ od_frontend_copy_in(od_client_t *client)
if (rc == -1)
return OD_RS_ESERVER_WRITE;
od_server_stat_sent(server, shapito_stream_used(stream));
/* copy complete or fail */
if (type == 'c' || type == 'f')
break;
@ -356,7 +354,7 @@ od_frontend_remote(od_client_t *client)
/* update request and recv stat */
od_server_stat_request(server);
od_server_stat_recv(server, shapito_stream_used(stream));
od_server_stat_recv_client(server, shapito_stream_used(stream));
rc = od_write(server->io, stream);
if (rc == -1)
@ -389,6 +387,8 @@ od_frontend_remote(od_client_t *client)
od_debug_server(&instance->logger, &server->id, NULL,
"%c", type);
/* update server recv stats */
od_server_stat_recv_server(server, shapito_stream_used(stream));
/* ErrorResponse */
if (type == 'E') {
@ -431,9 +431,6 @@ od_frontend_remote(od_client_t *client)
if (rc == -1)
return OD_RS_ECLIENT_WRITE;
/* update stats */
od_server_stat_sent(server, shapito_stream_used(stream));
/* switch to CopyIn mode */
rc = od_frontend_copy_in(client);
if (rc != OD_RS_OK)
@ -456,8 +453,6 @@ od_frontend_remote(od_client_t *client)
rc = od_write(client->io, stream);
if (rc == -1)
return OD_RS_ECLIENT_WRITE;
/* update stats */
od_server_stat_sent(server, shapito_stream_used(stream));
shapito_stream_reset(stream);
}
}

43
sources/periodic.c
View File

@ -53,6 +53,8 @@ od_periodic_stats_server(od_server_t *server, void *arg)
od_serverstat_t *stats = arg;
stats->query_time += od_atomic_u64_of(&server->stats.query_time);
stats->count_request += od_atomic_u64_of(&server->stats.count_request);
stats->recv_client += od_atomic_u64_of(&server->stats.recv_client);
stats->recv_server += od_atomic_u64_of(&server->stats.recv_server);
return 0;
}
@ -75,6 +77,7 @@ od_periodic_stats(od_router_t *router)
/* gather statistics per route server pool */
od_serverstat_t stats;
memset(&stats, 0, sizeof(stats));
od_serverpool_foreach(&route->server_pool, OD_SACTIVE,
od_periodic_stats_server,
&stats);
@ -87,10 +90,13 @@ od_periodic_stats(od_router_t *router)
int64_t reqs_diff;
reqs_diff = stats.count_request - route->periodic_stats.count_request;
uint64_t recv_client = 0;
uint64_t recv_server = 0;
uint64_t reqs = 0;
uint64_t query_time = 0;
if (reqs_diff >= 0)
{
/* request count */
uint64_t reqs_prev = 0;
reqs_prev = route->periodic_stats.count_request /
instance->scheme.stats_interval;
@ -102,6 +108,31 @@ od_periodic_stats(od_router_t *router)
reqs = (reqs_current - reqs_prev) /
instance->scheme.stats_interval;
/* recv client */
uint64_t recv_client_prev = 0;
recv_client_prev = route->periodic_stats.recv_client /
instance->scheme.stats_interval;
uint64_t recv_client_current = 0;
recv_client_current = stats.recv_client /
instance->scheme.stats_interval;
recv_client = (recv_client_current - recv_client_prev) /
instance->scheme.stats_interval;
/* recv server */
uint64_t recv_server_prev = 0;
recv_server_prev = route->periodic_stats.recv_server /
instance->scheme.stats_interval;
uint64_t recv_server_current = 0;
recv_server_current = stats.recv_server /
instance->scheme.stats_interval;
recv_server = (recv_server_current - recv_server_prev) /
instance->scheme.stats_interval;
/* query time */
if (reqs_diff > 0)
query_time = (stats.query_time - route->periodic_stats.query_time) /
(reqs_current - reqs_prev);
@ -111,7 +142,9 @@ od_periodic_stats(od_router_t *router)
route->periodic_stats = stats;
route->periodic_stats_avg.count_request = reqs;
route->periodic_stats_avg.query_time = query_time;
route->periodic_stats_avg.recv_client = recv_client;
route->periodic_stats_avg.recv_server = recv_server;
route->periodic_stats_avg.query_time = query_time;
if (instance->scheme.log_stats) {
od_log(&instance->logger,
@ -119,7 +152,9 @@ od_periodic_stats(od_router_t *router)
"pool_active %d, "
"pool_idle %d "
"rps %" PRIu64 " "
"query_time_us %" PRIu64,
"query_time_us %" PRIu64 " "
"recv_client_bytes %" PRIu64 " "
"recv_server_bytes %" PRIu64,
route->id.database_len,
route->id.database,
route->id.user_len,
@ -130,7 +165,9 @@ od_periodic_stats(od_router_t *router)
route->server_pool.count_active,
route->server_pool.count_idle,
reqs,
query_time);
query_time,
recv_client,
recv_server);
}
}
}

12
sources/route_pool.c
View File

@ -171,10 +171,14 @@ od_routepool_stats_mark(od_routepool_t *pool,
continue;
total->count_request += route->periodic_stats.count_request;
total->query_time += route->periodic_stats.query_time;
total->query_time += route->periodic_stats.query_time;
total->recv_client += route->periodic_stats.recv_client;
total->recv_server += route->periodic_stats.recv_server;
avg->count_request += route->periodic_stats_avg.count_request;
avg->query_time += route->periodic_stats_avg.query_time;
avg->count_request += route->periodic_stats_avg.count_request;
avg->query_time += route->periodic_stats_avg.query_time;
avg->recv_client += route->periodic_stats_avg.recv_client;
avg->recv_server += route->periodic_stats_avg.recv_server;
route->stats_mark++;
match++;
@ -216,6 +220,8 @@ od_routepool_stats(od_routepool_t *pool,
assert(match > 0);
avg.count_request /= match;
avg.query_time /= match;
avg.recv_client /= match;
avg.recv_server /= match;
int rc;
rc = callback(route->id.database, route->id.database_len, &total, &avg, arg);
if (rc == -1) {

17
sources/server.h
View File

@ -20,15 +20,10 @@ typedef enum
struct od_serverstat
{
/* total request count */
od_atomic_u64_t count_request;
/* bytes sent to server */
od_atomic_u64_t bytes_sent;
/* bytes received from clients */
od_atomic_u64_t bytes_recv;
/* total query time */
od_atomic_u64_t recv_server;
od_atomic_u64_t recv_client;
od_atomic_u64_t query_time;
/* start time of last request */
uint64_t query_time_start;
};
@ -130,15 +125,15 @@ od_server_stat_reply(od_server_t *server)
}
static inline void
od_server_stat_sent(od_server_t *server, uint64_t bytes)
od_server_stat_recv_server(od_server_t *server, uint64_t bytes)
{
od_atomic_u64_add(&server->stats.bytes_sent, bytes);
od_atomic_u64_add(&server->stats.recv_server, bytes);
}
static inline void
od_server_stat_recv(od_server_t *server, uint64_t bytes)
od_server_stat_recv_client(od_server_t *server, uint64_t bytes)
{
od_atomic_u64_add(&server->stats.bytes_recv, bytes);
od_atomic_u64_add(&server->stats.recv_client, bytes);
}
#endif /* OD_SERVER_H */