@cryptotaxi247 / netdata-1 / commits / 6393b2f53

Improve the impact of health code on netdata scalability (#8407)

* Add support for spawning processes without pipes. * Port health_alarm_execute() from mypopen() to netdata_spawn() * Make alarm notifications asynchronous within a single health thread iteration * Initial version of spawn server. * preliminary integration of spawn client with health

Markos Fountoulakis committed May 14, 2020 at 11:57 UTC 6393b2f535c993de9f341d2245ad8ba327694281

16 files changed +1216 -48

CMakeLists.txt

@@ -641,6 +641,13 @@ set(ACLK_PLUGIN_FILES

641 aclk/mqtt.h

642 )

643

644 +set(SPAWN_PLUGIN_FILES

645 + spawn/spawn.c

646 + spawn/spawn_server.c

647 + spawn/spawn_client.c

648 + spawn/spawn.h

649 + )

650 +

651 set(ACLK_STATIC_LIBS

652 ${CMAKE_SOURCE_DIR}/externaldeps/mosquitto/libmosquitto.a

653 ${CMAKE_SOURCE_DIR}/externaldeps/libwebsockets/libwebsockets.a

@@ -741,6 +748,7 @@ set(NETDATA_FILES

748 ${STREAMING_PLUGIN_FILES}

749 ${WEB_PLUGIN_FILES}

750 ${CLAIM_PLUGIN_FILES}

751 + ${SPAWN_PLUGIN_FILES}

752 )

753

754 set(NETDATACLI_FILES

Makefile.am

@@ -112,6 +112,7 @@ SUBDIRS += \

112 web \

113 claim \

114 aclk \

115 + spawn \

116 $(NULL)

117

118

@@ -497,6 +498,13 @@ endif

498

499

500

501 +SPAWN_PLUGIN_FILES = \

502 + spawn/spawn.c \

503 + spawn/spawn_server.c \

504 + spawn/spawn_client.c \

505 + spawn/spawn.h \

506 + $(NULL)

507 +

508 EXPORTING_ENGINE_FILES = \

509 exporting/exporting_engine.c \

510 exporting/exporting_engine.h \

@@ -595,6 +603,7 @@ NETDATA_FILES = \

603 $(WEB_PLUGIN_FILES) \

604 $(CLAIM_FILES) \

605 $(ACLK_FILES) \

606 + $(SPAWN_PLUGIN_FILES) \

607 $(NULL)

608

609 if FREEBSD

configure.ac

@@ -1490,6 +1490,7 @@ AC_CONFIG_FILES([

1490 web/server/static/Makefile

1491 claim/Makefile

1492 aclk/Makefile

1493 + spawn/Makefile

1494 ])

1495 AC_OUTPUT

1496

daemon/common.h

@@ -68,6 +68,9 @@

68 // netdata agent cloud link

69 #include "aclk/agent_cloud_link.h"

71 +// netdata agent spawn server

72 +#include "spawn/spawn.h"

73 +

74 // the netdata deamon

75 #include "daemon.h"

76 #include "main.h"

daemon/main.c

@@ -906,6 +906,11 @@ int main(int argc, char **argv) {

906 else i++;

907 }

908 }

909 + if (argc > 1 && strcmp(argv[1], SPAWN_SERVER_COMMAND_LINE_ARGUMENT) == 0) {

910 + // don't run netdata, this is the spawn server

911 + spawn_server();

912 + exit(0);

913 + }

914

915 // parse options

916 {

@@ -1377,6 +1382,9 @@ int main(int argc, char **argv) {

1382

1383 netdata_threads_init_after_fork((size_t)config_get_number(CONFIG_SECTION_GLOBAL, "pthread stack size", (long)default_stacksize));

1384

1385 + // fork the spawn server

1386 + spawn_init();

1387 +

1388 // ------------------------------------------------------------------------

1389 // initialize rrd, registry, health, rrdpush, etc.

1390

database/rrd.h

@@ -576,6 +576,7 @@ struct alarm_entry {

576 char *recipient;

577 time_t exec_run_timestamp;

578 int exec_code;

579 + uint64_t exec_spawn_serial;

580

581 char *source;

582 char *units;

@@ -601,6 +602,8 @@ struct alarm_entry {

602 time_t last_repeat;

603

604 struct alarm_entry *next;

605 + struct alarm_entry *next_in_progress;

606 + struct alarm_entry *prev_in_progress;

607 };

608

609

health/health.c

+72 -14

@@ -11,6 +11,46 @@ struct health_cmdapi_thread_status {

11 unsigned int default_health_enabled = 1;

12 char *silencers_filename;

14 +// the queue of executed alarm notifications that haven't been waited for yet

15 +static struct {

16 + ALARM_ENTRY *head; // oldest

17 + ALARM_ENTRY *tail; // latest

18 +} alarm_notifications_in_progress = {NULL, NULL};

19 +

20 +static inline void enqueue_alarm_notify_in_progress(ALARM_ENTRY *ae)

21 +{

22 + ae->prev_in_progress = NULL;

23 + ae->next_in_progress = NULL;

24 +

25 + if (NULL != alarm_notifications_in_progress.tail) {

26 + ae->prev_in_progress = alarm_notifications_in_progress.tail;

27 + alarm_notifications_in_progress.tail->next_in_progress = ae;

28 + }

29 + if (NULL == alarm_notifications_in_progress.head) {

30 + alarm_notifications_in_progress.head = ae;

31 + }

32 + alarm_notifications_in_progress.tail = ae;

33 +

34 +}

35 +

36 +static inline void unlink_alarm_notify_in_progress(ALARM_ENTRY *ae)

37 +{

38 + struct alarm_entry *prev = ae->prev_in_progress;

39 + struct alarm_entry *next = ae->next_in_progress;

40 +

41 + if (NULL != prev) {

42 + prev->next_in_progress = next;

43 + }

44 + if (NULL != next) {

45 + next->prev_in_progress = prev;

46 + }

47 + if (ae == alarm_notifications_in_progress.head) {

48 + alarm_notifications_in_progress.head = next;

49 + }

50 + if (ae == alarm_notifications_in_progress.tail) {

51 + alarm_notifications_in_progress.tail = prev;

52 + }

53 +}

54 // ----------------------------------------------------------------------------

55 // health initialization

@@ -265,7 +305,6 @@ static inline void health_alarm_execute(RRDHOST *host, ALARM_ENTRY *ae) {

305 }

306

307 static char command_to_run[ALARM_EXEC_COMMAND_LENGTH + 1];

268 - pid_t command_pid;

308

309 const char *exec = (ae->exec) ? ae->exec : host->health_default_exec;

310 const char *recipient = (ae->recipient) ? ae->recipient : host->health_default_recipient;

@@ -321,25 +360,30 @@ static inline void health_alarm_execute(RRDHOST *host, ALARM_ENTRY *ae) {

360 );

361

362 ae->flags |= HEALTH_ENTRY_FLAG_EXEC_RUN;

324 - ae->exec_run_timestamp = now_realtime_sec();

363 + ae->exec_run_timestamp = now_realtime_sec(); /* will be updated by real time after spawning */

364

365 debug(D_HEALTH, "executing command '%s'", command_to_run);

327 - FILE *fp = mypopen(command_to_run, &command_pid);

328 - if(!fp) {

329 - error("HEALTH: Cannot popen(\"%s\", \"r\").", command_to_run);

330 - goto done;

331 - }

332 - debug(D_HEALTH, "HEALTH reading from command (discarding command's output)");

333 - char buffer[100 + 1];

334 - while(fgets(buffer, 100, fp) != NULL) ;

335 - ae->exec_code = mypclose(fp, command_pid);

366 + ae->flags |= HEALTH_ENTRY_FLAG_EXEC_IN_PROGRESS;

367 + ae->exec_spawn_serial = spawn_enq_cmd(command_to_run);

368 + enqueue_alarm_notify_in_progress(ae);

369 +

370 + return; //health_alarm_wait_for_execution

371 +done:

372 + health_alarm_log_save(host, ae);

373 +}

374 +

375 +static inline void health_alarm_wait_for_execution(ALARM_ENTRY *ae) {

376 + if (!(ae->flags & HEALTH_ENTRY_FLAG_EXEC_IN_PROGRESS))

377 + return;

378 +

379 + spawn_wait_cmd(ae->exec_spawn_serial, &ae->exec_code, &ae->exec_run_timestamp);

380 debug(D_HEALTH, "done executing command - returned with code %d", ae->exec_code);

381 + ae->flags &= ~HEALTH_ENTRY_FLAG_EXEC_IN_PROGRESS;

382

383 if(ae->exec_code != 0)

384 ae->flags |= HEALTH_ENTRY_FLAG_EXEC_FAILED;

385

341 -done:

342 - health_alarm_log_save(host, ae);

386 + unlink_alarm_notify_in_progress(ae);

387 }

388

389 static inline void health_process_notifications(RRDHOST *host, ALARM_ENTRY *ae) {

@@ -401,6 +445,7 @@ static inline void health_alarm_log_process(RRDHOST *host) {

445 ALARM_ENTRY *t = ae->next;

446

447 if(likely(!alarm_entry_isrepeating(host, ae))) {

448 + health_alarm_wait_for_execution(ae);

449 health_alarm_log_free_one_nochecks_nounlink(ae);

450 host->health_log.count--;

451 }

@@ -945,6 +990,7 @@ void *health_main(void *ptr) {

990 rc->rrdcalc_flags |= RRDCALC_FLAG_RUN_ONCE;

991 health_process_notifications(host, ae);

992 debug(D_HEALTH, "Notification sent for the repeating alarm %u.", ae->alarm_id);

993 + health_alarm_wait_for_execution(ae);

994 health_alarm_log_free_one_nochecks_nounlink(ae);

995 }

996 }

@@ -959,11 +1005,23 @@ void *health_main(void *ptr) {

1005 // and cleanup

1006 health_alarm_log_process(host);

1007

962 - if (unlikely(netdata_exit))

1008 + if (unlikely(netdata_exit)) {

1009 + // wait for all notifications to finish before allowing health to be cleaned up

1010 + ALARM_ENTRY *ae;

1011 + while (NULL != (ae = alarm_notifications_in_progress.head)) {

1012 + health_alarm_wait_for_execution(ae);

1013 + }

1014 break;

1015 + }

1016

1017 } /* rrdhost_foreach */

1018

1019 + // wait for all notifications to finish before allowing health to be cleaned up

1020 + ALARM_ENTRY *ae;

1021 + while (NULL != (ae = alarm_notifications_in_progress.head)) {

1022 + health_alarm_wait_for_execution(ae);

1023 + }

1024 +

1025 rrd_unlock();

1026

1027

health/health.h

@@ -24,6 +24,7 @@ extern unsigned int default_health_enabled;

24 #define HEALTH_ENTRY_FLAG_EXEC_FAILED 0x00000008

25 #define HEALTH_ENTRY_FLAG_SILENCED 0x00000010

26 #define HEALTH_ENTRY_RUN_ONCE 0x00000020

27 +#define HEALTH_ENTRY_FLAG_EXEC_IN_PROGRESS 0x00000040

29 #define HEALTH_ENTRY_FLAG_SAVED 0x10000000

30 #define HEALTH_ENTRY_FLAG_NO_CLEAR_NOTIFICATION 0x80000000

libnetdata/popen/popen.c

+84 -34

@@ -78,8 +78,16 @@ static void myp_del(pid_t pid) {

78 #define PIPE_READ 0

79 #define PIPE_WRITE 1

81 -static inline FILE *custom_popene(const char *command, volatile pid_t *pidptr, char **env) {

82 - FILE *fp;

81 +/* custom_popene flag definitions */

82 +#define FLAG_CREATE_PIPE 1 // Create a pipe like popen() when set, otherwise set stdout to /dev/null

83 +#define FLAG_CLOSE_FD 2 // Close all file descriptors other than STDIN_FILENO, STDOUT_FILENO, STDERR_FILENO

84 +

85 +/*

86 + * Returns -1 on failure, 0 on success. When FLAG_CREATE_PIPE is set, on success set the FILE *fp pointer.

87 + */

88 +static inline int custom_popene(const char *command, volatile pid_t *pidptr, char **env, uint8_t flags, FILE **fpp) {

89 + FILE *fp = NULL;

90 + int ret = 0; // success by default

91 int pipefd[2], error;

92 pid_t pid;

93 char *const spawn_argv[] = {

@@ -91,23 +99,36 @@ static inline FILE *custom_popene(const char *command, volatile pid_t *pidptr, c

99 posix_spawnattr_t attr;

100 posix_spawn_file_actions_t fa;

101

94 - if (pipe(pipefd) == -1)

95 - return NULL;

96 - if ((fp = fdopen(pipefd[PIPE_READ], "r")) == NULL) {

97 - goto error_after_pipe;

102 + if (flags & FLAG_CREATE_PIPE) {

103 + if (pipe(pipefd) == -1)

104 + return -1;

105 + if ((fp = fdopen(pipefd[PIPE_READ], "r")) == NULL) {

106 + goto error_after_pipe;

107 + }

108 }

109

100 - // Mark all files to be closed by the exec() stage of posix_spawn()

101 - int i;

102 - for (i = (int) (sysconf(_SC_OPEN_MAX) - 1); i >= 0; i--)

103 - if(i != STDIN_FILENO && i != STDERR_FILENO)

104 - (void)fcntl(i, F_SETFD, FD_CLOEXEC);

110 + if (flags & FLAG_CLOSE_FD) {

111 + // Mark all files to be closed by the exec() stage of posix_spawn()

112 + int i;

113 + for (i = (int) (sysconf(_SC_OPEN_MAX) - 1); i >= 0; i--) {

114 + if (i != STDIN_FILENO && i != STDERR_FILENO)

115 + (void) fcntl(i, F_SETFD, FD_CLOEXEC);

116 + }

117 + }

118

119 if (!posix_spawn_file_actions_init(&fa)) {

107 - // move the pipe to stdout in the child

108 - if (posix_spawn_file_actions_adddup2(&fa, pipefd[PIPE_WRITE], STDOUT_FILENO)) {

109 - error("posix_spawn_file_actions_adddup2() failed");

110 - goto error_after_posix_spawn_file_actions_init;

120 + if (flags & FLAG_CREATE_PIPE) {

121 + // move the pipe to stdout in the child

122 + if (posix_spawn_file_actions_adddup2(&fa, pipefd[PIPE_WRITE], STDOUT_FILENO)) {

123 + error("posix_spawn_file_actions_adddup2() failed");

124 + goto error_after_posix_spawn_file_actions_init;

125 + }

126 + } else {

127 + // set stdout to /dev/null

128 + if (posix_spawn_file_actions_addopen(&fa, STDOUT_FILENO, "/dev/null", O_WRONLY, 0)) {

129 + error("posix_spawn_file_actions_addopen() failed");

130 + // this is not a fatal error

131 + }

132 }

133 } else {

134 error("posix_spawn_file_actions_init() failed.");

@@ -136,10 +157,16 @@ static inline FILE *custom_popene(const char *command, volatile pid_t *pidptr, c

157 } else {

158 myp_add_unlock();

159 error("Failed to spawn command: '%s' from parent pid %d.", command, getpid());

139 - fclose(fp);

140 - fp = NULL;

160 + if (flags & FLAG_CREATE_PIPE) {

161 + fclose(fp);

162 + }

163 + ret = -1;

164 + }

165 + if (flags & FLAG_CREATE_PIPE) {

166 + close(pipefd[PIPE_WRITE]);

167 + if (0 == ret) // on success set FILE * pointer

168 + *fpp = fp;

169 }

142 - close(pipefd[PIPE_WRITE]);

170

171 if (!error) {

172 // posix_spawnattr_init() succeeded

@@ -149,19 +176,21 @@ static inline FILE *custom_popene(const char *command, volatile pid_t *pidptr, c

176 if (posix_spawn_file_actions_destroy(&fa))

177 error("posix_spawn_file_actions_destroy");

178

152 - return fp;

179 + return ret;

180

181 error_after_posix_spawn_file_actions_init:

182 if (posix_spawn_file_actions_destroy(&fa))

183 error("posix_spawn_file_actions_destroy");

184 error_after_pipe:

158 - if (fp)

159 - fclose(fp);

160 - else

161 - close(pipefd[PIPE_READ]);

185 + if (flags & FLAG_CREATE_PIPE) {

186 + if (fp)

187 + fclose(fp);

188 + else

189 + close(pipefd[PIPE_READ]);

190

163 - close(pipefd[PIPE_WRITE]);

164 - return NULL;

191 + close(pipefd[PIPE_WRITE]);

192 + }

193 + return -1;

194 }

195

196 // See man environ

@@ -222,26 +251,37 @@ int myp_reap(pid_t pid) {

251 }

252

253 FILE *mypopen(const char *command, volatile pid_t *pidptr) {

225 - return custom_popene(command, pidptr, environ);

254 + FILE *fp = NULL;

255 + (void)custom_popene(command, pidptr, environ, FLAG_CREATE_PIPE | FLAG_CLOSE_FD, &fp);

256 + return fp;

257 }

258

259 FILE *mypopene(const char *command, volatile pid_t *pidptr, char **env) {

229 - return custom_popene(command, pidptr, env);

260 + FILE *fp = NULL;

261 + (void)custom_popene(command, pidptr, env, FLAG_CREATE_PIPE | FLAG_CLOSE_FD, &fp);

262 + return fp;

263 +}

264 +

265 +// returns 0 on success, -1 on failure

266 +int netdata_spawn(const char *command, volatile pid_t *pidptr) {

267 + return custom_popene(command, pidptr, environ, 0, NULL);

268 }

269

232 -int mypclose(FILE *fp, pid_t pid) {

270 +int custom_pclose(FILE *fp, pid_t pid) {

271 int ret;

272 siginfo_t info;

273

274 debug(D_EXIT, "Request to mypclose() on pid %d", pid);

275

238 - // close the pipe fd

239 - // this is required in musl

240 - // without it the childs do not exit

241 - close(fileno(fp));

276 + if (fp) {

277 + // close the pipe fd

278 + // this is required in musl

279 + // without it the childs do not exit

280 + close(fileno(fp));

281

243 - // close the pipe file pointer

244 - fclose(fp);

282 + // close the pipe file pointer

283 + fclose(fp);

284 + }

285

286 errno = 0;

287

@@ -285,3 +325,13 @@ int mypclose(FILE *fp, pid_t pid) {

325

326 return 0;

327 }

328 +

329 +int mypclose(FILE *fp, pid_t pid)

330 +{

331 + return custom_pclose(fp, pid);

332 +}

333 +

334 +int netdata_spawn_waitpid(pid_t pid)

335 +{

336 + return custom_pclose(NULL, pid);

337 +}

\ No newline at end of file

libnetdata/popen/popen.h

@@ -11,6 +11,8 @@

11 extern FILE *mypopen(const char *command, volatile pid_t *pidptr);

12 extern FILE *mypopene(const char *command, volatile pid_t *pidptr, char **env);

13 extern int mypclose(FILE *fp, pid_t pid);

14 +extern int netdata_spawn(const char *command, volatile pid_t *pidptr);

15 +extern int netdata_spawn_waitpid(pid_t pid);

16 extern void myp_init(void);

17 extern void myp_free(void);

18 extern int myp_reap(pid_t pid);

spawn/Makefile.am new

@@ -0,0 +1,9 @@

1 +# SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +AUTOMAKE_OPTIONS = subdir-objects

4 +MAINTAINERCLEANFILES = $(srcdir)/Makefile.in

5 +

6 +dist_noinst_DATA = \

7 + README.md \

8 + $(NULL)

9 +

spawn/README.md

spawn/spawn.c new

+289

@@ -0,0 +1,289 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "spawn.h"

4 +#include "../database/engine/rrdenginelib.h"

5 +

6 +static uv_thread_t thread;

7 +int spawn_thread_error;

8 +int spawn_thread_shutdown;

9 +

10 +struct spawn_queue spawn_cmd_queue;

11 +

12 +static struct spawn_cmd_info *create_spawn_cmd(char *command_to_run)

13 +{

14 + struct spawn_cmd_info *cmdinfo;

15 +

16 + cmdinfo = mallocz(sizeof(*cmdinfo));

17 + assert(0 == uv_cond_init(&cmdinfo->cond));

18 + assert(0 == uv_mutex_init(&cmdinfo->mutex));

19 + cmdinfo->serial = 0; /* invalid */

20 + cmdinfo->command_to_run = strdupz(command_to_run);

21 + cmdinfo->exit_status = -1; /* invalid */

22 + cmdinfo->pid = -1; /* invalid */

23 + cmdinfo->flags = 0;

24 +

25 + return cmdinfo;

26 +}

27 +

28 +void destroy_spawn_cmd(struct spawn_cmd_info *cmdinfo)

29 +{

30 + uv_cond_destroy(&cmdinfo->cond);

31 + uv_mutex_destroy(&cmdinfo->mutex);

32 +

33 + freez(cmdinfo->command_to_run);

34 + freez(cmdinfo);

35 +}

36 +

37 +int spawn_cmd_compare(void *a, void *b)

38 +{

39 + struct spawn_cmd_info *cmda = a, *cmdb = b;

40 +

41 + /* No need for mutex, serial will never change and the entries cannot be deallocated yet */

42 + if (cmda->serial < cmdb->serial) return -1;

43 + if (cmda->serial > cmdb->serial) return 1;

44 +

45 + return 0;

46 +}

47 +

48 +static void init_spawn_cmd_queue(void)

49 +{

50 + spawn_cmd_queue.cmd_tree.root = NULL;

51 + spawn_cmd_queue.cmd_tree.compar = spawn_cmd_compare;

52 + spawn_cmd_queue.size = 0;

53 + spawn_cmd_queue.latest_serial = 0;

54 + assert(0 == uv_cond_init(&spawn_cmd_queue.cond));

55 + assert(0 == uv_mutex_init(&spawn_cmd_queue.mutex));

56 +}

57 +

58 +/*

59 + * Returns serial number of the enqueued command

60 + */

61 +uint64_t spawn_enq_cmd(char *command_to_run)

62 +{

63 + unsigned queue_size;

64 + uint64_t serial;

65 + avl *avl_ret;

66 + struct spawn_cmd_info *cmdinfo;

67 +

68 + cmdinfo = create_spawn_cmd(command_to_run);

69 +

70 + /* wait for free space in queue */

71 + uv_mutex_lock(&spawn_cmd_queue.mutex);

72 + while ((queue_size = spawn_cmd_queue.size) == SPAWN_MAX_OUTSTANDING) {

73 + uv_cond_wait(&spawn_cmd_queue.cond, &spawn_cmd_queue.mutex);

74 + }

75 + assert(queue_size < SPAWN_MAX_OUTSTANDING);

76 + spawn_cmd_queue.size = queue_size + 1;

77 +

78 + serial = ++spawn_cmd_queue.latest_serial; /* 0 is invalid */

79 + cmdinfo->serial = serial; /* No need to take the cmd mutex since it is unreachable at the moment */

80 +

81 + /* enqueue command */

82 + avl_ret = avl_insert(&spawn_cmd_queue.cmd_tree, (avl *)cmdinfo);

83 + assert(avl_ret == (avl *)cmdinfo);

84 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

85 +

86 + /* wake up event loop */

87 + assert(0 == uv_async_send(&spawn_async));

88 + return serial;

89 +}

90 +

91 +/*

92 + * Blocks until command with serial finishes running. Only one thread is allowed to wait per command.

93 + */

94 +void spawn_wait_cmd(uint64_t serial, int *exit_status, time_t *exec_run_timestamp)

95 +{

96 + avl *avl_ret;

97 + struct spawn_cmd_info tmp, *cmdinfo;

98 +

99 + tmp.serial = serial;

100 +

101 + uv_mutex_lock(&spawn_cmd_queue.mutex);

102 + avl_ret = avl_search(&spawn_cmd_queue.cmd_tree, (avl *)&tmp);

103 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

104 +

105 + assert(avl_ret); /* Could be NULL if more than 1 threads wait for the command */

106 + cmdinfo = (struct spawn_cmd_info *)avl_ret;

107 +

108 + uv_mutex_lock(&cmdinfo->mutex);

109 + while (!(cmdinfo->flags & SPAWN_CMD_DONE)) {

110 + /* Only 1 thread is allowed to wait for this command to finish */

111 + uv_cond_wait(&cmdinfo->cond, &cmdinfo->mutex);

112 + }

113 + uv_mutex_unlock(&cmdinfo->mutex);

114 +

115 + spawn_deq_cmd(cmdinfo);

116 + *exit_status = cmdinfo->exit_status;

117 + *exec_run_timestamp = cmdinfo->exec_run_timestamp;

118 +

119 + destroy_spawn_cmd(cmdinfo);

120 +}

121 +

122 +void spawn_deq_cmd(struct spawn_cmd_info *cmdinfo)

123 +{

124 + unsigned queue_size;

125 + avl *avl_ret;

126 +

127 + uv_mutex_lock(&spawn_cmd_queue.mutex);

128 + queue_size = spawn_cmd_queue.size;

129 + assert(queue_size);

130 + /* dequeue command */

131 + avl_ret = avl_remove(&spawn_cmd_queue.cmd_tree, (avl *)cmdinfo);

132 + assert(avl_ret);

133 +

134 + spawn_cmd_queue.size = queue_size - 1;

135 +

136 + /* wake up callers */

137 + uv_cond_signal(&spawn_cmd_queue.cond);

138 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

139 +}

140 +

141 +/*

142 + * Must be called from the spawn client event loop context. This way no mutex is needed because the event loop is the

143 + * only writer as far as struct spawn_cmd_info entries are concerned.

144 + */

145 +static int find_unprocessed_spawn_cmd_cb(void *entry, void *data)

146 +{

147 + struct spawn_cmd_info **cmdinfop = data, *cmdinfo = entry;

148 +

149 + if (!(cmdinfo->flags & SPAWN_CMD_PROCESSED)) {

150 + *cmdinfop = cmdinfo;

151 + return -1; /* break tree traversal */

152 + }

153 + return 0; /* continue traversing */

154 +}

155 +

156 +struct spawn_cmd_info *spawn_get_unprocessed_cmd(void)

157 +{

158 + struct spawn_cmd_info *cmdinfo;

159 + unsigned queue_size;

160 + int ret;

161 +

162 + uv_mutex_lock(&spawn_cmd_queue.mutex);

163 + queue_size = spawn_cmd_queue.size;

164 + if (queue_size == 0) {

165 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

166 + return NULL;

167 + }

168 + /* find command */

169 + cmdinfo = NULL;

170 + ret = avl_traverse(&spawn_cmd_queue.cmd_tree, find_unprocessed_spawn_cmd_cb, (void *)&cmdinfo);

171 + if (-1 != ret) { /* no commands available for processing */

172 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

173 + return NULL;

174 + }

175 + uv_mutex_unlock(&spawn_cmd_queue.mutex);

176 +

177 + return cmdinfo;

178 +}

179 +

180 +/**

181 + * This function spawns a process that shares a libuv IPC pipe with the caller and performs spawn server duties.

182 + * The spawn server process will close all open file descriptors except for the pipe, UV_STDOUT_FD, and UV_STDERR_FD.

183 + * The caller has to be the netdata user as configured.

184 + *

185 + * @param loop the libuv loop of the caller context

186 + * @param spawn_channel the birectional libuv IPC pipe that the server and the caller will share

187 + * @param process the spawn server libuv process context

188 + * @return 0 on success or the libuv error code

189 + */

190 +int create_spawn_server(uv_loop_t *loop, uv_pipe_t *spawn_channel, uv_process_t *process)

191 +{

192 + uv_process_options_t options = {0};

193 + size_t exepath_size;

194 + char exepath[FILENAME_MAX];

195 + char *args[3];

196 + int ret;

197 +#define SPAWN_SERVER_DESCRIPTORS (3)

198 + uv_stdio_container_t stdio[SPAWN_SERVER_DESCRIPTORS];

199 +

200 + exepath_size = sizeof(exepath);

201 + ret = uv_exepath(exepath, &exepath_size);

202 + assert(ret == 0);

203 +

204 + exepath[exepath_size] = '\0';

205 + args[0] = exepath;

206 + args[1] = SPAWN_SERVER_COMMAND_LINE_ARGUMENT;

207 + args[2] = NULL;

208 +

209 + memset(&options, 0, sizeof(options));

210 + options.file = exepath;

211 + options.args = args;

212 + options.exit_cb = NULL; //exit_cb;

213 + options.stdio = stdio;

214 + options.stdio_count = SPAWN_SERVER_DESCRIPTORS;

215 +

216 + stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE | UV_WRITABLE_PIPE;

217 + stdio[0].data.stream = (uv_stream_t *)spawn_channel; /* bidirectional libuv pipe */

218 + stdio[1].flags = UV_INHERIT_FD;

219 + stdio[1].data.fd = 1 /* UV_STDOUT_FD */;

220 + stdio[2].flags = UV_INHERIT_FD;

221 + stdio[2].data.fd = 2 /* UV_STDERR_FD */;

222 +

223 + ret = uv_spawn(loop, process, &options); /* execute the netdata binary again as the netdata user */

224 + assert(ret == 0);

225 +

226 + return ret;

227 +}

228 +

229 +#define CONCURRENT_SPAWNS 16

230 +#define SPAWN_ITERATIONS 10000

231 +#undef CONCURRENT_STRESS_TEST

232 +

233 +void spawn_init(void)

234 +{

235 + struct completion completion;

236 + int error;

237 +

238 + info("Initializing spawn client.");

239 +

240 + init_spawn_cmd_queue();

241 +

242 + init_completion(&completion);

243 + error = uv_thread_create(&thread, spawn_client, &completion);

244 + if (error) {

245 + error("uv_thread_create(): %s", uv_strerror(error));

246 + goto after_error;

247 + }

248 + /* wait for spawn client thread to initialize */

249 + wait_for_completion(&completion);

250 + destroy_completion(&completion);

251 + uv_thread_set_name_np(thread, "DAEMON_SPAWN");

252 +

253 + if (spawn_thread_error) {

254 + error = uv_thread_join(&thread);

255 + if (error) {

256 + error("uv_thread_create(): %s", uv_strerror(error));

257 + }

258 + goto after_error;

259 + }

260 +#ifdef CONCURRENT_STRESS_TEST

261 + signals_reset();

262 + signals_unblock();

263 +

264 + sleep(60);

265 + uint64_t serial[CONCURRENT_SPAWNS];

266 + for (int j = 0 ; j < SPAWN_ITERATIONS ; ++j) {

267 + for (int i = 0; i < CONCURRENT_SPAWNS; ++i) {

268 + char cmd[64];

269 + sprintf(cmd, "echo CONCURRENT_STRESS_TEST %d 1>&2", j * CONCURRENT_SPAWNS + i + 1);

270 + serial[i] = spawn_enq_cmd(cmd);

271 + info("Queued command %s for spawning.", cmd);

272 + }

273 + int exit_status;

274 + time_t exec_run_timestamp;

275 + for (int i = 0; i < CONCURRENT_SPAWNS; ++i) {

276 + info("Started waiting for serial %llu exit status %d run timestamp %llu.", serial[i], exit_status,

277 + exec_run_timestamp);

278 + spawn_wait_cmd(serial[i], &exit_status, &exec_run_timestamp);

279 + info("Finished waiting for serial %llu exit status %d run timestamp %llu.", serial[i], exit_status,

280 + exec_run_timestamp);

281 + }

282 + }

283 + exit(0);

284 +#endif

285 + return;

286 +

287 + after_error:

288 + error("Failed to initialize spawn service. The alarms notifications will not be spawned.");

289 +}

spawn/spawn.h new

+109

@@ -0,0 +1,109 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#ifndef NETDATA_SPAWN_H

4 +#define NETDATA_SPAWN_H 1

5 +

6 +#include "../daemon/common.h"

7 +

8 +#define SPAWN_SERVER_COMMAND_LINE_ARGUMENT "--special-spawn-server"

9 +

10 +typedef enum spawn_protocol {

11 + SPAWN_PROT_EXEC_CMD = 0,

12 + SPAWN_PROT_SPAWN_RESULT,

13 + SPAWN_PROT_CMD_EXIT_STATUS

14 +} spawn_prot_t;

15 +

16 +struct spawn_prot_exec_cmd {

17 + uint16_t command_length;

18 + char command_to_run[];

19 +};

20 +

21 +struct spawn_prot_spawn_result {

22 + pid_t exec_pid; /* 0 if failed to spawn */

23 + time_t exec_run_timestamp; /* time of successfully spawning the command */

24 +};

25 +

26 +struct spawn_prot_cmd_exit_status {

27 + int exec_exit_status;

28 +};

29 +

30 +struct spawn_prot_header {

31 + spawn_prot_t opcode;

32 + void *handle;

33 +};

34 +

35 +#undef SPAWN_DEBUG /* define to enable debug prints */

36 +

37 +#define SPAWN_MAX_OUTSTANDING (32768)

38 +

39 +#define SPAWN_CMD_PROCESSED 0x00000001

40 +#define SPAWN_CMD_IN_PROGRESS 0x00000002

41 +#define SPAWN_CMD_FAILED_TO_SPAWN 0x00000004

42 +#define SPAWN_CMD_DONE 0x00000008

43 +

44 +struct spawn_cmd_info {

45 + avl avl;

46 +

47 + /* concurrency control per command */

48 + uv_mutex_t mutex;

49 + uv_cond_t cond; /* users block here until command has finished */

50 +

51 + uint64_t serial;

52 + char *command_to_run;

53 + int exit_status;

54 + pid_t pid;

55 + unsigned long flags;

56 + time_t exec_run_timestamp; /* time of successfully spawning the command */

57 +};

58 +

59 +/* spawn command queue */

60 +struct spawn_queue {

61 + avl_tree cmd_tree;

62 +

63 + /* concurrency control of command queue */

64 + uv_mutex_t mutex;

65 + uv_cond_t cond;

66 +

67 + volatile unsigned size;

68 + uint64_t latest_serial;

69 +};

70 +

71 +struct write_context {

72 + uv_write_t write_req;

73 + struct spawn_prot_header header;

74 + struct spawn_prot_cmd_exit_status exit_status;

75 + struct spawn_prot_spawn_result spawn_result;

76 + struct spawn_prot_exec_cmd payload;

77 +};

78 +

79 +extern int spawn_thread_error;

80 +extern int spawn_thread_shutdown;

81 +extern uv_async_t spawn_async;

82 +

83 +void spawn_init(void);

84 +void spawn_server(void);

85 +void spawn_client(void *arg);

86 +void destroy_spawn_cmd(struct spawn_cmd_info *cmdinfo);

87 +uint64_t spawn_enq_cmd(char *command_to_run);

88 +void spawn_wait_cmd(uint64_t serial, int *exit_status, time_t *exec_run_timestamp);

89 +void spawn_deq_cmd(struct spawn_cmd_info *cmdinfo);

90 +struct spawn_cmd_info *spawn_get_unprocessed_cmd(void);

91 +int create_spawn_server(uv_loop_t *loop, uv_pipe_t *spawn_channel, uv_process_t *process);

92 +

93 +/*

94 + * Copies from the source buffer to the protocol buffer. It advances the source buffer by the amount copied. It

95 + * subtracts the amount copied from the source length.

96 + */

97 +static inline void copy_to_prot_buffer(char *prot_buffer, unsigned *prot_buffer_len, unsigned max_to_copy,

98 + char **source, unsigned *source_len)

99 +{

100 + unsigned to_copy;

101 +

102 + to_copy = MIN(max_to_copy, *source_len);

103 + memcpy(prot_buffer + *prot_buffer_len, *source, to_copy);

104 + *prot_buffer_len += to_copy;

105 + *source += to_copy;

106 + *source_len -= to_copy;

107 +}

108 +

109 +#endif //NETDATA_SPAWN_H

spawn/spawn_client.c new

+241

@@ -0,0 +1,241 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "spawn.h"

4 +#include "../database/engine/rrdenginelib.h"

5 +

6 +static uv_process_t process;

7 +static uv_pipe_t spawn_channel;

8 +static uv_loop_t *loop;

9 +uv_async_t spawn_async;

10 +

11 +static char prot_buffer[MAX_COMMAND_LENGTH];

12 +static unsigned prot_buffer_len = 0;

13 +

14 +static void async_cb(uv_async_t *handle)

15 +{

16 + uv_stop(handle->loop);

17 +}

18 +

19 +static void after_pipe_write(uv_write_t* req, int status)

20 +{

21 + (void)status;

22 +#ifdef SPAWN_DEBUG

23 + info("CLIENT %s called status=%d", __func__, status);

24 +#endif

25 + freez(req->data);

26 +}

27 +

28 +static void client_parse_spawn_protocol(unsigned source_len, char *source)

29 +{

30 + unsigned required_len;

31 + struct spawn_prot_header *header;

32 + struct spawn_prot_spawn_result *spawn_result;

33 + struct spawn_prot_cmd_exit_status *exit_status;

34 + struct spawn_cmd_info *cmdinfo;

35 +

36 + while (source_len) {

37 + required_len = sizeof(*header);

38 + if (prot_buffer_len < required_len)

39 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

40 + if (prot_buffer_len < required_len)

41 + return; /* Source buffer ran out */

42 +

43 + header = (struct spawn_prot_header *)prot_buffer;

44 + cmdinfo = (struct spawn_cmd_info *)header->handle;

45 + assert(NULL != cmdinfo);

46 +

47 + switch(header->opcode) {

48 + case SPAWN_PROT_SPAWN_RESULT:

49 + required_len += sizeof(*spawn_result);

50 + if (prot_buffer_len < required_len)

51 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

52 + if (prot_buffer_len < required_len)

53 + return; /* Source buffer ran out */

54 +

55 + spawn_result = (struct spawn_prot_spawn_result *)(header + 1);

56 + uv_mutex_lock(&cmdinfo->mutex);

57 + cmdinfo->pid = spawn_result->exec_pid;

58 + if (0 == cmdinfo->pid) { /* Failed to spawn */

59 +#ifdef SPAWN_DEBUG

60 + info("CLIENT %s SPAWN_PROT_SPAWN_RESULT failed to spawn.", __func__);

61 +#endif

62 + cmdinfo->flags |= SPAWN_CMD_FAILED_TO_SPAWN | SPAWN_CMD_DONE;

63 + uv_cond_signal(&cmdinfo->cond);

64 + } else {

65 + cmdinfo->exec_run_timestamp = spawn_result->exec_run_timestamp;

66 + cmdinfo->flags |= SPAWN_CMD_IN_PROGRESS;

67 +#ifdef SPAWN_DEBUG

68 + info("CLIENT %s SPAWN_PROT_SPAWN_RESULT in progress.", __func__);

69 +#endif

70 + }

71 + uv_mutex_unlock(&cmdinfo->mutex);

72 + prot_buffer_len = 0;

73 + break;

74 + case SPAWN_PROT_CMD_EXIT_STATUS:

75 + required_len += sizeof(*exit_status);

76 + if (prot_buffer_len < required_len)

77 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

78 + if (prot_buffer_len < required_len)

79 + return; /* Source buffer ran out */

80 +

81 + exit_status = (struct spawn_prot_cmd_exit_status *)(header + 1);

82 + uv_mutex_lock(&cmdinfo->mutex);

83 + cmdinfo->exit_status = exit_status->exec_exit_status;

84 +#ifdef SPAWN_DEBUG

85 + info("CLIENT %s SPAWN_PROT_CMD_EXIT_STATUS %d.", __func__, exit_status->exec_exit_status);

86 +#endif

87 + cmdinfo->flags |= SPAWN_CMD_DONE;

88 + uv_cond_signal(&cmdinfo->cond);

89 + uv_mutex_unlock(&cmdinfo->mutex);

90 + prot_buffer_len = 0;

91 + break;

92 + default:

93 + assert(0);

94 + break;

95 + }

96 +

97 + }

98 +}

99 +

100 +static void on_pipe_read(uv_stream_t* pipe, ssize_t nread, const uv_buf_t* buf)

101 +{

102 + if (0 == nread) {

103 + info("%s: Zero bytes read from spawn pipe.", __func__);

104 + } else if (UV_EOF == nread) {

105 + info("EOF found in spawn pipe.");

106 + } else if (nread < 0) {

107 + error("%s: %s", __func__, uv_strerror(nread));

108 + }

109 +

110 + if (nread < 0) { /* stop stream due to EOF or error */

111 + (void)uv_read_stop((uv_stream_t *)pipe);

112 + } else if (nread) {

113 +#ifdef SPAWN_DEBUG

114 + info("CLIENT %s read %u", __func__, (unsigned)nread);

115 +#endif

116 + client_parse_spawn_protocol(nread, buf->base);

117 + }

118 + if (buf && buf->len) {

119 + freez(buf->base);

120 + }

121 +

122 + if (nread < 0) {

123 + uv_close((uv_handle_t *)pipe, NULL);

124 + }

125 +}

126 +

127 +static void on_read_alloc(uv_handle_t* handle,

128 + size_t suggested_size,

129 + uv_buf_t* buf)

130 +{

131 + (void)handle;

132 + buf->base = mallocz(suggested_size);

133 + buf->len = suggested_size;

134 +}

135 +

136 +static void spawn_process_cmd(struct spawn_cmd_info *cmdinfo)

137 +{

138 + int ret;

139 + uv_buf_t writebuf[3];

140 + struct write_context *write_ctx;

141 +

142 + write_ctx = mallocz(sizeof(*write_ctx));

143 + write_ctx->write_req.data = write_ctx;

144 +

145 + uv_mutex_lock(&cmdinfo->mutex);

146 + cmdinfo->flags |= SPAWN_CMD_PROCESSED;

147 + uv_mutex_unlock(&cmdinfo->mutex);

148 +

149 + write_ctx->header.opcode = SPAWN_PROT_EXEC_CMD;

150 + write_ctx->header.handle = cmdinfo;

151 + write_ctx->payload.command_length = strlen(cmdinfo->command_to_run);

152 +

153 + writebuf[0] = uv_buf_init((char *)&write_ctx->header, sizeof(write_ctx->header));

154 + writebuf[1] = uv_buf_init((char *)&write_ctx->payload, sizeof(write_ctx->payload));

155 + writebuf[2] = uv_buf_init((char *)cmdinfo->command_to_run, write_ctx->payload.command_length);

156 +

157 +#ifdef SPAWN_DEBUG

158 + info("CLIENT %s SPAWN_PROT_EXEC_CMD %u", __func__, (unsigned)cmdinfo->serial);

159 +#endif

160 + ret = uv_write(&write_ctx->write_req, (uv_stream_t *)&spawn_channel, writebuf, 3, after_pipe_write);

161 + assert(ret == 0);

162 +}

163 +

164 +void spawn_client(void *arg)

165 +{

166 + int ret;

167 + struct completion *completion = (struct completion *)arg;

168 +

169 + loop = mallocz(sizeof(uv_loop_t));

170 + ret = uv_loop_init(loop);

171 + if (ret) {

172 + error("uv_loop_init(): %s", uv_strerror(ret));

173 + spawn_thread_error = ret;

174 + goto error_after_loop_init;

175 + }

176 + loop->data = NULL;

177 +

178 + spawn_async.data = NULL;

179 + ret = uv_async_init(loop, &spawn_async, async_cb);

180 + if (ret) {

181 + error("uv_async_init(): %s", uv_strerror(ret));

182 + spawn_thread_error = ret;

183 + goto error_after_async_init;

184 + }

185 +

186 + ret = uv_pipe_init(loop, &spawn_channel, 1);

187 + if (ret) {

188 + error("uv_pipe_init(): %s", uv_strerror(ret));

189 + spawn_thread_error = ret;

190 + goto error_after_pipe_init;

191 + }

192 + assert(spawn_channel.ipc);

193 +

194 + ret = create_spawn_server(loop, &spawn_channel, &process);

195 + if (ret) {

196 + error("Failed to fork spawn server process.");

197 + spawn_thread_error = ret;

198 + goto error_after_spawn_server;

199 + }

200 +

201 + spawn_thread_error = 0;

202 + spawn_thread_shutdown = 0;

203 + /* wake up initialization thread */

204 + complete(completion);

205 +

206 + prot_buffer_len = 0;

207 + ret = uv_read_start((uv_stream_t *)&spawn_channel, on_read_alloc, on_pipe_read);

208 + assert(ret == 0);

209 +

210 + while (spawn_thread_shutdown == 0) {

211 + struct spawn_cmd_info *cmdinfo;

212 +

213 + uv_run(loop, UV_RUN_DEFAULT);

214 + while (NULL != (cmdinfo = spawn_get_unprocessed_cmd())) {

215 + spawn_process_cmd(cmdinfo);

216 + }

217 + }

218 + /* cleanup operations of the event loop */

219 + info("Shutting down spawn client event loop.");

220 + uv_close((uv_handle_t *)&spawn_channel, NULL);

221 + uv_close((uv_handle_t *)&spawn_async, NULL);

222 + uv_run(loop, UV_RUN_DEFAULT); /* flush all libuv handles */

223 +

224 + info("Shutting down spawn client loop complete.");

225 + assert(0 == uv_loop_close(loop));

226 +

227 + return;

228 +

229 +error_after_spawn_server:

230 + uv_close((uv_handle_t *)&spawn_channel, NULL);

231 +error_after_pipe_init:

232 + uv_close((uv_handle_t *)&spawn_async, NULL);

233 +error_after_async_init:

234 + uv_run(loop, UV_RUN_DEFAULT); /* flush all libuv handles */

235 + assert(0 == uv_loop_close(loop));

236 +error_after_loop_init:

237 + freez(loop);

238 +

239 + /* wake up initialization thread */

240 + complete(completion);

241 +}

spawn/spawn_server.c new

+377

@@ -0,0 +1,377 @@

1 +// SPDX-License-Identifier: GPL-3.0-or-later

2 +

3 +#include "spawn.h"

4 +

5 +static uv_loop_t *loop;

6 +static uv_pipe_t server_pipe;

7 +

8 +static int server_shutdown = 0;

9 +

10 +static uv_thread_t thread;

11 +

12 +/* spawn outstanding execution structure */

13 +static avl_tree_lock spawn_outstanding_exec_tree;

14 +

15 +static char prot_buffer[MAX_COMMAND_LENGTH];

16 +static unsigned prot_buffer_len = 0;

17 +

18 +struct spawn_execution_info {

19 + avl avl;

20 +

21 + void *handle;

22 + int exit_status;

23 + pid_t pid;

24 + struct spawn_execution_info *next;

25 +};

26 +

27 +int spawn_exec_compare(void *a, void *b)

28 +{

29 + struct spawn_execution_info *spwna = a, *spwnb = b;

30 +

31 + if (spwna->pid < spwnb->pid) return -1;

32 + if (spwna->pid > spwnb->pid) return 1;

33 +

34 + return 0;

35 +}

36 +

37 +/* wake up waiter thread to reap the spawned processes */

38 +static uv_mutex_t wait_children_mutex;

39 +static uv_cond_t wait_children_cond;

40 +static uint8_t spawned_processes;

41 +static struct spawn_execution_info *child_waited_list;

42 +static uv_async_t child_waited_async;

43 +

44 +static inline struct spawn_execution_info *dequeue_child_waited_list(void)

45 +{

46 + struct spawn_execution_info *exec_info;

47 +

48 + uv_mutex_lock(&wait_children_mutex);

49 + if (NULL == child_waited_list) {

50 + exec_info = NULL;

51 + } else {

52 + exec_info = child_waited_list;

53 + child_waited_list = exec_info->next;

54 + }

55 + uv_mutex_unlock(&wait_children_mutex);

56 +

57 + return exec_info;

58 +}

59 +

60 +static inline void enqueue_child_waited_list(struct spawn_execution_info *exec_info)

61 +{

62 + uv_mutex_lock(&wait_children_mutex);

63 + exec_info->next = child_waited_list;

64 + child_waited_list = exec_info;

65 + uv_mutex_unlock(&wait_children_mutex);

66 +}

67 +

68 +static void after_pipe_write(uv_write_t *req, int status)

69 +{

70 + (void)status;

71 +#ifdef SPAWN_DEBUG

72 + fprintf(stderr, "SERVER %s called status=%d\n", __func__, status);

73 +#endif

74 + freez(req->data);

75 +}

76 +

77 +static void child_waited_async_cb(uv_async_t *async_handle)

78 +{

79 + uv_buf_t writebuf[2];

80 + int ret;

81 + struct spawn_execution_info *exec_info;

82 + struct write_context *write_ctx;

83 +

84 + (void)async_handle;

85 + while (NULL != (exec_info = dequeue_child_waited_list())) {

86 + write_ctx = mallocz(sizeof(*write_ctx));

87 + write_ctx->write_req.data = write_ctx;

88 +

89 +

90 + write_ctx->header.opcode = SPAWN_PROT_CMD_EXIT_STATUS;

91 + write_ctx->header.handle = exec_info->handle;

92 + write_ctx->exit_status.exec_exit_status = exec_info->exit_status;

93 + writebuf[0] = uv_buf_init((char *) &write_ctx->header, sizeof(write_ctx->header));

94 + writebuf[1] = uv_buf_init((char *) &write_ctx->exit_status, sizeof(write_ctx->exit_status));

95 +#ifdef SPAWN_DEBUG

96 + fprintf(stderr, "SERVER %s SPAWN_PROT_CMD_EXIT_STATUS\n", __func__);

97 +#endif

98 + ret = uv_write(&write_ctx->write_req, (uv_stream_t *) &server_pipe, writebuf, 2, after_pipe_write);

99 + assert(ret == 0);

100 +

101 + freez(exec_info);

102 + }

103 +}

104 +

105 +static void wait_children(void *arg)

106 +{

107 + siginfo_t i;

108 + struct spawn_execution_info tmp, *exec_info;

109 + avl *ret_avl;

110 +

111 + (void)arg;

112 + while (!server_shutdown) {

113 + uv_mutex_lock(&wait_children_mutex);

114 + while (!spawned_processes) {

115 + uv_cond_wait(&wait_children_cond, &wait_children_mutex);

116 + }

117 + spawned_processes = 0;

118 + uv_mutex_unlock(&wait_children_mutex);

119 +

120 + while (!server_shutdown) {

121 + i.si_pid = 0;

122 + if (waitid(P_ALL, (id_t) 0, &i, WEXITED) == -1) {

123 + if (errno != ECHILD)

124 + fprintf(stderr, "SPAWN: Failed to wait: %s\n", strerror(errno));

125 + break;

126 + }

127 + if (i.si_pid == 0) {

128 + fprintf(stderr, "SPAWN: No child exited.\n");

129 + break;

130 + }

131 +#ifdef SPAWN_DEBUG

132 + fprintf(stderr, "SPAWN: Successfully waited for pid:%d.\n", (int) i.si_pid);

133 +#endif

134 + assert(CLD_EXITED == i.si_code);

135 + tmp.pid = (pid_t)i.si_pid;

136 + while (NULL == (ret_avl = avl_remove_lock(&spawn_outstanding_exec_tree, (avl *)&tmp))) {

137 + fprintf(stderr,

138 + "SPAWN: race condition detected, waiting for child process %d to be indexed.\n",

139 + (int)tmp.pid);

140 + (void)sleep_usec(10000); /* 10 msec */

141 + }

142 + exec_info = (struct spawn_execution_info *)ret_avl;

143 + exec_info->exit_status = i.si_status;

144 + enqueue_child_waited_list(exec_info);

145 +

146 + /* wake up event loop */

147 + assert(0 == uv_async_send(&child_waited_async));

148 + }

149 + }

150 +}

151 +

152 +void spawn_protocol_execute_command(void *handle, char *command_to_run, uint16_t command_length)

153 +{

154 + uv_buf_t writebuf[2];

155 + int ret;

156 + avl *avl_ret;

157 + struct spawn_execution_info *exec_info;

158 + struct write_context *write_ctx;

159 +

160 + write_ctx = mallocz(sizeof(*write_ctx));

161 + write_ctx->write_req.data = write_ctx;

162 +

163 + command_to_run[command_length] = '\0';

164 +#ifdef SPAWN_DEBUG

165 + fprintf(stderr, "SPAWN: executing command '%s'\n", command_to_run);

166 +#endif

167 + if (netdata_spawn(command_to_run, &write_ctx->spawn_result.exec_pid)) {

168 + fprintf(stderr, "SPAWN: Cannot spawn(\"%s\", \"r\").\n", command_to_run);

169 + write_ctx->spawn_result.exec_pid = 0;

170 + } else { /* successfully spawned command */

171 + write_ctx->spawn_result.exec_run_timestamp = now_realtime_sec();

172 +

173 + /* record it for when the process finishes execution */

174 + exec_info = mallocz(sizeof(*exec_info));

175 + exec_info->handle = handle;

176 + exec_info->pid = write_ctx->spawn_result.exec_pid;

177 + avl_ret = avl_insert_lock(&spawn_outstanding_exec_tree, (avl *)exec_info);

178 + assert(avl_ret == (avl *)exec_info);

179 +

180 + /* wake up the thread that blocks waiting for processes to exit */

181 + uv_mutex_lock(&wait_children_mutex);

182 + spawned_processes = 1;

183 + uv_cond_signal(&wait_children_cond);

184 + uv_mutex_unlock(&wait_children_mutex);

185 + }

186 +

187 + write_ctx->header.opcode = SPAWN_PROT_SPAWN_RESULT;

188 + write_ctx->header.handle = handle;

189 + writebuf[0] = uv_buf_init((char *)&write_ctx->header, sizeof(write_ctx->header));

190 + writebuf[1] = uv_buf_init((char *)&write_ctx->spawn_result, sizeof(write_ctx->spawn_result));

191 +#ifdef SPAWN_DEBUG

192 + fprintf(stderr, "SERVER %s SPAWN_PROT_SPAWN_RESULT\n", __func__);

193 +#endif

194 + ret = uv_write(&write_ctx->write_req, (uv_stream_t *)&server_pipe, writebuf, 2, after_pipe_write);

195 + assert(ret == 0);

196 +}

197 +

198 +static void server_parse_spawn_protocol(unsigned source_len, char *source)

199 +{

200 + unsigned required_len;

201 + struct spawn_prot_header *header;

202 + struct spawn_prot_exec_cmd *payload;

203 + uint16_t command_length;

204 +

205 + while (source_len) {

206 + required_len = sizeof(*header);

207 + if (prot_buffer_len < required_len)

208 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

209 + if (prot_buffer_len < required_len)

210 + return; /* Source buffer ran out */

211 +

212 + header = (struct spawn_prot_header *)prot_buffer;

213 + assert(SPAWN_PROT_EXEC_CMD == header->opcode);

214 + assert(NULL != header->handle);

215 +

216 + required_len += sizeof(*payload);

217 + if (prot_buffer_len < required_len)

218 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

219 + if (prot_buffer_len < required_len)

220 + return; /* Source buffer ran out */

221 +

222 + payload = (struct spawn_prot_exec_cmd *)(header + 1);

223 + command_length = payload->command_length;

224 +

225 + required_len += command_length;

226 + if (unlikely(required_len > MAX_COMMAND_LENGTH - 1)) {

227 + fprintf(stderr, "SPAWN: Ran out of protocol buffer space.\n");

228 + command_length = (MAX_COMMAND_LENGTH - 1) - (sizeof(*header) + sizeof(*payload));

229 + required_len = MAX_COMMAND_LENGTH - 1;

230 + }

231 + if (prot_buffer_len < required_len)

232 + copy_to_prot_buffer(prot_buffer, &prot_buffer_len, required_len - prot_buffer_len, &source, &source_len);

233 + if (prot_buffer_len < required_len)

234 + return; /* Source buffer ran out */

235 +

236 + spawn_protocol_execute_command(header->handle, payload->command_to_run, command_length);

237 + prot_buffer_len = 0;

238 + }

239 +}

240 +

241 +static void on_pipe_read(uv_stream_t *pipe, ssize_t nread, const uv_buf_t *buf)

242 +{

243 + if (0 == nread) {

244 + fprintf(stderr, "SERVER %s: Zero bytes read from spawn pipe.\n", __func__);

245 + } else if (UV_EOF == nread) {

246 + fprintf(stderr, "EOF found in spawn pipe.\n");

247 + } else if (nread < 0) {

248 + fprintf(stderr, "%s: %s\n", __func__, uv_strerror(nread));

249 + }

250 +

251 + if (nread < 0) { /* stop spawn server due to EOF or error */

252 + int error;

253 +

254 + uv_mutex_lock(&wait_children_mutex);

255 + server_shutdown = 1;

256 + spawned_processes = 1;

257 + uv_cond_signal(&wait_children_cond);

258 + uv_mutex_unlock(&wait_children_mutex);

259 +

260 + fprintf(stderr, "Shutting down spawn server event loop.\n");

261 + /* cleanup operations of the event loop */

262 + (void)uv_read_stop((uv_stream_t *) pipe);

263 + uv_close((uv_handle_t *)&server_pipe, NULL);

264 +

265 + error = uv_thread_join(&thread);

266 + if (error) {

267 + fprintf(stderr, "uv_thread_create(): %s", uv_strerror(error));

268 + }

269 + /* After joining it is safe to destroy child_waited_async */

270 + uv_close((uv_handle_t *)&child_waited_async, NULL);

271 + } else if (nread) {

272 +#ifdef SPAWN_DEBUG

273 + fprintf(stderr, "SERVER %s nread %u\n", __func__, (unsigned)nread);

274 +#endif

275 + server_parse_spawn_protocol(nread, buf->base);

276 + }

277 + if (buf && buf->len) {

278 + freez(buf->base);

279 + }

280 +}

281 +

282 +static void on_read_alloc(uv_handle_t *handle,

283 + size_t suggested_size,

284 + uv_buf_t* buf)

285 +{

286 + (void)handle;

287 + buf->base = mallocz(suggested_size);

288 + buf->len = suggested_size;

289 +}

290 +

291 +static void ignore_signal_handler(int signo) {

292 + /*

293 + * By having a signal handler we allow spawned processes to reset default signal dispositions. Setting SIG_IGN

294 + * would be inherited by the spawned children which is not desirable.

295 + */

296 + (void)signo;

297 +}

298 +

299 +void spawn_server(void)

300 +{

301 + int error;

302 +

303 + test_clock_boottime();

304 + test_clock_monotonic_coarse();

305 +

306 + // close all open file descriptors, except the standard ones

307 + // the caller may have left open files (lxc-attach has this issue)

308 + int fd;

309 + for(fd = (int)(sysconf(_SC_OPEN_MAX) - 1) ; fd > 2 ; --fd)

310 + if(fd_is_valid(fd))

311 + close(fd);

312 +

313 + // Have the libuv IPC pipe be closed when forking child processes

314 + (void) fcntl(0, F_SETFD, FD_CLOEXEC);

315 + fprintf(stderr, "Spawn server is up.\n");

316 +

317 + // Define signals we want to ignore

318 + struct sigaction sa;

319 + int signals_to_ignore[] = {SIGPIPE, SIGINT, SIGQUIT, SIGTERM, SIGHUP, SIGUSR1, SIGUSR2, SIGBUS, SIGCHLD};

320 + unsigned ignore_length = sizeof(signals_to_ignore) / sizeof(signals_to_ignore[0]);

321 +

322 + unsigned i;

323 + for (i = 0; i < ignore_length ; ++i) {

324 + sa.sa_flags = 0;

325 + sigemptyset(&sa.sa_mask);

326 + sa.sa_handler = ignore_signal_handler;

327 + if(sigaction(signals_to_ignore[i], &sa, NULL) == -1)

328 + fprintf(stderr, "SPAWN: Failed to change signal handler for signal: %d.\n", signals_to_ignore[i]);

329 + }

330 +

331 + signals_unblock();

332 +

333 + loop = uv_default_loop();

334 + loop->data = NULL;

335 +

336 + error = uv_pipe_init(loop, &server_pipe, 1);

337 + if (error) {

338 + fprintf(stderr, "uv_pipe_init(): %s\n", uv_strerror(error));

339 + exit(error);

340 + }

341 + assert(server_pipe.ipc);

342 +

343 + error = uv_pipe_open(&server_pipe, 0 /* UV_STDIN_FD */);

344 + if (error) {

345 + fprintf(stderr, "uv_pipe_open(): %s\n", uv_strerror(error));

346 + exit(error);

347 + }

348 + avl_init_lock(&spawn_outstanding_exec_tree, spawn_exec_compare);

349 +

350 + spawned_processes = 0;

351 + assert(0 == uv_cond_init(&wait_children_cond));

352 + assert(0 == uv_mutex_init(&wait_children_mutex));

353 + child_waited_list = NULL;

354 + error = uv_async_init(loop, &child_waited_async, child_waited_async_cb);

355 + if (error) {

356 + fprintf(stderr, "uv_async_init(): %s\n", uv_strerror(error));

357 + exit(error);

358 + }

359 +

360 + error = uv_thread_create(&thread, wait_children, NULL);

361 + if (error) {

362 + fprintf(stderr, "uv_thread_create(): %s\n", uv_strerror(error));

363 + exit(error);

364 + }

365 +

366 + prot_buffer_len = 0;

367 + error = uv_read_start((uv_stream_t *)&server_pipe, on_read_alloc, on_pipe_read);

368 + assert(error == 0);

369 +

370 + while (!server_shutdown) {

371 + uv_run(loop, UV_RUN_DEFAULT);

372 + }

373 + fprintf(stderr, "Shutting down spawn server loop complete.\n");

374 + assert(0 == uv_loop_close(loop));

375 +

376 + exit(0);

377 +}