@cryptotaxi247 / netdata-1 / commits / acd59f419

eBPF socket: function with event loop (#16004)

thiagoftsm committed Sep 19, 2023 at 09:42 UTC acd59f41955ca66d07343debac7da333e47c62fa

2 files changed +32 -74

collectors/ebpf.plugin/ebpf.c

@@ -3815,9 +3815,11 @@ static void ebpf_create_statistic_charts(int update_every)

3815 NETDATA_EBPF_ORDER_STAT_THREADS,

3816 update_every,

3817 NULL);

3818 + /*

3819 #ifdef NETDATA_DEV_MODE

3820 EBPF_PLUGIN_FUNCTIONS(EBPF_FUNCTION_THREAD, EBPF_PLUGIN_THREAD_FUNCTION_DESCRIPTION);

3821 #endif

3822 + */

3823

3824 ebpf_create_thread_chart(NETDATA_EBPF_LIFE_TIME,

3825 "Time remaining for thread.",

@@ -3825,9 +3827,11 @@ static void ebpf_create_statistic_charts(int update_every)

3827 NETDATA_EBPF_ORDER_STAT_LIFE_TIME,

3828 update_every,

3829 NULL);

3830 + /*

3831 #ifdef NETDATA_DEV_MODE

3832 EBPF_PLUGIN_FUNCTIONS(EBPF_FUNCTION_THREAD, EBPF_PLUGIN_THREAD_FUNCTION_DESCRIPTION);

3833 #endif

3834 + */

3835

3836 int i,j;

3837 char name[256];

collectors/ebpf.plugin/ebpf_functions.c

+28 -74

@@ -7,8 +7,6 @@

7 * EBPF FUNCTION COMMON

8 *****************************************************************/

10 -RW_SPINLOCK rw_spinlock; // protect the buffer

11 -

10 /**

11 * Function Start thread

12 *

@@ -49,7 +47,6 @@ static int ebpf_function_start_thread(ebpf_module_t *em, int period)

47 * @param thread_name name of the thread we are looking for.

48 *

49 * @return it returns a pointer for the module that has thread_name on success or NULL otherwise.

52 - */

