tests/test-sflow.c

   1 /*
   2  * Copyright (c) 2011, 2012, 2013 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 #define SFLOW_MAX_AGENTIP_STRLEN 64
  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[SFLOW_MAX_AGENTIP_STRLEN];
  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         snprintf(x->agentIPStr, SFLOW_MAX_AGENTIP_STRLEN,
 329                  "%04x:%04x:%04x:%04x",
 330                  x->agentAddr.a.ip6[0],
 331                  x->agentAddr.a.ip6[1],
 332                  x->agentAddr.a.ip6[2],
 333                  x->agentAddr.a.ip6[3]);
 334     } else {
 335         snprintf(x->agentIPStr, SFLOW_MAX_AGENTIP_STRLEN,
 336                  IP_FMT, IP_ARGS(x->agentAddr.a.ip4));
 337     }
 338
 339     /* Array of flow/counter samples. */
 340     samples = sflowxdr_next(x);
 341     for (s = 0; s < samples; s++) {
 342         uint32_t sType = sflowxdr_next(x);
 343         uint32_t sQuads = sflowxdr_next(x) >> 2;
 344         uint32_t sMark = sflowxdr_mark(x, sQuads);
 345         SFLOWXDR_assert(x, sflowxdr_more(x, sQuads));
 346
 347         switch (sType) {
 348         case SFLOW_COUNTERS_SAMPLE_EXPANDED:
 349         case SFLOW_COUNTERS_SAMPLE:
 350         {
 351             uint32_t csElements, e;
 352             uint32_t ceTag, ceQuads, ceMark, csEnd;
 353
 354             x->csSeqNo = sflowxdr_next(x);
 355             if (sType == SFLOW_COUNTERS_SAMPLE_EXPANDED) {
 356                 x->dsClass = sflowxdr_next(x);
 357                 x->dsIndex = sflowxdr_next(x);
 358             } else {
 359                 uint32_t dsCombined = sflowxdr_next(x);
 360                 x->dsClass = dsCombined >> 24;
 361                 x->dsIndex = dsCombined & 0x00FFFFFF;
 362             }
 363
 364             csElements = sflowxdr_next(x);
 365             for (e = 0; e < csElements; e++) {
 366                 SFLOWXDR_assert(x, sflowxdr_more(x,2));
 367                 ceTag = sflowxdr_next(x);
 368                 ceQuads = sflowxdr_next(x) >> 2;
 369                 ceMark = sflowxdr_mark(x, ceQuads);
 370                 SFLOWXDR_assert(x, sflowxdr_more(x,ceQuads));
 371                 /* Only care about selected structures.  Just record their
 372                  * offsets here. We'll read the fields out later. */
 373                 switch (ceTag) {
 374                 case SFLOW_TAG_CTR_IFCOUNTERS:
 375                     sflowxdr_mark_unique(x, &x->offset.IFCOUNTERS);
 376                     break;
 377
 378                     /* Add others here... */
 379                 }
 380
 381                 sflowxdr_skip(x, ceQuads);
 382                 SFLOWXDR_assert(x, sflowxdr_mark_ok(x, ceMark));
 383             }
 384
 385             csEnd = sflowxdr_mark(x, 0);
 386             process_counter_sample(x);
 387             /* Make sure we pick up the decoding where we left off. */
 388             sflowxdr_setc(x, csEnd);
 389
 390             /* Clear the offsets for the next sample. */
 391             memset(&x->offset, 0, sizeof x->offset);
 392         }
 393         break;
 394
 395         case SFLOW_FLOW_SAMPLE:
 396         case SFLOW_FLOW_SAMPLE_EXPANDED:
 397         {
 398             uint32_t fsElements, e;
 399             uint32_t feTag, feQuads, feMark, fsEnd;
 400             x->fsSeqNo = sflowxdr_next(x);
 401             if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
 402                 x->dsClass = sflowxdr_next(x);
 403                 x->dsIndex = sflowxdr_next(x);
 404             } else {
 405                 uint32_t dsCombined = sflowxdr_next(x);
 406                 x->dsClass = dsCombined >> 24;
 407                 x->dsIndex = dsCombined & 0x00FFFFFF;
 408             }
 409             x->meanSkipCount = sflowxdr_next(x);
 410             x->samplePool = sflowxdr_next(x);
 411             x->dropEvents = sflowxdr_next(x);
 412             if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
 413                 x->inputPortFormat = sflowxdr_next(x);
 414                 x->inputPort = sflowxdr_next(x);
 415                 x->outputPortFormat = sflowxdr_next(x);
 416                 x->outputPort = sflowxdr_next(x);
 417             } else {
 418                 uint32_t inp, outp;
 419
 420                 inp = sflowxdr_next(x);
 421                 outp = sflowxdr_next(x);
 422                 x->inputPortFormat = inp >> 30;
 423                 x->inputPort = inp & 0x3fffffff;
 424                 x->outputPortFormat = outp >> 30;
 425                 x->outputPort = outp & 0x3fffffff;
 426             }
 427             fsElements = sflowxdr_next(x);
 428             for (e = 0; e < fsElements; e++) {
 429                 SFLOWXDR_assert(x, sflowxdr_more(x,2));
 430                 feTag = sflowxdr_next(x);
 431                 feQuads = sflowxdr_next(x) >> 2;
 432                 feMark = sflowxdr_mark(x, feQuads);
 433                 SFLOWXDR_assert(x, sflowxdr_more(x,feQuads));
 434                 /* Only care about selected structures.  Just record their
 435                  * offsets here. We'll read the fields out below. */
 436                 switch (feTag) {
 437                 case SFLOW_TAG_PKT_HEADER:
 438                     sflowxdr_mark_unique(x, &x->offset.HEADER);
 439                     break;
 440
 441                 case SFLOW_TAG_PKT_SWITCH:
 442                     sflowxdr_mark_unique(x, &x->offset.SWITCH);
 443                     break;
 444
 445                     /* Add others here... */
 446                 }
 447
 448                 sflowxdr_skip(x, feQuads);
 449                 SFLOWXDR_assert(x, sflowxdr_mark_ok(x, feMark));
 450             }
 451
 452             fsEnd = sflowxdr_mark(x, 0);
 453             process_flow_sample(x);
 454             /* Make sure we pick up the decoding where we left off. */
 455             sflowxdr_setc(x, fsEnd);
 456
 457             /* Clear the offsets for the next counter/flow sample. */
 458             memset(&x->offset, 0, sizeof x->offset);
 459         }
 460         break;
 461
 462         default:
 463             /* Skip other sample types. */
 464             sflowxdr_skip(x, sQuads);
 465         }
 466         SFLOWXDR_assert(x, sflowxdr_mark_ok(x, sMark));
 467     }
 468 }
 469
 470 static void
 471 print_sflow(struct ofpbuf *buf)
 472 {
 473     char *dgram_buf;
 474     int dgram_len = buf->size;
 475     struct sflow_xdr xdrDatagram;
 476     struct sflow_xdr *x = &xdrDatagram;
 477
 478     memset(x, 0, sizeof *x);
 479     if (SFLOWXDR_try(x)) {
 480         SFLOWXDR_assert(x, (dgram_buf = ofpbuf_try_pull(buf, buf->size)));
 481         sflowxdr_init(x, dgram_buf, dgram_len);
 482         SFLOWXDR_assert(x, dgram_len >= SFLOW_MIN_LEN);
 483         process_datagram(x);
 484     } else {
 485         // CATCH
 486         printf("\n>>>>> ERROR in " __FILE__ " at line %u\n", x->errline);
 487     }
 488 }
 489
 490 int
 491 main(int argc, char *argv[])
 492 {
 493     struct unixctl_server *server;
 494     enum { MAX_RECV = 1500 };
 495     const char *target;
 496     struct ofpbuf buf;
 497     bool exiting = false;
 498     int error;
 499     int sock;
 500
 501     proctitle_init(argc, argv);
 502     set_program_name(argv[0]);
 503     parse_options(argc, argv);
 504
 505     if (argc - optind != 1) {
 506         ovs_fatal(0, "exactly one non-option argument required "
 507                   "(use --help for help)");
 508     }
 509     target = argv[optind];
 510
 511     sock = inet_open_passive(SOCK_DGRAM, target, 0, NULL, 0);
 512     if (sock < 0) {
 513         ovs_fatal(0, "%s: failed to open (%s)", argv[1], strerror(-sock));
 514     }
 515
 516     daemon_save_fd(STDOUT_FILENO);
 517     daemonize_start();
 518
 519     error = unixctl_server_create(NULL, &server);
 520     if (error) {
 521         ovs_fatal(error, "failed to create unixctl server");
 522     }
 523     unixctl_command_register("exit", "", 0, 0, test_sflow_exit, &exiting);
 524
 525     daemonize_complete();
 526
 527     ofpbuf_init(&buf, MAX_RECV);
 528     for (;;) {
 529         int retval;
 530
 531         unixctl_server_run(server);
 532
 533         ofpbuf_clear(&buf);
 534         do {
 535             retval = read(sock, buf.data, buf.allocated);
 536         } while (retval < 0 && errno == EINTR);
 537         if (retval > 0) {
 538             ofpbuf_put_uninit(&buf, retval);
 539             print_sflow(&buf);
 540             fflush(stdout);
 541         }
 542
 543         if (exiting) {
 544             break;
 545         }
 546
 547         poll_fd_wait(sock, POLLIN);
 548         unixctl_server_wait(server);
 549         poll_block();
 550     }
 551
 552     return 0;
 553 }
 554
 555 static void
 556 parse_options(int argc, char *argv[])
 557 {
 558     enum {
 559         DAEMON_OPTION_ENUMS,
 560         VLOG_OPTION_ENUMS
 561     };
 562     static const struct option long_options[] = {
 563         {"verbose", optional_argument, NULL, 'v'},
 564         {"help", no_argument, NULL, 'h'},
 565         DAEMON_LONG_OPTIONS,
 566         VLOG_LONG_OPTIONS,
 567         {NULL, 0, NULL, 0},
 568     };
 569     char *short_options = long_options_to_short_options(long_options);
 570
 571     for (;;) {
 572         int c = getopt_long(argc, argv, short_options, long_options, NULL);
 573         if (c == -1) {
 574             break;
 575         }
 576
 577         switch (c) {
 578         case 'h':
 579             usage();
 580
 581         DAEMON_OPTION_HANDLERS
 582         VLOG_OPTION_HANDLERS
 583
 584         case '?':
 585             exit(EXIT_FAILURE);
 586
 587         default:
 588             abort();
 589         }
 590     }
 591     free(short_options);
 592 }
 593
 594 static void
 595 usage(void)
 596 {
 597     printf("%s: sflow collector test utility\n"
 598            "usage: %s [OPTIONS] PORT[:IP]\n"
 599            "where PORT is the UDP port to listen on and IP is optionally\n"
 600            "the IP address to listen on.\n",
 601            program_name, program_name);
 602     daemon_usage();
 603     vlog_usage();
 604     printf("\nOther options:\n"
 605            "  -h, --help                  display this help message\n");
 606     exit(EXIT_SUCCESS);
 607 }
 608
 609 static void
 610 test_sflow_exit(struct unixctl_conn *conn,
 611                 int argc OVS_UNUSED, const char *argv[] OVS_UNUSED,
 612                 void *exiting_)
 613 {
 614     bool *exiting = exiting_;
 615     *exiting = true;
 616     unixctl_command_reply(conn, NULL);
 617 }