[Rework] Send health checks from a single worker

4 files changed, 53 insertions, 1 deletions

diff --git a/src/libserver/monitored.c b/src/libserver/monitored.c
index a280298c8..548dcf362 100644
--- a/src/libserver/monitored.c
+++ b/src/libserver/monitored.c
@@ -386,6 +386,13 @@ rspamd_monitored_ctx_config (struct rspamd_monitored_ctx *ctx,
 }
 
 
+struct event_base *
+rspamd_monitored_ctx_get_ev_base (struct rspamd_monitored_ctx *ctx)
+{
+	return ctx->ev_base;
+}
+
+
 struct rspamd_monitored *
 rspamd_monitored_create_ (struct rspamd_monitored_ctx *ctx,
 		const gchar *line,
@@ -580,5 +587,5 @@ rspamd_monitored_get_tag (struct rspamd_monitored *m,
 {
 	g_assert (m != NULL);
 
-	rspamd_strlcpy (m->tag, tag_out, RSPAMD_MONITORED_TAG_LEN);
+	rspamd_strlcpy (tag_out, m->tag, RSPAMD_MONITORED_TAG_LEN);
 }
\ No newline at end of file
diff --git a/src/libserver/monitored.h b/src/libserver/monitored.h
index 261227d35..4db41f9c2 100644
--- a/src/libserver/monitored.h
+++ b/src/libserver/monitored.h
@@ -57,6 +57,8 @@ void rspamd_monitored_ctx_config (struct rspamd_monitored_ctx *ctx,
 		mon_change_cb change_cb,
 		gpointer ud);
 
+struct event_base *rspamd_monitored_ctx_get_ev_base (struct rspamd_monitored_ctx *ctx);
+
 /**
  * Create monitored object
  * @param ctx context
diff --git a/src/libserver/worker_util.c b/src/libserver/worker_util.c
index fcd9a8be9..0c0cc9344 100644
--- a/src/libserver/worker_util.c
+++ b/src/libserver/worker_util.c
@@ -802,4 +802,37 @@ 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 event_base *ev_base;
+	guchar 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);
+	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;
+	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 event_base *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);
 }
\ No newline at end of file
diff --git a/src/libserver/worker_util.h b/src/libserver/worker_util.h
index b21a92bfc..5e33c940d 100644
--- a/src/libserver/worker_util.h
+++ b/src/libserver/worker_util.h
@@ -167,6 +167,16 @@ void rspamd_worker_session_cache_remove (void *cache, void *ptr);
 struct rspamd_worker *rspamd_fork_worker (struct rspamd_main *,
 		struct rspamd_worker_conf *, guint idx, struct event_base *ev_base);
 
+/**
+ * Initialise the main monitoring worker
+ * @param worker
+ * @param ev_base
+ * @param resolver
+ */
+void rspamd_worker_init_monitored (struct rspamd_worker *worker,
+		struct event_base *ev_base,
+		struct rspamd_dns_resolver *resolver);
+
 #define msg_err_main(...) rspamd_default_log_function (G_LOG_LEVEL_CRITICAL, \
         rspamd_main->server_pool->tag.tagname, rspamd_main->server_pool->tag.uid, \
         G_STRFUNC, \