1 // L2TPNS Clustering Stuff
3 char const *cvs_id_cluster
= "$Id: cluster.c,v 1.44 2005-06-28 14:48:19 bodea Exp $";
12 #include <sys/socket.h>
13 #include <netinet/in.h>
14 #include <arpa/inet.h>
15 #include <sys/ioctl.h>
31 * All cluster packets have the same format.
33 * One or more instances of
35 * a 32 bit 'extra' data dependant on the 'type'.
36 * zero or more bytes of structure data, dependant on the type.
41 extern int cluster_sockfd
; // The filedescriptor for the cluster communications port.
43 in_addr_t my_address
= 0; // The network address of my ethernet port.
44 static int walk_session_number
= 0; // The next session to send when doing the slow table walk.
45 static int walk_tunnel_number
= 0; // The next tunnel to send when doing the slow table walk.
46 int forked
= 0; // Sanity check: CLI must not diddle with heartbeat table
48 #define MAX_HEART_SIZE (8192) // Maximum size of heartbeat packet. Must be less than max IP packet size :)
49 #define MAX_CHANGES (MAX_HEART_SIZE/(sizeof(sessiont) + sizeof(int) ) - 2) // Assumes a session is the biggest type!
54 } cluster_changes
[MAX_CHANGES
]; // Queue of changed structures that need to go out when next heartbeat.
59 char data
[MAX_HEART_SIZE
];
60 } past_hearts
[HB_HISTORY_SIZE
]; // Ring buffer of heartbeats that we've recently sent out. Needed so
61 // we can re-transmit if needed.
68 } peers
[CLUSTER_MAX_SIZE
]; // List of all the peers we've heard from.
69 static int num_peers
; // Number of peers in list.
71 static int rle_decompress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
);
72 static int rle_compress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
);
75 // Create a listening socket
77 // This joins the cluster multi-cast group.
81 struct sockaddr_in addr
;
82 struct sockaddr_in interface_addr
;
87 config
->cluster_undefined_sessions
= MAXSESSION
-1;
88 config
->cluster_undefined_tunnels
= MAXTUNNEL
-1;
90 if (!config
->cluster_address
)
92 if (!*config
->cluster_interface
)
95 cluster_sockfd
= socket(AF_INET
, SOCK_DGRAM
, IPPROTO_UDP
);
97 memset(&addr
, 0, sizeof(addr
));
98 addr
.sin_family
= AF_INET
;
99 addr
.sin_port
= htons(CLUSTERPORT
);
100 addr
.sin_addr
.s_addr
= INADDR_ANY
;
101 setsockopt(cluster_sockfd
, SOL_SOCKET
, SO_REUSEADDR
, &addr
, sizeof(addr
));
103 opt
= fcntl(cluster_sockfd
, F_GETFL
, 0);
104 fcntl(cluster_sockfd
, F_SETFL
, opt
| O_NONBLOCK
);
106 if (bind(cluster_sockfd
, (void *) &addr
, sizeof(addr
)) < 0)
108 LOG(0, 0, 0, "Failed to bind cluster socket: %s\n", strerror(errno
));
112 strcpy(ifr
.ifr_name
, config
->cluster_interface
);
113 if (ioctl(cluster_sockfd
, SIOCGIFADDR
, &ifr
) < 0)
115 LOG(0, 0, 0, "Failed to get interface address for (%s): %s\n", config
->cluster_interface
, strerror(errno
));
119 memcpy(&interface_addr
, &ifr
.ifr_addr
, sizeof(interface_addr
));
120 my_address
= interface_addr
.sin_addr
.s_addr
;
122 // Join multicast group.
123 mreq
.imr_multiaddr
.s_addr
= config
->cluster_address
;
124 mreq
.imr_interface
= interface_addr
.sin_addr
;
127 opt
= 0; // Turn off multicast loopback.
128 setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_LOOP
, &opt
, sizeof(opt
));
130 if (setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_ADD_MEMBERSHIP
, &mreq
, sizeof(mreq
)) < 0)
132 LOG(0, 0, 0, "Failed to setsockopt (join mcast group): %s\n", strerror(errno
));
136 if (setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_IF
, &interface_addr
, sizeof(interface_addr
)) < 0)
138 LOG(0, 0, 0, "Failed to setsockopt (set mcast interface): %s\n", strerror(errno
));
142 config
->cluster_last_hb
= TIME
;
143 config
->cluster_seq_number
= -1;
145 return cluster_sockfd
;
150 // Send a chunk of data to the entire cluster (usually via the multicast
154 static int cluster_send_data(void *data
, int datalen
)
156 struct sockaddr_in addr
= {0};
158 if (!cluster_sockfd
) return -1;
159 if (!config
->cluster_address
) return 0;
161 addr
.sin_addr
.s_addr
= config
->cluster_address
;
162 addr
.sin_port
= htons(CLUSTERPORT
);
163 addr
.sin_family
= AF_INET
;
165 LOG(5, 0, 0, "Cluster send data: %d bytes\n", datalen
);
167 if (sendto(cluster_sockfd
, data
, datalen
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
169 LOG(0, 0, 0, "sendto: %s\n", strerror(errno
));
177 // Add a chunk of data to a heartbeat packet.
178 // Maintains the format. Assumes that the caller
179 // has passed in a big enough buffer!
181 static void add_type(char **p
, int type
, int more
, char *data
, int size
)
183 *((uint32_t *) (*p
)) = type
;
184 *p
+= sizeof(uint32_t);
186 *((uint32_t *)(*p
)) = more
;
187 *p
+= sizeof(uint32_t);
189 if (data
&& size
> 0) {
190 memcpy(*p
, data
, size
);
195 // advertise our presence via BGP or gratuitous ARP
196 static void advertise_routes(void)
200 bgp_enable_routing(1);
203 if (config
->send_garp
)
204 send_garp(config
->bind_address
); // Start taking traffic.
207 // withdraw our routes (BGP only)
208 static void withdraw_routes(void)
212 bgp_enable_routing(0);
216 static void cluster_uptodate(void)
218 if (config
->cluster_iam_uptodate
)
221 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
)
224 config
->cluster_iam_uptodate
= 1;
226 LOG(0, 0, 0, "Now uptodate with master.\n");
231 // Send a unicast UDP packet to a peer with 'data' as the
234 static int peer_send_data(in_addr_t peer
, char *data
, int size
)
236 struct sockaddr_in addr
= {0};
238 if (!cluster_sockfd
) return -1;
239 if (!config
->cluster_address
) return 0;
244 addr
.sin_addr
.s_addr
= peer
;
245 addr
.sin_port
= htons(CLUSTERPORT
);
246 addr
.sin_family
= AF_INET
;
248 LOG_HEX(5, "Peer send", data
, size
);
250 if (sendto(cluster_sockfd
, data
, size
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
252 LOG(0, 0, 0, "sendto: %s\n", strerror(errno
));
260 // Send a structured message to a peer with a single element of type 'type'.
262 static int peer_send_message(in_addr_t peer
, int type
, int more
, char *data
, int size
)
264 char buf
[65536]; // Vast overkill.
267 LOG(4, 0, 0, "Sending message to peer (type %d, more %d, size %d)\n", type
, more
, size
);
268 add_type(&p
, type
, more
, data
, size
);
270 return peer_send_data(peer
, buf
, (p
-buf
) );
273 // send a packet to the master
274 static int _forward_packet(char *data
, int size
, in_addr_t addr
, int port
, int type
)
276 char buf
[65536]; // Vast overkill.
279 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
282 LOG(4, 0, 0, "Forwarding packet from %s to master (size %d)\n", fmtaddr(addr
, 0), size
);
285 add_type(&p
, type
, addr
, (char *) &port
, sizeof(port
)); // ick. should be uint16_t
286 memcpy(p
, data
, size
);
289 return peer_send_data(config
->cluster_master_address
, buf
, (p
- buf
));
293 // Forward a state changing packet to the master.
295 // The master just processes the payload as if it had
296 // received it off the tun device.
298 int master_forward_packet(char *data
, int size
, in_addr_t addr
, int port
)
300 return _forward_packet(data
, size
, addr
, port
, C_FORWARD
);
303 // Forward a DAE RADIUS packet to the master.
304 int master_forward_dae_packet(char *data
, int size
, in_addr_t addr
, int port
)
306 return _forward_packet(data
, size
, addr
, port
, C_FORWARD_DAE
);
310 // Forward a throttled packet to the master for handling.
312 // The master just drops the packet into the appropriate
313 // token bucket queue, and lets normal processing take care
316 int master_throttle_packet(int tbfid
, char *data
, int size
)
318 char buf
[65536]; // Vast overkill.
321 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
324 LOG(4, 0, 0, "Throttling packet master (size %d, tbfid %d)\n", size
, tbfid
);
326 add_type(&p
, C_THROTTLE
, tbfid
, data
, size
);
328 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
) );
333 // Forward a walled garden packet to the master for handling.
335 // The master just writes the packet straight to the tun
336 // device (where is will normally loop through the
337 // firewall rules, and come back in on the tun device)
339 // (Note that this must be called with the tun header
340 // as the start of the data).
341 int master_garden_packet(sessionidt s
, char *data
, int size
)
343 char buf
[65536]; // Vast overkill.
346 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
349 LOG(4, 0, 0, "Walled garden packet to master (size %d)\n", size
);
351 add_type(&p
, C_GARDEN
, s
, data
, size
);
353 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
));
358 // Send a chunk of data as a heartbeat..
359 // We save it in the history buffer as we do so.
361 static void send_heartbeat(int seq
, char *data
, int size
)
365 if (size
> sizeof(past_hearts
[0].data
))
367 LOG(0, 0, 0, "Tried to heartbeat something larger than the maximum packet!\n");
371 i
= seq
% HB_HISTORY_SIZE
;
372 past_hearts
[i
].seq
= seq
;
373 past_hearts
[i
].size
= size
;
374 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
375 cluster_send_data(data
, size
);
379 // Send an 'i am alive' message to every machine in the cluster.
381 void cluster_send_ping(time_t basetime
)
383 char buff
[100 + sizeof(pingt
)];
387 if (config
->cluster_iam_master
&& basetime
) // We're heartbeating so no need to ping.
390 LOG(5, 0, 0, "Sending cluster ping...\n");
393 x
.addr
= config
->bind_address
;
394 x
.undef
= config
->cluster_undefined_sessions
+ config
->cluster_undefined_tunnels
;
395 x
.basetime
= basetime
;
397 add_type(&p
, C_PING
, basetime
, (char *) &x
, sizeof(x
));
398 cluster_send_data(buff
, (p
-buff
) );
402 // Walk the session counters looking for non-zero ones to send
403 // to the master. We send up to 600 of them at one time.
404 // We examine a maximum of 3000 sessions.
405 // (50k max session should mean that we normally
406 // examine the entire session table every 25 seconds).
408 #define MAX_B_RECS (600)
409 void master_update_counts(void)
412 bytest b
[MAX_B_RECS
+1];
414 if (config
->cluster_iam_master
) // Only happens on the slaves.
417 if (!config
->cluster_master_address
) // If we don't have a master, skip it for a while.
420 // C_BYTES format changed in 2.1.0 (cluster version 5)
421 // during upgrade from previous versions, hang onto our counters
422 // for a bit until the new master comes up
423 if (config
->cluster_last_hb_ver
< 5)
426 i
= MAX_B_RECS
* 5; // Examine max 3000 sessions;
427 if (config
->cluster_highest_sessionid
> i
)
428 i
= config
->cluster_highest_sessionid
;
430 for ( c
= 0; i
> 0 ; --i
) {
431 // Next session to look at.
432 walk_session_number
++;
433 if ( walk_session_number
> config
->cluster_highest_sessionid
)
434 walk_session_number
= 1;
436 if (!sess_local
[walk_session_number
].cin
&& !sess_local
[walk_session_number
].cout
)
437 continue; // Unchanged. Skip it.
439 b
[c
].sid
= walk_session_number
;
440 b
[c
].pin
= sess_local
[walk_session_number
].pin
;
441 b
[c
].pout
= sess_local
[walk_session_number
].pout
;
442 b
[c
].cin
= sess_local
[walk_session_number
].cin
;
443 b
[c
].cout
= sess_local
[walk_session_number
].cout
;
446 sess_local
[walk_session_number
].pin
= sess_local
[walk_session_number
].pout
= 0;
447 sess_local
[walk_session_number
].cin
= sess_local
[walk_session_number
].cout
= 0;
449 if (++c
> MAX_B_RECS
) // Send a max of 600 elements in a packet.
453 if (!c
) // Didn't find any that changes. Get out of here!
457 // Forward the data to the master.
458 LOG(4, 0, 0, "Sending byte counters to master (%d elements)\n", c
);
459 peer_send_message(config
->cluster_master_address
, C_BYTES
, c
, (char *) &b
, sizeof(b
[0]) * c
);
464 // On the master, check how our slaves are going. If
465 // one of them's not up-to-date we'll heartbeat faster.
466 // If we don't have any of them, then we need to turn
467 // on our own packet handling!
469 void cluster_check_slaves(void)
472 static int have_peers
= 0;
473 int had_peers
= have_peers
;
476 if (!config
->cluster_iam_master
)
477 return; // Only runs on the master...
479 config
->cluster_iam_uptodate
= 1; // cleared in loop below
481 for (i
= have_peers
= 0; i
< num_peers
; i
++)
483 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
) < t
)
484 continue; // Stale peer! Skip them.
486 if (!peers
[i
].basetime
)
487 continue; // Shutdown peer! Skip them.
489 if (peers
[i
].uptodate
)
492 config
->cluster_iam_uptodate
= 0; // Start fast heartbeats
495 // in a cluster, withdraw/add routes when we get a peer/lose peers
496 if (have_peers
!= had_peers
)
498 if (had_peers
< config
->cluster_master_min_adv
&&
499 have_peers
>= config
->cluster_master_min_adv
)
502 else if (had_peers
>= config
->cluster_master_min_adv
&&
503 have_peers
< config
->cluster_master_min_adv
)
509 // Check that we have a master. If it's been too
510 // long since we heard from a master then hold an election.
512 void cluster_check_master(void)
514 int i
, count
, tcount
, high_unique_id
= 0;
517 static int probed
= 0;
520 if (config
->cluster_iam_master
)
521 return; // Only runs on the slaves...
523 // If the master is late (missed 2 hearbeats by a second and a
524 // hair) it may be that the switch has dropped us from the
525 // multicast group, try unicasting probes to the master
526 // which will hopefully respond with a unicast heartbeat that
527 // will allow us to limp along until the querier next runs.
528 if (config
->cluster_master_address
529 && TIME
> (config
->cluster_last_hb
+ 2 * config
->cluster_hb_interval
+ 11))
531 if (!probed
|| (TIME
> (probed
+ 2 * config
->cluster_hb_interval
)))
534 LOG(1, 0, 0, "Heartbeat from master %.1fs late, probing...\n",
535 0.1 * (TIME
- (config
->cluster_last_hb
+ config
->cluster_hb_interval
)));
537 peer_send_message(config
->cluster_master_address
,
538 C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
540 } else { // We got a recent heartbeat; reset the probe flag.
544 if (TIME
< (config
->cluster_last_hb
+ config
->cluster_hb_timeout
))
545 return; // Everything's ok!
547 config
->cluster_last_hb
= TIME
+ 1; // Just the one election thanks.
548 config
->cluster_master_address
= 0;
550 LOG(0, 0, 0, "Master timed out! Holding election...\n");
552 // In the process of shutting down, can't be master
556 for (i
= have_peers
= 0; i
< num_peers
; i
++)
558 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
) < t
)
559 continue; // Stale peer! Skip them.
561 if (!peers
[i
].basetime
)
562 continue; // Shutdown peer! Skip them.
564 if (peers
[i
].basetime
< basetime
) {
565 LOG(1, 0, 0, "Expecting %s to become master\n", fmtaddr(peers
[i
].peer
, 0));
566 return; // They'll win the election. Get out of here.
569 if (peers
[i
].basetime
== basetime
&&
570 peers
[i
].peer
> my_address
) {
571 LOG(1, 0, 0, "Expecting %s to become master\n", fmtaddr(peers
[i
].peer
, 0));
572 return; // They'll win the election. Wait for them to come up.
575 if (peers
[i
].uptodate
)
579 // Wow. it's been ages since I last heard a heartbeat
580 // and I'm better than an of my peers so it's time
581 // to become a master!!!
583 config
->cluster_iam_master
= 1;
585 LOG(0, 0, 0, "I am declaring myself the master!\n");
587 if (have_peers
< config
->cluster_master_min_adv
)
592 if (config
->cluster_seq_number
== -1)
593 config
->cluster_seq_number
= 0;
596 // Go through and mark all the tunnels as defined.
597 // Count the highest used tunnel number as well.
599 config
->cluster_highest_tunnelid
= 0;
600 for (i
= 0, tcount
= 0; i
< MAXTUNNEL
; ++i
) {
601 if (tunnel
[i
].state
== TUNNELUNDEF
)
602 tunnel
[i
].state
= TUNNELFREE
;
604 if (tunnel
[i
].state
!= TUNNELFREE
&& i
> config
->cluster_highest_tunnelid
)
605 config
->cluster_highest_tunnelid
= i
;
609 // Go through and mark all the sessions as being defined.
610 // reset the idle timeouts.
611 // add temporary byte counters to permanent ones.
612 // Re-string the free list.
613 // Find the ID of the highest session.
616 config
->cluster_highest_sessionid
= 0;
617 for (i
= 0, count
= 0; i
< MAXSESSION
; ++i
) {
618 if (session
[i
].tunnel
== T_UNDEF
) {
619 session
[i
].tunnel
= T_FREE
;
623 if (!session
[i
].opened
) { // Unused session. Add to free list.
624 memset(&session
[i
], 0, sizeof(session
[i
]));
625 session
[i
].tunnel
= T_FREE
;
626 session
[last_free
].next
= i
;
632 // Reset idle timeouts..
633 session
[i
].last_packet
= time_now
;
635 // Reset die relative to our uptime rather than the old master's
636 if (session
[i
].die
) session
[i
].die
= TIME
;
638 // Accumulate un-sent byte/packet counters.
639 increment_counter(&session
[i
].cin
, &session
[i
].cin_wrap
, sess_local
[i
].cin
);
640 increment_counter(&session
[i
].cout
, &session
[i
].cout_wrap
, sess_local
[i
].cout
);
641 session
[i
].cin_delta
+= sess_local
[i
].cin
;
642 session
[i
].cout_delta
+= sess_local
[i
].cout
;
644 session
[i
].pin
+= sess_local
[i
].pin
;
645 session
[i
].pout
+= sess_local
[i
].pout
;
647 sess_local
[i
].cin
= sess_local
[i
].cout
= 0;
648 sess_local
[i
].pin
= sess_local
[i
].pout
= 0;
650 sess_local
[i
].radius
= 0; // Reset authentication as the radius blocks aren't up to date.
652 if (session
[i
].unique_id
>= high_unique_id
) // This is different to the index into the session table!!!
653 high_unique_id
= session
[i
].unique_id
+1;
655 session
[i
].tbf_in
= session
[i
].tbf_out
= 0; // Remove stale pointers from old master.
656 throttle_session(i
, session
[i
].throttle_in
, session
[i
].throttle_out
);
658 config
->cluster_highest_sessionid
= i
;
661 session
[last_free
].next
= 0; // End of chain.
662 last_id
= high_unique_id
; // Keep track of the highest used session ID.
666 rebuild_address_pool();
668 // If we're not the very first master, this is a big issue!
670 LOG(0, 0, 0, "Warning: Fixed %d uninitialized sessions in becoming master!\n", count
);
672 config
->cluster_undefined_sessions
= 0;
673 config
->cluster_undefined_tunnels
= 0;
674 config
->cluster_iam_uptodate
= 1; // assume all peers are up-to-date
676 // FIXME. We need to fix up the tunnel control message
677 // queue here! There's a number of other variables we
678 // should also update.
683 // Check that our session table is validly matching what the
684 // master has in mind.
686 // In particular, if we have too many sessions marked 'undefined'
687 // we fix it up here, and we ensure that the 'first free session'
690 static void cluster_check_sessions(int highsession
, int freesession_ptr
, int hightunnel
)
694 sessionfree
= freesession_ptr
; // Keep the freesession ptr valid.
696 if (config
->cluster_iam_uptodate
)
699 if (highsession
> config
->cluster_undefined_sessions
&& hightunnel
> config
->cluster_undefined_tunnels
)
702 // Clear out defined sessions, counting the number of
704 config
->cluster_undefined_sessions
= 0;
705 for (i
= 1 ; i
< MAXSESSION
; ++i
) {
706 if (i
> highsession
) {
707 if (session
[i
].tunnel
== T_UNDEF
) session
[i
].tunnel
= T_FREE
; // Defined.
711 if (session
[i
].tunnel
== T_UNDEF
)
712 ++config
->cluster_undefined_sessions
;
715 // Clear out defined tunnels, counting the number of
717 config
->cluster_undefined_tunnels
= 0;
718 for (i
= 1 ; i
< MAXTUNNEL
; ++i
) {
719 if (i
> hightunnel
) {
720 if (tunnel
[i
].state
== TUNNELUNDEF
) tunnel
[i
].state
= TUNNELFREE
; // Defined.
724 if (tunnel
[i
].state
== TUNNELUNDEF
)
725 ++config
->cluster_undefined_tunnels
;
729 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
) {
730 LOG(2, 0, 0, "Cleared undefined sessions/tunnels. %d sess (high %d), %d tunn (high %d)\n",
731 config
->cluster_undefined_sessions
, highsession
, config
->cluster_undefined_tunnels
, hightunnel
);
735 // Are we up to date?
737 if (!config
->cluster_iam_uptodate
)
741 static int hb_add_type(char **p
, int type
, int id
)
744 case C_CSESSION
: { // Compressed C_SESSION.
745 uint8_t c
[sizeof(sessiont
) * 2]; // Bigger than worst case.
746 uint8_t *d
= (uint8_t *) &session
[id
];
750 size
= rle_compress( &d
, sizeof(sessiont
), c
, sizeof(c
) );
752 // Did we compress the full structure, and is the size actually
754 if ( (d
- orig
) == sizeof(sessiont
) && size
< sizeof(sessiont
) ) {
755 add_type(p
, C_CSESSION
, id
, (char *) c
, size
);
758 // Failed to compress : Fall through.
760 case C_SESSION
: add_type(p
, C_SESSION
, id
,
761 (char *) &session
[id
], sizeof(sessiont
));
764 case C_CTUNNEL
: { // Compressed C_TUNNEL
765 uint8_t c
[sizeof(tunnelt
) * 2]; // Bigger than worst case.
766 uint8_t *d
= (uint8_t *) &tunnel
[id
];
770 size
= rle_compress( &d
, sizeof(tunnelt
), c
, sizeof(c
) );
772 // Did we compress the full structure, and is the size actually
774 if ( (d
- orig
) == sizeof(tunnelt
) && size
< sizeof(tunnelt
) ) {
775 add_type(p
, C_CTUNNEL
, id
, c
, size
);
778 // Failed to compress : Fall through.
780 case C_TUNNEL
: add_type(p
, C_TUNNEL
, id
,
781 (char *) &tunnel
[id
], sizeof(tunnelt
));
784 LOG(0, 0, 0, "Found an invalid type in heart queue! (%d)\n", type
);
792 // Send a heartbeat, incidently sending out any queued changes..
794 void cluster_heartbeat()
796 int i
, count
= 0, tcount
= 0;
797 char buff
[MAX_HEART_SIZE
+ sizeof(heartt
) + sizeof(int) ];
801 if (!config
->cluster_iam_master
) // Only the master does this.
804 config
->cluster_table_version
+= config
->cluster_num_changes
;
806 // Fill out the heartbeat header.
807 memset(&h
, 0, sizeof(h
));
809 h
.version
= HB_VERSION
;
810 h
.seq
= config
->cluster_seq_number
;
811 h
.basetime
= basetime
;
812 h
.clusterid
= config
->bind_address
; // Will this do??
813 h
.basetime
= basetime
;
814 h
.highsession
= config
->cluster_highest_sessionid
;
815 h
.freesession
= sessionfree
;
816 h
.hightunnel
= config
->cluster_highest_tunnelid
;
817 h
.size_sess
= sizeof(sessiont
); // Just in case.
818 h
.size_tunn
= sizeof(tunnelt
);
819 h
.interval
= config
->cluster_hb_interval
;
820 h
.timeout
= config
->cluster_hb_timeout
;
821 h
.table_version
= config
->cluster_table_version
;
823 add_type(&p
, C_HEARTBEAT
, HB_VERSION
, (char *) &h
, sizeof(h
));
825 for (i
= 0; i
< config
->cluster_num_changes
; ++i
) {
826 hb_add_type(&p
, cluster_changes
[i
].type
, cluster_changes
[i
].id
);
829 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
830 LOG(0, 0, 0, "FATAL: Overran the heartbeat buffer! This is fatal. Exiting. (size %d)\n", (int) (p
- buff
));
836 // Fill out the packet with sessions from the session table...
837 // (not forgetting to leave space so we can get some tunnels in too )
838 while ( (p
+ sizeof(uint32_t) * 2 + sizeof(sessiont
) * 2 ) < (buff
+ MAX_HEART_SIZE
) ) {
840 if (!walk_session_number
) // session #0 isn't valid.
841 ++walk_session_number
;
843 if (count
>= config
->cluster_highest_sessionid
) // If we're a small cluster, don't go wild.
846 hb_add_type(&p
, C_CSESSION
, walk_session_number
);
847 walk_session_number
= (1+walk_session_number
)%(config
->cluster_highest_sessionid
+1); // +1 avoids divide by zero.
849 ++count
; // Count the number of extra sessions we're sending.
853 // Fill out the packet with tunnels from the tunnel table...
854 // This effectively means we walk the tunnel table more quickly
855 // than the session table. This is good because stuffing up a
856 // tunnel is a much bigger deal than stuffing up a session.
858 while ( (p
+ sizeof(uint32_t) * 2 + sizeof(tunnelt
) ) < (buff
+ MAX_HEART_SIZE
) ) {
860 if (!walk_tunnel_number
) // tunnel #0 isn't valid.
861 ++walk_tunnel_number
;
863 if (tcount
>= config
->cluster_highest_tunnelid
)
866 hb_add_type(&p
, C_CTUNNEL
, walk_tunnel_number
);
867 walk_tunnel_number
= (1+walk_tunnel_number
)%(config
->cluster_highest_tunnelid
+1); // +1 avoids divide by zero.
873 // Did we do something wrong?
874 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
875 LOG(0, 0, 0, "Overran the heartbeat buffer now! This is fatal. Exiting. (size %d)\n", (int) (p
- buff
));
880 LOG(3, 0, 0, "Sending v%d heartbeat #%d, change #%" PRIu64
" with %d changes "
881 "(%d x-sess, %d x-tunnels, %d highsess, %d hightun, size %d)\n",
882 HB_VERSION
, h
.seq
, h
.table_version
, config
->cluster_num_changes
,
883 count
, tcount
, config
->cluster_highest_sessionid
,
884 config
->cluster_highest_tunnelid
, (int) (p
- buff
));
886 config
->cluster_num_changes
= 0;
888 send_heartbeat(h
.seq
, buff
, (p
-buff
) ); // Send out the heartbeat to the cluster, keeping a copy of it.
890 config
->cluster_seq_number
= (config
->cluster_seq_number
+1)%HB_MAX_SEQ
; // Next seq number to use.
894 // A structure of type 'type' has changed; Add it to the queue to send.
896 static int type_changed(int type
, int id
)
900 for (i
= 0 ; i
< config
->cluster_num_changes
; ++i
)
901 if ( cluster_changes
[i
].id
== id
&&
902 cluster_changes
[i
].type
== type
)
903 return 0; // Already marked for change.
905 cluster_changes
[i
].type
= type
;
906 cluster_changes
[i
].id
= id
;
907 ++config
->cluster_num_changes
;
909 if (config
->cluster_num_changes
> MAX_CHANGES
)
910 cluster_heartbeat(); // flush now
916 // A particular session has been changed!
917 int cluster_send_session(int sid
)
919 if (!config
->cluster_iam_master
) {
920 LOG(0, sid
, 0, "I'm not a master, but I just tried to change a session!\n");
925 LOG(0, sid
, 0, "cluster_send_session called from child process!\n");
929 return type_changed(C_CSESSION
, sid
);
932 // A particular tunnel has been changed!
933 int cluster_send_tunnel(int tid
)
935 if (!config
->cluster_iam_master
) {
936 LOG(0, 0, tid
, "I'm not a master, but I just tried to change a tunnel!\n");
940 return type_changed(C_CTUNNEL
, tid
);
945 // We're a master, and a slave has just told us that it's
946 // missed a packet. We'll resend it every packet since
947 // the last one it's seen.
949 static int cluster_catchup_slave(int seq
, in_addr_t slave
)
954 LOG(1, 0, 0, "Slave %s sent LASTSEEN with seq %d\n", fmtaddr(slave
, 0), seq
);
955 if (!config
->cluster_iam_master
) {
956 LOG(1, 0, 0, "Got LASTSEEN but I'm not a master! Redirecting it to %s.\n",
957 fmtaddr(config
->cluster_master_address
, 0));
959 peer_send_message(slave
, C_MASTER
, config
->cluster_master_address
, NULL
, 0);
963 diff
= config
->cluster_seq_number
- seq
; // How many packet do we need to send?
967 if (diff
>= HB_HISTORY_SIZE
) { // Ouch. We don't have the packet to send it!
968 LOG(0, 0, 0, "A slave asked for message %d when our seq number is %d. Killing it.\n",
969 seq
, config
->cluster_seq_number
);
970 return peer_send_message(slave
, C_KILL
, seq
, NULL
, 0);// Kill the slave. Nothing else to do.
973 LOG(1, 0, 0, "Sending %d catchup packets to slave %s\n", diff
, fmtaddr(slave
, 0) );
975 // Now resend every packet that it missed, in order.
976 while (seq
!= config
->cluster_seq_number
) {
977 s
= seq
% HB_HISTORY_SIZE
;
978 if (seq
!= past_hearts
[s
].seq
) {
979 LOG(0, 0, 0, "Tried to re-send heartbeat for %s but %d doesn't match %d! (%d,%d)\n",
980 fmtaddr(slave
, 0), seq
, past_hearts
[s
].seq
, s
, config
->cluster_seq_number
);
981 return -1; // What to do here!?
983 peer_send_data(slave
, past_hearts
[s
].data
, past_hearts
[s
].size
);
984 seq
= (seq
+1)%HB_MAX_SEQ
; // Increment to next seq number.
986 return 0; // All good!
990 // We've heard from another peer! Add it to the list
991 // that we select from at election time.
993 static int cluster_add_peer(in_addr_t peer
, time_t basetime
, pingt
*pp
, int size
)
999 // Allow for backward compatability.
1000 // Just the ping packet into a new structure to allow
1001 // for the possibility that we might have received
1002 // more or fewer elements than we were expecting.
1003 if (size
> sizeof(p
))
1006 memset( (void *) &p
, 0, sizeof(p
) );
1007 memcpy( (void *) &p
, (void *) pp
, size
);
1010 if (clusterid
!= config
->bind_address
)
1013 LOG(4, 0, 0, "Skipping ping from %s (different cluster)\n", fmtaddr(peer
, 0));
1017 for (i
= 0; i
< num_peers
; ++i
)
1019 if (peers
[i
].peer
!= peer
)
1022 // This peer already exists. Just update the timestamp.
1023 peers
[i
].basetime
= basetime
;
1024 peers
[i
].timestamp
= TIME
;
1025 peers
[i
].uptodate
= !p
.undef
;
1029 // Is this the master shutting down??
1030 if (peer
== config
->cluster_master_address
) {
1031 LOG(3, 0, 0, "Master %s %s\n", fmtaddr(config
->cluster_master_address
, 0),
1032 basetime
? "has restarted!" : "shutting down...");
1034 config
->cluster_master_address
= 0;
1035 config
->cluster_last_hb
= 0; // Force an election.
1036 cluster_check_master();
1041 LOG(4, 0, 0, "Adding %s as a peer\n", fmtaddr(peer
, 0));
1043 // Not found. Is there a stale slot to re-use?
1044 for (i
= 0; i
< num_peers
; ++i
)
1046 if (!peers
[i
].basetime
) // Shutdown
1049 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
* 10) < TIME
) // Stale.
1053 if (i
>= CLUSTER_MAX_SIZE
)
1056 LOG(0, 0, 0, "Tried to add %s as a peer, but I already have %d of them!\n", fmtaddr(peer
, 0), i
);
1060 peers
[i
].peer
= peer
;
1061 peers
[i
].basetime
= basetime
;
1062 peers
[i
].timestamp
= TIME
;
1063 peers
[i
].uptodate
= !p
.undef
;
1067 LOG(1, 0, 0, "Added %s as a new peer. Now %d peers\n", fmtaddr(peer
, 0), num_peers
);
1073 // A slave responds with C_MASTER when it gets a message which should have gone to a master.
1074 static int cluster_set_master(in_addr_t peer
, in_addr_t master
)
1076 if (config
->cluster_iam_master
) // Sanity...
1079 LOG(3, 0, 0, "Peer %s set the master to %s...\n", fmtaddr(peer
, 0),
1080 fmtaddr(master
, 1));
1082 config
->cluster_master_address
= master
;
1085 // catchup with new master
1086 peer_send_message(master
, C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
1088 // delay next election
1089 config
->cluster_last_hb
= TIME
;
1092 // run election (or reset "probed" if master was set)
1093 cluster_check_master();
1097 /* Handle the slave updating the byte counters for the master. */
1099 // Note that we don't mark the session as dirty; We rely on
1100 // the slow table walk to propogate this back out to the slaves.
1102 static int cluster_handle_bytes(char *data
, int size
)
1106 b
= (bytest
*) data
;
1108 LOG(3, 0, 0, "Got byte counter update (size %d)\n", size
);
1110 /* Loop around, adding the byte
1111 counts to each of the sessions. */
1113 while (size
>= sizeof(*b
) ) {
1114 if (b
->sid
> MAXSESSION
) {
1115 LOG(0, 0, 0, "Got C_BYTES with session #%d!\n", b
->sid
);
1116 return -1; /* Abort processing */
1119 session
[b
->sid
].pin
+= b
->pin
;
1120 session
[b
->sid
].pout
+= b
->pout
;
1122 increment_counter(&session
[b
->sid
].cin
, &session
[b
->sid
].cin_wrap
, b
->cin
);
1123 increment_counter(&session
[b
->sid
].cout
, &session
[b
->sid
].cout_wrap
, b
->cout
);
1125 session
[b
->sid
].cin_delta
+= b
->cin
;
1126 session
[b
->sid
].cout_delta
+= b
->cout
;
1129 session
[b
->sid
].last_packet
= time_now
; // Reset idle timer!
1136 LOG(0, 0, 0, "Got C_BYTES with %d bytes of trailing junk!\n", size
);
1142 // Handle receiving a session structure in a heartbeat packet.
1144 static int cluster_recv_session(int more
, uint8_t *p
)
1146 if (more
>= MAXSESSION
) {
1147 LOG(0, 0, 0, "DANGER: Received a heartbeat session id > MAXSESSION!\n");
1151 if (session
[more
].tunnel
== T_UNDEF
) {
1152 if (config
->cluster_iam_uptodate
) { // Sanity.
1153 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined session!\n");
1155 --config
->cluster_undefined_sessions
;
1159 load_session(more
, (sessiont
*) p
); // Copy session into session table..
1161 LOG(5, more
, 0, "Received session update (%d undef)\n", config
->cluster_undefined_sessions
);
1163 if (!config
->cluster_iam_uptodate
)
1164 cluster_uptodate(); // Check to see if we're up to date.
1169 static int cluster_recv_tunnel(int more
, uint8_t *p
)
1171 if (more
>= MAXTUNNEL
) {
1172 LOG(0, 0, 0, "DANGER: Received a tunnel session id > MAXTUNNEL!\n");
1176 if (tunnel
[more
].state
== TUNNELUNDEF
) {
1177 if (config
->cluster_iam_uptodate
) { // Sanity.
1178 LOG(0, 0, 0, "I thought I was uptodate but I just found an undefined tunnel!\n");
1180 --config
->cluster_undefined_tunnels
;
1184 memcpy(&tunnel
[more
], p
, sizeof(tunnel
[more
]) );
1187 // Clear tunnel control messages. These are dynamically allocated.
1188 // If we get unlucky, this may cause the tunnel to drop!
1190 tunnel
[more
].controls
= tunnel
[more
].controle
= NULL
;
1191 tunnel
[more
].controlc
= 0;
1193 LOG(5, 0, more
, "Received tunnel update\n");
1195 if (!config
->cluster_iam_uptodate
)
1196 cluster_uptodate(); // Check to see if we're up to date.
1202 // pre v5 heartbeat session structure
1209 unsigned long unique_id
;
1216 uint32_t total_cout
;
1218 uint16_t throttle_in
;
1219 uint16_t throttle_out
;
1223 in_addr_t dns1
, dns2
;
1224 routet route
[MAXROUTE
];
1230 uint8_t reserved_old_snoop
;
1231 uint8_t walled_garden
;
1233 char random_vector
[MAXTEL
];
1234 int random_vector_length
;
1236 char called
[MAXTEL
];
1237 char calling
[MAXTEL
];
1238 uint32_t tx_connect_speed
;
1239 uint32_t rx_connect_speed
;
1242 uint16_t snoop_port
;
1249 static uint8_t *convert_session(struct oldsession
*old
)
1251 static sessiont
new;
1254 memset(&new, 0, sizeof(new));
1256 new.next
= old
->next
;
1258 new.tunnel
= old
->tunnel
;
1259 new.l2tp_flags
= old
->l2tp_flags
;
1260 new.flags
= old
->flags
;
1262 new.ip_pool_index
= old
->ip_pool_index
;
1263 new.unique_id
= old
->unique_id
;
1266 new.magic
= old
->magic
;
1268 new.pout
= old
->pout
;
1269 new.cin
= old
->total_cin
;
1270 new.cout
= old
->total_cout
;
1271 new.cin_delta
= old
->cin
;
1272 new.cout_delta
= old
->cout
;
1273 new.throttle_in
= old
->throttle_in
;
1274 new.throttle_out
= old
->throttle_out
;
1275 new.filter_in
= old
->filter_in
;
1276 new.filter_out
= old
->filter_out
;
1278 new.opened
= old
->opened
;
1280 new.last_packet
= old
->last_packet
;
1281 new.dns1
= old
->dns1
;
1282 new.dns2
= old
->dns2
;
1283 new.tbf_in
= old
->tbf_in
;
1284 new.tbf_out
= old
->tbf_out
;
1285 new.random_vector_length
= old
->random_vector_length
;
1286 new.tx_connect_speed
= old
->tx_connect_speed
;
1287 new.rx_connect_speed
= old
->rx_connect_speed
;
1288 new.snoop_ip
= old
->snoop_ip
;
1289 new.snoop_port
= old
->snoop_port
;
1290 new.walled_garden
= old
->walled_garden
;
1292 memcpy(new.random_vector
, old
->random_vector
, sizeof(new.random_vector
));
1293 memcpy(new.user
, old
->user
, sizeof(new.user
));
1294 memcpy(new.called
, old
->called
, sizeof(new.called
));
1295 memcpy(new.calling
, old
->calling
, sizeof(new.calling
));
1297 for (i
= 0; i
< MAXROUTE
; i
++)
1298 memcpy(&new.route
[i
], &old
->route
[i
], sizeof(new.route
[i
]));
1300 return (uint8_t *) &new;
1304 // Process a heartbeat..
1306 // v3: added interval, timeout
1307 // v4: added table_version
1308 // v5: added ipv6, re-ordered session structure
1309 static int cluster_process_heartbeat(uint8_t *data
, int size
, int more
, uint8_t *p
, in_addr_t addr
)
1312 int s
= size
- (p
-data
);
1317 # error "need to update cluster_process_heartbeat()"
1320 // we handle versions 3 through 5
1321 if (hb_ver
< 3 || hb_ver
> HB_VERSION
) {
1322 LOG(0, 0, 0, "Received a heartbeat version that I don't support (%d)!\n", hb_ver
);
1323 return -1; // Ignore it??
1326 // Ok. It's a heartbeat packet from a cluster master!
1334 if (h
->clusterid
!= config
->bind_address
)
1335 return -1; // It's not part of our cluster.
1337 if (config
->cluster_iam_master
) { // Sanity...
1338 // Note that this MUST match the election process above!
1340 LOG(0, 0, 0, "I just got a heartbeat from master %s, but _I_ am the master!\n", fmtaddr(addr
, 0));
1342 LOG(0, 0, 0, "Heartbeat with zero basetime! Ignoring\n");
1343 return -1; // Skip it.
1347 if (h
->table_version
> config
->cluster_table_version
) {
1348 LOG(0, 0, 0, "They've seen more state changes (%" PRIu64
" vs my %" PRIu64
") so I'm gone!\n",
1349 h
->table_version
, config
->cluster_table_version
);
1354 if (h
->table_version
< config
->cluster_table_version
)
1358 if (basetime
> h
->basetime
) {
1359 LOG(0, 0, 0, "They're an older master than me so I'm gone!\n");
1364 if (basetime
< h
->basetime
)
1367 if (my_address
< addr
) { // Tie breaker.
1368 LOG(0, 0, 0, "They're a higher IP address than me, so I'm gone!\n");
1374 // Send it a unicast heartbeat to see give it a chance to die.
1375 // NOTE: It's actually safe to do seq-number - 1 without checking
1378 cluster_catchup_slave(config
->cluster_seq_number
- 1, addr
);
1380 return -1; // Skip it.
1384 // Try and guard against a stray master appearing.
1386 // Ignore heartbeats received from another master before the
1387 // timeout (less a smidgen) for the old master has elapsed.
1389 // Note that after a clean failover, the cluster_master_address
1390 // is cleared, so this doesn't run.
1392 if (config
->cluster_master_address
&& addr
!= config
->cluster_master_address
) {
1393 LOG(0, 0, 0, "Ignoring stray heartbeat from %s, current master %s has not yet timed out (last heartbeat %.1f seconds ago).\n",
1394 fmtaddr(addr
, 0), fmtaddr(config
->cluster_master_address
, 1),
1395 0.1 * (TIME
- config
->cluster_last_hb
));
1396 return -1; // ignore
1399 if (config
->cluster_seq_number
== -1) // Don't have one. Just align to the master...
1400 config
->cluster_seq_number
= h
->seq
;
1402 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1403 config
->cluster_last_hb_ver
= hb_ver
; // remember what cluster version the master is using
1405 if (config
->cluster_seq_number
!= h
->seq
) { // Out of sequence heartbeat!
1406 static int lastseen_seq
= 0;
1407 static time_t lastseen_time
= 0;
1409 // limit to once per second for a particular seq#
1410 int ask
= (config
->cluster_seq_number
!= lastseen_seq
|| time_now
!= lastseen_time
);
1412 LOG(1, 0, 0, "HB: Got seq# %d but was expecting %d. %s.\n",
1413 h
->seq
, config
->cluster_seq_number
,
1414 ask
? "Asking for resend" : "Ignoring");
1418 lastseen_seq
= config
->cluster_seq_number
;
1419 lastseen_time
= time_now
;
1420 peer_send_message(addr
, C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
1423 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1425 // Just drop the packet. The master will resend it as part of the catchup.
1429 // Save the packet in our buffer.
1430 // This is needed in case we become the master.
1431 config
->cluster_seq_number
= (h
->seq
+1)%HB_MAX_SEQ
;
1432 i
= h
->seq
% HB_HISTORY_SIZE
;
1433 past_hearts
[i
].seq
= h
->seq
;
1434 past_hearts
[i
].size
= size
;
1435 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
1438 // Check that we don't have too many undefined sessions, and
1439 // that the free session pointer is correct.
1440 cluster_check_sessions(h
->highsession
, h
->freesession
, h
->hightunnel
);
1442 if (h
->interval
!= config
->cluster_hb_interval
)
1444 LOG(2, 0, 0, "Master set ping/heartbeat interval to %u (was %u)\n",
1445 h
->interval
, config
->cluster_hb_interval
);
1447 config
->cluster_hb_interval
= h
->interval
;
1450 if (h
->timeout
!= config
->cluster_hb_timeout
)
1452 LOG(2, 0, 0, "Master set heartbeat timeout to %u (was %u)\n",
1453 h
->timeout
, config
->cluster_hb_timeout
);
1455 config
->cluster_hb_timeout
= h
->timeout
;
1458 // Ok. process the packet...
1461 type
= *((uint32_t *) p
);
1462 p
+= sizeof(uint32_t);
1463 s
-= sizeof(uint32_t);
1465 more
= *((uint32_t *) p
);
1466 p
+= sizeof(uint32_t);
1467 s
-= sizeof(uint32_t);
1470 case C_CSESSION
: { // Compressed session structure.
1471 uint8_t c
[ sizeof(sessiont
) + 2];
1473 uint8_t *orig_p
= p
;
1475 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
) );
1478 // session struct changed with v5
1481 if (size
!= sizeof(struct oldsession
)) {
1482 LOG(0, 0, 0, "DANGER: Received a v%d CSESSION that didn't decompress correctly!\n", hb_ver
);
1483 // Now what? Should exit! No-longer up to date!
1486 cluster_recv_session(more
, convert_session((struct oldsession
*) c
));
1490 if (size
!= sizeof(sessiont
) ) { // Ouch! Very very bad!
1491 LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n");
1492 // Now what? Should exit! No-longer up to date!
1496 cluster_recv_session(more
, c
);
1502 if (s
< sizeof(struct oldsession
))
1505 cluster_recv_session(more
, convert_session((struct oldsession
*) p
));
1507 p
+= sizeof(struct oldsession
);
1508 s
-= sizeof(struct oldsession
);
1512 if ( s
< sizeof(session
[more
]))
1515 cluster_recv_session(more
, p
);
1517 p
+= sizeof(session
[more
]);
1518 s
-= sizeof(session
[more
]);
1521 case C_CTUNNEL
: { // Compressed tunnel structure.
1522 uint8_t c
[ sizeof(tunnelt
) + 2];
1524 uint8_t *orig_p
= p
;
1526 size
= rle_decompress((uint8_t **) &p
, s
, c
, sizeof(c
));
1529 if (size
!= sizeof(tunnelt
) ) { // Ouch! Very very bad!
1530 LOG(0, 0, 0, "DANGER: Received a CSESSION that didn't decompress correctly!\n");
1531 // Now what? Should exit! No-longer up to date!
1535 cluster_recv_tunnel(more
, c
);
1540 if ( s
< sizeof(tunnel
[more
]))
1543 cluster_recv_tunnel(more
, p
);
1545 p
+= sizeof(tunnel
[more
]);
1546 s
-= sizeof(tunnel
[more
]);
1549 LOG(0, 0, 0, "DANGER: I received a heartbeat element where I didn't understand the type! (%d)\n", type
);
1550 return -1; // can't process any more of the packet!!
1554 if (config
->cluster_master_address
!= addr
)
1556 LOG(0, 0, 0, "My master just changed from %s to %s!\n",
1557 fmtaddr(config
->cluster_master_address
, 0), fmtaddr(addr
, 1));
1559 config
->cluster_master_address
= addr
;
1562 config
->cluster_last_hb
= TIME
; // Successfully received a heartbeat!
1563 config
->cluster_table_version
= h
->table_version
;
1567 LOG(0, 0, 0, "I got an incomplete heartbeat packet! This means I'm probably out of sync!!\n");
1572 // We got a packet on the cluster port!
1573 // Handle pings, lastseens, and heartbeats!
1575 int processcluster(char *data
, int size
, in_addr_t addr
)
1581 if (addr
== my_address
)
1582 return -1; // Ignore it. Something looped back the multicast!
1584 LOG(5, 0, 0, "Process cluster: %d bytes from %s\n", size
, fmtaddr(addr
, 0));
1586 if (s
<= 0) // Any data there??
1592 type
= *((uint32_t *) p
);
1593 p
+= sizeof(uint32_t);
1594 s
-= sizeof(uint32_t);
1596 more
= *((uint32_t *) p
);
1597 p
+= sizeof(uint32_t);
1598 s
-= sizeof(uint32_t);
1602 case C_PING
: // Update the peers table.
1603 return cluster_add_peer(addr
, more
, (pingt
*) p
, s
);
1605 case C_MASTER
: // Our master is wrong
1606 return cluster_set_master(addr
, more
);
1608 case C_LASTSEEN
: // Catch up a slave (slave missed a packet).
1609 return cluster_catchup_slave(more
, addr
);
1611 case C_FORWARD
: // Forwarded control packet. pass off to processudp.
1612 case C_FORWARD_DAE
: // Forwarded DAE packet. pass off to processdae.
1613 if (!config
->cluster_iam_master
)
1615 LOG(0, 0, 0, "I'm not the master, but I got a C_FORWARD_%s from %s?\n",
1616 type
== C_FORWARD_DAE
? "_DAE" : "", fmtaddr(addr
, 0));
1622 struct sockaddr_in a
;
1623 a
.sin_addr
.s_addr
= more
;
1625 a
.sin_port
= *(int *) p
;
1629 LOG(4, 0, 0, "Got a forwarded %spacket... (%s:%d)\n",
1630 type
== C_FORWARD_DAE
? "DAE " : "", fmtaddr(more
, 0), a
.sin_port
);
1633 if (type
== C_FORWARD_DAE
)
1634 processdae(p
, s
, &a
, sizeof(a
));
1636 processudp(p
, s
, &a
);
1641 case C_THROTTLE
: { // Receive a forwarded packet from a slave.
1642 if (!config
->cluster_iam_master
) {
1643 LOG(0, 0, 0, "I'm not the master, but I got a C_THROTTLE from %s?\n", fmtaddr(addr
, 0));
1647 tbf_queue_packet(more
, p
, s
); // The TBF id tells wether it goes in or out.
1651 // Receive a walled garden packet from a slave.
1652 if (!config
->cluster_iam_master
) {
1653 LOG(0, 0, 0, "I'm not the master, but I got a C_GARDEN from %s?\n", fmtaddr(addr
, 0));
1661 if (!config
->cluster_iam_master
) {
1662 LOG(0, 0, 0, "I'm not the master, but I got a C_BYTES from %s?\n", fmtaddr(addr
, 0));
1666 return cluster_handle_bytes(p
, s
);
1668 case C_KILL
: // The master asked us to die!? (usually because we're too out of date).
1669 if (config
->cluster_iam_master
) {
1670 LOG(0, 0, 0, "_I_ am master, but I received a C_KILL from %s! (Seq# %d)\n", fmtaddr(addr
, 0), more
);
1673 if (more
!= config
->cluster_seq_number
) {
1674 LOG(0, 0, 0, "The master asked us to die but the seq number didn't match!?\n");
1678 if (addr
!= config
->cluster_master_address
) {
1679 LOG(0, 0, 0, "Received a C_KILL from %s which doesn't match config->cluster_master_address (%s)\n",
1680 fmtaddr(addr
, 0), fmtaddr(config
->cluster_master_address
, 1));
1681 // We can only warn about it. The master might really have switched!
1684 LOG(0, 0, 0, "Received a valid C_KILL: I'm going to die now.\n");
1686 exit(0); // Lets be paranoid;
1687 return -1; // Just signalling the compiler.
1690 LOG(4, 0, 0, "Got a heartbeat from %s\n", fmtaddr(addr
, 0));
1691 return cluster_process_heartbeat(data
, size
, more
, p
, addr
);
1694 LOG(0, 0, 0, "Strange type packet received on cluster socket (%d)\n", type
);
1700 LOG(0, 0, 0, "I got a _short_ cluster heartbeat packet! This means I'm probably out of sync!!\n");
1704 //====================================================================================================
1706 int cmd_show_cluster(struct cli_def
*cli
, char *command
, char **argv
, int argc
)
1710 if (CLI_HELP_REQUESTED
)
1711 return CLI_HELP_NO_ARGS
;
1713 cli_print(cli
, "Cluster status : %s", config
->cluster_iam_master
? "Master" : "Slave" );
1714 cli_print(cli
, "My address : %s", fmtaddr(my_address
, 0));
1715 cli_print(cli
, "VIP address : %s", fmtaddr(config
->bind_address
, 0));
1716 cli_print(cli
, "Multicast address: %s", fmtaddr(config
->cluster_address
, 0));
1717 cli_print(cli
, "Multicast i'face : %s", config
->cluster_interface
);
1719 if (!config
->cluster_iam_master
) {
1720 cli_print(cli
, "My master : %s (last heartbeat %.1f seconds old)",
1721 config
->cluster_master_address
1722 ? fmtaddr(config
->cluster_master_address
, 0)
1724 0.1 * (TIME
- config
->cluster_last_hb
));
1725 cli_print(cli
, "Uptodate : %s", config
->cluster_iam_uptodate
? "Yes" : "No");
1726 cli_print(cli
, "Table version # : %" PRIu64
, config
->cluster_table_version
);
1727 cli_print(cli
, "Next sequence number expected: %d", config
->cluster_seq_number
);
1728 cli_print(cli
, "%d sessions undefined of %d", config
->cluster_undefined_sessions
, config
->cluster_highest_sessionid
);
1729 cli_print(cli
, "%d tunnels undefined of %d", config
->cluster_undefined_tunnels
, config
->cluster_highest_tunnelid
);
1731 cli_print(cli
, "Table version # : %" PRIu64
, config
->cluster_table_version
);
1732 cli_print(cli
, "Next heartbeat # : %d", config
->cluster_seq_number
);
1733 cli_print(cli
, "Highest session : %d", config
->cluster_highest_sessionid
);
1734 cli_print(cli
, "Highest tunnel : %d", config
->cluster_highest_tunnelid
);
1735 cli_print(cli
, "%d changes queued for sending", config
->cluster_num_changes
);
1737 cli_print(cli
, "%d peers.", num_peers
);
1740 cli_print(cli
, "%20s %10s %8s", "Address", "Basetime", "Age");
1741 for (i
= 0; i
< num_peers
; ++i
) {
1742 cli_print(cli
, "%20s %10u %8d", fmtaddr(peers
[i
].peer
, 0),
1743 peers
[i
].basetime
, TIME
- peers
[i
].timestamp
);
1749 // Simple run-length-encoding compression.
1751 // 1 byte < 128 = count of non-zero bytes following. // Not legal to be zero.
1752 // n non-zero bytes;
1754 // 1 byte > 128 = (count - 128) run of zero bytes. //
1756 // count == 0 indicates end of compressed stream.
1758 // Compress from 'src' into 'dst'. return number of bytes
1760 // Updates *src_p to indicate 1 past last bytes used.
1762 // We could get an extra byte in the zero runs by storing (count-1)
1763 // but I'm playing it safe.
1765 // Worst case is a 50% expansion in space required (trying to
1766 // compress { 0x00, 0x01 } * N )
1767 static int rle_compress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
)
1770 int orig_dsize
= dsize
;
1774 while (ssize
> 0 && dsize
> 2) {
1776 x
= dst
++; --dsize
; // Reserve space for count byte..
1778 if (*src
) { // Copy a run of non-zero bytes.
1779 while (*src
&& count
< 127 && ssize
> 0 && dsize
> 1) { // Count number of non-zero bytes.
1784 *x
= count
; // Store number of non-zero bytes. Guarenteed to be non-zero!
1786 } else { // Compress a run of zero bytes.
1787 while (*src
== 0 && count
< 127 && ssize
> 0) {
1796 *dst
++ = 0x0; // Add Stop byte.
1800 return (orig_dsize
- dsize
);
1804 // Decompress the buffer into **p.
1805 // 'psize' is the size of the decompression buffer available.
1807 // Returns the number of bytes decompressed.
1809 // Decompresses from '*src_p' into 'dst'.
1810 // Return the number of dst bytes used.
1811 // Updates the 'src_p' pointer to point to the
1812 // first un-used byte.
1813 static int rle_decompress(uint8_t **src_p
, int ssize
, uint8_t *dst
, int dsize
)
1816 int orig_dsize
= dsize
;
1819 while (ssize
>0 && dsize
> 0) { // While there's more to decompress, and there's room in the decompress buffer...
1820 count
= *src
++; --ssize
; // get the count byte from the source.
1821 if (count
== 0x0) // End marker reached? If so, finish.
1824 if (count
& 0x80) { // Decompress a run of zeros
1825 for (count
&= 0x7f ; count
> 0 && dsize
> 0; --count
) {
1829 } else { // Copy run of non-zero bytes.
1830 for ( ; count
> 0 && ssize
&& dsize
; --count
) { // Copy non-zero bytes across.
1837 return (orig_dsize
- dsize
);