[ovs.git] / tests / test-sflow.c

/*
 * Copyright (c) 2011, 2012, 2013, 2014 Nicira, Inc.
 * Copyright (c) 2013 InMon Corp.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <config.h>

#include <arpa/inet.h>
#include <errno.h>
#include <getopt.h>
#include <signal.h>
#include <stdlib.h>
#include <unistd.h>
#include <setjmp.h>

#include "command-line.h"
#include "daemon.h"
#include "dynamic-string.h"
#include "netflow.h"
#include "ofpbuf.h"
#include "packets.h"
#include "poll-loop.h"
#include "socket-util.h"
#include "unixctl.h"
#include "util.h"
#include "vlog.h"
#include "ovstest.h"

static void usage(void) NO_RETURN;
static void parse_options(int argc, char *argv[]);

static unixctl_cb_func test_sflow_exit;

/* Datagram. */
#define SFLOW_VERSION_5 5
#define SFLOW_MIN_LEN 36

/* Sample tag numbers. */
#define SFLOW_FLOW_SAMPLE 1
#define SFLOW_COUNTERS_SAMPLE 2
#define SFLOW_FLOW_SAMPLE_EXPANDED 3
#define SFLOW_COUNTERS_SAMPLE_EXPANDED 4

/* Structure element tag numbers. */
#define SFLOW_TAG_CTR_IFCOUNTERS 1
#define SFLOW_TAG_PKT_HEADER 1
#define SFLOW_TAG_PKT_SWITCH 1001

struct sflow_addr {
    enum {
        SFLOW_ADDRTYPE_undefined = 0,
        SFLOW_ADDRTYPE_IP4,
        SFLOW_ADDRTYPE_IP6
    } type;

    union {
        ovs_be32 ip4;
        ovs_be32 ip6[4];
    } a;
};

struct sflow_xdr {
    /* Exceptions. */
    jmp_buf env;
    int errline;

    /* Cursor. */
    ovs_be32 *datap;
    uint32_t i;
    uint32_t quads;

    /* Agent. */
    struct sflow_addr agentAddr;
    char agentIPStr[INET6_ADDRSTRLEN + 2];
    uint32_t subAgentId;
    uint32_t uptime_mS;

    /* Datasource. */
    uint32_t dsClass;
    uint32_t dsIndex;

    /* Sequence numbers. */
    uint32_t dgramSeqNo;
    uint32_t fsSeqNo;
    uint32_t csSeqNo;

    /* Structure offsets. */
    struct {
        uint32_t HEADER;
        uint32_t SWITCH;
        uint32_t IFCOUNTERS;
    } offset;

    /* Flow sample fields. */
    uint32_t meanSkipCount;
    uint32_t samplePool;
    uint32_t dropEvents;
    uint32_t inputPortFormat;
    uint32_t inputPort;
    uint32_t outputPortFormat;
    uint32_t outputPort;
};

#define SFLOWXDR_try(x) ((x->errline = setjmp(x->env)) == 0)
#define SFLOWXDR_throw(x) longjmp(x->env, __LINE__)
#define SFLOWXDR_assert(x, t) if (!(t)) SFLOWXDR_throw(x)

static void
sflowxdr_init(struct sflow_xdr *x, void *buf, size_t len)
{
    x->datap = buf;
    x->quads = len >> 2;
}

static uint32_t
sflowxdr_next(struct sflow_xdr *x)
{
    return ntohl(x->datap[x->i++]);
}

static ovs_be32
sflowxdr_next_n(struct sflow_xdr *x)
{
    return x->datap[x->i++];
}

static bool
sflowxdr_more(const struct sflow_xdr *x, uint32_t q)
{
    return q + x->i <= x->quads;
}

static void
sflowxdr_skip(struct sflow_xdr *x, uint32_t q)
{
    x->i += q;
}

static uint32_t
sflowxdr_mark(const struct sflow_xdr *x, uint32_t q)
{
    return x->i + q;
}

static bool
sflowxdr_mark_ok(const struct sflow_xdr *x, uint32_t m)
{
    return m == x->i;
}

static void
sflowxdr_mark_unique(struct sflow_xdr *x, uint32_t *pi)
{
    if (*pi) {
        SFLOWXDR_throw(x);
    }
    *pi = x->i;
}

static void
sflowxdr_setc(struct sflow_xdr *x, uint32_t j)
{
    x->i = j;
}

static const char *
sflowxdr_str(const struct sflow_xdr *x)
{
    return (const char *) (x->datap + x->i);
}

static uint64_t
sflowxdr_next_int64(struct sflow_xdr *x)
{
    uint64_t scratch;
    scratch = sflowxdr_next(x);
    scratch <<= 32;
    scratch += sflowxdr_next(x);
    return scratch;
}

