1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356 |
- /*
- * Copyright 2024 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 "rspamd.h"
- #include "lua/lua_common.h"
- #include "worker_util.h"
- #include "unix-std.h"
- #include "utlist.h"
- #include "ottery.h"
- #include "rspamd_control.h"
- #include "libserver/maps/map.h"
- #include "libserver/maps/map_private.h"
- #include "libserver/http/http_private.h"
- #include "libserver/http/http_router.h"
- #include "libutil/rrd.h"
-
- /* sys/resource.h */
- #ifdef HAVE_SYS_RESOURCE_H
- #include <sys/resource.h>
- #endif
- /* pwd and grp */
- #ifdef HAVE_PWD_H
- #include <pwd.h>
- #endif
- #ifdef HAVE_GRP_H
- #include <grp.h>
- #endif
- #ifdef HAVE_LIBUTIL_H
- #include <libutil.h>
- #endif
- #include "zlib.h"
-
- #ifdef HAVE_UCONTEXT_H
- #include <ucontext.h>
- #elif defined(HAVE_SYS_UCONTEXT_H)
- #include <sys/ucontext.h>
- #endif
-
- #ifdef HAVE_SYS_WAIT_H
- #include <sys/wait.h>
- #include <math.h>
-
- #endif
-
- #include "contrib/libev/ev.h"
- #include "libstat/stat_api.h"
-
- struct rspamd_worker *rspamd_current_worker = NULL;
-
- /* Forward declaration */
- static void rspamd_worker_heartbeat_start(struct rspamd_worker *,
- struct ev_loop *);
-
- static void rspamd_worker_ignore_signal(struct rspamd_worker_signal_handler *);
- /**
- * Return worker's control structure by its type
- * @param type
- * @return worker's control structure or NULL
- */
- worker_t *
- rspamd_get_worker_by_type(struct rspamd_config *cfg, GQuark type)
- {
- worker_t **pwrk;
-
- pwrk = cfg->compiled_workers;
- while (pwrk && *pwrk) {
- if (rspamd_check_worker(cfg, *pwrk)) {
- if (g_quark_from_string((*pwrk)->name) == type) {
- return *pwrk;
- }
- }
-
- pwrk++;
- }
-
- return NULL;
- }
-
- static void
- rspamd_worker_check_finished(EV_P_ ev_timer *w, int revents)
- {
- int *pnchecks = (int *) w->data;
-
- if (*pnchecks > SOFT_SHUTDOWN_TIME * 10) {
- msg_warn("terminating worker before finishing of terminate handlers");
- ev_break(EV_A_ EVBREAK_ONE);
- }
- else {
- int refcount = ev_active_cnt(EV_A);
-
- if (refcount == 1) {
- ev_break(EV_A_ EVBREAK_ONE);
- }
- else {
- ev_timer_again(EV_A_ w);
- }
- }
- }
-
- static gboolean
- rspamd_worker_finalize(gpointer user_data)
- {
- struct rspamd_task *task = user_data;
-
- if (!(task->flags & RSPAMD_TASK_FLAG_PROCESSING)) {
- msg_info_task("finishing actions has been processed, terminating");
- /* ev_break (task->event_loop, EVBREAK_ALL); */
- task->worker->state = rspamd_worker_wanna_die;
- rspamd_session_destroy(task->s);
-
- return TRUE;
- }
-
- return FALSE;
- }
-
- gboolean
- rspamd_worker_call_finish_handlers(struct rspamd_worker *worker)
- {
- struct rspamd_task *task;
- struct rspamd_config *cfg = worker->srv->cfg;
- struct rspamd_abstract_worker_ctx *ctx;
- struct rspamd_config_cfg_lua_script *sc;
-
- if (cfg->on_term_scripts) {
- ctx = (struct rspamd_abstract_worker_ctx *) worker->ctx;
- /* Create a fake task object for async events */
- task = rspamd_task_new(worker, cfg, NULL, NULL, ctx->event_loop, FALSE);
- task->resolver = ctx->resolver;
- task->flags |= RSPAMD_TASK_FLAG_PROCESSING;
- task->s = rspamd_session_create(task->task_pool,
- rspamd_worker_finalize,
- NULL,
- (event_finalizer_t) rspamd_task_free,
- task);
-
- DL_FOREACH(cfg->on_term_scripts, sc)
- {
- lua_call_finish_script(sc, task);
- }
-
- task->flags &= ~RSPAMD_TASK_FLAG_PROCESSING;
-
- if (rspamd_session_pending(task->s)) {
- return TRUE;
- }
- }
-
- return FALSE;
- }
-
- static void
- rspamd_worker_terminate_handlers(struct rspamd_worker *w)
- {
- if (w->nconns == 0 &&
- (!(w->flags & RSPAMD_WORKER_SCANNER) || w->srv->cfg->on_term_scripts == NULL)) {
- /*
- * We are here either:
- * - No active connections are represented
- * - No term scripts are registered
- * - Worker is not a scanner, so it can die safely
- */
- w->state = rspamd_worker_wanna_die;
- }
- else {
- if (w->nconns > 0) {
- /*
- * Wait until all connections are terminated
- */
- w->state = rspamd_worker_wait_connections;
- }
- else {
- /*
- * Start finish scripts
- */
- if (w->state != rspamd_worker_wait_final_scripts) {
- w->state = rspamd_worker_wait_final_scripts;
-
- if ((w->flags & RSPAMD_WORKER_SCANNER) &&
- rspamd_worker_call_finish_handlers(w)) {
- msg_info("performing async finishing actions");
- w->state = rspamd_worker_wait_final_scripts;
- }
- else {
- /*
- * We are done now
- */
- msg_info("no async finishing actions, terminating");
- w->state = rspamd_worker_wanna_die;
- }
- }
- }
- }
- }
-
- static void
- rspamd_worker_on_delayed_shutdown(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker *worker = (struct rspamd_worker *) w->data;
-
- worker->state = rspamd_worker_wanna_die;
- ev_timer_stop(EV_A_ w);
- ev_break(loop, EVBREAK_ALL);
- }
-
- static void
- rspamd_worker_shutdown_check(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker *worker = (struct rspamd_worker *) w->data;
-
- if (worker->state != rspamd_worker_wanna_die) {
- rspamd_worker_terminate_handlers(worker);
-
- if (worker->state == rspamd_worker_wanna_die) {
- /* We are done, kill event loop */
- ev_timer_stop(EV_A_ w);
- ev_break(EV_A_ EVBREAK_ALL);
- }
- else {
- /* Try again later */
- ev_timer_again(EV_A_ w);
- }
- }
- else {
- ev_timer_stop(EV_A_ w);
- ev_break(EV_A_ EVBREAK_ALL);
- }
- }
-
- static void
- rspamd_worker_shutdown_check_nconns(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker *worker = (struct rspamd_worker *) w->data;
-
- if (worker->state != rspamd_worker_wanna_die) {
- if (worker->state != rspamd_worker_wait_connections) {
- rspamd_worker_terminate_handlers(worker);
- }
-
- /* Check again, as rspamd_worker_terminate_handlers could change the worker's state */
- if (worker->state == rspamd_worker_wanna_die) {
- /* We are done, kill event loop */
- ev_timer_stop(EV_A_ w);
- ev_break(EV_A_ EVBREAK_ALL);
- }
- else {
- /* Try again later */
-
- if (worker->nconns > 0) {
- ev_timer_again(EV_A_ w);
- }
- else {
- /* No connections left, can close everything */
- ev_timer_stop(EV_A_ w);
- ev_break(EV_A_ EVBREAK_ALL);
- }
- }
- }
- else {
- ev_timer_stop(EV_A_ w);
- ev_break(EV_A_ EVBREAK_ALL);
- }
- }
-
- /*
- * Config reload is designed by sending sigusr2 to active workers and pending shutdown of them
- */
- static gboolean
- rspamd_worker_usr2_handler(struct rspamd_worker_signal_handler *sigh, void *arg)
- {
- /* Do not accept new connections, preparing to end worker's process */
- if (sigh->worker->state == rspamd_worker_state_running) {
- static ev_timer shutdown_ev, shutdown_check_ev;
- ev_tstamp shutdown_ts;
-
- if (sigh->worker->flags & RSPAMD_WORKER_NO_TERMINATE_DELAY) {
- shutdown_ts = 0.0;
- }
- else {
- shutdown_ts = MAX(SOFT_SHUTDOWN_TIME,
- sigh->worker->srv->cfg->task_timeout * 2.0);
- }
-
- rspamd_worker_ignore_signal(sigh);
- sigh->worker->state = rspamd_worker_state_terminating;
-
- rspamd_default_log_function(G_LOG_LEVEL_INFO,
- sigh->worker->srv->server_pool->tag.tagname,
- sigh->worker->srv->server_pool->tag.uid,
- G_STRFUNC,
- "worker's shutdown is pending in %.2f sec",
- shutdown_ts);
-
- /* Soft shutdown timer */
- shutdown_ev.data = sigh->worker;
- ev_timer_init(&shutdown_ev, rspamd_worker_on_delayed_shutdown,
- shutdown_ts, 0.0);
- ev_timer_start(sigh->event_loop, &shutdown_ev);
-
- if (!(sigh->worker->flags & RSPAMD_WORKER_NO_TERMINATE_DELAY)) {
- /* This timer checks if we are ready to die and is called frequently */
- shutdown_check_ev.data = sigh->worker;
- ev_timer_init(&shutdown_check_ev, rspamd_worker_shutdown_check,
- 0.5, 0.5);
- ev_timer_start(sigh->event_loop, &shutdown_check_ev);
- }
-
- rspamd_worker_stop_accept(sigh->worker);
- }
-
- /* No more signals */
- return FALSE;
- }
-
- /*
- * Reopen log is designed by sending sigusr1 to active workers and pending shutdown of them
- */
- static gboolean
- rspamd_worker_usr1_handler(struct rspamd_worker_signal_handler *sigh, void *arg)
- {
- struct rspamd_main *rspamd_main = sigh->worker->srv;
-
- rspamd_log_reopen(sigh->worker->srv->logger, rspamd_main->cfg, -1, -1);
- msg_info_main("logging reinitialised");
-
- /* Get more signals */
- return TRUE;
- }
-
- static gboolean
- rspamd_worker_term_handler(struct rspamd_worker_signal_handler *sigh, void *arg)
- {
- if (sigh->worker->state == rspamd_worker_state_running) {
- static ev_timer shutdown_ev, shutdown_check_ev;
- ev_tstamp shutdown_ts;
-
- if (sigh->worker->flags & RSPAMD_WORKER_NO_TERMINATE_DELAY) {
- shutdown_ts = 0.0;
- }
- else {
- shutdown_ts = MAX(SOFT_SHUTDOWN_TIME,
- sigh->worker->srv->cfg->task_timeout * 2.0);
- }
-
- rspamd_worker_ignore_signal(sigh);
- sigh->worker->state = rspamd_worker_state_terminating;
- rspamd_default_log_function(G_LOG_LEVEL_INFO,
- sigh->worker->srv->server_pool->tag.tagname,
- sigh->worker->srv->server_pool->tag.uid,
- G_STRFUNC,
- "terminating in up to %.0f second after receiving signal %s",
- shutdown_ts,
- g_strsignal(sigh->signo));
-
- rspamd_worker_stop_accept(sigh->worker);
- rspamd_worker_terminate_handlers(sigh->worker);
-
- /* Check if we are ready to die */
- if (sigh->worker->state != rspamd_worker_wanna_die) {
- /* This timer is called when we have no choices but to die */
- shutdown_ev.data = sigh->worker;
- ev_timer_init(&shutdown_ev, rspamd_worker_on_delayed_shutdown,
- shutdown_ts, 0.0);
- ev_timer_start(sigh->event_loop, &shutdown_ev);
-
- if (!(sigh->worker->flags & RSPAMD_WORKER_NO_TERMINATE_DELAY)) {
- /* This timer checks if we are ready to die and is called frequently */
- shutdown_check_ev.data = sigh->worker;
- ev_timer_init(&shutdown_check_ev, rspamd_worker_shutdown_check,
- 0.5, 0.5);
- ev_timer_start(sigh->event_loop, &shutdown_check_ev);
- }
- else {
- /* This timer checks if we have no active connections pending and terminates once all conns are done */
- shutdown_check_ev.data = sigh->worker;
- ev_timer_init(&shutdown_check_ev, rspamd_worker_shutdown_check_nconns,
- 0.5, 0.5);
- ev_timer_start(sigh->event_loop, &shutdown_check_ev);
- }
- }
- else {
- /* Flag to die has been already set */
- ev_break(sigh->event_loop, EVBREAK_ALL);
- }
- }
-
- /* Stop reacting on signals */
- return FALSE;
- }
-
- static void
- rspamd_worker_signal_handle(EV_P_ ev_signal *w, int revents)
- {
- struct rspamd_worker_signal_handler *sigh =
- (struct rspamd_worker_signal_handler *) w->data;
- struct rspamd_worker_signal_handler_elt *cb, *cbtmp;
-
- /* Call all signal handlers registered */
- DL_FOREACH_SAFE(sigh->cb, cb, cbtmp)
- {
- if (!cb->handler(sigh, cb->handler_data)) {
- DL_DELETE(sigh->cb, cb);
- g_free(cb);
- }
- }
- }
-
- static void
- rspamd_worker_ignore_signal(struct rspamd_worker_signal_handler *sigh)
- {
- sigset_t set;
-
- ev_signal_stop(sigh->event_loop, &sigh->ev_sig);
- sigemptyset(&set);
- sigaddset(&set, sigh->signo);
- sigprocmask(SIG_BLOCK, &set, NULL);
- }
-
- static void
- rspamd_worker_default_signal(int signo)
- {
- struct sigaction sig;
-
- sigemptyset(&sig.sa_mask);
- sigaddset(&sig.sa_mask, signo);
- sig.sa_handler = SIG_DFL;
- sig.sa_flags = 0;
- sigaction(signo, &sig, NULL);
- }
-
- static void
- rspamd_sigh_free(void *p)
- {
- struct rspamd_worker_signal_handler *sigh = p;
- struct rspamd_worker_signal_handler_elt *cb, *tmp;
-
- DL_FOREACH_SAFE(sigh->cb, cb, tmp)
- {
- DL_DELETE(sigh->cb, cb);
- g_free(cb);
- }
-
- ev_signal_stop(sigh->event_loop, &sigh->ev_sig);
- rspamd_worker_default_signal(sigh->signo);
- g_free(sigh);
- }
-
- void rspamd_worker_set_signal_handler(int signo, struct rspamd_worker *worker,
- struct ev_loop *event_loop,
- rspamd_worker_signal_cb_t handler,
- void *handler_data)
- {
- struct rspamd_worker_signal_handler *sigh;
- struct rspamd_worker_signal_handler_elt *cb;
-
- sigh = g_hash_table_lookup(worker->signal_events, GINT_TO_POINTER(signo));
-
- if (sigh == NULL) {
- sigh = g_malloc0(sizeof(*sigh));
- sigh->signo = signo;
- sigh->worker = worker;
- sigh->event_loop = event_loop;
- sigh->enabled = TRUE;
-
- sigh->ev_sig.data = sigh;
- ev_signal_init(&sigh->ev_sig, rspamd_worker_signal_handle, signo);
- ev_signal_start(event_loop, &sigh->ev_sig);
-
- g_hash_table_insert(worker->signal_events,
- GINT_TO_POINTER(signo),
- sigh);
- }
-
- cb = g_malloc0(sizeof(*cb));
- cb->handler = handler;
- cb->handler_data = handler_data;
- DL_APPEND(sigh->cb, cb);
- }
-
- void rspamd_worker_init_signals(struct rspamd_worker *worker,
- struct ev_loop *event_loop)
- {
- /* A set of terminating signals */
- rspamd_worker_set_signal_handler(SIGTERM, worker, event_loop,
- rspamd_worker_term_handler, NULL);
- rspamd_worker_set_signal_handler(SIGINT, worker, event_loop,
- rspamd_worker_term_handler, NULL);
- rspamd_worker_set_signal_handler(SIGHUP, worker, event_loop,
- rspamd_worker_term_handler, NULL);
-
- /* Special purpose signals */
- rspamd_worker_set_signal_handler(SIGUSR1, worker, event_loop,
- rspamd_worker_usr1_handler, NULL);
- rspamd_worker_set_signal_handler(SIGUSR2, worker, event_loop,
- rspamd_worker_usr2_handler, NULL);
- }
-
-
- struct ev_loop *
- rspamd_prepare_worker(struct rspamd_worker *worker, const char *name,
- rspamd_accept_handler hdl)
- {
- struct ev_loop *event_loop;
- GList *cur;
- struct rspamd_worker_listen_socket *ls;
- struct rspamd_worker_accept_event *accept_ev;
-
- worker->signal_events = g_hash_table_new_full(g_direct_hash, g_direct_equal,
- NULL, rspamd_sigh_free);
-
- event_loop = ev_loop_new(rspamd_config_ev_backend_get(worker->srv->cfg));
-
- worker->srv->event_loop = event_loop;
-
- rspamd_worker_init_signals(worker, event_loop);
- rspamd_control_worker_add_default_cmd_handlers(worker, event_loop);
- rspamd_worker_heartbeat_start(worker, event_loop);
- rspamd_redis_pool_config(worker->srv->cfg->redis_pool,
- worker->srv->cfg, event_loop);
-
- /* Accept all sockets */
- if (hdl) {
- cur = worker->cf->listen_socks;
-
- while (cur) {
- ls = cur->data;
-
- if (ls->fd != -1) {
- accept_ev = g_malloc0(sizeof(*accept_ev));
- accept_ev->event_loop = event_loop;
- accept_ev->accept_ev.data = worker;
- ev_io_init(&accept_ev->accept_ev, hdl, ls->fd, EV_READ);
- ev_io_start(event_loop, &accept_ev->accept_ev);
-
- DL_APPEND(worker->accept_events, accept_ev);
- }
-
- cur = g_list_next(cur);
- }
- }
-
- return event_loop;
- }
-
- void rspamd_worker_stop_accept(struct rspamd_worker *worker)
- {
- struct rspamd_worker_accept_event *cur, *tmp;
-
- /* Remove all events */
- DL_FOREACH_SAFE(worker->accept_events, cur, tmp)
- {
-
- if (ev_can_stop(&cur->accept_ev)) {
- ev_io_stop(cur->event_loop, &cur->accept_ev);
- }
-
-
- if (ev_can_stop(&cur->throttling_ev)) {
- ev_timer_stop(cur->event_loop, &cur->throttling_ev);
- }
-
- g_free(cur);
- }
-
- /* XXX: we need to do it much later */
- #if 0
- g_hash_table_iter_init (&it, worker->signal_events);
-
- while (g_hash_table_iter_next (&it, &k, &v)) {
- sigh = (struct rspamd_worker_signal_handler *)v;
- g_hash_table_iter_steal (&it);
-
- if (sigh->enabled) {
- event_del (&sigh->ev);
- }
-
- g_free (sigh);
- }
-
- g_hash_table_unref (worker->signal_events);
- #endif
- }
-
- static rspamd_fstring_t *
- rspamd_controller_maybe_compress(struct rspamd_http_connection_entry *entry,
- rspamd_fstring_t *buf, struct rspamd_http_message *msg)
- {
- if (entry->support_gzip) {
- if (rspamd_fstring_gzip(&buf)) {
- rspamd_http_message_add_header(msg, "Content-Encoding", "gzip");
- }
- }
-
- return buf;
- }
-
- void rspamd_controller_send_error(struct rspamd_http_connection_entry *entry,
- int code, const char *error_msg, ...)
- {
- struct rspamd_http_message *msg;
- va_list args;
- rspamd_fstring_t *reply;
-
- msg = rspamd_http_new_message(HTTP_RESPONSE);
-
- va_start(args, error_msg);
- msg->status = rspamd_fstring_new();
- rspamd_vprintf_fstring(&msg->status, error_msg, args);
- va_end(args);
-
- msg->date = time(NULL);
- msg->code = code;
- reply = rspamd_fstring_sized_new(msg->status->len + 16);
- rspamd_printf_fstring(&reply, "{\"error\":\"%V\"}", msg->status);
- rspamd_http_message_set_body_from_fstring_steal(msg,
- rspamd_controller_maybe_compress(entry, reply, msg));
- rspamd_http_connection_reset(entry->conn);
- rspamd_http_router_insert_headers(entry->rt, msg);
- rspamd_http_connection_write_message(entry->conn,
- msg,
- NULL,
- "application/json",
- entry,
- entry->rt->timeout);
- entry->is_reply = TRUE;
- }
-
- void rspamd_controller_send_openmetrics(struct rspamd_http_connection_entry *entry,
- rspamd_fstring_t *str)
- {
- struct rspamd_http_message *msg;
-
- msg = rspamd_http_new_message(HTTP_RESPONSE);
- msg->date = time(NULL);
- msg->code = 200;
- msg->status = rspamd_fstring_new_init("OK", 2);
-
- rspamd_http_message_set_body_from_fstring_steal(msg,
- rspamd_controller_maybe_compress(entry, str, msg));
- rspamd_http_connection_reset(entry->conn);
- rspamd_http_router_insert_headers(entry->rt, msg);
- rspamd_http_connection_write_message(entry->conn,
- msg,
- NULL,
- "application/openmetrics-text; version=1.0.0; charset=utf-8",
- entry,
- entry->rt->timeout);
- entry->is_reply = TRUE;
- }
-
- void rspamd_controller_send_string(struct rspamd_http_connection_entry *entry,
- const char *str)
- {
- struct rspamd_http_message *msg;
- rspamd_fstring_t *reply;
-
- msg = rspamd_http_new_message(HTTP_RESPONSE);
- msg->date = time(NULL);
- msg->code = 200;
- msg->status = rspamd_fstring_new_init("OK", 2);
-
- if (str) {
- reply = rspamd_fstring_new_init(str, strlen(str));
- }
- else {
- reply = rspamd_fstring_new_init("null", 4);
- }
-
- rspamd_http_message_set_body_from_fstring_steal(msg,
- rspamd_controller_maybe_compress(entry, reply, msg));
- rspamd_http_connection_reset(entry->conn);
- rspamd_http_router_insert_headers(entry->rt, msg);
- rspamd_http_connection_write_message(entry->conn,
- msg,
- NULL,
- "application/json",
- entry,
- entry->rt->timeout);
- entry->is_reply = TRUE;
- }
-
- void rspamd_controller_send_ucl(struct rspamd_http_connection_entry *entry,
- ucl_object_t *obj)
- {
- struct rspamd_http_message *msg;
- rspamd_fstring_t *reply;
-
- msg = rspamd_http_new_message(HTTP_RESPONSE);
- msg->date = time(NULL);
- msg->code = 200;
- msg->status = rspamd_fstring_new_init("OK", 2);
- reply = rspamd_fstring_sized_new(BUFSIZ);
- rspamd_ucl_emit_fstring(obj, UCL_EMIT_JSON_COMPACT, &reply);
- rspamd_http_message_set_body_from_fstring_steal(msg,
- rspamd_controller_maybe_compress(entry, reply, msg));
- rspamd_http_connection_reset(entry->conn);
- rspamd_http_router_insert_headers(entry->rt, msg);
- rspamd_http_connection_write_message(entry->conn,
- msg,
- NULL,
- "application/json",
- entry,
- entry->rt->timeout);
- entry->is_reply = TRUE;
- }
-
- static void
- rspamd_worker_drop_priv(struct rspamd_main *rspamd_main)
- {
- if (rspamd_main->is_privileged) {
- if (setgid(rspamd_main->workers_gid) == -1) {
- msg_err_main("cannot setgid to %d (%s), aborting",
- (int) rspamd_main->workers_gid,
- strerror(errno));
- exit(-errno);
- }
-
- if (rspamd_main->cfg->rspamd_user &&
- initgroups(rspamd_main->cfg->rspamd_user,
- rspamd_main->workers_gid) == -1) {
- msg_err_main("initgroups failed (%s), aborting", strerror(errno));
- exit(-errno);
- }
-
- if (setuid(rspamd_main->workers_uid) == -1) {
- msg_err_main("cannot setuid to %d (%s), aborting",
- (int) rspamd_main->workers_uid,
- strerror(errno));
- exit(-errno);
- }
- }
- }
-
- static void
- rspamd_worker_set_limits(struct rspamd_main *rspamd_main,
- struct rspamd_worker_conf *cf)
- {
- struct rlimit rlmt;
-
- if (cf->rlimit_nofile != 0) {
- rlmt.rlim_cur = (rlim_t) cf->rlimit_nofile;
- rlmt.rlim_max = (rlim_t) cf->rlimit_nofile;
-
- if (setrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
- msg_warn_main("cannot set files rlimit: %L, %s",
- cf->rlimit_nofile,
- strerror(errno));
- }
-
- memset(&rlmt, 0, sizeof(rlmt));
-
- if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
- msg_warn_main("cannot get max files rlimit: %HL, %s",
- cf->rlimit_maxcore,
- strerror(errno));
- }
- else {
- msg_info_main("set max file descriptors limit: %HL cur and %HL max",
- (uint64_t) rlmt.rlim_cur,
- (uint64_t) rlmt.rlim_max);
- }
- }
- else {
- /* Just report */
- if (getrlimit(RLIMIT_NOFILE, &rlmt) == -1) {
- msg_warn_main("cannot get max files rlimit: %HL, %s",
- cf->rlimit_maxcore,
- strerror(errno));
- }
- else {
- msg_info_main("use system max file descriptors limit: %HL cur and %HL max",
- (uint64_t) rlmt.rlim_cur,
- (uint64_t) rlmt.rlim_max);
- }
- }
-
- if (rspamd_main->cores_throttling) {
- msg_info_main("disable core files for the new worker as limits are reached");
- rlmt.rlim_cur = 0;
- rlmt.rlim_max = 0;
-
- if (setrlimit(RLIMIT_CORE, &rlmt) == -1) {
- msg_warn_main("cannot disable core dumps: error when setting limits: %s",
- strerror(errno));
- }
- }
- else {
- if (cf->rlimit_maxcore != 0) {
- rlmt.rlim_cur = (rlim_t) cf->rlimit_maxcore;
- rlmt.rlim_max = (rlim_t) cf->rlimit_maxcore;
-
- if (setrlimit(RLIMIT_CORE, &rlmt) == -1) {
- msg_warn_main("cannot set max core size limit: %HL, %s",
- cf->rlimit_maxcore,
- strerror(errno));
- }
-
- /* Ensure that we did it */
- memset(&rlmt, 0, sizeof(rlmt));
-
- if (getrlimit(RLIMIT_CORE, &rlmt) == -1) {
- msg_warn_main("cannot get max core size rlimit: %HL, %s",
- cf->rlimit_maxcore,
- strerror(errno));
- }
- else {
- if (rlmt.rlim_cur != cf->rlimit_maxcore ||
- rlmt.rlim_max != cf->rlimit_maxcore) {
- msg_warn_main("setting of core file limits was unsuccessful: "
- "%HL was wanted, "
- "but we have %HL cur and %HL max",
- cf->rlimit_maxcore,
- (uint64_t) rlmt.rlim_cur,
- (uint64_t) rlmt.rlim_max);
- }
- else {
- msg_info_main("set max core size limit: %HL cur and %HL max",
- (uint64_t) rlmt.rlim_cur,
- (uint64_t) rlmt.rlim_max);
- }
- }
- }
- else {
- /* Just report */
- if (getrlimit(RLIMIT_CORE, &rlmt) == -1) {
- msg_warn_main("cannot get max core size limit: %HL, %s",
- cf->rlimit_maxcore,
- strerror(errno));
- }
- else {
- msg_info_main("use system max core size limit: %HL cur and %HL max",
- (uint64_t) rlmt.rlim_cur,
- (uint64_t) rlmt.rlim_max);
- }
- }
- }
- }
-
- static void
- rspamd_worker_on_term(EV_P_ ev_child *w, int revents)
- {
- struct rspamd_worker *wrk = (struct rspamd_worker *) w->data;
-
- if (wrk->ppid == getpid()) {
- if (wrk->term_handler) {
- wrk->term_handler(EV_A_ w, wrk->srv, wrk);
- }
- else {
- rspamd_check_termination_clause(wrk->srv, wrk, w->rstatus);
- }
- }
- else {
- /* Ignore SIGCHLD for not our children... */
- }
- }
-
- static void
- rspamd_worker_heartbeat_cb(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker *wrk = (struct rspamd_worker *) w->data;
- struct rspamd_srv_command cmd;
-
- memset(&cmd, 0, sizeof(cmd));
- cmd.type = RSPAMD_SRV_HEARTBEAT;
- rspamd_srv_send_command(wrk, EV_A, &cmd, -1, NULL, NULL);
- }
-
- static void
- rspamd_worker_heartbeat_start(struct rspamd_worker *wrk, struct ev_loop *event_loop)
- {
- wrk->hb.heartbeat_ev.data = (void *) wrk;
- ev_timer_init(&wrk->hb.heartbeat_ev, rspamd_worker_heartbeat_cb,
- 0.0, wrk->srv->cfg->heartbeat_interval);
- ev_timer_start(event_loop, &wrk->hb.heartbeat_ev);
- }
-
- static void
- rspamd_main_heartbeat_cb(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker *wrk = (struct rspamd_worker *) w->data;
- double time_from_last = ev_time();
- struct rspamd_main *rspamd_main;
- static struct rspamd_control_command cmd;
- struct tm tm;
- char timebuf[64];
- char usec_buf[16];
- int r;
-
- time_from_last -= wrk->hb.last_event;
- rspamd_main = wrk->srv;
-
- if (wrk->hb.last_event > 0 &&
- time_from_last > 0 &&
- time_from_last >= rspamd_main->cfg->heartbeat_interval * 2) {
-
- rspamd_localtime(wrk->hb.last_event, &tm);
- r = strftime(timebuf, sizeof(timebuf), "%F %H:%M:%S", &tm);
- rspamd_snprintf(usec_buf, sizeof(usec_buf), "%.5f",
- wrk->hb.last_event - (double) (time_t) wrk->hb.last_event);
- rspamd_snprintf(timebuf + r, sizeof(timebuf) - r,
- "%s", usec_buf + 1);
-
- if (wrk->hb.nbeats > 0) {
- /* First time lost event */
- cmd.type = RSPAMD_CONTROL_CHILD_CHANGE;
- cmd.cmd.child_change.what = rspamd_child_offline;
- cmd.cmd.child_change.pid = wrk->pid;
- rspamd_control_broadcast_srv_cmd(rspamd_main, &cmd, wrk->pid);
- msg_warn_main("lost heartbeat from worker type %s with pid %P, "
- "last beat on: %s (%L beats received previously)",
- g_quark_to_string(wrk->type), wrk->pid,
- timebuf,
- wrk->hb.nbeats);
- wrk->hb.nbeats = -1;
- /* TODO: send notify about worker problem */
- }
- else {
- wrk->hb.nbeats--;
- msg_warn_main("lost %L heartbeat from worker type %s with pid %P, "
- "last beat on: %s",
- -(wrk->hb.nbeats),
- g_quark_to_string(wrk->type),
- wrk->pid,
- timebuf);
-
- if (rspamd_main->cfg->heartbeats_loss_max > 0 &&
- -(wrk->hb.nbeats) >= rspamd_main->cfg->heartbeats_loss_max) {
-
-
- if (-(wrk->hb.nbeats) > rspamd_main->cfg->heartbeats_loss_max + 1) {
- msg_err_main("force kill worker type %s with pid %P, "
- "last beat on: %s; %L heartbeat lost",
- g_quark_to_string(wrk->type),
- wrk->pid,
- timebuf,
- -(wrk->hb.nbeats));
- kill(wrk->pid, SIGKILL);
- }
- else {
- msg_err_main("terminate worker type %s with pid %P, "
- "last beat on: %s; %L heartbeat lost",
- g_quark_to_string(wrk->type),
- wrk->pid,
- timebuf,
- -(wrk->hb.nbeats));
- kill(wrk->pid, SIGTERM);
- }
- }
- }
- }
- else if (wrk->hb.nbeats < 0) {
- rspamd_localtime(wrk->hb.last_event, &tm);
- r = strftime(timebuf, sizeof(timebuf), "%F %H:%M:%S", &tm);
- rspamd_snprintf(usec_buf, sizeof(usec_buf), "%.5f",
- wrk->hb.last_event - (double) (time_t) wrk->hb.last_event);
- rspamd_snprintf(timebuf + r, sizeof(timebuf) - r,
- "%s", usec_buf + 1);
-
- cmd.type = RSPAMD_CONTROL_CHILD_CHANGE;
- cmd.cmd.child_change.what = rspamd_child_online;
- cmd.cmd.child_change.pid = wrk->pid;
- rspamd_control_broadcast_srv_cmd(rspamd_main, &cmd, wrk->pid);
- msg_info_main("received heartbeat from worker type %s with pid %P, "
- "last beat on: %s (%L beats lost previously)",
- g_quark_to_string(wrk->type), wrk->pid,
- timebuf,
- -(wrk->hb.nbeats));
- wrk->hb.nbeats = 1;
- /* TODO: send notify about worker restoration */
- }
- }
-
- static void
- rspamd_main_heartbeat_start(struct rspamd_worker *wrk, struct ev_loop *event_loop)
- {
- wrk->hb.heartbeat_ev.data = (void *) wrk;
- ev_timer_init(&wrk->hb.heartbeat_ev, rspamd_main_heartbeat_cb,
- 0.0, wrk->srv->cfg->heartbeat_interval * 2);
- ev_timer_start(event_loop, &wrk->hb.heartbeat_ev);
- }
-
- static bool
- rspamd_maybe_reuseport_socket(struct rspamd_worker_listen_socket *ls)
- {
- if (ls->is_systemd) {
- /* No need to reuseport */
- return true;
- }
-
- if (ls->fd != -1 && rspamd_inet_address_get_af(ls->addr) == AF_UNIX) {
- /* Just try listen */
-
- if (listen(ls->fd, -1) == -1) {
- return false;
- }
-
- return true;
- }
-
- #if defined(SO_REUSEPORT) && defined(SO_REUSEADDR) && defined(LINUX)
- int nfd = -1;
-
- if (ls->type == RSPAMD_WORKER_SOCKET_UDP) {
- nfd = rspamd_inet_address_listen(ls->addr,
- (ls->type == RSPAMD_WORKER_SOCKET_UDP ? SOCK_DGRAM : SOCK_STREAM),
- RSPAMD_INET_ADDRESS_LISTEN_ASYNC | RSPAMD_INET_ADDRESS_LISTEN_REUSEPORT,
- -1);
-
- if (nfd == -1) {
- msg_warn("cannot create reuseport listen socket for %d: %s",
- ls->fd, strerror(errno));
- nfd = ls->fd;
- }
- else {
- if (ls->fd != -1) {
- close(ls->fd);
- }
- ls->fd = nfd;
- nfd = -1;
- }
- }
- else {
- /*
- * Reuseport is broken with the current architecture, so it is easier not
- * to use it at all
- */
- nfd = ls->fd;
- }
- #endif
-
- #if 0
- /* This needed merely if we have reuseport for tcp, but for now it is disabled */
- /* This means that we have an fd with no listening enabled */
- if (nfd != -1) {
- if (ls->type == RSPAMD_WORKER_SOCKET_TCP) {
- if (listen (nfd, -1) == -1) {
- return false;
- }
- }
- }
- #endif
-
- return true;
- }
-
- /**
- * Handles worker after fork returned zero
- * @param wrk
- * @param rspamd_main
- * @param cf
- * @param listen_sockets
- */
- static void __attribute__((noreturn))
- rspamd_handle_child_fork(struct rspamd_worker *wrk,
- struct rspamd_main *rspamd_main,
- struct rspamd_worker_conf *cf,
- GHashTable *listen_sockets)
- {
- int rc;
- struct rlimit rlim;
-
- /* Update pid for logging */
- rspamd_log_on_fork(cf->type, rspamd_main->cfg, rspamd_main->logger);
- wrk->pid = getpid();
-
- /* Init PRNG after fork */
- rc = ottery_init(rspamd_main->cfg->libs_ctx->ottery_cfg);
- if (rc != OTTERY_ERR_NONE) {
- msg_err_main("cannot initialize PRNG: %d", rc);
- abort();
- }
-
- rspamd_random_seed_fast();
- #ifdef HAVE_EVUTIL_RNG_INIT
- evutil_secure_rng_init();
- #endif
-
- /*
- * Libev stores all signals in a global table, so
- * previous handlers must be explicitly detached and forgotten
- * before starting a new loop
- */
- ev_signal_stop(rspamd_main->event_loop, &rspamd_main->int_ev);
- ev_signal_stop(rspamd_main->event_loop, &rspamd_main->term_ev);
- ev_signal_stop(rspamd_main->event_loop, &rspamd_main->hup_ev);
- ev_signal_stop(rspamd_main->event_loop, &rspamd_main->usr1_ev);
- /* Remove the inherited event base */
- ev_loop_destroy(rspamd_main->event_loop);
- rspamd_main->event_loop = NULL;
-
- /* Close unused sockets */
- GHashTableIter it;
- gpointer k, v;
-
-
- g_hash_table_iter_init(&it, listen_sockets);
-
- /*
- * Close listen sockets of not our process (inherited from other forks)
- */
- while (g_hash_table_iter_next(&it, &k, &v)) {
- GList *elt = (GList *) v;
- GList *our = cf->listen_socks;
-
- if (g_list_position(our, elt) == -1) {
- GList *cur = elt;
-
- while (cur) {
- struct rspamd_worker_listen_socket *ls =
- (struct rspamd_worker_listen_socket *) cur->data;
-
- if (ls->fd != -1 && close(ls->fd) == -1) {
- msg_err("cannot close fd %d (addr = %s): %s",
- ls->fd,
- rspamd_inet_address_to_string_pretty(ls->addr),
- strerror(errno));
- }
-
- ls->fd = -1;
-
- cur = g_list_next(cur);
- }
- }
- }
-
- /* Reuseport before dropping privs */
- GList *cur = cf->listen_socks;
-
- while (cur) {
- struct rspamd_worker_listen_socket *ls =
- (struct rspamd_worker_listen_socket *) cur->data;
-
- if (!rspamd_maybe_reuseport_socket(ls)) {
- msg_err("cannot listen on socket %s: %s",
- rspamd_inet_address_to_string_pretty(ls->addr),
- strerror(errno));
- }
-
- cur = g_list_next(cur);
- }
-
- /* Drop privileges */
- rspamd_worker_drop_priv(rspamd_main);
- /* Set limits */
- rspamd_worker_set_limits(rspamd_main, cf);
- /* Re-set stack limit */
- getrlimit(RLIMIT_STACK, &rlim);
- rlim.rlim_cur = 100 * 1024 * 1024;
- rlim.rlim_max = rlim.rlim_cur;
- setrlimit(RLIMIT_STACK, &rlim);
-
- if (cf->bind_conf) {
- rspamd_setproctitle("%s process (%s)", cf->worker->name,
- cf->bind_conf->bind_line);
- }
- else {
- rspamd_setproctitle("%s process", cf->worker->name);
- }
-
- if (rspamd_main->pfh) {
- rspamd_pidfile_close(rspamd_main->pfh);
- }
-
- if (rspamd_main->cfg->log_silent_workers) {
- rspamd_log_set_log_level(rspamd_main->logger, G_LOG_LEVEL_MESSAGE);
- }
-
- wrk->start_time = rspamd_get_calendar_ticks();
-
- if (cf->bind_conf) {
- GString *listen_conf_stringified = g_string_new(NULL);
- struct rspamd_worker_bind_conf *cur_conf;
-
- LL_FOREACH(cf->bind_conf, cur_conf)
- {
- if (cur_conf->next) {
- rspamd_printf_gstring(listen_conf_stringified, "%s, ",
- cur_conf->bind_line);
- }
- else {
- rspamd_printf_gstring(listen_conf_stringified, "%s",
- cur_conf->bind_line);
- }
- }
-
- msg_info_main("starting %s process %P (%d); listen on: %v",
- cf->worker->name,
- getpid(), wrk->index, listen_conf_stringified);
- g_string_free(listen_conf_stringified, TRUE);
- }
- else {
- msg_info_main("starting %s process %P (%d); no listen",
- cf->worker->name,
- getpid(), wrk->index);
- }
- /* Close parent part of socketpair */
- close(wrk->control_pipe[0]);
- close(wrk->srv_pipe[0]);
- /*
- * Read comments in `rspamd_handle_main_fork` for details why these channel
- * is blocking.
- */
- rspamd_socket_nonblocking(wrk->control_pipe[1]);
- #if 0
- rspamd_socket_nonblocking (wrk->srv_pipe[1]);
- #endif
- rspamd_main->cfg->cur_worker = wrk;
- /* Execute worker (this function should not return normally!) */
- cf->worker->worker_start_func(wrk);
- /* To distinguish from normal termination */
- exit(EXIT_FAILURE);
- }
-
- static void
- rspamd_handle_main_fork(struct rspamd_worker *wrk,
- struct rspamd_main *rspamd_main,
- struct rspamd_worker_conf *cf,
- struct ev_loop *ev_base)
- {
- /* Close worker part of socketpair */
- close(wrk->control_pipe[1]);
- close(wrk->srv_pipe[1]);
-
- /*
- * There are no reasons why control pipes are blocking: the messages
- * there are rare and are strictly bounded by command sizes, so if we block
- * on some pipe, it is ok, as we still poll that for all operations.
- * It is also impossible to block on writing in normal conditions.
- * And if the conditions are not normal, e.g. a worker is unresponsive, then
- * we can safely think that the non-blocking behaviour as it is implemented
- * currently will not make things better, as it would lead to incomplete
- * reads/writes that are not handled anyhow and are totally broken from the
- * beginning.
- */
- #if 0
- rspamd_socket_nonblocking (wrk->srv_pipe[0]);
- #endif
- rspamd_socket_nonblocking(wrk->control_pipe[0]);
-
- rspamd_srv_start_watching(rspamd_main, wrk, ev_base);
- /* Child event */
- wrk->cld_ev.data = wrk;
- ev_child_init(&wrk->cld_ev, rspamd_worker_on_term, wrk->pid, 0);
- ev_child_start(rspamd_main->event_loop, &wrk->cld_ev);
- /* Heartbeats */
- rspamd_main_heartbeat_start(wrk, rspamd_main->event_loop);
- /* Insert worker into worker's table, pid is index */
- g_hash_table_insert(rspamd_main->workers,
- GSIZE_TO_POINTER(wrk->pid), wrk);
-
- #if defined(SO_REUSEPORT) && defined(SO_REUSEADDR) && defined(LINUX)
- /*
- * Close listen sockets in the main process once a child is handling them,
- * if we have reuseport
- */
- GList *cur = cf->listen_socks;
-
- while (cur) {
- struct rspamd_worker_listen_socket *ls =
- (struct rspamd_worker_listen_socket *) cur->data;
-
- if (ls->fd != -1 && ls->type == RSPAMD_WORKER_SOCKET_UDP) {
- close(ls->fd);
- ls->fd = -1;
- }
-
- cur = g_list_next(cur);
- }
- #endif
- }
-
- #ifndef SOCK_SEQPACKET
- #define SOCK_SEQPACKET SOCK_DGRAM
- #endif
- struct rspamd_worker *
- rspamd_fork_worker(struct rspamd_main *rspamd_main,
- struct rspamd_worker_conf *cf,
- unsigned int index,
- struct ev_loop *ev_base,
- rspamd_worker_term_cb term_handler,
- GHashTable *listen_sockets)
- {
- struct rspamd_worker *wrk;
-
- /* Starting worker process */
- wrk = (struct rspamd_worker *) g_malloc0(sizeof(struct rspamd_worker));
-
- if (!rspamd_socketpair(wrk->control_pipe, SOCK_SEQPACKET)) {
- msg_err("socketpair failure: %s", strerror(errno));
- rspamd_hard_terminate(rspamd_main);
- }
-
- if (!rspamd_socketpair(wrk->srv_pipe, SOCK_SEQPACKET)) {
- msg_err("socketpair failure: %s", strerror(errno));
- rspamd_hard_terminate(rspamd_main);
- }
-
- if (cf->bind_conf) {
- msg_info_main("prepare to fork process %s (%d); listen on: %s",
- cf->worker->name,
- index, cf->bind_conf->name);
- }
- else {
- msg_info_main("prepare to fork process %s (%d), no bind socket",
- cf->worker->name,
- index);
- }
-
- wrk->srv = rspamd_main;
- wrk->type = cf->type;
- wrk->cf = cf;
- wrk->flags = cf->worker->flags;
- REF_RETAIN(cf);
- wrk->index = index;
- wrk->ctx = cf->ctx;
- wrk->ppid = getpid();
- wrk->pid = fork();
- wrk->cores_throttled = rspamd_main->cores_throttling;
- wrk->term_handler = term_handler;
- wrk->control_events_pending = g_hash_table_new_full(g_direct_hash, g_direct_equal,
- NULL, rspamd_pending_control_free);
-
- switch (wrk->pid) {
- case 0:
- rspamd_current_worker = wrk;
- rspamd_handle_child_fork(wrk, rspamd_main, cf, listen_sockets);
- break;
- case -1:
- msg_err_main("cannot fork main process: %s", strerror(errno));
-
- if (rspamd_main->pfh) {
- rspamd_pidfile_remove(rspamd_main->pfh);
- }
-
- rspamd_hard_terminate(rspamd_main);
- break;
- default:
- rspamd_handle_main_fork(wrk, rspamd_main, cf, ev_base);
- break;
- }
-
- return wrk;
- }
-
- void rspamd_worker_block_signals(void)
- {
- sigset_t set;
-
- sigemptyset(&set);
- sigaddset(&set, SIGTERM);
- sigaddset(&set, SIGINT);
- sigaddset(&set, SIGHUP);
- sigaddset(&set, SIGUSR1);
- sigaddset(&set, SIGUSR2);
- sigprocmask(SIG_BLOCK, &set, NULL);
- }
-
- void rspamd_worker_unblock_signals(void)
- {
- sigset_t set;
-
- sigemptyset(&set);
- sigaddset(&set, SIGTERM);
- sigaddset(&set, SIGINT);
- sigaddset(&set, SIGHUP);
- sigaddset(&set, SIGUSR1);
- sigaddset(&set, SIGUSR2);
- sigprocmask(SIG_UNBLOCK, &set, NULL);
- }
-
- void rspamd_hard_terminate(struct rspamd_main *rspamd_main)
- {
- GHashTableIter it;
- gpointer k, v;
- struct rspamd_worker *w;
- sigset_t set;
-
- /* Block all signals */
- sigemptyset(&set);
- sigaddset(&set, SIGTERM);
- sigaddset(&set, SIGINT);
- sigaddset(&set, SIGHUP);
- sigaddset(&set, SIGUSR1);
- sigaddset(&set, SIGUSR2);
- sigaddset(&set, SIGCHLD);
- sigprocmask(SIG_BLOCK, &set, NULL);
-
- /* We need to terminate all workers that might be already spawned */
- rspamd_worker_block_signals();
- g_hash_table_iter_init(&it, rspamd_main->workers);
-
- while (g_hash_table_iter_next(&it, &k, &v)) {
- w = v;
- msg_err_main("kill worker %P as Rspamd is terminating due to "
- "an unrecoverable error",
- w->pid);
- kill(w->pid, SIGKILL);
- }
-
- msg_err_main("shutting down Rspamd due to fatal error");
-
- rspamd_log_close(rspamd_main->logger);
- exit(EXIT_FAILURE);
- }
-
- gboolean
- rspamd_worker_is_scanner(struct rspamd_worker *w)
- {
-
- if (w) {
- return !!(w->flags & RSPAMD_WORKER_SCANNER);
- }
-
- return FALSE;
- }
-
- gboolean
- rspamd_worker_is_primary_controller(struct rspamd_worker *w)
- {
-
- if (w) {
- return !!(w->flags & RSPAMD_WORKER_CONTROLLER) && w->index == 0;
- }
-
- return FALSE;
- }
-
- gboolean
- rspamd_worker_check_controller_presence(struct rspamd_worker *w)
- {
- if (w->index == 0) {
- GQuark our_type = w->type;
- gboolean controller_seen = FALSE;
- GList *cur;
-
- enum {
- low_priority_worker,
- high_priority_worker
- } our_priority;
-
- if (our_type == g_quark_from_static_string("rspamd_proxy")) {
- our_priority = low_priority_worker;
- }
- else if (our_type == g_quark_from_static_string("normal")) {
- our_priority = high_priority_worker;
- }
- else {
- msg_err("function is called for a wrong worker type: %s", g_quark_to_string(our_type));
- return FALSE;
- }
-
- cur = w->srv->cfg->workers;
-
- while (cur) {
- struct rspamd_worker_conf *cf;
-
- cf = (struct rspamd_worker_conf *) cur->data;
-
- if (our_priority == low_priority_worker) {
- if ((cf->type == g_quark_from_static_string("controller")) ||
- (cf->type == g_quark_from_static_string("normal"))) {
-
- if (cf->enabled && cf->count >= 0) {
- controller_seen = TRUE;
- break;
- }
- }
- }
- else {
- if (cf->type == g_quark_from_static_string("controller")) {
- if (cf->enabled && cf->count >= 0) {
- controller_seen = TRUE;
- break;
- }
- }
- }
-
- cur = g_list_next(cur);
- }
-
- if (!controller_seen) {
- msg_info("no controller or normal workers defined, execute "
- "controller periodics in this worker");
- w->flags |= RSPAMD_WORKER_CONTROLLER;
- return TRUE;
- }
- }
-
- return FALSE;
- }
-
- struct rspamd_worker_session_elt {
- void *ptr;
- unsigned int *pref;
- const char *tag;
- time_t when;
- };
-
- struct rspamd_worker_session_cache {
- struct ev_loop *ev_base;
- GHashTable *cache;
- struct rspamd_config *cfg;
- struct ev_timer periodic;
- };
-
- static int
- rspamd_session_cache_sort_cmp(gconstpointer pa, gconstpointer pb)
- {
- const struct rspamd_worker_session_elt
- *e1 = *(const struct rspamd_worker_session_elt **) pa,
- *e2 = *(const struct rspamd_worker_session_elt **) pb;
-
- return e2->when < e1->when;
- }
-
- static void
- rspamd_sessions_cache_periodic(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker_session_cache *c =
- (struct rspamd_worker_session_cache *) w->data;
- GHashTableIter it;
- char timebuf[32];
- gpointer k, v;
- struct rspamd_worker_session_elt *elt;
- struct tm tms;
- GPtrArray *res;
- unsigned int i;
-
- if (g_hash_table_size(c->cache) > c->cfg->max_sessions_cache) {
- res = g_ptr_array_sized_new(g_hash_table_size(c->cache));
- g_hash_table_iter_init(&it, c->cache);
-
- while (g_hash_table_iter_next(&it, &k, &v)) {
- g_ptr_array_add(res, v);
- }
-
- msg_err("sessions cache is overflowed %d elements where %d is limit",
- (int) res->len, (int) c->cfg->max_sessions_cache);
- g_ptr_array_sort(res, rspamd_session_cache_sort_cmp);
-
- PTR_ARRAY_FOREACH(res, i, elt)
- {
- rspamd_localtime(elt->when, &tms);
- strftime(timebuf, sizeof(timebuf), "%F %H:%M:%S", &tms);
-
- msg_warn("redundant session; ptr: %p, "
- "tag: %s, refcount: %d, time: %s",
- elt->ptr, elt->tag ? elt->tag : "unknown",
- elt->pref ? *elt->pref : 0,
- timebuf);
- }
- }
-
- ev_timer_again(EV_A_ w);
- }
-
- void *
- rspamd_worker_session_cache_new(struct rspamd_worker *w,
- struct ev_loop *ev_base)
- {
- struct rspamd_worker_session_cache *c;
- static const double periodic_interval = 60.0;
-
- c = g_malloc0(sizeof(*c));
- c->ev_base = ev_base;
- c->cache = g_hash_table_new_full(g_direct_hash, g_direct_equal,
- NULL, g_free);
- c->cfg = w->srv->cfg;
- c->periodic.data = c;
- ev_timer_init(&c->periodic, rspamd_sessions_cache_periodic, periodic_interval,
- periodic_interval);
- ev_timer_start(ev_base, &c->periodic);
-
- return c;
- }
-
-
- void rspamd_worker_session_cache_add(void *cache, const char *tag,
- unsigned int *pref, void *ptr)
- {
- struct rspamd_worker_session_cache *c = cache;
- struct rspamd_worker_session_elt *elt;
-
- elt = g_malloc0(sizeof(*elt));
- elt->pref = pref;
- elt->ptr = ptr;
- elt->tag = tag;
- elt->when = time(NULL);
-
- g_hash_table_insert(c->cache, elt->ptr, elt);
- }
-
-
- void rspamd_worker_session_cache_remove(void *cache, void *ptr)
- {
- struct rspamd_worker_session_cache *c = cache;
-
- g_hash_table_remove(c->cache, ptr);
- }
-
- static void
- rspamd_worker_monitored_on_change(struct rspamd_monitored_ctx *ctx,
- struct rspamd_monitored *m, gboolean alive,
- void *ud)
- {
- struct rspamd_worker *worker = ud;
- struct rspamd_config *cfg = worker->srv->cfg;
- struct ev_loop *ev_base;
- unsigned char tag[RSPAMD_MONITORED_TAG_LEN];
- static struct rspamd_srv_command srv_cmd;
-
- rspamd_monitored_get_tag(m, tag);
- ev_base = rspamd_monitored_ctx_get_ev_base(ctx);
- memset(&srv_cmd, 0, sizeof(srv_cmd));
- srv_cmd.type = RSPAMD_SRV_MONITORED_CHANGE;
- rspamd_strlcpy(srv_cmd.cmd.monitored_change.tag, tag,
- sizeof(srv_cmd.cmd.monitored_change.tag));
- srv_cmd.cmd.monitored_change.alive = alive;
- srv_cmd.cmd.monitored_change.sender = getpid();
- msg_info_config("broadcast monitored update for %s: %s",
- srv_cmd.cmd.monitored_change.tag, alive ? "alive" : "dead");
-
- rspamd_srv_send_command(worker, ev_base, &srv_cmd, -1, NULL, NULL);
- }
-
- void rspamd_worker_init_monitored(struct rspamd_worker *worker,
- struct ev_loop *ev_base,
- struct rspamd_dns_resolver *resolver)
- {
- rspamd_monitored_ctx_config(worker->srv->cfg->monitored_ctx,
- worker->srv->cfg, ev_base, resolver->r,
- rspamd_worker_monitored_on_change, worker);
- }
-
- #ifdef HAVE_SA_SIGINFO
-
- static struct rspamd_main *saved_main = NULL;
- static gboolean
- rspamd_crash_propagate(gpointer key, gpointer value, gpointer unused)
- {
- struct rspamd_worker *w = value;
-
- /* Kill children softly */
- kill(w->pid, SIGTERM);
-
- return TRUE;
- }
-
- #ifdef BACKWARD_ENABLE
- /* See backtrace.cxx */
- extern void rspamd_print_crash(void);
- #endif
-
- static void
- rspamd_crash_sig_handler(int sig, siginfo_t *info, void *ctx)
- {
- struct sigaction sa;
- ucontext_t *uap = ctx;
- pid_t pid;
-
- pid = getpid();
- msg_err("caught fatal signal %d(%s), "
- "pid: %P, trace: ",
- sig, strsignal(sig), pid);
- (void) uap;
- #ifdef BACKWARD_ENABLE
- rspamd_print_crash();
- #endif
- msg_err("please see Rspamd FAQ to learn how to dump core files and how to "
- "fill a bug report");
-
- if (saved_main) {
- if (pid == saved_main->pid) {
- /*
- * Main process has crashed, propagate crash further to trigger
- * monitoring alerts and mass panic
- */
- g_hash_table_foreach_remove(saved_main->workers,
- rspamd_crash_propagate, NULL);
- }
- }
-
- /*
- * Invoke signal with the default handler
- */
- sigemptyset(&sa.sa_mask);
- sa.sa_handler = SIG_DFL;
- sa.sa_flags = 0;
- sigaction(sig, &sa, NULL);
- kill(pid, sig);
- }
- #endif
-
- RSPAMD_NO_SANITIZE void
- rspamd_set_crash_handler(struct rspamd_main *rspamd_main)
- {
- #ifdef HAVE_SA_SIGINFO
- struct sigaction sa;
-
- #ifdef HAVE_SIGALTSTACK
- void *stack_mem;
- stack_t ss;
- memset(&ss, 0, sizeof ss);
-
- ss.ss_size = MAX(SIGSTKSZ, 8192 * 4);
- stack_mem = g_malloc0(ss.ss_size);
- ss.ss_sp = stack_mem;
- sigaltstack(&ss, NULL);
- #endif
- saved_main = rspamd_main;
- sigemptyset(&sa.sa_mask);
- sa.sa_sigaction = &rspamd_crash_sig_handler;
- sa.sa_flags = SA_RESTART | SA_SIGINFO | SA_ONSTACK;
- sigaction(SIGSEGV, &sa, NULL);
- sigaction(SIGBUS, &sa, NULL);
- sigaction(SIGABRT, &sa, NULL);
- sigaction(SIGFPE, &sa, NULL);
- sigaction(SIGSYS, &sa, NULL);
- #endif
- }
-
- RSPAMD_NO_SANITIZE void rspamd_unset_crash_handler(struct rspamd_main *unused_)
- {
- #ifdef HAVE_SIGALTSTACK
- int ret;
- stack_t ss;
- ret = sigaltstack(NULL, &ss);
-
- if (ret != -1) {
- if (ss.ss_size > 0 && ss.ss_sp) {
- g_free(ss.ss_sp);
- }
-
- ss.ss_size = 0;
- ss.ss_sp = NULL;
- #ifdef SS_DISABLE
- ss.ss_flags |= SS_DISABLE;
- #endif
- sigaltstack(&ss, NULL);
- }
- #endif
- }
-
- static void
- rspamd_enable_accept_event(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_worker_accept_event *ac_ev =
- (struct rspamd_worker_accept_event *) w->data;
-
- ev_timer_stop(EV_A_ w);
- ev_io_start(EV_A_ & ac_ev->accept_ev);
- }
-
- void rspamd_worker_throttle_accept_events(int sock, void *data)
- {
- struct rspamd_worker_accept_event *head, *cur;
- const double throttling = 0.5;
-
- head = (struct rspamd_worker_accept_event *) data;
-
- DL_FOREACH(head, cur)
- {
-
- ev_io_stop(cur->event_loop, &cur->accept_ev);
- cur->throttling_ev.data = cur;
- ev_timer_init(&cur->throttling_ev, rspamd_enable_accept_event,
- throttling, 0.0);
- ev_timer_start(cur->event_loop, &cur->throttling_ev);
- }
- }
-
- gboolean
- rspamd_check_termination_clause(struct rspamd_main *rspamd_main,
- struct rspamd_worker *wrk,
- int res)
- {
- gboolean need_refork = TRUE;
-
- if (wrk->state != rspamd_worker_state_running || rspamd_main->wanna_die ||
- (wrk->flags & RSPAMD_WORKER_OLD_CONFIG)) {
- /* Do not refork workers that are intended to be terminated */
- need_refork = FALSE;
- }
-
- if (WIFEXITED(res) && WEXITSTATUS(res) == 0) {
- /* Normal worker termination, do not fork one more */
-
- if (wrk->flags & RSPAMD_WORKER_OLD_CONFIG) {
- /* Never re-fork old workers */
- msg_info_main("%s process %P terminated normally",
- g_quark_to_string(wrk->type),
- wrk->pid);
- need_refork = FALSE;
- }
- else {
- if (wrk->hb.nbeats < 0 && rspamd_main->cfg->heartbeats_loss_max > 0 &&
- -(wrk->hb.nbeats) >= rspamd_main->cfg->heartbeats_loss_max) {
- msg_info_main("%s process %P terminated normally, but lost %L "
- "heartbeats, refork it",
- g_quark_to_string(wrk->type),
- wrk->pid,
- -(wrk->hb.nbeats));
- need_refork = TRUE;
- }
- else {
- msg_info_main("%s process %P terminated normally",
- g_quark_to_string(wrk->type),
- wrk->pid);
- need_refork = FALSE;
- }
- }
- }
- else {
- if (WIFSIGNALED(res)) {
- #ifdef WCOREDUMP
- if (WCOREDUMP(res)) {
- msg_warn_main(
- "%s process %P terminated abnormally by signal: %s"
- " and created core file; please see Rspamd FAQ "
- "to learn how to extract data from core file and "
- "fill a bug report",
- g_quark_to_string(wrk->type),
- wrk->pid,
- g_strsignal(WTERMSIG(res)));
- }
- else {
- #ifdef HAVE_SYS_RESOURCE_H
- struct rlimit rlmt;
- (void) getrlimit(RLIMIT_CORE, &rlmt);
-
- msg_warn_main(
- "%s process %P terminated abnormally with exit code %d by "
- "signal: %s"
- " but NOT created core file (throttled=%s); "
- "core file limits: %L current, %L max",
- g_quark_to_string(wrk->type),
- wrk->pid,
- WEXITSTATUS(res),
- g_strsignal(WTERMSIG(res)),
- wrk->cores_throttled ? "yes" : "no",
- (int64_t) rlmt.rlim_cur,
- (int64_t) rlmt.rlim_max);
- #else
- msg_warn_main(
- "%s process %P terminated abnormally with exit code %d by signal: %s"
- " but NOT created core file (throttled=%s); ",
- g_quark_to_string(wrk->type),
- wrk->pid, WEXITSTATUS(res),
- g_strsignal(WTERMSIG(res)),
- wrk->cores_throttled ? "yes" : "no");
- #endif
- }
- #else
- msg_warn_main(
- "%s process %P terminated abnormally with exit code %d by signal: %s",
- g_quark_to_string(wrk->type),
- wrk->pid, WEXITSTATUS(res),
- g_strsignal(WTERMSIG(res)));
- #endif
- if (WTERMSIG(res) == SIGUSR2) {
- /*
- * It is actually race condition when not started process
- * has been requested to be reloaded.
- *
- * We shouldn't refork on this
- */
- need_refork = FALSE;
- }
- }
- else {
- msg_warn_main("%s process %P terminated abnormally "
- "(but it was not killed by a signal) "
- "with exit code %d",
- g_quark_to_string(wrk->type),
- wrk->pid,
- WEXITSTATUS(res));
- }
- }
-
- return need_refork;
- }
-
- #ifdef WITH_HYPERSCAN
- gboolean
- rspamd_worker_hyperscan_ready(struct rspamd_main *rspamd_main,
- struct rspamd_worker *worker, int fd,
- int attached_fd,
- struct rspamd_control_command *cmd,
- gpointer ud)
- {
- struct rspamd_control_reply rep;
- struct rspamd_re_cache *cache = worker->srv->cfg->re_cache;
-
- memset(&rep, 0, sizeof(rep));
- rep.type = RSPAMD_CONTROL_HYPERSCAN_LOADED;
-
- if (rspamd_re_cache_is_hs_loaded(cache) != RSPAMD_HYPERSCAN_LOADED_FULL ||
- cmd->cmd.hs_loaded.forced) {
-
- msg_info("loading hyperscan expressions after receiving compilation "
- "notice: %s",
- (rspamd_re_cache_is_hs_loaded(cache) != RSPAMD_HYPERSCAN_LOADED_FULL) ? "new db" : "forced update");
- rep.reply.hs_loaded.status = rspamd_re_cache_load_hyperscan(
- worker->srv->cfg->re_cache, cmd->cmd.hs_loaded.cache_dir, false);
- }
-
- if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {
- msg_err("cannot write reply to the control socket: %s",
- strerror(errno));
- }
-
- return TRUE;
- }
- #endif /* With Hyperscan */
-
- gboolean
- rspamd_worker_check_context(gpointer ctx, uint64_t magic)
- {
- struct rspamd_abstract_worker_ctx *actx = (struct rspamd_abstract_worker_ctx *) ctx;
-
- return actx->magic == magic;
- }
-
- static gboolean
- rspamd_worker_log_pipe_handler(struct rspamd_main *rspamd_main,
- struct rspamd_worker *worker, int fd,
- int attached_fd,
- struct rspamd_control_command *cmd,
- gpointer ud)
- {
- struct rspamd_config *cfg = ud;
- struct rspamd_worker_log_pipe *lp;
- struct rspamd_control_reply rep;
-
- memset(&rep, 0, sizeof(rep));
- rep.type = RSPAMD_CONTROL_LOG_PIPE;
-
- if (attached_fd != -1) {
- lp = g_malloc0(sizeof(*lp));
- lp->fd = attached_fd;
- lp->type = cmd->cmd.log_pipe.type;
-
- DL_APPEND(cfg->log_pipes, lp);
- msg_info("added new log pipe");
- }
- else {
- rep.reply.log_pipe.status = ENOENT;
- msg_err("cannot attach log pipe: invalid fd");
- }
-
- if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {
- msg_err("cannot write reply to the control socket: %s",
- strerror(errno));
- }
-
- return TRUE;
- }
-
- static gboolean
- rspamd_worker_monitored_handler(struct rspamd_main *rspamd_main,
- struct rspamd_worker *worker, int fd,
- int attached_fd,
- struct rspamd_control_command *cmd,
- gpointer ud)
- {
- struct rspamd_control_reply rep;
- struct rspamd_monitored *m;
- struct rspamd_monitored_ctx *mctx = worker->srv->cfg->monitored_ctx;
- struct rspamd_config *cfg = ud;
-
- memset(&rep, 0, sizeof(rep));
- rep.type = RSPAMD_CONTROL_MONITORED_CHANGE;
-
- if (cmd->cmd.monitored_change.sender != getpid()) {
- m = rspamd_monitored_by_tag(mctx, cmd->cmd.monitored_change.tag);
-
- if (m != NULL) {
- rspamd_monitored_set_alive(m, cmd->cmd.monitored_change.alive);
- rep.reply.monitored_change.status = 1;
- msg_info_config("updated monitored status for %s: %s",
- cmd->cmd.monitored_change.tag,
- cmd->cmd.monitored_change.alive ? "alive" : "dead");
- }
- else {
- msg_err("cannot find monitored by tag: %*s", 32,
- cmd->cmd.monitored_change.tag);
- rep.reply.monitored_change.status = 0;
- }
- }
-
- if (write(fd, &rep, sizeof(rep)) != sizeof(rep)) {
- msg_err("cannot write reply to the control socket: %s",
- strerror(errno));
- }
-
- return TRUE;
- }
-
- void rspamd_worker_init_scanner(struct rspamd_worker *worker,
- struct ev_loop *ev_base,
- struct rspamd_dns_resolver *resolver,
- struct rspamd_lang_detector **plang_det)
- {
- rspamd_stat_init(worker->srv->cfg, ev_base);
- #ifdef WITH_HYPERSCAN
- rspamd_control_worker_add_cmd_handler(worker,
- RSPAMD_CONTROL_HYPERSCAN_LOADED,
- rspamd_worker_hyperscan_ready,
- NULL);
- #endif
- rspamd_control_worker_add_cmd_handler(worker,
- RSPAMD_CONTROL_LOG_PIPE,
- rspamd_worker_log_pipe_handler,
- worker->srv->cfg);
- rspamd_control_worker_add_cmd_handler(worker,
- RSPAMD_CONTROL_MONITORED_CHANGE,
- rspamd_worker_monitored_handler,
- worker->srv->cfg);
-
- *plang_det = worker->srv->cfg->lang_det;
- }
-
- void rspamd_controller_store_saved_stats(struct rspamd_main *rspamd_main,
- struct rspamd_config *cfg)
- {
- struct rspamd_stat *stat;
- ucl_object_t *top, *sub;
- struct ucl_emitter_functions *efuncs;
- int i, fd;
- FILE *fp;
- char fpath[PATH_MAX];
-
- if (cfg->stats_file == NULL) {
- return;
- }
-
- rspamd_snprintf(fpath, sizeof(fpath), "%s.XXXXXXXX", cfg->stats_file);
- fd = g_mkstemp_full(fpath, O_WRONLY | O_TRUNC, 00644);
-
- if (fd == -1) {
- msg_err_config("cannot open for writing controller stats from %s: %s",
- fpath, strerror(errno));
- return;
- }
-
- fp = fdopen(fd, "w");
- stat = rspamd_main->stat;
-
- top = ucl_object_typed_new(UCL_OBJECT);
- ucl_object_insert_key(top, ucl_object_fromint(stat->messages_scanned), "scanned", 0, false);
- ucl_object_insert_key(top, ucl_object_fromint(stat->messages_learned), "learned", 0, false);
-
- if (stat->messages_scanned > 0) {
- sub = ucl_object_typed_new(UCL_OBJECT);
- for (i = METRIC_ACTION_REJECT; i <= METRIC_ACTION_NOACTION; i++) {
- ucl_object_insert_key(sub,
- ucl_object_fromint(stat->actions_stat[i]),
- rspamd_action_to_str(i), 0, false);
- }
- ucl_object_insert_key(top, sub, "actions", 0, false);
- }
-
- ucl_object_insert_key(top,
- ucl_object_fromint(stat->connections_count),
- "connections", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(stat->control_connections_count),
- "control_connections", 0, false);
-
- efuncs = ucl_object_emit_file_funcs(fp);
- if (!ucl_object_emit_full(top, UCL_EMIT_JSON_COMPACT,
- efuncs, NULL)) {
- msg_err_config("cannot write stats to %s: %s",
- fpath, strerror(errno));
-
- unlink(fpath);
- }
- else {
- if (rename(fpath, cfg->stats_file) == -1) {
- msg_err_config("cannot rename stats from %s to %s: %s",
- fpath, cfg->stats_file, strerror(errno));
- }
- }
-
- ucl_object_unref(top);
- fclose(fp);
- ucl_object_emit_funcs_free(efuncs);
- }
-
- static ev_timer rrd_timer;
-
- void rspamd_controller_on_terminate(struct rspamd_worker *worker,
- struct rspamd_rrd_file *rrd)
- {
- struct rspamd_abstract_worker_ctx *ctx;
-
- ctx = (struct rspamd_abstract_worker_ctx *) worker->ctx;
- rspamd_controller_store_saved_stats(worker->srv, worker->srv->cfg);
-
- if (rrd) {
- ev_timer_stop(ctx->event_loop, &rrd_timer);
- msg_info("closing rrd file: %s", rrd->filename);
- rspamd_rrd_close(rrd);
- }
- }
-
- static void
- rspamd_controller_load_saved_stats(struct rspamd_main *rspamd_main,
- struct rspamd_config *cfg)
- {
- struct ucl_parser *parser;
- ucl_object_t *obj;
- const ucl_object_t *elt, *subelt;
- struct rspamd_stat *stat, stat_copy;
- int i;
-
- if (cfg->stats_file == NULL) {
- return;
- }
-
- if (access(cfg->stats_file, R_OK) == -1) {
- msg_err_config("cannot load controller stats from %s: %s",
- cfg->stats_file, strerror(errno));
- return;
- }
-
- parser = ucl_parser_new(0);
-
- if (!ucl_parser_add_file(parser, cfg->stats_file)) {
- msg_err_config("cannot parse controller stats from %s: %s",
- cfg->stats_file, ucl_parser_get_error(parser));
- ucl_parser_free(parser);
-
- return;
- }
-
- obj = ucl_parser_get_object(parser);
- ucl_parser_free(parser);
-
- stat = rspamd_main->stat;
- memcpy(&stat_copy, stat, sizeof(stat_copy));
-
- elt = ucl_object_lookup(obj, "scanned");
-
- if (elt != NULL && ucl_object_type(elt) == UCL_INT) {
- stat_copy.messages_scanned = ucl_object_toint(elt);
- }
-
- elt = ucl_object_lookup(obj, "learned");
-
- if (elt != NULL && ucl_object_type(elt) == UCL_INT) {
- stat_copy.messages_learned = ucl_object_toint(elt);
- }
-
- elt = ucl_object_lookup(obj, "actions");
-
- if (elt != NULL) {
- for (i = METRIC_ACTION_REJECT; i <= METRIC_ACTION_NOACTION; i++) {
- subelt = ucl_object_lookup(elt, rspamd_action_to_str(i));
-
- if (subelt && ucl_object_type(subelt) == UCL_INT) {
- stat_copy.actions_stat[i] = ucl_object_toint(subelt);
- }
- }
- }
-
- elt = ucl_object_lookup(obj, "connections_count");
-
- if (elt != NULL && ucl_object_type(elt) == UCL_INT) {
- stat_copy.connections_count = ucl_object_toint(elt);
- }
-
- elt = ucl_object_lookup(obj, "control_connections_count");
-
- if (elt != NULL && ucl_object_type(elt) == UCL_INT) {
- stat_copy.control_connections_count = ucl_object_toint(elt);
- }
-
- ucl_object_unref(obj);
- memcpy(stat, &stat_copy, sizeof(stat_copy));
- }
-
- struct rspamd_controller_periodics_cbdata {
- struct rspamd_worker *worker;
- struct rspamd_rrd_file *rrd;
- struct rspamd_stat *stat;
- ev_timer save_stats_event;
- };
-
- static void
- rspamd_controller_rrd_update(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_controller_periodics_cbdata *cbd =
- (struct rspamd_controller_periodics_cbdata *) w->data;
- struct rspamd_stat *stat;
- GArray ar;
- double points[METRIC_ACTION_MAX];
- GError *err = NULL;
- unsigned int i;
-
- g_assert(cbd->rrd != NULL);
- stat = cbd->stat;
-
- for (i = METRIC_ACTION_REJECT; i < METRIC_ACTION_MAX; i++) {
- points[i] = stat->actions_stat[i];
- }
-
- ar.data = (char *) points;
- ar.len = sizeof(points);
-
- if (!rspamd_rrd_add_record(cbd->rrd, &ar, rspamd_get_calendar_ticks(),
- &err)) {
- msg_err("cannot update rrd file: %e", err);
- g_error_free(err);
- }
-
- /* Plan new event */
- ev_timer_again(EV_A_ w);
- }
-
- static void
- rspamd_controller_stats_save_periodic(EV_P_ ev_timer *w, int revents)
- {
- struct rspamd_controller_periodics_cbdata *cbd =
- (struct rspamd_controller_periodics_cbdata *) w->data;
-
- rspamd_controller_store_saved_stats(cbd->worker->srv, cbd->worker->srv->cfg);
- ev_timer_again(EV_A_ w);
- }
-
- void rspamd_worker_init_controller(struct rspamd_worker *worker,
- struct rspamd_rrd_file **prrd)
- {
- struct rspamd_abstract_worker_ctx *ctx;
- static const ev_tstamp rrd_update_time = 1.0;
-
- ctx = (struct rspamd_abstract_worker_ctx *) worker->ctx;
- rspamd_controller_load_saved_stats(worker->srv, worker->srv->cfg);
-
- if (worker->index == 0) {
- /* Enable periodics and other stuff */
- static struct rspamd_controller_periodics_cbdata cbd;
- const ev_tstamp save_stats_interval = 60; /* 1 minute */
-
- memset(&cbd, 0, sizeof(cbd));
- cbd.save_stats_event.data = &cbd;
- cbd.worker = worker;
- cbd.stat = worker->srv->stat;
-
- ev_timer_init(&cbd.save_stats_event,
- rspamd_controller_stats_save_periodic,
- save_stats_interval, save_stats_interval);
- ev_timer_start(ctx->event_loop, &cbd.save_stats_event);
-
- rspamd_map_watch(worker->srv->cfg, ctx->event_loop,
- ctx->resolver, worker,
- RSPAMD_MAP_WATCH_PRIMARY_CONTROLLER);
-
- if (prrd != NULL) {
- if (ctx->cfg->rrd_file && worker->index == 0) {
- GError *rrd_err = NULL;
-
- *prrd = rspamd_rrd_file_default(ctx->cfg->rrd_file, &rrd_err);
-
- if (*prrd) {
- cbd.rrd = *prrd;
- rrd_timer.data = &cbd;
- ev_timer_init(&rrd_timer, rspamd_controller_rrd_update,
- rrd_update_time, rrd_update_time);
- ev_timer_start(ctx->event_loop, &rrd_timer);
- }
- else if (rrd_err) {
- msg_err("cannot load rrd from %s: %e", ctx->cfg->rrd_file,
- rrd_err);
- g_error_free(rrd_err);
- }
- else {
- msg_err("cannot load rrd from %s: unknown error",
- ctx->cfg->rrd_file);
- }
- }
- else {
- *prrd = NULL;
- }
- }
-
- if (!ctx->cfg->disable_monitored) {
- rspamd_worker_init_monitored(worker,
- ctx->event_loop, ctx->resolver);
- }
- }
- else {
- rspamd_map_watch(worker->srv->cfg, ctx->event_loop,
- ctx->resolver, worker, RSPAMD_MAP_WATCH_SCANNER);
- }
- }
-
- double
- rspamd_worker_check_and_adjust_timeout(struct rspamd_config *cfg, double timeout)
- {
- if (isnan(timeout)) {
- /* Use implicit timeout from cfg->task_timeout */
- timeout = cfg->task_timeout;
- }
-
- if (isnan(timeout)) {
- return timeout;
- }
-
- struct rspamd_symcache_timeout_result *tres = rspamd_symcache_get_max_timeout(cfg->cache);
- g_assert(tres != 0);
-
- if (tres->max_timeout > timeout) {
- msg_info_config("configured task_timeout %.2f is less than maximum symbols cache timeout %.2f; "
- "some symbols can be terminated before checks",
- timeout, tres->max_timeout);
- GString *buf = g_string_sized_new(512);
- static const int max_displayed_items = 12;
-
- for (int i = 0; i < MIN(tres->nitems, max_displayed_items); i++) {
- if (i == 0) {
- rspamd_printf_gstring(buf, "%s(%.2f)",
- rspamd_symcache_item_name((struct rspamd_symcache_item *) tres->items[i].item),
- tres->items[i].timeout);
- }
- else {
- rspamd_printf_gstring(buf, "; %s(%.2f)",
- rspamd_symcache_item_name((struct rspamd_symcache_item *) tres->items[i].item),
- tres->items[i].timeout);
- }
- }
- msg_info_config("list of top %d symbols by execution time: %v",
- (int) MIN(tres->nitems, max_displayed_items),
- buf);
-
- g_string_free(buf, TRUE);
- }
-
- rspamd_symcache_timeout_result_free(tres);
-
- /* TODO: maybe adjust timeout */
- return timeout;
- }
|