1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
|
#include "grok.h"
#include "grok_program.h"
#include "grok_input.h"
#include "grok_matchconf.h"
#include <sys/time.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <time.h>
#include <unistd.h>
#include <event.h>
#include <signal.h>
#include <fcntl.h>
static void *_event_init = NULL;
void _collection_sigchld(int sig, short what, void *data);
grok_collection_t *grok_collection_init() {
grok_collection_t *gcol;
gcol = calloc(1, sizeof(grok_collection_t));
gcol->nprograms = 0;
gcol->program_size = 10;
gcol->programs = calloc(gcol->program_size, sizeof(grok_program_t));
gcol->ebase = event_init();
gcol->exit_code = 0;
gcol->ev_sigchld = malloc(sizeof(struct event));
signal_set(gcol->ev_sigchld, SIGCHLD, _collection_sigchld, gcol);
signal_add(gcol->ev_sigchld, NULL);
return gcol;
}
void grok_collection_check_end_state(grok_collection_t *gcol) {
int still_alive = 0;
int p, i, m;
int reaction_count = 0;
for (p = 0; p < gcol->nprograms; p++) {
grok_program_t *gprog = gcol->programs[p];
reaction_count += gprog->reactions;
for (i = 0; i < gprog->ninputs; i++) {
grok_input_t *ginput = &gprog->inputs[i];
still_alive += (ginput->done == 0);
}
for (m = 0; m < gprog->nmatchconfigs; m++) {
grok_matchconf_t *gmc = &gprog->matchconfigs[m];
still_alive += (gmc->pid != 0);
}
}
if (still_alive == 0) {
struct timeval nodelay = { 0, 0 };
grok_log(gcol, LOG_PROGRAM,
"No more subprocesses are running. Breaking event loop now.");
/* Cleanup */
grok_matchconfig_global_cleanup();
event_base_loopexit(gcol->ebase, &nodelay);
if (reaction_count == 0) {
gcol->exit_code = 1;
}
}
}
void grok_collection_add(grok_collection_t *gcol, grok_program_t *gprog) {
int i = 0;
grok_log(gcol, LOG_PROGRAM, "Adding %d inputs", gprog->ninputs);
for (i = 0; i < gprog->ninputs; i++) {
grok_log(gprog, LOG_PROGRAM, "Adding input %d", i);
gprog->inputs[i].gprog = gprog;
grok_program_add_input(gprog, gprog->inputs + i);
}
gcol->nprograms++;
if (gcol->nprograms == gcol->program_size) {
gcol->program_size *= 2;
gcol->programs = realloc(gcol->programs,
gcol->program_size * sizeof(grok_collection_t *));
}
gcol->programs[gcol->nprograms - 1] = gprog;
gprog->gcol = gcol;
}
void _collection_sigchld(int sig, short what, void *data) {
grok_collection_t *gcol = (grok_collection_t*)data;
struct timeval nodelay = { 0, 0 };
int i = 0;
int prognum;
int pid, status;
grok_log(gcol, LOG_PROGRAM, "SIGCHLD received");
while ((pid = waitpid(-1, &status, WNOHANG)) > 0) {
grok_log(gcol, LOG_PROGRAM, "Found dead child pid %d", pid);
for (prognum = 0; prognum < gcol->nprograms; prognum++) {
grok_program_t *gprog = gcol->programs[prognum];
/* we found a dead child. Look for an input_process it belongs to,
* then see if we should restart it
*/
for (i = 0; i < gprog->nmatchconfigs; i++) {
grok_matchconf_t *gmc = (gprog->matchconfigs + i);
if (gmc->pid != pid)
continue;
grok_log(gcol, LOG_PROGRAM, "Pid %d is a matchconf shell", pid);
gmc->pid = 0;
}
for (i = 0; i < gprog->ninputs; i++) {
grok_input_t *ginput = (gprog->inputs + i);
if (ginput->type != I_PROCESS)
continue;
grok_input_process_t *gipt = &(ginput->source.process);
if (gipt->pid != pid)
continue;
grok_log(gcol, LOG_PROGRAM, "Pid %d is an exec process", pid);
/* use ginput's log values */
grok_log(ginput, LOG_PROGRAM, "Reaped child pid %d. Was process '%s'",
pid, gipt->cmd);
if (PROCESS_SHOULD_RESTART(gipt)) {
/* Calculate the restart delay */
/* XXX: This code should go in the eof handler */
struct timeval restart_delay = { 0, 0 };
if (gipt->run_interval > 0) {
struct timeval interval = { gipt->run_interval, 0 };
struct timeval duration;
struct timeval now;
gettimeofday(&now, NULL);
timersub(&now, &(gipt->start_time), &duration);
timersub(&interval, &duration, &restart_delay);
}
if (gipt->min_restart_delay > 0) {
struct timeval fixed_delay = { gipt->min_restart_delay, 0 };
if (timercmp(&restart_delay, &fixed_delay, <)) {
restart_delay.tv_sec = fixed_delay.tv_sec;
restart_delay.tv_usec = fixed_delay.tv_usec;
}
}
grok_log(ginput, LOG_PROGRAM,
"Scheduling process restart in %d.%d seconds: %s",
restart_delay.tv_sec, restart_delay.tv_usec, gipt->cmd);
ginput->restart_delay.tv_sec = restart_delay.tv_sec;
ginput->restart_delay.tv_usec = restart_delay.tv_usec;
} else {
grok_log(gprog, LOG_PROGRAM, "Not restarting process '%s'", gipt->cmd);
}
event_once(-1, EV_TIMEOUT, grok_input_eof_handler, ginput, &nodelay);
} /* end for looping over gprog's inputs */
} /* end for looping over gcol's programs */
} /* while waitpid */
grok_collection_check_end_state(gcol);
}
void grok_collection_loop(grok_collection_t *gcol) {
event_base_dispatch(gcol->ebase);
}
|
