]> git.proxmox.com Git - mirror_frr.git/blob - zebra/zebra_fpm.c
Merge pull request #5356 from v00lk/master
[mirror_frr.git] / zebra / zebra_fpm.c
1 /*
2 * Main implementation file for interface to Forwarding Plane Manager.
3 *
4 * Copyright (C) 2012 by Open Source Routing.
5 * Copyright (C) 2012 by Internet Systems Consortium, Inc. ("ISC")
6 *
7 * This file is part of GNU Zebra.
8 *
9 * GNU Zebra is free software; you can redistribute it and/or modify it
10 * under the terms of the GNU General Public License as published by the
11 * Free Software Foundation; either version 2, or (at your option) any
12 * later version.
13 *
14 * GNU Zebra is distributed in the hope that it will be useful, but
15 * WITHOUT ANY WARRANTY; without even the implied warranty of
16 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
17 * General Public License for more details.
18 *
19 * You should have received a copy of the GNU General Public License along
20 * with this program; see the file COPYING; if not, write to the Free Software
21 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
22 */
23
24 #include <zebra.h>
25
26 #include "log.h"
27 #include "libfrr.h"
28 #include "stream.h"
29 #include "thread.h"
30 #include "network.h"
31 #include "command.h"
32 #include "version.h"
33 #include "jhash.h"
34
35 #include "zebra/rib.h"
36 #include "zebra/zserv.h"
37 #include "zebra/zebra_ns.h"
38 #include "zebra/zebra_vrf.h"
39 #include "zebra/zebra_errors.h"
40 #include "zebra/zebra_memory.h"
41
42 #include "fpm/fpm.h"
43 #include "zebra_fpm_private.h"
44 #include "zebra/zebra_router.h"
45 #include "zebra_vxlan_private.h"
46
47 DEFINE_MTYPE_STATIC(ZEBRA, FPM_MAC_INFO, "FPM_MAC_INFO");
48
49 /*
50 * Interval at which we attempt to connect to the FPM.
51 */
52 #define ZFPM_CONNECT_RETRY_IVL 5
53
54 /*
55 * Sizes of outgoing and incoming stream buffers for writing/reading
56 * FPM messages.
57 */
58 #define ZFPM_OBUF_SIZE (2 * FPM_MAX_MSG_LEN)
59 #define ZFPM_IBUF_SIZE (FPM_MAX_MSG_LEN)
60
61 /*
62 * The maximum number of times the FPM socket write callback can call
63 * 'write' before it yields.
64 */
65 #define ZFPM_MAX_WRITES_PER_RUN 10
66
67 /*
68 * Interval over which we collect statistics.
69 */
70 #define ZFPM_STATS_IVL_SECS 10
71 #define FPM_MAX_MAC_MSG_LEN 512
72
73 static void zfpm_iterate_rmac_table(struct hash_backet *backet, void *args);
74
75 /*
76 * Structure that holds state for iterating over all route_node
77 * structures that are candidates for being communicated to the FPM.
78 */
79 typedef struct zfpm_rnodes_iter_t_ {
80 rib_tables_iter_t tables_iter;
81 route_table_iter_t iter;
82 } zfpm_rnodes_iter_t;
83
84 /*
85 * Statistics.
86 */
87 typedef struct zfpm_stats_t_ {
88 unsigned long connect_calls;
89 unsigned long connect_no_sock;
90
91 unsigned long read_cb_calls;
92
93 unsigned long write_cb_calls;
94 unsigned long write_calls;
95 unsigned long partial_writes;
96 unsigned long max_writes_hit;
97 unsigned long t_write_yields;
98
99 unsigned long nop_deletes_skipped;
100 unsigned long route_adds;
101 unsigned long route_dels;
102
103 unsigned long updates_triggered;
104 unsigned long redundant_triggers;
105
106 unsigned long dests_del_after_update;
107
108 unsigned long t_conn_down_starts;
109 unsigned long t_conn_down_dests_processed;
110 unsigned long t_conn_down_yields;
111 unsigned long t_conn_down_finishes;
112
113 unsigned long t_conn_up_starts;
114 unsigned long t_conn_up_dests_processed;
115 unsigned long t_conn_up_yields;
116 unsigned long t_conn_up_aborts;
117 unsigned long t_conn_up_finishes;
118
119 } zfpm_stats_t;
120
121 /*
122 * States for the FPM state machine.
123 */
124 typedef enum {
125
126 /*
127 * In this state we are not yet ready to connect to the FPM. This
128 * can happen when this module is disabled, or if we're cleaning up
129 * after a connection has gone down.
130 */
131 ZFPM_STATE_IDLE,
132
133 /*
134 * Ready to talk to the FPM and periodically trying to connect to
135 * it.
136 */
137 ZFPM_STATE_ACTIVE,
138
139 /*
140 * In the middle of bringing up a TCP connection. Specifically,
141 * waiting for a connect() call to complete asynchronously.
142 */
143 ZFPM_STATE_CONNECTING,
144
145 /*
146 * TCP connection to the FPM is up.
147 */
148 ZFPM_STATE_ESTABLISHED
149
150 } zfpm_state_t;
151
152 /*
153 * Message format to be used to communicate with the FPM.
154 */
155 typedef enum {
156 ZFPM_MSG_FORMAT_NONE,
157 ZFPM_MSG_FORMAT_NETLINK,
158 ZFPM_MSG_FORMAT_PROTOBUF,
159 } zfpm_msg_format_e;
160 /*
161 * Globals.
162 */
163 typedef struct zfpm_glob_t_ {
164
165 /*
166 * True if the FPM module has been enabled.
167 */
168 int enabled;
169
170 /*
171 * Message format to be used to communicate with the fpm.
172 */
173 zfpm_msg_format_e message_format;
174
175 struct thread_master *master;
176
177 zfpm_state_t state;
178
179 in_addr_t fpm_server;
180 /*
181 * Port on which the FPM is running.
182 */
183 int fpm_port;
184
185 /*
186 * List of rib_dest_t structures to be processed
187 */
188 TAILQ_HEAD(zfpm_dest_q, rib_dest_t_) dest_q;
189
190 /*
191 * List of fpm_mac_info structures to be processed
192 */
193 TAILQ_HEAD(zfpm_mac_q, fpm_mac_info_t) mac_q;
194
195 /*
196 * Hash table of fpm_mac_info_t entries
197 *
198 * While adding fpm_mac_info_t for a MAC to the mac_q,
199 * it is possible that another fpm_mac_info_t node for the this MAC
200 * is already present in the queue.
201 * This is possible in the case of consecutive add->delete operations.
202 * To avoid such duplicate insertions in the mac_q,
203 * define a hash table for fpm_mac_info_t which can be looked up
204 * to see if an fpm_mac_info_t node for a MAC is already present
205 * in the mac_q.
206 */
207 struct hash *fpm_mac_info_table;
208
209 /*
210 * Stream socket to the FPM.
211 */
212 int sock;
213
214 /*
215 * Buffers for messages to/from the FPM.
216 */
217 struct stream *obuf;
218 struct stream *ibuf;
219
220 /*
221 * Threads for I/O.
222 */
223 struct thread *t_connect;
224 struct thread *t_write;
225 struct thread *t_read;
226
227 /*
228 * Thread to clean up after the TCP connection to the FPM goes down
229 * and the state that belongs to it.
230 */
231 struct thread *t_conn_down;
232
233 struct {
234 zfpm_rnodes_iter_t iter;
235 } t_conn_down_state;
236
237 /*
238 * Thread to take actions once the TCP conn to the FPM comes up, and
239 * the state that belongs to it.
240 */
241 struct thread *t_conn_up;
242
243 struct {
244 zfpm_rnodes_iter_t iter;
245 } t_conn_up_state;
246
247 unsigned long connect_calls;
248 time_t last_connect_call_time;
249
250 /*
251 * Stats from the start of the current statistics interval up to
252 * now. These are the counters we typically update in the code.
253 */
254 zfpm_stats_t stats;
255
256 /*
257 * Statistics that were gathered in the last collection interval.
258 */
259 zfpm_stats_t last_ivl_stats;
260
261 /*
262 * Cumulative stats from the last clear to the start of the current
263 * statistics interval.
264 */
265 zfpm_stats_t cumulative_stats;
266
267 /*
268 * Stats interval timer.
269 */
270 struct thread *t_stats;
271
272 /*
273 * If non-zero, the last time when statistics were cleared.
274 */
275 time_t last_stats_clear_time;
276
277 } zfpm_glob_t;
278
279 static zfpm_glob_t zfpm_glob_space;
280 static zfpm_glob_t *zfpm_g = &zfpm_glob_space;
281
282 static int zfpm_trigger_update(struct route_node *rn, const char *reason);
283
284 static int zfpm_read_cb(struct thread *thread);
285 static int zfpm_write_cb(struct thread *thread);
286
287 static void zfpm_set_state(zfpm_state_t state, const char *reason);
288 static void zfpm_start_connect_timer(const char *reason);
289 static void zfpm_start_stats_timer(void);
290 static void zfpm_mac_info_del(struct fpm_mac_info_t *fpm_mac);
291
292 /*
293 * zfpm_thread_should_yield
294 */
295 static inline int zfpm_thread_should_yield(struct thread *t)
296 {
297 return thread_should_yield(t);
298 }
299
300 /*
301 * zfpm_state_to_str
302 */
303 static const char *zfpm_state_to_str(zfpm_state_t state)
304 {
305 switch (state) {
306
307 case ZFPM_STATE_IDLE:
308 return "idle";
309
310 case ZFPM_STATE_ACTIVE:
311 return "active";
312
313 case ZFPM_STATE_CONNECTING:
314 return "connecting";
315
316 case ZFPM_STATE_ESTABLISHED:
317 return "established";
318
319 default:
320 return "unknown";
321 }
322 }
323
324 /*
325 * zfpm_get_elapsed_time
326 *
327 * Returns the time elapsed (in seconds) since the given time.
328 */
329 static time_t zfpm_get_elapsed_time(time_t reference)
330 {
331 time_t now;
332
333 now = monotime(NULL);
334
335 if (now < reference) {
336 assert(0);
337 return 0;
338 }
339
340 return now - reference;
341 }
342
343 /*
344 * zfpm_rnodes_iter_init
345 */
346 static inline void zfpm_rnodes_iter_init(zfpm_rnodes_iter_t *iter)
347 {
348 memset(iter, 0, sizeof(*iter));
349 rib_tables_iter_init(&iter->tables_iter);
350
351 /*
352 * This is a hack, but it makes implementing 'next' easier by
353 * ensuring that route_table_iter_next() will return NULL the first
354 * time we call it.
355 */
356 route_table_iter_init(&iter->iter, NULL);
357 route_table_iter_cleanup(&iter->iter);
358 }
359
360 /*
361 * zfpm_rnodes_iter_next
362 */
363 static inline struct route_node *zfpm_rnodes_iter_next(zfpm_rnodes_iter_t *iter)
364 {
365 struct route_node *rn;
366 struct route_table *table;
367
368 while (1) {
369 rn = route_table_iter_next(&iter->iter);
370 if (rn)
371 return rn;
372
373 /*
374 * We've made our way through this table, go to the next one.
375 */
376 route_table_iter_cleanup(&iter->iter);
377
378 table = rib_tables_iter_next(&iter->tables_iter);
379
380 if (!table)
381 return NULL;
382
383 route_table_iter_init(&iter->iter, table);
384 }
385
386 return NULL;
387 }
388
389 /*
390 * zfpm_rnodes_iter_pause
391 */
392 static inline void zfpm_rnodes_iter_pause(zfpm_rnodes_iter_t *iter)
393 {
394 route_table_iter_pause(&iter->iter);
395 }
396
397 /*
398 * zfpm_rnodes_iter_cleanup
399 */
400 static inline void zfpm_rnodes_iter_cleanup(zfpm_rnodes_iter_t *iter)
401 {
402 route_table_iter_cleanup(&iter->iter);
403 rib_tables_iter_cleanup(&iter->tables_iter);
404 }
405
406 /*
407 * zfpm_stats_init
408 *
409 * Initialize a statistics block.
410 */
411 static inline void zfpm_stats_init(zfpm_stats_t *stats)
412 {
413 memset(stats, 0, sizeof(*stats));
414 }
415
416 /*
417 * zfpm_stats_reset
418 */
419 static inline void zfpm_stats_reset(zfpm_stats_t *stats)
420 {
421 zfpm_stats_init(stats);
422 }
423
424 /*
425 * zfpm_stats_copy
426 */
427 static inline void zfpm_stats_copy(const zfpm_stats_t *src, zfpm_stats_t *dest)
428 {
429 memcpy(dest, src, sizeof(*dest));
430 }
431
432 /*
433 * zfpm_stats_compose
434 *
435 * Total up the statistics in two stats structures ('s1 and 's2') and
436 * return the result in the third argument, 'result'. Note that the
437 * pointer 'result' may be the same as 's1' or 's2'.
438 *
439 * For simplicity, the implementation below assumes that the stats
440 * structure is composed entirely of counters. This can easily be
441 * changed when necessary.
442 */
443 static void zfpm_stats_compose(const zfpm_stats_t *s1, const zfpm_stats_t *s2,
444 zfpm_stats_t *result)
445 {
446 const unsigned long *p1, *p2;
447 unsigned long *result_p;
448 int i, num_counters;
449
450 p1 = (const unsigned long *)s1;
451 p2 = (const unsigned long *)s2;
452 result_p = (unsigned long *)result;
453
454 num_counters = (sizeof(zfpm_stats_t) / sizeof(unsigned long));
455
456 for (i = 0; i < num_counters; i++) {
457 result_p[i] = p1[i] + p2[i];
458 }
459 }
460
461 /*
462 * zfpm_read_on
463 */
464 static inline void zfpm_read_on(void)
465 {
466 assert(!zfpm_g->t_read);
467 assert(zfpm_g->sock >= 0);
468
469 thread_add_read(zfpm_g->master, zfpm_read_cb, 0, zfpm_g->sock,
470 &zfpm_g->t_read);
471 }
472
473 /*
474 * zfpm_write_on
475 */
476 static inline void zfpm_write_on(void)
477 {
478 assert(!zfpm_g->t_write);
479 assert(zfpm_g->sock >= 0);
480
481 thread_add_write(zfpm_g->master, zfpm_write_cb, 0, zfpm_g->sock,
482 &zfpm_g->t_write);
483 }
484
485 /*
486 * zfpm_read_off
487 */
488 static inline void zfpm_read_off(void)
489 {
490 THREAD_READ_OFF(zfpm_g->t_read);
491 }
492
493 /*
494 * zfpm_write_off
495 */
496 static inline void zfpm_write_off(void)
497 {
498 THREAD_WRITE_OFF(zfpm_g->t_write);
499 }
500
501 static inline void zfpm_connect_off(void)
502 {
503 THREAD_TIMER_OFF(zfpm_g->t_connect);
504 }
505
506 /*
507 * zfpm_conn_up_thread_cb
508 *
509 * Callback for actions to be taken when the connection to the FPM
510 * comes up.
511 */
512 static int zfpm_conn_up_thread_cb(struct thread *thread)
513 {
514 struct route_node *rnode;
515 zfpm_rnodes_iter_t *iter;
516 rib_dest_t *dest;
517
518 zfpm_g->t_conn_up = NULL;
519
520 iter = &zfpm_g->t_conn_up_state.iter;
521
522 if (zfpm_g->state != ZFPM_STATE_ESTABLISHED) {
523 zfpm_debug(
524 "Connection not up anymore, conn_up thread aborting");
525 zfpm_g->stats.t_conn_up_aborts++;
526 goto done;
527 }
528
529 /* Enqueue FPM updates for all the RMAC entries */
530 hash_iterate(zrouter.l3vni_table, zfpm_iterate_rmac_table, NULL);
531
532 while ((rnode = zfpm_rnodes_iter_next(iter))) {
533 dest = rib_dest_from_rnode(rnode);
534
535 if (dest) {
536 zfpm_g->stats.t_conn_up_dests_processed++;
537 zfpm_trigger_update(rnode, NULL);
538 }
539
540 /*
541 * Yield if need be.
542 */
543 if (!zfpm_thread_should_yield(thread))
544 continue;
545
546 zfpm_g->stats.t_conn_up_yields++;
547 zfpm_rnodes_iter_pause(iter);
548 zfpm_g->t_conn_up = NULL;
549 thread_add_timer_msec(zfpm_g->master, zfpm_conn_up_thread_cb,
550 NULL, 0, &zfpm_g->t_conn_up);
551 return 0;
552 }
553
554 zfpm_g->stats.t_conn_up_finishes++;
555
556 done:
557 zfpm_rnodes_iter_cleanup(iter);
558 return 0;
559 }
560
561 /*
562 * zfpm_connection_up
563 *
564 * Called when the connection to the FPM comes up.
565 */
566 static void zfpm_connection_up(const char *detail)
567 {
568 assert(zfpm_g->sock >= 0);
569 zfpm_read_on();
570 zfpm_write_on();
571 zfpm_set_state(ZFPM_STATE_ESTABLISHED, detail);
572
573 /*
574 * Start thread to push existing routes to the FPM.
575 */
576 assert(!zfpm_g->t_conn_up);
577
578 zfpm_rnodes_iter_init(&zfpm_g->t_conn_up_state.iter);
579
580 zfpm_debug("Starting conn_up thread");
581 zfpm_g->t_conn_up = NULL;
582 thread_add_timer_msec(zfpm_g->master, zfpm_conn_up_thread_cb, NULL, 0,
583 &zfpm_g->t_conn_up);
584 zfpm_g->stats.t_conn_up_starts++;
585 }
586
587 /*
588 * zfpm_connect_check
589 *
590 * Check if an asynchronous connect() to the FPM is complete.
591 */
592 static void zfpm_connect_check(void)
593 {
594 int status;
595 socklen_t slen;
596 int ret;
597
598 zfpm_read_off();
599 zfpm_write_off();
600
601 slen = sizeof(status);
602 ret = getsockopt(zfpm_g->sock, SOL_SOCKET, SO_ERROR, (void *)&status,
603 &slen);
604
605 if (ret >= 0 && status == 0) {
606 zfpm_connection_up("async connect complete");
607 return;
608 }
609
610 /*
611 * getsockopt() failed or indicated an error on the socket.
612 */
613 close(zfpm_g->sock);
614 zfpm_g->sock = -1;
615
616 zfpm_start_connect_timer("getsockopt() after async connect failed");
617 return;
618 }
619
620 /*
621 * zfpm_conn_down_thread_cb
622 *
623 * Callback that is invoked to clean up state after the TCP connection
624 * to the FPM goes down.
625 */
626 static int zfpm_conn_down_thread_cb(struct thread *thread)
627 {
628 struct route_node *rnode;
629 zfpm_rnodes_iter_t *iter;
630 rib_dest_t *dest;
631 struct fpm_mac_info_t *mac = NULL;
632
633 assert(zfpm_g->state == ZFPM_STATE_IDLE);
634
635 /*
636 * Delink and free all fpm_mac_info_t nodes
637 * in the mac_q and fpm_mac_info_hash
638 */
639 while ((mac = TAILQ_FIRST(&zfpm_g->mac_q)) != NULL)
640 zfpm_mac_info_del(mac);
641
642 zfpm_g->t_conn_down = NULL;
643
644 iter = &zfpm_g->t_conn_down_state.iter;
645
646 while ((rnode = zfpm_rnodes_iter_next(iter))) {
647 dest = rib_dest_from_rnode(rnode);
648
649 if (dest) {
650 if (CHECK_FLAG(dest->flags, RIB_DEST_UPDATE_FPM)) {
651 TAILQ_REMOVE(&zfpm_g->dest_q, dest,
652 fpm_q_entries);
653 }
654
655 UNSET_FLAG(dest->flags, RIB_DEST_UPDATE_FPM);
656 UNSET_FLAG(dest->flags, RIB_DEST_SENT_TO_FPM);
657
658 zfpm_g->stats.t_conn_down_dests_processed++;
659
660 /*
661 * Check if the dest should be deleted.
662 */
663 rib_gc_dest(rnode);
664 }
665
666 /*
667 * Yield if need be.
668 */
669 if (!zfpm_thread_should_yield(thread))
670 continue;
671
672 zfpm_g->stats.t_conn_down_yields++;
673 zfpm_rnodes_iter_pause(iter);
674 zfpm_g->t_conn_down = NULL;
675 thread_add_timer_msec(zfpm_g->master, zfpm_conn_down_thread_cb,
676 NULL, 0, &zfpm_g->t_conn_down);
677 return 0;
678 }
679
680 zfpm_g->stats.t_conn_down_finishes++;
681 zfpm_rnodes_iter_cleanup(iter);
682
683 /*
684 * Start the process of connecting to the FPM again.
685 */
686 zfpm_start_connect_timer("cleanup complete");
687 return 0;
688 }
689
690 /*
691 * zfpm_connection_down
692 *
693 * Called when the connection to the FPM has gone down.
694 */
695 static void zfpm_connection_down(const char *detail)
696 {
697 if (!detail)
698 detail = "unknown";
699
700 assert(zfpm_g->state == ZFPM_STATE_ESTABLISHED);
701
702 zlog_info("connection to the FPM has gone down: %s", detail);
703
704 zfpm_read_off();
705 zfpm_write_off();
706
707 stream_reset(zfpm_g->ibuf);
708 stream_reset(zfpm_g->obuf);
709
710 if (zfpm_g->sock >= 0) {
711 close(zfpm_g->sock);
712 zfpm_g->sock = -1;
713 }
714
715 /*
716 * Start thread to clean up state after the connection goes down.
717 */
718 assert(!zfpm_g->t_conn_down);
719 zfpm_rnodes_iter_init(&zfpm_g->t_conn_down_state.iter);
720 zfpm_g->t_conn_down = NULL;
721 thread_add_timer_msec(zfpm_g->master, zfpm_conn_down_thread_cb, NULL, 0,
722 &zfpm_g->t_conn_down);
723 zfpm_g->stats.t_conn_down_starts++;
724
725 zfpm_set_state(ZFPM_STATE_IDLE, detail);
726 }
727
728 /*
729 * zfpm_read_cb
730 */
731 static int zfpm_read_cb(struct thread *thread)
732 {
733 size_t already;
734 struct stream *ibuf;
735 uint16_t msg_len;
736 fpm_msg_hdr_t *hdr;
737
738 zfpm_g->stats.read_cb_calls++;
739
740 /*
741 * Check if async connect is now done.
742 */
743 if (zfpm_g->state == ZFPM_STATE_CONNECTING) {
744 zfpm_connect_check();
745 return 0;
746 }
747
748 assert(zfpm_g->state == ZFPM_STATE_ESTABLISHED);
749 assert(zfpm_g->sock >= 0);
750
751 ibuf = zfpm_g->ibuf;
752
753 already = stream_get_endp(ibuf);
754 if (already < FPM_MSG_HDR_LEN) {
755 ssize_t nbyte;
756
757 nbyte = stream_read_try(ibuf, zfpm_g->sock,
758 FPM_MSG_HDR_LEN - already);
759 if (nbyte == 0 || nbyte == -1) {
760 if (nbyte == -1) {
761 char buffer[1024];
762
763 sprintf(buffer, "closed socket in read(%d): %s",
764 errno, safe_strerror(errno));
765 zfpm_connection_down(buffer);
766 } else
767 zfpm_connection_down("closed socket in read");
768 return 0;
769 }
770
771 if (nbyte != (ssize_t)(FPM_MSG_HDR_LEN - already))
772 goto done;
773
774 already = FPM_MSG_HDR_LEN;
775 }
776
777 stream_set_getp(ibuf, 0);
778
779 hdr = (fpm_msg_hdr_t *)stream_pnt(ibuf);
780
781 if (!fpm_msg_hdr_ok(hdr)) {
782 zfpm_connection_down("invalid message header");
783 return 0;
784 }
785
786 msg_len = fpm_msg_len(hdr);
787
788 /*
789 * Read out the rest of the packet.
790 */
791 if (already < msg_len) {
792 ssize_t nbyte;
793
794 nbyte = stream_read_try(ibuf, zfpm_g->sock, msg_len - already);
795
796 if (nbyte == 0 || nbyte == -1) {
797 if (nbyte == -1) {
798 char buffer[1024];
799
800 sprintf(buffer, "failed to read message(%d) %s",
801 errno, safe_strerror(errno));
802 zfpm_connection_down(buffer);
803 } else
804 zfpm_connection_down("failed to read message");
805 return 0;
806 }
807
808 if (nbyte != (ssize_t)(msg_len - already))
809 goto done;
810 }
811
812 /*
813 * Just throw it away for now.
814 */
815 stream_reset(ibuf);
816
817 done:
818 zfpm_read_on();
819 return 0;
820 }
821
822 static bool zfpm_updates_pending(void)
823 {
824 if (!(TAILQ_EMPTY(&zfpm_g->dest_q)) || !(TAILQ_EMPTY(&zfpm_g->mac_q)))
825 return true;
826
827 return false;
828 }
829
830 /*
831 * zfpm_writes_pending
832 *
833 * Returns true if we may have something to write to the FPM.
834 */
835 static int zfpm_writes_pending(void)
836 {
837
838 /*
839 * Check if there is any data in the outbound buffer that has not
840 * been written to the socket yet.
841 */
842 if (stream_get_endp(zfpm_g->obuf) - stream_get_getp(zfpm_g->obuf))
843 return 1;
844
845 /*
846 * Check if there are any updates scheduled on the outbound queues.
847 */
848 if (zfpm_updates_pending())
849 return 1;
850
851 return 0;
852 }
853
854 /*
855 * zfpm_encode_route
856 *
857 * Encode a message to the FPM with information about the given route.
858 *
859 * Returns the number of bytes written to the buffer. 0 or a negative
860 * value indicates an error.
861 */
862 static inline int zfpm_encode_route(rib_dest_t *dest, struct route_entry *re,
863 char *in_buf, size_t in_buf_len,
864 fpm_msg_type_e *msg_type)
865 {
866 size_t len;
867 #ifdef HAVE_NETLINK
868 int cmd;
869 #endif
870 len = 0;
871
872 *msg_type = FPM_MSG_TYPE_NONE;
873
874 switch (zfpm_g->message_format) {
875
876 case ZFPM_MSG_FORMAT_PROTOBUF:
877 #ifdef HAVE_PROTOBUF
878 len = zfpm_protobuf_encode_route(dest, re, (uint8_t *)in_buf,
879 in_buf_len);
880 *msg_type = FPM_MSG_TYPE_PROTOBUF;
881 #endif
882 break;
883
884 case ZFPM_MSG_FORMAT_NETLINK:
885 #ifdef HAVE_NETLINK
886 *msg_type = FPM_MSG_TYPE_NETLINK;
887 cmd = re ? RTM_NEWROUTE : RTM_DELROUTE;
888 len = zfpm_netlink_encode_route(cmd, dest, re, in_buf,
889 in_buf_len);
890 assert(fpm_msg_align(len) == len);
891 *msg_type = FPM_MSG_TYPE_NETLINK;
892 #endif /* HAVE_NETLINK */
893 break;
894
895 default:
896 break;
897 }
898
899 return len;
900 }
901
902 /*
903 * zfpm_route_for_update
904 *
905 * Returns the re that is to be sent to the FPM for a given dest.
906 */
907 struct route_entry *zfpm_route_for_update(rib_dest_t *dest)
908 {
909 return dest->selected_fib;
910 }
911
912 /*
913 * Define an enum for return codes for queue processing functions
914 *
915 * FPM_WRITE_STOP: This return code indicates that the write buffer is full.
916 * Stop processing all the queues and empty the buffer by writing its content
917 * to the socket.
918 *
919 * FPM_GOTO_NEXT_Q: This return code indicates that either this queue is
920 * empty or we have processed enough updates from this queue.
921 * So, move on to the next queue.
922 */
923 enum {
924 FPM_WRITE_STOP = 0,
925 FPM_GOTO_NEXT_Q = 1
926 };
927
928 #define FPM_QUEUE_PROCESS_LIMIT 10000
929
930 /*
931 * zfpm_build_route_updates
932 *
933 * Process the dest_q queue and write FPM messages to the outbound buffer.
934 */
935 static int zfpm_build_route_updates(void)
936 {
937 struct stream *s;
938 rib_dest_t *dest;
939 unsigned char *buf, *data, *buf_end;
940 size_t msg_len;
941 size_t data_len;
942 fpm_msg_hdr_t *hdr;
943 struct route_entry *re;
944 int is_add, write_msg;
945 fpm_msg_type_e msg_type;
946 uint16_t q_limit;
947
948 if (TAILQ_EMPTY(&zfpm_g->dest_q))
949 return FPM_GOTO_NEXT_Q;
950
951 s = zfpm_g->obuf;
952 q_limit = FPM_QUEUE_PROCESS_LIMIT;
953
954 do {
955 /*
956 * Make sure there is enough space to write another message.
957 */
958 if (STREAM_WRITEABLE(s) < FPM_MAX_MSG_LEN)
959 return FPM_WRITE_STOP;
960
961 buf = STREAM_DATA(s) + stream_get_endp(s);
962 buf_end = buf + STREAM_WRITEABLE(s);
963
964 dest = TAILQ_FIRST(&zfpm_g->dest_q);
965 if (!dest)
966 return FPM_GOTO_NEXT_Q;
967
968 assert(CHECK_FLAG(dest->flags, RIB_DEST_UPDATE_FPM));
969
970 hdr = (fpm_msg_hdr_t *)buf;
971 hdr->version = FPM_PROTO_VERSION;
972
973 data = fpm_msg_data(hdr);
974
975 re = zfpm_route_for_update(dest);
976 is_add = re ? 1 : 0;
977
978 write_msg = 1;
979
980 /*
981 * If this is a route deletion, and we have not sent the route
982 * to
983 * the FPM previously, skip it.
984 */
985 if (!is_add && !CHECK_FLAG(dest->flags, RIB_DEST_SENT_TO_FPM)) {
986 write_msg = 0;
987 zfpm_g->stats.nop_deletes_skipped++;
988 }
989
990 if (write_msg) {
991 data_len = zfpm_encode_route(dest, re, (char *)data,
992 buf_end - data, &msg_type);
993
994 assert(data_len);
995 if (data_len) {
996 hdr->msg_type = msg_type;
997 msg_len = fpm_data_len_to_msg_len(data_len);
998 hdr->msg_len = htons(msg_len);
999 stream_forward_endp(s, msg_len);
1000
1001 if (is_add)
1002 zfpm_g->stats.route_adds++;
1003 else
1004 zfpm_g->stats.route_dels++;
1005 }
1006 }
1007
1008 /*
1009 * Remove the dest from the queue, and reset the flag.
1010 */
1011 UNSET_FLAG(dest->flags, RIB_DEST_UPDATE_FPM);
1012 TAILQ_REMOVE(&zfpm_g->dest_q, dest, fpm_q_entries);
1013
1014 if (is_add) {
1015 SET_FLAG(dest->flags, RIB_DEST_SENT_TO_FPM);
1016 } else {
1017 UNSET_FLAG(dest->flags, RIB_DEST_SENT_TO_FPM);
1018 }
1019
1020 /*
1021 * Delete the destination if necessary.
1022 */
1023 if (rib_gc_dest(dest->rnode))
1024 zfpm_g->stats.dests_del_after_update++;
1025
1026 q_limit--;
1027 if (q_limit == 0) {
1028 /*
1029 * We have processed enough updates in this queue.
1030 * Now yield for other queues.
1031 */
1032 return FPM_GOTO_NEXT_Q;
1033 }
1034 } while (true);
1035 }
1036
1037 /*
1038 * zfpm_encode_mac
1039 *
1040 * Encode a message to FPM with information about the given MAC.
1041 *
1042 * Returns the number of bytes written to the buffer.
1043 */
1044 static inline int zfpm_encode_mac(struct fpm_mac_info_t *mac, char *in_buf,
1045 size_t in_buf_len, fpm_msg_type_e *msg_type)
1046 {
1047 size_t len = 0;
1048
1049 *msg_type = FPM_MSG_TYPE_NONE;
1050
1051 switch (zfpm_g->message_format) {
1052
1053 case ZFPM_MSG_FORMAT_NONE:
1054 break;
1055 case ZFPM_MSG_FORMAT_NETLINK:
1056 #ifdef HAVE_NETLINK
1057 len = zfpm_netlink_encode_mac(mac, in_buf, in_buf_len);
1058 assert(fpm_msg_align(len) == len);
1059 *msg_type = FPM_MSG_TYPE_NETLINK;
1060 #endif /* HAVE_NETLINK */
1061 break;
1062 case ZFPM_MSG_FORMAT_PROTOBUF:
1063 break;
1064 }
1065 return len;
1066 }
1067
1068 static int zfpm_build_mac_updates(void)
1069 {
1070 struct stream *s;
1071 struct fpm_mac_info_t *mac;
1072 unsigned char *buf, *data, *buf_end;
1073 fpm_msg_hdr_t *hdr;
1074 size_t data_len, msg_len;
1075 fpm_msg_type_e msg_type;
1076 uint16_t q_limit;
1077
1078 if (TAILQ_EMPTY(&zfpm_g->mac_q))
1079 return FPM_GOTO_NEXT_Q;
1080
1081 s = zfpm_g->obuf;
1082 q_limit = FPM_QUEUE_PROCESS_LIMIT;
1083
1084 do {
1085 /* Make sure there is enough space to write another message. */
1086 if (STREAM_WRITEABLE(s) < FPM_MAX_MAC_MSG_LEN)
1087 return FPM_WRITE_STOP;
1088
1089 buf = STREAM_DATA(s) + stream_get_endp(s);
1090 buf_end = buf + STREAM_WRITEABLE(s);
1091
1092 mac = TAILQ_FIRST(&zfpm_g->mac_q);
1093 if (!mac)
1094 return FPM_GOTO_NEXT_Q;
1095
1096 /* Check for no-op */
1097 if (!CHECK_FLAG(mac->fpm_flags, ZEBRA_MAC_UPDATE_FPM)) {
1098 zfpm_g->stats.nop_deletes_skipped++;
1099 zfpm_mac_info_del(mac);
1100 continue;
1101 }
1102
1103 hdr = (fpm_msg_hdr_t *)buf;
1104 hdr->version = FPM_PROTO_VERSION;
1105
1106 data = fpm_msg_data(hdr);
1107 data_len = zfpm_encode_mac(mac, (char *)data, buf_end - data,
1108 &msg_type);
1109 assert(data_len);
1110
1111 hdr->msg_type = msg_type;
1112 msg_len = fpm_data_len_to_msg_len(data_len);
1113 hdr->msg_len = htons(msg_len);
1114 stream_forward_endp(s, msg_len);
1115
1116 /* Remove the MAC from the queue, and delete it. */
1117 zfpm_mac_info_del(mac);
1118
1119 q_limit--;
1120 if (q_limit == 0) {
1121 /*
1122 * We have processed enough updates in this queue.
1123 * Now yield for other queues.
1124 */
1125 return FPM_GOTO_NEXT_Q;
1126 }
1127 } while (1);
1128 }
1129
1130 /*
1131 * zfpm_build_updates
1132 *
1133 * Process the outgoing queues and write messages to the outbound
1134 * buffer.
1135 */
1136 static void zfpm_build_updates(void)
1137 {
1138 struct stream *s;
1139
1140 s = zfpm_g->obuf;
1141 assert(stream_empty(s));
1142
1143 do {
1144 /*
1145 * Stop processing the queues if zfpm_g->obuf is full
1146 * or we do not have more updates to process
1147 */
1148 if (zfpm_build_mac_updates() == FPM_WRITE_STOP)
1149 break;
1150 if (zfpm_build_route_updates() == FPM_WRITE_STOP)
1151 break;
1152 } while (zfpm_updates_pending());
1153 }
1154
1155 /*
1156 * zfpm_write_cb
1157 */
1158 static int zfpm_write_cb(struct thread *thread)
1159 {
1160 struct stream *s;
1161 int num_writes;
1162
1163 zfpm_g->stats.write_cb_calls++;
1164
1165 /*
1166 * Check if async connect is now done.
1167 */
1168 if (zfpm_g->state == ZFPM_STATE_CONNECTING) {
1169 zfpm_connect_check();
1170 return 0;
1171 }
1172
1173 assert(zfpm_g->state == ZFPM_STATE_ESTABLISHED);
1174 assert(zfpm_g->sock >= 0);
1175
1176 num_writes = 0;
1177
1178 do {
1179 int bytes_to_write, bytes_written;
1180
1181 s = zfpm_g->obuf;
1182
1183 /*
1184 * If the stream is empty, try fill it up with data.
1185 */
1186 if (stream_empty(s)) {
1187 zfpm_build_updates();
1188 }
1189
1190 bytes_to_write = stream_get_endp(s) - stream_get_getp(s);
1191 if (!bytes_to_write)
1192 break;
1193
1194 bytes_written =
1195 write(zfpm_g->sock, stream_pnt(s), bytes_to_write);
1196 zfpm_g->stats.write_calls++;
1197 num_writes++;
1198
1199 if (bytes_written < 0) {
1200 if (ERRNO_IO_RETRY(errno))
1201 break;
1202
1203 zfpm_connection_down("failed to write to socket");
1204 return 0;
1205 }
1206
1207 if (bytes_written != bytes_to_write) {
1208
1209 /*
1210 * Partial write.
1211 */
1212 stream_forward_getp(s, bytes_written);
1213 zfpm_g->stats.partial_writes++;
1214 break;
1215 }
1216
1217 /*
1218 * We've written out the entire contents of the stream.
1219 */
1220 stream_reset(s);
1221
1222 if (num_writes >= ZFPM_MAX_WRITES_PER_RUN) {
1223 zfpm_g->stats.max_writes_hit++;
1224 break;
1225 }
1226
1227 if (zfpm_thread_should_yield(thread)) {
1228 zfpm_g->stats.t_write_yields++;
1229 break;
1230 }
1231 } while (1);
1232
1233 if (zfpm_writes_pending())
1234 zfpm_write_on();
1235
1236 return 0;
1237 }
1238
1239 /*
1240 * zfpm_connect_cb
1241 */
1242 static int zfpm_connect_cb(struct thread *t)
1243 {
1244 int sock, ret;
1245 struct sockaddr_in serv;
1246
1247 assert(zfpm_g->state == ZFPM_STATE_ACTIVE);
1248
1249 sock = socket(AF_INET, SOCK_STREAM, 0);
1250 if (sock < 0) {
1251 zlog_err("Failed to create socket for connect(): %s",
1252 strerror(errno));
1253 zfpm_g->stats.connect_no_sock++;
1254 return 0;
1255 }
1256
1257 set_nonblocking(sock);
1258
1259 /* Make server socket. */
1260 memset(&serv, 0, sizeof(serv));
1261 serv.sin_family = AF_INET;
1262 serv.sin_port = htons(zfpm_g->fpm_port);
1263 #ifdef HAVE_STRUCT_SOCKADDR_IN_SIN_LEN
1264 serv.sin_len = sizeof(struct sockaddr_in);
1265 #endif /* HAVE_STRUCT_SOCKADDR_IN_SIN_LEN */
1266 if (!zfpm_g->fpm_server)
1267 serv.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
1268 else
1269 serv.sin_addr.s_addr = (zfpm_g->fpm_server);
1270
1271 /*
1272 * Connect to the FPM.
1273 */
1274 zfpm_g->connect_calls++;
1275 zfpm_g->stats.connect_calls++;
1276 zfpm_g->last_connect_call_time = monotime(NULL);
1277
1278 ret = connect(sock, (struct sockaddr *)&serv, sizeof(serv));
1279 if (ret >= 0) {
1280 zfpm_g->sock = sock;
1281 zfpm_connection_up("connect succeeded");
1282 return 1;
1283 }
1284
1285 if (errno == EINPROGRESS) {
1286 zfpm_g->sock = sock;
1287 zfpm_read_on();
1288 zfpm_write_on();
1289 zfpm_set_state(ZFPM_STATE_CONNECTING,
1290 "async connect in progress");
1291 return 0;
1292 }
1293
1294 zlog_info("can't connect to FPM %d: %s", sock, safe_strerror(errno));
1295 close(sock);
1296
1297 /*
1298 * Restart timer for retrying connection.
1299 */
1300 zfpm_start_connect_timer("connect() failed");
1301 return 0;
1302 }
1303
1304 /*
1305 * zfpm_set_state
1306 *
1307 * Move state machine into the given state.
1308 */
1309 static void zfpm_set_state(zfpm_state_t state, const char *reason)
1310 {
1311 zfpm_state_t cur_state = zfpm_g->state;
1312
1313 if (!reason)
1314 reason = "Unknown";
1315
1316 if (state == cur_state)
1317 return;
1318
1319 zfpm_debug("beginning state transition %s -> %s. Reason: %s",
1320 zfpm_state_to_str(cur_state), zfpm_state_to_str(state),
1321 reason);
1322
1323 switch (state) {
1324
1325 case ZFPM_STATE_IDLE:
1326 assert(cur_state == ZFPM_STATE_ESTABLISHED);
1327 break;
1328
1329 case ZFPM_STATE_ACTIVE:
1330 assert(cur_state == ZFPM_STATE_IDLE
1331 || cur_state == ZFPM_STATE_CONNECTING);
1332 assert(zfpm_g->t_connect);
1333 break;
1334
1335 case ZFPM_STATE_CONNECTING:
1336 assert(zfpm_g->sock);
1337 assert(cur_state == ZFPM_STATE_ACTIVE);
1338 assert(zfpm_g->t_read);
1339 assert(zfpm_g->t_write);
1340 break;
1341
1342 case ZFPM_STATE_ESTABLISHED:
1343 assert(cur_state == ZFPM_STATE_ACTIVE
1344 || cur_state == ZFPM_STATE_CONNECTING);
1345 assert(zfpm_g->sock);
1346 assert(zfpm_g->t_read);
1347 assert(zfpm_g->t_write);
1348 break;
1349 }
1350
1351 zfpm_g->state = state;
1352 }
1353
1354 /*
1355 * zfpm_calc_connect_delay
1356 *
1357 * Returns the number of seconds after which we should attempt to
1358 * reconnect to the FPM.
1359 */
1360 static long zfpm_calc_connect_delay(void)
1361 {
1362 time_t elapsed;
1363
1364 /*
1365 * Return 0 if this is our first attempt to connect.
1366 */
1367 if (zfpm_g->connect_calls == 0) {
1368 return 0;
1369 }
1370
1371 elapsed = zfpm_get_elapsed_time(zfpm_g->last_connect_call_time);
1372
1373 if (elapsed > ZFPM_CONNECT_RETRY_IVL) {
1374 return 0;
1375 }
1376
1377 return ZFPM_CONNECT_RETRY_IVL - elapsed;
1378 }
1379
1380 /*
1381 * zfpm_start_connect_timer
1382 */
1383 static void zfpm_start_connect_timer(const char *reason)
1384 {
1385 long delay_secs;
1386
1387 assert(!zfpm_g->t_connect);
1388 assert(zfpm_g->sock < 0);
1389
1390 assert(zfpm_g->state == ZFPM_STATE_IDLE
1391 || zfpm_g->state == ZFPM_STATE_ACTIVE
1392 || zfpm_g->state == ZFPM_STATE_CONNECTING);
1393
1394 delay_secs = zfpm_calc_connect_delay();
1395 zfpm_debug("scheduling connect in %ld seconds", delay_secs);
1396
1397 thread_add_timer(zfpm_g->master, zfpm_connect_cb, 0, delay_secs,
1398 &zfpm_g->t_connect);
1399 zfpm_set_state(ZFPM_STATE_ACTIVE, reason);
1400 }
1401
1402 /*
1403 * zfpm_is_enabled
1404 *
1405 * Returns true if the zebra FPM module has been enabled.
1406 */
1407 static inline int zfpm_is_enabled(void)
1408 {
1409 return zfpm_g->enabled;
1410 }
1411
1412 /*
1413 * zfpm_conn_is_up
1414 *
1415 * Returns true if the connection to the FPM is up.
1416 */
1417 static inline int zfpm_conn_is_up(void)
1418 {
1419 if (zfpm_g->state != ZFPM_STATE_ESTABLISHED)
1420 return 0;
1421
1422 assert(zfpm_g->sock >= 0);
1423
1424 return 1;
1425 }
1426
1427 /*
1428 * zfpm_trigger_update
1429 *
1430 * The zebra code invokes this function to indicate that we should
1431 * send an update to the FPM about the given route_node.
1432 */
1433 static int zfpm_trigger_update(struct route_node *rn, const char *reason)
1434 {
1435 rib_dest_t *dest;
1436 char buf[PREFIX_STRLEN];
1437
1438 /*
1439 * Ignore if the connection is down. We will update the FPM about
1440 * all destinations once the connection comes up.
1441 */
1442 if (!zfpm_conn_is_up())
1443 return 0;
1444
1445 dest = rib_dest_from_rnode(rn);
1446
1447 if (CHECK_FLAG(dest->flags, RIB_DEST_UPDATE_FPM)) {
1448 zfpm_g->stats.redundant_triggers++;
1449 return 0;
1450 }
1451
1452 if (reason) {
1453 zfpm_debug("%s triggering update to FPM - Reason: %s",
1454 prefix2str(&rn->p, buf, sizeof(buf)), reason);
1455 }
1456
1457 SET_FLAG(dest->flags, RIB_DEST_UPDATE_FPM);
1458 TAILQ_INSERT_TAIL(&zfpm_g->dest_q, dest, fpm_q_entries);
1459 zfpm_g->stats.updates_triggered++;
1460
1461 /*
1462 * Make sure that writes are enabled.
1463 */
1464 if (zfpm_g->t_write)
1465 return 0;
1466
1467 zfpm_write_on();
1468 return 0;
1469 }
1470
1471 /*
1472 * Generate Key for FPM MAC info hash entry
1473 * Key is generated using MAC address and VNI id which should be sufficient
1474 * to provide uniqueness
1475 */
1476 static unsigned int zfpm_mac_info_hash_keymake(const void *p)
1477 {
1478 struct fpm_mac_info_t *fpm_mac = (struct fpm_mac_info_t *)p;
1479 uint32_t mac_key;
1480
1481 mac_key = jhash(fpm_mac->macaddr.octet, ETH_ALEN, 0xa5a5a55a);
1482
1483 return jhash_2words(mac_key, fpm_mac->vni, 0);
1484 }
1485
1486 /*
1487 * Compare function for FPM MAC info hash lookup
1488 */
1489 static bool zfpm_mac_info_cmp(const void *p1, const void *p2)
1490 {
1491 const struct fpm_mac_info_t *fpm_mac1 = p1;
1492 const struct fpm_mac_info_t *fpm_mac2 = p2;
1493
1494 if (memcmp(fpm_mac1->macaddr.octet, fpm_mac2->macaddr.octet, ETH_ALEN)
1495 != 0)
1496 return false;
1497 if (fpm_mac1->r_vtep_ip.s_addr != fpm_mac2->r_vtep_ip.s_addr)
1498 return false;
1499 if (fpm_mac1->vni != fpm_mac2->vni)
1500 return false;
1501
1502 return true;
1503 }
1504
1505 /*
1506 * Lookup FPM MAC info hash entry.
1507 */
1508 static struct fpm_mac_info_t *zfpm_mac_info_lookup(struct fpm_mac_info_t *key)
1509 {
1510 return hash_lookup(zfpm_g->fpm_mac_info_table, key);
1511 }
1512
1513 /*
1514 * Callback to allocate fpm_mac_info_t structure.
1515 */
1516 static void *zfpm_mac_info_alloc(void *p)
1517 {
1518 const struct fpm_mac_info_t *key = p;
1519 struct fpm_mac_info_t *fpm_mac;
1520
1521 fpm_mac = XCALLOC(MTYPE_FPM_MAC_INFO, sizeof(struct fpm_mac_info_t));
1522
1523 memcpy(&fpm_mac->macaddr, &key->macaddr, ETH_ALEN);
1524 memcpy(&fpm_mac->r_vtep_ip, &key->r_vtep_ip, sizeof(struct in_addr));
1525 fpm_mac->vni = key->vni;
1526
1527 return (void *)fpm_mac;
1528 }
1529
1530 /*
1531 * Delink and free fpm_mac_info_t.
1532 */
1533 static void zfpm_mac_info_del(struct fpm_mac_info_t *fpm_mac)
1534 {
1535 hash_release(zfpm_g->fpm_mac_info_table, fpm_mac);
1536 TAILQ_REMOVE(&zfpm_g->mac_q, fpm_mac, fpm_mac_q_entries);
1537 XFREE(MTYPE_FPM_MAC_INFO, fpm_mac);
1538 }
1539
1540 /*
1541 * zfpm_trigger_rmac_update
1542 *
1543 * Zebra code invokes this function to indicate that we should
1544 * send an update to FPM for given MAC entry.
1545 *
1546 * This function checks if we already have enqueued an update for this RMAC,
1547 * If yes, update the same fpm_mac_info_t. Else, create and enqueue an update.
1548 */
1549 static int zfpm_trigger_rmac_update(zebra_mac_t *rmac, zebra_l3vni_t *zl3vni,
1550 bool delete, const char *reason)
1551 {
1552 char buf[ETHER_ADDR_STRLEN];
1553 struct fpm_mac_info_t *fpm_mac, key;
1554 struct interface *vxlan_if, *svi_if;
1555
1556 /*
1557 * Ignore if the connection is down. We will update the FPM about
1558 * all destinations once the connection comes up.
1559 */
1560 if (!zfpm_conn_is_up())
1561 return 0;
1562
1563 if (reason) {
1564 zfpm_debug("triggering update to FPM - Reason: %s - %s",
1565 reason,
1566 prefix_mac2str(&rmac->macaddr, buf, sizeof(buf)));
1567 }
1568
1569 vxlan_if = zl3vni_map_to_vxlan_if(zl3vni);
1570 svi_if = zl3vni_map_to_svi_if(zl3vni);
1571
1572 memset(&key, 0, sizeof(struct fpm_mac_info_t));
1573
1574 memcpy(&key.macaddr, &rmac->macaddr, ETH_ALEN);
1575 key.r_vtep_ip.s_addr = rmac->fwd_info.r_vtep_ip.s_addr;
1576 key.vni = zl3vni->vni;
1577
1578 /* Check if this MAC is already present in the queue. */
1579 fpm_mac = zfpm_mac_info_lookup(&key);
1580
1581 if (fpm_mac) {
1582 if (!!CHECK_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_DELETE_FPM)
1583 == delete) {
1584 /*
1585 * MAC is already present in the queue
1586 * with the same op as this one. Do nothing
1587 */
1588 zfpm_g->stats.redundant_triggers++;
1589 return 0;
1590 }
1591
1592 /*
1593 * A new op for an already existing fpm_mac_info_t node.
1594 * Update the existing node for the new op.
1595 */
1596 if (!delete) {
1597 /*
1598 * New op is "add". Previous op is "delete".
1599 * Update the fpm_mac_info_t for the new add.
1600 */
1601 fpm_mac->zebra_flags = rmac->flags;
1602
1603 fpm_mac->vxlan_if = vxlan_if ? vxlan_if->ifindex : 0;
1604 fpm_mac->svi_if = svi_if ? svi_if->ifindex : 0;
1605
1606 UNSET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_DELETE_FPM);
1607 SET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_UPDATE_FPM);
1608 } else {
1609 /*
1610 * New op is "delete". Previous op is "add".
1611 * Thus, no-op. Unset ZEBRA_MAC_UPDATE_FPM flag.
1612 */
1613 SET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_DELETE_FPM);
1614 UNSET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_UPDATE_FPM);
1615 }
1616
1617 return 0;
1618 }
1619
1620 fpm_mac = hash_get(zfpm_g->fpm_mac_info_table, &key,
1621 zfpm_mac_info_alloc);
1622 if (!fpm_mac)
1623 return 0;
1624
1625 fpm_mac->zebra_flags = rmac->flags;
1626 fpm_mac->vxlan_if = vxlan_if ? vxlan_if->ifindex : 0;
1627 fpm_mac->svi_if = svi_if ? svi_if->ifindex : 0;
1628
1629 SET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_UPDATE_FPM);
1630 if (delete)
1631 SET_FLAG(fpm_mac->fpm_flags, ZEBRA_MAC_DELETE_FPM);
1632
1633 TAILQ_INSERT_TAIL(&zfpm_g->mac_q, fpm_mac, fpm_mac_q_entries);
1634
1635 zfpm_g->stats.updates_triggered++;
1636
1637 /* If writes are already enabled, return. */
1638 if (zfpm_g->t_write)
1639 return 0;
1640
1641 zfpm_write_on();
1642 return 0;
1643 }
1644
1645 /*
1646 * This function is called when the FPM connections is established.
1647 * Iterate over all the RMAC entries for the given L3VNI
1648 * and enqueue the RMAC for FPM processing.
1649 */
1650 static void zfpm_trigger_rmac_update_wrapper(struct hash_backet *backet,
1651 void *args)
1652 {
1653 zebra_mac_t *zrmac = (zebra_mac_t *)backet->data;
1654 zebra_l3vni_t *zl3vni = (zebra_l3vni_t *)args;
1655
1656 zfpm_trigger_rmac_update(zrmac, zl3vni, false, "RMAC added");
1657 }
1658
1659 /*
1660 * This function is called when the FPM connections is established.
1661 * This function iterates over all the L3VNIs to trigger
1662 * FPM updates for RMACs currently available.
1663 */
1664 static void zfpm_iterate_rmac_table(struct hash_backet *backet, void *args)
1665 {
1666 zebra_l3vni_t *zl3vni = (zebra_l3vni_t *)backet->data;
1667
1668 hash_iterate(zl3vni->rmac_table, zfpm_trigger_rmac_update_wrapper,
1669 (void *)zl3vni);
1670 }
1671
1672 /*
1673 * zfpm_stats_timer_cb
1674 */
1675 static int zfpm_stats_timer_cb(struct thread *t)
1676 {
1677 zfpm_g->t_stats = NULL;
1678
1679 /*
1680 * Remember the stats collected in the last interval for display
1681 * purposes.
1682 */
1683 zfpm_stats_copy(&zfpm_g->stats, &zfpm_g->last_ivl_stats);
1684
1685 /*
1686 * Add the current set of stats into the cumulative statistics.
1687 */
1688 zfpm_stats_compose(&zfpm_g->cumulative_stats, &zfpm_g->stats,
1689 &zfpm_g->cumulative_stats);
1690
1691 /*
1692 * Start collecting stats afresh over the next interval.
1693 */
1694 zfpm_stats_reset(&zfpm_g->stats);
1695
1696 zfpm_start_stats_timer();
1697
1698 return 0;
1699 }
1700
1701 /*
1702 * zfpm_stop_stats_timer
1703 */
1704 static void zfpm_stop_stats_timer(void)
1705 {
1706 if (!zfpm_g->t_stats)
1707 return;
1708
1709 zfpm_debug("Stopping existing stats timer");
1710 THREAD_TIMER_OFF(zfpm_g->t_stats);
1711 }
1712
1713 /*
1714 * zfpm_start_stats_timer
1715 */
1716 void zfpm_start_stats_timer(void)
1717 {
1718 assert(!zfpm_g->t_stats);
1719
1720 thread_add_timer(zfpm_g->master, zfpm_stats_timer_cb, 0,
1721 ZFPM_STATS_IVL_SECS, &zfpm_g->t_stats);
1722 }
1723
1724 /*
1725 * Helper macro for zfpm_show_stats() below.
1726 */
1727 #define ZFPM_SHOW_STAT(counter) \
1728 do { \
1729 vty_out(vty, "%-40s %10lu %16lu\n", #counter, \
1730 total_stats.counter, zfpm_g->last_ivl_stats.counter); \
1731 } while (0)
1732
1733 /*
1734 * zfpm_show_stats
1735 */
1736 static void zfpm_show_stats(struct vty *vty)
1737 {
1738 zfpm_stats_t total_stats;
1739 time_t elapsed;
1740
1741 vty_out(vty, "\n%-40s %10s Last %2d secs\n\n", "Counter", "Total",
1742 ZFPM_STATS_IVL_SECS);
1743
1744 /*
1745 * Compute the total stats up to this instant.
1746 */
1747 zfpm_stats_compose(&zfpm_g->cumulative_stats, &zfpm_g->stats,
1748 &total_stats);
1749
1750 ZFPM_SHOW_STAT(connect_calls);
1751 ZFPM_SHOW_STAT(connect_no_sock);
1752 ZFPM_SHOW_STAT(read_cb_calls);
1753 ZFPM_SHOW_STAT(write_cb_calls);
1754 ZFPM_SHOW_STAT(write_calls);
1755 ZFPM_SHOW_STAT(partial_writes);
1756 ZFPM_SHOW_STAT(max_writes_hit);
1757 ZFPM_SHOW_STAT(t_write_yields);
1758 ZFPM_SHOW_STAT(nop_deletes_skipped);
1759 ZFPM_SHOW_STAT(route_adds);
1760 ZFPM_SHOW_STAT(route_dels);
1761 ZFPM_SHOW_STAT(updates_triggered);
1762 ZFPM_SHOW_STAT(redundant_triggers);
1763 ZFPM_SHOW_STAT(dests_del_after_update);
1764 ZFPM_SHOW_STAT(t_conn_down_starts);
1765 ZFPM_SHOW_STAT(t_conn_down_dests_processed);
1766 ZFPM_SHOW_STAT(t_conn_down_yields);
1767 ZFPM_SHOW_STAT(t_conn_down_finishes);
1768 ZFPM_SHOW_STAT(t_conn_up_starts);
1769 ZFPM_SHOW_STAT(t_conn_up_dests_processed);
1770 ZFPM_SHOW_STAT(t_conn_up_yields);
1771 ZFPM_SHOW_STAT(t_conn_up_aborts);
1772 ZFPM_SHOW_STAT(t_conn_up_finishes);
1773
1774 if (!zfpm_g->last_stats_clear_time)
1775 return;
1776
1777 elapsed = zfpm_get_elapsed_time(zfpm_g->last_stats_clear_time);
1778
1779 vty_out(vty, "\nStats were cleared %lu seconds ago\n",
1780 (unsigned long)elapsed);
1781 }
1782
1783 /*
1784 * zfpm_clear_stats
1785 */
1786 static void zfpm_clear_stats(struct vty *vty)
1787 {
1788 if (!zfpm_is_enabled()) {
1789 vty_out(vty, "The FPM module is not enabled...\n");
1790 return;
1791 }
1792
1793 zfpm_stats_reset(&zfpm_g->stats);
1794 zfpm_stats_reset(&zfpm_g->last_ivl_stats);
1795 zfpm_stats_reset(&zfpm_g->cumulative_stats);
1796
1797 zfpm_stop_stats_timer();
1798 zfpm_start_stats_timer();
1799
1800 zfpm_g->last_stats_clear_time = monotime(NULL);
1801
1802 vty_out(vty, "Cleared FPM stats\n");
1803 }
1804
1805 /*
1806 * show_zebra_fpm_stats
1807 */
1808 DEFUN (show_zebra_fpm_stats,
1809 show_zebra_fpm_stats_cmd,
1810 "show zebra fpm stats",
1811 SHOW_STR
1812 ZEBRA_STR
1813 "Forwarding Path Manager information\n"
1814 "Statistics\n")
1815 {
1816 zfpm_show_stats(vty);
1817 return CMD_SUCCESS;
1818 }
1819
1820 /*
1821 * clear_zebra_fpm_stats
1822 */
1823 DEFUN (clear_zebra_fpm_stats,
1824 clear_zebra_fpm_stats_cmd,
1825 "clear zebra fpm stats",
1826 CLEAR_STR
1827 ZEBRA_STR
1828 "Clear Forwarding Path Manager information\n"
1829 "Statistics\n")
1830 {
1831 zfpm_clear_stats(vty);
1832 return CMD_SUCCESS;
1833 }
1834
1835 /*
1836 * update fpm connection information
1837 */
1838 DEFUN ( fpm_remote_ip,
1839 fpm_remote_ip_cmd,
1840 "fpm connection ip A.B.C.D port (1-65535)",
1841 "fpm connection remote ip and port\n"
1842 "Remote fpm server ip A.B.C.D\n"
1843 "Enter ip ")
1844 {
1845
1846 in_addr_t fpm_server;
1847 uint32_t port_no;
1848
1849 fpm_server = inet_addr(argv[3]->arg);
1850 if (fpm_server == INADDR_NONE)
1851 return CMD_ERR_INCOMPLETE;
1852
1853 port_no = atoi(argv[5]->arg);
1854 if (port_no < TCP_MIN_PORT || port_no > TCP_MAX_PORT)
1855 return CMD_ERR_INCOMPLETE;
1856
1857 zfpm_g->fpm_server = fpm_server;
1858 zfpm_g->fpm_port = port_no;
1859
1860
1861 return CMD_SUCCESS;
1862 }
1863
1864 DEFUN ( no_fpm_remote_ip,
1865 no_fpm_remote_ip_cmd,
1866 "no fpm connection ip A.B.C.D port (1-65535)",
1867 "fpm connection remote ip and port\n"
1868 "Connection\n"
1869 "Remote fpm server ip A.B.C.D\n"
1870 "Enter ip ")
1871 {
1872 if (zfpm_g->fpm_server != inet_addr(argv[4]->arg)
1873 || zfpm_g->fpm_port != atoi(argv[6]->arg))
1874 return CMD_ERR_NO_MATCH;
1875
1876 zfpm_g->fpm_server = FPM_DEFAULT_IP;
1877 zfpm_g->fpm_port = FPM_DEFAULT_PORT;
1878
1879 return CMD_SUCCESS;
1880 }
1881
1882 /*
1883 * zfpm_init_message_format
1884 */
1885 static inline void zfpm_init_message_format(const char *format)
1886 {
1887 int have_netlink, have_protobuf;
1888
1889 #ifdef HAVE_NETLINK
1890 have_netlink = 1;
1891 #else
1892 have_netlink = 0;
1893 #endif
1894
1895 #ifdef HAVE_PROTOBUF
1896 have_protobuf = 1;
1897 #else
1898 have_protobuf = 0;
1899 #endif
1900
1901 zfpm_g->message_format = ZFPM_MSG_FORMAT_NONE;
1902
1903 if (!format) {
1904 if (have_netlink) {
1905 zfpm_g->message_format = ZFPM_MSG_FORMAT_NETLINK;
1906 } else if (have_protobuf) {
1907 zfpm_g->message_format = ZFPM_MSG_FORMAT_PROTOBUF;
1908 }
1909 return;
1910 }
1911
1912 if (!strcmp("netlink", format)) {
1913 if (!have_netlink) {
1914 flog_err(EC_ZEBRA_NETLINK_NOT_AVAILABLE,
1915 "FPM netlink message format is not available");
1916 return;
1917 }
1918 zfpm_g->message_format = ZFPM_MSG_FORMAT_NETLINK;
1919 return;
1920 }
1921
1922 if (!strcmp("protobuf", format)) {
1923 if (!have_protobuf) {
1924 flog_err(
1925 EC_ZEBRA_PROTOBUF_NOT_AVAILABLE,
1926 "FPM protobuf message format is not available");
1927 return;
1928 }
1929 flog_warn(EC_ZEBRA_PROTOBUF_NOT_AVAILABLE,
1930 "FPM protobuf message format is deprecated and scheduled to be removed. "
1931 "Please convert to using netlink format or contact dev@lists.frrouting.org with your use case.");
1932 zfpm_g->message_format = ZFPM_MSG_FORMAT_PROTOBUF;
1933 return;
1934 }
1935
1936 flog_warn(EC_ZEBRA_FPM_FORMAT_UNKNOWN, "Unknown fpm format '%s'",
1937 format);
1938 }
1939
1940 /**
1941 * fpm_remote_srv_write
1942 *
1943 * Module to write remote fpm connection
1944 *
1945 * Returns ZERO on success.
1946 */
1947
1948 static int fpm_remote_srv_write(struct vty *vty)
1949 {
1950 struct in_addr in;
1951
1952 in.s_addr = zfpm_g->fpm_server;
1953
1954 if ((zfpm_g->fpm_server != FPM_DEFAULT_IP
1955 && zfpm_g->fpm_server != INADDR_ANY)
1956 || (zfpm_g->fpm_port != FPM_DEFAULT_PORT && zfpm_g->fpm_port != 0))
1957 vty_out(vty, "fpm connection ip %s port %d\n", inet_ntoa(in),
1958 zfpm_g->fpm_port);
1959
1960 return 0;
1961 }
1962
1963
1964 /* Zebra node */
1965 static struct cmd_node zebra_node = {ZEBRA_NODE, "", 1};
1966
1967
1968 /**
1969 * zfpm_init
1970 *
1971 * One-time initialization of the Zebra FPM module.
1972 *
1973 * @param[in] port port at which FPM is running.
1974 * @param[in] enable true if the zebra FPM module should be enabled
1975 * @param[in] format to use to talk to the FPM. Can be 'netink' or 'protobuf'.
1976 *
1977 * Returns true on success.
1978 */
1979 static int zfpm_init(struct thread_master *master)
1980 {
1981 int enable = 1;
1982 uint16_t port = 0;
1983 const char *format = THIS_MODULE->load_args;
1984
1985 memset(zfpm_g, 0, sizeof(*zfpm_g));
1986 zfpm_g->master = master;
1987 TAILQ_INIT(&zfpm_g->dest_q);
1988 TAILQ_INIT(&zfpm_g->mac_q);
1989
1990 /* Create hash table for fpm_mac_info_t enties */
1991 zfpm_g->fpm_mac_info_table = hash_create(zfpm_mac_info_hash_keymake,
1992 zfpm_mac_info_cmp,
1993 "FPM MAC info hash table");
1994
1995 zfpm_g->sock = -1;
1996 zfpm_g->state = ZFPM_STATE_IDLE;
1997
1998 zfpm_stats_init(&zfpm_g->stats);
1999 zfpm_stats_init(&zfpm_g->last_ivl_stats);
2000 zfpm_stats_init(&zfpm_g->cumulative_stats);
2001
2002 install_node(&zebra_node, fpm_remote_srv_write);
2003 install_element(ENABLE_NODE, &show_zebra_fpm_stats_cmd);
2004 install_element(ENABLE_NODE, &clear_zebra_fpm_stats_cmd);
2005 install_element(CONFIG_NODE, &fpm_remote_ip_cmd);
2006 install_element(CONFIG_NODE, &no_fpm_remote_ip_cmd);
2007
2008 zfpm_init_message_format(format);
2009
2010 /*
2011 * Disable FPM interface if no suitable format is available.
2012 */
2013 if (zfpm_g->message_format == ZFPM_MSG_FORMAT_NONE)
2014 enable = 0;
2015
2016 zfpm_g->enabled = enable;
2017
2018 if (!zfpm_g->fpm_server)
2019 zfpm_g->fpm_server = FPM_DEFAULT_IP;
2020
2021 if (!port)
2022 port = FPM_DEFAULT_PORT;
2023
2024 zfpm_g->fpm_port = port;
2025
2026 zfpm_g->obuf = stream_new(ZFPM_OBUF_SIZE);
2027 zfpm_g->ibuf = stream_new(ZFPM_IBUF_SIZE);
2028
2029 zfpm_start_stats_timer();
2030 zfpm_start_connect_timer("initialized");
2031 return 0;
2032 }
2033
2034 static int zfpm_fini(void)
2035 {
2036 zfpm_write_off();
2037 zfpm_read_off();
2038 zfpm_connect_off();
2039
2040 zfpm_stop_stats_timer();
2041
2042 hook_unregister(rib_update, zfpm_trigger_update);
2043 return 0;
2044 }
2045
2046 static int zebra_fpm_module_init(void)
2047 {
2048 hook_register(rib_update, zfpm_trigger_update);
2049 hook_register(zebra_rmac_update, zfpm_trigger_rmac_update);
2050 hook_register(frr_late_init, zfpm_init);
2051 hook_register(frr_early_fini, zfpm_fini);
2052 return 0;
2053 }
2054
2055 FRR_MODULE_SETUP(.name = "zebra_fpm", .version = FRR_VERSION,
2056 .description = "zebra FPM (Forwarding Plane Manager) module",
2057 .init = zebra_fpm_module_init, )