static void
process_counter_sample(struct sflow_xdr *x)
{
    if (x->offset.IFCOUNTERS) {
        sflowxdr_setc(x, x->offset.IFCOUNTERS);
        printf("IFCOUNTERS");
        printf(" dgramSeqNo=%"PRIu32, x->dgramSeqNo);
        printf(" ds=%s>%"PRIu32":%"PRIu32,
               x->agentIPStr, x->dsClass, x->dsIndex);
        printf(" csSeqNo=%"PRIu32, x->csSeqNo);
        printf(" ifindex=%"PRIu32, sflowxdr_next(x));
        printf(" type=%"PRIu32, sflowxdr_next(x));
        printf(" ifspeed=%"PRIu64, sflowxdr_next_int64(x));
        printf(" direction=%"PRIu32, sflowxdr_next(x));
        printf(" status=%"PRIu32, sflowxdr_next(x));
        printf(" in_octets=%"PRIu64, sflowxdr_next_int64(x));
        printf(" in_unicasts=%"PRIu32, sflowxdr_next(x));
        printf(" in_multicasts=%"PRIu32, sflowxdr_next(x));
        printf(" in_broadcasts=%"PRIu32, sflowxdr_next(x));
        printf(" in_discards=%"PRIu32, sflowxdr_next(x));
        printf(" in_errors=%"PRIu32, sflowxdr_next(x));
        printf(" in_unknownprotos=%"PRIu32, sflowxdr_next(x));
        printf(" out_octets=%"PRIu64, sflowxdr_next_int64(x));
        printf(" out_unicasts=%"PRIu32, sflowxdr_next(x));
        printf(" out_multicasts=%"PRIu32, sflowxdr_next(x));
        printf(" out_broadcasts=%"PRIu32, sflowxdr_next(x));
        printf(" out_discards=%"PRIu32, sflowxdr_next(x));
        printf(" out_errors=%"PRIu32, sflowxdr_next(x));
        printf(" promiscuous=%"PRIu32, sflowxdr_next(x));
        printf("\n");
    }
}

static char
bin_to_hex(int hexit)
{
    return "0123456789ABCDEF"[hexit];
}

static int
print_hex(const char *a, int len, char *buf, int bufLen)
{
    unsigned char nextByte;
    int b = 0;
    int i;

    for (i = 0; i < len; i++) {
        if (b > bufLen - 10) {
            break;
        }
        nextByte = a[i];
        buf[b++] = bin_to_hex(nextByte >> 4);
        buf[b++] = bin_to_hex(nextByte & 0x0f);
        if (i < len - 1) {
            buf[b++] = '-';
        }
    }
    buf[b] = '\0';
    return b;
}

#define SFLOW_HEX_SCRATCH 1024

static void
process_flow_sample(struct sflow_xdr *x)
{
    if (x->offset.HEADER) {
        uint32_t headerLen;
        char scratch[SFLOW_HEX_SCRATCH];

        printf("HEADER");
        printf(" dgramSeqNo=%"PRIu32, x->dgramSeqNo);
        printf(" ds=%s>%"PRIu32":%"PRIu32,
               x->agentIPStr, x->dsClass, x->dsIndex);
        printf(" fsSeqNo=%"PRIu32, x->fsSeqNo);

        if (x->offset.SWITCH) {
            sflowxdr_setc(x, x->offset.SWITCH);
            printf(" in_vlan=%"PRIu32, sflowxdr_next(x));
            printf(" in_priority=%"PRIu32, sflowxdr_next(x));
            printf(" out_vlan=%"PRIu32, sflowxdr_next(x));
            printf(" out_priority=%"PRIu32, sflowxdr_next(x));
        }

        sflowxdr_setc(x, x->offset.HEADER);
        printf(" meanSkip=%"PRIu32, x->meanSkipCount);
        printf(" samplePool=%"PRIu32, x->samplePool);
        printf(" dropEvents=%"PRIu32, x->dropEvents);
        printf(" in_ifindex=%"PRIu32, x->inputPort);
        printf(" in_format=%"PRIu32, x->inputPortFormat);
        printf(" out_ifindex=%"PRIu32, x->outputPort);
        printf(" out_format=%"PRIu32, x->outputPortFormat);
        printf(" hdr_prot=%"PRIu32, sflowxdr_next(x));
        printf(" pkt_len=%"PRIu32, sflowxdr_next(x));
        printf(" stripped=%"PRIu32, sflowxdr_next(x));
        headerLen = sflowxdr_next(x);
        printf(" hdr_len=%"PRIu32, headerLen);
        print_hex(sflowxdr_str(x), headerLen, scratch, SFLOW_HEX_SCRATCH);
        printf(" hdr=%s", scratch);
        printf("\n");
    }
}

