JKBfEbs1

1#include <signal.h>
2
3#include "parser.h"
4
5static char ulog_log[1024];
6static char parsed_offset_file[1024];
7static char mysql_host[1024];
8static char mysql_user[1024];
9static char mysql_password[1024];
10static char mysql_database[1024];
11
12static const char* output_log_path;
13static MYSQL *connection;
14
15static struct task_queue* global_queue;
16pthread_mutex_t global_queue_mutex = PTHREAD_MUTEX_INITIALIZER;
17static char do_worker;
18pthread_mutex_t worker_mutex = PTHREAD_MUTEX_INITIALIZER;
19pthread_t worker_thread;
20
21static fpos_t raw_log_offset;
22static FILE* raw_log_offset_file;
23
24int create_offset(const char* path) {
25    raw_log_offset_file = fopen(path, "rw");
26    if (raw_log_offset_file == NULL) {
27        return 1;
28    }
29    return 0;
30}
31
32void read_offset() {
33    fseek(raw_log_offset_file, 0, SEEK_END); 
34    long int size = ftell(raw_log_offset_file);
35    fseek(raw_log_offset_file, 0, SEEK_SET);
36
37    if (size != 0) {
38        rewind(raw_log_offset_file);
39        fread(&raw_log_offset, sizeof (fpos_t), 1, raw_log_offset_file);
40    }    
41}
42
43void write_offset() {
44    rewind(raw_log_offset_file);
45    fwrite(&raw_log_offset, sizeof(fpos_t), 1, raw_log_offset_file);
46}
47
48void destroy_offset() {
49    fclose(raw_log_offset_file);
50}
51
52void read_config(const char* config_path) {
53    FILE* config;
54    char line_buffer[LINE_BUFFER_SIZE];
55
56    config = fopen(config_path, "r+");
57    
58    if (config == NULL) {
59        syslog(LOG_NOTICE, "can't read config. aborting");
60        exit(EXIT_FAILURE);
61    } else {
62        while (!feof(config)) {
63            if (fgets(line_buffer, LINE_BUFFER_SIZE, config) != NULL) {
64                if (strlen(line_buffer) == 0) {
65                    continue;
66                }
67
68                if (strncmp(line_buffer, "ulog_log=", 9) == 0) { /* извлекаем путь до журнала ulogd */
69                    strcpy(ulog_log, strtrim_right(line_buffer + 9, '\n'));
70                }
71                if (strncmp(line_buffer, "parsed_offset_file=", 11) == 0) { /* извлекаем путь до смещения в файле лога */
72                    strcpy(parsed_offset_file, strtrim_right(line_buffer + 11, '\n'));
73                }
74                if (strncmp(line_buffer, "mysql_host=", 11) == 0) { /* извлекаем адрес СУБД сервера */
75                    strcpy(mysql_host, strtrim_right(line_buffer + 11, '\n'));
76                }
77                if (strncmp(line_buffer, "mysql_user=", 11) == 0) { /* извлекаем имя пользователя СУБД */
78                    strcpy(mysql_user, strtrim_right(line_buffer + 11, '\n'));
79                }
80                if (strncmp(line_buffer, "mysql_password=", 15) == 0) { /* извлекаем пароль пользователя СУБД */
81                    strcpy(mysql_password, strtrim_right(line_buffer + 15, '\n'));
82                }
83                if (strncmp(line_buffer, "mysql_database=", 15) == 0) { /* извлекаем имя БД СУБД */
84                    strcpy(mysql_database, strtrim_right(line_buffer + 15, '\n'));
85                }
86            }
87        }
88        fclose(config);
89    }
90}
91
92int connect_to_database() {
93    /*printf("MySQL client version: %s\n", mysql_get_client_info());*/
94    connection = mysql_init(NULL);
95
96    if (connection == NULL) {
97        fprintf(stdout, "Database error %u: %s\n", mysql_errno(connection), mysql_error(connection));
98        return 1;
99    }
100
101    if (mysql_real_connect(connection, mysql_host, mysql_user, mysql_password, mysql_database, 0, NULL, 0) == NULL) {
102        printf("Database error %u: %s\n", mysql_errno(connection), mysql_error(connection));
103        return 1;
104    }
105
106    return 0;
107}
108
109void disconnect_from_database() {
110    if (connection != NULL) {
111        mysql_close(connection);
112    }
113}
114
115void run_parse_log_service(const char* config_path, const char* offset_path, char with_clear) {
116    FILE* raw_log;
117    char line_buffer[LINE_BUFFER_SIZE];
118    struct task* t = NULL;
119    //fpos_t file_pos;
120    unsigned long file_size = 0;
121    //char first_open = 1;
122
123    struct stat st;
124    
125    char year[YEAR_LEN];
126    char month[MONTH_LEN];
127    char day[DAY_LEN];
128    char timed[TIME_LEN];
129    char src[IP_ADDRESS_LEN];
130    char dst[IP_ADDRESS_LEN];
131    char packet_len[PACKET_LEN];
132    char proto[PROTOCOL_TYPE_LEN];
133    char dst_port[PORT_LEN];
134    
135    time_t timer;
136    struct tm *ts;
137
138    read_config(config_path);
139
140    if (connect_to_database() != 0) {
141        syslog(LOG_NOTICE, "can't connect to database. aborting");
142        exit(EXIT_FAILURE);
143    }
144
145    if (with_clear) {
146        syslog(LOG_NOTICE, "clear database");
147        clear_data();
148    }
149
150    global_queue = create_task_queue();
151
152    do_worker = 1;
153    worker_thread = pthread_create(&worker_thread, NULL, worker, NULL);
154
155    if (create_offset(offset_path) == 1) {
156        syslog(LOG_NOTICE, "can't create offset file. aborting");
157        exit(EXIT_FAILURE);
158    }
159    read_offset();
160    
161    while (1) {
162        raw_log = fopen(ulog_log, "r+");      
163        if (raw_log == NULL) {
164            syslog(LOG_NOTICE, "can't open ulogd log. aborting");
165            break;
166        } else {
167            stat(ulog_log, &st);
168            file_size = st.st_size;
169
170            fsetpos(raw_log, &raw_log_offset);
171
172            while (!feof(raw_log)) {
173                //fgetpos(raw_log, &file_pos);
174                fgetpos(raw_log, &raw_log_offset);
175                //printf("=%d\n", file_pp);
176                if (fgets(line_buffer, LINE_BUFFER_SIZE, raw_log) != NULL) {
177                    if (strlen(line_buffer) == 0) {
178                        continue;
179                    }
180                    year[0]         = '\0';
181                    month[0]        = '\0';
182                    day[0]          = '\0';
183                    timed[0]        = '\0';
184                    src[0]          = '\0';
185                    dst[0]          = '\0';
186                    packet_len[0]   = '\0';
187                    proto[0]        = '\0';
188                    dst_port[0]     = '\0';
189
190                    parse_raw_line(line_buffer, month, day, timed, src, dst, packet_len, proto, dst_port);
191
192                    timer = time(NULL);
193                    ts = localtime(&timer);
194                    strftime(year, sizeof(year), "%Y", ts);
195                    
196                    t = create_task(year, month, day, timed, src, dst, packet_len, proto, dst_port);
197
198                    pthread_mutex_lock(&global_queue_mutex);
199                    add_task_to_task_queue(global_queue, t);
200                    pthread_mutex_unlock(&global_queue_mutex);
201                }
202            }
203            fclose(raw_log);
204
205            write_offset();
206
207            syslog(LOG_NOTICE, "await new raw data");
208            while (1) {
209                stat(ulog_log, &st);
210                if (st.st_size > file_size) {
211                    break;
212                }
213                sleep(1);
214            }
215        }
216    }
217}
218
219void* worker(void* args) {
220    struct task_queue* tq = create_task_queue();
221    if (global_queue->size > TASK_BUFFER_SIZE) {
222        do_task_queue(global_queue);
223    }
224
225    while (1) {
226        pthread_mutex_lock(&worker_mutex);
227        if (!do_worker) {
228            break;
229        }
230        pthread_mutex_unlock(&worker_mutex);
231
232        if (global_queue->size > TASK_BUFFER_SIZE) {
233            pthread_mutex_lock(&global_queue_mutex);
234            tq->head = global_queue->head;
235            tq->tail = global_queue->tail;
236            tq->size = global_queue->size;
237
238            global_queue->head = NULL;
239            global_queue->tail = NULL;
240            global_queue->size = 0;
241            
242            pthread_mutex_unlock(&global_queue_mutex);
243            
244            do_task_queue(tq);
245        }
246        sleep(1);
247    }
248    
249    do_task_queue(tq);
250    destroy_task_queue(tq);
251}
252
253void parse_raw_line(char* line, char* month, char* day, char* time, char* src, char* dst, char* len, char* proto, char* dst_port) {
254    char *line_part;
255    int part_counter = 0;
256    char dst_found = 0;
257    char len_found = 0;
258    char proto_found = 0;
259    char dst_port_found = 0;
260
261    line_part = strtok(line, " ");
262    while (line_part != NULL) {
263        if (part_counter == 0) { /* копируем месяц */
264            strcpy(month, line_part);
265        } else if (part_counter == 1) { /* копируем день */
266            strcpy(day, line_part);
267        } else if (part_counter == 2) { /* копируем время */
268            strcpy(time, line_part);
269        } else if (part_counter == 4) { /* адрес клиента */
270            strncpy(src, line_part, strlen(line_part) - 1);
271        }
272
273        if (!dst_found && strncmp(line_part, "DST=", 4) == 0) { /* извлекаем адрес назначения */
274            strcpy(dst, line_part + 4);
275            dst_found = 1;
276        }
277
278        if (!len_found && strncmp(line_part, "LEN=", 4) == 0) { /* извлекаем длину пакета */
279            strcpy(len, line_part + 4);
280            len_found = 1;
281        }
282
283        if (!proto_found && strncmp(line_part, "PROTO=", 6) == 0) { /* извлекаем тип протокола */
284            strcpy(proto, line_part + 6);
285            proto_found = 1;
286        }
287
288        if (!dst_port_found && strncmp(line_part, "DPT=", 4) == 0) { /* извлекаем тип порт назначения */
289            strcpy(dst_port, line_part + 4);
290            dst_port_found = 1;
291        }
292
293        line_part = strtok(NULL, " ");
294        ++part_counter;
295    }
296}
297
298int do_task(struct task* t) {
299    static char query[1024];
300    static char month[3];
301    
302    sprintf(query, "CREATE TABLE IF NOT EXISTS `%s` (id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT, ts DATETIME, proto CHAR(10), ip CHAR(15), dst_port CHAR(5), rx INT UNSIGNED, PRIMARY KEY (id))", t->src);
303
304    if (mysql_query(connection, query) != 0) {
305        return 1;
306    }
307
308    if (str_month_to_num(month, t->month) != 0) {
309        return 1;
310    }
311    
312    sprintf(query, "INSERT INTO `%s` (ts, proto, ip, dst_port, rx) VALUES ('%s-%s-%s %s', '%s', '%s', '%s', '%s')", t->src, t->year, month, t->day, t->time, t->proto, t->dst, t->dst_port, t->packet_len);
313
314    if (mysql_query(connection, query) != 0) {
315        return 1;
316    }
317
318    return 0;
319}
320
321int clear_data() {
322    static char query[1024];
323    MYSQL_RES *result;
324    MYSQL_ROW row;
325
326    sprintf(query, "SHOW TABLES");
327
328    if (mysql_query(connection, query) != 0) {
329        return 1;
330    }
331    result = mysql_store_result(connection);
332
333    while ((row = mysql_fetch_row(result))) {
334        sprintf(query, "DROP TABLE `%s`", row[0]);
335        if (mysql_query(connection, query) != 0) {
336            return 1;
337        }
338    }
339    mysql_free_result(result);
340
341    return 0;
342}
343
344void signal_handler(int sig) {
345    switch (sig) {
346        case SIGTERM:
347        case SIGINT:
348        case SIGQUIT:
349            pthread_mutex_lock(&worker_mutex);
350            do_worker = 0;
351            pthread_mutex_unlock(&worker_mutex);
352            pthread_join(worker_thread, NULL);
353
354            destroy_task_queue(global_queue);
355
356            destroy_offset();
357
358            disconnect_from_database();
359            syslog(LOG_NOTICE, "stopped on signal % d", sig);
360            closelog();
361            exit(0);
362            break;
363        default:
364            //syslog(LOG_NOTICE, "got signal %d. ignore...", sig);
365            break;
366    }
367}