12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333233423352336233723382339234023412342234323442345234623472348234923502351235223532354235523562357235823592360236123622363236423652366236723682369237023712372237323742375237623772378237923802381238223832384238523862387238823892390239123922393239423952396239723982399240024012402240324042405240624072408240924102411241224132414241524162417241824192420242124222423242424252426242724282429243024312432243324342435243624372438243924402441244224432444244524462447244824492450245124522453245424552456245724582459246024612462246324642465246624672468246924702471247224732474247524762477247824792480248124822483248424852486248724882489249024912492249324942495249624972498249925002501250225032504250525062507250825092510251125122513251425152516251725182519252025212522252325242525252625272528252925302531253225332534253525362537253825392540254125422543254425452546254725482549255025512552255325542555255625572558 |
- /*
- * 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;
- }
-
- ucl_object_t *
- rspamd_worker_metrics_object(struct rspamd_config *cfg, struct rspamd_stat *stat, ev_tstamp uptime)
- {
- rspamd_mempool_stat_t mem_st;
- memset(&mem_st, 0, sizeof(mem_st));
- rspamd_mempool_stat(&mem_st);
-
- ucl_object_t *top = ucl_object_typed_new(UCL_OBJECT);
-
- ucl_object_insert_key(top, ucl_object_fromstring(RVERSION), "version", 0, false);
- ucl_object_insert_key(top, ucl_object_fromstring(cfg->checksum), "config_id", 0, false);
- ucl_object_insert_key(top, ucl_object_fromdouble(uptime), "uptime", 0, false);
- 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);
-
- gsize cnt = MAX_AVG_TIME_SLOTS;
- float sum = rspamd_sum_floats(stat->avg_time.avg_time, &cnt);
-
- ucl_object_insert_key(top,
- ucl_object_fromdouble(cnt > 0 ? (double) sum / cnt : 0.0), "avg_scan_time", 0, false);
-
- unsigned spam = 0, ham = 0;
-
- if (stat->messages_scanned > 0) {
- ucl_object_t *sub = ucl_object_typed_new(UCL_OBJECT);
- for (int 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);
- if (i < METRIC_ACTION_GREYLIST) {
- spam += stat->actions_stat[i];
- }
- else {
- ham += stat->actions_stat[i];
- }
- }
- ucl_object_insert_key(top, sub, "actions", 0, false);
- }
-
- ucl_object_insert_key(top, ucl_object_fromint(spam), "spam_count", 0, false);
- ucl_object_insert_key(top, ucl_object_fromint(ham), "ham_count", 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);
-
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.pools_allocated), "pools_allocated", 0,
- false);
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.pools_freed), "pools_freed", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.bytes_allocated), "bytes_allocated", 0,
- false);
- ucl_object_insert_key(top,
- ucl_object_fromint(
- mem_st.chunks_allocated),
- "chunks_allocated", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.shared_chunks_allocated),
- "shared_chunks_allocated", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.chunks_freed), "chunks_freed", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(
- mem_st.oversized_chunks),
- "chunks_oversized", 0, false);
- ucl_object_insert_key(top,
- ucl_object_fromint(mem_st.fragmented_size), "fragmented", 0, false);
-
- return top;
- }
-
- rspamd_fstring_t *
- rspamd_metrics_to_prometheus_string(const ucl_object_t *top)
- {
- rspamd_fstring_t *output = rspamd_fstring_sized_new(1024);
-
- rspamd_printf_fstring(&output, "# HELP rspamd_build_info A metric with a constant '1' value "
- "labeled by version from which rspamd was built.\n");
- rspamd_printf_fstring(&output, "# TYPE rspamd_build_info gauge\n");
- rspamd_printf_fstring(&output, "rspamd_build_info{version=\"%s\"} 1\n",
- ucl_object_tostring(ucl_object_lookup(top, "version")));
- rspamd_printf_fstring(&output, "# HELP rspamd_config A metric with a constant '1' value "
- "labeled by id of the current config.\n");
- rspamd_printf_fstring(&output, "# TYPE rspamd_config gauge\n");
- rspamd_printf_fstring(&output, "rspamd_config{id=\"%s\"} 1\n",
- ucl_object_tostring(ucl_object_lookup(top, "config_id")));
-
-
- rspamd_metrics_add_integer(&output, top,
- "rspamd_scan_time_average",
- "gauge",
- "Average messages scan time.",
- "avg_scan_time");
- rspamd_metrics_add_integer(&output, top,
- "process_start_time_seconds",
- "gauge",
- "Start time of the process since unix epoch in seconds.",
- "start_time");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_read_only",
- "gauge",
- "Whether the rspamd instance is read-only.",
- "read_only");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_scanned_total",
- "counter",
- "Scanned messages.",
- "scanned");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_learned_total",
- "counter",
- "Learned messages.",
- "learned");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_spam_total",
- "counter",
- "Messages classified as spam.",
- "spam_count");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_ham_total",
- "counter",
- "Messages classified as ham.",
- "ham_count");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_connections",
- "gauge",
- "Active connections.",
- "connections");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_control_connections_total",
- "gauge",
- "Control connections.",
- "control_connections");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_pools_allocated",
- "gauge",
- "Pools allocated.",
- "pools_allocated");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_pools_freed",
- "gauge",
- "Pools freed.",
- "pools_freed");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_allocated_bytes",
- "gauge",
- "Bytes allocated.",
- "bytes_allocated");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_chunks_allocated",
- "gauge",
- "Memory pools: current chunks allocated.",
- "chunks_allocated");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_shared_chunks_allocated",
- "gauge",
- "Memory pools: current shared chunks allocated.",
- "shared_chunks_allocated");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_chunks_freed",
- "gauge",
- "Memory pools: current chunks freed.",
- "chunks_freed");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_chunks_oversized",
- "gauge",
- "Memory pools: current chunks oversized (needs extra allocation/fragmentation).",
- "chunks_oversized");
- rspamd_metrics_add_integer(&output, top,
- "rspamd_fragmented",
- "gauge",
- "Memory pools: fragmented memory waste.",
- "fragmented");
-
- const ucl_object_t *acts_obj = ucl_object_lookup(top, "actions");
-
- if (acts_obj) {
- rspamd_printf_fstring(&output, "# HELP rspamd_actions_total Actions labelled by action type.\n");
- rspamd_printf_fstring(&output, "# TYPE rspamd_actions_total counter\n");
- for (int i = METRIC_ACTION_REJECT; i <= METRIC_ACTION_NOACTION; i++) {
- const char *str_act = rspamd_action_to_str(i);
- const ucl_object_t *act = ucl_object_lookup(acts_obj, str_act);
-
- if (act) {
- rspamd_printf_fstring(&output, "rspamd_actions_total{type=\"%s\"} %L\n",
- str_act,
- ucl_object_toint(act));
- }
- else {
- rspamd_printf_fstring(&output, "rspamd_actions_total{type=\"%s\"} 0\n",
- str_act);
- }
- }
- }
-
- /* Must be finalized and freed by caller */
- return output;
- }
|