static void
process_datagram(struct sflow_xdr *x)
{
    uint32_t samples, s;

    SFLOWXDR_assert(x, (sflowxdr_next(x) == SFLOW_VERSION_5));

    /* Read the sFlow header. */
    x->agentAddr.type = sflowxdr_next(x);
    switch (x->agentAddr.type) {
    case SFLOW_ADDRTYPE_IP4:
        x->agentAddr.a.ip4 = sflowxdr_next_n(x);
        break;

    case SFLOW_ADDRTYPE_IP6:
        x->agentAddr.a.ip6[0] = sflowxdr_next_n(x);
        x->agentAddr.a.ip6[1] = sflowxdr_next_n(x);
        x->agentAddr.a.ip6[2] = sflowxdr_next_n(x);
        x->agentAddr.a.ip6[3] = sflowxdr_next_n(x);
        break;

    case SFLOW_ADDRTYPE_undefined:
    default:
        SFLOWXDR_throw(x);
        break;
    }
    x->subAgentId = sflowxdr_next(x);
    x->dgramSeqNo = sflowxdr_next(x);
    x->uptime_mS = sflowxdr_next(x);

    /* Store the agent address as a string. */
    if (x->agentAddr.type == SFLOW_ADDRTYPE_IP6) {
        char ipstr[INET6_ADDRSTRLEN];
        inet_ntop(AF_INET6, (const void *) &x->agentAddr.a.ip6,
                  ipstr, INET6_ADDRSTRLEN);
        snprintf(x->agentIPStr, sizeof x->agentIPStr, "[%s]", ipstr);
    } else {
        snprintf(x->agentIPStr, sizeof x->agentIPStr,
                 IP_FMT, IP_ARGS(x->agentAddr.a.ip4));
    }

    /* Array of flow/counter samples. */
    samples = sflowxdr_next(x);
    for (s = 0; s < samples; s++) {
        uint32_t sType = sflowxdr_next(x);
        uint32_t sQuads = sflowxdr_next(x) >> 2;
        uint32_t sMark = sflowxdr_mark(x, sQuads);
        SFLOWXDR_assert(x, sflowxdr_more(x, sQuads));

        switch (sType) {
        case SFLOW_COUNTERS_SAMPLE_EXPANDED:
        case SFLOW_COUNTERS_SAMPLE:
        {
            uint32_t csElements, e;
            uint32_t ceTag, ceQuads, ceMark, csEnd;

            x->csSeqNo = sflowxdr_next(x);
            if (sType == SFLOW_COUNTERS_SAMPLE_EXPANDED) {
                x->dsClass = sflowxdr_next(x);
                x->dsIndex = sflowxdr_next(x);
            } else {
                uint32_t dsCombined = sflowxdr_next(x);
                x->dsClass = dsCombined >> 24;
                x->dsIndex = dsCombined & 0x00FFFFFF;
            }

            csElements = sflowxdr_next(x);
            for (e = 0; e < csElements; e++) {
                SFLOWXDR_assert(x, sflowxdr_more(x,2));
                ceTag = sflowxdr_next(x);
                ceQuads = sflowxdr_next(x) >> 2;
                ceMark = sflowxdr_mark(x, ceQuads);
                SFLOWXDR_assert(x, sflowxdr_more(x,ceQuads));
                /* Only care about selected structures.  Just record their
                 * offsets here. We'll read the fields out later. */
                switch (ceTag) {
                case SFLOW_TAG_CTR_IFCOUNTERS:
                    sflowxdr_mark_unique(x, &x->offset.IFCOUNTERS);
                    break;

                    /* Add others here... */
                }

                sflowxdr_skip(x, ceQuads);
                SFLOWXDR_assert(x, sflowxdr_mark_ok(x, ceMark));
            }

            csEnd = sflowxdr_mark(x, 0);
            process_counter_sample(x);
            /* Make sure we pick up the decoding where we left off. */
            sflowxdr_setc(x, csEnd);

            /* Clear the offsets for the next sample. */
            memset(&x->offset, 0, sizeof x->offset);
        }
        break;

        case SFLOW_FLOW_SAMPLE:
        case SFLOW_FLOW_SAMPLE_EXPANDED:
        {
            uint32_t fsElements, e;
            uint32_t feTag, feQuads, feMark, fsEnd;
            x->fsSeqNo = sflowxdr_next(x);
            if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
                x->dsClass = sflowxdr_next(x);
                x->dsIndex = sflowxdr_next(x);
            } else {
                uint32_t dsCombined = sflowxdr_next(x);
                x->dsClass = dsCombined >> 24;
                x->dsIndex = dsCombined & 0x00FFFFFF;
            }
            x->meanSkipCount = sflowxdr_next(x);
            x->samplePool = sflowxdr_next(x);
            x->dropEvents = sflowxdr_next(x);
            if (sType == SFLOW_FLOW_SAMPLE_EXPANDED) {
                x->inputPortFormat = sflowxdr_next(x);
                x->inputPort = sflowxdr_next(x);
                x->outputPortFormat = sflowxdr_next(x);
                x->outputPort = sflowxdr_next(x);
            } else {
                uint32_t inp, outp;

                inp = sflowxdr_next(x);
                outp = sflowxdr_next(x);
                x->inputPortFormat = inp >> 30;
                x->inputPort = inp & 0x3fffffff;
                x->outputPortFormat = outp >> 30;
                x->outputPort = outp & 0x3fffffff;
            }
            fsElements = sflowxdr_next(x);
            for (e = 0; e < fsElements; e++) {
                SFLOWXDR_assert(x, sflowxdr_more(x,2));
                feTag = sflowxdr_next(x);
                feQuads = sflowxdr_next(x) >> 2;
                feMark = sflowxdr_mark(x, feQuads);
                SFLOWXDR_assert(x, sflowxdr_more(x,feQuads));
                /* Only care about selected structures.  Just record their
                 * offsets here. We'll read the fields out below. */
                switch (feTag) {
                case SFLOW_TAG_PKT_HEADER:
                    sflowxdr_mark_unique(x, &x->offset.HEADER);
                    break;

                case SFLOW_TAG_PKT_SWITCH:
                    sflowxdr_mark_unique(x, &x->offset.SWITCH);
                    break;

                    /* Add others here... */
                }

                sflowxdr_skip(x, feQuads);
                SFLOWXDR_assert(x, sflowxdr_mark_ok(x, feMark));
            }

            fsEnd = sflowxdr_mark(x, 0);
            process_flow_sample(x);
            /* Make sure we pick up the decoding where we left off. */
            sflowxdr_setc(x, fsEnd);

            /* Clear the offsets for the next counter/flow sample. */
            memset(&x->offset, 0, sizeof x->offset);
        }
        break;

        default:
            /* Skip other sample types. */
            sflowxdr_skip(x, sQuads);
        }
        SFLOWXDR_assert(x, sflowxdr_mark_ok(x, sMark));
    }
}

