src/wccp.cc

   1 /*
   2  * Copyright (C) 1996-2023 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 80    WCCP Support */
  10
  11 #include "squid.h"
  12
  13 #if USE_WCCP
  14 #include "comm.h"
  15 #include "comm/Connection.h"
  16 #include "comm/Loops.h"
  17 #include "event.h"
  18 #include "fatal.h"
  19 #include "SquidConfig.h"
  20 #include "wccp.h"
  21
  22 #define WCCP_PORT 2048
  23 #define WCCP_REVISION 0
  24 #define WCCP_ACTIVE_CACHES 32
  25 #define WCCP_HASH_SIZE 32
  26 #define WCCP_BUCKETS 256
  27 #define WCCP_CACHE_LEN 4
  28
  29 #define WCCP_HERE_I_AM 7
  30 #define WCCP_I_SEE_YOU 8
  31 #define WCCP_ASSIGN_BUCKET 9
  32
  33 struct wccp_here_i_am_t {
  34     int type;
  35     int version;
  36     int revision;
  37     char hash[WCCP_HASH_SIZE];
  38     int reserved;
  39     int id;
  40 };
  41
  42 struct wccp_cache_entry_t {
  43     struct in_addr ip_addr;  // WCCP on-the-wire in 32-bit IPv4-only.
  44     int revision;
  45     char hash[WCCP_HASH_SIZE];
  46     int reserved;
  47 };
  48
  49 struct wccp_i_see_you_t {
  50     int32_t type;
  51     int32_t version;
  52     int32_t change;
  53     int32_t id;
  54     int32_t number;
  55
  56     struct wccp_cache_entry_t wccp_cache_entry[WCCP_ACTIVE_CACHES];
  57 };
  58
  59 struct wccp_assign_bucket_t {
  60     int type;
  61     int id;
  62     int number;
  63 };
  64
  65 static int theWccpConnection = -1;
  66
  67 static struct wccp_here_i_am_t wccp_here_i_am;
  68
  69 static struct wccp_i_see_you_t wccp_i_see_you;
  70 static int last_change;
  71 static int last_id;
  72 static int last_assign_buckets_change;
  73 static unsigned int number_caches;
  74
  75 static Ip::Address local_ip;
  76
  77 static PF wccpHandleUdp;
  78 static int wccpLowestIP(void);
  79 static EVH wccpHereIam;
  80 static void wccpAssignBuckets(void);
  81
  82 /*
  83  * The functions used during startup:
  84  * wccpInit
  85  * wccpConnectionOpen
  86  * wccpConnectionShutdown
  87  * wccpConnectionClose
  88  */
  89
  90 void
  91 wccpInit(void)
  92 {
  93     debugs(80, 5, "wccpInit: Called");
  94     memset(&wccp_here_i_am, '\0', sizeof(wccp_here_i_am));
  95     wccp_here_i_am.type = htonl(WCCP_HERE_I_AM);
  96     wccp_here_i_am.version = htonl(Config.Wccp.version);
  97     wccp_here_i_am.revision = htonl(WCCP_REVISION);
  98     last_change = 0;
  99     last_id = 0;
 100     last_assign_buckets_change = 0;
 101     number_caches = 0;
 102
 103     if (!Config.Wccp.router.isAnyAddr())
 104         if (!eventFind(wccpHereIam, nullptr))
 105             eventAdd("wccpHereIam", wccpHereIam, nullptr, 5.0, 1);
 106 }
 107
 108 void
 109 wccpConnectionOpen(void)
 110 {
 111     debugs(80, 5, "wccpConnectionOpen: Called");
 112
 113     if (Config.Wccp.router.isAnyAddr()) {
 114         debugs(80, 2, "WCCPv1 disabled.");
 115         return;
 116     }
 117
 118     if ( !Config.Wccp.router.setIPv4() ) {
 119         debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Router " << Config.Wccp.router << " is not an IPv4 address.");
 120         return;
 121     }
 122
 123     if ( !Config.Wccp.address.setIPv4() ) {
 124         debugs(80, DBG_CRITICAL, "WCCPv1 Disabled. Local address " << Config.Wccp.address << " is not an IPv4 address.");
 125         return;
 126     }
 127
 128     Config.Wccp.address.port(WCCP_PORT);
 129     Config.Wccp.router.port(WCCP_PORT);
 130
 131     theWccpConnection = comm_open_listener(SOCK_DGRAM,
 132                                            IPPROTO_UDP,
 133                                            Config.Wccp.address,
 134                                            COMM_NONBLOCKING,
 135                                            "WCCP Socket");
 136
 137     if (theWccpConnection < 0)
 138         fatal("Cannot open WCCP Port");
 139
 140     Comm::SetSelect(theWccpConnection, COMM_SELECT_READ, wccpHandleUdp, nullptr, 0);
 141
 142     debugs(80, DBG_IMPORTANT, "Accepting WCCPv1 messages on " << Config.Wccp.address << ", FD " << theWccpConnection << ".");
 143
 144     // Sadly WCCP only does IPv4
 145
 146     struct sockaddr_in router;
 147     Config.Wccp.router.getSockAddr(router);
 148     if (connect(theWccpConnection, (struct sockaddr*)&router, sizeof(router)))
 149         fatal("Unable to connect WCCP out socket");
 150
 151     struct sockaddr_in local;
 152     memset(&local, '\0', sizeof(local));
 153     socklen_t slen = sizeof(local);
 154     if (getsockname(theWccpConnection, (struct sockaddr*)&local, &slen))
 155         fatal("Unable to getsockname on WCCP out socket");
 156
 157     local_ip = local;
 158 }
 159
 160 void
 161 wccpConnectionClose(void)
 162 {
 163     if (theWccpConnection > -1) {
 164         debugs(80, DBG_IMPORTANT, "FD " << theWccpConnection << " Closing WCCPv1 socket");
 165         comm_close(theWccpConnection);
 166         theWccpConnection = -1;
 167     }
 168 }
 169
 170 /*
 171  * Functions for handling the requests.
 172  */
 173
 174 /*
 175  * Accept the UDP packet
 176  */
 177 static void
 178 wccpHandleUdp(int sock, void *)
 179 {
 180     Ip::Address from;
 181     int len;
 182
 183     debugs(80, 6, "wccpHandleUdp: Called.");
 184
 185     Comm::SetSelect(sock, COMM_SELECT_READ, wccpHandleUdp, nullptr, 0);
 186
 187     memset(&wccp_i_see_you, '\0', sizeof(wccp_i_see_you));
 188
 189     len = comm_udp_recvfrom(sock,
 190                             (void *) &wccp_i_see_you,
 191                             sizeof(wccp_i_see_you),
 192                             0,
 193                             from);
 194     debugs(80, 3, "wccpHandleUdp: " << len << " bytes WCCP pkt from " << from <<
 195            ": type=" <<
 196            (unsigned) ntohl(wccp_i_see_you.type) << ", version=" <<
 197            (unsigned) ntohl(wccp_i_see_you.version) << ", change=" <<
 198            (unsigned) ntohl(wccp_i_see_you.change) << ", id=" <<
 199            (unsigned) ntohl(wccp_i_see_you.id) << ", number=" <<
 200            (unsigned) ntohl(wccp_i_see_you.number));
 201
 202     if (len < 0)
 203         return;
 204
 205     if (from != Config.Wccp.router)
 206         return;
 207
 208     if ((unsigned) ntohl(wccp_i_see_you.version) != (unsigned) Config.Wccp.version)
 209         return;
 210
 211     if (ntohl(wccp_i_see_you.type) != WCCP_I_SEE_YOU)
 212         return;
 213
 214     if (ntohl(wccp_i_see_you.number) > WCCP_ACTIVE_CACHES) {
 215         debugs(80, DBG_IMPORTANT, "Ignoring WCCP_I_SEE_YOU from " <<
 216                from << " with number of caches set to " <<
 217                (int) ntohl(wccp_i_see_you.number));
 218
 219         return;
 220     }
 221
 222     last_id = wccp_i_see_you.id;
 223
 224     if ((0 == last_change) && (number_caches == (unsigned) ntohl(wccp_i_see_you.number))) {
 225         if (last_assign_buckets_change == wccp_i_see_you.change) {
 226             /*
 227              * After a WCCP_ASSIGN_BUCKET message, the router should
 228              * update the change value.  If not, maybe the route didn't
 229              * receive our WCCP_ASSIGN_BUCKET message, so send it again.
 230              *
 231              * Don't update change here.  Instead, fall through to
 232              * the next block to call wccpAssignBuckets() again.
 233              */
 234             (void) 0;
 235         } else {
 236             last_change = wccp_i_see_you.change;
 237             return;
 238         }
 239     }
 240
 241     if (last_change != wccp_i_see_you.change) {
 242         last_change = wccp_i_see_you.change;
 243
 244         if (wccpLowestIP() && wccp_i_see_you.number) {
 245             last_assign_buckets_change = last_change;
 246             wccpAssignBuckets();
 247         }
 248     }
 249 }
 250
 251 static int
 252 wccpLowestIP(void)
 253 {
 254     unsigned int loop;
 255     int found = 0;
 256
 257     /*
 258      * We sanity checked wccp_i_see_you.number back in wccpHandleUdp()
 259      */
 260
 261     for (loop = 0; loop < (unsigned) ntohl(wccp_i_see_you.number); ++loop) {
 262         assert(loop < WCCP_ACTIVE_CACHES);
 263
 264         if (local_ip > wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
 265             return 0;
 266
 267         if (local_ip == wccp_i_see_you.wccp_cache_entry[loop].ip_addr)
 268             found = 1;
 269     }
 270
 271     return found;
 272 }
 273
 274 static void
 275 wccpHereIam(void *)
 276 {
 277     debugs(80, 6, "wccpHereIam: Called");
 278
 279     wccp_here_i_am.id = last_id;
 280     double interval = 10.0; // TODO: make this configurable, possibly negotiate with the router.
 281     ssize_t sent = comm_udp_send(theWccpConnection, &wccp_here_i_am, sizeof(wccp_here_i_am), 0);
 282
 283     // if we failed to send the whole lot, try again at a shorter interval (20%)
 284     if (sent != sizeof(wccp_here_i_am)) {
 285         int xerrno = errno;
 286         debugs(80, 2, "ERROR: failed to send WCCP HERE_I_AM packet: " << xstrerr(xerrno));
 287         interval = 2.0;
 288     }
 289
 290     if (!eventFind(wccpHereIam, nullptr))
 291         eventAdd("wccpHereIam", wccpHereIam, nullptr, interval, 1);
 292 }
 293
 294 static void
 295 wccpAssignBuckets(void)
 296 {
 297
 298     struct wccp_assign_bucket_t *wccp_assign_bucket;
 299     int wab_len;
 300     char *buckets;
 301     int buckets_per_cache;
 302     unsigned int loop;
 303     int bucket = 0;
 304     int *caches;
 305     int cache_len;
 306     char *buf;
 307
 308     debugs(80, 6, "wccpAssignBuckets: Called");
 309     number_caches = ntohl(wccp_i_see_you.number);
 310
 311     assert(number_caches > 0);
 312     assert(number_caches <= WCCP_ACTIVE_CACHES);
 313
 314     wab_len = sizeof(struct wccp_assign_bucket_t);
 315
 316     cache_len = WCCP_CACHE_LEN * number_caches;
 317
 318     buf = (char *)xmalloc(wab_len +
 319                           WCCP_BUCKETS +
 320                           cache_len);
 321
 322     wccp_assign_bucket = (struct wccp_assign_bucket_t *) buf;
 323
 324     caches = (int *) (buf + wab_len);
 325
 326     buckets = buf + wab_len + cache_len;
 327
 328     memset(wccp_assign_bucket, '\0', sizeof(*wccp_assign_bucket));
 329
 330     memset(buckets, 0xFF, WCCP_BUCKETS);
 331
 332     buckets_per_cache = WCCP_BUCKETS / number_caches;
 333
 334     for (loop = 0; loop < number_caches; ++loop) {
 335         int i;
 336         memcpy(&caches[loop],
 337                &wccp_i_see_you.wccp_cache_entry[loop].ip_addr,
 338                sizeof(*caches));
 339
 340         for (i = 0; i < buckets_per_cache; ++i) {
 341             assert(bucket < WCCP_BUCKETS);
 342             buckets[bucket] = loop;
 343             ++bucket;
 344         }
 345     }
 346
 347     while (bucket < WCCP_BUCKETS) {
 348         buckets[bucket] = number_caches - 1;
 349         ++bucket;
 350     }
 351
 352     wccp_assign_bucket->type = htonl(WCCP_ASSIGN_BUCKET);
 353     wccp_assign_bucket->id = wccp_i_see_you.id;
 354     wccp_assign_bucket->number = wccp_i_see_you.number;
 355
 356     comm_udp_send(theWccpConnection,
 357                   buf,
 358                   wab_len + WCCP_BUCKETS + cache_len,
 359                   0);
 360     last_change = 0;
 361     xfree(buf);
 362 }
 363
 364 #endif /* USE_WCCP */
 365