src/plugins/nat/nat_ha.c

   1 /*
   2  * Copyright (c) 2019 Cisco and/or its affiliates.
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at:
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include "nat_ha.h"
  17 #include <vnet/udp/udp_local.h>
  18 #include <nat/nat.h>
  19 #include <vppinfra/atomics.h>
  20
  21 /* number of retries */
  22 #define NAT_HA_RETRIES 3
  23
  24 #define foreach_nat_ha_counter           \
  25 _(RECV_ADD, "add-event-recv", 0)         \
  26 _(RECV_DEL, "del-event-recv", 1)         \
  27 _(RECV_REFRESH, "refresh-event-recv", 2) \
  28 _(SEND_ADD, "add-event-send", 3)         \
  29 _(SEND_DEL, "del-event-send", 4)         \
  30 _(SEND_REFRESH, "refresh-event-send", 5) \
  31 _(RECV_ACK, "ack-recv", 6)               \
  32 _(SEND_ACK, "ack-send", 7)               \
  33 _(RETRY_COUNT, "retry-count", 8)         \
  34 _(MISSED_COUNT, "missed-count", 9)
  35
  36 /* NAT HA protocol version */
  37 #define NAT_HA_VERSION 0x01
  38
  39 /* NAT HA protocol flags */
  40 #define NAT_HA_FLAG_ACK 0x01
  41
  42 /* NAT HA event types */
  43 typedef enum
  44 {
  45   NAT_HA_ADD = 1,
  46   NAT_HA_DEL,
  47   NAT_HA_REFRESH,
  48 } nat_ha_event_type_t;
  49
  50 /* NAT HA protocol header */
  51 typedef struct
  52 {
  53   /* version */
  54   u8 version;
  55   /* flags */
  56   u8 flags;
  57   /* event count */
  58   u16 count;
  59   /* sequence number */
  60   u32 sequence_number;
  61   /* thread index where events originated */
  62   u32 thread_index;
  63 } __attribute__ ((packed)) nat_ha_message_header_t;
  64
  65 /* NAT HA protocol event data */
  66 typedef struct
  67 {
  68   /* event type */
  69   u8 event_type;
  70   /* session data */
  71   u8 protocol;
  72   u16 flags;
  73   u32 in_addr;
  74   u32 out_addr;
  75   u16 in_port;
  76   u16 out_port;
  77   u32 eh_addr;
  78   u32 ehn_addr;
  79   u16 eh_port;
  80   u16 ehn_port;
  81   u32 fib_index;
  82   u32 total_pkts;
  83   u64 total_bytes;
  84 } __attribute__ ((packed)) nat_ha_event_t;
  85
  86 typedef enum
  87 {
  88 #define _(N, s, v) NAT_HA_COUNTER_##N = v,
  89   foreach_nat_ha_counter
  90 #undef _
  91   NAT_HA_N_COUNTERS
  92 } nat_ha_counter_t;
  93
  94 /* data waiting for ACK */
  95 typedef struct
  96 {
  97   /* sequence number */
  98   u32 seq;
  99   /* retry count */
 100   u32 retry_count;
 101   /* next retry time */
 102   f64 retry_timer;
 103   /* 1 if HA resync */
 104   u8 is_resync;
 105   /* packet data */
 106   u8 *data;
 107 } nat_ha_resend_entry_t;
 108
 109 /* per thread data */
 110 typedef struct
 111 {
 112   /* buffer under construction */
 113   vlib_buffer_t *state_sync_buffer;
 114   /* frame containing NAT HA buffers */
 115   vlib_frame_t *state_sync_frame;
 116   /* number of events */
 117   u16 state_sync_count;
 118   /* next event offset */
 119   u32 state_sync_next_event_offset;
 120   /* data waiting for ACK */
 121   nat_ha_resend_entry_t *resend_queue;
 122 } nat_ha_per_thread_data_t;
 123
 124 /* NAT HA settings */
 125 typedef struct nat_ha_main_s
 126 {
 127   u8 enabled;
 128   /* local IP address and UDP port */
 129   ip4_address_t src_ip_address;
 130   u16 src_port;
 131   /* failvoer IP address and UDP port */
 132   ip4_address_t dst_ip_address;
 133   u16 dst_port;
 134   /* path MTU between local and failover */
 135   u32 state_sync_path_mtu;
 136   /* number of seconds after which to send session counters refresh */
 137   u32 session_refresh_interval;
 138   /* counters */
 139   vlib_simple_counter_main_t counters[NAT_HA_N_COUNTERS];
 140   vlib_main_t *vlib_main;
 141   /* sequence number counter */
 142   u32 sequence_number;
 143   /* 1 if resync in progress */
 144   u8 in_resync;
 145   /* number of remaing ACK for resync */
 146   u32 resync_ack_count;
 147   /* number of missed ACK for resync */
 148   u32 resync_ack_missed;
 149   /* resync data */
 150   nat_ha_resync_event_cb_t event_callback;
 151   u32 client_index;
 152   u32 pid;
 153   /* call back functions for received HA events on failover */
 154   nat_ha_sadd_cb_t sadd_cb;
 155   nat_ha_sdel_cb_t sdel_cb;
 156   nat_ha_sref_cb_t sref_cb;
 157   /* per thread data */
 158   u32 num_workers;
 159   nat_ha_per_thread_data_t *per_thread_data;
 160   /* worker handoff frame-queue index */
 161   u32 fq_index;
 162 } nat_ha_main_t;
 163
 164 nat_ha_main_t nat_ha_main;
 165 vlib_node_registration_t nat_ha_process_node;
 166 vlib_node_registration_t nat_ha_worker_node;
 167 vlib_node_registration_t nat_ha_node;
 168 vlib_node_registration_t nat_ha_handoff_node;
 169
 170 static void
 171 nat_ha_resync_fin (void)
 172 {
 173   nat_ha_main_t *ha = &nat_ha_main;
 174
 175   /* if no more resync ACK remainig we are done */
 176   if (ha->resync_ack_count)
 177     return;
 178
 179   ha->in_resync = 0;
 180   if (ha->resync_ack_missed)
 181     {
 182       nat_elog_info ("resync completed with result FAILED");
 183     }
 184   else
 185     {
 186       nat_elog_info ("resync completed with result SUCCESS");
 187     }
 188   if (ha->event_callback)
 189     ha->event_callback (ha->client_index, ha->pid, ha->resync_ack_missed);
 190 }
 191
 192 /* cache HA NAT data waiting for ACK */
 193 static int
 194 nat_ha_resend_queue_add (u32 seq, u8 * data, u8 data_len, u8 is_resync,
 195                          u32 thread_index)
 196 {
 197   nat_ha_main_t *ha = &nat_ha_main;
 198   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 199   nat_ha_resend_entry_t *entry;
 200   f64 now = vlib_time_now (ha->vlib_main);
 201
 202   vec_add2 (td->resend_queue, entry, 1);
 203   clib_memset (entry, 0, sizeof (*entry));
 204   entry->retry_timer = now + 2.0;
 205   entry->seq = seq;
 206   entry->is_resync = is_resync;
 207   vec_add (entry->data, data, data_len);
 208
 209   return 0;
 210 }
 211
 212 static_always_inline void
 213 nat_ha_ack_recv (u32 seq, u32 thread_index)
 214 {
 215   nat_ha_main_t *ha = &nat_ha_main;
 216   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 217   u32 i;
 218
 219   vec_foreach_index (i, td->resend_queue)
 220   {
 221     if (td->resend_queue[i].seq != seq)
 222       continue;
 223
 224     vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_ACK],
 225                                    thread_index, 0, 1);
 226     /* ACK received remove cached data */
 227     if (td->resend_queue[i].is_resync)
 228       {
 229         clib_atomic_fetch_sub (&ha->resync_ack_count, 1);
 230         nat_ha_resync_fin ();
 231       }
 232     vec_free (td->resend_queue[i].data);
 233     vec_del1 (td->resend_queue, i);
 234     nat_elog_debug_X1 ("ACK for seq %d received", "i4",
 235                        clib_net_to_host_u32 (seq));
 236
 237     return;
 238   }
 239 }
 240
 241 /* scan non-ACKed HA NAT for retry */
 242 static void
 243 nat_ha_resend_scan (f64 now, u32 thread_index)
 244 {
 245   nat_ha_main_t *ha = &nat_ha_main;
 246   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 247   u32 i, *del, *to_delete = 0;
 248   vlib_main_t *vm = ha->vlib_main;
 249   vlib_buffer_t *b = 0;
 250   vlib_frame_t *f;
 251   u32 bi, *to_next;
 252   ip4_header_t *ip;
 253
 254   vec_foreach_index (i, td->resend_queue)
 255   {
 256     if (td->resend_queue[i].retry_timer > now)
 257       continue;
 258
 259     /* maximum retry reached delete cached data */
 260     if (td->resend_queue[i].retry_count >= NAT_HA_RETRIES)
 261       {
 262         nat_elog_notice_X1 ("seq %d missed", "i4",
 263                             clib_net_to_host_u32 (td->resend_queue[i].seq));
 264         if (td->resend_queue[i].is_resync)
 265           {
 266             clib_atomic_fetch_add (&ha->resync_ack_missed, 1);
 267             clib_atomic_fetch_sub (&ha->resync_ack_count, 1);
 268             nat_ha_resync_fin ();
 269           }
 270         vec_add1 (to_delete, i);
 271         vlib_increment_simple_counter (&ha->counters
 272                                        [NAT_HA_COUNTER_MISSED_COUNT],
 273                                        thread_index, 0, 1);
 274         continue;
 275       }
 276
 277     /* retry to send non-ACKed data */
 278     nat_elog_debug_X1 ("state sync seq %d resend", "i4",
 279                        clib_net_to_host_u32 (td->resend_queue[i].seq));
 280     td->resend_queue[i].retry_count++;
 281     vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RETRY_COUNT],
 282                                    thread_index, 0, 1);
 283     if (vlib_buffer_alloc (vm, &bi, 1) != 1)
 284       {
 285         nat_elog_warn ("HA NAT state sync can't allocate buffer");
 286         return;
 287       }
 288     b = vlib_get_buffer (vm, bi);
 289     b->current_length = vec_len (td->resend_queue[i].data);
 290     b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 291     b->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
 292     vnet_buffer (b)->sw_if_index[VLIB_RX] = 0;
 293     vnet_buffer (b)->sw_if_index[VLIB_TX] = 0;
 294     ip = vlib_buffer_get_current (b);
 295     clib_memcpy (ip, td->resend_queue[i].data,
 296                  vec_len (td->resend_queue[i].data));
 297     f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
 298     to_next = vlib_frame_vector_args (f);
 299     to_next[0] = bi;
 300     f->n_vectors = 1;
 301     vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
 302     td->resend_queue[i].retry_timer = now + 2.0;
 303   }
 304
 305   vec_foreach (del, to_delete)
 306   {
 307     vec_free (td->resend_queue[*del].data);
 308     vec_del1 (td->resend_queue, *del);
 309   }
 310   vec_free (to_delete);
 311 }
 312
 313 void
 314 nat_ha_enable (nat_ha_sadd_cb_t sadd_cb,
 315                nat_ha_sdel_cb_t sdel_cb, nat_ha_sref_cb_t sref_cb)
 316 {
 317   nat_ha_main_t *ha = &nat_ha_main;
 318
 319   ha->sadd_cb = sadd_cb;
 320   ha->sdel_cb = sdel_cb;
 321   ha->sref_cb = sref_cb;
 322
 323   ha->enabled = 1;
 324 }
 325
 326 void
 327 nat_ha_disable ()
 328 {
 329   nat_ha_main_t *ha = &nat_ha_main;
 330   ha->dst_port = 0;
 331   ha->enabled = 0;
 332 }
 333
 334 void
 335 nat_ha_init (vlib_main_t * vm, u32 num_workers, u32 num_threads)
 336 {
 337   nat_ha_main_t *ha = &nat_ha_main;
 338   clib_memset (ha, 0, sizeof (*ha));
 339
 340   ha->vlib_main = vm;
 341   ha->fq_index = ~0;
 342
 343   ha->num_workers = num_workers;
 344   vec_validate (ha->per_thread_data, num_threads);
 345
 346 #define _(N, s, v) ha->counters[v].name = s;          \
 347   ha->counters[v].stat_segment_name = "/nat44/ha/" s; \
 348   vlib_validate_simple_counter(&ha->counters[v], 0);  \
 349   vlib_zero_simple_counter(&ha->counters[v], 0);
 350   foreach_nat_ha_counter
 351 #undef _
 352 }
 353
 354 int
 355 nat_ha_set_listener (ip4_address_t * addr, u16 port, u32 path_mtu)
 356 {
 357   nat_ha_main_t *ha = &nat_ha_main;
 358
 359   /* unregister previously set UDP port */
 360   if (ha->src_port)
 361     udp_unregister_dst_port (ha->vlib_main, ha->src_port, 1);
 362
 363   ha->src_ip_address.as_u32 = addr->as_u32;
 364   ha->src_port = port;
 365   ha->state_sync_path_mtu = path_mtu;
 366
 367   if (port)
 368     {
 369       /* if multiple worker threads first go to handoff node */
 370       if (ha->num_workers > 1)
 371         {
 372           if (ha->fq_index == ~0)
 373             ha->fq_index = vlib_frame_queue_main_init (nat_ha_node.index, 0);
 374           udp_register_dst_port (ha->vlib_main, port,
 375                                  nat_ha_handoff_node.index, 1);
 376         }
 377       else
 378         {
 379           udp_register_dst_port (ha->vlib_main, port, nat_ha_node.index, 1);
 380         }
 381       nat_elog_info_X1 ("HA listening on port %d for state sync", "i4", port);
 382     }
 383
 384   return 0;
 385 }
 386
 387 void
 388 nat_ha_get_listener (ip4_address_t * addr, u16 * port, u32 * path_mtu)
 389 {
 390   nat_ha_main_t *ha = &nat_ha_main;
 391
 392   addr->as_u32 = ha->src_ip_address.as_u32;
 393   *port = ha->src_port;
 394   *path_mtu = ha->state_sync_path_mtu;
 395 }
 396
 397 int
 398 nat_ha_set_failover (ip4_address_t * addr, u16 port,
 399                      u32 session_refresh_interval)
 400 {
 401   nat_ha_main_t *ha = &nat_ha_main;
 402
 403   ha->dst_ip_address.as_u32 = addr->as_u32;
 404   ha->dst_port = port;
 405   ha->session_refresh_interval = session_refresh_interval;
 406
 407   vlib_process_signal_event (ha->vlib_main, nat_ha_process_node.index, 1, 0);
 408
 409   return 0;
 410 }
 411
 412 void
 413 nat_ha_get_failover (ip4_address_t * addr, u16 * port,
 414                      u32 * session_refresh_interval)
 415 {
 416   nat_ha_main_t *ha = &nat_ha_main;
 417
 418   addr->as_u32 = ha->dst_ip_address.as_u32;
 419   *port = ha->dst_port;
 420   *session_refresh_interval = ha->session_refresh_interval;
 421 }
 422
 423 static_always_inline void
 424 nat_ha_recv_add (nat_ha_event_t * event, f64 now, u32 thread_index)
 425 {
 426   nat_ha_main_t *ha = &nat_ha_main;
 427   ip4_address_t in_addr, out_addr, eh_addr, ehn_addr;
 428   u32 fib_index;
 429   u16 flags;
 430
 431   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_ADD],
 432                                  thread_index, 0, 1);
 433
 434   in_addr.as_u32 = event->in_addr;
 435   out_addr.as_u32 = event->out_addr;
 436   eh_addr.as_u32 = event->eh_addr;
 437   ehn_addr.as_u32 = event->ehn_addr;
 438   fib_index = clib_net_to_host_u32 (event->fib_index);
 439   flags = clib_net_to_host_u16 (event->flags);
 440
 441   ha->sadd_cb (&in_addr, event->in_port, &out_addr, event->out_port, &eh_addr,
 442                event->eh_port, &ehn_addr, event->ehn_port, event->protocol,
 443                fib_index, flags, thread_index);
 444 }
 445
 446 static_always_inline void
 447 nat_ha_recv_del (nat_ha_event_t * event, u32 thread_index)
 448 {
 449   nat_ha_main_t *ha = &nat_ha_main;
 450   ip4_address_t out_addr, eh_addr;
 451   u32 fib_index;
 452
 453   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_DEL],
 454                                  thread_index, 0, 1);
 455
 456   out_addr.as_u32 = event->out_addr;
 457   eh_addr.as_u32 = event->eh_addr;
 458   fib_index = clib_net_to_host_u32 (event->fib_index);
 459
 460   ha->sdel_cb (&out_addr, event->out_port, &eh_addr, event->eh_port,
 461                event->protocol, fib_index, thread_index);
 462 }
 463
 464 static_always_inline void
 465 nat_ha_recv_refresh (nat_ha_event_t * event, f64 now, u32 thread_index)
 466 {
 467   nat_ha_main_t *ha = &nat_ha_main;
 468   ip4_address_t out_addr, eh_addr;
 469   u32 fib_index, total_pkts;
 470   u64 total_bytes;
 471
 472   vlib_increment_simple_counter (&ha->counters[NAT_HA_COUNTER_RECV_REFRESH],
 473                                  thread_index, 0, 1);
 474
 475   out_addr.as_u32 = event->out_addr;
 476   eh_addr.as_u32 = event->eh_addr;
 477   fib_index = clib_net_to_host_u32 (event->fib_index);
 478   total_pkts = clib_net_to_host_u32 (event->total_pkts);
 479   total_bytes = clib_net_to_host_u64 (event->total_bytes);
 480
 481   ha->sref_cb (&out_addr, event->out_port, &eh_addr, event->eh_port,
 482                event->protocol, fib_index, total_pkts, total_bytes,
 483                thread_index);
 484 }
 485
 486 /* process received NAT HA event */
 487 static_always_inline void
 488 nat_ha_event_process (nat_ha_event_t * event, f64 now, u32 thread_index)
 489 {
 490   switch (event->event_type)
 491     {
 492     case NAT_HA_ADD:
 493       nat_ha_recv_add (event, now, thread_index);
 494       break;
 495     case NAT_HA_DEL:
 496       nat_ha_recv_del (event, thread_index);
 497       break;
 498     case NAT_HA_REFRESH:
 499       nat_ha_recv_refresh (event, now, thread_index);
 500       break;
 501     default:
 502       nat_elog_notice_X1 ("Unsupported HA event type %d", "i4",
 503                           event->event_type);
 504       break;
 505     }
 506 }
 507
 508 static inline void
 509 nat_ha_header_create (vlib_buffer_t * b, u32 * offset, u32 thread_index)
 510 {
 511   nat_ha_main_t *ha = &nat_ha_main;
 512   nat_ha_message_header_t *h;
 513   ip4_header_t *ip;
 514   udp_header_t *udp;
 515   u32 sequence_number;
 516
 517   b->current_data = 0;
 518   b->current_length = sizeof (*ip) + sizeof (*udp) + sizeof (*h);
 519   b->flags |= VLIB_BUFFER_TOTAL_LENGTH_VALID;
 520   b->flags |= VNET_BUFFER_F_LOCALLY_ORIGINATED;
 521   vnet_buffer (b)->sw_if_index[VLIB_RX] = 0;
 522   vnet_buffer (b)->sw_if_index[VLIB_TX] = 0;
 523   ip = vlib_buffer_get_current (b);
 524   udp = (udp_header_t *) (ip + 1);
 525   h = (nat_ha_message_header_t *) (udp + 1);
 526
 527   /* IP header */
 528   ip->ip_version_and_header_length = 0x45;
 529   ip->ttl = 254;
 530   ip->protocol = IP_PROTOCOL_UDP;
 531   ip->flags_and_fragment_offset =
 532     clib_host_to_net_u16 (IP4_HEADER_FLAG_DONT_FRAGMENT);
 533   ip->src_address.as_u32 = ha->src_ip_address.as_u32;
 534   ip->dst_address.as_u32 = ha->dst_ip_address.as_u32;
 535   /* UDP header */
 536   udp->src_port = clib_host_to_net_u16 (ha->src_port);
 537   udp->dst_port = clib_host_to_net_u16 (ha->dst_port);
 538   udp->checksum = 0;
 539
 540   /* NAT HA protocol header */
 541   h->version = NAT_HA_VERSION;
 542   h->flags = 0;
 543   h->count = 0;
 544   h->thread_index = clib_host_to_net_u32 (thread_index);
 545   sequence_number = clib_atomic_fetch_add (&ha->sequence_number, 1);
 546   h->sequence_number = clib_host_to_net_u32 (sequence_number);
 547
 548   *offset =
 549     sizeof (ip4_header_t) + sizeof (udp_header_t) +
 550     sizeof (nat_ha_message_header_t);
 551 }
 552
 553 static inline void
 554 nat_ha_send (vlib_frame_t * f, vlib_buffer_t * b, u8 is_resync,
 555              u32 thread_index)
 556 {
 557   nat_ha_main_t *ha = &nat_ha_main;
 558   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 559   nat_ha_message_header_t *h;
 560   ip4_header_t *ip;
 561   udp_header_t *udp;
 562   vlib_main_t *vm = vlib_mains[thread_index];
 563
 564   ip = vlib_buffer_get_current (b);
 565   udp = ip4_next_header (ip);
 566   h = (nat_ha_message_header_t *) (udp + 1);
 567
 568   h->count = clib_host_to_net_u16 (td->state_sync_count);
 569
 570   ip->length = clib_host_to_net_u16 (b->current_length);
 571   ip->checksum = ip4_header_checksum (ip);
 572   udp->length = clib_host_to_net_u16 (b->current_length - sizeof (*ip));
 573
 574   nat_ha_resend_queue_add (h->sequence_number, (u8 *) ip, b->current_length,
 575                            is_resync, thread_index);
 576
 577   vlib_put_frame_to_node (vm, ip4_lookup_node.index, f);
 578 }
 579
 580 /* add NAT HA protocol event */
 581 static_always_inline void
 582 nat_ha_event_add (nat_ha_event_t * event, u8 do_flush, u32 thread_index,
 583                   u8 is_resync)
 584 {
 585   nat_ha_main_t *ha = &nat_ha_main;
 586   nat_ha_per_thread_data_t *td = &ha->per_thread_data[thread_index];
 587   vlib_main_t *vm = vlib_mains[thread_index];
 588   vlib_buffer_t *b = 0;
 589   vlib_frame_t *f;
 590   u32 bi = ~0, offset;
 591
 592   b = td->state_sync_buffer;
 593
 594   if (PREDICT_FALSE (b == 0))
 595     {
 596       if (do_flush)
 597         return;
 598
 599       if (vlib_buffer_alloc (vm, &bi, 1) != 1)
 600         {
 601           nat_elog_warn ("HA NAT state sync can't allocate buffer");
 602           return;
 603         }
 604
 605       b = td->state_sync_buffer = vlib_get_buffer (vm, bi);
 606       clib_memset (vnet_buffer (b), 0, sizeof (*vnet_buffer (b)));
 607       VLIB_BUFFER_TRACE_TRAJECTORY_INIT (b);
 608       offset = 0;
 609     }
 610   else
 611     {
 612       bi = vlib_get_buffer_index (vm, b);
 613       offset = td->state_sync_next_event_offset;
 614     }
 615
 616   f = td->state_sync_frame;
 617   if (PREDICT_FALSE (f == 0))
 618     {
 619       u32 *to_next;
 620       f = vlib_get_frame_to_node (vm, ip4_lookup_node.index);
 621       td->state_sync_frame = f;
 622       to_next = vlib_frame_vector_args (f);
 623       to_next[0] = bi;
 624       f->n_vectors = 1;
 625     }
 626
 627   if (PREDICT_FALSE (td->state_sync_count == 0))
 628     nat_ha_header_create (b, &offset, thread_index);
 629
 630   if (PREDICT_TRUE (do_flush == 0))
 631     {
 632       clib_memcpy_fast (b->data + offset, event, sizeof (*event));
 633       offset += sizeof (*event);
 634       td->state_sync_count++;
 635       b->current_length += sizeof (*event);
 636
 637       switch (event->event_type)
 638         {
 639         case NAT_HA_ADD:
 640           vlib_increment_simple_counter (&ha->counters
 641                                          [NAT_HA_COUNTER_SEND_ADD],
 642                                          thread_index, 0, 1);
 643           break;
 644         case NAT_HA_DEL:
 645           vlib_increment_simple_counter (&ha->counters
 646                                          [NAT_HA_COUNTER_SEND_DEL],
 647                                          thread_index, 0, 1);
 648           break;
 649         case NAT_HA_REFRESH:
 650           vlib_increment_simple_counter (&ha->counters
 651                                          [NAT_HA_COUNTER_SEND_REFRESH],
 652                                          thread_index, 0, 1);
 653           break;
 654         default:
 655           break;
 656         }
 657     }
 658
 659   if (PREDICT_FALSE
 660       (do_flush || offset + (sizeof (*event)) > ha->state_sync_path_mtu))
 661     {
 662       nat_ha_send (f, b, is_resync, thread_index);
 663       td->state_sync_buffer = 0;
 664       td->state_sync_frame = 0;
 665       td->state_sync_count = 0;
 666       offset = 0;
 667       if (is_resync)
 668         {
 669           clib_atomic_fetch_add (&ha->resync_ack_count, 1);
 670           nat_ha_resync_fin ();
 671         }
 672     }
 673
 674   td->state_sync_next_event_offset = offset;
 675 }
 676
 677 #define skip_if_disabled()          \
 678 do {                                \
 679   nat_ha_main_t *ha = &nat_ha_main; \
 680   if (PREDICT_TRUE (!ha->dst_port)) \
 681     return;                         \
 682 } while (0)
 683
 684 void
 685 nat_ha_flush (u8 is_resync)
 686 {
 687   skip_if_disabled ();
 688   nat_ha_event_add (0, 1, 0, is_resync);
 689 }
 690
 691 void
 692 nat_ha_sadd (ip4_address_t * in_addr, u16 in_port, ip4_address_t * out_addr,
 693              u16 out_port, ip4_address_t * eh_addr, u16 eh_port,
 694              ip4_address_t * ehn_addr, u16 ehn_port, u8 proto, u32 fib_index,
 695              u16 flags, u32 thread_index, u8 is_resync)
 696 {
 697   nat_ha_event_t event;
 698
 699   skip_if_disabled ();
 700
 701   clib_memset (&event, 0, sizeof (event));
 702   event.event_type = NAT_HA_ADD;
 703   event.flags = clib_host_to_net_u16 (flags);
 704   event.in_addr = in_addr->as_u32;
 705   event.in_port = in_port;
 706   event.out_addr = out_addr->as_u32;
 707   event.out_port = out_port;
 708   event.eh_addr = eh_addr->as_u32;
 709   event.eh_port = eh_port;
 710   event.ehn_addr = ehn_addr->as_u32;
 711   event.ehn_port = ehn_port;
 712   event.fib_index = clib_host_to_net_u32 (fib_index);
 713   event.protocol = proto;
 714   nat_ha_event_add (&event, 0, thread_index, is_resync);
 715 }
 716
 717 void
 718 nat_ha_sdel (ip4_address_t * out_addr, u16 out_port, ip4_address_t * eh_addr,
 719              u16 eh_port, u8 proto, u32 fib_index, u32 thread_index)
 720 {
 721   nat_ha_event_t event;
 722
 723   skip_if_disabled ();
 724
 725   clib_memset (&event, 0, sizeof (event));
 726   event.event_type = NAT_HA_DEL;
 727   event.out_addr = out_addr->as_u32;
 728   event.out_port = out_port;
 729   event.eh_addr = eh_addr->as_u32;
 730   event.eh_port = eh_port;
 731   event.fib_index = clib_host_to_net_u32 (fib_index);
 732   event.protocol = proto;
 733   nat_ha_event_add (&event, 0, thread_index, 0);
 734 }
 735
 736 void
 737 nat_ha_sref (ip4_address_t * out_addr, u16 out_port, ip4_address_t * eh_addr,
 738              u16 eh_port, u8 proto, u32 fib_index, u32 total_pkts,
 739              u64 total_bytes, u32 thread_index, f64 * last_refreshed, f64 now)
 740 {
 741   nat_ha_main_t *ha = &nat_ha_main;
 742   nat_ha_event_t event;
 743
 744   skip_if_disabled ();
 745
 746   if ((*last_refreshed + ha->session_refresh_interval) > now)
 747     return;
 748
 749   *last_refreshed = now;
 750   clib_memset (&event, 0, sizeof (event));
 751   event.event_type = NAT_HA_REFRESH;
 752   event.out_addr = out_addr->as_u32;
 753   event.out_port = out_port;
 754   event.eh_addr = eh_addr->as_u32;
 755   event.eh_port = eh_port;
 756   event.fib_index = clib_host_to_net_u32 (fib_index);
 757   event.protocol = proto;
 758   event.total_pkts = clib_host_to_net_u32 (total_pkts);
 759   event.total_bytes = clib_host_to_net_u64 (total_bytes);
 760   nat_ha_event_add (&event, 0, thread_index, 0);
 761 }
 762
 763 static_always_inline u8
 764 plugin_enabled ()
 765 {
 766   nat_ha_main_t *ha = &nat_ha_main;
 767   return ha->enabled;
 768 }
 769
 770 /* per thread process waiting for interrupt */
 771 static uword
 772 nat_ha_worker_fn (vlib_main_t * vm, vlib_node_runtime_t * rt,
 773                   vlib_frame_t * f)
 774 {
 775   u32 thread_index = vm->thread_index;
 776
 777   if (plugin_enabled () == 0)
 778     return 0;
 779
 780   /* flush HA NAT data under construction */
 781   nat_ha_event_add (0, 1, thread_index, 0);
 782   /* scan if we need to resend some non-ACKed data */
 783   nat_ha_resend_scan (vlib_time_now (vm), thread_index);
 784   return 0;
 785 }
 786
 787 /* *INDENT-OFF* */
 788 VLIB_REGISTER_NODE (nat_ha_worker_node) = {
 789     .function = nat_ha_worker_fn,
 790     .type = VLIB_NODE_TYPE_INPUT,
 791     .state = VLIB_NODE_STATE_INTERRUPT,
 792     .name = "nat-ha-worker",
 793 };
 794 /* *INDENT-ON* */
 795
 796 /* periodically send interrupt to each thread */
 797 static uword
 798 nat_ha_process (vlib_main_t * vm, vlib_node_runtime_t * rt, vlib_frame_t * f)
 799 {
 800   nat_ha_main_t *ha = &nat_ha_main;
 801   uword event_type;
 802   uword *event_data = 0;
 803   u32 ti;
 804
 805   vlib_process_wait_for_event (vm);
 806   event_type = vlib_process_get_events (vm, &event_data);
 807   if (event_type)
 808     nat_elog_info ("nat-ha-process: bogus kickoff event received");
 809   vec_reset_length (event_data);
 810
 811   while (1)
 812     {
 813       vlib_process_wait_for_event_or_clock (vm, 1.0);
 814       event_type = vlib_process_get_events (vm, &event_data);
 815       vec_reset_length (event_data);
 816       for (ti = 0; ti < vec_len (vlib_mains); ti++)
 817         {
 818           if (ti >= vec_len (ha->per_thread_data))
 819             continue;
 820
 821           vlib_node_set_interrupt_pending (vlib_mains[ti],
 822                                            nat_ha_worker_node.index);
 823         }
 824     }
 825
 826   return 0;
 827 }
 828
 829 /* *INDENT-OFF* */
 830 VLIB_REGISTER_NODE (nat_ha_process_node) = {
 831     .function = nat_ha_process,
 832     .type = VLIB_NODE_TYPE_PROCESS,
 833     .name = "nat-ha-process",
 834 };
 835 /* *INDENT-ON* */
 836
 837 void
 838 nat_ha_get_resync_status (u8 * in_resync, u32 * resync_ack_missed)
 839 {
 840   nat_ha_main_t *ha = &nat_ha_main;
 841
 842   *in_resync = ha->in_resync;
 843   *resync_ack_missed = ha->resync_ack_missed;
 844 }
 845
 846 typedef struct
 847 {
 848   ip4_address_t addr;
 849   u32 event_count;
 850 } nat_ha_trace_t;
 851
 852 static u8 *
 853 format_nat_ha_trace (u8 * s, va_list * args)
 854 {
 855   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
 856   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
 857   nat_ha_trace_t *t = va_arg (*args, nat_ha_trace_t *);
 858
 859   s =
 860     format (s, "nat-ha: %u events from %U", t->event_count,
 861             format_ip4_address, &t->addr);
 862
 863   return s;
 864 }
 865
 866 typedef enum
 867 {
 868   NAT_HA_NEXT_IP4_LOOKUP,
 869   NAT_HA_NEXT_DROP,
 870   NAT_HA_N_NEXT,
 871 } nat_ha_next_t;
 872
 873 #define foreach_nat_ha_error   \
 874 _(PROCESSED, "pkts-processed") \
 875 _(BAD_VERSION, "bad-version")
 876
 877 typedef enum
 878 {
 879 #define _(sym, str) NAT_HA_ERROR_##sym,
 880   foreach_nat_ha_error
 881 #undef _
 882     NAT_HA_N_ERROR,
 883 } nat_ha_error_t;
 884
 885 static char *nat_ha_error_strings[] = {
 886 #define _(sym, str) str,
 887   foreach_nat_ha_error
 888 #undef _
 889 };
 890
 891 /* process received HA NAT protocol messages */
 892 static uword
 893 nat_ha_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
 894                 vlib_frame_t * frame)
 895 {
 896   u32 n_left_from, *from, next_index, *to_next;
 897   f64 now = vlib_time_now (vm);
 898   u32 thread_index = vm->thread_index;
 899   u32 pkts_processed = 0;
 900   ip4_main_t *i4m = &ip4_main;
 901   u8 host_config_ttl = i4m->host_config.ttl;
 902   nat_ha_main_t *ha = &nat_ha_main;
 903
 904   from = vlib_frame_vector_args (frame);
 905   n_left_from = frame->n_vectors;
 906   next_index = node->cached_next_index;
 907
 908   while (n_left_from > 0)
 909     {
 910       u32 n_left_to_next;
 911
 912       vlib_get_next_frame (vm, node, next_index, to_next, n_left_to_next);
 913
 914       while (n_left_from > 0 && n_left_to_next > 0)
 915         {
 916           u32 bi0, next0, src_addr0, dst_addr0;;
 917           vlib_buffer_t *b0;
 918           nat_ha_message_header_t *h0;
 919           nat_ha_event_t *e0;
 920           u16 event_count0, src_port0, dst_port0, old_len0;
 921           ip4_header_t *ip0;
 922           udp_header_t *udp0;
 923           ip_csum_t sum0;
 924
 925           bi0 = from[0];
 926           to_next[0] = bi0;
 927           from += 1;
 928           to_next += 1;
 929           n_left_from -= 1;
 930           n_left_to_next -= 1;
 931
 932           b0 = vlib_get_buffer (vm, bi0);
 933           h0 = vlib_buffer_get_current (b0);
 934           vlib_buffer_advance (b0, -sizeof (*udp0));
 935           udp0 = vlib_buffer_get_current (b0);
 936           vlib_buffer_advance (b0, -sizeof (*ip0));
 937           ip0 = vlib_buffer_get_current (b0);
 938
 939           next0 = NAT_HA_NEXT_DROP;
 940
 941           if (h0->version != NAT_HA_VERSION)
 942             {
 943               b0->error = node->errors[NAT_HA_ERROR_BAD_VERSION];
 944               goto done0;
 945             }
 946
 947           event_count0 = clib_net_to_host_u16 (h0->count);
 948           /* ACK for previously send data */
 949           if (!event_count0 && (h0->flags & NAT_HA_FLAG_ACK))
 950             {
 951               nat_ha_ack_recv (h0->sequence_number, thread_index);
 952               b0->error = node->errors[NAT_HA_ERROR_PROCESSED];
 953               goto done0;
 954             }
 955
 956           e0 = (nat_ha_event_t *) (h0 + 1);
 957
 958           /* process each event */
 959           while (event_count0)
 960             {
 961               nat_ha_event_process (e0, now, thread_index);
 962               event_count0--;
 963               e0 = (nat_ha_event_t *) ((u8 *) e0 + sizeof (nat_ha_event_t));
 964             }
 965
 966           next0 = NAT_HA_NEXT_IP4_LOOKUP;
 967           pkts_processed++;
 968
 969           /* reply with ACK */
 970           b0->current_length = sizeof (*ip0) + sizeof (*udp0) + sizeof (*h0);
 971
 972           src_addr0 = ip0->src_address.data_u32;
 973           dst_addr0 = ip0->dst_address.data_u32;
 974           ip0->src_address.data_u32 = dst_addr0;
 975           ip0->dst_address.data_u32 = src_addr0;
 976           old_len0 = ip0->length;
 977           ip0->length = clib_host_to_net_u16 (b0->current_length);
 978
 979           sum0 = ip0->checksum;
 980           sum0 = ip_csum_update (sum0, ip0->ttl, host_config_ttl,
 981                                  ip4_header_t, ttl);
 982           ip0->ttl = host_config_ttl;
 983           sum0 =
 984             ip_csum_update (sum0, old_len0, ip0->length, ip4_header_t,
 985                             length);
 986           ip0->checksum = ip_csum_fold (sum0);
 987
 988           udp0->checksum = 0;
 989           src_port0 = udp0->src_port;
 990           dst_port0 = udp0->dst_port;
 991           udp0->src_port = dst_port0;
 992           udp0->dst_port = src_port0;
 993           udp0->length =
 994             clib_host_to_net_u16 (b0->current_length - sizeof (*ip0));
 995
 996           h0->flags = NAT_HA_FLAG_ACK;
 997           h0->count = 0;
 998           vlib_increment_simple_counter (&ha->counters
 999                                          [NAT_HA_COUNTER_SEND_ACK],
1000                                          thread_index, 0, 1);
1001
1002         done0:
1003           if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE)
1004                              && (b0->flags & VLIB_BUFFER_IS_TRACED)))
1005             {
1006               nat_ha_trace_t *t = vlib_add_trace (vm, node, b0, sizeof (*t));
1007               ip4_header_t *ip =
1008                 (void *) (b0->data + vnet_buffer (b0)->l3_hdr_offset);
1009               t->event_count = clib_net_to_host_u16 (h0->count);
1010               t->addr.as_u32 = ip->src_address.data_u32;
1011             }
1012
1013           vlib_validate_buffer_enqueue_x1 (vm, node, next_index,
1014                                            to_next, n_left_to_next,
1015                                            bi0, next0);
1016         }
1017
1018       vlib_put_next_frame (vm, node, next_index, n_left_to_next);
1019     }
1020
1021   vlib_node_increment_counter (vm, nat_ha_node.index,
1022                                NAT_HA_ERROR_PROCESSED, pkts_processed);
1023
1024   return frame->n_vectors;
1025 }
1026
1027 /* *INDENT-OFF* */
1028 VLIB_REGISTER_NODE (nat_ha_node) = {
1029   .function = nat_ha_node_fn,
1030   .name = "nat-ha",
1031   .vector_size = sizeof (u32),
1032   .format_trace = format_nat_ha_trace,
1033   .type = VLIB_NODE_TYPE_INTERNAL,
1034   .n_errors = ARRAY_LEN (nat_ha_error_strings),
1035   .error_strings = nat_ha_error_strings,
1036   .n_next_nodes = NAT_HA_N_NEXT,
1037   .next_nodes = {
1038      [NAT_HA_NEXT_IP4_LOOKUP] = "ip4-lookup",
1039      [NAT_HA_NEXT_DROP] = "error-drop",
1040   },
1041 };
1042 /* *INDENT-ON* */
1043
1044 typedef struct
1045 {
1046   u32 next_worker_index;
1047   u8 in2out;
1048 } nat_ha_handoff_trace_t;
1049
1050 #define foreach_nat_ha_handoff_error  \
1051 _(CONGESTION_DROP, "congestion drop") \
1052 _(SAME_WORKER, "same worker")         \
1053 _(DO_HANDOFF, "do handoff")
1054
1055 typedef enum
1056 {
1057 #define _(sym,str) NAT_HA_HANDOFF_ERROR_##sym,
1058   foreach_nat_ha_handoff_error
1059 #undef _
1060     NAT44_HANDOFF_N_ERROR,
1061 } nat_ha_handoff_error_t;
1062
1063 static char *nat_ha_handoff_error_strings[] = {
1064 #define _(sym,string) string,
1065   foreach_nat_ha_handoff_error
1066 #undef _
1067 };
1068
1069 static u8 *
1070 format_nat_ha_handoff_trace (u8 * s, va_list * args)
1071 {
1072   CLIB_UNUSED (vlib_main_t * vm) = va_arg (*args, vlib_main_t *);
1073   CLIB_UNUSED (vlib_node_t * node) = va_arg (*args, vlib_node_t *);
1074   nat_ha_handoff_trace_t *t = va_arg (*args, nat_ha_handoff_trace_t *);
1075
1076   s =
1077     format (s, "NAT_HA_WORKER_HANDOFF: next-worker %d", t->next_worker_index);
1078
1079   return s;
1080 }
1081
1082 /* do worker handoff based on thread_index in NAT HA protcol header */
1083 static uword
1084 nat_ha_handoff_node_fn (vlib_main_t * vm, vlib_node_runtime_t * node,
1085                         vlib_frame_t * frame)
1086 {
1087   nat_ha_main_t *ha = &nat_ha_main;
1088   vlib_buffer_t *bufs[VLIB_FRAME_SIZE], **b;
1089   u32 n_enq, n_left_from, *from;
1090   u16 thread_indices[VLIB_FRAME_SIZE], *ti;
1091   u32 thread_index = vm->thread_index;
1092   u32 do_handoff = 0, same_worker = 0;
1093
1094   from = vlib_frame_vector_args (frame);
1095   n_left_from = frame->n_vectors;
1096   vlib_get_buffers (vm, from, bufs, n_left_from);
1097
1098   b = bufs;
1099   ti = thread_indices;
1100
1101   while (n_left_from > 0)
1102     {
1103       nat_ha_message_header_t *h0;
1104
1105       h0 = vlib_buffer_get_current (b[0]);
1106       ti[0] = clib_net_to_host_u32 (h0->thread_index);
1107
1108       if (ti[0] != thread_index)
1109         do_handoff++;
1110       else
1111         same_worker++;
1112
1113       if (PREDICT_FALSE ((node->flags & VLIB_NODE_FLAG_TRACE)
1114                          && (b[0]->flags & VLIB_BUFFER_IS_TRACED)))
1115         {
1116           nat_ha_handoff_trace_t *t =
1117             vlib_add_trace (vm, node, b[0], sizeof (*t));
1118           t->next_worker_index = ti[0];
1119         }
1120
1121       n_left_from -= 1;
1122       ti += 1;
1123       b += 1;
1124     }
1125
1126   n_enq =
1127     vlib_buffer_enqueue_to_thread (vm, ha->fq_index, from, thread_indices,
1128                                    frame->n_vectors, 1);
1129
1130   if (n_enq < frame->n_vectors)
1131     vlib_node_increment_counter (vm, node->node_index,
1132                                  NAT_HA_HANDOFF_ERROR_CONGESTION_DROP,
1133                                  frame->n_vectors - n_enq);
1134   vlib_node_increment_counter (vm, node->node_index,
1135                                NAT_HA_HANDOFF_ERROR_SAME_WORKER, same_worker);
1136   vlib_node_increment_counter (vm, node->node_index,
1137                                NAT_HA_HANDOFF_ERROR_DO_HANDOFF, do_handoff);
1138   return frame->n_vectors;
1139 }
1140
1141 int
1142 nat_ha_resync (u32 client_index, u32 pid,
1143                nat_ha_resync_event_cb_t event_callback)
1144 {
1145   return 0;
1146 }
1147
1148 /* *INDENT-OFF* */
1149 VLIB_REGISTER_NODE (nat_ha_handoff_node) = {
1150   .function = nat_ha_handoff_node_fn,
1151   .name = "nat-ha-handoff",
1152   .vector_size = sizeof (u32),
1153   .format_trace = format_nat_ha_handoff_trace,
1154   .type = VLIB_NODE_TYPE_INTERNAL,
1155   .n_errors = ARRAY_LEN(nat_ha_handoff_error_strings),
1156   .error_strings = nat_ha_handoff_error_strings,
1157   .n_next_nodes = 1,
1158   .next_nodes = {
1159     [0] = "error-drop",
1160   },
1161 };
1162 /* *INDENT-ON* */
1163
1164 /*
1165  * fd.io coding-style-patch-verification: ON
1166  *
1167  * Local Variables:
1168  * eval: (c-set-style "gnu")
1169  * End:
1170  */