+++ /dev/null
-/*-
- * Copyright 2016 Vsevolod Stakhov
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-#include "config.h"
-
-#include "libutil/util.h"
-#include "libserver/cfg_file.h"
-#include "libserver/cfg_rcl.h"
-#include "libserver/worker_util.h"
-#include "libserver/rspamd_control.h"
-#include "libutil/addr.h"
-#include "lua/lua_common.h"
-#include "unix-std.h"
-#include "utlist.h"
-
-#ifdef HAVE_GLOB_H
-#include <glob.h>
-#endif
-
-static gpointer init_log_helper (struct rspamd_config *cfg);
-static void start_log_helper (struct rspamd_worker *worker);
-
-worker_t log_helper_worker = {
- "log_helper", /* Name */
- init_log_helper, /* Init function */
- start_log_helper, /* Start function */
- RSPAMD_WORKER_UNIQUE | RSPAMD_WORKER_KILLABLE,
- RSPAMD_WORKER_SOCKET_NONE, /* No socket */
- RSPAMD_WORKER_VER /* Version info */
-};
-
-static const guint64 rspamd_log_helper_magic = 0x1090bb46aaa74c9aULL;
-
-/*
- * Worker's context
- */
-struct log_helper_ctx {
- guint64 magic;
- /* Events base */
- struct ev_loop *ev_base;
- /* DNS resolver */
- struct rspamd_dns_resolver *resolver;
- /* Config */
- struct rspamd_config *cfg;
- /* END OF COMMON PART */
- struct event log_ev;
- struct rspamd_worker_lua_script *scripts;
- lua_State *L;
- gint pair[2];
-};
-
-static gpointer
-init_log_helper (struct rspamd_config *cfg)
-{
- struct log_helper_ctx *ctx;
- GQuark type;
-
- type = g_quark_try_string ("log_helper");
- (void)type;
- ctx = rspamd_mempool_alloc (cfg->cfg_pool, sizeof (*ctx));
-
- ctx->magic = rspamd_log_helper_magic;
- ctx->cfg = cfg;
-
- return ctx;
-}
-
-static void
-rspamd_log_helper_read (gint fd, short what, gpointer ud)
-{
- struct log_helper_ctx *ctx = ud;
- guchar buf[8192];
- gssize r;
- guint32 n, i, nextra;
- struct rspamd_protocol_log_message_sum *sm;
- struct rspamd_worker_lua_script *sc;
- struct rspamd_config **pcfg;
- struct ev_loop **pevbase;
-
- r = read (fd, buf, sizeof (buf));
-
- if (r >= (gssize)sizeof (struct rspamd_protocol_log_message_sum)) {
- memcpy (&n, buf, sizeof (n));
- memcpy (&nextra, buf + sizeof (n), sizeof (nextra));
-
- if (n + nextra !=
- (r - sizeof (*sm)) / sizeof (struct rspamd_protocol_log_symbol_result)) {
- msg_warn ("cannot read data from log pipe: bad length: %d elements "
- "announced but %d available", n + nextra,
- (gint)((r - sizeof (*sm)) /
- sizeof (struct rspamd_protocol_log_symbol_result)));
- }
- else {
- sm = g_malloc (r);
- memcpy (sm, buf, r);
-
- DL_FOREACH (ctx->scripts, sc) {
- lua_rawgeti (ctx->L, LUA_REGISTRYINDEX, sc->cbref);
- lua_pushnumber (ctx->L, sm->score);
- lua_pushnumber (ctx->L, sm->required_score);
-
- lua_createtable (ctx->L, n, 0);
- for (i = 0; i < n; i ++) {
- lua_createtable (ctx->L, 2, 0);
- lua_pushinteger (ctx->L, sm->results[i].id);
- lua_rawseti (ctx->L, -2, 1);
- lua_pushnumber (ctx->L, sm->results[i].score);
- lua_rawseti (ctx->L, -2, 2);
-
- lua_rawseti (ctx->L, -2, (i + 1));
- }
-
- pcfg = lua_newuserdata (ctx->L, sizeof (*pcfg));
- *pcfg = ctx->cfg;
- rspamd_lua_setclass (ctx->L, "rspamd{config}", -1);
- lua_pushinteger (ctx->L, sm->settings_id);
-
- lua_createtable (ctx->L, nextra, 0);
- for (i = 0; i < nextra; i ++) {
- lua_createtable (ctx->L, 2, 0);
- lua_pushinteger (ctx->L, sm->results[i + n].id);
- lua_rawseti (ctx->L, -2, 1);
- lua_pushnumber (ctx->L, sm->results[i + n].score);
- lua_rawseti (ctx->L, -2, 2);
-
- lua_rawseti (ctx->L, -2, (i + 1));
- }
-
- pevbase = lua_newuserdata (ctx->L, sizeof (*pevbase));
- *pevbase = ctx->ev_base;
- rspamd_lua_setclass (ctx->L, "rspamd{ev_base}", -1);
-
- if (lua_pcall (ctx->L, 7, 0, 0) != 0) {
- msg_err ("error executing log handler code: %s",
- lua_tostring (ctx->L, -1));
- lua_pop (ctx->L, 1);
- }
- }
-
- g_free (sm);
- }
- }
- else if (r == -1) {
- if (errno != EAGAIN && errno != EINTR) {
- msg_warn ("cannot read data from log pipe: %s", strerror (errno));
- event_del (&ctx->log_ev);
- }
- }
- else if (r == 0) {
- msg_warn ("cannot read data from log pipe: EOF");
- event_del (&ctx->log_ev);
- }
-}
-
-static void
-rspamd_log_helper_reply_handler (struct rspamd_worker *worker,
- struct rspamd_srv_reply *rep, gint rep_fd,
- gpointer ud)
-{
- struct log_helper_ctx *ctx = ud;
-
- close (ctx->pair[1]);
- msg_info ("start waiting for log events");
- event_set (&ctx->log_ev, ctx->pair[0], EV_READ | EV_PERSIST,
- rspamd_log_helper_read, ctx);
- event_base_set (ctx->ev_base, &ctx->log_ev);
- event_add (&ctx->log_ev, NULL);
-}
-
-static void
-start_log_helper (struct rspamd_worker *worker)
-{
- struct log_helper_ctx *ctx = worker->ctx;
- gssize r = -1;
- gint nscripts = 0;
- struct rspamd_worker_lua_script *tmp;
- static struct rspamd_srv_command srv_cmd;
-
- ctx->ev_base = rspamd_prepare_worker (worker,
- "log_helper",
- NULL);
- ctx->cfg = worker->srv->cfg;
- ctx->scripts = worker->cf->scripts;
- ctx->L = ctx->cfg->lua_state;
- ctx->resolver = rspamd_dns_resolver_init (worker->srv->logger,
- ctx->ev_base,
- worker->srv->cfg);
- rspamd_upstreams_library_config (worker->srv->cfg, ctx->cfg->ups_ctx,
- ctx->ev_base, ctx->resolver->r);
-
- DL_COUNT (worker->cf->scripts, tmp, nscripts);
- msg_info ("started log_helper worker with %d scripts", nscripts);
-
- r = rspamd_socketpair (ctx->pair, FALSE);
-
- if (r == -1) {
- msg_err ("cannot create socketpair: %s, exiting now", strerror (errno));
- /* Prevent new processes spawning */
- exit (EXIT_SUCCESS);
- }
-
- memset (&srv_cmd, 0, sizeof (srv_cmd));
- srv_cmd.type = RSPAMD_SRV_LOG_PIPE;
- srv_cmd.cmd.log_pipe.type = RSPAMD_LOG_PIPE_SYMBOLS;
-
-
- /* Wait for startup being completed */
- rspamd_mempool_lock_mutex (worker->srv->start_mtx);
- rspamd_srv_send_command (worker, ctx->ev_base, &srv_cmd, ctx->pair[1],
- rspamd_log_helper_reply_handler, ctx);
- rspamd_mempool_unlock_mutex (worker->srv->start_mtx);
- rspamd_lua_run_postloads (ctx->cfg->lua_state, ctx->cfg, ctx->ev_base,
- worker);
- event_base_loop (ctx->ev_base, 0);
- close (ctx->pair[0]);
- rspamd_worker_block_signals ();
-
- REF_RELEASE (ctx->cfg);
- rspamd_log_close (worker->srv->logger, TRUE);
-
- exit (EXIT_SUCCESS);
-}
projects / rspamd.git / commitdiff