src/libcharon/plugins/ha/ha_segments.c

   1 /*
   2  * Copyright (C) 2008 Martin Willi
   3  * Hochschule fuer Technik Rapperswil
   4  *
   5  * This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the GNU General Public License as published by the
   7  * Free Software Foundation; either version 2 of the License, or (at your
   8  * option) any later version.  See <http://www.fsf.org/copyleft/gpl.txt>.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
  12  * or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  13  * for more details.
  14  */
  15
  16 #include "ha_segments.h"
  17
  18 #include <threading/mutex.h>
  19 #include <threading/condvar.h>
  20 #include <utils/linked_list.h>
  21 #include <threading/thread.h>
  22 #include <processing/jobs/callback_job.h>
  23
  24 #define DEFAULT_HEARTBEAT_DELAY 1000
  25 #define DEFAULT_HEARTBEAT_TIMEOUT 2100
  26
  27 typedef struct private_ha_segments_t private_ha_segments_t;
  28
  29 /**
  30  * Private data of an ha_segments_t object.
  31  */
  32 struct private_ha_segments_t {
  33
  34         /**
  35          * Public ha_segments_t interface.
  36          */
  37         ha_segments_t public;
  38
  39         /**
  40          * communication socket
  41          */
  42         ha_socket_t *socket;
  43
  44         /**
  45          * Sync tunnel, if any
  46          */
  47         ha_tunnel_t *tunnel;
  48
  49         /**
  50          * Interface to control segments at kernel level
  51          */
  52         ha_kernel_t *kernel;
  53
  54         /**
  55          * Mutex to lock segment manipulation
  56          */
  57         mutex_t *mutex;
  58
  59         /**
  60          * Condvar to wait for heartbeats
  61          */
  62         condvar_t *condvar;
  63
  64         /**
  65          * Job checking for heartbeats
  66          */
  67         callback_job_t *job;
  68
  69         /**
  70          * Total number of ClusterIP segments
  71          */
  72         u_int count;
  73
  74         /**
  75          * mask of active segments
  76          */
  77         segment_mask_t active;
  78
  79         /**
  80          * Node number
  81          */
  82         u_int node;
  83
  84         /**
  85          * Interval we send hearbeats
  86          */
  87         int heartbeat_delay;
  88
  89         /**
  90          * Timeout for heartbeats received from other node
  91          */
  92         int heartbeat_timeout;
  93 };
  94
  95 /**
  96  * Log currently active segments
  97  */
  98 static void log_segments(private_ha_segments_t *this, bool activated,
  99                                                  u_int segment)
 100 {
 101         char buf[64] = "none", *pos = buf;
 102         int i;
 103         bool first = TRUE;
 104
 105         for (i = 1; i <= this->count; i++)
 106         {
 107                 if (this->active & SEGMENTS_BIT(i))
 108                 {
 109                         if (first)
 110                         {
 111                                 first = FALSE;
 112                         }
 113                         else
 114                         {
 115                                 pos += snprintf(pos, buf + sizeof(buf) - pos, ",");
 116                         }
 117                         pos += snprintf(pos, buf + sizeof(buf) - pos, "%d", i);
 118                 }
 119         }
 120         DBG1(DBG_CFG, "HA segment %d %sactivated, now active: %s",
 121                  segment, activated ? "" : "de", buf);
 122 }
 123
 124 /**
 125  * Enable/Disable a specific segment
 126  */
 127 static void enable_disable(private_ha_segments_t *this, u_int segment,
 128                                                    bool enable, bool notify)
 129 {
 130         ike_sa_t *ike_sa;
 131         enumerator_t *enumerator;
 132         ike_sa_state_t old, new;
 133         ha_message_t *message = NULL;
 134         ha_message_type_t type;
 135         bool changes = FALSE;
 136
 137         if (segment > this->count)
 138         {
 139                 return;
 140         }
 141
 142         if (enable)
 143         {
 144                 old = IKE_PASSIVE;
 145                 new = IKE_ESTABLISHED;
 146                 type = HA_SEGMENT_TAKE;
 147                 if (!(this->active & SEGMENTS_BIT(segment)))
 148                 {
 149                         this->active |= SEGMENTS_BIT(segment);
 150                         this->kernel->activate(this->kernel, segment);
 151                         changes = TRUE;
 152                 }
 153         }
 154         else
 155         {
 156                 old = IKE_ESTABLISHED;
 157                 new = IKE_PASSIVE;
 158                 type = HA_SEGMENT_DROP;
 159                 if (this->active & SEGMENTS_BIT(segment))
 160                 {
 161                         this->active &= ~SEGMENTS_BIT(segment);
 162                         this->kernel->deactivate(this->kernel, segment);
 163                         changes = TRUE;
 164                 }
 165         }
 166
 167         if (changes)
 168         {
 169                 enumerator = charon->ike_sa_manager->create_enumerator(
 170                                                                                                 charon->ike_sa_manager, TRUE);
 171                 while (enumerator->enumerate(enumerator, &ike_sa))
 172                 {
 173                         if (ike_sa->get_state(ike_sa) != old)
 174                         {
 175                                 continue;
 176                         }
 177                         if (this->tunnel && this->tunnel->is_sa(this->tunnel, ike_sa))
 178                         {
 179                                 continue;
 180                         }
 181                         if (this->kernel->get_segment(this->kernel,
 182                                                                         ike_sa->get_other_host(ike_sa)) == segment)
 183                         {
 184                                 ike_sa->set_state(ike_sa, new);
 185                         }
 186                 }
 187                 enumerator->destroy(enumerator);
 188                 log_segments(this, enable, segment);
 189         }
 190
 191         if (notify)
 192         {
 193                 message = ha_message_create(type);
 194                 message->add_attribute(message, HA_SEGMENT, segment);
 195                 this->socket->push(this->socket, message);
 196                 message->destroy(message);
 197         }
 198 }
 199
 200 /**
 201  * Enable/Disable all or a specific segment, do locking
 202  */
 203 static void enable_disable_all(private_ha_segments_t *this, u_int segment,
 204                                                            bool enable, bool notify)
 205 {
 206         int i;
 207
 208         this->mutex->lock(this->mutex);
 209         if (segment == 0)
 210         {
 211                 for (i = 1; i <= this->count; i++)
 212                 {
 213                         enable_disable(this, i, enable, notify);
 214                 }
 215         }
 216         else
 217         {
 218                 enable_disable(this, segment, enable, notify);
 219         }
 220         this->mutex->unlock(this->mutex);
 221 }
 222
 223 METHOD(ha_segments_t, activate, void,
 224         private_ha_segments_t *this, u_int segment, bool notify)
 225 {
 226         enable_disable_all(this, segment, TRUE, notify);
 227 }
 228
 229 METHOD(ha_segments_t, deactivate, void,
 230         private_ha_segments_t *this, u_int segment, bool notify)
 231 {
 232         enable_disable_all(this, segment, FALSE, notify);
 233 }
 234
 235 METHOD(listener_t, alert_hook, bool,
 236         private_ha_segments_t *this, ike_sa_t *ike_sa, alert_t alert, va_list args)
 237 {
 238         if (alert == ALERT_SHUTDOWN_SIGNAL)
 239         {
 240                 if (this->job)
 241                 {
 242                         DBG1(DBG_CFG, "HA heartbeat active, dropping all segments");
 243                         deactivate(this, 0, TRUE);
 244                 }
 245                 else
 246                 {
 247                         DBG1(DBG_CFG, "no HA heartbeat active, closing IKE_SAs");
 248                 }
 249         }
 250         return TRUE;
 251 }
 252
 253 /**
 254  * Monitor heartbeat activity of remote node
 255  */
 256 static job_requeue_t watchdog(private_ha_segments_t *this)
 257 {
 258         bool timeout, oldstate;
 259
 260         this->mutex->lock(this->mutex);
 261         thread_cleanup_push((void*)this->mutex->unlock, this->mutex);
 262         oldstate = thread_cancelability(TRUE);
 263         timeout = this->condvar->timed_wait(this->condvar, this->mutex,
 264                                                                                 this->heartbeat_timeout);
 265         thread_cancelability(oldstate);
 266         thread_cleanup_pop(TRUE);
 267         if (timeout)
 268         {
 269                 DBG1(DBG_CFG, "no heartbeat received, taking all segments");
 270                 activate(this, 0, TRUE);
 271                 /* disable heartbeat detection util we get one */
 272                 this->job = NULL;
 273                 return JOB_REQUEUE_NONE;
 274         }
 275         return JOB_REQUEUE_DIRECT;
 276 }
 277
 278 /**
 279  * Start the heartbeat detection thread
 280  */
 281 static void start_watchdog(private_ha_segments_t *this)
 282 {
 283         this->job = callback_job_create((callback_job_cb_t)watchdog,
 284                                                                         this, NULL, NULL);
 285         lib->processor->queue_job(lib->processor, (job_t*)this->job);
 286 }
 287
 288 METHOD(ha_segments_t, handle_status, void,
 289         private_ha_segments_t *this, segment_mask_t mask)
 290 {
 291         segment_mask_t missing;
 292         int i;
 293
 294         this->mutex->lock(this->mutex);
 295
 296         missing = ~(this->active | mask);
 297
 298         for (i = 1; i <= this->count; i++)
 299         {
 300                 if (missing & SEGMENTS_BIT(i))
 301                 {
 302                         if (this->node == i % 2)
 303                         {
 304                                 DBG1(DBG_CFG, "HA segment %d was not handled, taking", i);
 305                                 enable_disable(this, i, TRUE, TRUE);
 306                         }
 307                         else
 308                         {
 309                                 DBG1(DBG_CFG, "HA segment %d was not handled, dropping", i);
 310                                 enable_disable(this, i, FALSE, TRUE);
 311                         }
 312                 }
 313         }
 314
 315         this->mutex->unlock(this->mutex);
 316         this->condvar->signal(this->condvar);
 317
 318         if (!this->job)
 319         {
 320                 DBG1(DBG_CFG, "received heartbeat, reenabling watchdog");
 321                 start_watchdog(this);
 322         }
 323 }
 324
 325 /**
 326  * Send a status message with our active segments
 327  */
 328 static job_requeue_t send_status(private_ha_segments_t *this)
 329 {
 330         ha_message_t *message;
 331         int i;
 332
 333         message = ha_message_create(HA_STATUS);
 334
 335         for (i = 1; i <= this->count; i++)
 336         {
 337                 if (this->active & SEGMENTS_BIT(i))
 338                 {
 339                         message->add_attribute(message, HA_SEGMENT, i);
 340                 }
 341         }
 342
 343         this->socket->push(this->socket, message);
 344         message->destroy(message);
 345
 346         /* schedule next invocation */
 347         lib->scheduler->schedule_job_ms(lib->scheduler, (job_t*)
 348                                                                         callback_job_create((callback_job_cb_t)
 349                                                                                 send_status, this, NULL, NULL),
 350                                                                         this->heartbeat_delay);
 351
 352         return JOB_REQUEUE_NONE;
 353 }
 354
 355 METHOD(ha_segments_t, is_active, bool,
 356         private_ha_segments_t *this, u_int segment)
 357 {
 358         return (this->active & SEGMENTS_BIT(segment)) != 0;
 359 }
 360
 361 METHOD(ha_segments_t, destroy, void,
 362         private_ha_segments_t *this)
 363 {
 364         if (this->job)
 365         {
 366                 this->job->cancel(this->job);
 367         }
 368         this->mutex->destroy(this->mutex);
 369         this->condvar->destroy(this->condvar);
 370         free(this);
 371 }
 372
 373 /**
 374  * See header
 375  */
 376 ha_segments_t *ha_segments_create(ha_socket_t *socket, ha_kernel_t *kernel,
 377                                                                   ha_tunnel_t *tunnel, u_int count, u_int node,
 378                                                                   bool monitor)
 379 {
 380         private_ha_segments_t *this;
 381
 382         INIT(this,
 383                 .public = {
 384                         .listener = {
 385                                 .alert = _alert_hook,
 386                         },
 387                         .activate = _activate,
 388                         .deactivate = _deactivate,
 389                         .handle_status = _handle_status,
 390                         .is_active = _is_active,
 391                         .destroy = _destroy,
 392                 },
 393                 .socket = socket,
 394                 .tunnel = tunnel,
 395                 .kernel = kernel,
 396                 .count = count,
 397                 .node = node,
 398                 .mutex = mutex_create(MUTEX_TYPE_DEFAULT),
 399                 .condvar = condvar_create(CONDVAR_TYPE_DEFAULT),
 400                 .heartbeat_delay = lib->settings->get_int(lib->settings,
 401                         "charon.plugins.ha.heartbeat_delay", DEFAULT_HEARTBEAT_DELAY),
 402                 .heartbeat_timeout = lib->settings->get_int(lib->settings,
 403                         "charon.plugins.ha.heartbeat_timeout", DEFAULT_HEARTBEAT_TIMEOUT),
 404         );
 405
 406         if (monitor)
 407         {
 408                 DBG1(DBG_CFG, "starting HA heartbeat, delay %dms, timeout %dms",
 409                          this->heartbeat_delay, this->heartbeat_timeout);
 410                 send_status(this);
 411                 start_watchdog(this);
 412         }
 413
 414         return &this->public;
 415 }
 416