1 /*-
2 * collectd - src/dpdkstat.c
3 * MIT License
4 *
5 * Copyright(c) 2016 Intel Corporation. All rights reserved.
6 *
7 * Permission is hereby granted, free of charge, to any person obtaining a copy
8 * of this software and associated documentation files (the "Software"), to deal
9 * in the Software without restriction, including without limitation the rights
10 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
11 * copies of the Software, and to permit persons to whom the Software is
12 * furnished to do so, subject to the following conditions:
13 *
14 * The above copyright notice and this permission notice shall be included in
15 * all copies or substantial portions of the Software.
16 *
17 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
18 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
19 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
20 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
21 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
22 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
23 * SOFTWARE.
24 *
25 * Authors:
26 * Maryam Tahhan <maryam.tahhan@intel.com>
27 * Harry van Haaren <harry.van.haaren@intel.com>
28 * Taras Chornyi <tarasx.chornyi@intel.com>
29 */
31 #include "collectd.h"
33 #include "common.h" /* auxiliary functions */
34 #include "plugin.h" /* plugin_register_*, plugin_dispatch_values */
35 #include "utils_time.h"
37 #include <getopt.h>
38 #include <poll.h>
39 #include <semaphore.h>
40 #include <sys/mman.h>
41 #include <sys/queue.h>
43 #include <rte_atomic.h>
44 #include <rte_branch_prediction.h>
45 #include <rte_common.h>
46 #include <rte_config.h>
47 #include <rte_debug.h>
48 #include <rte_debug.h>
49 #include <rte_eal.h>
50 #include <rte_ethdev.h>
51 #include <rte_launch.h>
52 #include <rte_lcore.h>
53 #include <rte_log.h>
54 #include <rte_malloc.h>
55 #include <rte_memory.h>
56 #include <rte_memzone.h>
57 #include <rte_per_lcore.h>
58 #include <rte_string_fns.h>
59 #include <rte_tailq.h>
60 #include <rte_version.h>
62 #define DPDK_DEFAULT_RTE_CONFIG "/var/run/.rte_config"
63 #define DPDK_MAX_ARGC 8
64 #define DPDKSTAT_MAX_BUFFER_SIZE (4096 * 4)
65 #define DPDK_SHM_NAME "dpdk_collectd_stats_shm"
66 #define ERR_BUF_SIZE 1024
67 #define REINIT_SHM 1
68 #define RESET 1
69 #define NO_RESET 0
71 #define RTE_VERSION_16_07 RTE_VERSION_NUM(16, 7, 0, 16)
73 #if RTE_VERSION < RTE_VERSION_16_07
74 #define DPDK_STATS_XSTAT_GET_VALUE(ctx, index) ctx->xstats[index].value
75 #define DPDK_STATS_XSTAT_GET_NAME(ctx, index) ctx->xstats[index].name
76 #define DPDK_STATS_CTX_GET_XSTAT_SIZE sizeof(struct rte_eth_xstats)
77 #define DPDK_STATS_CTX_INIT(ctx) \
78 do { \
79 ctx->xstats = (struct rte_eth_xstats *)&ctx->raw_data[0]; \
80 } while (0)
81 #else
82 #define DPDK_STATS_XSTAT_GET_VALUE(ctx, index) ctx->xstats[index].value
83 #define DPDK_STATS_XSTAT_GET_NAME(ctx, index) ctx->xnames[index].name
84 #define DPDK_STATS_CTX_GET_XSTAT_SIZE \
85 (sizeof(struct rte_eth_xstat) + sizeof(struct rte_eth_xstat_name))
86 #define DPDK_STATS_CTX_INIT(ctx) \
87 do { \
88 ctx->xstats = (struct rte_eth_xstat *)&ctx->raw_data[0]; \
89 ctx->xnames = \
90 (struct rte_eth_xstat_name *)&ctx \
91 ->raw_data[ctx->num_xstats * sizeof(struct rte_eth_xstat)]; \
92 } while (0)
93 #endif
95 enum DPDK_HELPER_ACTION {
96 DPDK_HELPER_ACTION_COUNT_STATS,
97 DPDK_HELPER_ACTION_SEND_STATS,
98 };
100 enum DPDK_HELPER_STATUS {
101 DPDK_HELPER_NOT_INITIALIZED = 0,
102 DPDK_HELPER_WAITING_ON_PRIMARY,
103 DPDK_HELPER_INITIALIZING_EAL,
104 DPDK_HELPER_ALIVE_SENDING_STATS,
105 DPDK_HELPER_GRACEFUL_QUIT,
106 };
108 struct dpdk_config_s {
109 /* General DPDK params */
110 char coremask[DATA_MAX_NAME_LEN];
111 char memory_channels[DATA_MAX_NAME_LEN];
112 char socket_memory[DATA_MAX_NAME_LEN];
113 char process_type[DATA_MAX_NAME_LEN];
114 char file_prefix[DATA_MAX_NAME_LEN];
115 cdtime_t interval;
116 uint32_t eal_initialized;
117 uint32_t enabled_port_mask;
118 char port_name[RTE_MAX_ETHPORTS][DATA_MAX_NAME_LEN];
119 uint32_t eal_argc;
120 /* Helper info */
121 int collectd_reinit_shm;
122 pid_t helper_pid;
123 sem_t sema_helper_get_stats;
124 sem_t sema_stats_in_shm;
125 int helper_pipes[2];
126 enum DPDK_HELPER_STATUS helper_status;
127 enum DPDK_HELPER_ACTION helper_action;
128 /* xstats info */
129 uint32_t num_ports;
130 uint32_t num_xstats;
131 cdtime_t port_read_time[RTE_MAX_ETHPORTS];
132 uint32_t num_stats_in_port[RTE_MAX_ETHPORTS];
133 struct rte_eth_link link_status[RTE_MAX_ETHPORTS];
134 #if RTE_VERSION < RTE_VERSION_16_07
135 struct rte_eth_xstats *xstats;
136 #else
137 struct rte_eth_xstat *xstats;
138 struct rte_eth_xstat_name *xnames;
139 #endif
140 char *raw_data;
141 /* rte_eth_xstats from here on until the end of the SHM */
142 };
143 typedef struct dpdk_config_s dpdk_config_t;
145 static int g_configured;
146 static dpdk_config_t *g_configuration;
148 static void dpdk_config_init_default(void);
149 static int dpdk_config(oconfig_item_t *ci);
150 static int dpdk_helper_init_eal(void);
151 static int dpdk_helper_run(void);
152 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action);
153 static int dpdk_init(void);
154 static int dpdk_read(user_data_t *ud);
155 static int dpdk_shm_cleanup(void);
156 static int dpdk_shm_init(size_t size);
158 /* Write the default configuration to the g_configuration instances */
159 static void dpdk_config_init_default(void) {
160 g_configuration->interval = plugin_get_interval();
161 if (g_configuration->interval == cf_get_default_interval())
162 WARNING("dpdkstat: No time interval was configured, default value %lu ms "
163 "is set",
164 CDTIME_T_TO_MS(g_configuration->interval));
165 /* Default is all ports enabled */
166 g_configuration->enabled_port_mask = ~0;
167 g_configuration->eal_argc = DPDK_MAX_ARGC;
168 g_configuration->eal_initialized = 0;
169 ssnprintf(g_configuration->coremask, DATA_MAX_NAME_LEN, "%s", "0xf");
170 ssnprintf(g_configuration->memory_channels, DATA_MAX_NAME_LEN, "%s", "1");
171 ssnprintf(g_configuration->process_type, DATA_MAX_NAME_LEN, "%s",
172 "secondary");
173 ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN, "%s",
174 DPDK_DEFAULT_RTE_CONFIG);
176 for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
177 g_configuration->port_name[i][0] = 0;
178 }
180 static int dpdk_config(oconfig_item_t *ci) {
181 int port_counter = 0;
182 /* Allocate g_configuration and
183 * initialize a POSIX SHared Memory (SHM) object.
184 */
185 int err = dpdk_shm_init(sizeof(dpdk_config_t));
186 if (err) {
187 char errbuf[ERR_BUF_SIZE];
188 ERROR("dpdkstat: error in shm_init, %s",
189 sstrerror(errno, errbuf, sizeof(errbuf)));
190 return -1;
191 }
193 /* Set defaults for config, overwritten by loop if config item exists */
194 dpdk_config_init_default();
196 for (int i = 0; i < ci->children_num; i++) {
197 oconfig_item_t *child = ci->children + i;
199 if (strcasecmp("Coremask", child->key) == 0) {
200 cf_util_get_string_buffer(child, g_configuration->coremask,
201 sizeof(g_configuration->coremask));
202 DEBUG("dpdkstat:COREMASK %s ", g_configuration->coremask);
203 } else if (strcasecmp("MemoryChannels", child->key) == 0) {
204 cf_util_get_string_buffer(child, g_configuration->memory_channels,
205 sizeof(g_configuration->memory_channels));
206 DEBUG("dpdkstat:Memory Channels %s ", g_configuration->memory_channels);
207 } else if (strcasecmp("SocketMemory", child->key) == 0) {
208 cf_util_get_string_buffer(child, g_configuration->socket_memory,
209 sizeof(g_configuration->memory_channels));
210 DEBUG("dpdkstat: socket mem %s ", g_configuration->socket_memory);
211 } else if (strcasecmp("ProcessType", child->key) == 0) {
212 cf_util_get_string_buffer(child, g_configuration->process_type,
213 sizeof(g_configuration->process_type));
214 DEBUG("dpdkstat: proc type %s ", g_configuration->process_type);
215 } else if ((strcasecmp("FilePrefix", child->key) == 0) &&
216 (child->values[0].type == OCONFIG_TYPE_STRING)) {
217 ssnprintf(g_configuration->file_prefix, DATA_MAX_NAME_LEN,
218 "/var/run/.%s_config", child->values[0].value.string);
219 DEBUG("dpdkstat: file prefix %s ", g_configuration->file_prefix);
220 } else if ((strcasecmp("EnabledPortMask", child->key) == 0) &&
221 (child->values[0].type == OCONFIG_TYPE_NUMBER)) {
222 g_configuration->enabled_port_mask =
223 (uint32_t)child->values[0].value.number;
224 DEBUG("dpdkstat: Enabled Port Mask %u",
225 g_configuration->enabled_port_mask);
226 } else if (strcasecmp("PortName", child->key) == 0) {
227 cf_util_get_string_buffer(
228 child, g_configuration->port_name[port_counter],
229 sizeof(g_configuration->port_name[port_counter]));
230 DEBUG("dpdkstat: Port %d Name: %s ", port_counter,
231 g_configuration->port_name[port_counter]);
232 port_counter++;
233 } else {
234 WARNING("dpdkstat: The config option \"%s\" is unknown.", child->key);
235 }
236 } /* End for (int i = 0; i < ci->children_num; i++)*/
237 g_configured = 1; /* Bypass configuration in dpdk_shm_init(). */
239 return 0;
240 }
242 /*
243 * Allocate g_configuration and initialize SHared Memory (SHM)
244 * for config and helper process
245 */
246 static int dpdk_shm_init(size_t size) {
247 /*
248 * Check if SHM is already configured: when config items are provided, the
249 * config function initializes SHM. If there is no config, then init() will
250 * just return.
251 */
252 if (g_configuration)
253 return 0;
255 char errbuf[ERR_BUF_SIZE];
257 /* Create and open a new object, or open an existing object. */
258 int fd = shm_open(DPDK_SHM_NAME, O_CREAT | O_TRUNC | O_RDWR, 0666);
259 if (fd < 0) {
260 WARNING("dpdkstat:Failed to open %s as SHM:%s", DPDK_SHM_NAME,
261 sstrerror(errno, errbuf, sizeof(errbuf)));
262 goto fail;
263 }
264 /* Set the size of the shared memory object. */
265 int ret = ftruncate(fd, size);
266 if (ret != 0) {
267 WARNING("dpdkstat:Failed to resize SHM:%s",
268 sstrerror(errno, errbuf, sizeof(errbuf)));
269 goto fail_close;
270 }
271 /* Map the shared memory object into this process' virtual address space. */
272 g_configuration = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
273 if (g_configuration == MAP_FAILED) {
274 WARNING("dpdkstat:Failed to mmap SHM:%s",
275 sstrerror(errno, errbuf, sizeof(errbuf)));
276 goto fail_close;
277 }
278 /*
279 * Close the file descriptor, the shared memory object still exists
280 * and can only be removed by calling shm_unlink().
281 */
282 close(fd);
284 /* Initialize g_configuration. */
285 memset(g_configuration, 0, size);
287 /* Initialize the semaphores for SHM use */
288 int err = sem_init(&g_configuration->sema_helper_get_stats, 1, 0);
289 if (err) {
290 ERROR("dpdkstat semaphore init failed: %s",
291 sstrerror(errno, errbuf, sizeof(errbuf)));
292 goto fail_close;
293 }
294 err = sem_init(&g_configuration->sema_stats_in_shm, 1, 0);
295 if (err) {
296 ERROR("dpdkstat semaphore init failed: %s",
297 sstrerror(errno, errbuf, sizeof(errbuf)));
298 goto fail_close;
299 }
301 g_configuration->xstats = NULL;
303 return 0;
305 fail_close:
306 close(fd);
307 fail:
308 /* Reset to zero, as it was set to MAP_FAILED aka: (void *)-1. Avoid
309 * an issue if collectd attempts to run this plugin failure.
310 */
311 g_configuration = 0;
312 return -1;
313 }
315 static int dpdk_re_init_shm() {
316 dpdk_config_t temp_config;
317 memcpy(&temp_config, g_configuration, sizeof(dpdk_config_t));
318 DEBUG("dpdkstat: %s: ports %" PRIu32 ", xstats %" PRIu32, __func__,
319 temp_config.num_ports, temp_config.num_xstats);
321 size_t shm_xstats_size =
322 sizeof(dpdk_config_t) +
323 (DPDK_STATS_CTX_GET_XSTAT_SIZE * g_configuration->num_xstats);
324 DEBUG("=== SHM new size for %" PRIu32 " xstats", g_configuration->num_xstats);
326 int err = dpdk_shm_cleanup();
327 if (err) {
328 ERROR("dpdkstat: Error in shm_cleanup in %s", __func__);
329 return err;
330 }
331 err = dpdk_shm_init(shm_xstats_size);
332 if (err) {
333 WARNING("dpdkstat: Error in shm_init in %s", __func__);
334 return err;
335 }
336 /* If the XML config() function has been run, don't re-initialize defaults */
337 if (!g_configured)
338 dpdk_config_init_default();
340 memcpy(g_configuration, &temp_config, sizeof(dpdk_config_t));
341 g_configuration->collectd_reinit_shm = 0;
342 g_configuration->raw_data = (char *)(g_configuration + 1);
343 DPDK_STATS_CTX_INIT(g_configuration);
344 return 0;
345 }
347 static int dpdk_init(void) {
348 int err = dpdk_shm_init(sizeof(dpdk_config_t));
349 if (err) {
350 ERROR("dpdkstat: %s : error %d in shm_init()", __func__, err);
351 return err;
352 }
354 /* If the XML config() function has been run, dont re-initialize defaults */
355 if (!g_configured) {
356 dpdk_config_init_default();
357 }
359 return 0;
360 }
362 static int dpdk_helper_stop(int reset) {
363 g_configuration->helper_status = DPDK_HELPER_GRACEFUL_QUIT;
364 if (reset) {
365 g_configuration->eal_initialized = 0;
366 g_configuration->num_ports = 0;
367 g_configuration->xstats = NULL;
368 g_configuration->num_xstats = 0;
369 for (int i = 0; i < RTE_MAX_ETHPORTS; i++)
370 g_configuration->num_stats_in_port[i] = 0;
371 }
372 close(g_configuration->helper_pipes[1]);
373 int err = kill(g_configuration->helper_pid, SIGKILL);
374 if (err) {
375 char errbuf[ERR_BUF_SIZE];
376 WARNING("dpdkstat: error sending kill to helper: %s",
377 sstrerror(errno, errbuf, sizeof(errbuf)));
378 }
380 return 0;
381 }
383 static int dpdk_helper_spawn(enum DPDK_HELPER_ACTION action) {
384 char errbuf[ERR_BUF_SIZE];
385 g_configuration->eal_initialized = 0;
386 g_configuration->helper_action = action;
387 /*
388 * Create a pipe for helper stdout back to collectd. This is necessary for
389 * logging EAL failures, as rte_eal_init() calls rte_panic().
390 */
391 if (pipe(g_configuration->helper_pipes) != 0) {
392 DEBUG("dpdkstat: Could not create helper pipe: %s",
393 sstrerror(errno, errbuf, sizeof(errbuf)));
394 return -1;
395 }
397 int pipe0_flags = fcntl(g_configuration->helper_pipes[0], F_GETFL, 0);
398 int pipe1_flags = fcntl(g_configuration->helper_pipes[1], F_GETFL, 0);
399 if (pipe0_flags == -1 || pipe1_flags == -1) {
400 WARNING("dpdkstat: Failed setting up pipe flags: %s",
401 sstrerror(errno, errbuf, sizeof(errbuf)));
402 }
403 int pipe0_err = fcntl(g_configuration->helper_pipes[0], F_SETFL,
404 pipe1_flags | O_NONBLOCK);
405 int pipe1_err = fcntl(g_configuration->helper_pipes[1], F_SETFL,
406 pipe0_flags | O_NONBLOCK);
407 if (pipe0_err == -1 || pipe1_err == -1) {
408 WARNING("dpdkstat: Failed setting up pipes: %s",
409 sstrerror(errno, errbuf, sizeof(errbuf)));
410 }
412 pid_t pid = fork();
413 if (pid > 0) {
414 close(g_configuration->helper_pipes[1]);
415 g_configuration->helper_pid = pid;
416 DEBUG("dpdkstat: helper pid %lu", (long)g_configuration->helper_pid);
417 /* Kick helper once its alive to have it start processing */
418 sem_post(&g_configuration->sema_helper_get_stats);
419 } else if (pid == 0) {
420 /* Replace stdout with a pipe to collectd. */
421 close(g_configuration->helper_pipes[0]);
422 close(STDOUT_FILENO);
423 dup2(g_configuration->helper_pipes[1], STDOUT_FILENO);
424 dpdk_helper_run();
425 exit(0);
426 } else {
427 ERROR("dpdkstat: Failed to fork helper process: %s",
428 sstrerror(errno, errbuf, sizeof(errbuf)));
429 return -1;
430 }
431 return 0;
432 }
434 /*
435 * Initialize the DPDK EAL, if this returns, EAL is successfully initialized.
436 * On failure, the EAL prints an error message, and the helper process exits.
437 */
438 static int dpdk_helper_init_eal(void) {
439 g_configuration->helper_status = DPDK_HELPER_INITIALIZING_EAL;
440 char *argp[(g_configuration->eal_argc) + 1];
441 int i = 0;
443 argp[i++] = "collectd-dpdk";
444 if (strcasecmp(g_configuration->coremask, "") != 0) {
445 argp[i++] = "-c";
446 argp[i++] = g_configuration->coremask;
447 }
448 if (strcasecmp(g_configuration->memory_channels, "") != 0) {
449 argp[i++] = "-n";
450 argp[i++] = g_configuration->memory_channels;
451 }
452 if (strcasecmp(g_configuration->socket_memory, "") != 0) {
453 argp[i++] = "--socket-mem";
454 argp[i++] = g_configuration->socket_memory;
455 }
456 if (strcasecmp(g_configuration->file_prefix, "") != 0 &&
457 strcasecmp(g_configuration->file_prefix, DPDK_DEFAULT_RTE_CONFIG) != 0) {
458 argp[i++] = "--file-prefix";
459 argp[i++] = g_configuration->file_prefix;
460 }
461 if (strcasecmp(g_configuration->process_type, "") != 0) {
462 argp[i++] = "--proc-type";
463 argp[i++] = g_configuration->process_type;
464 }
465 g_configuration->eal_argc = i;
467 g_configuration->eal_initialized = 1;
468 int ret = rte_eal_init(g_configuration->eal_argc, argp);
469 if (ret < 0) {
470 g_configuration->eal_initialized = 0;
471 return ret;
472 }
473 return 0;
474 }
476 static int dpdk_helper_run(void) {
477 char errbuf[ERR_BUF_SIZE];
478 pid_t ppid = getppid();
479 g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
481 while (1) {
482 /* sem_timedwait() to avoid blocking forever */
483 cdtime_t now = cdtime();
484 cdtime_t safety_period = MS_TO_CDTIME_T(1500);
485 int ret =
486 sem_timedwait(&g_configuration->sema_helper_get_stats,
487 &CDTIME_T_TO_TIMESPEC(now + safety_period +
488 g_configuration->interval * 2));
490 if (ret == -1 && errno == ETIMEDOUT) {
491 ERROR("dpdkstat-helper: sem timedwait()"
492 " timeout, did collectd terminate?");
493 dpdk_helper_stop(RESET);
494 }
495 /* Parent PID change means collectd died so quit the helper process. */
496 if (ppid != getppid()) {
497 WARNING("dpdkstat-helper: parent PID changed, quitting.");
498 dpdk_helper_stop(RESET);
499 }
501 /* Checking for DPDK primary process. */
502 if (!rte_eal_primary_proc_alive(g_configuration->file_prefix)) {
503 if (g_configuration->eal_initialized) {
504 WARNING("dpdkstat-helper: no primary alive but EAL initialized:"
505 " quitting.");
506 dpdk_helper_stop(RESET);
507 }
508 g_configuration->helper_status = DPDK_HELPER_WAITING_ON_PRIMARY;
509 /* Back to start of while() - waiting for primary process */
510 continue;
511 }
513 if (!g_configuration->eal_initialized) {
514 /* Initialize EAL. */
515 int ret = dpdk_helper_init_eal();
516 if (ret != 0) {
517 WARNING("ERROR INITIALIZING EAL");
518 dpdk_helper_stop(RESET);
519 }
520 }
522 g_configuration->helper_status = DPDK_HELPER_ALIVE_SENDING_STATS;
524 uint8_t nb_ports = rte_eth_dev_count();
525 if (nb_ports == 0) {
526 DEBUG("dpdkstat-helper: No DPDK ports available. "
527 "Check bound devices to DPDK driver.");
528 dpdk_helper_stop(RESET);
529 }
531 if (nb_ports > RTE_MAX_ETHPORTS)
532 nb_ports = RTE_MAX_ETHPORTS;
534 int len = 0, enabled_port_count = 0, num_xstats = 0;
535 for (uint8_t i = 0; i < nb_ports; i++) {
536 if (!(g_configuration->enabled_port_mask & (1 << i)))
537 continue;
539 if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
540 #if RTE_VERSION >= RTE_VERSION_16_07
541 len = rte_eth_xstats_get_names(i, NULL, 0);
542 #else
543 len = rte_eth_xstats_get(i, NULL, 0);
544 #endif
545 if (len < 0) {
546 ERROR("dpdkstat-helper: Cannot get xstats count on port %" PRIu8, i);
547 break;
548 }
549 num_xstats += len;
550 g_configuration->num_stats_in_port[enabled_port_count] = len;
551 enabled_port_count++;
552 continue;
553 } else {
554 len = g_configuration->num_stats_in_port[enabled_port_count];
555 g_configuration->port_read_time[enabled_port_count] = cdtime();
556 ret = rte_eth_xstats_get(
557 i, g_configuration->xstats + num_xstats,
558 g_configuration->num_stats_in_port[enabled_port_count]);
559 if (ret < 0 || ret != len) {
560 DEBUG("dpdkstat-helper: Error reading xstats on port %" PRIu8
561 " len = %d",
562 i, len);
563 break;
564 }
565 #if RTE_VERSION >= RTE_VERSION_16_07
566 ret = rte_eth_xstats_get_names(i, g_configuration->xnames + num_xstats,
567 len);
568 if (ret < 0 || ret != len) {
569 ERROR("dpdkstat-helper: Error reading xstat names (port=%d; len=%d)",
570 i, len);
571 break;
572 }
573 #endif
574 num_xstats += g_configuration->num_stats_in_port[enabled_port_count];
575 enabled_port_count++;
576 }
577 } /* for (nb_ports) */
579 if (g_configuration->helper_action == DPDK_HELPER_ACTION_COUNT_STATS) {
580 g_configuration->num_ports = enabled_port_count;
581 g_configuration->num_xstats = num_xstats;
582 DEBUG("dpdkstat-helper ports: %" PRIu32 ", num stats: %" PRIu32,
583 g_configuration->num_ports, g_configuration->num_xstats);
584 /* Exit, allowing collectd to re-init SHM to the right size */
585 g_configuration->collectd_reinit_shm = REINIT_SHM;
586 dpdk_helper_stop(NO_RESET);
587 }
588 /* Now kick collectd send thread to send the stats */
589 int err = sem_post(&g_configuration->sema_stats_in_shm);
590 if (err) {
591 WARNING("dpdkstat: error posting semaphore to helper %s",
592 sstrerror(errno, errbuf, sizeof(errbuf)));
593 dpdk_helper_stop(RESET);
594 }
595 } /* while(1) */
597 return 0;
598 }
600 static void dpdk_submit_xstats(const char *dev_name, int count,
601 uint32_t counters, cdtime_t port_read_time) {
602 for (uint32_t j = 0; j < counters; j++) {
603 value_list_t vl = VALUE_LIST_INIT;
604 char *counter_name;
605 char *type_end;
607 vl.values = &(value_t){.derive = (derive_t)DPDK_STATS_XSTAT_GET_VALUE(
608 g_configuration, count + j)};
609 vl.values_len = 1; /* Submit stats one at a time */
610 vl.time = port_read_time;
611 sstrncpy(vl.plugin, "dpdkstat", sizeof(vl.plugin));
612 sstrncpy(vl.plugin_instance, dev_name, sizeof(vl.plugin_instance));
613 counter_name = DPDK_STATS_XSTAT_GET_NAME(g_configuration, count + j);
614 if (counter_name == NULL) {
615 WARNING("dpdkstat: Failed to get counter name.");
616 return;
617 }
619 type_end = strrchr(counter_name, '_');
621 if ((type_end != NULL) &&
622 (strncmp(counter_name, "rx_", strlen("rx_")) == 0)) {
623 if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
624 sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
625 } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
626 sstrncpy(vl.type, "if_rx_dropped", sizeof(vl.type));
627 } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
628 sstrncpy(vl.type, "if_rx_octets", sizeof(vl.type));
629 } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
630 sstrncpy(vl.type, "if_rx_packets", sizeof(vl.type));
631 } else if (strncmp(type_end, "_placement", strlen("_placement")) == 0) {
632 sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
633 } else if (strncmp(type_end, "_buff", strlen("_buff")) == 0) {
634 sstrncpy(vl.type, "if_rx_errors", sizeof(vl.type));
635 } else {
636 /* Does not fit obvious type: use a more generic one */
637 sstrncpy(vl.type, "derive", sizeof(vl.type));
638 }
640 } else if ((type_end != NULL) &&
641 (strncmp(counter_name, "tx_", strlen("tx_"))) == 0) {
642 if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
643 sstrncpy(vl.type, "if_tx_errors", sizeof(vl.type));
644 } else if (strncmp(type_end, "_dropped", strlen("_dropped")) == 0) {
645 sstrncpy(vl.type, "if_tx_dropped", sizeof(vl.type));
646 } else if (strncmp(type_end, "_bytes", strlen("_bytes")) == 0) {
647 sstrncpy(vl.type, "if_tx_octets", sizeof(vl.type));
648 } else if (strncmp(type_end, "_packets", strlen("_packets")) == 0) {
649 sstrncpy(vl.type, "if_tx_packets", sizeof(vl.type));
650 } else {
651 /* Does not fit obvious type: use a more generic one */
652 sstrncpy(vl.type, "derive", sizeof(vl.type));
653 }
654 } else if ((type_end != NULL) &&
655 (strncmp(counter_name, "flow_", strlen("flow_"))) == 0) {
657 if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
658 sstrncpy(vl.type, "operations", sizeof(vl.type));
659 } else if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
660 sstrncpy(vl.type, "errors", sizeof(vl.type));
661 } else if (strncmp(type_end, "_filters", strlen("_filters")) == 0) {
662 sstrncpy(vl.type, "filter_result", sizeof(vl.type));
663 }
664 } else if ((type_end != NULL) &&
665 (strncmp(counter_name, "mac_", strlen("mac_"))) == 0) {
666 if (strncmp(type_end, "_errors", strlen("_errors")) == 0) {
667 sstrncpy(vl.type, "errors", sizeof(vl.type));
668 }
669 } else {
670 /* Does not fit obvious type, or strrchr error:
671 * use a more generic type */
672 sstrncpy(vl.type, "derive", sizeof(vl.type));
673 }
675 sstrncpy(vl.type_instance, counter_name, sizeof(vl.type_instance));
676 plugin_dispatch_values(&vl);
677 }
678 }
680 static int dpdk_read(user_data_t *ud) {
681 int ret = 0;
683 /*
684 * Check if SHM flag is set to be re-initialized. AKA DPDK ports have been
685 * counted, so re-init SHM to be large enough to fit all the statistics.
686 */
687 if (g_configuration->collectd_reinit_shm) {
688 DEBUG("dpdkstat: read() now reinit SHM then launching send-thread");
689 dpdk_re_init_shm();
690 }
692 /*
693 * Check if DPDK proc is alive, and has already counted port / stats. This
694 * must be done in dpdk_read(), because the DPDK primary process may not be
695 * alive at dpdk_init() time.
696 */
697 if (g_configuration->helper_status == DPDK_HELPER_NOT_INITIALIZED ||
698 g_configuration->helper_status == DPDK_HELPER_GRACEFUL_QUIT) {
699 int action = DPDK_HELPER_ACTION_SEND_STATS;
700 if (g_configuration->num_xstats == 0)
701 action = DPDK_HELPER_ACTION_COUNT_STATS;
702 /* Spawn the helper thread to count stats or to read stats. */
703 int err = dpdk_helper_spawn(action);
704 if (err) {
705 char errbuf[ERR_BUF_SIZE];
706 ERROR("dpdkstat: error spawning helper %s",
707 sstrerror(errno, errbuf, sizeof(errbuf)));
708 return -1;
709 }
710 }
712 pid_t ws = waitpid(g_configuration->helper_pid, NULL, WNOHANG);
713 /*
714 * Conditions under which to respawn helper:
715 * waitpid() fails, helper process died (or quit), so respawn
716 */
717 _Bool respawn_helper = 0;
718 if (ws != 0) {
719 respawn_helper = 1;
720 }
722 char buf[DPDKSTAT_MAX_BUFFER_SIZE];
723 char out[DPDKSTAT_MAX_BUFFER_SIZE];
725 /* non blocking check on helper logging pipe */
726 struct pollfd fds = {
727 .fd = g_configuration->helper_pipes[0], .events = POLLIN,
728 };
729 int data_avail = poll(&fds, 1, 0);
730 if (data_avail < 0) {
731 char errbuf[ERR_BUF_SIZE];
732 if (errno != EINTR || errno != EAGAIN)
733 ERROR("dpdkstats: poll(2) failed: %s",
734 sstrerror(errno, errbuf, sizeof(errbuf)));
735 }
736 while (data_avail) {
737 int nbytes = read(g_configuration->helper_pipes[0], buf, sizeof(buf));
738 if (nbytes <= 0)
739 break;
740 ssnprintf(out, nbytes, "%s", buf);
741 DEBUG("dpdkstat: helper-proc: %s", out);
742 }
744 if (respawn_helper) {
745 if (g_configuration->helper_pid)
746 dpdk_helper_stop(RESET);
747 dpdk_helper_spawn(DPDK_HELPER_ACTION_COUNT_STATS);
748 }
750 /* Kick helper process through SHM */
751 sem_post(&g_configuration->sema_helper_get_stats);
753 cdtime_t now = cdtime();
754 ret = sem_timedwait(&g_configuration->sema_stats_in_shm,
755 &CDTIME_T_TO_TIMESPEC(now + g_configuration->interval));
756 if (ret == -1) {
757 if (errno == ETIMEDOUT)
758 DEBUG(
759 "dpdkstat: timeout in collectd thread: is a DPDK Primary running? ");
760 return 0;
761 }
763 /* Dispatch the stats.*/
764 uint32_t count = 0, port_num = 0;
766 for (uint32_t i = 0; i < g_configuration->num_ports; i++) {
767 char dev_name[64];
768 cdtime_t port_read_time = g_configuration->port_read_time[i];
769 uint32_t counters_num = g_configuration->num_stats_in_port[i];
770 size_t ports_max = CHAR_BIT * sizeof(g_configuration->enabled_port_mask);
771 for (size_t j = port_num; j < ports_max; j++) {
772 if ((g_configuration->enabled_port_mask & (1 << j)) != 0)
773 break;
774 port_num++;
775 }
777 if (g_configuration->port_name[i][0] != 0)
778 ssnprintf(dev_name, sizeof(dev_name), "%s",
779 g_configuration->port_name[i]);
780 else
781 ssnprintf(dev_name, sizeof(dev_name), "port.%" PRIu32, port_num);
782 dpdk_submit_xstats(dev_name, count, counters_num, port_read_time);
783 count += counters_num;
784 port_num++;
785 } /* for each port */
786 return 0;
787 }
789 static int dpdk_shm_cleanup(void) {
790 int ret = munmap(g_configuration, sizeof(dpdk_config_t));
791 g_configuration = 0;
792 if (ret) {
793 ERROR("dpdkstat: munmap returned %d", ret);
794 return ret;
795 }
796 ret = shm_unlink(DPDK_SHM_NAME);
797 if (ret) {
798 ERROR("dpdkstat: shm_unlink returned %d", ret);
799 return ret;
800 }
801 return 0;
802 }
804 static int dpdk_shutdown(void) {
805 int ret = 0;
806 char errbuf[ERR_BUF_SIZE];
807 close(g_configuration->helper_pipes[1]);
808 int err = kill(g_configuration->helper_pid, SIGKILL);
809 if (err) {
810 ERROR("dpdkstat: error sending sigkill to helper %s",
811 sstrerror(errno, errbuf, sizeof(errbuf)));
812 ret = -1;
813 }
814 err = dpdk_shm_cleanup();
815 if (err) {
816 ERROR("dpdkstat: error cleaning up SHM: %s",
817 sstrerror(errno, errbuf, sizeof(errbuf)));
818 ret = -1;
819 }
821 return ret;
822 }
824 void module_register(void) {
825 plugin_register_complex_config("dpdkstat", dpdk_config);
826 plugin_register_init("dpdkstat", dpdk_init);
827 plugin_register_complex_read(NULL, "dpdkstat", dpdk_read, 0, NULL);
828 plugin_register_shutdown("dpdkstat", dpdk_shutdown);
829 }