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