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