tests/test-sflow.c

   1 /*
   2  * Copyright (c) 2011, 2012, 2013, 2014 Nicira, Inc.
   3  * Copyright (c) 2013 InMon Corp.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at:
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  */
  17
  18 #include <config.h>
  19
  20 #include <arpa/inet.h>
  21 #include <errno.h>
  22 #include <getopt.h>
  23 #include <signal.h>
  24 #include <stdlib.h>
  25 #include <unistd.h>
  26 #include <setjmp.h>
  27
  28 #include "command-line.h"
  29 #include "daemon.h"
  30 #include "dynamic-string.h"
  31 #include "netflow.h"
  32 #include "ofpbuf.h"
  33 #include "packets.h"
  34 #include "poll-loop.h"
  35 #include "socket-util.h"
  36 #include "unixctl.h"
  37 #include "util.h"
  38 #include "vlog.h"
  39 #include "ovstest.h"
  40
  41 static void usage(void) NO_RETURN;
  42 static void parse_options(int argc, char *argv[]);
  43
  44 static unixctl_cb_func test_sflow_exit;
  45
  46 /* Datagram. */
  47 #define SFLOW_VERSION_5 5
  48 #define SFLOW_MIN_LEN 36
  49
  50 /* Sample tag numbers. */
  51 #define SFLOW_FLOW_SAMPLE 1
  52 #define SFLOW_COUNTERS_SAMPLE 2
  53 #define SFLOW_FLOW_SAMPLE_EXPANDED 3
  54 #define SFLOW_COUNTERS_SAMPLE_EXPANDED 4
  55
  56 /* Structure element tag numbers. */
  57 #define SFLOW_TAG_CTR_IFCOUNTERS 1
  58 #define SFLOW_TAG_PKT_HEADER 1
  59 #define SFLOW_TAG_PKT_SWITCH 1001
  60
  61 struct sflow_addr {
  62     enum {
  63         SFLOW_ADDRTYPE_undefined = 0,
  64         SFLOW_ADDRTYPE_IP4,
  65         SFLOW_ADDRTYPE_IP6
  66     } type;
  67
  68     union {
  69         ovs_be32 ip4;
  70         ovs_be32 ip6[4];
  71     } a;
  72 };
  73
  74 struct sflow_xdr {
  75     /* Exceptions. */
  76     jmp_buf env;
  77     int errline;
  78
  79     /* Cursor. */
  80     ovs_be32 *datap;
  81     uint32_t i;
  82     uint32_t quads;
  83
  84     /* Agent. */
  85     struct sflow_addr agentAddr;
  86     char agentIPStr[INET6_ADDRSTRLEN + 2];
  87     uint32_t subAgentId;
  88     uint32_t uptime_mS;
  89
  90     /* Datasource. */
  91     uint32_t dsClass;
  92     uint32_t dsIndex;
  93
  94     /* Sequence numbers. */
  95     uint32_t dgramSeqNo;
  96     uint32_t fsSeqNo;
  97     uint32_t csSeqNo;
  98
  99     /* Structure offsets. */
 100     struct {
 101         uint32_t HEADER;
 102         uint32_t SWITCH;
 103         uint32_t IFCOUNTERS;
 104     } offset;
 105
 106     /* Flow sample fields. */
 107     uint32_t meanSkipCount;
 108     uint32_t samplePool;
 109     uint32_t dropEvents;
 110     uint32_t inputPortFormat;
 111     uint32_t inputPort;
 112     uint32_t outputPortFormat;
 113     uint32_t outputPort;
 114 };
 115
 116 #define SFLOWXDR_try(x) ((x->errline = setjmp(x->env)) == 0)
 117 #define SFLOWXDR_throw(x) longjmp(x->env, __LINE__)
 118 #define SFLOWXDR_assert(x, t) if (!(t)) SFLOWXDR_throw(x)
 119
 120 static void
 121 sflowxdr_init(struct sflow_xdr *x, void *buf, size_t len)
 122 {
 123     x->datap = buf;
 124     x->quads = len >> 2;
 125 }
 126
 127 static uint32_t
 128 sflowxdr_next(struct sflow_xdr *x)
 129 {
 130     return ntohl(x->datap[x->i++]);
 131 }
 132
 133 static ovs_be32
 134 sflowxdr_next_n(struct sflow_xdr *x)
 135 {
 136     return x->datap[x->i++];
 137 }
 138
 139 static bool
 140 sflowxdr_more(const struct sflow_xdr *x, uint32_t q)
 141 {
 142     return q + x->i <= x->quads;
 143 }
 144
 145 static void
 146 sflowxdr_skip(struct sflow_xdr *x, uint32_t q)
 147 {
 148     x->i += q;
 149 }
 150
 151 static uint32_t
 152 sflowxdr_mark(const struct sflow_xdr *x, uint32_t q)
 153 {
 154     return x->i + q;
 155 }
 156
 157 static bool
 158 sflowxdr_mark_ok(const struct sflow_xdr *x, uint32_t m)
 159 {
 160     return m == x->i;
 161 }
 162
 163 static void
 164 sflowxdr_mark_unique(struct sflow_xdr *x, uint32_t *pi)
 165 {
 166     if (*pi) {
 167         SFLOWXDR_throw(x);
 168     }
 169     *pi = x->i;
 170 }
 171
 172 static void
 173 sflowxdr_setc(struct sflow_xdr *x, uint32_t j)
 174 {
 175     x->i = j;
 176 }
 177
 178 static const char *
 179 sflowxdr_str(const struct sflow_xdr *x)
 180 {
 181     return (const char *) (x->datap + x->i);
 182 }
 183
 184 static uint64_t
 185 sflowxdr_next_int64(struct sflow_xdr *x)
 186 {
 187     uint64_t scratch;
 188     scratch = sflowxdr_next(x);
 189     scratch <<= 32;
 190     scratch += sflowxdr_next(x);
 191     return scratch;
 192 }
 193
 194 static void
 195 process_counter_sample(struct sflow_xdr *x)
 196 {
 197     if (x->offset.IFCOUNTERS) {
 198         sflowxdr_setc(x, x->offset.IFCOUNTERS);
 199         printf("IFCOUNTERS");
 200         printf(" dgramSeqNo=%"PRIu32, x->dgramSeqNo);
 201         printf(" ds=%s>%"PRIu32":%"PRIu32,
 202                x->agentIPStr, x->dsClass, x->dsIndex);
 203         printf(" csSeqNo=%"PRIu32, x->csSeqNo);
 204         printf(" ifindex=%"PRIu32, sflowxdr_next(x));
 205         printf(" type=%"PRIu32, sflowxdr_next(x));
 206         printf(" ifspeed=%"PRIu64, sflowxdr_next_int64(x));
 207         printf(" direction=%"PRIu32, sflowxdr_next(x));
 208         printf(" status=%"PRIu32, sflowxdr_next(x));
 209         printf(" in_octets=%"PRIu64, sflowxdr_next_int64(x));
 210         printf(" in_unicasts=%"PRIu32, sflowxdr_next(x));
 211         printf(" in_multicasts=%"PRIu32, sflowxdr_next(x));
 212         printf(" in_broadcasts=%"PRIu32, sflowxdr_next(x));
 213         printf(" in_discards=%"PRIu32, sflowxdr_next(x));
 214         printf(" in_errors=%"PRIu32, sflowxdr_next(x));
 215         printf(" in_unknownprotos=%"PRIu32, sflowxdr_next(x));
 216         printf(" out_octets=%"PRIu64, sflowxdr_next_int64(x));
 217         printf(" out_unicasts=%"PRIu32, sflowxdr_next(x));
 218         printf(" out_multicasts=%"PRIu32, sflowxdr_next(x));
 219         printf(" out_broadcasts=%"PRIu32, sflowxdr_next(x));
 220         printf(" out_discards=%"PRIu32, sflowxdr_next(x));
 221         printf(" out_errors=%"PRIu32, sflowxdr_next(x));
 222         printf(" promiscuous=%"PRIu32, sflowxdr_next(x));
 223         printf("\n");
 224     }
 225 }
 226
 227 static char
 228 bin_to_hex(int hexit)
 229 {
 230     return "0123456789ABCDEF"[hexit];
 231 }
 232
 233 static int
 234 print_hex(const char *a, int len, char *buf, int bufLen)
 235 {
 236     unsigned char nextByte;
 237     int b = 0;
 238     int i;
 239
 240     for (i = 0; i < len; i++) {
 241         if (b > bufLen - 10) {
 242             break;
 243         }
 244         nextByte = a[i];
 245         buf[b++] = bin_to_hex(nextByte >> 4);
 246         buf[b++] = bin_to_hex(nextByte & 0x0f);
 247         if (i < len - 1) {
 248             buf[b++] = '-';
 249         }
 250     }
 251     buf[b] = '\0';
 252     return b;
 253 }
 254
 255 #define SFLOW_HEX_SCRATCH 1024
 256
 257 static void
 258 process_flow_sample(struct sflow_xdr *x)
 259 {
 260     if (x->offset.HEADER) {
 261         uint32_t headerLen;
 262         char scratch[SFLOW_HEX_SCRATCH];
 263
 264         printf("HEADER");
 265         printf(" dgramSeqNo=%"PRIu32, x->dgramSeqNo);
 266         printf(" ds=%s>%"PRIu32":%"PRIu32,
 267                x->agentIPStr, x->dsClass, x->dsIndex);
 268         printf(" fsSeqNo=%"PRIu32, x->fsSeqNo);
 269
 270         if (x->offset.SWITCH) {
 271             sflowxdr_setc(x, x->offset.SWITCH);
 272             printf(" in_vlan=%"PRIu32, sflowxdr_next(x));
 273             printf(" in_priority=%"PRIu32, sflowxdr_next(x));
 274             printf(" out_vlan=%"PRIu32, sflowxdr_next(x));
 275             printf(" out_priority=%"PRIu32, sflowxdr_next(x));
 276         }
 277
 278         sflowxdr_setc(x, x->offset.HEADER);
 279         printf(" meanSkip=%"PRIu32, x->meanSkipCount);
 280         printf(" samplePool=%"PRIu32, x->samplePool);
 281         printf(" dropEvents=%"PRIu32, x->dropEvents);
 282         printf(" in_ifindex=%"PRIu32, x->inputPort);
 283         printf(" in_format=%"PRIu32, x->inputPortFormat);
 284         printf(" out_ifindex=%"PRIu32, x->outputPort);
 285         printf(" out_format=%"PRIu32, x->outputPortFormat);
 286         printf(" hdr_prot=%"PRIu32, sflowxdr_next(x));
 287         printf(" pkt_len=%"PRIu32, sflowxdr_next(x));
 288         printf(" stripped=%"PRIu32, sflowxdr_next(x));
 289         headerLen = sflowxdr_next(x);
 290         printf(" hdr_len=%"PRIu32, headerLen);
 291         print_hex(sflowxdr_str(x), headerLen, scratch, SFLOW_HEX_SCRATCH);
 292         printf(" hdr=%s", scratch);
 293         printf("\n");
 294     }
 295 }
 296
 297 static void
 298 process_datagram(struct sflow_xdr *x)
 299 {
 300     uint32_t samples, s;
 301
 302     SFLOWXDR_assert(x, (sflowxdr_next(x) == SFLOW_VERSION_5));
 303
 304     /* Read the sFlow header. */
 305     x->agentAddr.type = sflowxdr_next(x);
 306     switch (x->agentAddr.type) {
 307     case SFLOW_ADDRTYPE_IP4:
 308         x->agentAddr.a.ip4 = sflowxdr_next_n(x);
 309         break;
 310
 311     case SFLOW_ADDRTYPE_IP6:
 312         x->agentAddr.a.ip6[0] = sflowxdr_next_n(x);
 313         x->agentAddr.a.ip6[1] = sflowxdr_next_n(x);
 314         x->agentAddr.a.ip6[2] = sflowxdr_next_n(x);
 315         x->agentAddr.a.ip6[3] = sflowxdr_next_n(x);
 316         break;
 317
 318     case SFLOW_ADDRTYPE_undefined:
 319     default:
 320         SFLOWXDR_throw(x);
 321         break;
 322     }
 323     x->subAgentId = sflowxdr_next(x);
 324     x->dgramSeqNo = sflowxdr_next(x);
 325     x->uptime_mS = sflowxdr_next(x);
 326
 327     /* Store the agent address as a string. */
 328     if (x->agentAddr.type == SFLOW_ADDRTYPE_IP6) {
 329         char ipstr[INET6_ADDRSTRLEN];
 330         inet_ntop(AF_INET6, (const void *) &x->agentAddr.a.ip6,
 331                   ipstr, INET6_ADDRSTRLEN);
 332         snprintf(x->agentIPStr, sizeof x->agentIPStr, "[%s]", ipstr);
 333     } else {
 334         snprintf(x->agentIPStr, sizeof x->agentIPStr,
 335                  IP_FMT, IP_ARGS(x->agentAddr.a.ip4));
 336     }
 337
 338     /* Array of flow/counter samples. */
 339     samples = sflowxdr_next(x);
 340     for (s = 0; s < samples; s++) {
 341         uint32_t sType = sflowxdr_next(x);
 342         uint32_t sQuads = sflowxdr_next(x) >> 2;
 343         uint32_t sMark = sflowxdr_mark(x, sQuads);
 344         SFLOWXDR_assert(x, sflowxdr_more(x, sQuads));
 345
 346         switch (sType) {
 347         case SFLOW_COUNTERS_SAMPLE_EXPANDED:
 348         case SFLOW_COUNTERS_SAMPLE:
 349         {
 350             uint32_t csElements, e;
 351             uint32_t ceTag, ceQuads, ceMark, csEnd;
 352
 353             x->csSeqNo = sflowxdr_next(x);
 354             if (sType == SFLOW_COUNTERS_SAMPLE_EXPANDED) {
 355                 x->dsClass = sflowxdr_next(x);
 356                 x->dsIndex = sflowxdr_next(x);
 357             } else {
 358                 uint32_t dsCombined = sflowxdr_next(x);
 359                 x->dsClass = dsCombined >> 24;
 360                 x->dsIndex = dsCombined & 0x00FFFFFF;
 361             }
 362
 363             csElements = sflowxdr_next(x);
 364             for (e = 0; e < csElements; e++) {
 365                 SFLOWXDR_assert(x, sflowxdr_more(x,2));
 366                 ceTag = sflowxdr_next(x);
 367                 ceQuads = sflowxdr_next(x) >> 2;
 368                 ceMark = sflowxdr_mark(x, ceQuads);
 369                 SFLOWXDR_assert(x, sflowxdr_more(x,ceQuads));
 370                 /* Only care about selected structures.  Just record their
 371                  * offsets here. We'll read the fields out later. */
 372                 switch (ceTag) {
 373                 case SFLOW_TAG_CTR_IFCOUNTERS:
 374                     sflowxdr_mark_unique(x, &x->offset.IFCOUNTERS);
 375                     break;
 376
 377                     /* Add others here... */
 378                 }
 379
 380                 sflowxdr_skip(x, ceQuads);
 381                 SFLOWXDR_assert(x, sflowxdr_mark_ok(x, ceMark));
 382             }
 383
 384             csEnd = sflowxdr_mark(x, 0);
 385             process_counter_sample(x);
 386             /* Make sure we pick up the decoding where we left off. */
 387             sflowxdr_setc(x, csEnd);
 388
 389             /* Clear the offsets for the next sample. */
 390             memset(&x->offset, 0, sizeof x->offset);
 391         }
 392         break;
 393
 394         case SFLOW_FLOW_SAMPLE:
 395         case SFLOW_FLOW_SAMPLE_EXPANDED:
 396         {
 397             uint32_t fsElements, e;
 398             uint32_t feTag, feQuads, feMark, fsEnd;
 399             x->fsSeqNo = sflowxdr_next(x);
 400             if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
 401                 x->dsClass = sflowxdr_next(x);
 402                 x->dsIndex = sflowxdr_next(x);
 403             } else {
 404                 uint32_t dsCombined = sflowxdr_next(x);
 405                 x->dsClass = dsCombined >> 24;
 406                 x->dsIndex = dsCombined & 0x00FFFFFF;
 407             }
 408             x->meanSkipCount = sflowxdr_next(x);
 409             x->samplePool = sflowxdr_next(x);
 410             x->dropEvents = sflowxdr_next(x);
 411             if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
 412                 x->inputPortFormat = sflowxdr_next(x);
 413                 x->inputPort = sflowxdr_next(x);
 414                 x->outputPortFormat = sflowxdr_next(x);
 415                 x->outputPort = sflowxdr_next(x);
 416             } else {
 417                 uint32_t inp, outp;
 418
 419                 inp = sflowxdr_next(x);
 420                 outp = sflowxdr_next(x);
 421                 x->inputPortFormat = inp >> 30;
 422                 x->inputPort = inp & 0x3fffffff;
 423                 x->outputPortFormat = outp >> 30;
 424                 x->outputPort = outp & 0x3fffffff;
 425             }
 426             fsElements = sflowxdr_next(x);
 427             for (e = 0; e < fsElements; e++) {
 428                 SFLOWXDR_assert(x, sflowxdr_more(x,2));
 429                 feTag = sflowxdr_next(x);
 430                 feQuads = sflowxdr_next(x) >> 2;
 431                 feMark = sflowxdr_mark(x, feQuads);
 432                 SFLOWXDR_assert(x, sflowxdr_more(x,feQuads));
 433                 /* Only care about selected structures.  Just record their
 434                  * offsets here. We'll read the fields out below. */
 435                 switch (feTag) {
 436                 case SFLOW_TAG_PKT_HEADER:
 437                     sflowxdr_mark_unique(x, &x->offset.HEADER);
 438                     break;
 439
 440                 case SFLOW_TAG_PKT_SWITCH:
 441                     sflowxdr_mark_unique(x, &x->offset.SWITCH);
 442                     break;
 443
 444                     /* Add others here... */
 445                 }
 446
 447                 sflowxdr_skip(x, feQuads);
 448                 SFLOWXDR_assert(x, sflowxdr_mark_ok(x, feMark));
 449             }
 450
 451             fsEnd = sflowxdr_mark(x, 0);
 452             process_flow_sample(x);
 453             /* Make sure we pick up the decoding where we left off. */
 454             sflowxdr_setc(x, fsEnd);
 455
 456             /* Clear the offsets for the next counter/flow sample. */
 457             memset(&x->offset, 0, sizeof x->offset);
 458         }
 459         break;
 460
 461         default:
 462             /* Skip other sample types. */
 463             sflowxdr_skip(x, sQuads);
 464         }
 465         SFLOWXDR_assert(x, sflowxdr_mark_ok(x, sMark));
 466     }
 467 }
 468
 469 static void
 470 print_sflow(struct ofpbuf *buf)
 471 {
 472     char *dgram_buf;
 473     int dgram_len = ofpbuf_size(buf);
 474     struct sflow_xdr xdrDatagram;
 475     struct sflow_xdr *x = &xdrDatagram;
 476
 477     memset(x, 0, sizeof *x);
 478     if (SFLOWXDR_try(x)) {
 479         SFLOWXDR_assert(x, (dgram_buf = ofpbuf_try_pull(buf, ofpbuf_size(buf))));
 480         sflowxdr_init(x, dgram_buf, dgram_len);
 481         SFLOWXDR_assert(x, dgram_len >= SFLOW_MIN_LEN);
 482         process_datagram(x);
 483     } else {
 484         // CATCH
 485         printf("\n>>>>> ERROR in " __FILE__ " at line %u\n", x->errline);
 486     }
 487 }
 488
 489 static void
 490 test_sflow_main(int argc, char *argv[])
 491 {
 492     struct unixctl_server *server;
 493     enum { MAX_RECV = 1500 };
 494     const char *target;
 495     struct ofpbuf buf;
 496     bool exiting = false;
 497     int error;
 498     int sock;
 499
 500     proctitle_init(argc, argv);
 501     set_program_name(argv[0]);
 502     parse_options(argc, argv);
 503
 504     if (argc - optind != 1) {
 505         ovs_fatal(0, "exactly one non-option argument required "
 506                   "(use --help for help)");
 507     }
 508     target = argv[optind];
 509
 510     sock = inet_open_passive(SOCK_DGRAM, target, 0, NULL, 0);
 511     if (sock < 0) {
 512         ovs_fatal(0, "%s: failed to open (%s)", argv[1], ovs_strerror(-sock));
 513     }
 514
 515     daemon_save_fd(STDOUT_FILENO);
 516     daemonize_start();
 517
 518     error = unixctl_server_create(NULL, &server);
 519     if (error) {
 520         ovs_fatal(error, "failed to create unixctl server");
 521     }
 522     unixctl_command_register("exit", "", 0, 0, test_sflow_exit, &exiting);
 523
 524     daemonize_complete();
 525
 526     ofpbuf_init(&buf, MAX_RECV);
 527     for (;;) {
 528         int retval;
 529
 530         unixctl_server_run(server);
 531
 532         ofpbuf_clear(&buf);
 533         do {
 534             retval = read(sock, ofpbuf_data(&buf), buf.allocated);
 535         } while (retval < 0 && errno == EINTR);
 536         if (retval > 0) {
 537             ofpbuf_put_uninit(&buf, retval);
 538             print_sflow(&buf);
 539             fflush(stdout);
 540         }
 541
 542         if (exiting) {
 543             break;
 544         }
 545
 546         poll_fd_wait(sock, POLLIN);
 547         unixctl_server_wait(server);
 548         poll_block();
 549     }
 550 }
 551
 552 static void
 553 parse_options(int argc, char *argv[])
 554 {
 555     enum {
 556         DAEMON_OPTION_ENUMS,
 557         VLOG_OPTION_ENUMS
 558     };
 559     static const struct option long_options[] = {
 560         {"verbose", optional_argument, NULL, 'v'},
 561         {"help", no_argument, NULL, 'h'},
 562         DAEMON_LONG_OPTIONS,
 563         VLOG_LONG_OPTIONS,
 564         {NULL, 0, NULL, 0},
 565     };
 566     char *short_options = long_options_to_short_options(long_options);
 567
 568     for (;;) {
 569         int c = getopt_long(argc, argv, short_options, long_options, NULL);
 570         if (c == -1) {
 571             break;
 572         }
 573
 574         switch (c) {
 575         case 'h':
 576             usage();
 577
 578         DAEMON_OPTION_HANDLERS
 579         VLOG_OPTION_HANDLERS
 580
 581         case '?':
 582             exit(EXIT_FAILURE);
 583
 584         default:
 585             abort();
 586         }
 587     }
 588     free(short_options);
 589 }
 590
 591 static void
 592 usage(void)
 593 {
 594     printf("%s: sflow collector test utility\n"
 595            "usage: %s [OPTIONS] PORT[:IP]\n"
 596            "where PORT is the UDP port to listen on and IP is optionally\n"
 597            "the IP address to listen on.\n",
 598            program_name, program_name);
 599     daemon_usage();
 600     vlog_usage();
 601     printf("\nOther options:\n"
 602            "  -h, --help                  display this help message\n");
 603     exit(EXIT_SUCCESS);
 604 }
 605
 606 static void
 607 test_sflow_exit(struct unixctl_conn *conn,
 608                 int argc OVS_UNUSED, const char *argv[] OVS_UNUSED,
 609                 void *exiting_)
 610 {
 611     bool *exiting = exiting_;
 612     *exiting = true;
 613     unixctl_command_reply(conn, NULL);
 614 }
 615
 616 OVSTEST_REGISTER("test-sflow", test_sflow_main);