50 ebpf_module_t *ebpf_functions_select_module(const char *thread_name) {

51 int i;

52 for (i = 0; i < EBPF_MODULE_FUNCTION_IDX; i++) {

@@ -60,6 +57,7 @@ ebpf_module_t *ebpf_functions_select_module(const char *thread_name) {

58 return NULL;

59 }

60 + */

62 /*****************************************************************

63 * EBPF HELP FUNCTIONS

@@ -71,7 +69,6 @@ ebpf_module_t *ebpf_functions_select_module(const char *thread_name) {

69 * Shows help with all options accepted by thread function.

70 *

71 * @param transaction the transaction id that Netdata sent for this function execution

74 -*/

72 static void ebpf_function_thread_manipulation_help(const char *transaction) {

73 BUFFER *wb = buffer_create(0, NULL);

74 buffer_sprintf(wb, "%s",

@@ -94,12 +91,11 @@ static void ebpf_function_thread_manipulation_help(const char *transaction) {

91 "Filters can be combined. Each filter can be given only one time.\n"

92 );

97 - pthread_mutex_lock(&lock);

94 pluginsd_function_result_to_stdout(transaction, HTTP_RESP_OK, "text/plain", now_realtime_sec() + 3600, wb);

99 - pthread_mutex_unlock(&lock);

96 buffer_free(wb);

97 }

98 +*/

100 /*****************************************************************

101 * EBPF ERROR FUNCTIONS

@@ -115,9 +111,7 @@ static void ebpf_function_thread_manipulation_help(const char *transaction) {

111 * @param msg the error message

112 */

113 static void ebpf_function_error(const char *transaction, int code, const char *msg) {

118 - pthread_mutex_lock(&lock);

114 pluginsd_function_json_error_to_stdout(transaction, code, msg);

120 - pthread_mutex_unlock(&lock);

115 }

116

117 /*****************************************************************

@@ -135,7 +129,6 @@ static void ebpf_function_error(const char *transaction, int code, const char *m

129 * @param line_max Number of arguments given

130 * @param timeout The function timeout

131 * @param em The structure with thread information

138 - */

132 static void ebpf_function_thread_manipulation(const char *transaction,

133 char *function __maybe_unused,

134 char *line_buffer __maybe_unused,

@@ -370,12 +363,11 @@ static void ebpf_function_thread_manipulation(const char *transaction,

363 buffer_json_finalize(wb);

364

365 // Lock necessary to avoid race condition

373 - pthread_mutex_lock(&lock);

366 pluginsd_function_result_to_stdout(transaction, HTTP_RESP_OK, "application/json", expires, wb);

375 - pthread_mutex_unlock(&lock);

367

368 buffer_free(wb);

369 }

370 + */

371

372 /*****************************************************************

373 * EBPF SOCKET FUNCTION

@@ -389,7 +381,6 @@ static void ebpf_function_thread_manipulation(const char *transaction,

381 * @param transaction the transaction id that Netdata sent for this function execution

382 */

383 static void ebpf_function_socket_help(const char *transaction) {

392 - pthread_mutex_lock(&lock);

384 pluginsd_function_result_begin_to_stdout(transaction, HTTP_RESP_OK, "text/plain", now_realtime_sec() + 3600);

385 fprintf(stdout, "%s",

386 "ebpf.plugin / socket\n"

@@ -427,7 +418,6 @@ static void ebpf_function_socket_help(const char *transaction) {

418 );

419 pluginsd_function_result_end_to_stdout();

420 fflush(stdout);

430 - pthread_mutex_unlock(&lock);

421 }

422

423 /**

@@ -644,20 +634,16 @@ void ebpf_socket_read_open_connections(BUFFER *buf, struct ebpf_module *em)

634 *

635 * @param transaction the transaction id that Netdata sent for this function execution

636 * @param function function name and arguments given to thread.

647 - * @param line_buffer buffer used to parse args

648 - * @param line_max Number of arguments given

637 * @param timeout The function timeout

650 - * @param em The structure with thread information

638 + * @param cancelled Variable used to store function status.

639 */

640 static void ebpf_function_socket_manipulation(const char *transaction,

641 char *function __maybe_unused,

654 - char *line_buffer __maybe_unused,

655 - int line_max __maybe_unused,

642 int timeout __maybe_unused,

657 - ebpf_module_t *em)

643 + bool *cancelled __maybe_unused)

644 {

659 - UNUSED(line_buffer);

645 UNUSED(timeout);

646 + ebpf_module_t *em = &ebpf_modules[EBPF_MODULE_SOCKET_IDX];

647

648 char *words[PLUGINSD_MAX_WORDS] = {NULL};

649 size_t num_words = quoted_strings_splitter_pluginsd(function, words, PLUGINSD_MAX_WORDS);

@@ -1058,14 +1044,12 @@ static void ebpf_function_socket_manipulation(const char *transaction,

1044 buffer_json_finalize(wb);

1045

1046 // Lock necessary to avoid race condition

1061 - pthread_mutex_lock(&lock);

1047 pluginsd_function_result_begin_to_stdout(transaction, HTTP_RESP_OK, "application/json", expires);

1048

1049 fwrite(buffer_tostring(wb), buffer_strlen(wb), 1, stdout);

1050

1051 pluginsd_function_result_end_to_stdout();

1052 fflush(stdout);

1068 - pthread_mutex_unlock(&lock);

1053

1054 buffer_free(wb);

1055 }

@@ -1083,61 +1067,31 @@ static void ebpf_function_socket_manipulation(const char *transaction,

1067 */

1068 void *ebpf_function_thread(void *ptr)

1069 {

1086 - ebpf_module_t *em = (ebpf_module_t *)ptr;

1087 - char buffer[PLUGINSD_LINE_MAX + 1];

1088 -

1089 - rw_spinlock_init(&rw_spinlock);

1090 - char *s = NULL;

1091 - while(!ebpf_exit_plugin && (s = fgets(buffer, PLUGINSD_LINE_MAX, stdin))) {

1092 - char *words[PLUGINSD_MAX_WORDS] = { NULL };

1093 - size_t num_words = quoted_strings_splitter_pluginsd(buffer, words, PLUGINSD_MAX_WORDS);

1094 -

1095 - const char *keyword = get_word(words, num_words, 0);

1096 -

1097 - if(keyword && strcmp(keyword, PLUGINSD_KEYWORD_FUNCTION) == 0) {

1098 - char *transaction = get_word(words, num_words, 1);

1099 - char *timeout_s = get_word(words, num_words, 2);

1100 - char *function = get_word(words, num_words, 3);

1101 -

1102 - if(!transaction || !*transaction || !timeout_s || !*timeout_s || !function || !*function) {

1103 - netdata_log_error("Received incomplete %s (transaction = '%s', timeout = '%s', function = '%s'). Ignoring it.",

1104 - keyword,

1105 - transaction?transaction:"(unset)",

1106 - timeout_s?timeout_s:"(unset)",

1107 - function?function:"(unset)");

1108 - }

1109 - else {

1110 - int timeout = str2i(timeout_s);

1111 - rw_spinlock_write_lock(&rw_spinlock);

1112 - if (!strncmp(function, EBPF_FUNCTION_THREAD, sizeof(EBPF_FUNCTION_THREAD) - 1))

1113 - ebpf_function_thread_manipulation(transaction,

1114 - function,

1115 - buffer,

1116 - PLUGINSD_LINE_MAX + 1,

1117 - timeout,

1118 - em);

1119 - else if (!strncmp(function, EBPF_FUNCTION_SOCKET, sizeof(EBPF_FUNCTION_SOCKET) - 1))

1120 - ebpf_function_socket_manipulation(transaction,

1121 - function,

1122 - buffer,

1123 - PLUGINSD_LINE_MAX + 1,

1124 - timeout,

1125 - &ebpf_modules[EBPF_MODULE_SOCKET_IDX]);

1126 - else

1127 - ebpf_function_error(transaction,

1128 - HTTP_RESP_NOT_FOUND,

1129 - "No function with this name found in ebpf.plugin.");

1070 + (void)ptr;

1071

1131 - rw_spinlock_write_unlock(&rw_spinlock);

1132 - }

1072 + bool ebpf_function_plugin_exit = false;

1073 + struct functions_evloop_globals *wg = functions_evloop_init(1,

1074 + "EBPF",

1075 + &lock,

1076 + &ebpf_function_plugin_exit);

1077 +

1078 + functions_evloop_add_function(wg,

1079 + "ebpf_socket",

1080 + ebpf_function_socket_manipulation,

1081 + PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT);

1082 +

1083 + heartbeat_t hb;

1084 + heartbeat_init(&hb);

1085 + while(!ebpf_exit_plugin) {

1086 + (void)heartbeat_next(&hb, USEC_PER_SEC);

1087 +

1088 + if (ebpf_function_plugin_exit) {

1089 + pthread_mutex_lock(&ebpf_exit_cleanup);

1090 + ebpf_stop_threads(0);

1091 + pthread_mutex_unlock(&ebpf_exit_cleanup);

1092 + break;

1093 }

1134 - else

1135 - netdata_log_error("Received unknown command: %s", keyword ? keyword : "(unset)");

1094 }

1095

1138 - if(!s || feof(stdin) || ferror(stdin)) {

1139 - ebpf_stop_threads(SIGQUIT);

1140 - netdata_log_error("Received error on stdin.");

1141 - }

1096 return NULL;

1097 }