1 // L2TPNS Clustering Stuff
3 char const *cvs_id_cluster
= "$Id: cluster.c,v 1.6 2004/07/05 06:54:01 bodea Exp $";
8 #include <sys/socket.h>
9 #include <netinet/in.h>
10 #include <arpa/inet.h>
11 #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 int cluster_sockfd
= 0; // The filedescriptor for the cluster communications port.
43 ipt 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.
47 static int hsess
, fsess
; // Saved copies of the highest used session id, and the first free one.
49 #define MAX_HEART_SIZE (8192) // Maximum size of heartbeat packet. Must be less than max IP packet size :)
50 #define MAX_CHANGES (MAX_HEART_SIZE/(sizeof(sessiont) + sizeof(int) ) - 2) // Assumes a session is the biggest type!
55 } cluster_changes
[MAX_CHANGES
]; // Queue of changed structures that need to go out when next heartbeat.
60 char data
[MAX_HEART_SIZE
];
61 } past_hearts
[HB_HISTORY_SIZE
]; // Ring buffer of heartbeats that we've recently sent out. Needed so
62 // we can re-transmit if needed.
69 } peers
[CLUSTER_MAX_SIZE
]; // List of all the peers we've heard from.
70 static int num_peers
; // Number of peers in list.
71 static int have_peers
; // At least one peer
73 int rle_decompress(u8
** src_p
, int ssize
, u8
*dst
, int dsize
);
74 int rle_compress(u8
** src_p
, int ssize
, u8
*dst
, int dsize
);
77 // Create a listening socket
79 // This joins the cluster multi-cast group.
83 struct sockaddr_in addr
;
84 struct sockaddr_in interface_addr
;
89 config
->cluster_undefined_sessions
= MAXSESSION
-1;
90 config
->cluster_undefined_tunnels
= MAXTUNNEL
-1;
92 if (!config
->cluster_address
)
94 if (!*config
->cluster_interface
)
97 cluster_sockfd
= socket(AF_INET
, SOCK_DGRAM
, UDP
);
99 memset(&addr
, 0, sizeof(addr
));
100 addr
.sin_family
= AF_INET
;
101 addr
.sin_port
= htons(CLUSTERPORT
);
102 addr
.sin_addr
.s_addr
= INADDR_ANY
;
103 setsockopt(cluster_sockfd
, SOL_SOCKET
, SO_REUSEADDR
, &addr
, sizeof(addr
));
105 if (bind(cluster_sockfd
, (void *) &addr
, sizeof(addr
)) < 0)
107 log(0, 0, 0, 0, "Failed to bind cluster socket: %s\n", strerror(errno
));
111 strcpy(ifr
.ifr_name
, config
->cluster_interface
);
112 if (ioctl(cluster_sockfd
, SIOCGIFADDR
, &ifr
) < 0) {
113 log(0, 0, 0, 0, "Failed to get interface address for (%s): %s\n", config
->cluster_interface
, strerror(errno
));
117 memcpy(&interface_addr
, &ifr
.ifr_addr
, sizeof(interface_addr
) );
118 my_address
= interface_addr
.sin_addr
.s_addr
;
120 // Join multicast group.
121 mreq
.imr_multiaddr
.s_addr
= config
->cluster_address
;
122 mreq
.imr_interface
= interface_addr
.sin_addr
;
125 opt
= 0; // Turn off multicast loopback.
126 setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_LOOP
, &opt
, sizeof(opt
));
128 if (setsockopt(cluster_sockfd
, IPPROTO_IP
, IP_ADD_MEMBERSHIP
, &mreq
, sizeof(mreq
)) < 0) {
129 log(0, 0, 0, 0, "Failed to setsockopt (join mcast group): %s\n", strerror(errno
));
133 if (setsockopt (cluster_sockfd
, IPPROTO_IP
, IP_MULTICAST_IF
, &interface_addr
, sizeof(interface_addr
)) < 0) {
134 log(0, 0, 0, 0, "Failed to setsockopt (set mcast interface): %s\n", strerror(errno
));
138 config
->cluster_last_hb
= TIME
;
139 config
->cluster_seq_number
= -1;
141 return cluster_sockfd
;
146 // Send a chunk of data to the entire cluster (usually via the multicast
150 int cluster_send_data(void *data
, int datalen
)
152 struct sockaddr_in addr
= {0};
154 if (!cluster_sockfd
) return -1;
155 if (!config
->cluster_address
) return 0;
157 addr
.sin_addr
.s_addr
= config
->cluster_address
;
158 addr
.sin_port
= htons(CLUSTERPORT
);
159 addr
.sin_family
= AF_INET
;
161 // log_hex(4, "Cluster send", data, datalen); // VERY big data packets. How about we don't..
163 log(5,0,0,0, "Cluster send data: %d bytes\n", datalen
);
165 if (sendto(cluster_sockfd
, data
, datalen
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
167 log(0, 0, 0, 0, "sendto: %s\n", strerror(errno
));
175 // Add a chunk of data to a heartbeat packet.
176 // Maintains the format. Assumes that the caller
177 // has passed in a big enough buffer!
179 static void add_type(char ** p
, int type
, int more
, char * data
, int size
)
181 * ( (u32
*)(*p
) ) = type
;
184 * ( (u32
*)(*p
) ) = more
;
187 if (data
&& size
> 0) {
188 memcpy(*p
, data
, size
);
193 void cluster_uptodate(void)
195 if (config
->cluster_iam_uptodate
)
198 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
)
201 config
->cluster_iam_uptodate
= 1;
203 log(0,0,0,0, "Now uptodate with master.\n");
205 // If we're not a master, or if we have no slaves
206 // then start taking traffic..
207 if (!config
->cluster_iam_master
|| !have_peers
)
211 bgp_enable_routing(1);
214 if (config
->send_garp
)
215 send_garp(config
->bind_address
); // Start taking traffic.
220 // Send a unicast UDP packet to a peer with 'data' as the
223 int peer_send_data(u32 peer
, char * data
, int size
)
225 struct sockaddr_in addr
= {0};
227 if (!cluster_sockfd
) return -1;
228 if (!config
->cluster_address
) return 0;
233 addr
.sin_addr
.s_addr
= peer
;
234 addr
.sin_port
= htons(CLUSTERPORT
);
235 addr
.sin_family
= AF_INET
;
237 log_hex(5, "Peer send", data
, size
);
239 if (sendto(cluster_sockfd
, data
, size
, MSG_NOSIGNAL
, (void *) &addr
, sizeof(addr
)) < 0)
241 log(0, 0, 0, 0, "sendto: %s\n", strerror(errno
));
249 // Send a structured message to a peer with a single element of type 'type'.
251 int peer_send_message(u32 peer
, int type
, int more
, char * data
, int size
)
253 char buf
[65536]; // Vast overkill.
256 log(4,0,0,0, "Sending message to peer (type %d, more %d, size %d)\n", type
, more
, size
);
257 add_type(&p
, type
, more
, data
, size
);
259 return peer_send_data(peer
, buf
, (p
-buf
) );
263 // Forward a state changing packet to the master.
265 // The master just processes the payload as if it had
266 // received it off the tap device.
268 int master_forward_packet(char * data
, int size
, u32 addr
, int port
)
270 char buf
[65536]; // Vast overkill.
273 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
276 log(4,0,0,0, "Forwarding packet from %s to master (size %d)\n", inet_toa(addr
), size
);
279 add_type(&p
, C_FORWARD
, addr
, (char*) &port
, sizeof(port
) );
280 memcpy(p
, data
, size
);
283 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
) );
288 // Forward a throttled packet to the master for handling.
290 // The master just drops the packet into the appropriate
291 // token bucket queue, and lets normal processing take care
294 int master_throttle_packet(int tbfid
, char * data
, int size
)
296 char buf
[65536]; // Vast overkill.
299 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
302 log(4,0,0,0, "Throttling packet master (size %d, tbfid %d)\n", size
, tbfid
);
304 add_type(&p
, C_THROTTLE
, tbfid
, data
, size
);
306 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
) );
311 // Forward a walled garden packet to the master for handling.
313 // The master just writes the packet straight to the tun
314 // device (where is will normally loop through the
315 // firewall rules, and come back in on the tun device)
317 // (Note that this must be called with the tun header
318 // as the start of the data).
319 int master_garden_packet(sessionidt s
, char *data
, int size
)
321 char buf
[65536]; // Vast overkill.
324 if (!config
->cluster_master_address
) // No election has been held yet. Just skip it.
327 log(4,0,0,0, "Walled garden packet to master (size %d)\n", size
);
329 add_type(&p
, C_GARDEN
, s
, data
, size
);
331 return peer_send_data(config
->cluster_master_address
, buf
, (p
-buf
));
336 // Send a chunk of data as a heartbeat..
337 // We save it in the history buffer as we do so.
339 static void send_heartbeat(int seq
, char * data
, int size
)
343 if (size
> sizeof(past_hearts
[0].data
)) {
344 log(0,0,0,0, "Tried to heartbeat something larger than the maximum packet!\n");
348 i
= seq
% HB_HISTORY_SIZE
;
349 past_hearts
[i
].seq
= seq
;
350 past_hearts
[i
].size
= size
;
351 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
352 cluster_send_data(data
, size
);
356 // Send an 'i am alive' message to every machine in the cluster.
358 void cluster_send_ping(time_t basetime
)
360 char buff
[100 + sizeof(pingt
)];
364 if (config
->cluster_iam_master
&& basetime
) // We're heartbeating so no need to ping.
367 log(5,0,0,0, "Sending cluster ping...\n");
370 x
.addr
= config
->bind_address
;
371 x
.undef
= config
->cluster_undefined_sessions
+ config
->cluster_undefined_tunnels
;
372 x
.basetime
= basetime
;
374 add_type(&p
, C_PING
, basetime
, (char *) &x
, sizeof(x
));
375 cluster_send_data(buff
, (p
-buff
) );
379 // Walk the session counters looking for non-zero ones to send
380 // to the master. We send up to 100 of them at one time.
381 // We examine a maximum of 2000 sessions.
382 // (50k max session should mean that we normally
383 // examine the entire session table every 25 seconds).
385 #define MAX_B_RECS (400)
386 void master_update_counts(void)
389 bytest b
[MAX_B_RECS
+1];
391 if (config
->cluster_iam_master
) // Only happens on the slaves.
394 if (!config
->cluster_master_address
) // If we don't have a master, skip it for a while.
397 i
= MAX_B_RECS
* 5; // Examine max 2000 sessions;
398 if (config
->cluster_highest_sessionid
> i
)
399 i
= config
->cluster_highest_sessionid
;
401 for ( c
= 0; i
> 0 ; --i
) {
402 // Next session to look at.
403 walk_session_number
++;
404 if ( walk_session_number
> config
->cluster_highest_sessionid
)
405 walk_session_number
= 1;
407 if (!sess_count
[walk_session_number
].cin
&& !sess_count
[walk_session_number
].cout
)
408 continue; // Unused. Skip it.
410 b
[c
].sid
= walk_session_number
;
411 b
[c
].in
= sess_count
[walk_session_number
].cin
;
412 b
[c
].out
= sess_count
[walk_session_number
].cout
;
414 if (++c
> MAX_B_RECS
) // Send a max of 400 elements in a packet.
418 sess_count
[walk_session_number
].cin
= sess_count
[walk_session_number
].cout
= 0;
421 if (!c
) // Didn't find any that changes. Get out of here!
425 // Forward the data to the master.
426 log(4,0,0,0, "Sending byte counters to master (%d elements)\n", c
);
427 peer_send_message(config
->cluster_master_address
, C_BYTES
, c
, (char*) &b
, sizeof(b
[0]) * c
);
432 // Check that we have a master. If it's been too
433 // long since we heard from a master then hold an election.
435 void cluster_check_master(void)
437 int i
, count
, tcount
, high_sid
= 0;
439 int had_peers
= have_peers
;
441 static int probed
= 0;
443 // Is the master late? If so, try probing it...
444 if (TIME
> (config
->cluster_last_hb
+ config
->cluster_hb_timeout
/8 + 11)) {
446 if (config
->cluster_master_address
) {
447 peer_send_message(config
->cluster_master_address
,
448 C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
452 } else { // We got a recent heartbeat; reset the probe flag.
456 if (TIME
< (config
->cluster_last_hb
+ config
->cluster_hb_timeout
) )
457 return; // Everything's ok. return.
459 if (!config
->cluster_iam_master
)
460 log(0,0,0,0, "Master timed out! Holding election...\n");
462 config
->cluster_last_hb
= TIME
+ 1;
464 for (i
= have_peers
= 0; i
< num_peers
; ++i
) {
465 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
) < t
)
466 continue; // Stale peer! Skip them.
468 if (!peers
[i
].basetime
)
469 continue; // Shutdown peer! Skip them.
472 if (peers
[i
].basetime
< basetime
) {
473 log(1,0,0,0, "Expecting %s to become master\n", inet_toa(peers
[i
].peer
) );
474 return; // They'll win the election. Get out of here.
477 if (peers
[i
].basetime
== basetime
&&
478 peers
[i
].peer
> my_address
) {
479 log(1,0,0,0, "Expecting %s to become master\n", inet_toa(peers
[i
].peer
) );
480 return; // They'll win the election. Wait for them to come up.
484 if (config
->cluster_iam_master
) // If we're the master, we've already won
487 // master lost all slaves, need to handle traffic ourself
488 if (bgp_configured
&& had_peers
&& !have_peers
)
489 bgp_enable_routing(1);
494 // Wow. it's been ages since I last heard a heartbeat
495 // and I'm better than an of my peers so it's time
496 // to become a master!!!
498 config
->cluster_iam_master
= 1;
499 config
->cluster_master_address
= 0;
501 log(0,0,0,0, "I am declaring myself the master!\n");
504 if (bgp_configured
&& have_peers
)
505 bgp_enable_routing(0); /* stop handling traffic */
508 if (config
->cluster_seq_number
== -1)
509 config
->cluster_seq_number
= 0;
512 // Go through and mark all the tunnels as defined.
513 // Count the highest used tunnel number as well.
515 config
->cluster_highest_tunnelid
= 0;
516 for (i
= 0, tcount
= 0; i
< MAXTUNNEL
; ++i
) {
517 if (tunnel
[i
].state
== TUNNELUNDEF
)
518 tunnel
[i
].state
= TUNNELFREE
;
520 if (tunnel
[i
].state
!= TUNNELFREE
&& i
> config
->cluster_highest_tunnelid
)
521 config
->cluster_highest_tunnelid
= i
;
525 // Go through and mark all the sessions as being defined.
526 // reset the idle timeouts.
527 // add temporary byte counters to permanent ones.
528 // Re-string the free list.
529 // Find the ID of the highest session.
532 config
->cluster_highest_sessionid
= 0;
533 for (i
= 0, count
= 0; i
< MAXSESSION
; ++i
) {
534 if (session
[i
].tunnel
== T_UNDEF
) {
535 session
[i
].tunnel
= T_FREE
;
539 if (session
[i
].tunnel
== T_FREE
) { // Unused session. Add to free list.
540 session
[last_free
].next
= i
;
545 // Reset all the idle timeouts..
546 session
[i
].last_packet
= time_now
;
548 // Accumulate un-sent byte counters.
549 session
[i
].cin
+= sess_count
[i
].cin
;
550 session
[i
].cout
+= sess_count
[i
].cout
;
551 session
[i
].total_cin
+= sess_count
[i
].cin
;
552 session
[i
].total_cout
+= sess_count
[i
].cout
;
554 sess_count
[i
].cin
= sess_count
[i
].cout
= 0;
556 session
[i
].radius
= 0; // Reset authentication as the radius blocks aren't up to date.
558 if (session
[i
].sid
>= high_sid
) // This is different to the index into the session table!!!
559 high_sid
= session
[i
].sid
+1;
562 session
[i
].tbf_in
= session
[i
].tbf_out
= 0; // Remove stale pointers from old master.
563 throttle_session(i
, session
[i
].throttle
);
565 // I'm unsure about this. --mo
566 // It's potentially a good thing, but it could send a
568 // if (session[i].throttle)
569 // cluster_send_session(s); // Tell the slaves about the new tbf indexes.
571 if (session
[i
].tunnel
!= T_FREE
&& i
> config
->cluster_highest_sessionid
)
572 config
->cluster_highest_sessionid
= i
;
576 session
[last_free
].next
= 0; // End of chain.
577 last_sid
= high_sid
; // Keep track of the highest used session ID.
581 rebuild_address_pool();
583 // If we're not the very first master, this is a big issue!
585 log(0,0,0,0, "Warning: Fixed %d uninitialized sessions in becoming master!\n", count
);
587 config
->cluster_undefined_sessions
= 0;
588 config
->cluster_undefined_tunnels
= 0;
591 // FIXME. We need to fix up the tunnel control message
592 // queue here! There's a number of other variables we
593 // should also update.
599 // Check that our session table is validly matching what the
600 // master has in mind.
602 // In particular, if we have too many sessions marked 'undefined'
603 // we fix it up here, and we ensure that the 'first free session'
606 static void cluster_check_sessions(int highsession
, int freesession_ptr
, int hightunnel
)
610 sessionfree
= freesession_ptr
; // Keep the freesession ptr valid.
612 if (config
->cluster_iam_uptodate
)
615 if (highsession
> config
->cluster_undefined_sessions
&& hightunnel
> config
->cluster_undefined_tunnels
)
618 // Clear out defined sessions, counting the number of
620 config
->cluster_undefined_sessions
= 0;
621 for (i
= 1 ; i
< MAXSESSION
; ++i
) {
622 if (i
> highsession
) {
623 session
[i
].tunnel
= 0; // Defined.
626 if (session
[i
].tunnel
!= T_UNDEF
)
628 ++config
->cluster_undefined_sessions
;
631 // Clear out defined tunnels, counting the number of
633 config
->cluster_undefined_tunnels
= 0;
634 for (i
= 1 ; i
< MAXTUNNEL
; ++i
) {
635 if (i
> hightunnel
) {
636 tunnel
[i
].state
= TUNNELFREE
; // Defined.
639 if (tunnel
[i
].state
!= TUNNELUNDEF
)
641 ++config
->cluster_undefined_tunnels
;
645 if (config
->cluster_undefined_sessions
|| config
->cluster_undefined_tunnels
) {
646 log(2,0,0,0, "Cleared undefined sessions/tunnels. %d sess (high %d), %d tunn (high %d)\n",
647 config
->cluster_undefined_sessions
, highsession
, config
->cluster_undefined_tunnels
, hightunnel
);
651 // Are we up to date?
653 if (!config
->cluster_iam_uptodate
)
657 int hb_add_type(char **p
, int type
, int id
)
660 case C_CSESSION
: { // Compressed C_SESSION.
661 u8 c
[sizeof(sessiont
) * 2]; // Bigger than worst case.
662 u8
*d
= (u8
*) &session
[id
];
666 size
= rle_compress( &d
, sizeof(sessiont
), c
, sizeof(c
) );
668 // Did we compress the full structure, and is the size actually
670 if ( (d
- orig
) == sizeof(sessiont
) && size
< sizeof(sessiont
) ) {
671 add_type(p
, C_CSESSION
, id
, (char*) c
, size
);
674 // Failed to compress : Fall through.
676 case C_SESSION
: add_type(p
, C_SESSION
, id
,
677 (char*) &session
[id
], sizeof(sessiont
));
680 case C_CTUNNEL
: { // Compressed C_TUNNEL
681 u8 c
[sizeof(tunnelt
) * 2]; // Bigger than worst case.
682 u8
*d
= (u8
*) &tunnel
[id
];
686 size
= rle_compress( &d
, sizeof(tunnelt
), c
, sizeof(c
) );
688 // Did we compress the full structure, and is the size actually
690 if ( (d
- orig
) == sizeof(tunnelt
) && size
< sizeof(tunnelt
) ) {
691 add_type(p
, C_CTUNNEL
, id
, c
, size
);
694 // Failed to compress : Fall through.
696 case C_TUNNEL
: add_type(p
, C_TUNNEL
, id
,
697 (char*) &tunnel
[id
], sizeof(tunnelt
));
700 log(0,0,0,0, "Found an invalid type in heart queue! (%d)\n", type
);
707 // Send a heartbeat, incidently sending out any queued changes..
709 void cluster_heartbeat(int highsession
, int freesession
, int hightunnel
)
711 int i
, count
= 0, tcount
= 0;
712 char buff
[MAX_HEART_SIZE
+ sizeof(heartt
) + sizeof(int) ];
716 if (!config
->cluster_iam_master
) // Only the master does this.
721 // Fill out the heartbeat header.
722 h
.version
= HB_VERSION
;
723 h
.seq
= config
->cluster_seq_number
;
724 h
.basetime
= basetime
;
725 h
.clusterid
= config
->bind_address
; // Will this do??
726 h
.basetime
= basetime
;
727 h
.highsession
= highsession
;
728 h
.freesession
= freesession
;
729 h
.hightunnel
= hightunnel
;
730 h
.size_sess
= sizeof(sessiont
); // Just in case.
731 h
.size_tunn
= sizeof(tunnelt
);
733 add_type(&p
, C_HEARTBEAT
, HB_VERSION
, (char*) &h
, sizeof(h
) );
735 for (i
= 0; i
< config
->cluster_num_changes
; ++i
) {
736 hb_add_type(&p
, cluster_changes
[i
].type
, cluster_changes
[i
].id
);
739 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
740 log(0,0,0,0, "FATAL: Overran the heartbeat buffer! This is fatal. Exiting. (size %d)\n", p
- buff
);
745 // Fill out the packet with sessions from the session table...
746 // (not forgetting to leave space so we can get some tunnels in too )
747 while ( (p
+ sizeof(u32
) * 2 + sizeof(sessiont
) * 2 ) < (buff
+ MAX_HEART_SIZE
) ) {
749 if (!walk_session_number
) // session #0 isn't valid.
750 ++walk_session_number
;
752 if (count
>= highsession
) // If we're a small cluster, don't go wild.
755 hb_add_type(&p
, C_CSESSION
, walk_session_number
);
756 walk_session_number
= (1+walk_session_number
)%(highsession
+1); // +1 avoids divide by zero.
758 ++count
; // Count the number of extra sessions we're sending.
762 // Fill out the packet with tunnels from the tunnel table...
763 // This effectively means we walk the tunnel table more quickly
764 // than the session table. This is good because stuffing up a
765 // tunnel is a much bigger deal than stuffing up a session.
767 while ( (p
+ sizeof(u32
) * 2 + sizeof(tunnelt
) ) < (buff
+ MAX_HEART_SIZE
) ) {
769 if (!walk_tunnel_number
) // tunnel #0 isn't valid.
770 ++walk_tunnel_number
;
772 if (tcount
>= config
->cluster_highest_tunnelid
)
775 hb_add_type(&p
, C_CTUNNEL
, walk_tunnel_number
);
776 walk_tunnel_number
= (1+walk_tunnel_number
)%(config
->cluster_highest_tunnelid
+1); // +1 avoids divide by zero.
782 // Did we do something wrong?
783 if (p
> (buff
+ sizeof(buff
))) { // Did we somehow manage to overun the buffer?
784 log(0,0,0,0, "Overran the heartbeat buffer now! This is fatal. Exiting. (size %d)\n", p
- buff
);
788 log(3,0,0,0, "Sending heartbeat #%d with %d changes (%d x-sess, %d x-tunnels, %d highsess, %d hightun size %d)\n",
789 h
.seq
, config
->cluster_num_changes
, count
, tcount
, config
->cluster_highest_sessionid
,
790 config
->cluster_highest_tunnelid
, (p
-buff
));
792 config
->cluster_num_changes
= 0;
794 send_heartbeat(h
.seq
, buff
, (p
-buff
) ); // Send out the heartbeat to the cluster, keeping a copy of it.
796 config
->cluster_seq_number
= (config
->cluster_seq_number
+1)%HB_MAX_SEQ
; // Next seq number to use.
800 // A structure of type 'type' has changed; Add it to the queue to send.
802 int type_changed(int type
, int id
)
806 for (i
= 0 ; i
< config
->cluster_num_changes
; ++i
)
807 if ( cluster_changes
[i
].id
== id
&&
808 cluster_changes
[i
].type
== type
)
809 return 0; // Already marked for change.
811 cluster_changes
[i
].type
= type
;
812 cluster_changes
[i
].id
= id
;
813 ++config
->cluster_num_changes
;
815 if (config
->cluster_num_changes
> MAX_CHANGES
)
816 cluster_heartbeat(config
->cluster_highest_sessionid
, fsess
, config
->cluster_highest_tunnelid
);
822 // A particular session has been changed!
823 int cluster_send_session(int sid
)
825 if (!config
->cluster_iam_master
) {
826 log(0,0,sid
,0, "I'm not a master, but I just tried to change a session!\n");
830 return type_changed(C_CSESSION
, sid
);
833 // A particular tunnel has been changed!
834 int cluster_send_tunnel(int tid
)
836 if (!config
->cluster_iam_master
) {
837 log(0,0,0,tid
, "I'm not a master, but I just tried to change a tunnel!\n");
841 return type_changed(C_CTUNNEL
, tid
);
846 // We're a master, and a slave has just told us that it's
847 // missed a packet. We'll resend it every packet since
848 // the last one it's seen.
850 int cluster_catchup_slave(int seq
, u32 slave
)
855 log(1,0,0,0, "Slave %s sent LASTSEEN with seq %d\n", inet_toa(slave
), seq
);
857 diff
= config
->cluster_seq_number
- seq
; // How many packet do we need to send?
861 if (diff
>= HB_HISTORY_SIZE
) { // Ouch. We don't have the packet to send it!
862 log(0,0,0,0, "A slaved asked for message %d when our seq number is %d. Killing it.\n",
863 seq
, config
->cluster_seq_number
);
864 return peer_send_message(slave
, C_KILL
, seq
, NULL
, 0);// Kill the slave. Nothing else to do.
867 // Now resend every packet that it missed, in order.
868 while (seq
!= config
->cluster_seq_number
) {
869 s
= seq
%HB_HISTORY_SIZE
;
870 if (seq
!= past_hearts
[s
].seq
) {
871 log(0,0,0,0, "Tried to re-send heartbeat for %s but %d doesn't match %d! (%d,%d)\n",
872 inet_toa(slave
), seq
, past_hearts
[s
].seq
, s
, config
->cluster_seq_number
);
873 return -1; // What to do here!?
875 peer_send_data(slave
, past_hearts
[s
].data
, past_hearts
[s
].size
);
876 seq
= (seq
+1)%HB_MAX_SEQ
; // Increment to next seq number.
878 return 0; // All good!
882 // We've heard from another peer! Add it to the list
883 // that we select from at election time.
885 int cluster_add_peer(u32 peer
, time_t basetime
, pingt
*p
)
891 if (clusterid
!= config
->bind_address
)
894 log(4,0,0,0, "Skipping ping from %s (different cluster)\n", inet_toa(peer
));
898 // Is this the master shutting down??
899 if (peer
== config
->cluster_master_address
&& !basetime
) {
900 config
->cluster_master_address
= 0;
901 config
->cluster_last_hb
= 0; // Force an election.
902 cluster_check_master();
906 for (i
= 0; i
< num_peers
; ++i
)
908 if (peers
[i
].peer
!= peer
)
911 // This peer already exists. Just update the timestamp.
912 peers
[i
].basetime
= basetime
;
913 peers
[i
].timestamp
= TIME
;
919 log(4,0,0,0, "Adding %s as a peer\n", inet_toa(peer
));
921 // Not found. Is there a stale slot to re-use?
922 for (i
= 0; i
< num_peers
; ++i
)
924 if (peers
[i
].peer
!= peer
)
926 if ((peers
[i
].timestamp
+ config
->cluster_hb_timeout
* 10) < TIME
) // Stale.
930 if (i
>= CLUSTER_MAX_SIZE
)
933 log(0,0,0,0, "Tried to add %s as a peer, but I already have %d of them!\n", inet_toa(peer
), i
);
937 peers
[i
].peer
= peer
;
938 peers
[i
].basetime
= basetime
;
939 peers
[i
].timestamp
= TIME
;
943 log(1,0,0,0, "Added %s as a new peer. Now %d peers\n", inet_toa(peer
), num_peers
);
947 /* drop routes if we've now got a peer */
948 if (bgp_configured
&& config
->cluster_iam_master
&& !have_peers
)
949 bgp_enable_routing(0);
957 /* Handle the slave updating the byte counters for the master. */
959 // Note that we don't mark the session as dirty; We rely on
960 // the slow table walk to propogate this back out to the slaves.
962 int cluster_handle_bytes(char * data
, int size
)
968 log(3,0,0,0, "Got byte counter update (size %d)\n", size
);
970 /* Loop around, adding the byte
971 counts to each of the sessions. */
973 while (size
>= sizeof(*b
) ) {
974 if (b
->sid
> MAXSESSION
) {
975 log(0,0,0,0, "Got C_BYTES with session #%d!\n", b
->sid
);
976 return -1; /* Abort processing */
979 session
[b
->sid
].total_cin
+= b
->in
;
980 session
[b
->sid
].total_cout
+= b
->out
;
982 session
[b
->sid
].cin
+= b
->in
;
983 session
[b
->sid
].cout
+= b
->out
;
984 session
[b
->sid
].last_packet
= time_now
; // Reset idle timer!
991 log(0,0,0,0, "Got C_BYTES with %d bytes of trailing junk!\n", size
);
997 // Handle receiving a session structure in a heartbeat packet.
999 static int cluster_recv_session(int more
, u8
* p
)
1001 if (more
>= MAXSESSION
) {
1002 log(0,0,0,0, "DANGER: Received a heartbeat session id > MAXSESSION!\n");
1006 if (session
[more
].tunnel
== T_UNDEF
) {
1007 if (config
->cluster_iam_uptodate
) { // Sanity.
1008 log(0,0,0,0, "I thought I was uptodate but I just found an undefined session!\n");
1010 --config
->cluster_undefined_sessions
;
1014 load_session(more
, (sessiont
*) p
); // Copy session into session table..
1016 log(5,0,more
,0, "Received session update (%d undef)\n", config
->cluster_undefined_sessions
);
1018 if (!config
->cluster_iam_uptodate
)
1019 cluster_uptodate(); // Check to see if we're up to date.
1023 static int cluster_recv_tunnel(int more
, u8
*p
)
1025 if (more
>= MAXTUNNEL
) {
1026 log(0,0,0,0, "DANGER: Received a tunnel session id > MAXTUNNEL!\n");
1030 if (tunnel
[more
].state
== TUNNELUNDEF
) {
1031 if (config
->cluster_iam_uptodate
) { // Sanity.
1032 log(0,0,0,0, "I thought I was uptodate but I just found an undefined tunnel!\n");
1034 --config
->cluster_undefined_tunnels
;
1038 memcpy(&tunnel
[more
], p
, sizeof(tunnel
[more
]) );
1041 // Clear tunnel control messages. These are dynamically allocated.
1042 // If we get unlucky, this may cause the tunnel to drop!
1044 tunnel
[more
].controls
= tunnel
[more
].controle
= NULL
;
1045 tunnel
[more
].controlc
= 0;
1047 log(5,0,0,more
, "Received tunnel update\n");
1049 if (!config
->cluster_iam_uptodate
)
1050 cluster_uptodate(); // Check to see if we're up to date.
1057 // Process a version one heartbeat..
1059 static int cluster_process_heartbeat_v2(u8
* data
, int size
, int more
, u8
* p
, u32 addr
)
1062 int s
= size
- (p
-data
);
1065 if (more
!= HB_VERSION
) {
1066 log(0,0,0,0, "Received a heartbeat version that I don't understand!\n");
1067 return -1; // Ignore it??
1069 // Ok. It's a heartbeat packet from a cluster master!
1078 if (h
->clusterid
!= config
->bind_address
)
1079 return -1; // It's not part of our cluster.
1081 if (config
->cluster_iam_master
) { // Sanity...
1082 // Note that this MUST match the election process above!
1084 log(0,0,0,0, "I just got a packet claiming to be from a master but _I_ am the master!\n");
1086 log(0,0,0,0, "Heartbeat from addr %s with zero basetime!\n", inet_toa(addr
) );
1087 return -1; // Skip it.
1089 if (basetime
> h
->basetime
) {
1090 log(0,0,0,0, "They're (%s) an older master than me so I'm gone!\n", inet_toa(addr
));
1094 if (basetime
== h
->basetime
&& my_address
< addr
) { // Tie breaker.
1095 log(0,0,0,0, "They're a higher IP address than me, so I'm gone!\n");
1099 return -1; // Skip it.
1102 if (config
->cluster_seq_number
== -1) // Don't have one. Just align to the master...
1103 config
->cluster_seq_number
= h
->seq
;
1105 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1107 if (config
->cluster_seq_number
!= h
->seq
) { // Out of sequence heartbeat!
1108 log(1,0,0,0, "HB: Got seq# %d but was expecting %d. asking for resend.\n", h
->seq
, config
->cluster_seq_number
);
1110 peer_send_message(addr
, C_LASTSEEN
, config
->cluster_seq_number
, NULL
, 0);
1112 config
->cluster_last_hb
= TIME
; // Reset to ensure that we don't become master!!
1114 // Just drop the packet. The master will resend it as part of the catchup.
1118 // Save the packet in our buffer.
1119 // This is needed in case we become the master.
1120 config
->cluster_seq_number
= (h
->seq
+1)%HB_MAX_SEQ
;
1121 i
= h
->seq
% HB_HISTORY_SIZE
;
1122 past_hearts
[i
].seq
= h
->seq
;
1123 past_hearts
[i
].size
= size
;
1124 memcpy(&past_hearts
[i
].data
, data
, size
); // Save it.
1127 // Check that we don't have too many undefined sessions, and
1128 // that the free session pointer is correct.
1129 cluster_check_sessions(h
->highsession
, h
->freesession
, h
->hightunnel
);
1131 // Ok. process the packet...
1134 type
= * ((u32
*) p
);
1138 more
= * ((u32
*) p
);
1143 case C_CSESSION
: { // Compressed session structure.
1144 u8 c
[ sizeof(sessiont
) + 2];
1148 size
= rle_decompress((u8
**) &p
, s
, c
, sizeof(c
) );
1151 if (size
!= sizeof(sessiont
) ) { // Ouch! Very very bad!
1152 log(0,0,0,0, "DANGER: Received a CSESSION that didn't decompress correctly!\n");
1153 // Now what? Should exit! No-longer up to date!
1157 cluster_recv_session(more
, c
);
1161 if ( s
< sizeof(session
[more
]))
1164 cluster_recv_session(more
, p
);
1166 p
+= sizeof(session
[more
]);
1167 s
-= sizeof(session
[more
]);
1170 case C_CTUNNEL
: { // Compressed tunnel structure.
1171 u8 c
[ sizeof(tunnelt
) + 2];
1175 size
= rle_decompress( (u8
**) &p
, s
, c
, sizeof(c
) );
1178 if (size
!= sizeof(tunnelt
) ) { // Ouch! Very very bad!
1179 log(0,0,0,0, "DANGER: Received a CSESSION that didn't decompress correctly!\n");
1180 // Now what? Should exit! No-longer up to date!
1184 cluster_recv_tunnel(more
, c
);
1189 if ( s
< sizeof(tunnel
[more
]))
1192 cluster_recv_tunnel(more
, p
);
1194 p
+= sizeof(tunnel
[more
]);
1195 s
-= sizeof(tunnel
[more
]);
1198 log(0,0,0,0, "DANGER: I received a heartbeat element where I didn't understand the type! (%d)\n", type
);
1199 return -1; // can't process any more of the packet!!
1202 if (config
->cluster_master_address
!= addr
)
1205 str
= strdup(inet_toa(config
->cluster_master_address
));
1206 log(0,0,0,0, "My master just changed from %s to %s!\n", str
, inet_toa(addr
));
1210 config
->cluster_master_address
= addr
;
1211 config
->cluster_last_hb
= TIME
; // Successfully received a heartbeat!
1215 log(0,0,0,0, "I got an incomplete heartbeat packet! This means I'm probably out of sync!!\n");
1220 // We got a packet on the cluster port!
1221 // Handle pings, lastseens, and heartbeats!
1223 int processcluster(char * data
, int size
, u32 addr
)
1229 if (addr
== my_address
)
1230 return -1; // Ignore it. Something looped back the multicast!
1232 log(5,0,0,0, "Process cluster: %d bytes from %s\n", size
, inet_toa(addr
));
1234 if (s
<= 0) // Any data there??
1240 type
= * ((u32
*) p
);
1244 more
= * ((u32
*) p
);
1249 case C_PING
: // Update the peers table.
1250 return cluster_add_peer(addr
, more
, (pingt
*)p
);
1252 case C_LASTSEEN
: // Catch up a slave (slave missed a packet).
1253 return cluster_catchup_slave(more
, addr
);
1255 case C_FORWARD
: { // Forwarded control packet. pass off to processudp.
1256 struct sockaddr_in a
;
1257 a
.sin_addr
.s_addr
= more
;
1259 a
.sin_port
= * (int*) p
;
1263 if (!config
->cluster_iam_master
) { // huh?
1264 log(0,0,0,0, "I'm not the master, but I got a C_FORWARD from %s?\n", inet_toa(addr
));
1268 log(4,0,0,0, "Got a forwarded packet... (%s:%d)\n", inet_toa(more
), a
.sin_port
);
1270 processudp(p
, s
, &a
);
1273 case C_THROTTLE
: { // Receive a forwarded packet from a slave.
1274 if (!config
->cluster_iam_master
) {
1275 log(0,0,0,0, "I'm not the master, but I got a C_THROTTLE from %s?\n", inet_toa(addr
));
1279 tbf_queue_packet(more
, p
, s
); // The TBF id tells wether it goes in or out.
1283 // Receive a walled garden packet from a slave.
1284 if (!config
->cluster_iam_master
) {
1285 log(0,0,0,0, "I'm not the master, but I got a C_GARDEN from %s?\n", inet_toa(addr
));
1293 return cluster_handle_bytes(p
, s
);
1295 case C_KILL
: // The master asked us to die!? (usually because we're too out of date).
1296 if (config
->cluster_iam_master
) {
1297 log(0,0,0,0, "_I_ am master, but I received a C_KILL from %s! (Seq# %d)\n", inet_toa(addr
), more
);
1300 if (more
!= config
->cluster_seq_number
) {
1301 log(0,0,0,0, "The master asked us to die but the seq number didn't match!?\n");
1305 if (addr
!= config
->cluster_master_address
) {
1306 log(0,0,0,0, "Received a C_KILL from %s which doesn't match config->cluster_master_address (%x)\n",
1307 inet_toa(addr
), config
->cluster_master_address
);
1308 // We can only warn about it. The master might really have switched!
1311 log(0,0,0,0, "Received a valid C_KILL: I'm going to die now.\n");
1313 exit(0); // Lets be paranoid;
1314 return -1; // Just signalling the compiler.
1317 log(4,0,0,0, "Got a heartbeat from %s\n", inet_toa(addr
));
1319 return cluster_process_heartbeat_v2(data
, size
, more
, p
, addr
);
1322 log(0,0,0,0, "Strange type packet received on cluster socket (%d)\n", type
);
1327 log(0,0,0,0, "I got an cluster heartbeat packet! This means I'm probably out of sync!!\n");
1331 //====================================================================================================
1333 int cmd_show_cluster(struct cli_def
*cli
, char *command
, char **argv
, int argc
)
1337 if (CLI_HELP_REQUESTED
)
1338 return CLI_HELP_NO_ARGS
;
1340 cli_print(cli
, "Cluster status : %s", config
->cluster_iam_master
? "Master" : "Slave" );
1341 cli_print(cli
, "My address : %s", inet_toa(my_address
));
1342 cli_print(cli
, "VIP address : %s", inet_toa(config
->bind_address
));
1343 cli_print(cli
, "Multicast address: %s", inet_toa(config
->cluster_address
));
1344 cli_print(cli
, "Multicast i'face : %s", config
->cluster_interface
);
1346 if (!config
->cluster_iam_master
) {
1347 cli_print(cli
, "My master : %s (last heartbeat %.1f seconds old)",
1348 config
->cluster_master_address
? inet_toa(config
->cluster_master_address
) : "Not defined",
1349 0.1 * (TIME
- config
->cluster_last_hb
));
1350 cli_print(cli
, "Uptodate : %s", config
->cluster_iam_uptodate
? "Yes" : "No");
1351 cli_print(cli
, "Next sequence number expected: %d", config
->cluster_seq_number
);
1352 cli_print(cli
, "%d sessions undefined of %d", config
->cluster_undefined_sessions
, config
->cluster_highest_sessionid
);
1353 cli_print(cli
, "%d tunnels undefined of %d", config
->cluster_undefined_tunnels
, config
->cluster_highest_tunnelid
);
1355 cli_print(cli
, "Next heartbeat # : %d", config
->cluster_seq_number
);
1356 cli_print(cli
, "Highest session : %d", config
->cluster_highest_sessionid
);
1357 cli_print(cli
, "Highest tunnel : %d", config
->cluster_highest_tunnelid
);
1358 cli_print(cli
, "%d changes queued for sending", config
->cluster_num_changes
);
1360 cli_print(cli
, "%d peers.", num_peers
);
1363 cli_print(cli
, "%20s %10s %8s", "Address", "Basetime", "Age");
1364 for (i
= 0; i
< num_peers
; ++i
) {
1365 cli_print(cli
, "%20s %10d %8d", inet_toa(peers
[i
].peer
),
1366 peers
[i
].basetime
, TIME
- peers
[i
].timestamp
);
1372 // Simple run-length-encoding compression.
1374 // 1 byte < 128 = count of non-zero bytes following. // Not legal to be zero.
1375 // n non-zero bytes;
1377 // 1 byte > 128 = (count - 128) run of zero bytes. //
1379 // count == 0 indicates end of compressed stream.
1381 // Compress from 'src' into 'dst'. return number of bytes
1383 // Updates *src_p to indicate 1 past last bytes used.
1385 // We could get an extra byte in the zero runs by storing (count-1)
1386 // but I'm playing it safe.
1388 // Worst case is a 50% expansion in space required (trying to
1389 // compress { 0x00, 0x01 } * N )
1390 int rle_compress(u8
** src_p
, int ssize
, u8
*dst
, int dsize
)
1393 int orig_dsize
= dsize
;
1397 while (ssize
> 0 && dsize
> 2) {
1399 x
= dst
++; --dsize
; // Reserve space for count byte..
1401 if (*src
) { // Copy a run of non-zero bytes.
1402 while (*src
&& count
< 127 && ssize
> 0 && dsize
> 1) { // Count number of non-zero bytes.
1407 *x
= count
; // Store number of non-zero bytes. Guarenteed to be non-zero!
1409 } else { // Compress a run of zero bytes.
1410 while (*src
== 0 && count
< 127 && ssize
> 0) {
1419 *dst
++ = 0x0; // Add Stop byte.
1423 return (orig_dsize
- dsize
);
1427 // Decompress the buffer into **p.
1428 // 'psize' is the size of the decompression buffer available.
1430 // Returns the number of bytes decompressed.
1432 // Decompresses from '*src_p' into 'dst'.
1433 // Return the number of dst bytes used.
1434 // Updates the 'src_p' pointer to point to the
1435 // first un-used byte.
1436 int rle_decompress(u8
** src_p
, int ssize
, u8
*dst
, int dsize
)
1439 int orig_dsize
= dsize
;
1440 char * src
= *src_p
;
1442 while (ssize
>0 && dsize
> 0) { // While there's more to decompress, and there's room in the decompress buffer...
1443 count
= *src
++; --ssize
; // get the count byte from the source.
1444 if (count
== 0x0) // End marker reached? If so, finish.
1447 if (count
& 0x80) { // Decompress a run of zeros
1448 for (count
&= 0x7f ; count
> 0 && dsize
> 0; --count
) {
1452 } else { // Copy run of non-zero bytes.
1453 for ( ; count
> 0 && ssize
&& dsize
; --count
) { // Copy non-zero bytes across.
1460 return (orig_dsize
- dsize
);