static void
print_sflow(struct ofpbuf *buf)
{
    char *dgram_buf;
    int dgram_len = ofpbuf_size(buf);
    struct sflow_xdr xdrDatagram;
    struct sflow_xdr *x = &xdrDatagram;

    memset(x, 0, sizeof *x);
    if (SFLOWXDR_try(x)) {
        SFLOWXDR_assert(x, (dgram_buf = ofpbuf_try_pull(buf, ofpbuf_size(buf))));
        sflowxdr_init(x, dgram_buf, dgram_len);
        SFLOWXDR_assert(x, dgram_len >= SFLOW_MIN_LEN);
        process_datagram(x);
    } else {
        // CATCH
        printf("\n>>>>> ERROR in " __FILE__ " at line %u\n", x->errline);
    }
}

static void
test_sflow_main(int argc, char *argv[])
{
    struct unixctl_server *server;
    enum { MAX_RECV = 1500 };
    const char *target;
    struct ofpbuf buf;
    bool exiting = false;
    int error;
    int sock;

    proctitle_init(argc, argv);
    set_program_name(argv[0]);
    parse_options(argc, argv);

    if (argc - optind != 1) {
        ovs_fatal(0, "exactly one non-option argument required "
                  "(use --help for help)");
    }
    target = argv[optind];

    sock = inet_open_passive(SOCK_DGRAM, target, 0, NULL, 0);
    if (sock < 0) {
        ovs_fatal(0, "%s: failed to open (%s)", argv[1], ovs_strerror(-sock));
    }

    daemon_save_fd(STDOUT_FILENO);
    daemonize_start();

    error = unixctl_server_create(NULL, &server);
    if (error) {
        ovs_fatal(error, "failed to create unixctl server");
    }
    unixctl_command_register("exit", "", 0, 0, test_sflow_exit, &exiting);

    daemonize_complete();

    ofpbuf_init(&buf, MAX_RECV);
    for (;;) {
        int retval;

        unixctl_server_run(server);

        ofpbuf_clear(&buf);
        do {
            retval = read(sock, ofpbuf_data(&buf), buf.allocated);
        } while (retval < 0 && errno == EINTR);
        if (retval > 0) {
            ofpbuf_put_uninit(&buf, retval);
            print_sflow(&buf);
            fflush(stdout);
        }

        if (exiting) {
            break;
        }

        poll_fd_wait(sock, POLLIN);
        unixctl_server_wait(server);
        poll_block();
    }
}

