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