static void
parse_options(int argc, char *argv[])
{
    enum {
        DAEMON_OPTION_ENUMS,
        VLOG_OPTION_ENUMS
    };
    static const struct option long_options[] = {
        {"verbose", optional_argument, NULL, 'v'},
        {"help", no_argument, NULL, 'h'},
        DAEMON_LONG_OPTIONS,
        VLOG_LONG_OPTIONS,
        {NULL, 0, NULL, 0},
    };
    char *short_options = long_options_to_short_options(long_options);

    for (;;) {
        int c = getopt_long(argc, argv, short_options, long_options, NULL);
        if (c == -1) {
            break;
        }

        switch (c) {
        case 'h':
            usage();

        DAEMON_OPTION_HANDLERS
        VLOG_OPTION_HANDLERS

        case '?':
            exit(EXIT_FAILURE);

        default:
            abort();
        }
    }
    free(short_options);
}

static void
usage(void)
{
    printf("%s: sflow collector test utility\n"
           "usage: %s [OPTIONS] PORT[:IP]\n"
           "where PORT is the UDP port to listen on and IP is optionally\n"
           "the IP address to listen on.\n",
           program_name, program_name);
    daemon_usage();
    vlog_usage();
    printf("\nOther options:\n"
           "  -h, --help                  display this help message\n");
    exit(EXIT_SUCCESS);
}

static void
test_sflow_exit(struct unixctl_conn *conn,
                int argc OVS_UNUSED, const char *argv[] OVS_UNUSED,
                void *exiting_)
{
    bool *exiting = exiting_;
    *exiting = true;
    unixctl_command_reply(conn, NULL);
}

OVSTEST_REGISTER("test-sflow", test_sflow_main);
Commit	Line	Data
8073dd31	1	/*
e731d71b	2	* Copyright (c) 2011, 2012, 2013, 2014 Nicira, Inc.
8073dd31 NM	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
b7807e4f	20	#include <arpa/inet.h>
8073dd31 NM	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"
eadd1644	39	#include "ovstest.h"
8073dd31 NM	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
8073dd31 NM	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;
e731d71b	86	char agentIPStr[INET6_ADDRSTRLEN + 2];
8073dd31 NM	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) {
e731d71b AS	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);
8073dd31	333	} else {
e731d71b	334	snprintf(x->agentIPStr, sizeof x->agentIPStr,
8073dd31 NM	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;
1f317cb5	473	int dgram_len = ofpbuf_size(buf);
8073dd31 NM	474	struct sflow_xdr xdrDatagram;
	475	struct sflow_xdr *x = &xdrDatagram;
	476
	477	memset(x, 0, sizeof *x);
	478	if (SFLOWXDR_try(x)) {
1f317cb5	479	SFLOWXDR_assert(x, (dgram_buf = ofpbuf_try_pull(buf, ofpbuf_size(buf))));
8073dd31 NM	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
eadd1644 AZ	489	static void
eadd1644 AZ	490	test_sflow_main(int argc, char *argv[])
8073dd31 NM	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) {
10a89ef0	512	ovs_fatal(0, "%s: failed to open (%s)", argv[1], ovs_strerror(-sock));
8073dd31 NM	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 {
1f317cb5	534	retval = read(sock, ofpbuf_data(&buf), buf.allocated);
8073dd31 NM	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	}
8073dd31 NM	550	}
	551
	552	static void
	553	parse_options(int argc, char *argv[])
	554	{
	555	enum {
977529e4 BP	556	DAEMON_OPTION_ENUMS,
977529e4 BP	557	VLOG_OPTION_ENUMS
8073dd31	558	};
07fc4ed3	559	static const struct option long_options[] = {
8073dd31 NM	560	{"verbose", optional_argument, NULL, 'v'},
	561	{"help", no_argument, NULL, 'h'},
	562	DAEMON_LONG_OPTIONS,
977529e4	563	VLOG_LONG_OPTIONS,
8073dd31 NM	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
977529e4 BP	578	DAEMON_OPTION_HANDLERS
977529e4 BP	579	VLOG_OPTION_HANDLERS
8073dd31 NM	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	}
eadd1644 AZ	615
eadd1644 AZ	616	OVSTEST_REGISTER("test-sflow", test_sflow_main);